From f7be1e0597234debab771a3766e7ca73a70cb78f Mon Sep 17 00:00:00 2001
From: Matt Benjamin <mbenjamin@redhat.com>
Date: Mon, 9 Jul 2018 17:21:22 -0400
Subject: [PATCH] rgw: Swift SLO size_bytes member is optional

For various reasons, the current SLO manifest processing logic
in RGWGetObj cannot deal properly with a manifest stored without
explicit value(s) for size_bytes, but size_bytes is optional.

Work around this by storing the component values of size_bytes for
each segment--or fail if any is invalid, as this is also correct
Swift behavior.

Fixes (part 2 of): http://tracker.ceph.com/issues/18936

Signed-off-by: Matt Benjamin <mbenjamin@redhat.com>
---
 src/rgw/rgw_op.cc         |   2 +-
 src/rgw/rgw_op.h          |   1 -
 src/rgw/rgw_rest_swift.cc | 101 +++++++++++++++++++++++++++++++++++++-
 src/rgw/rgw_rest_swift.h  |   2 +
 4 files changed, 103 insertions(+), 3 deletions(-)

diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index d0f770617a7..5003582ffac 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -1645,7 +1645,7 @@ int RGWGetObj::handle_slo_manifest(bufferlist& bl)
 
     slo_parts[total_len] = part;
     total_len += part.size;
-  }
+  } /* foreach entry */
 
   complete_etag(etag_sum, &lo_etag);
 
diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index b1bd7dd5b5b..21f037917a0 100644
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -1053,7 +1053,6 @@ public:
 
   virtual RGWPutObjProcessor *select_processor(RGWObjectCtx& obj_ctx, bool *is_multipart);
   void dispose_processor(RGWPutObjDataProcessor *processor);
-
   int verify_permission() override;
   void pre_exec() override;
   void execute() override;
diff --git a/src/rgw/rgw_rest_swift.cc b/src/rgw/rgw_rest_swift.cc
index 33d6fc8dc8f..214e01b2cf3 100644
--- a/src/rgw/rgw_rest_swift.cc
+++ b/src/rgw/rgw_rest_swift.cc
@@ -816,6 +816,93 @@ int RGWPutObj_ObjStore_SWIFT::verify_permission()
   }
 }
 
+int RGWPutObj_ObjStore_SWIFT::update_slo_segment_size(rgw_slo_entry& entry) {
+
+  int r = 0;
+  const string& path = entry.path;
+
+  /* If the path starts with slashes, strip them all. */
+  const size_t pos_init = path.find_first_not_of('/');
+
+  if (pos_init == string::npos) {
+    return -EINVAL;
+  }
+
+  const size_t pos_sep = path.find('/', pos_init);
+  if (pos_sep == string::npos) {
+    return -EINVAL;
+  }
+
+  string bucket_name = path.substr(pos_init, pos_sep - pos_init);
+  string obj_name = path.substr(pos_sep + 1);
+
+  rgw_bucket bucket;
+
+  if (bucket_name.compare(s->bucket.name) != 0) {
+    RGWBucketInfo bucket_info;
+    map<string, bufferlist> bucket_attrs;
+    RGWObjectCtx obj_ctx(store);
+    r = store->get_bucket_info(obj_ctx, s->user->user_id.tenant,
+			       bucket_name, bucket_info, nullptr,
+			       &bucket_attrs);
+    if (r < 0) {
+      ldpp_dout(this, 0) << "could not get bucket info for bucket="
+			 << bucket_name << dendl;
+      return r;
+    }
+    bucket = bucket_info.bucket;
+  } else {
+    bucket = s->bucket;
+  }
+
+  /* fetch the stored size of the seg (or error if not valid) */
+  rgw_obj_key slo_key(obj_name);
+  rgw_obj slo_seg(bucket, slo_key);
+
+  /* no prefetch */
+  RGWObjectCtx obj_ctx(store);
+  obj_ctx.obj.set_atomic(slo_seg);
+
+  RGWRados::Object op_target(store, s->bucket_info, obj_ctx, slo_seg);
+  RGWRados::Object::Read read_op(&op_target);
+
+  bool compressed;
+  RGWCompressionInfo cs_info;
+  map<std::string, buffer::list> attrs;
+  uint64_t size_bytes{0};
+
+  read_op.params.attrs = &attrs;
+  read_op.params.obj_size = &size_bytes;
+
+  r = read_op.prepare();
+  if (r < 0) {
+    return r;
+  }
+
+  r = rgw_compression_info_from_attrset(attrs, compressed, cs_info);
+  if (r < 0) {
+    return -EIO;
+  }
+
+  if (compressed) {
+    size_bytes = cs_info.orig_size;
+  }
+
+  /* "When the PUT operation sees the multipart-manifest=put query
+   * parameter, it reads the request body and verifies that each
+   * segment object exists and that the sizes and ETags match. If
+   * there is a mismatch, the PUT operation fails."
+   */
+  if (entry.size_bytes &&
+      (entry.size_bytes != size_bytes)) {
+    return -EINVAL;
+  }
+
+  entry.size_bytes = size_bytes;
+
+  return 0;
+} /* RGWPutObj_ObjStore_SWIFT::update_slo_segment_sizes */
+
 int RGWPutObj_ObjStore_SWIFT::get_params()
 {
   if (s->has_bad_meta) {
@@ -891,9 +978,21 @@ int RGWPutObj_ObjStore_SWIFT::get_params()
 
     MD5 etag_sum;
     uint64_t total_size = 0;
-    for (const auto& entry : slo_info->entries) {
+    for (auto& entry : slo_info->entries) {
       etag_sum.Update((const unsigned char *)entry.etag.c_str(),
                       entry.etag.length());
+
+      /* if size_bytes == 0, it should be replaced with the
+       * real segment size (which could be 0);  this follows from the
+       * fact that Swift requires all segments to exist, but permits
+       * the size_bytes element to be omitted from the SLO manifest, see
+       * https://docs.openstack.org/swift/latest/api/large_objects.html
+       */
+      r = update_slo_segment_size(entry);
+      if (r < 0) {
+	return r;
+      }
+
       total_size += entry.size_bytes;
 
       ldout(s->cct, 20) << "slo_part: " << entry.path
diff --git a/src/rgw/rgw_rest_swift.h b/src/rgw/rgw_rest_swift.h
index 6ed9346d5b8..8210f53049f 100644
--- a/src/rgw/rgw_rest_swift.h
+++ b/src/rgw/rgw_rest_swift.h
@@ -121,6 +121,8 @@ public:
   RGWPutObj_ObjStore_SWIFT() {}
   ~RGWPutObj_ObjStore_SWIFT() override {}
 
+  int update_slo_segment_size(rgw_slo_entry& entry);
+
   int verify_permission() override;
   int get_params() override;
   void send_response() override;
-- 
2.47.3