rgw: emit bucket logging record for lifecycle multipart upload abort

Signed-off-by: ShreeJejurikar <shreemj8@gmail.com>

diff --git a/src/rgw/rgw_lc.cc b/src/rgw/rgw_lc.cc
index 8b6adba2b8c791e8cebcc2089471eda09a7d7723..941c955e4b61f89a3ee4e76893bfee526e70a211 100644 (file)
--- 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 9e87c30e86c6bf48a241cab033f87a4c373d1ad4..f798ccc72cafc7a1559b934c1b08d69fbf87a0e5 100644 (file)
--- 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."""