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