Merge pull request ceph#55657 from ceph/wip-rgw-meta-topic

rgw: replicate v2 topic/notification metadata

Reviewed-by: Krunal Chheda <[email protected]>
Reviewed-by: Yuval Lifshitz <[email protected]>
Reviewed-by: Casey Bodley <[email protected]>
Reviewed-by: Matt Benjamin <[email protected]>

Author: cbodley

qa/tasks/notification_tests.py

@@ -298,6 +298,9 @@ def task(ctx,config):
                     {
                     'port':endpoint.port,
                     'host':endpoint.dns_name,
+                    'zonegroup':'default',
+                    'cluster':'noname',
+                    'version':'v2'
                     },
                 's3 main':{}
             }

src/rgw/CMakeLists.txt

@@ -193,7 +193,8 @@ set(librgw_common_srcs
   driver/rados/rgw_trim_mdlog.cc
   driver/rados/rgw_user.cc
   driver/rados/rgw_zone.cc
-  driver/rados/sync_fairness.cc)
+  driver/rados/sync_fairness.cc
+  driver/rados/topic.cc)
 
 list(APPEND librgw_common_srcs
   driver/immutable_config/store.cc

src/rgw/driver/rados/rgw_bucket.cc

@@ -1336,17 +1336,6 @@ static int bucket_stats(rgw::sal::Driver* driver,
     }
   }
 
-  // bucket notifications
-  RGWPubSub ps(driver, tenant_name);
-  rgw_pubsub_bucket_topics result;
-  const RGWPubSub::Bucket b(ps, bucket.get());
-  ret = b.get_topics(dpp, result, y);
-  if (ret < 0 && ret != -ENOENT) {
-    cerr << "ERROR: could not get topics: " << cpp_strerror(-ret) << std::endl;
-    return -ret;
-  }
-  result.dump(formatter);
-
   // TODO: bucket CORS
   // TODO: bucket LC
   formatter->close_section();
@@ -2121,6 +2110,92 @@ int RGWMetadataHandlerPut_Bucket::put_post(const DoutPrefixProvider *dpp)
   return ret;
 }
 
