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 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(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(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(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(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(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(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(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(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(completion->get_io_obj(), oid,
 231                                      &completion->get_header(),
 232                                      completion->get_completion());
 233 }
 234
 235 int RGWMetadataLog::trim(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(oid, from_time, end_time, start_marker,
 242                                end_marker, nullptr, null_yield);
 243 }
 244
 245 int RGWMetadataLog::lock_exclusive(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(svc.zone->get_zone_params().log_pool, oid, duration, zone_id, owner_id);
 250 }
 251
 252 int RGWMetadataLog::unlock(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(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) override {
 310     return -ENOTSUP;
 311   }
 312
 313   int put(string& entry, RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
 314           optional_yield y, RGWMDLogSyncType type, bool from_remote_zone) override {
 315     return -ENOTSUP;
 316   }
 317
 318   int remove(string& entry, RGWObjVersionTracker& objv_tracker, optional_yield y) 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              RGWMDLogStatus op_type,
 327              std::function<int()> f) {
 328     return -ENOTSUP;
 329   }
 330
 331   int list_keys_init(const string& marker, void **phandle) override {
 332     iter_data *data = new iter_data;
 333     list<string> sections;
 334     mgr->get_sections(sections);
 335     for (auto& s : sections) {
 336       data->sections.insert(s);
 337     }
 338     data->iter = data->sections.lower_bound(marker);
 339
 340     *phandle = data;
 341
 342     return 0;
 343   }
 344   int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) override  {
 345     iter_data *data = static_cast<iter_data *>(handle);
 346     for (int i = 0; i < max && data->iter != data->sections.end(); ++i, ++(data->iter)) {
 347       keys.push_back(*data->iter);
 348     }
 349
 350     *truncated = (data->iter != data->sections.end());
 351
 352     return 0;
 353   }
 354   void list_keys_complete(void *handle) override {
 355     iter_data *data = static_cast<iter_data *>(handle);
 356
 357     delete data;
 358   }
 359
 360   virtual string get_marker(void *handle) override {
 361     iter_data *data = static_cast<iter_data *>(handle);
 362
 363     if (data->iter != data->sections.end()) {
 364       return *(data->iter);
 365     }
 366
 367     return string();
 368   }
 369 };
 370
 371 RGWMetadataManager::RGWMetadataManager(RGWSI_Meta *_meta_svc)
 372   : cct(_meta_svc->ctx()), meta_svc(_meta_svc)
 373 {
 374   md_top_handler.reset(new RGWMetadataTopHandler(meta_svc, this));
 375 }
 376
 377 RGWMetadataManager::~RGWMetadataManager()
 378 {
 379 }
 380
 381 int RGWMetadataHandler::attach(RGWMetadataManager *manager)
 382 {
 383   return manager->register_handler(this);
 384 }
 385
 386 RGWMetadataHandler_GenericMetaBE::Put::Put(RGWMetadataHandler_GenericMetaBE *_handler,
 387                                            RGWSI_MetaBackend_Handler::Op *_op,
 388                                            string& _entry, RGWMetadataObject *_obj,
 389                                            RGWObjVersionTracker& _objv_tracker,
 390                                            optional_yield _y,
 391                                            RGWMDLogSyncType _type, bool _from_remote_zone):
 392   handler(_handler), op(_op),
 393   entry(_entry), obj(_obj),
 394   objv_tracker(_objv_tracker),
 395   apply_type(_type),
 396   y(_y),
 397   from_remote_zone(_from_remote_zone)
 398 {
 399 }
 400
 401 RGWMetadataHandlerPut_SObj::RGWMetadataHandlerPut_SObj(RGWMetadataHandler_GenericMetaBE *handler, RGWSI_MetaBackend_Handler::Op *op,
 402                                                        string& entry, RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
 403                                                        optional_yield y,
 404                                                        RGWMDLogSyncType type, bool from_remote_zone) : Put(handler, op, entry, obj, objv_tracker, y, type, from_remote_zone) {
 405 }
 406
 407 RGWMetadataHandlerPut_SObj::~RGWMetadataHandlerPut_SObj() {
 408 }
 409
 410 int RGWMetadataHandlerPut_SObj::put_pre()
 411 {
 412   int ret = get(&old_obj);
 413   if (ret < 0 && ret != -ENOENT) {
 414     return ret;
 415   }
 416   exists = (ret != -ENOENT);
 417
 418   oo.reset(old_obj);
 419
 420   auto old_ver = (!old_obj ? obj_version() : old_obj->get_version());
 421   auto old_mtime = (!old_obj ? ceph::real_time() : old_obj->get_mtime());
 422
 423   // are we actually going to perform this put, or is it too old?
 424   if (!handler->check_versions(exists, old_ver, old_mtime,
 425                                objv_tracker.write_version, obj->get_mtime(),
 426                                apply_type)) {
 427     return STATUS_NO_APPLY;
 428   }
 429
 430   objv_tracker.read_version = old_ver; /* maintain the obj version we just read */
 431
 432   return 0;
 433 }
 434
 435 int RGWMetadataHandlerPut_SObj::put()
 436 {
 437   int ret = put_check();
 438   if (ret != 0) {
 439     return ret;
 440   }
 441
 442   return put_checked();
 443 }
 444
 445 int RGWMetadataHandlerPut_SObj::put_checked()
 446 {
 447   RGWSI_MBSObj_PutParams params(obj->get_pattrs(), obj->get_mtime());
 448
 449   encode_obj(&params.bl);
 450
 451   int ret = op->put(entry, params, &objv_tracker, y);
 452   if (ret < 0) {
 453     return ret;
 454   }
 455
 456   return 0;
 457 }
 458
 459 int RGWMetadataHandler_GenericMetaBE::do_put_operate(Put *put_op)
 460 {
 461   int r = put_op->put_pre();
 462   if (r != 0) { /* r can also be STATUS_NO_APPLY */
 463     return r;
 464   }
 465
 466   r = put_op->put();
 467   if (r != 0) {
 468     return r;
 469   }
 470
 471   r = put_op->put_post();
 472   if (r != 0) {  /* e.g., -error or STATUS_APPLIED */
 473     return r;
 474   }
 475
 476   return 0;
 477 }
 478
 479 int RGWMetadataHandler_GenericMetaBE::get(string& entry, RGWMetadataObject **obj, optional_yield y)
 480 {
 481   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 482     return do_get(op, entry, obj, y);
 483   });
 484 }
 485
 486 int RGWMetadataHandler_GenericMetaBE::put(string& entry, RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
 487                                           optional_yield y, RGWMDLogSyncType type, bool from_remote_zone)
 488 {
 489   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 490     return do_put(op, entry, obj, objv_tracker, y, type, from_remote_zone);
 491   });
 492 }
 493
 494 int RGWMetadataHandler_GenericMetaBE::remove(string& entry, RGWObjVersionTracker& objv_tracker, optional_yield y)
 495 {
 496   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 497     return do_remove(op, entry, objv_tracker, y);
 498   });
 499 }
 500
 501 int RGWMetadataHandler_GenericMetaBE::mutate(const string& entry,
 502                                              const ceph::real_time& mtime,
 503                                              RGWObjVersionTracker *objv_tracker,
 504                                              optional_yield y,
 505                                              RGWMDLogStatus op_type,
 506                                              std::function<int()> f)
 507 {
 508   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 509     RGWSI_MetaBackend::MutateParams params(mtime, op_type);
 510     return op->mutate(entry,
 511                       params,
 512                       objv_tracker,
 513                       y,
 514                       f);
 515   });
 516 }
 517
 518 int RGWMetadataHandler_GenericMetaBE::get_shard_id(const string& entry, int *shard_id)
 519 {
 520   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
 521     return op->get_shard_id(entry, shard_id);
 522   });
 523 }
 524
 525 int RGWMetadataHandler_GenericMetaBE::list_keys_init(const string& marker, void **phandle)
 526 {
 527   auto op = std::make_unique<RGWSI_MetaBackend_Handler::Op_ManagedCtx>(be_handler);
 528
 529   int ret = op->list_init(marker);
 530   if (ret < 0) {
 531     return ret;
 532   }
 533
 534   *phandle = (void *)op.release();
 535
 536   return 0;
 537 }
 538
 539 int RGWMetadataHandler_GenericMetaBE::list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)
 540 {
 541   auto op = static_cast<RGWSI_MetaBackend_Handler::Op_ManagedCtx *>(handle);
 542
 543   int ret = op->list_next(max, &keys, truncated);
 544   if (ret < 0 && ret != -ENOENT) {
 545     return ret;
 546   }
 547   if (ret == -ENOENT) {
 548     if (truncated) {
 549       *truncated = false;
 550     }
 551     return 0;
 552   }
 553
 554   return 0;
 555 }
 556
 557 void RGWMetadataHandler_GenericMetaBE::list_keys_complete(void *handle)
 558 {
 559   auto op = static_cast<RGWSI_MetaBackend_Handler::Op_ManagedCtx *>(handle);
 560   delete op;
 561 }
 562
 563 string RGWMetadataHandler_GenericMetaBE::get_marker(void *handle)
 564 {
 565   auto op = static_cast<RGWSI_MetaBackend_Handler::Op_ManagedCtx *>(handle);
 566   string marker;
 567   int r = op->list_get_marker(&marker);
 568   if (r < 0) {
 569     ldout(cct, 0) << "ERROR: " << __func__ << "(): list_get_marker() returned: r=" << r << dendl;
 570     /* not much else to do */
 571   }
 572
 573   return marker;
 574 }
 575
 576 int RGWMetadataManager::register_handler(RGWMetadataHandler *handler)
 577 {
 578   string type = handler->get_type();
 579
 580   if (handlers.find(type) != handlers.end())
 581     return -EEXIST;
 582
 583   handlers[type] = handler;
 584
 585   return 0;
 586 }
 587
 588 RGWMetadataHandler *RGWMetadataManager::get_handler(const string& type)
 589 {
 590   map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
 591   if (iter == handlers.end())
 592     return NULL;
 593
 594   return iter->second;
 595 }
 596
 597 void RGWMetadataManager::parse_metadata_key(const string& metadata_key, string& type, string& entry)
 598 {
 599   auto pos = metadata_key.find(':');
 600   if (pos == string::npos) {
 601     type = metadata_key;
 602   } else {
 603     type = metadata_key.substr(0, pos);
 604     entry = metadata_key.substr(pos + 1);
 605   }
 606 }
 607
 608 int RGWMetadataManager::find_handler(const string& metadata_key, RGWMetadataHandler **handler, string& entry)
 609 {
 610   string type;
 611
 612   parse_metadata_key(metadata_key, type, entry);
 613
 614   if (type.empty()) {
 615     *handler = md_top_handler.get();
 616     return 0;
 617   }
 618
 619   map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
 620   if (iter == handlers.end())
 621     return -ENOENT;
 622
 623   *handler = iter->second;
 624
 625   return 0;
 626
 627 }
 628
 629 int RGWMetadataManager::get(string& metadata_key, Formatter *f, optional_yield y)
 630 {
 631   RGWMetadataHandler *handler;
 632   string entry;
 633   int ret = find_handler(metadata_key, &handler, entry);
 634   if (ret < 0) {
 635     return ret;
 636   }
 637
 638   RGWMetadataObject *obj;
 639
 640   ret = handler->get(entry, &obj, y);
 641   if (ret < 0) {
 642     return ret;
 643   }
 644
 645   f->open_object_section("metadata_info");
 646   encode_json("key", metadata_key, f);
 647   encode_json("ver", obj->get_version(), f);
 648   real_time mtime = obj->get_mtime();
 649   if (!real_clock::is_zero(mtime)) {
 650     utime_t ut(mtime);
 651     encode_json("mtime", ut, f);
 652   }
 653   encode_json("data", *obj, f);
 654   f->close_section();
 655
 656   delete obj;
 657
 658   return 0;
 659 }
 660
 661 int RGWMetadataManager::put(string& metadata_key, bufferlist& bl,
 662                             optional_yield y,
 663                             RGWMDLogSyncType sync_type,
 664                             bool from_remote_zone,
 665                             obj_version *existing_version)
 666 {
 667   RGWMetadataHandler *handler;
 668   string entry;
 669
 670   int ret = find_handler(metadata_key, &handler, entry);
 671   if (ret < 0) {
 672     return ret;
 673   }
 674
 675   JSONParser parser;
 676   if (!parser.parse(bl.c_str(), bl.length())) {
 677     return -EINVAL;
 678   }
 679
 680   RGWObjVersionTracker objv_tracker;
 681
 682   obj_version *objv = &objv_tracker.write_version;
 683
 684   utime_t mtime;
 685
 686   try {
 687     JSONDecoder::decode_json("key", metadata_key, &parser);
 688     JSONDecoder::decode_json("ver", *objv, &parser);
 689     JSONDecoder::decode_json("mtime", mtime, &parser);
 690   } catch (JSONDecoder::err& e) {
 691     return -EINVAL;
 692   }
 693
 694   JSONObj *jo = parser.find_obj("data");
 695   if (!jo) {
 696     return -EINVAL;
 697   }
 698
 699   RGWMetadataObject *obj = handler->get_meta_obj(jo, *objv, mtime.to_real_time());
 700   if (!obj) {
 701     return -EINVAL;
 702   }
 703
 704   ret = handler->put(entry, obj, objv_tracker, y, sync_type, from_remote_zone);
 705   if (existing_version) {
 706     *existing_version = objv_tracker.read_version;
 707   }
 708
 709   delete obj;
 710
 711   return ret;
 712 }
 713
 714 int RGWMetadataManager::remove(string& metadata_key, optional_yield y)
 715 {
 716   RGWMetadataHandler *handler;
 717   string entry;
 718
 719   int ret = find_handler(metadata_key, &handler, entry);
 720   if (ret < 0) {
 721     return ret;
 722   }
 723
 724   RGWMetadataObject *obj;
 725   ret = handler->get(entry, &obj, y);
 726   if (ret < 0) {
 727     return ret;
 728   }
 729   RGWObjVersionTracker objv_tracker;
 730   objv_tracker.read_version = obj->get_version();
 731   delete obj;
 732
 733   return handler->remove(entry, objv_tracker, y);
 734 }
 735
 736 int RGWMetadataManager::mutate(const string& metadata_key,
 737                                const ceph::real_time& mtime,
 738                                RGWObjVersionTracker *objv_tracker,
 739                                optional_yield y,
 740                                RGWMDLogStatus op_type,
 741                                std::function<int()> f)
 742 {
 743   RGWMetadataHandler *handler;
 744   string entry;
 745
 746   int ret = find_handler(metadata_key, &handler, entry);
 747   if (ret < 0) {
 748     return ret;
 749   }
 750
 751   return handler->mutate(entry, mtime, objv_tracker, y, op_type, f);
 752 }
 753
 754 int RGWMetadataManager::get_shard_id(const string& section, const string& entry, int *shard_id)
 755 {
 756   RGWMetadataHandler *handler = get_handler(section);
 757   if (!handler) {
 758     return -EINVAL;
 759   }
 760
 761   return handler->get_shard_id(entry, shard_id);
 762 }
 763
 764 struct list_keys_handle {
 765   void *handle;
 766   RGWMetadataHandler *handler;
 767 };
 768
 769 int RGWMetadataManager::list_keys_init(const string& section, void **handle)
 770 {
 771   return list_keys_init(section, string(), handle);
 772 }
 773
 774 int RGWMetadataManager::list_keys_init(const string& section,
 775                                        const string& marker, void **handle)
 776 {
 777   string entry;
 778   RGWMetadataHandler *handler;
 779
 780   int ret;
 781
 782   ret = find_handler(section, &handler, entry);
 783   if (ret < 0) {
 784     return -ENOENT;
 785   }
 786
 787   list_keys_handle *h = new list_keys_handle;
 788   h->handler = handler;
 789   ret = handler->list_keys_init(marker, &h->handle);
 790   if (ret < 0) {
 791     delete h;
 792     return ret;
 793   }
 794
 795   *handle = (void *)h;
 796
 797   return 0;
 798 }
 799
 800 int RGWMetadataManager::list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)
 801 {
 802   list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 803
 804   RGWMetadataHandler *handler = h->handler;
 805
 806   return handler->list_keys_next(h->handle, max, keys, truncated);
 807 }
 808
 809 void RGWMetadataManager::list_keys_complete(void *handle)
 810 {
 811   list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 812
 813   RGWMetadataHandler *handler = h->handler;
 814
 815   handler->list_keys_complete(h->handle);
 816   delete h;
 817 }
 818
 819 string RGWMetadataManager::get_marker(void *handle)
 820 {
 821   list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 822
 823   return h->handler->get_marker(h->handle);
 824 }
 825
 826 void RGWMetadataManager::dump_log_entry(cls_log_entry& entry, Formatter *f)
 827 {
 828   f->open_object_section("entry");
 829   f->dump_string("id", entry.id);
 830   f->dump_string("section", entry.section);
 831   f->dump_string("name", entry.name);
 832   entry.timestamp.gmtime_nsec(f->dump_stream("timestamp"));
 833
 834   try {
 835     RGWMetadataLogData log_data;
 836     auto iter = entry.data.cbegin();
 837     decode(log_data, iter);
 838
 839     encode_json("data", log_data, f);
 840   } catch (buffer::error& err) {
 841     lderr(cct) << "failed to decode log entry: " << entry.section << ":" << entry.name<< " ts=" << entry.timestamp << dendl;
 842   }
 843   f->close_section();
 844 }
 845
 846 void RGWMetadataManager::get_sections(list<string>& sections)
 847 {
 848   for (map<string, RGWMetadataHandler *>::iterator iter = handlers.begin(); iter != handlers.end(); ++iter) {
 849     sections.push_back(iter->first);
 850   }
 851 }
 852