ceph/src/rgw/rgw_metadata.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #include <boost/intrusive_ptr.hpp>
   5 #include "common/ceph_json.h"
   6 #include "common/errno.h"
   7 #include "rgw_metadata.h"
   8 #include "rgw_coroutine.h"
   9 #include "cls/version/cls_version_types.h"
  10
  11 #include "rgw_zone.h"
  12 #include "rgw_tools.h"
  13 #include "rgw_mdlog.h"
  14 #include "rgw_sal.h"
  15
  16 #include "rgw_cr_rados.h"
  17
  18 #include "services/svc_zone.h"
  19 #include "services/svc_meta.h"
  20 #include "services/svc_meta_be.h"
  21 #include "services/svc_meta_be_sobj.h"
  22 #include "services/svc_cls.h"
  23
  24 #include "include/ceph_assert.h"
  25
  26 #include <boost/asio/yield.hpp>
  27
  28 #define dout_subsys ceph_subsys_rgw
  29
  30 const std::string RGWMetadataLogHistory::oid = "meta.history";
  31
  32 void LogStatusDump::dump(Formatter *f) const {
  33   string s;
  34   switch (status) {
  35     case MDLOG_STATUS_WRITE:
  36       s = "write";
  37       break;
  38     case MDLOG_STATUS_SETATTRS:
  39       s = "set_attrs";
  40       break;
  41     case MDLOG_STATUS_REMOVE:
  42       s = "remove";
  43       break;
  44     case MDLOG_STATUS_COMPLETE:
  45       s = "complete";
  46       break;
  47     case MDLOG_STATUS_ABORT:
  48       s = "abort";
  49       break;
  50     default:
  51       s = "unknown";
  52       break;
  53   }
  54   encode_json("status", s, f);
  55 }
  56
  57 void RGWMetadataLogData::encode(bufferlist& bl) const {
  58   ENCODE_START(1, 1, bl);
  59   encode(read_version, bl);
  60   encode(write_version, bl);
  61   uint32_t s = (uint32_t)status;
  62   encode(s, bl);
  63   ENCODE_FINISH(bl);
  64 }
  65
  66 void RGWMetadataLogData::decode(bufferlist::const_iterator& bl) {
  67    DECODE_START(1, bl);
  68    decode(read_version, bl);
  69    decode(write_version, bl);
  70    uint32_t s;
  71    decode(s, bl);
  72    status = (RGWMDLogStatus)s;
  73    DECODE_FINISH(bl);
  74 }
  75
  76 void RGWMetadataLogData::dump(Formatter *f) const {
  77   encode_json("read_version", read_version, f);
  78   encode_json("write_version", write_version, f);
  79   encode_json("status", LogStatusDump(status), f);
  80 }
  81
  82 void decode_json_obj(RGWMDLogStatus& status, JSONObj *obj) {
  83   string s;
  84   JSONDecoder::decode_json("status", s, obj);
  85   if (s == "complete") {
  86     status = MDLOG_STATUS_COMPLETE;
  87   } else if (s == "write") {
  88     status = MDLOG_STATUS_WRITE;
  89   } else if (s == "remove") {
  90     status = MDLOG_STATUS_REMOVE;
  91   } else if (s == "set_attrs") {
  92     status = MDLOG_STATUS_SETATTRS;
  93   } else if (s == "abort") {
  94     status = MDLOG_STATUS_ABORT;
  95   } else {
  96     status = MDLOG_STATUS_UNKNOWN;
  97   }
  98 }
  99
 100 void RGWMetadataLogData::decode_json(JSONObj *obj) {
 101   JSONDecoder::decode_json("read_version", read_version, obj);
 102   JSONDecoder::decode_json("write_version", write_version, obj);
 103   JSONDecoder::decode_json("status", status, obj);
 104 }
 105
 106
 107 int RGWMetadataLog::add_entry(const DoutPrefixProvider *dpp, const string& hash_key, const string& section, const string& key, bufferlist& bl) {
 108   if (!svc.zone->need_to_log_metadata())
 109     return 0;
 110
 111   string oid;
 112   int shard_id;
 113
 114   rgw_shard_name(prefix, cct->_conf->rgw_md_log_max_shards, hash_key, oid, &shard_id);
 115   mark_modified(shard_id);
 116   real_time now = real_clock::now();
 117   return svc.cls->timelog.add(dpp, oid, now, section, key, bl, null_yield);
 118 }
 119
 120 int RGWMetadataLog::get_shard_id(const string& hash_key, int *shard_id)
 121 {
 122   string oid;
 123
 124   rgw_shard_name(prefix, cct->_conf->rgw_md_log_max_shards, hash_key, oid, shard_id);
 125   return 0;
 126 }
 127
 128 int RGWMetadataLog::store_entries_in_shard(const DoutPrefixProvider *dpp, list<cls_log_entry>& entries, int shard_id, librados::AioCompletion *completion)
 129 {
 130   string oid;
 131
 132   mark_modified(shard_id);
 133   rgw_shard_name(prefix, shard_id, oid);
 134   return svc.cls->timelog.add(dpp, oid, entries, completion, false, null_yield);
 135 }
 136
 137 void RGWMetadataLog::init_list_entries(int shard_id, const real_time& from_time, const real_time& end_time,
 138                                        const string& marker, void **handle)
 139 {
 140   LogListCtx *ctx = new LogListCtx();
 141
 142   ctx->cur_shard = shard_id;
 143   ctx->from_time = from_time;
 144   ctx->end_time  = end_time;
 145   ctx->marker    = marker;
 146
 147   get_shard_oid(ctx->cur_shard, ctx->cur_oid);
 148
 149   *handle = (void *)ctx;
 150 }
 151
 152 void RGWMetadataLog::complete_list_entries(void *handle) {
 153   LogListCtx *ctx = static_cast<LogListCtx *>(handle);
 154   delete ctx;
 155 }
 156
 157 int RGWMetadataLog::list_entries(const DoutPrefixProvider *dpp, void *handle,
 158                                  int max_entries,
 159                                  list<cls_log_entry>& entries,
 160                                  string *last_marker,
 161                                  bool *truncated) {
 162   LogListCtx *ctx = static_cast<LogListCtx *>(handle);
 163
 164   if (!max_entries) {
 165     *truncated = false;
 166     return 0;
 167   }
 168
 169   std::string next_marker;
 170   int ret = svc.cls->timelog.list(dpp, ctx->cur_oid, ctx->from_time, ctx->end_time,
 171                                   max_entries, entries, ctx->marker,
 172                                   &next_marker, truncated, null_yield);
 173   if ((ret < 0) && (ret != -ENOENT))
 174     return ret;
 175
 176   ctx->marker = std::move(next_marker);
 177   if (last_marker) {
 178     *last_marker = ctx->marker;
 179   }
 180
 181   if (ret == -ENOENT)
 182     *truncated = false;
 183
 184   return 0;
 185 }
 186
 187 int RGWMetadataLog::get_info(const DoutPrefixProvider *dpp, int shard_id, RGWMetadataLogInfo *info)
 188 {
 189   string oid;
 190   get_shard_oid(shard_id, oid);
 191
 192   cls_log_header header;
 193
 194   int ret = svc.cls->timelog.info(dpp, oid, &header, null_yield);
 195   if ((ret < 0) && (ret != -ENOENT))
 196     return ret;
 197
 198   info->marker = header.max_marker;
 199   info->last_update = header.max_time.to_real_time();
 200
 201   return 0;
 202 }
 203
 204 static void _mdlog_info_completion(librados::completion_t cb, void *arg)
 205 {
 206   auto infoc = static_cast<RGWMetadataLogInfoCompletion *>(arg);
 207   infoc->finish(cb);
 208   infoc->put(); // drop the ref from get_info_async()
 209 }
 210
 211 RGWMetadataLogInfoCompletion::RGWMetadataLogInfoCompletion(info_callback_t cb)
 212   : completion(librados::Rados::aio_create_completion((void *)this,
 213                                                       _mdlog_info_completion)),
 214     callback(cb)
 215 {
 216 }
 217
 218 RGWMetadataLogInfoCompletion::~RGWMetadataLogInfoCompletion()
 219 {
 220   completion->release();
 221 }
 222
 223 int RGWMetadataLog::get_info_async(const DoutPrefixProvider *dpp, int shard_id, RGWMetadataLogInfoCompletion *completion)
 224 {
 225   string oid;
 226   get_shard_oid(shard_id, oid);
 227
 228   completion->get(); // hold a ref until the completion fires
 229
 230   return svc.cls->timelog.info_async(dpp, completion->get_io_obj(), oid,
 231                                      &completion->get_header(),
 232                                      completion->get_completion());
 233 }
 234
 235 int RGWMetadataLog::trim(const DoutPrefixProvider *dpp, int shard_id, const real_time& from_time, const real_time& end_time,
 236                          const string& start_marker, const string& end_marker)
 237 {
 238   string oid;
 239   get_shard_oid(shard_id, oid);
 240
 241   return svc.cls->timelog.trim(dpp, oid, from_time, end_time, start_marker,
 242                                end_marker, nullptr, null_yield);
 243 }
 244
 245 int RGWMetadataLog::lock_exclusive(const DoutPrefixProvider *dpp, int shard_id, timespan duration, string& zone_id, string& owner_id) {
 246   string oid;
 247   get_shard_oid(shard_id, oid);
 248
 249   return svc.cls->lock.lock_exclusive(dpp, svc.zone->get_zone_params().log_pool, oid, duration, zone_id, owner_id);
 250 }
 251
 252 int RGWMetadataLog::unlock(const DoutPrefixProvider *dpp, int shard_id, string& zone_id, string& owner_id) {
 253   string oid;
 254   get_shard_oid(shard_id, oid);
 255
 256   return svc.cls->lock.unlock(dpp, svc.zone->get_zone_params().log_pool, oid, zone_id, owner_id);
 257 }
 258
 259 void RGWMetadataLog::mark_modified(int shard_id)
 260 {
 261   lock.get_read();
 262   if (modified_shards.find(shard_id) != modified_shards.end()) {
 263     lock.unlock();
 264     return;
 265   }
 266   lock.unlock();
 267
 268   std::unique_lock wl{lock};
 269   modified_shards.insert(shard_id);
 270 }
 271
 272 void RGWMetadataLog::read_clear_modified(set<int> &modified)
 273 {
 274   std::unique_lock wl{lock};
 275   modified.swap(modified_shards);
 276   modified_shards.clear();
 277 }
 278
 279 obj_version& RGWMetadataObject::get_version()
 280 {
 281   return objv;
 282 }
 283
 284 class RGWMetadataTopHandler : public RGWMetadataHandler {
 285   struct iter_data {
 286     set<string> sections;
 287     set<string>::iterator iter;
 288   };
 289
 290   struct Svc {
 291     RGWSI_Meta *meta{nullptr};
 292   } svc;
 293
 294   RGWMetadataManager *mgr;
 295
 296 public:
 297   RGWMetadataTopHandler(RGWSI_Meta *meta_svc,
 298                         RGWMetadataManager *_mgr) : mgr(_mgr) {
 299     base_init(meta_svc->ctx());
 300     svc.meta = meta_svc;
 301   }
 302
 303   string get_type() override { return string(); }
 304
 305   RGWMetadataObject *get_meta_obj(JSONObj *jo, const obj_version& objv, const ceph::real_time& mtime) {
 306     return new RGWMetadataObject;
 307   }
 308
 309   int get(string& entry, RGWMetadataObject **obj, optional_yield y, const DoutPrefixProvider *dpp) override {
 310     return -ENOTSUP;
 311   }
 312
 313   int put(string& entry, RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
 314           optional_yield y, const DoutPrefixProvider *dpp, RGWMDLogSyncType type, bool from_remote_zone) override {
 315     return -ENOTSUP;
 316   }
 317
 318   int remove(string& entry, RGWObjVersionTracker& objv_tracker, optional_yield y, const DoutPrefixProvider *dpp) override {
 319     return -ENOTSUP;
 320   }
 321
 322   int mutate(const string& entry,
 323              const ceph::real_time& mtime,
 324              RGWObjVersionTracker *objv_tracker,
 325              optional_yield y,
 326              const DoutPrefixProvider *dpp,
 327              RGWMDLogStatus op_type,
 328              std::function<int()> f) {
 329     return -ENOTSUP;
 330   }
 331
 332   int list_keys_init(const DoutPrefixProvider *dpp, const string& marker, void **phandle) override {
 333     iter_data *data = new iter_data;
 334     list<string> sections;
 335     mgr->get_sections(sections);
 336     for (auto& s : sections) {
 337       data->sections.insert(s);
 338     }
 339     data->iter = data->sections.lower_bound(marker);
 340
 341     *phandle = data;
 342
 343     return 0;
 344   }
 345   int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) override  {
 346     iter_data *data = static_cast<iter_data *>(handle);
 347     for (int i = 0; i < max && data->iter != data->sections.end(); ++i, ++(data->iter)) {
 348       keys.push_back(*data->iter);
 349     }
 350
 351     *truncated = (data->iter != data->sections.end());
 352
 353     return 0;
 354   }
 355   void list_keys_complete(void *handle) override {
 356     iter_data *data = static_cast<iter_data *>(handle);
 357
 358     delete data;
 359   }
 360
 361   virtual string get_marker(void *handle) override {
 362     iter_data *data = static_cast<iter_data *>(handle);
 363
 364     if (data->iter != data->sections.end()) {
 365       return *(data->iter);
 366     }
 367
 368     return string();
 369   }
 370 };
 371
 372 RGWMetadataManager::RGWMetadataManager(RGWSI_Meta *_meta_svc)
 373   : cct(_meta_svc->ctx()), meta_svc(_meta_svc)
 374 {
 375   md_top_handler.reset(new RGWMetadataTopHandler(meta_svc, this));
 376 }
 377
 378 RGWMetadataManager::~RGWMetadataManager()
 379 {
 380 }
 381
 382 int RGWMetadataHandler::attach(RGWMetadataManager *manager)
 383 {
 384   return manager->register_handler(this);
 385 }
 386
 387 RGWMetadataHandler_GenericMetaBE::Put::Put(RGWMetadataHandler_GenericMetaBE *_handler,
 388                                            RGWSI_MetaBackend_Handler::Op *_op,
 389                                            string& _entry, RGWMetadataObject *_obj,
 390                                            RGWObjVersionTracker& _objv_tracker,
 391                                            optional_yield _y,
 392                                            RGWMDLogSyncType _type, bool _from_remote_zone):
 393   handler(_handler), op(_op),
 394   entry(_entry), obj(_obj),
 395   objv_tracker(_objv_tracker),
 396   apply_type(_type),
 397   y(_y),
 398   from_remote_zone(_from_remote_zone)
 399 {
 400 }
 401
 402 RGWMetadataHandlerPut_SObj::RGWMetadataHandlerPut_SObj(RGWMetadataHandler_GenericMetaBE *handler, RGWSI_MetaBackend_Handler::Op *op,
 403                                                        string& entry, RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
 404                                                        optional_yield y,
 405                                                        RGWMDLogSyncType type, bool from_remote_zone) : Put(handler, op, entry, obj, objv_tracker, y, type, from_remote_zone) {
 406 }
 407
 408 RGWMetadataHandlerPut_SObj::~RGWMetadataHandlerPut_SObj() {
 409 }
 410
 411 int RGWMetadataHandlerPut_SObj::put_pre(const DoutPrefixProvider *dpp)
 412 {
 413   int ret = get(&old_obj, dpp);
 414   if (ret < 0 && ret != -ENOENT) {
 415     return ret;
 416   }
 417   exists = (ret != -ENOENT);
 418
 419   oo.reset(old_obj);
 420
 421   auto old_ver = (!old_obj ? obj_version() : old_obj->get_version());
 422   auto old_mtime = (!old_obj ? ceph::real_time() : old_obj->get_mtime());
 423
 424   // are we actually going to perform this put, or is it too old?
 425   if (!handler->check_versions(exists, old_ver, old_mtime,
 426                                objv_tracker.write_version, obj->get_mtime(),
 427                                apply_type)) {
 428     return STATUS_NO_APPLY;
 429   }
 430
 431   objv_tracker.read_version = old_ver; /* maintain the obj version we just read */
 432
 433   return 0;
 434 }
 435
 436 int RGWMetadataHandlerPut_SObj::put(const DoutPrefixProvider *dpp)
 437 {
 438   int ret = put_check(dpp);
 439   if (ret != 0) {
 440     return ret;
 441   }
 442
 443   return put_checked(dpp);
 444 }
 445
 446 int RGWMetadataHandlerPut_SObj::put_checked(const DoutPrefixProvider *dpp)
 447 {
 448   RGWSI_MBSObj_PutParams params(obj->get_pattrs(), obj->get_mtime());
 449
 450   encode_obj(&params.bl);
 451
 452   int ret = op->put(entry, params, &objv_tracker, y, dpp);
 453   if (ret < 0) {
 454     return ret;
 455   }
 456
 457   return 0;
 458 }
 459
 460 int RGWMetadataHandler_GenericMetaBE::do_put_operate(Put *put_op, const DoutPrefixProvider *dpp)
 461 {
 462   int r = put_op->put_pre(dpp);
 463   if (r != 0) { /* r can also be STATUS_NO_APPLY */
 464     return r;
 465   }
 466
 467   r = put_op->put(dpp);
 468   if (r != 0) {
 469     return r;
 470   }
 471
 472   r = put_op->put_post(dpp);
 473   if (r != 0) {  /* e.g., -error or STATUS_APPLIED */
 474     return r;
 475   }
 476
 477   return 0;
 478 }
 479
 480 int RGWMetadataHandler_GenericMetaBE::get(string& entry, RGWMetadataObject **obj, optional_yield y, const DoutPrefixProvider *dpp)
 481 {
 482   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 483     return do_get(op, entry, obj, y, dpp);
 484   });
 485 }
 486
 487 int RGWMetadataHandler_GenericMetaBE::put(string& entry, RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
 488                                           optional_yield y, const DoutPrefixProvider *dpp, RGWMDLogSyncType type, bool from_remote_zone)
 489 {
 490   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 491     return do_put(op, entry, obj, objv_tracker, y, dpp, type, from_remote_zone);
 492   });
 493 }
 494
 495 int RGWMetadataHandler_GenericMetaBE::remove(string& entry, RGWObjVersionTracker& objv_tracker, optional_yield y, const DoutPrefixProvider *dpp)
 496 {
 497   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 498     return do_remove(op, entry, objv_tracker, y, dpp);
 499   });
 500 }
 501
 502 int RGWMetadataHandler_GenericMetaBE::mutate(const string& entry,
 503                                              const ceph::real_time& mtime,
 504                                              RGWObjVersionTracker *objv_tracker,
 505                                              optional_yield y,
 506                                              const DoutPrefixProvider *dpp,
 507                                              RGWMDLogStatus op_type,
 508                                              std::function<int()> f)
 509 {
 510   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 511     RGWSI_MetaBackend::MutateParams params(mtime, op_type);
 512     return op->mutate(entry,
 513                       params,
 514                       objv_tracker,
 515                       y,
 516                       f,
 517                       dpp);
 518   });
 519 }
 520
 521 int RGWMetadataHandler_GenericMetaBE::get_shard_id(const string& entry, int *shard_id)
 522 {
 523   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 524     return op->get_shard_id(entry, shard_id);
 525   });
 526 }
 527
 528 int RGWMetadataHandler_GenericMetaBE::list_keys_init(const DoutPrefixProvider *dpp, const string& marker, void **phandle)
 529 {
 530   auto op = std::make_unique<RGWSI_MetaBackend_Handler::Op_ManagedCtx>(be_handler);
 531
 532   int ret = op->list_init(dpp, marker);
 533   if (ret < 0) {
 534     return ret;
 535   }
 536
 537   *phandle = (void *)op.release();
 538
 539   return 0;
 540 }
 541
 542 int RGWMetadataHandler_GenericMetaBE::list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)
 543 {
 544   auto op = static_cast<RGWSI_MetaBackend_Handler::Op_ManagedCtx *>(handle);
 545
 546   int ret = op->list_next(max, &keys, truncated);
 547   if (ret < 0 && ret != -ENOENT) {
 548     return ret;
 549   }
 550   if (ret == -ENOENT) {
 551     if (truncated) {
 552       *truncated = false;
 553     }
 554     return 0;
 555   }
 556
 557   return 0;
 558 }
 559
 560 void RGWMetadataHandler_GenericMetaBE::list_keys_complete(void *handle)
 561 {
 562   auto op = static_cast<RGWSI_MetaBackend_Handler::Op_ManagedCtx *>(handle);
 563   delete op;
 564 }
 565
 566 string RGWMetadataHandler_GenericMetaBE::get_marker(void *handle)
 567 {
 568   auto op = static_cast<RGWSI_MetaBackend_Handler::Op_ManagedCtx *>(handle);
 569   string marker;
 570   int r = op->list_get_marker(&marker);
 571   if (r < 0) {
 572     ldout(cct, 0) << "ERROR: " << __func__ << "(): list_get_marker() returned: r=" << r << dendl;
 573     /* not much else to do */
 574   }
 575
 576   return marker;
 577 }
 578
 579 int RGWMetadataManager::register_handler(RGWMetadataHandler *handler)
 580 {
 581   string type = handler->get_type();
 582
 583   if (handlers.find(type) != handlers.end())
 584     return -EEXIST;
 585
 586   handlers[type] = handler;
 587
 588   return 0;
 589 }
 590
 591 RGWMetadataHandler *RGWMetadataManager::get_handler(const string& type)
 592 {
 593   map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
 594   if (iter == handlers.end())
 595     return NULL;
 596
 597   return iter->second;
 598 }
 599
 600 void RGWMetadataManager::parse_metadata_key(const string& metadata_key, string& type, string& entry)
 601 {
 602   auto pos = metadata_key.find(':');
 603   if (pos == string::npos) {
 604     type = metadata_key;
 605   } else {
 606     type = metadata_key.substr(0, pos);
 607     entry = metadata_key.substr(pos + 1);
 608   }
 609 }
 610
 611 int RGWMetadataManager::find_handler(const string& metadata_key, RGWMetadataHandler **handler, string& entry)
 612 {
 613   string type;
 614
 615   parse_metadata_key(metadata_key, type, entry);
 616
 617   if (type.empty()) {
 618     *handler = md_top_handler.get();
 619     return 0;
 620   }
 621
 622   map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
 623   if (iter == handlers.end())
 624     return -ENOENT;
 625
 626   *handler = iter->second;
 627
 628   return 0;
 629
 630 }
 631
 632 int RGWMetadataManager::get(string& metadata_key, Formatter *f, optional_yield y, const DoutPrefixProvider *dpp)
 633 {
 634   RGWMetadataHandler *handler;
 635   string entry;
 636   int ret = find_handler(metadata_key, &handler, entry);
 637   if (ret < 0) {
 638     return ret;
 639   }
 640
 641   RGWMetadataObject *obj;
 642
 643   ret = handler->get(entry, &obj, y, dpp);
 644   if (ret < 0) {
 645     return ret;
 646   }
 647
 648   f->open_object_section("metadata_info");
 649   encode_json("key", metadata_key, f);
 650   encode_json("ver", obj->get_version(), f);
 651   real_time mtime = obj->get_mtime();
 652   if (!real_clock::is_zero(mtime)) {
 653     utime_t ut(mtime);
 654     encode_json("mtime", ut, f);
 655   }
 656   encode_json("data", *obj, f);
 657   f->close_section();
 658
 659   delete obj;
 660
 661   return 0;
 662 }
 663
 664 int RGWMetadataManager::put(string& metadata_key, bufferlist& bl,
 665                             optional_yield y,
 666                             const DoutPrefixProvider *dpp,
 667                             RGWMDLogSyncType sync_type,
 668                             bool from_remote_zone,
 669                             obj_version *existing_version)
 670 {
 671   RGWMetadataHandler *handler;
 672   string entry;
 673
 674   int ret = find_handler(metadata_key, &handler, entry);
 675   if (ret < 0) {
 676     return ret;
 677   }
 678
 679   JSONParser parser;
 680   if (!parser.parse(bl.c_str(), bl.length())) {
 681     return -EINVAL;
 682   }
 683
 684   RGWObjVersionTracker objv_tracker;
 685
 686   obj_version *objv = &objv_tracker.write_version;
 687
 688   utime_t mtime;
 689
 690   try {
 691     JSONDecoder::decode_json("key", metadata_key, &parser);
 692     JSONDecoder::decode_json("ver", *objv, &parser);
 693     JSONDecoder::decode_json("mtime", mtime, &parser);
 694   } catch (JSONDecoder::err& e) {
 695     return -EINVAL;
 696   }
 697
 698   JSONObj *jo = parser.find_obj("data");
 699   if (!jo) {
 700     return -EINVAL;
 701   }
 702
 703   RGWMetadataObject *obj = handler->get_meta_obj(jo, *objv, mtime.to_real_time());
 704   if (!obj) {
 705     return -EINVAL;
 706   }
 707
 708   ret = handler->put(entry, obj, objv_tracker, y, dpp, sync_type, from_remote_zone);
 709   if (existing_version) {
 710     *existing_version = objv_tracker.read_version;
 711   }
 712
 713   delete obj;
 714
 715   return ret;
 716 }
 717
 718 int RGWMetadataManager::remove(string& metadata_key, optional_yield y, const DoutPrefixProvider *dpp)
 719 {
 720   RGWMetadataHandler *handler;
 721   string entry;
 722
 723   int ret = find_handler(metadata_key, &handler, entry);
 724   if (ret < 0) {
 725     return ret;
 726   }
 727
 728   RGWMetadataObject *obj;
 729   ret = handler->get(entry, &obj, y, dpp);
 730   if (ret < 0) {
 731     return ret;
 732   }
 733   RGWObjVersionTracker objv_tracker;
 734   objv_tracker.read_version = obj->get_version();
 735   delete obj;
 736
 737   return handler->remove(entry, objv_tracker, y, dpp);
 738 }
 739
 740 int RGWMetadataManager::mutate(const string& metadata_key,
 741                                const ceph::real_time& mtime,
 742                                RGWObjVersionTracker *objv_tracker,
 743                                optional_yield y,
 744                                const DoutPrefixProvider *dpp,
 745                                RGWMDLogStatus op_type,
 746                                std::function<int()> f)
 747 {
 748   RGWMetadataHandler *handler;
 749   string entry;
 750
 751   int ret = find_handler(metadata_key, &handler, entry);
 752   if (ret < 0) {
 753     return ret;
 754   }
 755
 756   return handler->mutate(entry, mtime, objv_tracker, y, dpp, op_type, f);
 757 }
 758
 759 int RGWMetadataManager::get_shard_id(const string& section, const string& entry, int *shard_id)
 760 {
 761   RGWMetadataHandler *handler = get_handler(section);
 762   if (!handler) {
 763     return -EINVAL;
 764   }
 765
 766   return handler->get_shard_id(entry, shard_id);
 767 }
 768
 769 struct list_keys_handle {
 770   void *handle;
 771   RGWMetadataHandler *handler;
 772 };
 773
 774 int RGWMetadataManager::list_keys_init(const DoutPrefixProvider *dpp, const string& section, void **handle)
 775 {
 776   return list_keys_init(dpp, section, string(), handle);
 777 }
 778
 779 int RGWMetadataManager::list_keys_init(const DoutPrefixProvider *dpp, const string& section,
 780                                        const string& marker, void **handle)
 781 {
 782   string entry;
 783   RGWMetadataHandler *handler;
 784
 785   int ret;
 786
 787   ret = find_handler(section, &handler, entry);
 788   if (ret < 0) {
 789     return -ENOENT;
 790   }
 791
 792   list_keys_handle *h = new list_keys_handle;
 793   h->handler = handler;
 794   ret = handler->list_keys_init(dpp, marker, &h->handle);
 795   if (ret < 0) {
 796     delete h;
 797     return ret;
 798   }
 799
 800   *handle = (void *)h;
 801
 802   return 0;
 803 }
 804
 805 int RGWMetadataManager::list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)
 806 {
 807   list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 808
 809   RGWMetadataHandler *handler = h->handler;
 810
 811   return handler->list_keys_next(h->handle, max, keys, truncated);
 812 }
 813
 814 void RGWMetadataManager::list_keys_complete(void *handle)
 815 {
 816   list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 817
 818   RGWMetadataHandler *handler = h->handler;
 819
 820   handler->list_keys_complete(h->handle);
 821   delete h;
 822 }
 823
 824 string RGWMetadataManager::get_marker(void *handle)
 825 {
 826   list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 827
 828   return h->handler->get_marker(h->handle);
 829 }
 830
 831 void RGWMetadataManager::dump_log_entry(cls_log_entry& entry, Formatter *f)
 832 {
 833   f->open_object_section("entry");
 834   f->dump_string("id", entry.id);
 835   f->dump_string("section", entry.section);
 836   f->dump_string("name", entry.name);
 837   entry.timestamp.gmtime_nsec(f->dump_stream("timestamp"));
 838
 839   try {
 840     RGWMetadataLogData log_data;
 841     auto iter = entry.data.cbegin();
 842     decode(log_data, iter);
 843
 844     encode_json("data", log_data, f);
 845   } catch (buffer::error& err) {
 846     lderr(cct) << "failed to decode log entry: " << entry.section << ":" << entry.name<< " ts=" << entry.timestamp << dendl;
 847   }
 848   f->close_section();
 849 }
 850
 851 void RGWMetadataManager::get_sections(list<string>& sections)
 852 {
 853   for (map<string, RGWMetadataHandler *>::iterator iter = handlers.begin(); iter != handlers.end(); ++iter) {
 854     sections.push_back(iter->first);
 855   }
 856 }
 857