+int update_bucket_topic_mappings(const DoutPrefixProvider* dpp,
+                                 RGWBucketCompleteInfo* orig_bci,
+                                 RGWBucketCompleteInfo* current_bci,
+                                 rgw::sal::Driver* driver) {
+  const auto decode_attrs = [](const rgw::sal::Attrs& attrs,
+                               rgw_pubsub_bucket_topics& bucket_topics) -> int {
+    auto iter = attrs.find(RGW_ATTR_BUCKET_NOTIFICATION);
+    if (iter == attrs.end()) {
+      return 0;
+    }
+    try {
+      const auto& bl = iter->second;
+      auto biter = bl.cbegin();
+      bucket_topics.decode(biter);
+    } catch (buffer::error& err) {
+      return -EIO;
+    }
+    return 0;
+  };
+  std::string bucket_name;
+  std::string bucket_tenant;
+  rgw_pubsub_bucket_topics old_bucket_topics;
+  if (orig_bci) {
+    auto ret = decode_attrs(orig_bci->attrs, old_bucket_topics);
+    if (ret < 0) {
+      ldpp_dout(dpp, 1)
+          << "ERROR: failed to decode OLD bucket topics for bucket: "
+          << orig_bci->info.bucket.name << dendl;
+      return ret;
+    }
+    bucket_name = orig_bci->info.bucket.name;
+    bucket_tenant = orig_bci->info.bucket.tenant;
+  }
+  rgw_pubsub_bucket_topics current_bucket_topics;
+  if (current_bci) {
+    auto ret = decode_attrs(current_bci->attrs, current_bucket_topics);
+    if (ret < 0) {
+      ldpp_dout(dpp, 1)
+          << "ERROR: failed to decode current bucket topics for bucket: "
+          << current_bci->info.bucket.name << dendl;
+      return ret;
+    }
+    bucket_name = current_bci->info.bucket.name;
+    bucket_tenant = current_bci->info.bucket.tenant;
+  }
+  // fetch the list of subscribed topics stored inside old_bucket attrs.
+  std::unordered_map<std::string, rgw_pubsub_topic> old_topics;
+  for (const auto& [_, topic_filter] : old_bucket_topics.topics) {
+    old_topics[topic_filter.topic.name] = topic_filter.topic;
+  }
+  // fetch the list of subscribed topics stored inside current_bucket attrs.
+  std::unordered_map<std::string, rgw_pubsub_topic> current_topics;
+  for (const auto& [_, topic_filter] : current_bucket_topics.topics) {
+    current_topics[topic_filter.topic.name] = topic_filter.topic;
+  }
+  // traverse thru old topics and check if they are not in current, then delete
+  // the mapping, if present in both current and old then delete from current
+  // set as we do not need to update those mapping.
+  int ret = 0;
+  for (const auto& [topic_name, topic] : old_topics) {
+    auto it = current_topics.find(topic_name);
+    if (it == current_topics.end()) {
+      const auto op_ret = driver->update_bucket_topic_mapping(
+          topic, rgw_make_bucket_entry_name(bucket_tenant, bucket_name),
+          /*add_mapping=*/false, null_yield, dpp);
+      if (op_ret < 0) {
+        ret = op_ret;
+      }
+    } else {
+      // already that attr is present, so do not update the mapping.
+      current_topics.erase(it);
+    }
+  }
+  // traverse thru current topics and check if they are any present, then add
+  // the mapping.
+  for (const auto& [topic_name, topic] : current_topics) {
+    const auto op_ret = driver->update_bucket_topic_mapping(
+        topic, rgw_make_bucket_entry_name(bucket_tenant, bucket_name),
+        /*add_mapping=*/true, null_yield, dpp);
+    if (op_ret < 0) {
+      ret = op_ret;
+    }
+  }
+  return ret;
+}
+
 static void get_md5_digest(const RGWBucketEntryPoint *be, string& md5_digest) {
 
    char md5[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 1];
@@ -2432,7 +2507,14 @@ class RGWBucketInstanceMetadataHandler : public RGWBucketInstanceMetadataHandler
     if (ret < 0 && ret != -ENOENT)
       return ret;
 
-    return svc.bucket->remove_bucket_instance_info(ctx, entry, bci.info, &bci.info.objv_tracker, y, dpp);
+    ret = svc.bucket->remove_bucket_instance_info(
+        ctx, entry, bci.info, &bci.info.objv_tracker, y, dpp);
+    if (ret < 0)
+      return ret;
+    ret = update_bucket_topic_mappings(dpp, &bci, /*current_bci=*/nullptr,
+                                       driver);
+    // update_bucket_topic_mapping error is swallowed.
+    return 0;
   }
 
   int call(std::function<int(RGWSI_Bucket_BI_Ctx& ctx)> f) {
@@ -2637,6 +2719,21 @@ int RGWMetadataHandlerPut_BucketInstance::put_post(const DoutPrefixProvider *dpp
     }
   } /* update lc */
 
+  /* update bucket topic mapping */
+  {
+    auto* orig_obj = static_cast<RGWBucketInstanceMetadataObject*>(old_obj);
+    auto* orig_bci = (orig_obj ? &orig_obj->get_bci() : nullptr);
+    ret = update_bucket_topic_mappings(dpp, orig_bci, &bci, bihandler->driver);
+    if (ret < 0) {
+      ldpp_dout(dpp, 0) << __func__
+                        << " failed to apply bucket topic mapping for "
+                        << bci.info.bucket.name << dendl;
+      return ret;
+    }
+    ldpp_dout(dpp, 20) << __func__
+                       << " successfully applied bucket topic mapping for "
+                       << bci.info.bucket.name << dendl;
+  }
   return STATUS_APPLIED;
 }
 

src/rgw/driver/rados/rgw_cr_rados.cc

@@ -809,7 +809,7 @@ int RGWAsyncFetchRemoteObj::_send_request(const DoutPrefixProvider *dpp)
                   req_id, null_yield);
 
         auto notify_res = static_cast<rgw::sal::RadosNotification*>(notify.get())->get_reservation();
