Compacted topic uploads

[abhijat]

Cover letter

Describe in plain language the motivation (bug, feature, etc.) behind the change in this PR and how the included commits address it.

Fixes #ISSUE-NUMBER, Fixes #ISSUE-NUMBER, ...

Backport Required

• [ ] not a bug fix
• [ ] issue does not exist in previous branches
• [ ] papercut/not impactful enough to backport
• [ ] v22.2.x
• [ ] v22.1.x
• [ ] v21.11.x

UX changes

Describe in plain language how this PR affects an end-user. What topic flags, configuration flags, command line flags, deprecation policies etc are added/changed.

Release notes

[mmedenjak]

Does this fix https://github.com/redpanda-data/redpanda/issues/6377?

[abhijat]

Does this fix #6377?

Yes eventually, currently it's an experiment to test out some things. I have to add a few more things and rebase other prs.

[abhijat]

• [x] read locking for all segments in concatenated upload
• [x] add compacted upload results to batch result
• [x] property to turn off compacted segment uploads defaulting to false
• [x] ntp archiver tests
• [x] ducktape tests

[Lazin]

In ntp_archiver_service.cc line 401 we have this code:

    auto path = cloud_storage::generate_remote_segment_path(
      _ntp, _rev, candidate.exposed_name, _start_term);

which has to be changed. It uses generate_remote_segment_path which always generates the old style segment paths. The problem is that this path doesn't include the committed offset so when you will re-upload the compacted segment it will re-write the old non-compacted segment since the names are the same. Instead of this function the remote_partition::generate_remote_segment_name is supposed to be used. This function will generate different paths depending on segment_meta.

[abhijat]

In ntp_archiver_service.cc line 401 we have this code:

    auto path = cloud_storage::generate_remote_segment_path(
      _ntp, _rev, candidate.exposed_name, _start_term);

which has to be changed. It uses generate_remote_segment_path which always generates the old style segment paths. The problem is that this path doesn't include the committed offset so when you will re-upload the compacted segment it will re-write the old non-compacted segment since the names are the same. Instead of this function the remote_partition::generate_remote_segment_name is supposed to be used. This function will generate different paths depending on segment_meta.

Is the new function to use partition_manifest::generate_remote_segment_path? I have changed to the following:

    cloud_storage::partition_manifest::key key{
      .base_offset = candidate.starting_offset,
      .term = _start_term,
    };

    cloud_storage::partition_manifest::value val{
      // Using compaction info from the "head" segment, although the candidate
      // might contain many segments
      .is_compacted = candidate.source->is_compacted_segment()
                      && candidate.source->finished_self_compaction(),
      .size_bytes = candidate.content_length,
      .base_offset = candidate.starting_offset,
      .committed_offset = candidate.final_offset,
      .base_timestamp = candidate.base_timestamp,
      .max_timestamp = candidate.max_timestamp,
      .ntp_revision = _rev,
      .archiver_term = _start_term,
      .version = cloud_storage::segment_meta_version::v2,
    };

    auto path = manifest().generate_remote_segment_path(_ntp, key, val);

[abhijat]

In ntp_archiver_service.cc line 401 we have this code:

    auto path = cloud_storage::generate_remote_segment_path(
      _ntp, _rev, candidate.exposed_name, _start_term);

which has to be changed. It uses generate_remote_segment_path which always generates the old style segment paths. The problem is that this path doesn't include the committed offset so when you will re-upload the compacted segment it will re-write the old non-compacted segment since the names are the same. Instead of this function the remote_partition::generate_remote_segment_name is supposed to be used. This function will generate different paths depending on segment_meta.

Is the new function to use partition_manifest::generate_remote_segment_path? I have changed to the following:

    ...
    auto path = manifest().generate_remote_segment_path(_ntp, key, val);

@Lazin one issue I am seeing after I made this change is that the URL contains the archiver term twice (.5.5):

Uploading segment to path {"264fcd31/kafka/test-topic/42_0/251-999-5-v1.log.5.5"}

I guess this is not expected behaviour?

In partition_manifest::generate_remote_segment_path archiver term is appended twice to path, once in partition_manifest::generate_remote_segment_name and then again in cloud_storage::generate_remote_segment_path. Is some other method supposed to be used here?

[abhijat]

need to look into several failing ducktape tests

