rgw: add HTTPManager to BucketTrimManager

Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 98e307cb3054f4f4f3fca1099734504796bd2426..3931b640daf8e4d8be321fca50e8faff93e8d8ab 100644 (file)
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -6733,6 +6733,14 @@ next:
   }
 
   if (opt_cmd == OPT_BILOG_AUTOTRIM) {
+    RGWCoroutinesManager crs(store->ctx(), store->get_cr_registry());
+    RGWHTTPManager http(store->ctx(), crs.get_completion_mgr());
+    int ret = http.set_threaded();
+    if (ret < 0) {
+      cerr << "failed to initialize http client with " << cpp_strerror(ret) << std::endl;
+      return -ret;
+    }
+
     rgw::BucketTrimConfig config;
     configure_bucket_trim(store->ctx(), config);
 
@@ -6742,9 +6750,7 @@ next:
       cerr << "trim manager init failed with " << cpp_strerror(ret) << std::endl;
       return -ret;
     }
-
-    RGWCoroutinesManager crs(store->ctx(), store->get_cr_registry());
-    ret = crs.run(trim.create_admin_bucket_trim_cr());
+    ret = crs.run(trim.create_admin_bucket_trim_cr(&http));
     if (ret < 0) {
       cerr << "automated bilog trim failed with " << cpp_strerror(ret) << std::endl;
       return -ret;

diff --git a/src/rgw/rgw_sync_log_trim.cc b/src/rgw/rgw_sync_log_trim.cc
index 260848ac9c2311b92fe2e46e348c302b06a2ba94..74acfe4d3e4903ffa2a64a49da8ced3fab9bf019 100644 (file)
--- a/src/rgw/rgw_sync_log_trim.cc
+++ b/src/rgw/rgw_sync_log_trim.cc
@@ -283,14 +283,16 @@ struct BucketTrimObserver {
 /// trim the bilog of all of the given bucket instance's shards
 class BucketTrimInstanceCR : public RGWCoroutine {
   RGWRados *const store;
+  RGWHTTPManager *const http;
   BucketTrimObserver *const observer;
   std::string bucket_instance;
 
  public:
-  BucketTrimInstanceCR(RGWRados *store, BucketTrimObserver *observer,
+  BucketTrimInstanceCR(RGWRados *store, RGWHTTPManager *http,
+                       BucketTrimObserver *observer,
                        const std::string& bucket_instance)
     : RGWCoroutine(store->ctx()), store(store),
-      observer(observer),
+      http(http), observer(observer),
       bucket_instance(bucket_instance)
   {}
   int operate() {
@@ -302,15 +304,17 @@ class BucketTrimInstanceCR : public RGWCoroutine {
 /// trim each bucket instance while limiting the number of concurrent operations
 class BucketTrimInstanceCollectCR : public RGWShardCollectCR {
   RGWRados *const store;
+  RGWHTTPManager *const http;
   BucketTrimObserver *const observer;
   std::vector<std::string>::const_iterator bucket;
   std::vector<std::string>::const_iterator end;
  public:
-  BucketTrimInstanceCollectCR(RGWRados *store, BucketTrimObserver *observer,
+  BucketTrimInstanceCollectCR(RGWRados *store, RGWHTTPManager *http,
+                              BucketTrimObserver *observer,
                               const std::vector<std::string>& buckets,
                               int max_concurrent)
     : RGWShardCollectCR(store->ctx(), max_concurrent),
-      store(store), observer(observer),
+      store(store), http(http), observer(observer),
       bucket(buckets.begin()), end(buckets.end())
   {}
   bool spawn_next() override;
@@ -321,7 +325,7 @@ bool BucketTrimInstanceCollectCR::spawn_next()
   if (bucket == end) {
     return false;
   }
-  spawn(new BucketTrimInstanceCR(store, observer, *bucket), false);
+  spawn(new BucketTrimInstanceCR(store, http, observer, *bucket), false);
   ++bucket;
   return true;
 }
@@ -499,6 +503,7 @@ class MetadataListCR : public RGWSimpleCoroutine {
 
 class BucketTrimCR : public RGWCoroutine {
   RGWRados *const store;
+  RGWHTTPManager *const http;
   const BucketTrimConfig& config;
   BucketTrimObserver *const observer;
   const rgw_raw_obj& obj;
@@ -511,9 +516,10 @@ class BucketTrimCR : public RGWCoroutine {
 
   static const std::string section; //< metadata section for bucket instances
  public:
-  BucketTrimCR(RGWRados *store, const BucketTrimConfig& config,
-               BucketTrimObserver *observer, const rgw_raw_obj& obj)
-    : RGWCoroutine(store->ctx()), store(store), config(config),
+  BucketTrimCR(RGWRados *store, RGWHTTPManager *http,
+               const BucketTrimConfig& config, BucketTrimObserver *observer,
+               const rgw_raw_obj& obj)
+    : RGWCoroutine(store->ctx()), store(store), http(http), config(config),
       observer(observer), obj(obj), counter(config.counter_size)
   {}
 
@@ -610,7 +616,7 @@ int BucketTrimCR::operate()
     // trim bucket instances with limited concurrency
     set_status("trimming buckets");
     ldout(cct, 4) << "collected " << buckets.size() << " buckets for trim" << dendl;
-    yield call(new BucketTrimInstanceCollectCR(store, observer, buckets,
+    yield call(new BucketTrimInstanceCollectCR(store, http, observer, buckets,
                                                config.concurrent_buckets));
     // ignore errors from individual buckets
 
@@ -636,6 +642,7 @@ int BucketTrimCR::operate()
 
 class BucketTrimPollCR : public RGWCoroutine {
   RGWRados *const store;
+  RGWHTTPManager *const http;
   const BucketTrimConfig& config;
   BucketTrimObserver *const observer;
   const rgw_raw_obj& obj;
@@ -643,10 +650,11 @@ class BucketTrimPollCR : public RGWCoroutine {
   const std::string cookie;
 
  public:
-  BucketTrimPollCR(RGWRados *store, const BucketTrimConfig& config,
+  BucketTrimPollCR(RGWRados *store, RGWHTTPManager *http,
+                   const BucketTrimConfig& config,
                    BucketTrimObserver *observer, const rgw_raw_obj& obj)
-    : RGWCoroutine(store->ctx()), store(store), config(config),
-      observer(observer), obj(obj),
+    : RGWCoroutine(store->ctx()), store(store), http(http),
+      config(config), observer(observer), obj(obj),
       cookie(RGWSimpleRadosLockCR::gen_random_cookie(cct))
   {}
 
@@ -671,7 +679,7 @@ int BucketTrimPollCR::operate()
       }
 
       set_status("trimming");
-      yield call(new BucketTrimCR(store, config, observer, obj));
+      yield call(new BucketTrimCR(store, http, config, observer, obj));
       if (retcode < 0) {
         // on errors, unlock so other gateways can try
         set_status("unlocking");
@@ -834,16 +842,16 @@ void BucketTrimManager::on_bucket_changed(const boost::string_view& bucket)
   impl->counter.insert(bucket.to_string());
 }
 
-RGWCoroutine* BucketTrimManager::create_bucket_trim_cr()
+RGWCoroutine* BucketTrimManager::create_bucket_trim_cr(RGWHTTPManager *http)
 {
-  return new BucketTrimPollCR(impl->store, impl->config,
+  return new BucketTrimPollCR(impl->store, http, impl->config,
                               impl.get(), impl->status_obj);
 }
 
-RGWCoroutine* BucketTrimManager::create_admin_bucket_trim_cr()
+RGWCoroutine* BucketTrimManager::create_admin_bucket_trim_cr(RGWHTTPManager *http)
 {
   // return the trim coroutine without any polling
-  return new BucketTrimCR(impl->store, impl->config,
+  return new BucketTrimCR(impl->store, http, impl->config,
                           impl.get(), impl->status_obj);
 }
 

diff --git a/src/rgw/rgw_sync_log_trim.h b/src/rgw/rgw_sync_log_trim.h
index 9585426927fdc6ef235c66a7858edd2b405f7c4c..3d61b08dc1bcc8483b663ba51734db7b09f5b743 100644 (file)
--- a/src/rgw/rgw_sync_log_trim.h
+++ b/src/rgw/rgw_sync_log_trim.h
@@ -23,6 +23,7 @@
 
 class CephContext;
 class RGWCoroutine;
+class RGWHTTPManager;
 class RGWRados;
 
 namespace rgw {
@@ -76,10 +77,10 @@ class BucketTrimManager : public BucketChangeObserver {
   void on_bucket_changed(const boost::string_view& bucket_instance) override;
 
   /// create a coroutine to run the bucket trim process every trim interval
-  RGWCoroutine* create_bucket_trim_cr();
+  RGWCoroutine* create_bucket_trim_cr(RGWHTTPManager *http);
 
   /// create a coroutine to trim buckets directly via radosgw-admin
-  RGWCoroutine* create_admin_bucket_trim_cr();
+  RGWCoroutine* create_admin_bucket_trim_cr(RGWHTTPManager *http);
 };
 
 /// provides persistent storage for the trim manager's current position in the