osd,librados: a rados op to create chunked object

Signed-off-by: Myoungwon Oh <omwmw@sk.com>

diff --git a/src/include/rados.h b/src/include/rados.h
index a5dcbfaa76d2c4233132cbbdc2abdb4975e25ce3..81effb4e0c94c730abf6d175edb8a2d972837fad 100644 (file)
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -310,6 +310,7 @@ extern int ceph_release_from_features(uint64_t features);
                                                                            \
        /* Extensible */                                                    \
        f(SET_REDIRECT, __CEPH_OSD_OP(WR, DATA, 39),    "set-redirect")     \
+       f(SET_CHUNK,    __CEPH_OSD_OP(WR, DATA, 40),    "set-chunk")        \
                                                                            \
        /** attrs **/                                                       \
        /* read */                                                          \

diff --git a/src/include/rados/librados.hpp b/src/include/rados/librados.hpp
index f4cd777df036e15a3b34266c034ad6704516770b..6e2a9d707212646a6f3c06648f2fbc07cbc31047 100644 (file)
--- a/src/include/rados/librados.hpp
+++ b/src/include/rados/librados.hpp
@@ -471,6 +471,9 @@ namespace librados
      */
     void set_redirect(const std::string& tgt_obj, const IoCtx& tgt_ioctx,
                      uint64_t tgt_version);
+    void set_chunk(uint64_t src_offset, uint64_t src_length, const IoCtx& tgt_ioctx,
+                   std::string tgt_oid, uint64_t tgt_offset);
+
 
     friend class IoCtx;
   };

diff --git a/src/librados/librados.cc b/src/librados/librados.cc
index e9157abcf91ecdfc4f9a230dc2319550f1982e30..0ca930f6bfd76a59c5a84734080ea4eafa96a781 100644 (file)
--- a/src/librados/librados.cc
+++ b/src/librados/librados.cc
@@ -614,6 +614,17 @@ void librados::ObjectWriteOperation::set_redirect(const std::string& tgt_obj,
                          tgt_ioctx.io_ctx_impl->oloc, tgt_version);
 }
 
