From b126cffe85e95d94e7e048bd9e51a97ec8840938 Mon Sep 17 00:00:00 2001 From: ShreeJejurikar Date: Thu, 11 Jun 2026 16:39:48 +0530 Subject: [PATCH] rgw: emit bucket logging record for lifecycle multipart upload abort Signed-off-by: ShreeJejurikar --- src/rgw/rgw_lc.cc | 14 +++-- .../rgw/bucket_logging/test_bucket_logging.py | 62 +++++++++++++++++++ 2 files changed, 72 insertions(+), 4 deletions(-) diff --git a/src/rgw/rgw_lc.cc b/src/rgw/rgw_lc.cc index 8b6adba2b8c..941c955e4b6 100644 --- a/src/rgw/rgw_lc.cc +++ b/src/rgw/rgw_lc.cc @@ -707,8 +707,9 @@ static void send_notification(const DoutPrefixProvider* dpp, } } -// Op name used in journal records for LC-driven deletions. +// Op names used in records for LC-driven actions. static const std::string lifecycle_delete_object_op = "LIFECYCLE.DELETE.OBJECT"; +static const std::string lifecycle_delete_upload_op = "LIFECYCLE.DELETE.UPLOAD"; // LC has no req_state, so the source bucket's owner stands in as the // identity for the log-bucket write permission check. @@ -720,7 +721,8 @@ static void send_log_record(const DoutPrefixProvider* dpp, const std::string& etag, uint64_t size, const std::string& version_id, - const std::string& op_name) { + const std::string& op_name, + rgw::bucketlogging::LoggingType logging_type) { rgw::bucketlogging::record_input input; input.bucket = bucket; input.user_or_account = to_string(bucket->get_owner()); @@ -729,7 +731,7 @@ static void send_log_record(const DoutPrefixProvider* dpp, const int ret = rgw::bucketlogging::log_record( driver, - rgw::bucketlogging::LoggingType::Journal, + logging_type, obj, input, op_name, etag, size, dpp, y, /*async_completion=*/true, /*log_source_bucket=*/false); @@ -836,7 +838,8 @@ static int remove_expired_obj(const DoutPrefixProvider* dpp, } if (log_op_name) { send_log_record(dpp, y, driver, obj.get(), oc.bucket, etag, size, - version_id, *log_op_name); + version_id, *log_op_name, + rgw::bucketlogging::LoggingType::Journal); } } @@ -1001,6 +1004,9 @@ int RGWLC::handle_multipart_expiration(rgw::sal::Bucket* target, const auto event_type = rgw::notify::ObjectExpirationAbortMPU; send_notification(this, y, driver, sal_obj.get(), target, etag, size, obj.key.instance, {event_type}); + send_log_record(this, y, driver, sal_obj.get(), target, etag, size, + obj.key.instance, lifecycle_delete_upload_op, + rgw::bucketlogging::LoggingType::Standard); if (perfcounter) { perfcounter->inc(l_rgw_lc_abort_mpu, 1); } diff --git a/src/test/rgw/bucket_logging/test_bucket_logging.py b/src/test/rgw/bucket_logging/test_bucket_logging.py index 9e87c30e86c..f798ccc72ca 100644 --- a/src/test/rgw/bucket_logging/test_bucket_logging.py +++ b/src/test/rgw/bucket_logging/test_bucket_logging.py @@ -381,6 +381,28 @@ def read_journal_records(s3_client, log_bucket, source_bucket, prefix=None, sett return records +def wait_for_mpu_gone(s3_client, bucket, upload_id, timeout=30, interval=1): + deadline = time.time() + timeout + while time.time() < deadline: + resp = s3_client.list_multipart_uploads(Bucket=bucket) + if not any(u.get('UploadId') == upload_id for u in resp.get('Uploads', [])): + return True + time.sleep(interval) + return False + + +def abort_pending_mpus(s3_client, bucket): + try: + resp = s3_client.list_multipart_uploads(Bucket=bucket) + except ClientError: + return + for u in resp.get('Uploads', []): + try: + s3_client.abort_multipart_upload(Bucket=bucket, Key=u['Key'], UploadId=u['UploadId']) + except ClientError: + pass + + @pytest.fixture def lc_fast(): set_lc_debug_interval(5) @@ -1015,6 +1037,46 @@ def test_lc_current_expiration_dm_branch_does_not_log(s3_client, logging_type, l cleanup_bucket(s3_client, log_bucket) +@pytest.mark.basic_test +def test_lc_abort_mpu_logs_standard_record(s3_client, logging_type, lc_fast): + """LC AbortIncompleteMultipartUpload emits a Standard-mode LIFECYCLE.DELETE.UPLOAD record.""" + if logging_type != 'Standard': + pytest.skip("AbortIncompleteMultipartUpload logging is Standard-mode-only") + source = gen_bucket_name("lc-mpu-source") + log_bucket = gen_bucket_name("lc-mpu-log") + key = 'incomplete-mpu.bin' + + try: + assert create_bucket_with_logging(s3_client, source, log_bucket) + apply_lc_config(s3_client, source, [ + make_lc_rule(AbortIncompleteMultipartUpload={'DaysAfterInitiation': 1}) + ]) + + upload_id = s3_client.create_multipart_upload(Bucket=source, Key=key)['UploadId'] + + time.sleep(lc_fast + 2) + trigger_lc_processing() + assert wait_for_mpu_gone(s3_client, source, upload_id), "LC did not abort the incomplete multipart upload" + + time.sleep(5) + admin(['bucket', 'logging', 'flush', '--bucket', source]) + resp = s3_client.list_objects_v2(Bucket=log_bucket, Prefix=f'{source}/') + log_keys = [obj['Key'] for obj in resp.get('Contents', [])] + assert log_keys, "no log object emitted for MPU abort" + + bodies = [] + for log_key in log_keys: + body = s3_client.get_object(Bucket=log_bucket, Key=log_key)['Body'].read().decode('utf-8') + bodies.append(body) + joined = '\n'.join(bodies) + assert 'LIFECYCLE.DELETE.UPLOAD' in joined, f"LIFECYCLE.DELETE.UPLOAD not found in log: {joined!r}" + + finally: + abort_pending_mpus(s3_client, source) + cleanup_bucket(s3_client, source) + cleanup_bucket(s3_client, log_bucket) + + @pytest.mark.basic_test def test_lc_runs_safely_without_logging_config(s3_client, lc_fast): """LC runs successfully on a bucket without bucket-logging configured.""" -- 2.47.3