-        int ret = rgw::notify::publish_reserve(dpp, rgw::notify::ObjectSyncedCreate, notify_res, &obj_tags);
+        int ret = rgw::notify::publish_reserve(dpp, *store->svc()->site, rgw::notify::ObjectSyncedCreate, notify_res, &obj_tags);
         if (ret < 0) {
           ldpp_dout(dpp, 1) << "ERROR: reserving notification failed, with error: " << ret << dendl;
           // no need to return, the sync already happened

src/rgw/driver/rados/rgw_metadata_lister.h

@@ -0,0 +1,73 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright contributors to the Ceph project
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include <algorithm>
+#include <iterator>
+#include <list>
+#include <string>
+#include <vector>
+#include "services/svc_sys_obj.h"
+
+class DoutPrefixProvider;
+
+class RGWMetadataLister {
+  RGWSI_SysObj::Pool pool;
+  RGWSI_SysObj::Pool::Op listing;
+
+  virtual void filter_transform(std::vector<std::string>& oids,
+                                std::list<std::string>& keys) {
+    // use all oids as keys
+    std::move(oids.begin(), oids.end(), std::back_inserter(keys));
+  }
+
+ public:
+  explicit RGWMetadataLister(RGWSI_SysObj::Pool pool)
+    : pool(pool), listing(this->pool) {}
+  virtual ~RGWMetadataLister() {}
+
+  int init(const DoutPrefixProvider* dpp,
+           const std::string& marker,
+           const std::string& prefix)
+  {
+    return listing.init(dpp, marker, prefix);
+  }
+
+  int get_next(const DoutPrefixProvider* dpp, int max,
+               std::list<std::string>& keys, bool* truncated)
+  {
+    std::vector<std::string> oids;
+    int r = listing.get_next(dpp, max, &oids, truncated);
+    if (r == -ENOENT) {
+      if (truncated) {
+        *truncated = false;
+      }
+      return 0;
+    }
+    if (r < 0) {
+      return r;
+    }
+    filter_transform(oids, keys);
+    return 0;
+  }
+
+  std::string get_marker()
+  {
+    std::string marker;
+    listing.get_marker(&marker);
+    return marker;
+  }
+};

src/rgw/driver/rados/rgw_notify.cc

@@ -14,7 +14,9 @@
 #include "rgw_sal_rados.h"
 #include "rgw_pubsub.h"
 #include "rgw_pubsub_push.h"
+#include "rgw_zone_features.h"
 #include "rgw_perf_counters.h"
+#include "services/svc_zone.h"
 #include "common/dout.h"
 #include <chrono>
 
@@ -119,6 +121,7 @@ class Manager : public DoutPrefixProvider {
   const uint32_t stale_reservations_period_s;
   const uint32_t reservations_cleanup_period_s;
   queues_persistency_tracker topics_persistency_tracker;
+  const SiteConfig& site;
 public:
   rgw::sal::RadosStore& rados_store;
 
@@ -489,10 +492,11 @@ class Manager : public DoutPrefixProvider {
           std::string tenant_name;
           // TODO: extract tenant name from queue_name once it is fixed
           uint64_t size_to_migrate = 0;
-          RGWPubSub ps(&rados_store, tenant_name);
+          RGWPubSub ps(&rados_store, tenant_name, site);
 
           rgw_pubsub_topic topic;
-          auto ret_of_get_topic = ps.get_topic(this, queue_name, topic, optional_yield(io_context, yield));
+          auto ret_of_get_topic = ps.get_topic(this, queue_name, topic,
+                           optional_yield(io_context, yield), nullptr);
           if (ret_of_get_topic < 0) {
             // we can't migrate entries without topic info
             ldpp_dout(this, 1) << "ERROR: failed to fetch topic: " << queue_name << " error: "
@@ -666,7 +670,8 @@ class Manager : public DoutPrefixProvider {
   Manager(CephContext* _cct, uint32_t _max_queue_size, uint32_t _queues_update_period_ms, 
           uint32_t _queues_update_retry_ms, uint32_t _queue_idle_sleep_us, u_int32_t failover_time_ms, 
           uint32_t _stale_reservations_period_s, uint32_t _reservations_cleanup_period_s,
-          uint32_t _worker_count, rgw::sal::RadosStore* store) :
+          uint32_t _worker_count, rgw::sal::RadosStore* store,
+          const SiteConfig& site) :
     max_queue_size(_max_queue_size),
     queues_update_period_ms(_queues_update_period_ms),
     queues_update_retry_ms(_queues_update_retry_ms),
@@ -678,6 +683,7 @@ class Manager : public DoutPrefixProvider {
     worker_count(_worker_count),
     stale_reservations_period_s(_stale_reservations_period_s),
     reservations_cleanup_period_s(_reservations_cleanup_period_s),
+    site(site),
     rados_store(*store)
     {
       spawn::spawn(io_context, [this](spawn::yield_context yield) {
@@ -750,7 +756,8 @@ constexpr uint32_t WORKER_COUNT = 1;                 // 1 worker thread
 constexpr uint32_t STALE_RESERVATIONS_PERIOD_S = 120;   // cleanup reservations that are more than 2 minutes old
 constexpr uint32_t RESERVATIONS_CLEANUP_PERIOD_S = 30; // reservation cleanup every 30 seconds
 
-bool init(CephContext* cct, rgw::sal::RadosStore* store, const DoutPrefixProvider *dpp) {
+bool init(CephContext* cct, rgw::sal::RadosStore* store,
+          const SiteConfig& site, const DoutPrefixProvider *dpp) {
   if (s_manager) {
     return false;
   }
@@ -760,7 +767,7 @@ bool init(CephContext* cct, rgw::sal::RadosStore* store, const DoutPrefixProvide
       IDLE_TIMEOUT_USEC, FAILOVER_TIME_MSEC, 
       STALE_RESERVATIONS_PERIOD_S, RESERVATIONS_CLEANUP_PERIOD_S,
       WORKER_COUNT,
-      store);
+      store, site);
   return true;
 }
 
@@ -975,17 +982,40 @@ static inline bool notification_match(reservation_t& res,
 }
 
   int publish_reserve(const DoutPrefixProvider* dpp,
+		      const SiteConfig& site,
 		      EventType event_type,
 		      reservation_t& res,
 		      const RGWObjTags* req_tags)
 {
-  const RGWPubSub ps(res.store, res.user_tenant);
-  const RGWPubSub::Bucket ps_bucket(ps, res.bucket);
   rgw_pubsub_bucket_topics bucket_topics;
-  auto rc = ps_bucket.get_topics(res.dpp, bucket_topics, res.yield);
-  if (rc < 0) {
-    // failed to fetch bucket topics
-    return rc;
+  if (all_zonegroups_support(site, zone_features::notification_v2) &&
+      res.store->stat_topics_v1(res.user_tenant, res.yield, res.dpp) == -ENOENT) {
+    auto ret = 0;
+    if (!res.s) {
+      //  for non S3-request caller (e.g., lifecycle, ObjectSync), bucket attrs
+      //  are not loaded, so force to reload the bucket, that reloads the attr.
+      // for non S3-request caller, res.s is nullptr
+      ret = res.bucket->load_bucket(dpp, res.yield);
+      if (ret < 0) {
+        ldpp_dout(dpp, 1)
+            << "ERROR: failed to reload bucket: '" << res.bucket->get_name()
+            << "' to get bucket notification attrs with error ret= " << ret
+            << dendl;
+        return ret;
+      }
+    }
+    ret = get_bucket_notifications(dpp, res.bucket, bucket_topics);
+    if (ret < 0) {
+      return ret;
+    }
+  } else {
+    const RGWPubSub ps(res.store, res.user_tenant, site);
+    const RGWPubSub::Bucket ps_bucket(ps, res.bucket);
+    auto rc = ps_bucket.get_topics(res.dpp, bucket_topics, res.yield);
+    if (rc < 0) {
+      // failed to fetch bucket topics
+      return rc;
+    }
   }
   for (const auto& bucket_topic : bucket_topics.topics) {
     const rgw_pubsub_topic_filter& topic_filter = bucket_topic.second;
@@ -1026,7 +1056,29 @@ static inline bool notification_match(reservation_t& res,
         return ret;
       }
     }
-    res.topics.emplace_back(topic_filter.s3_id, topic_cfg, res_id);
+    // load the topic,if there is change in topic config while it's stored in
+    // notification.
+    rgw_pubsub_topic result;
+    const RGWPubSub ps(res.store, res.user_tenant, site);
+    auto ret = ps.get_topic(res.dpp, topic_cfg.name, result, res.yield, nullptr);
+    if (ret < 0) {
+      ldpp_dout(res.dpp, 1)
+          << "INFO: failed to load topic: " << topic_cfg.name
+          << ". error: " << ret
+          << " while reserving persistent notification event" << dendl;
+      if (ret == -ENOENT) {
+        // either the topic is deleted but the corresponding notification still
+        // exist or in v2 mode the notification could have synced first but
+        // topic is not synced yet.
+        return 0;
+      }
+      ldpp_dout(res.dpp, 1)
+          << "WARN: Using the stored topic from bucket notification struct."
+          << dendl;
+      res.topics.emplace_back(topic_filter.s3_id, topic_cfg, res_id);
+    } else {
+      res.topics.emplace_back(topic_filter.s3_id, result, res_id);
+    }
   }
   return 0;
 }