rgw/logging: support source and destination buckets on different tenants

Signed-off-by: Yuval Lifshitz <ylifshit@ibm.com>

3 files changed, 77 insertions, 44 deletions

diff --git a/doc/radosgw/bucket_logging.rst b/doc/radosgw/bucket_logging.rst
index cb9f8465d20..1ac4e546a4c 100644
--- a/doc/radosgw/bucket_logging.rst
+++ b/doc/radosgw/bucket_logging.rst
@@ -72,7 +72,7 @@ has the following format:
 
 ::
 
-  <prefix><bucket owner>/<source region>/<bucket name>/<year>/<month>/<day>/<year-month-day-hour-minute-second>-<16 bytes unique-id>
+  <prefix><bucket owner>/<source region>/[tenant:]<bucket name>/<year>/<month>/<day>/<year-month-day-hour-minute-second>-<16 bytes unique-id>
 
 For example:
 
@@ -90,7 +90,7 @@ Journal
 minimum amount of data used for journaling bucket changes (this is a Ceph extension).
 
   - bucket owner (or dash if empty)
-  - bucket name (or dash if empty)
+  - bucket name (or dash if empty). in the format: ``[tenant:]<bucket name>``
   - time in the following format: ``[day/month/year:hour:minute:second timezone]``
   - object key (or dash if empty)
   - operation in the following format: ``WEBSITE/REST.<HTTP method>.<resource>``
@@ -111,7 +111,7 @@ Standard
 based on `AWS Logging Record Format`_.
   
   - bucket owner (or dash if empty)
-  - bucket name (or dash if empty)
+  - bucket name (or dash if empty). in the format: ``[tenant:]<bucket name>``
   - time
   - remote IP (not supported, always a dash)
   - user or account (or dash if empty)
diff --git a/src/rgw/rgw_bucket_logging.cc b/src/rgw/rgw_bucket_logging.cc
index d24a53024f1..efc4ab3d7f4 100644
--- a/src/rgw/rgw_bucket_logging.cc
+++ b/src/rgw/rgw_bucket_logging.cc
@@ -206,6 +206,13 @@ ceph::coarse_real_time time_from_name(const std::string& obj_name, const DoutPre
   return extracted_time;
 }
 
