Merge pull request ceph#57249 from cbodley/wip-65668

cbodley · web-flow · commit 744b88ce9aba · 2024-05-21T18:31:24.000+01:00
rgw/notify: decouple add/remove_persistent_topic() from Manager

Reviewed-by: Krunal Chheda &lt;kchheda3@bloomberg.net&gt;
Reviewed-by: Yuval Lifshitz &lt;ylifshit@redhat.com&gt;
diff --git a/src/rgw/driver/rados/rgw_notify.cc b/src/rgw/driver/rados/rgw_notify.cc
@@ -124,7 +124,6 @@ void publish_commit_completion(rados_completion_t completion, void *arg) {
 
 class Manager : public DoutPrefixProvider {
   bool shutdown = false;
-  const size_t max_queue_size;
   const uint32_t queues_update_period_ms;
   const uint32_t queues_update_retry_ms;
   const uint32_t queue_idle_sleep_us;
@@ -764,12 +763,11 @@ class Manager : public DoutPrefixProvider {
   }
 
   // ctor: start all threads
-  Manager(CephContext* _cct, uint32_t _max_queue_size, uint32_t _queues_update_period_ms, 
+  Manager(CephContext* _cct, 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,
           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),
     queue_idle_sleep_us(_queue_idle_sleep_us),
@@ -783,39 +781,6 @@ class Manager : public DoutPrefixProvider {
     site(site),
     rados_store(*store)
     {}
-
-  int add_persistent_topic(const std::string& topic_queue, optional_yield y) {
-    if (topic_queue == Q_LIST_OBJECT_NAME) {
-      ldpp_dout(this, 1) << "ERROR: topic name cannot be: " << Q_LIST_OBJECT_NAME << " (conflict with queue list object name)" << dendl;
-      return -EINVAL;
-    }
-    librados::ObjectWriteOperation op;
-    op.create(true);
-    cls_2pc_queue_init(op, topic_queue, max_queue_size);
-    auto& rados_ioctx = rados_store.getRados()->get_notif_pool_ctx();
-    auto ret = rgw_rados_operate(this, rados_ioctx, topic_queue, &op, y);
-    if (ret == -EEXIST) {
-      // queue already exists - nothing to do
-      ldpp_dout(this, 20) << "INFO: queue for topic: " << topic_queue << " already exists. nothing to do" << dendl;
-      return 0;
-    }
-    if (ret < 0) {
-      // failed to create queue
-      ldpp_dout(this, 1) << "ERROR: failed to create queue for topic: " << topic_queue << ". error: " << ret << dendl;
-      return ret;
-    }
-   
-    bufferlist empty_bl;
-    std::map<std::string, bufferlist> new_topic{{topic_queue, empty_bl}};
-    op.omap_set(new_topic);
-    ret = rgw_rados_operate(this, rados_ioctx, Q_LIST_OBJECT_NAME, &op, y);
-    if (ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to add queue: " << topic_queue << " to queue list. error: " << ret << dendl;
-      return ret;
-    } 
-    ldpp_dout(this, 20) << "INFO: queue: " << topic_queue << " added to queue list"  << dendl;
-    return 0;
-  }
 };
 
 std::unique_ptr<Manager> s_manager;
@@ -839,7 +804,7 @@ bool init(const DoutPrefixProvider* dpp, rgw::sal::RadosStore* store,
     return false;
   }
   // TODO: take conf from CephContext
-  s_manager = std::make_unique<Manager>(dpp->get_cct(), MAX_QUEUE_SIZE, 
+  s_manager = std::make_unique<Manager>(dpp->get_cct(),
       Q_LIST_UPDATE_MSEC, Q_LIST_RETRY_MSEC, 
       IDLE_TIMEOUT_USEC, FAILOVER_TIME_MSEC, 
       STALE_RESERVATIONS_PERIOD_S, RESERVATIONS_CLEANUP_PERIOD_S,
@@ -856,11 +821,38 @@ void shutdown() {
   s_manager.reset();
 }
 
-int add_persistent_topic(const std::string& topic_name, optional_yield y) {
-  if (!s_manager) {
-    return -EAGAIN;
+int add_persistent_topic(const DoutPrefixProvider* dpp, librados::IoCtx& rados_ioctx,
+                         const std::string& topic_queue, optional_yield y)
+{
+  if (topic_queue == Q_LIST_OBJECT_NAME) {
+    ldpp_dout(dpp, 1) << "ERROR: topic name cannot be: " << Q_LIST_OBJECT_NAME << " (conflict with queue list object name)" << dendl;
+    return -EINVAL;
   }
-  return s_manager->add_persistent_topic(topic_name, y);
+  librados::ObjectWriteOperation op;
+  op.create(true);
+  cls_2pc_queue_init(op, topic_queue, MAX_QUEUE_SIZE);
+  auto ret = rgw_rados_operate(dpp, rados_ioctx, topic_queue, &op, y);
+  if (ret == -EEXIST) {
+    // queue already exists - nothing to do
+    ldpp_dout(dpp, 20) << "INFO: queue for topic: " << topic_queue << " already exists. nothing to do" << dendl;
+    return 0;
+  }
+  if (ret < 0) {
+    // failed to create queue
+    ldpp_dout(dpp, 1) << "ERROR: failed to create queue for topic: " << topic_queue << ". error: " << ret << dendl;
+    return ret;
+  }
+
+  bufferlist empty_bl;
+  std::map<std::string, bufferlist> new_topic{{topic_queue, empty_bl}};
+  op.omap_set(new_topic);
+  ret = rgw_rados_operate(dpp, rados_ioctx, Q_LIST_OBJECT_NAME, &op, y);
+  if (ret < 0) {
+    ldpp_dout(dpp, 1) << "ERROR: failed to add queue: " << topic_queue << " to queue list. error: " << ret << dendl;
+    return ret;
+  }
+  ldpp_dout(dpp, 20) << "INFO: queue: " << topic_queue << " added to queue list"  << dendl;
+  return 0;
 }
 
 int remove_persistent_topic(const DoutPrefixProvider* dpp, librados::IoCtx& rados_ioctx, const std::string& topic_queue, optional_yield y) {
@@ -889,13 +881,6 @@ int remove_persistent_topic(const DoutPrefixProvider* dpp, librados::IoCtx& rado
   return 0;
 }
 
-int remove_persistent_topic(const std::string& topic_queue, optional_yield y) {
-  if (!s_manager) {
-    return -EAGAIN;
-  }
-  return remove_persistent_topic(s_manager.get(), s_manager->rados_store.getRados()->get_notif_pool_ctx(), topic_queue, y);
-}
-
 rgw::sal::Object* get_object_with_attributes(
   const reservation_t& res, rgw::sal::Object* obj) {
   // in case of copy obj, the tags and metadata are taken from source
diff --git a/src/rgw/driver/rados/rgw_notify.h b/src/rgw/driver/rados/rgw_notify.h
@@ -34,13 +34,10 @@ void shutdown();
 
 // create persistent delivery queue for a topic (endpoint)
 // this operation also add a topic queue to the common (to all RGWs) list of all topics
-int add_persistent_topic(const std::string& topic_queue, optional_yield y);
+int add_persistent_topic(const DoutPrefixProvider* dpp, librados::IoCtx& rados_ioctx, const std::string& topic_queue, optional_yield y);
 
 // remove persistent delivery queue for a topic (endpoint)
 // this operation also remove the topic queue from the common (to all RGWs) list of all topics
-int remove_persistent_topic(const std::string& topic_queue, optional_yield y);
-
-// same as the above, expect you need to provide the IoCtx, the above uses rgw::notify::Manager::rados_ioctx
 int remove_persistent_topic(const DoutPrefixProvider* dpp, librados::IoCtx& rados_ioctx, const std::string& topic_queue, optional_yield y);
 
 // struct holding reservation information
diff --git a/src/rgw/driver/rados/rgw_sal_rados.cc b/src/rgw/driver/rados/rgw_sal_rados.cc
@@ -1726,6 +1726,22 @@ int RadosStore::list_account_topics(const DoutPrefixProvider* dpp,
                                 listing.topics, listing.next_marker);
 }
 
+int RadosStore::add_persistent_topic(const DoutPrefixProvider* dpp,
+                                     optional_yield y,
+                                     const std::string& topic_queue)
+{
+  return rgw::notify::add_persistent_topic(
+      dpp, getRados()->get_notif_pool_ctx(), topic_queue, y);
+}
+
+int RadosStore::remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                        optional_yield y,
+                                        const std::string& topic_queue)
+{
+  return rgw::notify::remove_persistent_topic(
+      dpp, getRados()->get_notif_pool_ctx(), topic_queue, y);
+}
+
 int RadosStore::remove_bucket_mapping_from_topics(
     const rgw_pubsub_bucket_topics& bucket_topics,
     const std::string& bucket_key,
diff --git a/src/rgw/driver/rados/rgw_sal_rados.h b/src/rgw/driver/rados/rgw_sal_rados.h
@@ -319,6 +319,12 @@ class RadosStore : public StoreDriver {
                             std::string_view marker,
                             uint32_t max_items,
                             TopicList& listing) override;
+    int add_persistent_topic(const DoutPrefixProvider* dpp,
+                             optional_yield y,
+                             const std::string& topic_queue) override;
+    int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                optional_yield y,
+                                const std::string& topic_queue) override;
     int update_bucket_topic_mapping(const rgw_pubsub_topic& topic,
                                     const std::string& bucket_key,
                                     bool add_mapping,
diff --git a/src/rgw/driver/rados/topic.cc b/src/rgw/driver/rados/topic.cc
@@ -354,7 +354,11 @@ class MetadataHandler : public RGWMetadataHandler {
     }
     if (!info.dest.push_endpoint.empty() && info.dest.persistent &&
         !info.dest.persistent_queue.empty()) {
-      r = rgw::notify::add_persistent_topic(info.dest.persistent_queue, y);
+      librados::IoCtx ioctx;
+      r = rgw_init_ioctx(dpp, &rados, zone.notif_pool, ioctx, true, false);
+      if (r >= 0) {
+        r = rgw::notify::add_persistent_topic(dpp, ioctx, info.dest.persistent_queue, y);
+      }
       if (r < 0) {
         ldpp_dout(dpp, 1) << "ERROR: failed to create queue for persistent topic "
             << info.dest.persistent_queue << " with: " << cpp_strerror(r) << dendl;
@@ -388,7 +392,11 @@ class MetadataHandler : public RGWMetadataHandler {
     if (!dest.push_endpoint.empty() && dest.persistent &&
         !dest.persistent_queue.empty()) {
       // delete persistent topic queue
-      r = rgw::notify::remove_persistent_topic(dest.persistent_queue, y);
+      librados::IoCtx ioctx;
+      r = rgw_init_ioctx(dpp, &rados, zone.notif_pool, ioctx, true, false);
+      if (r >= 0) {
+        r = rgw::notify::remove_persistent_topic(dpp, ioctx, dest.persistent_queue, y);
+      }
       if (r < 0 && r != -ENOENT) {
         ldpp_dout(dpp, 1) << "Failed to delete queue for persistent topic: "
                           << name << " with error: " << r << dendl;
diff --git a/src/rgw/rgw_pubsub.cc b/src/rgw/rgw_pubsub.cc
@@ -1091,7 +1091,7 @@ int RGWPubSub::remove_topic_v2(const DoutPrefixProvider* dpp,
   const rgw_pubsub_dest& dest = topic.dest;
   if (!dest.push_endpoint.empty() && dest.persistent &&
       !dest.persistent_queue.empty()) {
-    ret = rgw::notify::remove_persistent_topic(dest.persistent_queue, y);
+    ret = driver->remove_persistent_topic(dpp, y, dest.persistent_queue);
     if (ret < 0 && ret != -ENOENT) {
       ldpp_dout(dpp, 1) << "WARNING: failed to remove queue for "
           "persistent topic: " << cpp_strerror(ret) << dendl;
@@ -1138,7 +1138,7 @@ int RGWPubSub::remove_topic(const DoutPrefixProvider *dpp, const std::string& na
 
   if (!dest.push_endpoint.empty() && dest.persistent &&
       !dest.persistent_queue.empty()) {
-    ret = rgw::notify::remove_persistent_topic(dest.persistent_queue, y);
+    ret = driver->remove_persistent_topic(dpp, y, dest.persistent_queue);
     if (ret < 0 && ret != -ENOENT) {
       ldpp_dout(dpp, 1) << "WARNING: failed to remove queue for "
           "persistent topic: " << cpp_strerror(ret) << dendl;
diff --git a/src/rgw/rgw_rest_pubsub.cc b/src/rgw/rgw_rest_pubsub.cc
@@ -415,7 +415,7 @@ void RGWPSCreateTopicOp::execute(optional_yield y) {
     dest.persistent_queue = string_cat_reserve(
         get_account_or_tenant(s->owner.id), ":", topic_name);
 
-    op_ret = rgw::notify::add_persistent_topic(dest.persistent_queue, s->yield);
+    op_ret = driver->add_persistent_topic(this, y, dest.persistent_queue);
     if (op_ret < 0) {
       ldpp_dout(this, 1) << "CreateTopic Action failed to create queue for "
                             "persistent topics. error:"
@@ -874,7 +874,7 @@ void RGWPSSetTopicAttributesOp::execute(optional_yield y) {
     dest.persistent_queue = string_cat_reserve(
         get_account_or_tenant(s->owner.id), ":", topic_name);
 
-    op_ret = rgw::notify::add_persistent_topic(dest.persistent_queue, s->yield);
+    op_ret = driver->add_persistent_topic(this, y, dest.persistent_queue);
     if (op_ret < 0) {
       ldpp_dout(this, 4)
           << "SetTopicAttributes Action failed to create queue for "
@@ -884,7 +884,7 @@ void RGWPSSetTopicAttributesOp::execute(optional_yield y) {
     }
   } else if (already_persistent) {
     // changing the persistent topic to non-persistent.
-    op_ret = rgw::notify::remove_persistent_topic(result.dest.persistent_queue, s->yield);
+    op_ret = driver->remove_persistent_topic(this, y, result.dest.persistent_queue);
     if (op_ret != -ENOENT && op_ret < 0) {
       ldpp_dout(this, 4) << "SetTopicAttributes Action failed to remove queue "
                             "for persistent topics. error:"
diff --git a/src/rgw/rgw_sal.h b/src/rgw/rgw_sal.h
@@ -542,6 +542,16 @@ class Driver {
                                     std::string_view marker,
                                     uint32_t max_items,
                                     TopicList& listing) = 0;
+
+    // TODO: backends should manage persistent topic queues internally on
+    // write_topic_v2()/remove_topic_v2()
+    virtual int add_persistent_topic(const DoutPrefixProvider* dpp,
+                                     optional_yield y,
+                                     const std::string& topic_queue) = 0;
+    virtual int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                        optional_yield y,
+                                        const std::string& topic_queue) = 0;
+
     /** Update the bucket-topic mapping in the store, if |add_mapping|=true then
      * adding the |bucket_key| |topic| mapping to store, else delete the
      * |bucket_key| |topic| mapping from the store.  The |bucket_key| is
diff --git a/src/rgw/rgw_sal_dbstore.cc b/src/rgw/rgw_sal_dbstore.cc
@@ -1912,6 +1912,20 @@ namespace rgw::sal {
     return -ENOTSUP;
   }
 
+  int DBStore::add_persistent_topic(const DoutPrefixProvider* dpp,
+                                    optional_yield y,
+                                    const std::string& topic_queue)
+  {
+    return -ENOTSUP;
+  }
+
+  int DBStore::remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                       optional_yield y,
+                                       const std::string& topic_queue)
+  {
+    return -ENOTSUP;
+  }
+
   RGWLC* DBStore::get_rgwlc(void) {
     return lc;
   }
diff --git a/src/rgw/rgw_sal_dbstore.h b/src/rgw/rgw_sal_dbstore.h
@@ -887,6 +887,13 @@ class DBNotification : public StoreNotification {
                           uint32_t max_items,
                           TopicList& listing) override;
 
+      int add_persistent_topic(const DoutPrefixProvider* dpp,
+                               optional_yield y,
+                               const std::string& topic_queue) override;
+      int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                  optional_yield y,
+                                  const std::string& topic_queue) override;
+
       virtual RGWLC* get_rgwlc(void) override;
       virtual RGWCoroutinesManagerRegistry* get_cr_registry() override { return NULL; }
       virtual int log_usage(const DoutPrefixProvider *dpp, std::map<rgw_user_bucket, RGWUsageBatch>& usage_info, optional_yield y) override;
diff --git a/src/rgw/rgw_sal_filter.cc b/src/rgw/rgw_sal_filter.cc
@@ -463,6 +463,20 @@ std::unique_ptr<Notification> FilterDriver::get_notification(
   return std::make_unique<FilterNotification>(std::move(n));
 }
 
+int FilterDriver::add_persistent_topic(const DoutPrefixProvider* dpp,
+                                       optional_yield y,
+                                       const std::string& topic_queue)
+{
+  return next->add_persistent_topic(dpp, y, topic_queue);
+}
+
+int FilterDriver::remove_persistent_topic(const DoutPrefixProvider* dpp,
+                                          optional_yield y,
+                                          const std::string& topic_queue)
+{
+  return next->remove_persistent_topic(dpp, y, topic_queue);
+}
+
 RGWLC* FilterDriver::get_rgwlc()
 {
   return next->get_rgwlc();
diff --git a/src/rgw/rgw_sal_filter.h b/src/rgw/rgw_sal_filter.h
@@ -353,6 +353,12 @@ class FilterDriver : public Driver {
     return next->list_account_topics(dpp, y, account_id, marker,
                                      max_items, listing);
   }
+  int add_persistent_topic(const DoutPrefixProvider* dpp,
+                           optional_yield y,
+                           const std::string& topic_queue) override;
+  int remove_persistent_topic(const DoutPrefixProvider* dpp,
+                              optional_yield y,
+                              const std::string& topic_queue) override;
   int update_bucket_topic_mapping(const rgw_pubsub_topic& topic,
                                   const std::string& bucket_key,
                                   bool add_mapping,
diff --git a/src/test/rgw/bucket_notification/test_bn.py b/src/test/rgw/bucket_notification/test_bn.py
@@ -656,7 +656,7 @@ def test_ps_s3_topic_admin_on_master():
                  'arn:aws:sns:' + zonegroup + ':' + tenant + ':' + topic_name + '_1')
 
     endpoint_address = 'http://127.0.0.1:9001'
-    endpoint_args = 'push-endpoint='+endpoint_address
+    endpoint_args = 'push-endpoint='+endpoint_address+'&persistent=true'
     topic_conf2 = PSTopicS3(conn, topic_name+'_2', zonegroup, endpoint_args=endpoint_args)
     topic_arn2 = topic_conf2.set_config()
     assert_equal(topic_arn2,