+void librados::ObjectWriteOperation::set_chunk(uint64_t src_offset,
+                                              uint64_t src_length,
+                                              const IoCtx& tgt_ioctx,
+                                              string tgt_oid,
+                                              uint64_t tgt_offset)
+{
+  ::ObjectOperation *o = &impl->o;
+  o->set_chunk(src_offset, src_length, 
+              tgt_ioctx.io_ctx_impl->oloc, object_t(tgt_oid), tgt_offset);
+}
+
 void librados::ObjectWriteOperation::tmap_put(const bufferlist &bl)
 {
   ::ObjectOperation *o = &impl->o;

diff --git a/src/osd/PrimaryLogPG.cc b/src/osd/PrimaryLogPG.cc
index 99a4230da78950aca6ca5d91d09fb4c67951be8a..8aa077ffe955a3b81db13218facd41cd217f4b78 100644 (file)
--- a/src/osd/PrimaryLogPG.cc
+++ b/src/osd/PrimaryLogPG.cc
@@ -6105,6 +6105,68 @@ int PrimaryLogPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
 
       break;
 
+    case CEPH_OSD_OP_SET_CHUNK:
+      ++ctx->num_write;
+      {
+       if (pool.info.is_tier()) {
+         result = -EINVAL;
+         break;
+       }
+       if (!obs.exists) {
+         result = -ENOENT;
+         break;
+       }
+       if (get_osdmap()->require_osd_release < CEPH_RELEASE_LUMINOUS) {
+         result = -EOPNOTSUPP;
+         break;
+       }
+
+       object_locator_t tgt_oloc;
+       uint64_t src_offset, src_length, tgt_offset;
+       object_t tgt_name;
+       try {
+         ::decode(src_offset, bp);
+         ::decode(src_length, bp);
+         ::decode(tgt_oloc, bp);
+         ::decode(tgt_name, bp);
+         ::decode(tgt_offset, bp);
+       }
+       catch (buffer::error& e) {
+         result = -EINVAL;
+         goto fail;
+       }
+       
+       if (!src_length) {
+         result = -EINVAL;
+         goto fail;
+       }
+
+        if (!oi.manifest.is_chunked()) {
+          oi.manifest.clear();
+        }
+
+        pg_t raw_pg;
+       chunk_info_t chunk_info;
+       hobject_t target(tgt_name, tgt_oloc.key, snapid_t(),
+                        raw_pg.ps(), raw_pg.pool(),
+                        tgt_oloc.nspace);
+       get_osdmap()->object_locator_to_pg(tgt_name, tgt_oloc, raw_pg);
+       chunk_info.flags = chunk_info_t::FLAG_MISSING;
+       chunk_info.oid = target;
+       chunk_info.offset = tgt_offset;
+       chunk_info.length= src_length;
+       oi.manifest.chunk_map[src_offset] = chunk_info;
+
+        oi.set_flag(object_info_t::FLAG_MANIFEST);
+        oi.manifest.type = object_manifest_t::TYPE_CHUNKED;
+        ctx->modify = true;
+
+       dout(10) << "set-chunked oid:" << oi.soid << " user_version: " << oi.user_version 
+                << " chunk_info: " << chunk_info << dendl;
+      }
+
+      break;
+
       // -- object attrs --
       
     case CEPH_OSD_OP_SETXATTR:

diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index f13e65bb6c622f1725fb4ca1399509d9c7c38b63..5826a92e3d01d9b0b3ddb9449623fa987998e07d 100644 (file)
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -4548,7 +4548,7 @@ struct chunk_info_t {
   hobject_t oid;
   uint32_t flags;   // FLAG_*
 
-  chunk_info_t() : length(0), flags(0) { }
+  chunk_info_t() : offset(0), length(0), flags(0) { }
 
   static string get_flag_string(uint64_t flags) {
     string r;

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 3358d6bb24e8cb1b62b5213d514a03511c039b50..ee6bababf02e8a813267574018d635ccd15e6e81 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -1141,6 +1141,16 @@ struct ObjectOperation {
     ::encode(tgt_oloc, osd_op.indata);
   }
 
+  void set_chunk(uint64_t src_offset, uint64_t src_length, object_locator_t tgt_oloc,
+                object_t tgt_oid, uint64_t tgt_offset) {
+    OSDOp& osd_op = add_op(CEPH_OSD_OP_SET_CHUNK);
+    ::encode(src_offset, osd_op.indata);
+    ::encode(src_length, osd_op.indata);
+    ::encode(tgt_oloc, osd_op.indata);
+    ::encode(tgt_oid, osd_op.indata);
+    ::encode(tgt_offset, osd_op.indata);
+  }
+
   void set_alloc_hint(uint64_t expected_object_size,
                       uint64_t expected_write_size,
                      uint32_t flags) {

diff --git a/src/tools/rados/rados.cc b/src/tools/rados/rados.cc
index eb2dd24b57a0945531d24b9ff7d9b9f31abbf90b..c7d53cbccac90686da1268dd9459588612597169 100644 (file)
--- a/src/tools/rados/rados.cc
+++ b/src/tools/rados/rados.cc
@@ -122,6 +122,10 @@ void usage(ostream& out)
 "   listwatchers <obj-name>          list the watchers of this object\n"
 "   set-alloc-hint <obj-name> <expected-object-size> <expected-write-size>\n"
 "                                    set allocation hint for an object\n"
+"   set-redirect <object A> --target-pool <caspool> <target object A>\n"
+"                                    set redirect target\n"
+"   set-chunk <object A> <offset> <length> --target-pool <caspool> <target object A> <taget-offset>\n"
+"                                    convert an object to chunked object\n"
 "\n"
 "IMPORT AND EXPORT\n"
 "   export [filename]\n"
@@ -3486,6 +3490,54 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts,
       cerr << "error set-redirect " << pool_name << "/" << nargs[1] << " => " << target << "/" << target_obj << ": " << cpp_strerror(ret) << std::endl;
       goto out;
     }
+  } else if (strcmp(nargs[0], "set-chunk") == 0) {
+    if (!pool_name)
+      usage_exit();
+
+    const char *target = target_pool_name;
+    if (!target)
+      target = pool_name;
+
+    uint64_t offset;
+    uint64_t length;
+    uint64_t tgt_offset;
+    string tgt_oid;
+    if (nargs.size() < 6) {
+      usage_exit();
+    } else {
+      char* endptr = NULL;
+      offset = strtoull(nargs[2], &endptr, 10);
+      if (*endptr) {
+       cerr << "Invalid value for size: '" << nargs[2] << "'" << std::endl;
+       ret = -EINVAL;
+       goto out;
+      }
+      length = strtoull(nargs[3], &endptr, 10);
+      if (*endptr) {
+       cerr << "Invalid value for size: '" << nargs[2] << "'" << std::endl;
+       ret = -EINVAL;
+       goto out;
+      }
+      tgt_oid = string(nargs[4]);
+      tgt_offset = strtoull(nargs[5], &endptr, 10);
+      if (*endptr) {
+       cerr << "Invalid value for size: '" << nargs[2] << "'" << std::endl;
+       ret = -EINVAL;
+       goto out;
+      }
+    }
+
+    IoCtx target_ctx;
+    ret = rados.ioctx_create(target, target_ctx);
+    ObjectWriteOperation op;
+    op.set_chunk(offset, length, target_ctx, tgt_oid, tgt_offset);
+    ret = io_ctx.operate(nargs[1], &op);
+    if (ret < 0) {
+      cerr << "error set-chunk " << pool_name << "/" << nargs[1] << " " << " offset " << offset
+           << " length " << length << " target_pool " << target 
+           << "tgt_offset: " << tgt_offset << " : " << cpp_strerror(ret) << std::endl;
+      goto out;
+    }
   } else if (strcmp(nargs[0], "export") == 0) {
     // export [filename]
     if (!pool_name || nargs.size() > 2) {