Merge pull request #53514 from yuvalif/wip-61535-reef

reef: RGW:notifications: persistent topics are not deleted via radosgw-admin

Reviewed-by: Shilpa Jagannath <smanjara@redhat.com>

src/rgw/driver/rados/rgw_notify.cc

@@ -52,14 +52,15 @@ auto make_stack_allocator() {
   return boost::context::protected_fixedsize_stack{128*1024};
 }
 
+const std::string Q_LIST_OBJECT_NAME = "queues_list_object";
+
 class Manager : public DoutPrefixProvider {
   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;
   const utime_t failover_time;
   CephContext* const cct;
-  librados::IoCtx& rados_ioctx;
   static constexpr auto COOKIE_LEN = 16;
   const std::string lock_cookie;
   boost::asio::io_context io_context;
@@ -68,8 +69,9 @@ class Manager : public DoutPrefixProvider {
   std::vector<std::thread> workers;
   const uint32_t stale_reservations_period_s;
   const uint32_t reservations_cleanup_period_s;
-
-  const std::string Q_LIST_OBJECT_NAME = "queues_list_object";
+public:
+  librados::IoCtx& rados_ioctx;
+private:
 
   CephContext *get_cct() const override { return cct; }
   unsigned get_subsys() const override { return dout_subsys; }
@@ -481,12 +483,12 @@ class Manager : public DoutPrefixProvider {
     queue_idle_sleep_us(_queue_idle_sleep_us),
     failover_time(std::chrono::milliseconds(failover_time_ms)),
     cct(_cct),
-    rados_ioctx(store->getRados()->get_notif_pool_ctx()),
     lock_cookie(gen_rand_alphanumeric(cct, COOKIE_LEN)),
     work_guard(boost::asio::make_work_guard(io_context)),
     worker_count(_worker_count),
     stale_reservations_period_s(_stale_reservations_period_s),
-    reservations_cleanup_period_s(_reservations_cleanup_period_s)
+    reservations_cleanup_period_s(_reservations_cleanup_period_s),
+    rados_ioctx(store->getRados()->get_notif_pool_ctx())
     {
       spawn::spawn(io_context, [this] (yield_context yield) {
             process_queues(yield);
@@ -541,32 +543,6 @@ class Manager : public DoutPrefixProvider {
     ldpp_dout(this, 20) << "INFO: queue: " << topic_name << " added to queue list"  << dendl;
     return 0;
   }
-
-  int remove_persistent_topic(const std::string& topic_name, optional_yield y) {
-    librados::ObjectWriteOperation op;
-    op.remove();
-    auto ret = rgw_rados_operate(this, rados_ioctx, topic_name, &op, y);
-    if (ret == -ENOENT) {
-      // queue already removed - nothing to do
-      ldpp_dout(this, 20) << "INFO: queue for topic: " << topic_name << " already removed. nothing to do" << dendl;
-      return 0;
-    }
-    if (ret < 0) {
-      // failed to remove queue
-      ldpp_dout(this, 1) << "ERROR: failed to remove queue for topic: " << topic_name << ". error: " << ret << dendl;
-      return ret;
-    }
-
-    std::set<std::string> topic_to_remove{{topic_name}};
-    op.omap_rm_keys(topic_to_remove);
-    ret = rgw_rados_operate(this, rados_ioctx, Q_LIST_OBJECT_NAME, &op, y);
-    if (ret < 0) {
-      ldpp_dout(this, 1) << "ERROR: failed to remove queue: " << topic_name << " from queue list. error: " << ret << dendl;
-      return ret;
-    } 
-    ldpp_dout(this, 20) << "INFO: queue: " << topic_name << " removed from queue list"  << dendl;
-    return 0;
-  }
 };
 
 // singleton manager
@@ -609,11 +585,37 @@ int add_persistent_topic(const std::string& topic_name, optional_yield y) {
   return s_manager->add_persistent_topic(topic_name, y);
 }
 
+int remove_persistent_topic(const DoutPrefixProvider* dpp, librados::IoCtx& rados_ioctx, const std::string& topic_name, optional_yield y) {
+  librados::ObjectWriteOperation op;
+  op.remove();
+  auto ret = rgw_rados_operate(dpp, rados_ioctx, topic_name, &op, y);
+  if (ret == -ENOENT) {
+    // queue already removed - nothing to do
+    ldpp_dout(dpp, 20) << "INFO: queue for topic: " << topic_name << " already removed. nothing to do" << dendl;
+    return 0;
+  }
+  if (ret < 0) {
+    // failed to remove queue
+    ldpp_dout(dpp, 1) << "ERROR: failed to remove queue for topic: " << topic_name << ". error: " << ret << dendl;
+    return ret;
+  }
+
+  std::set<std::string> topic_to_remove{{topic_name}};
+  op.omap_rm_keys(topic_to_remove);
+  ret = rgw_rados_operate(dpp, rados_ioctx, Q_LIST_OBJECT_NAME, &op, y);
+  if (ret < 0) {
+    ldpp_dout(dpp, 1) << "ERROR: failed to remove queue: " << topic_name << " from queue list. error: " << ret << dendl;
+    return ret;
+  }
+  ldpp_dout(dpp, 20) << "INFO: queue: " << topic_name << " removed from queue list"  << dendl;
+  return 0;
+}
+
 int remove_persistent_topic(const std::string& topic_name, optional_yield y) {
   if (!s_manager) {
     return -EAGAIN;
   }
-  return s_manager->remove_persistent_topic(topic_name, y);
+  return remove_persistent_topic(s_manager, s_manager->rados_ioctx, topic_name, y);
 }
 
 rgw::sal::Object* get_object_with_atttributes(

src/rgw/driver/rados/rgw_notify.h

@@ -38,6 +38,9 @@ int add_persistent_topic(const std::string& topic_name, optional_yield y);
 // this operation also remove the topic name from the common (to all RGWs) list of all topics
 int remove_persistent_topic(const std::string& topic_name, 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_name, optional_yield y);
+
 // struct holding reservation information
 // populated in the publish_reserve call
 // then used to commit or abort the reservation

src/rgw/rgw_admin.cc

@@ -10566,6 +10566,12 @@ int main(int argc, const char **argv)
       return EINVAL;
     }
 
+    ret = rgw::notify::remove_persistent_topic(dpp(), static_cast<rgw::sal::RadosStore*>(driver)->getRados()->get_notif_pool_ctx(), topic_name, null_yield);
+    if (ret < 0) {
+      cerr << "ERROR: could not remove persistent topic: " << cpp_strerror(-ret) << std::endl;
+      return -ret;
+    }
+
     RGWPubSub ps(driver, tenant);
 
     ret = ps.remove_topic(dpp(), topic_name, null_yield);