From 80934a1c1702c7ee4b6e84f578dca0a71c7accc2 Mon Sep 17 00:00:00 2001
From: Radoslaw Zarzynski <rzarzyns@redhat.com>
Date: Tue, 27 Aug 2019 16:43:14 +0200
Subject: [PATCH] crimson/osd: add support for CEPH_OSD_OP_OMAPGETVALSBYKEYS.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>
---
 src/crimson/osd/objclass.cc     | 24 ++++++++++++++++++++-
 src/crimson/osd/ops_executer.cc |  6 ++++++
 src/crimson/osd/pg_backend.cc   | 38 +++++++++++++++++++++++++++++++++
 src/crimson/osd/pg_backend.h    |  5 +++++
 4 files changed, 72 insertions(+), 1 deletion(-)

diff --git a/src/crimson/osd/objclass.cc b/src/crimson/osd/objclass.cc
index 2d4f0f5797043..e238b7107a89d 100644
--- a/src/crimson/osd/objclass.cc
+++ b/src/crimson/osd/objclass.cc
@@ -290,7 +290,29 @@ int cls_cxx_map_get_val(cls_method_context_t hctx,
                         const string &key,
                         bufferlist *outbl)
 {
-  return 0;
+  OSDOp op{ CEPH_OSD_OP_OMAPGETVALSBYKEYS };
+  {
+    std::set<std::string> k{key};
+    encode(k, op.indata);
+  }
+  try {
+    reinterpret_cast<ceph::osd::OpsExecuter*>(hctx)->do_osd_op(op).get();
+  } catch (ceph::osd::error& e) {
+    return -e.code().value();
+  }
+  std::map<std::string, ceph::bufferlist> m;
+  try {
+    auto iter = op.outdata.cbegin();
+    decode(m, iter);
+  } catch (buffer::error&) {
+    return -EIO;
+  }
+  if (auto iter = std::begin(m); iter != std::end(m)) {
+    *outbl = std::move(iter->second);
+    return 0;
+  } else {
+    return -ENOENT;
+  }
 }
 
 int cls_cxx_map_set_val(cls_method_context_t hctx,
diff --git a/src/crimson/osd/ops_executer.cc b/src/crimson/osd/ops_executer.cc
index ccbbf343257e2..387b521503e59 100644
--- a/src/crimson/osd/ops_executer.cc
+++ b/src/crimson/osd/ops_executer.cc
@@ -410,6 +410,12 @@ OpsExecuter::do_osd_op(OSDOp& osd_op)
     // @dzafman this isn't possible yet. Maybe it could be accomplished
     // before crimson's readiness and we'd luckily don't need to carry.
     return dont_do_legacy_op();
+
+  // OMAP
+  case CEPH_OSD_OP_OMAPGETVALSBYKEYS:
+    return do_read_op([&osd_op] (auto& backend, const auto& os) {
+      return backend.omap_get_vals_by_keys(os, osd_op);
+    });
   default:
     logger().warn("unknown op {}", ceph_osd_op_name(op.op));
     throw std::runtime_error(
diff --git a/src/crimson/osd/pg_backend.cc b/src/crimson/osd/pg_backend.cc
index 144fb7d05c81b..8291fa416b641 100644
--- a/src/crimson/osd/pg_backend.cc
+++ b/src/crimson/osd/pg_backend.cc
@@ -484,3 +484,41 @@ seastar::future<ceph::bufferptr> PGBackend::getxattr(
 {
   return store->get_attr(coll, ghobject_t{soid}, key);
 }
+
+static seastar::future<ceph::os::FuturizedStore::omap_values_t>
+maybe_get_omap_vals_by_keys(
+  auto& store,
+  const auto& coll,
+  const auto& oi,
+  const auto& keys_to_get)
+{
+  if (oi.is_omap()) {
+    return store->omap_get_values(coll, ghobject_t{oi.soid}, keys_to_get);
+  } else {
+    return seastar::make_ready_future<ceph::os::FuturizedStore::omap_values_t>(
+      ceph::os::FuturizedStore::omap_values_t{});
+  }
+}
+
+seastar::future<> PGBackend::omap_get_vals_by_keys(
+  const ObjectState& os,
+  OSDOp& osd_op) const
+{
+  std::set<std::string> keys_to_get;
+  try {
+    auto p = osd_op.indata.cbegin();
+    decode(keys_to_get, p);
+  } catch (buffer::error&) {
+    throw ceph::osd::invalid_argument();
+  }
+
+  return maybe_get_omap_vals_by_keys(store, coll, os.oi, keys_to_get).then(
+    [&osd_op] (ceph::os::FuturizedStore::omap_values_t vals) {
+      encode(vals, osd_op.outdata);
+      return seastar::now();
+    });
+
+  // TODO:
+  //ctx->delta_stats.num_rd_kb += shift_round_up(osd_op.outdata.length(), 10);
+  //ctx->delta_stats.num_rd++;
+}
diff --git a/src/crimson/osd/pg_backend.h b/src/crimson/osd/pg_backend.h
index 837c3ed1aae22..b4553157f98a2 100644
--- a/src/crimson/osd/pg_backend.h
+++ b/src/crimson/osd/pg_backend.h
@@ -88,6 +88,11 @@ public:
     const hobject_t& soid,
     std::string_view key) const;
 
+  // OMAP
+  seastar::future<> omap_get_vals_by_keys(
+    const ObjectState& os,
+    OSDOp& osd_op) const;
+
   virtual void got_rep_op_reply(const MOSDRepOpReply&) {}
 
 protected:
-- 
2.39.5