[DNM] rgw: sync plugin to AWS S3

src/rgw/CMakeLists.txt

@@ -58,6 +58,7 @@ set(rgw_a_srcs
   rgw_sync.cc
   rgw_data_sync.cc
   rgw_sync_module.cc
+  rgw_sync_module_aws.cc
   rgw_sync_module_es.cc
   rgw_sync_module_log.cc
   rgw_period_history.cc

src/rgw/rgw_cr_rest.h

@@ -7,25 +7,30 @@
 #include "rgw_coroutine.h"
 #include "rgw_rest_conn.h"
 
-template <class T>
-class RGWReadRESTResourceCR : public RGWSimpleCoroutine {
+class RGWReadRawRESTResourceCR : public RGWSimpleCoroutine{
+  bufferlist *result;
+ protected:
   RGWRESTConn *conn;
   RGWHTTPManager *http_manager;
   string path;
   param_vec_t params;
-  T *result;
-
+ public:
   boost::intrusive_ptr<RGWRESTReadResource> http_op;
+  RGWReadRawRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                           RGWHTTPManager *_http_manager, const string& _path,
+                           rgw_http_param_pair *params, bufferlist *_result)
+    : RGWSimpleCoroutine(_cct), result(_result), conn(_conn), http_manager(_http_manager),
+    path(_path), params(make_param_list(params))
+  {}
 
-public:
-  RGWReadRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
-                        RGWHTTPManager *_http_manager, const string& _path,
-                        rgw_http_param_pair *params, T *_result)
-    : RGWSimpleCoroutine(_cct), conn(_conn), http_manager(_http_manager),
-      path(_path), params(make_param_list(params)), result(_result)
+ RGWReadRawRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                          RGWHTTPManager *_http_manager, const string& _path,
+                          rgw_http_param_pair *params)
+   : RGWSimpleCoroutine(_cct), conn(_conn), http_manager(_http_manager),
+    path(_path), params(make_param_list(params))
   {}
 
-  ~RGWReadRESTResourceCR() override {
+  ~RGWReadRawRESTResourceCR() override {
     request_cleanup();
   }
 
@@ -46,8 +51,16 @@ class RGWReadRESTResourceCR : public RGWSimpleCoroutine {
     return 0;
   }
 
+  virtual int wait_result() {
+    return http_op->wait_bl(result);
+  }
+
   int request_complete() override {
-    int ret = http_op->wait(result);
+
+    int ret;
+
+    ret = wait_result();
+
     auto op = std::move(http_op); // release ref on return
     if (ret < 0) {
       error_stream << "http operation failed: " << op->to_str()
@@ -65,48 +78,84 @@ class RGWReadRESTResourceCR : public RGWSimpleCoroutine {
       http_op = NULL;
     }
   }
+
 };
 
-template <class S, class T>
-class RGWSendRESTResourceCR : public RGWSimpleCoroutine {
+
+template <class T>
+class RGWReadRESTResourceCR : public RGWReadRawRESTResourceCR {
+  T *result;
+ public:
+ RGWReadRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                       RGWHTTPManager *_http_manager, const string& _path,
+                       rgw_http_param_pair *params, T *_result)
+   : RGWReadRawRESTResourceCR(_cct, _conn, _http_manager, _path, params), result(_result)
+  {}
+
+  int wait_result() override {
+    return http_op->wait(result);
+  }
+
+};
+
+template <class T>
+class RGWSendRawRESTResourceCR: public RGWSimpleCoroutine {
+ protected:
   RGWRESTConn *conn;
   RGWHTTPManager *http_manager;
   string method;
   string path;
   param_vec_t params;
   T *result;
-  S input;
-
+  bufferlist input_bl;
+  bool send_content_length=false;
   boost::intrusive_ptr<RGWRESTSendResource> http_op;
 
-public:
-  RGWSendRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
-                        RGWHTTPManager *_http_manager,
-                        const string& _method, const string& _path,
-                        rgw_http_param_pair *_params, S& _input, T *_result)
-    : RGWSimpleCoroutine(_cct), conn(_conn), http_manager(_http_manager),
-      method(_method), path(_path), params(make_param_list(_params)), result(_result),
-      input(_input)
-  {}
+ public:
+ RGWSendRawRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                          RGWHTTPManager *_http_manager,
+                          const string& _method, const string& _path,
+                          rgw_http_param_pair *_params, bufferlist& _input, T *_result, bool _send_content_length)
+   : RGWSimpleCoroutine(_cct), conn(_conn), http_manager(_http_manager),
+    method(_method), path(_path), params(make_param_list(_params)), result(_result),
+    input_bl(_input), send_content_length(_send_content_length)
+    {}
+
+ RGWSendRawRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                          RGWHTTPManager *_http_manager,
+                          const string& _method, const string& _path,
+                          rgw_http_param_pair *_params, T *_result)
+   : RGWSimpleCoroutine(_cct), conn(_conn), http_manager(_http_manager),
+    method(_method), path(_path), params(make_param_list(_params)), result(_result)
+    {}
 