+std::string full_bucket_name(const std::unique_ptr<rgw::sal::Bucket>& bucket) {
+  if (bucket->get_tenant().empty()) {
+    return bucket->get_name();
+  }
+  return fmt::format("{}:{}", bucket->get_tenant(), bucket->get_name());
+}
+
 int new_logging_object(const configuration& conf,
     const std::unique_ptr<rgw::sal::Bucket>& bucket,
     std::string& obj_name,
@@ -235,7 +242,7 @@ int new_logging_object(const configuration& conf,
           conf.target_prefix,
           to_string(bucket->get_owner()),
           source_region,
-          bucket->get_name(),
+          full_bucket_name(bucket),
           t,
           t,
           unique);
@@ -270,8 +277,11 @@ int rollover_logging_object(const configuration& conf,
     optional_yield y,
     bool must_commit,
     RGWObjVersionTracker* objv_tracker) {
-  if (conf.target_bucket != bucket->get_name()) {
-    ldpp_dout(dpp, 1) << "ERROR: bucket name mismatch: '" << conf.target_bucket << "' != '" << bucket->get_name() << "'" << dendl;
+  std::string target_bucket_name;
+  std::string target_tenant_name;
+  std::ignore = rgw_parse_url_bucket(conf.target_bucket, bucket->get_tenant(), target_tenant_name, target_bucket_name);
+  if (target_bucket_name != bucket->get_name() || target_tenant_name != bucket->get_tenant()) {
+    ldpp_dout(dpp, 1) << "ERROR: bucket name mismatch: '" << conf.target_bucket << "' != '" << full_bucket_name(bucket) << "'" << dendl;
     return -EINVAL;
   }
   const auto old_obj = obj_name;
@@ -357,11 +367,19 @@ int log_record(rgw::sal::Driver* driver,
     ldpp_dout(dpp, 1) << "ERROR: only bucket operations are logged" << dendl;
     return -EINVAL;
   }
+  std::string target_bucket_name;
+  std::string target_tenant_name;
+  auto ret = rgw_parse_url_bucket(conf.target_bucket, s->bucket_tenant, target_tenant_name, target_bucket_name);
+  if (ret < 0) {
+    ldpp_dout(dpp, 1) << "ERROR: failed to parse target bucket '" << conf.target_bucket << "', ret = " << ret << dendl;
+    return ret;
+  }
+  const rgw_bucket target_bucket_id(target_tenant_name, target_bucket_name);
   std::unique_ptr<rgw::sal::Bucket> target_bucket;
-  auto ret = driver->load_bucket(dpp, rgw_bucket(s->bucket_tenant, conf.target_bucket),
+  ret = driver->load_bucket(dpp, target_bucket_id,
                                &target_bucket, y);
   if (ret < 0) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get target logging bucket '" << conf.target_bucket << "'. ret = " << ret << dendl;
+    ldpp_dout(dpp, 1) << "ERROR: failed to get target logging bucket '" << target_bucket_id << "'. ret = " << ret << dendl;
     return ret;
   }
   std::string obj_name;
@@ -420,7 +438,7 @@ int log_record(rgw::sal::Driver* driver,
     bucket_name = s->src_bucket_name;
   } else {
     bucket_owner = to_string( s->bucket->get_owner());
-    bucket_name = s->bucket->get_name();
+    bucket_name = full_bucket_name(s->bucket);
   }
 
   switch (conf.logging_type) {
@@ -459,7 +477,7 @@ int log_record(rgw::sal::Driver* driver,
     case LoggingType::Journal:
       record = fmt::format("{} {} [{:%d/%b/%Y:%H:%M:%S %z}] {} {} {} {} {}",
         dash_if_empty(to_string(s->bucket->get_owner())),
-        dash_if_empty(s->bucket->get_name()),
+        dash_if_empty(full_bucket_name(s->bucket)),
         t,
         op_name,
         dash_if_empty_or_null(obj, obj->get_name()),
@@ -543,10 +561,10 @@ int log_record(rgw::sal::Driver* driver,
         return 0;
       }
     }
-    ldpp_dout(dpp, 20) << "INFO: found matching logging configuration of bucket '" << s->bucket->get_name() << 
+    ldpp_dout(dpp, 20) << "INFO: found matching logging configuration of bucket '" << full_bucket_name(s->bucket) << 
       "' configuration: " << configuration.to_json_str() << dendl;
     if (auto ret = log_record(driver, obj, s, op_name, etag, size, configuration, dpp, y, async_completion, log_source_bucket); ret < 0) { 
-      ldpp_dout(dpp, 1) << "ERROR: failed to perform logging for bucket '" << s->bucket->get_name() << 
+      ldpp_dout(dpp, 1) << "ERROR: failed to perform logging for bucket '" << full_bucket_name(s->bucket) << 
         "'. ret=" << ret << dendl;
       return ret;
     }
diff --git a/src/rgw/rgw_rest_bucket_logging.cc b/src/rgw/rgw_rest_bucket_logging.cc
index ed12ce855a9..0e7b53120db 100644
--- a/src/rgw/rgw_rest_bucket_logging.cc
+++ b/src/rgw/rgw_rest_bucket_logging.cc
@@ -58,16 +58,15 @@ public:
       return;
     }
 
-    std::unique_ptr<rgw::sal::Bucket> bucket;
-    op_ret = driver->load_bucket(this, rgw_bucket(s->bucket_tenant, s->bucket_name),
-                                 &bucket, y);
+    const rgw_bucket src_bucket_id(s->bucket_tenant, s->bucket_name);
+    std::unique_ptr<rgw::sal::Bucket> src_bucket;
+    op_ret = driver->load_bucket(this, src_bucket_id,
+                                 &src_bucket, y);
     if (op_ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to get bucket '" << 
-        (s->bucket_tenant.empty() ? s->bucket_name : s->bucket_tenant + ":" + s->bucket_name) << 
-        "' info, ret = " << op_ret << dendl;
+      ldpp_dout(this, 1) << "ERROR: failed to get bucket '" << src_bucket_id << "', ret = " << op_ret << dendl;
       return;
     }
-    if (auto iter = bucket->get_attrs().find(RGW_ATTR_BUCKET_LOGGING); iter != bucket->get_attrs().end()) {
+    if (auto iter = src_bucket->get_attrs().find(RGW_ATTR_BUCKET_LOGGING); iter != src_bucket->get_attrs().end()) {
       try {
         configuration.enabled = true;
         decode(configuration, iter->second);
@@ -78,10 +77,10 @@ public:
         return;
       }
     } else {
-      ldpp_dout(this, 5) << "WARNING: no logging configuration on bucket '" << bucket->get_name() << "'" << dendl;
+      ldpp_dout(this, 5) << "WARNING: no logging configuration on bucket '" << src_bucket_id << "'" << dendl;
       return;
     }
-    ldpp_dout(this, 20) << "INFO: found logging configuration on bucket '" << bucket->get_name() << "'" 
+    ldpp_dout(this, 20) << "INFO: found logging configuration on bucket '" << src_bucket_id << "'" 
       << "'. configuration: " << configuration.to_json_str() << dendl;
   }
 
@@ -159,32 +158,40 @@ class RGWPutBucketLoggingOp : public RGWDefaultResponseOp {
       return;
     }
 
-    std::unique_ptr<rgw::sal::Bucket> bucket;
-    op_ret = driver->load_bucket(this, rgw_bucket(s->bucket_tenant, s->bucket_name),
-                                 &bucket, y);
+    const rgw_bucket src_bucket_id(s->bucket_tenant, s->bucket_name);
+    std::unique_ptr<rgw::sal::Bucket> src_bucket;
+    op_ret = driver->load_bucket(this, src_bucket_id,
+                                 &src_bucket, y);
     if (op_ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to get bucket '" << s->bucket_name << "', ret = " << op_ret << dendl;
+      ldpp_dout(this, 1) << "ERROR: failed to get bucket '" << src_bucket_id << "', ret = " << op_ret << dendl;
       return;
     }
 
-
-    auto& attrs = bucket->get_attrs();
+    auto& attrs = src_bucket->get_attrs();
     if (!configuration.enabled) {
       if (auto iter = attrs.find(RGW_ATTR_BUCKET_LOGGING); iter != attrs.end()) {
         attrs.erase(iter);
       }
     } else {
+      std::string target_bucket_name;
+      std::string target_tenant_name;
+      op_ret = rgw_parse_url_bucket(configuration.target_bucket, s->bucket_tenant, target_tenant_name, target_bucket_name);
+      if (op_ret < 0) {
+        ldpp_dout(this, 1) << "ERROR: failed to parse target bucket '" << configuration.target_bucket << "', ret = " << op_ret << dendl;
+        return;
+      }
+      const rgw_bucket target_bucket_id(target_tenant_name, target_bucket_name);
       std::unique_ptr<rgw::sal::Bucket> target_bucket;
-      op_ret = driver->load_bucket(this, rgw_bucket(s->bucket_tenant, configuration.target_bucket),
+      op_ret = driver->load_bucket(this, target_bucket_id,
                                    &target_bucket, y);
       if (op_ret < 0) {
-        ldpp_dout(this, 1) << "ERROR: failed to get target bucket '" << configuration.target_bucket << "', ret = " << op_ret << dendl;
+        ldpp_dout(this, 1) << "ERROR: failed to get target bucket '" << target_bucket_id << "', ret = " << op_ret << dendl;
         return;
       }
       const auto& target_attrs = target_bucket->get_attrs();
       if (target_attrs.find(RGW_ATTR_BUCKET_LOGGING) != target_attrs.end()) {
         // target bucket must not have logging set on it
-        ldpp_dout(this, 1) << "ERROR: logging target bucket '" << configuration.target_bucket << "', is configured with bucket logging" << dendl;
+        ldpp_dout(this, 1) << "ERROR: logging target bucket '" << target_bucket_id << "', is configured with bucket logging" << dendl;
         op_ret = -EINVAL;
         return;
       }
@@ -196,21 +203,20 @@ class RGWPutBucketLoggingOp : public RGWDefaultResponseOp {
       // if we do, how do we maintain it when bucket logging changes?
     }
     // TODO: use retry_raced_bucket_write from rgw_op.cc
-    op_ret = bucket->merge_and_store_attrs(this, attrs, y);
+    op_ret = src_bucket->merge_and_store_attrs(this, attrs, y);
     if (op_ret < 0) {
       ldpp_dout(this, 1) << "ERROR: failed to set logging attribute '" << RGW_ATTR_BUCKET_LOGGING << "' to bucket '" << 
-        bucket->get_name() << "', ret = " << op_ret << dendl;
+        src_bucket->get_name() << "', ret = " << op_ret << dendl;
       return;
     }
 
     ldpp_dout(this, 20) << "INFO: " << (configuration.enabled ? "wrote" : "removed")
-      << " logging configuration. bucket '" << bucket->get_name() << "'. configuration: " <<
+      << " logging configuration. bucket '" << src_bucket_id << "'. configuration: " <<
       configuration.to_json_str() << dendl;
   }
 };
 
 // Post /<bucket name>/?logging
-// actual configuration is XML encoded in the body of the message
 class RGWPostBucketLoggingOp : public RGWDefaultResponseOp {
   int verify_permission(optional_yield y) override {
     auto [has_s3_existing_tag, has_s3_resource_tag] = rgw_check_policy_condition(this, s, false);
@@ -234,17 +240,18 @@ class RGWPostBucketLoggingOp : public RGWDefaultResponseOp {
       return;
     }
 
-    std::unique_ptr<rgw::sal::Bucket> bucket;
-    op_ret = driver->load_bucket(this, rgw_bucket(s->bucket_tenant, s->bucket_name),
-                                 &bucket, y);
+    const rgw_bucket src_bucket_id(s->bucket_tenant, s->bucket_name);
+    std::unique_ptr<rgw::sal::Bucket> src_bucket;
+    op_ret = driver->load_bucket(this, src_bucket_id,
+                                 &src_bucket, y);
     if (op_ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to get bucket '" << s->bucket_name << "', ret = " << op_ret << dendl;
+      ldpp_dout(this, 1) << "ERROR: failed to get bucket '" << src_bucket_id << "', ret = " << op_ret << dendl;
       return;
     }
-    const auto& bucket_attrs = bucket->get_attrs();
+    const auto& bucket_attrs = src_bucket->get_attrs();
     auto iter = bucket_attrs.find(RGW_ATTR_BUCKET_LOGGING);
     if (iter == bucket_attrs.end()) {
-      ldpp_dout(this, 1) << "WARNING: no logging configured on bucket" << dendl;
+      ldpp_dout(this, 1) << "WARNING: no logging configured on bucket '" << src_bucket_id << "'" << dendl;
       return;
     }
     rgw::bucketlogging::configuration configuration;
@@ -258,24 +265,32 @@ class RGWPostBucketLoggingOp : public RGWDefaultResponseOp {
       return;
     }
 
+    std::string target_bucket_name;
+    std::string target_tenant_name;
+    op_ret = rgw_parse_url_bucket(configuration.target_bucket, s->bucket_tenant, target_tenant_name, target_bucket_name);
+    if (op_ret < 0) {
+      ldpp_dout(this, 1) << "ERROR: failed to parse target bucket '" << configuration.target_bucket << "', ret = " << op_ret << dendl;
+      return;
+    }
+    const rgw_bucket target_bucket_id(target_tenant_name, target_bucket_name);
     std::unique_ptr<rgw::sal::Bucket> target_bucket;
-    op_ret = driver->load_bucket(this, rgw_bucket(s->bucket_tenant, configuration.target_bucket),
+    op_ret = driver->load_bucket(this, target_bucket_id,
                                  &target_bucket, y);
     if (op_ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to get target bucket '" << configuration.target_bucket << "', ret = " << op_ret << dendl;
+      ldpp_dout(this, 1) << "ERROR: failed to get target bucket '" << target_bucket_id << "', ret = " << op_ret << dendl;
       return;
     }
     std::string obj_name;
     RGWObjVersionTracker objv_tracker;
     op_ret = target_bucket->get_logging_object_name(obj_name, configuration.target_prefix, null_yield, this, &objv_tracker);
     if (op_ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to get pending logging object name from target bucket '" << configuration.target_bucket << "'" << dendl;
+      ldpp_dout(this, 1) << "ERROR: failed to get pending logging object name from target bucket '" << target_bucket_id << "'" << dendl;
       return;
     }
     op_ret = rgw::bucketlogging::rollover_logging_object(configuration, target_bucket, obj_name, this, null_yield, true, &objv_tracker);
     if (op_ret < 0) {
       ldpp_dout(this, 1) << "ERROR: failed to flush pending logging object '" << obj_name
-               << "' to target bucket '" << configuration.target_bucket << "'" << dendl;
+               << "' to target bucket '" << target_bucket_id << "'" << dendl;
       return;
     }
     ldpp_dout(this, 20) << "flushed pending logging object '" << obj_name