Merge pull request #458 from ceph/wip-5693

Wip 5693
Reviewed-by: Greg Farnum <greg@inktank.com>

10 files changed, 111 insertions, 17 deletions

diff --git a/PendingReleaseNotes b/PendingReleaseNotes
index a9880942b5a..7a9adf7293e 100644
--- a/PendingReleaseNotes
+++ b/PendingReleaseNotes
@@ -19,3 +19,9 @@ v0.67
   commandline tool.  ceph_rest_api.py can be used as a WSGI application
   for deployment in a more-capable web server.  See ceph-rest-api.8
   for more.
+
+* rgw copy object operation may return extra progress info during the
+  operation. At this point it will only happen when doing cross zone
+  copy operations. The S3 response will now return extra <Progress>
+  field under the <CopyResult> container. The Swift response will
+  now send the progress as a json array.
diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index defb71ee514..ff23ba56232 100644
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -675,6 +675,8 @@ OPTION(rgw_md_log_max_shards, OPT_INT, 64) // max shards for metadata log
 OPTION(rgw_num_zone_opstate_shards, OPT_INT, 128) // max shards for keeping inter-region copy progress info
 OPTION(rgw_opstate_ratelimit_sec, OPT_INT, 30) // min time between opstate updates on a single upload (0 for disabling ratelimit)
 OPTION(rgw_curl_wait_timeout_ms, OPT_INT, 1000) // timeout for certain curl calls
+OPTION(rgw_copy_obj_progress, OPT_BOOL, true) // should dump progress during long copy operations?
+OPTION(rgw_copy_obj_progress_every_bytes, OPT_INT, 1024 * 1024) // min bytes between copy progress output
 
 OPTION(rgw_data_log_window, OPT_INT, 30) // data log entries window (in seconds)
 OPTION(rgw_data_log_changes_size, OPT_INT, 1000) // number of in-memory entries to hold for data changes log
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index 45477486ccc..7760a2f5c52 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -1654,6 +1654,25 @@ int RGWCopyObj::init_common()
   return 0;
 }
 
+static void copy_obj_progress_cb(off_t ofs, void *param)
+{
+  RGWCopyObj *op = static_cast<RGWCopyObj *>(param);
+  op->progress_cb(ofs);
+}
+
+void RGWCopyObj::progress_cb(off_t ofs)
+{
+  if (!s->cct->_conf->rgw_copy_obj_progress)
+    return;
+
+  if (ofs - last_ofs < s->cct->_conf->rgw_copy_obj_progress_every_bytes)
+    return;
+
+  send_partial_response(ofs);
+
+  last_ofs = ofs;
+}
+
 void RGWCopyObj::execute()
 {
   rgw_obj src_obj, dst_obj;
@@ -1685,7 +1704,9 @@ void RGWCopyObj::execute()
                         replace_attrs,
                         attrs, RGW_OBJ_CATEGORY_MAIN,
                         &s->req_id, /* use req_id as tag */
-                        &s->err);
+                        &s->err,
+                        copy_obj_progress_cb, (void *)this
+                        );
 }
 
 int RGWGetACLs::verify_permission()
diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index e107b90a155..5da2e4f472c 100644
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -438,6 +438,8 @@ protected:
   string client_id;
   string op_id;
 
+  off_t last_ofs;
+
 
   int init_common();
 
@@ -460,6 +462,7 @@ public:
     ret = 0;
     mtime = 0;
     replace_attrs = false;
+    last_ofs = 0;
   }
 
   virtual void init(RGWRados *store, struct req_state *s, RGWHandler *h) {
@@ -468,9 +471,11 @@ public:
   }
   int verify_permission();
   void execute();
+  void progress_cb(off_t ofs);
 
   virtual int init_dest_policy() { return 0; }
   virtual int get_params() = 0;
+  virtual void send_partial_response(off_t ofs) {}
   virtual void send_response() = 0;
   virtual const string name() { return "copy_obj"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 0c7b22a42d3..8af03b03a8f 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -2397,9 +2397,16 @@ class RGWRadosPutObj : public RGWGetDataCB
   rgw_obj obj;
   RGWPutObjProcessor_Atomic *processor;
   RGWOpStateSingleOp *opstate;
+  void (*progress_cb)(off_t, void *);
+  void *progress_data;
 public:
-  RGWRadosPutObj(RGWPutObjProcessor_Atomic *p, RGWOpStateSingleOp *_ops) : processor(p), opstate(_ops) {}
+  RGWRadosPutObj(RGWPutObjProcessor_Atomic *p, RGWOpStateSingleOp *_ops,
+                 void (*_progress_cb)(off_t, void *), void *_progress_data) : processor(p), opstate(_ops),
+                                                                       progress_cb(_progress_cb),
+                                                                       progress_data(_progress_data) {}
   int handle_data(bufferlist& bl, off_t ofs, off_t len) {
+    progress_cb(ofs, progress_data);
+
     void *handle;
     int ret = processor->handle_data(bl, ofs, &handle);
     if (ret < 0)
@@ -2477,7 +2484,9 @@ int RGWRados::copy_obj(void *ctx,
                map<string, bufferlist>& attrs,
                RGWObjCategory category,
                string *ptag,
-               struct rgw_err *err)
+               struct rgw_err *err,
+               void (*progress_cb)(off_t, void *),
+               void *progress_data)
 {
   int ret;
   uint64_t total_len, obj_size;
@@ -2545,7 +2554,7 @@ int RGWRados::copy_obj(void *ctx,
       ldout(cct, 0) << "ERROR: failed to set opstate ret=" << ret << dendl;
       return ret;
     }
-    RGWRadosPutObj cb(&processor, &opstate);
+    RGWRadosPutObj cb(&processor, &opstate, progress_cb, progress_data);
     string etag;
     map<string, string> req_headers;
     time_t set_mtime;
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index c9924e0dc56..bcc40900299 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -1121,7 +1121,9 @@ public:
                map<std::string, bufferlist>& attrs,
                RGWObjCategory category,
                string *ptag,
-               struct rgw_err *err);
+               struct rgw_err *err,
+               void (*progress_cb)(off_t, void *),
+               void *progress_data);
 
   int copy_obj_data(void *ctx,
 	       void *handle, off_t end,
diff --git a/src/rgw/rgw_rest_s3.cc b/src/rgw/rgw_rest_s3.cc
index 66f6652ec6a..e131eeee28d 100644
--- a/src/rgw/rgw_rest_s3.cc
+++ b/src/rgw/rgw_rest_s3.cc
@@ -1300,15 +1300,33 @@ int RGWCopyObj_ObjStore_S3::get_params()
   return 0;
 }
 