-  ~RGWSendRESTResourceCR() override {
+
+
+  ~RGWSendRawRESTResourceCR() override {
     request_cleanup();
   }
 
+  void set_input_bl(bufferlist bl){
+    input_bl = std::move(bl);
+  }
+
   int send_request() override {
+
+
+    param_vec_t p;
+    if (send_content_length){
+      lsubdout(cct, rgw, 0) << "abhi: sending content length of " << input_bl.length() << dendl;
+      string content_length = to_string(input_bl.length());
+      p.push_back(param_pair_t("CONTENT_LENGTH",content_length));
+    }
+
+
     auto op = boost::intrusive_ptr<RGWRESTSendResource>(
-        new RGWRESTSendResource(conn, method, path, params, NULL, http_manager));
+        new RGWRESTSendResource(conn, method, path, params, nullptr, http_manager));
 
     op->set_user_info((void *)stack);
 
-    JSONFormatter jf;
-    encode_json("data", input, &jf);
-    std::stringstream ss;
-    jf.flush(ss);
-    bufferlist bl;
-    bl.append(ss.str());
-
-    int ret = op->aio_send(bl);
+    int ret = op->aio_send(input_bl);
     if (ret < 0) {
       lsubdout(cct, rgw, 0) << "ERROR: failed to send request" << dendl;
       op->put();
@@ -145,6 +194,26 @@ class RGWSendRESTResourceCR : public RGWSimpleCoroutine {
   }
 };
 
+template <class S, class T>
+class RGWSendRESTResourceCR : public RGWSendRawRESTResourceCR<T> {
+ public:
+  RGWSendRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                           RGWHTTPManager *_http_manager,
+                           const string& _method, const string& _path,
+                        rgw_http_param_pair *_params,S& _input, T *_result)
+    : RGWSendRawRESTResourceCR<T>(_cct, _conn, _http_manager, _method, _path, _params, _result){
+
+    JSONFormatter jf;
+    encode_json("data", _input, &jf);
+    std::stringstream ss;
+    jf.flush(ss);
+    //bufferlist bl;
+    this->input_bl.append(ss.str());
+    //set_input_bl(std::move(bl));
+  }
+
+};
+
 template <class S, class T>
 class RGWPostRESTResourceCR : public RGWSendRESTResourceCR<S, T> {
 public:
@@ -157,6 +226,18 @@ class RGWPostRESTResourceCR : public RGWSendRESTResourceCR<S, T> {
                             _params, _input, _result) {}
 };
 
+template <class T>
+class RGWPutRawRESTResourceCR: public RGWSendRawRESTResourceCR <T> {
+ public:
+  RGWPutRawRESTResourceCR(CephContext *_cct, RGWRESTConn *_conn,
+                          RGWHTTPManager *_http_manager,
+                          const string& _path,
+                          rgw_http_param_pair *_params, bufferlist& _input, T *_result)
+    : RGWSendRawRESTResourceCR<T>(_cct, _conn, _http_manager, "PUT", _path, _params, _input, _result, true){}
+
+};
+
+
 template <class S, class T>
 class RGWPutRESTResourceCR : public RGWSendRESTResourceCR<S, T> {
 public:
@@ -165,8 +246,8 @@ class RGWPutRESTResourceCR : public RGWSendRESTResourceCR<S, T> {
                         const string& _path,
                         rgw_http_param_pair *_params, S& _input, T *_result)
     : RGWSendRESTResourceCR<S, T>(_cct, _conn, _http_manager,
-                            "PUT", _path,
-                            _params, _input, _result) {}
+                                  "PUT", _path,
+                                  _params, _input, _result) {}
 };
 
 class RGWDeleteRESTResourceCR : public RGWSimpleCoroutine {

src/rgw/rgw_rest_client.cc

@@ -687,6 +687,15 @@ int RGWRESTStreamRWRequest::get_resource(RGWAccessKey& key, map<string, string>&
     pmanager = mgr;
   }
 
+  // Not sure if this is the place to set a send_size, curl otherwise sets
+  // chunked option and doesn't send content length anymore
+  uint64_t send_size = (size_t)(outbl.length() - write_ofs);
+
+  if (send_size > 0){
+    ldout(cct,20) << "Setting content length as " << send_size << dendl;
+    set_send_length(send_size);
+  }
+
   int r = pmanager->add_request(this, new_info.method, new_url.c_str());
   if (r < 0)
     return r;

src/rgw/rgw_rest_conn.h

@@ -60,6 +60,10 @@ class RGWRESTConn
 public:
 
   RGWRESTConn(CephContext *_cct, RGWRados *store, const string& _remote_id, const list<string>& endpoints);
+
+  RGWRESTConn(CephContext *_cct, const string& _remote_id, const list<string>& endpoints, RGWAccessKey _cred):
+    cct(_cct), endpoints(endpoints.begin(),endpoints.end()), key(std::move(_cred)), remote_id(_remote_id) {}
+
   int get_url(string& endpoint);
   string get_url();
   const string& get_self_zonegroup() {

src/rgw/rgw_sync_module.cc

@@ -7,6 +7,7 @@
 
 #include "rgw_sync_module_log.h"
 #include "rgw_sync_module_es.h"
+#include "rgw_sync_module_aws.h"
 
 #define dout_subsys ceph_subsys_rgw
 
@@ -62,4 +63,7 @@ void rgw_register_sync_modules(RGWSyncModulesManager *modules_manager)
 
   RGWSyncModuleRef es_module(std::make_shared<RGWElasticSyncModule>());
   modules_manager->register_module("elasticsearch", es_module);
+
+  RGWSyncModuleRef aws_module(std::make_shared<RGWAWSSyncModule>());
+  modules_manager->register_module("aws", aws_module);
 }