[abhijat]

#6893 created to reduce the review surface area for this large PR (mostly isolated changes related to http test utils)

[abhijat]

@Lazin I have added the three fields to scheduled_upload.meta . The delta offset end calculation looks like

    auto ot_state = _partition->get_offset_translator_state();
    auto delta = base - model::offset_cast(ot_state->from_log_offset(base));
    auto delta_offset_end = upload.final_offset
                            - model::offset_cast(
                              ot_state->from_log_offset(upload.final_offset));

does this look correct?

[abhijat]

@Lazin changes for missing offset after compaction as discussed in this commit https://github.com/redpanda-data/redpanda/pull/6455/commits/0f2cebce3b4f3ec4fe741aa34855cfd5dcfd2e6e

[jcsp]

This is ready to be rebased now that Vlad's PR merged

[abhijat]

release build test failure is #6840 for which a fix is in progress #6842

https://buildkite.com/redpanda/redpanda/builds/17601#01842cd4-fc42-44c8-962f-64d7d6e4d632

AssertionError("NTP NTP(ns='kafka', topic='panda-topic', partition=0) the restored partition is too small 9733470. The original is 10500746 bytes which 767276 bytes larger.")
  | Traceback (most recent call last):
  | File "/usr/local/lib/python3.10/dist-packages/ducktape/tests/runner_client.py", line 135, in run
  | data = self.run_test()
  | File "/usr/local/lib/python3.10/dist-packages/ducktape/tests/runner_client.py", line 227, in run_test
  | return self.test_context.function(self.test)
  | File "/root/tests/rptest/services/cluster.py", line 35, in wrapped
  | r = f(self, *args, **kwargs)
  | File "/root/tests/rptest/tests/topic_recovery_test.py", line 1331, in test_fast1
  | self.do_run(test_case)
  | File "/root/tests/rptest/tests/topic_recovery_test.py", line 1254, in do_run
  | test_case.validate_cluster(baseline, restored)
  | File "/root/tests/rptest/tests/topic_recovery_test.py", line 698, in validate_cluster
  | verify_file_layout(baseline,
  | File "/root/tests/rptest/utils/si_utils.py", line 205, in verify_file_layout
  | assert delta <= BLOCK_SIZE, \
  | AssertionError: NTP NTP(ns='kafka', topic='panda-topic', partition=0) the restored partition is too small 9733470. The original is 10500746 bytes which 767276 bytes larger.

[abhijat]

debug failure is instance of #6892 https://buildkite.com/redpanda/redpanda/builds/17601#01842cd4-fc44-4ae3-afc4-e0b01aa3a617

[abhijat]

/ci-repeat 5

[abhijat]

https://buildkite.com/redpanda/redpanda/builds/17607#01842d5c-caa8-4cdb-a3f0-62c21a40fa99

instance of #6991

test_id:    rptest.tests.upgrade_test.UpgradeFromPriorFeatureVersionCloudStorageTest.test_rolling_upgrade
--
  | status:     FAIL
  | run time:   16.720 seconds
  |  
  |  
  | KeyError('3-1-v1.log')
  | Traceback (most recent call last):
  | File "/root/tests/rptest/services/redpanda.py", line 1984, in get_sizes
  | stat = node.account.sftp_client.lstat(
  | File "/usr/local/lib/python3.10/dist-packages/paramiko/sftp_client.py", line 511, in lstat
  | t, msg = self._request(CMD_LSTAT, path)
  | File "/usr/local/lib/python3.10/dist-packages/paramiko/sftp_client.py", line 822, in _request
  | return self._read_response(num)
  | File "/usr/local/lib/python3.10/dist-packages/paramiko/sftp_client.py", line 874, in _read_response
  | self._convert_status(msg)
  | File "/usr/local/lib/python3.10/dist-packages/paramiko/sftp_client.py", line 903, in _convert_status
  | raise IOError(errno.ENOENT, text)
  | FileNotFoundError: [Errno 2] No such file

[jcsp]

^[[0m[ERROR:2022-11-01 12:09:02,525]: Failed to import rptest.tests.retention_policy_test, which may indicate a broken test that cannot be loaded: ImportError: cannot import name 'S3View' from 'rptest.utils.si_utils' (/root/tests/rptest/utils/si_utils.py)^M