-void RGWCopyObj_ObjStore_S3::send_response()
+void RGWCopyObj_ObjStore_S3::send_partial_response(off_t ofs)
 {
-  if (ret)
+  if (!sent_header) {
+    if (ret)
     set_req_state_err(s, ret);
-  dump_errno(s);
+    dump_errno(s);
+
+    end_header(s, "binary/octet-stream");
+    if (ret == 0) {
+      s->formatter->open_object_section("CopyObjectResult");
+    }
+    sent_header = true;
+  } else {
+    /* Send progress field. Note that this diverge from the original S3
+     * spec. We do this in order to keep connection alive.
+     */
+    s->formatter->dump_int("Progress", (uint64_t)ofs);
+  }
+  rgw_flush_formatter(s, s->formatter);
+}
+
+void RGWCopyObj_ObjStore_S3::send_response()
+{
+  if (!sent_header)
+    send_partial_response(0);
 
-  end_header(s, "binary/octet-stream");
   if (ret == 0) {
-    s->formatter->open_object_section("CopyObjectResult");
     dump_time(s, "LastModified", &mtime);
     map<string, bufferlist>::iterator iter = attrs.find(RGW_ATTR_ETAG);
     if (iter != attrs.end()) {
diff --git a/src/rgw/rgw_rest_s3.h b/src/rgw/rgw_rest_s3.h
index e2a1b0b92eb..a0af4eac9fd 100644
--- a/src/rgw/rgw_rest_s3.h
+++ b/src/rgw/rgw_rest_s3.h
@@ -143,12 +143,14 @@ public:
 };
 
 class RGWCopyObj_ObjStore_S3 : public RGWCopyObj_ObjStore {
+  bool sent_header;
 public:
-  RGWCopyObj_ObjStore_S3() {}
+  RGWCopyObj_ObjStore_S3() : sent_header(false) {}
   ~RGWCopyObj_ObjStore_S3() {}
 
   int init_dest_policy();
   int get_params();
+  void send_partial_response(off_t ofs);
   void send_response();
 };
 
diff --git a/src/rgw/rgw_rest_swift.cc b/src/rgw/rgw_rest_swift.cc
index 157158e7ed7..80438a6556d 100644
--- a/src/rgw/rgw_rest_swift.cc
+++ b/src/rgw/rgw_rest_swift.cc
@@ -475,13 +475,40 @@ int RGWCopyObj_ObjStore_SWIFT::get_params()
   return 0;
 }
 
+void RGWCopyObj_ObjStore_SWIFT::send_partial_response(off_t ofs)
+{
+  if (!sent_header) {
+    if (!ret)
+      ret = STATUS_CREATED;
+    set_req_state_err(s, ret);
+    dump_errno(s);
+    end_header(s);
+
+    /* Send progress information. Note that this diverge from the original swift
+     * spec. We do this in order to keep connection alive.
+     */
+    if (ret == 0) {
+      s->formatter->open_array_section("progress");
+    }
+    sent_header = true;
+  } else {
+    s->formatter->dump_int("ofs", (uint64_t)ofs);
+  }
+  rgw_flush_formatter(s, s->formatter);
+}
+
 void RGWCopyObj_ObjStore_SWIFT::send_response()
 {
-  if (!ret)
-    ret = STATUS_CREATED;
-  set_req_state_err(s, ret);
-  dump_errno(s);
-  end_header(s);
+  if (!sent_header) {
+   if (!ret)
+      ret = STATUS_CREATED;
+    set_req_state_err(s, ret);
+    dump_errno(s);
+    end_header(s);
+  } else {
+    s->formatter->close_section();
+    rgw_flush_formatter(s, s->formatter);
+  }
 }
 
 int RGWGetObj_ObjStore_SWIFT::send_response_data(bufferlist& bl, off_t bl_ofs, off_t bl_len)
diff --git a/src/rgw/rgw_rest_swift.h b/src/rgw/rgw_rest_swift.h
index e4b6f0bccee..1c23ab29204 100644
--- a/src/rgw/rgw_rest_swift.h
+++ b/src/rgw/rgw_rest_swift.h
@@ -100,13 +100,15 @@ public:
 };
 
 class RGWCopyObj_ObjStore_SWIFT : public RGWCopyObj_ObjStore {
+  bool sent_header;
 public:
-  RGWCopyObj_ObjStore_SWIFT() {}
+  RGWCopyObj_ObjStore_SWIFT() : sent_header(false) {}
   ~RGWCopyObj_ObjStore_SWIFT() {}
 
   int init_dest_policy();
   int get_params();
   void send_response();
+  void send_partial_response(off_t ofs);
 };
 
 class RGWGetACLs_ObjStore_SWIFT : public RGWGetACLs_ObjStore {