1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
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"
11 #include "rgw_rados.h"
12 #include "rgw_tools.h"
14 #include "rgw_cr_rados.h"
16 #include "include/assert.h"
17 #include <boost/asio/yield.hpp>
19 #define dout_subsys ceph_subsys_rgw
21 void LogStatusDump::dump(Formatter *f) const {
24 case MDLOG_STATUS_WRITE:
27 case MDLOG_STATUS_SETATTRS:
30 case MDLOG_STATUS_REMOVE:
33 case MDLOG_STATUS_COMPLETE:
36 case MDLOG_STATUS_ABORT:
43 encode_json("status", s, f);
46 void RGWMetadataLogData::encode(bufferlist& bl) const {
47 ENCODE_START(1, 1, bl);
48 ::encode(read_version, bl);
49 ::encode(write_version, bl);
50 uint32_t s = (uint32_t)status;
55 void RGWMetadataLogData::decode(bufferlist::iterator& bl) {
57 ::decode(read_version, bl);
58 ::decode(write_version, bl);
61 status = (RGWMDLogStatus)s;
65 void RGWMetadataLogData::dump(Formatter *f) const {
66 encode_json("read_version", read_version, f);
67 encode_json("write_version", write_version, f);
68 encode_json("status", LogStatusDump(status), f);
71 void decode_json_obj(RGWMDLogStatus& status, JSONObj *obj) {
73 JSONDecoder::decode_json("status", s, obj);
74 if (s == "complete") {
75 status = MDLOG_STATUS_COMPLETE;
76 } else if (s == "write") {
77 status = MDLOG_STATUS_WRITE;
78 } else if (s == "remove") {
79 status = MDLOG_STATUS_REMOVE;
80 } else if (s == "set_attrs") {
81 status = MDLOG_STATUS_SETATTRS;
82 } else if (s == "abort") {
83 status = MDLOG_STATUS_ABORT;
85 status = MDLOG_STATUS_UNKNOWN;
89 void RGWMetadataLogData::decode_json(JSONObj *obj) {
90 JSONDecoder::decode_json("read_version", read_version, obj);
91 JSONDecoder::decode_json("write_version", write_version, obj);
92 JSONDecoder::decode_json("status", status, obj);
96 int RGWMetadataLog::add_entry(RGWMetadataHandler *handler, const string& section, const string& key, bufferlist& bl) {
97 if (!store->need_to_log_metadata())
103 handler->get_hash_key(section, key, hash_key);
106 store->shard_name(prefix, cct->_conf->rgw_md_log_max_shards, hash_key, oid, &shard_id);
107 mark_modified(shard_id);
108 real_time now = real_clock::now();
109 return store->time_log_add(oid, now, section, key, bl);
112 int RGWMetadataLog::store_entries_in_shard(list<cls_log_entry>& entries, int shard_id, librados::AioCompletion *completion)
116 mark_modified(shard_id);
117 store->shard_name(prefix, shard_id, oid);
118 return store->time_log_add(oid, entries, completion, false);
121 void RGWMetadataLog::init_list_entries(int shard_id, const real_time& from_time, const real_time& end_time,
122 string& marker, void **handle)
124 LogListCtx *ctx = new LogListCtx();
126 ctx->cur_shard = shard_id;
127 ctx->from_time = from_time;
128 ctx->end_time = end_time;
129 ctx->marker = marker;
131 get_shard_oid(ctx->cur_shard, ctx->cur_oid);
133 *handle = (void *)ctx;
136 void RGWMetadataLog::complete_list_entries(void *handle) {
137 LogListCtx *ctx = static_cast<LogListCtx *>(handle);
141 int RGWMetadataLog::list_entries(void *handle,
143 list<cls_log_entry>& entries,
146 LogListCtx *ctx = static_cast<LogListCtx *>(handle);
153 std::string next_marker;
154 int ret = store->time_log_list(ctx->cur_oid, ctx->from_time, ctx->end_time,
155 max_entries, entries, ctx->marker,
156 &next_marker, truncated);
157 if ((ret < 0) && (ret != -ENOENT))
160 ctx->marker = std::move(next_marker);
162 *last_marker = ctx->marker;
171 int RGWMetadataLog::get_info(int shard_id, RGWMetadataLogInfo *info)
174 get_shard_oid(shard_id, oid);
176 cls_log_header header;
178 int ret = store->time_log_info(oid, &header);
179 if ((ret < 0) && (ret != -ENOENT))
182 info->marker = header.max_marker;
183 info->last_update = header.max_time.to_real_time();
188 static void _mdlog_info_completion(librados::completion_t cb, void *arg)
190 auto infoc = static_cast<RGWMetadataLogInfoCompletion *>(arg);
192 infoc->put(); // drop the ref from get_info_async()
195 RGWMetadataLogInfoCompletion::RGWMetadataLogInfoCompletion(info_callback_t cb)
196 : completion(librados::Rados::aio_create_completion((void *)this, nullptr,
197 _mdlog_info_completion)),
202 RGWMetadataLogInfoCompletion::~RGWMetadataLogInfoCompletion()
204 completion->release();
207 int RGWMetadataLog::get_info_async(int shard_id, RGWMetadataLogInfoCompletion *completion)
210 get_shard_oid(shard_id, oid);
212 completion->get(); // hold a ref until the completion fires
214 return store->time_log_info_async(completion->get_io_ctx(), oid,
215 &completion->get_header(),
216 completion->get_completion());
219 int RGWMetadataLog::trim(int shard_id, const real_time& from_time, const real_time& end_time,
220 const string& start_marker, const string& end_marker)
223 get_shard_oid(shard_id, oid);
227 ret = store->time_log_trim(oid, from_time, end_time, start_marker, end_marker);
229 if (ret == -ENOENT || ret == -ENODATA)
235 int RGWMetadataLog::lock_exclusive(int shard_id, timespan duration, string& zone_id, string& owner_id) {
237 get_shard_oid(shard_id, oid);
239 return store->lock_exclusive(store->get_zone_params().log_pool, oid, duration, zone_id, owner_id);
242 int RGWMetadataLog::unlock(int shard_id, string& zone_id, string& owner_id) {
244 get_shard_oid(shard_id, oid);
246 return store->unlock(store->get_zone_params().log_pool, oid, zone_id, owner_id);
249 void RGWMetadataLog::mark_modified(int shard_id)
252 if (modified_shards.find(shard_id) != modified_shards.end()) {
258 RWLock::WLocker wl(lock);
259 modified_shards.insert(shard_id);
262 void RGWMetadataLog::read_clear_modified(set<int> &modified)
264 RWLock::WLocker wl(lock);
265 modified.swap(modified_shards);
266 modified_shards.clear();
269 obj_version& RGWMetadataObject::get_version()
274 class RGWMetadataTopHandler : public RGWMetadataHandler {
276 set<string> sections;
277 set<string>::iterator iter;
281 RGWMetadataTopHandler() {}
283 string get_type() override { return string(); }
285 int get(RGWRados *store, string& entry, RGWMetadataObject **obj) override { return -ENOTSUP; }
286 int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
287 real_time mtime, JSONObj *obj, sync_type_t sync_type) override { return -ENOTSUP; }
289 virtual void get_pool_and_oid(RGWRados *store, const string& key, rgw_pool& pool, string& oid) override {}
291 int remove(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker) override { return -ENOTSUP; }
293 int list_keys_init(RGWRados *store, const string& marker, void **phandle) override {
294 iter_data *data = new iter_data;
295 list<string> sections;
296 store->meta_mgr->get_sections(sections);
297 for (auto& s : sections) {
298 data->sections.insert(s);
300 data->iter = data->sections.lower_bound(marker);
306 int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) override {
307 iter_data *data = static_cast<iter_data *>(handle);
308 for (int i = 0; i < max && data->iter != data->sections.end(); ++i, ++(data->iter)) {
309 keys.push_back(*data->iter);
312 *truncated = (data->iter != data->sections.end());
316 void list_keys_complete(void *handle) override {
317 iter_data *data = static_cast<iter_data *>(handle);
322 virtual string get_marker(void *handle) {
323 iter_data *data = static_cast<iter_data *>(handle);
325 if (data->iter != data->sections.end()) {
326 return *(data->iter);
333 static RGWMetadataTopHandler md_top_handler;
336 RGWMetadataManager::RGWMetadataManager(CephContext *_cct, RGWRados *_store)
337 : cct(_cct), store(_store)
341 RGWMetadataManager::~RGWMetadataManager()
343 map<string, RGWMetadataHandler *>::iterator iter;
345 for (iter = handlers.begin(); iter != handlers.end(); ++iter) {
352 const std::string RGWMetadataLogHistory::oid = "meta.history";
356 int read_history(RGWRados *store, RGWMetadataLogHistory *state,
357 RGWObjVersionTracker *objv_tracker)
359 RGWObjectCtx ctx{store};
360 auto& pool = store->get_zone_params().log_pool;
361 const auto& oid = RGWMetadataLogHistory::oid;
363 int ret = rgw_get_system_obj(store, ctx, pool, oid, bl, objv_tracker, nullptr);
370 } catch (buffer::error& e) {
371 ldout(store->ctx(), 1) << "failed to decode the mdlog history: "
372 << e.what() << dendl;
378 int write_history(RGWRados *store, const RGWMetadataLogHistory& state,
379 RGWObjVersionTracker *objv_tracker, bool exclusive = false)
384 auto& pool = store->get_zone_params().log_pool;
385 const auto& oid = RGWMetadataLogHistory::oid;
386 return rgw_put_system_obj(store, pool, oid, bl.c_str(), bl.length(),
387 exclusive, objv_tracker, real_time{});
390 using Cursor = RGWPeriodHistory::Cursor;
392 /// read the mdlog history and use it to initialize the given cursor
393 class ReadHistoryCR : public RGWCoroutine {
396 RGWObjVersionTracker *objv_tracker;
397 RGWMetadataLogHistory state;
399 ReadHistoryCR(RGWRados *store, Cursor *cursor,
400 RGWObjVersionTracker *objv_tracker)
401 : RGWCoroutine(store->ctx()), store(store), cursor(cursor),
402 objv_tracker(objv_tracker)
408 rgw_raw_obj obj{store->get_zone_params().log_pool,
409 RGWMetadataLogHistory::oid};
410 constexpr bool empty_on_enoent = false;
412 using ReadCR = RGWSimpleRadosReadCR<RGWMetadataLogHistory>;
413 call(new ReadCR(store->get_async_rados(), store, obj,
414 &state, empty_on_enoent, objv_tracker));
417 ldout(cct, 1) << "failed to read mdlog history: "
418 << cpp_strerror(retcode) << dendl;
419 return set_cr_error(retcode);
421 *cursor = store->period_history->lookup(state.oldest_realm_epoch);
423 return set_cr_error(cursor->get_error());
426 ldout(cct, 10) << "read mdlog history with oldest period id="
427 << state.oldest_period_id << " realm_epoch="
428 << state.oldest_realm_epoch << dendl;
429 return set_cr_done();
435 /// write the given cursor to the mdlog history
436 class WriteHistoryCR : public RGWCoroutine {
439 RGWObjVersionTracker *objv;
440 RGWMetadataLogHistory state;
442 WriteHistoryCR(RGWRados *store, const Cursor& cursor,
443 RGWObjVersionTracker *objv)
444 : RGWCoroutine(store->ctx()), store(store), cursor(cursor), objv(objv)
449 state.oldest_period_id = cursor.get_period().get_id();
450 state.oldest_realm_epoch = cursor.get_epoch();
453 rgw_raw_obj obj{store->get_zone_params().log_pool,
454 RGWMetadataLogHistory::oid};
456 using WriteCR = RGWSimpleRadosWriteCR<RGWMetadataLogHistory>;
457 call(new WriteCR(store->get_async_rados(), store, obj, state, objv));
460 ldout(cct, 1) << "failed to write mdlog history: "
461 << cpp_strerror(retcode) << dendl;
462 return set_cr_error(retcode);
465 ldout(cct, 10) << "wrote mdlog history with oldest period id="
466 << state.oldest_period_id << " realm_epoch="
467 << state.oldest_realm_epoch << dendl;
468 return set_cr_done();
474 /// update the mdlog history to reflect trimmed logs
475 class TrimHistoryCR : public RGWCoroutine {
477 const Cursor cursor; //< cursor to trimmed period
478 RGWObjVersionTracker *objv; //< to prevent racing updates
479 Cursor next; //< target cursor for oldest log period
480 Cursor existing; //< existing cursor read from disk
483 TrimHistoryCR(RGWRados *store, Cursor cursor, RGWObjVersionTracker *objv)
484 : RGWCoroutine(store->ctx()),
485 store(store), cursor(cursor), objv(objv), next(cursor)
487 next.next(); // advance past cursor
492 // read an existing history, and write the new history if it's newer
493 yield call(new ReadHistoryCR(store, &existing, objv));
495 return set_cr_error(retcode);
497 // reject older trims with ECANCELED
498 if (cursor.get_epoch() < existing.get_epoch()) {
499 ldout(cct, 4) << "found oldest log epoch=" << existing.get_epoch()
500 << ", rejecting trim at epoch=" << cursor.get_epoch() << dendl;
501 return set_cr_error(-ECANCELED);
503 // overwrite with updated history
504 yield call(new WriteHistoryCR(store, next, objv));
506 return set_cr_error(retcode);
508 return set_cr_done();
514 // traverse all the way back to the beginning of the period history, and
515 // return a cursor to the first period in a fully attached history
516 Cursor find_oldest_period(RGWRados *store)
518 auto cct = store->ctx();
519 auto cursor = store->period_history->get_current();
522 // advance to the period's predecessor
523 if (!cursor.has_prev()) {
524 auto& predecessor = cursor.get_period().get_predecessor();
525 if (predecessor.empty()) {
526 // this is the first period, so our logs must start here
527 ldout(cct, 10) << "find_oldest_period returning first "
528 "period " << cursor.get_period().get_id() << dendl;
531 // pull the predecessor and add it to our history
533 int r = store->period_puller->pull(predecessor, period);
537 auto prev = store->period_history->insert(std::move(period));
541 ldout(cct, 20) << "find_oldest_period advancing to "
542 "predecessor period " << predecessor << dendl;
543 assert(cursor.has_prev());
547 ldout(cct, 10) << "find_oldest_period returning empty cursor" << dendl;
551 } // anonymous namespace
553 Cursor RGWMetadataManager::init_oldest_log_period()
555 // read the mdlog history
556 RGWMetadataLogHistory state;
557 RGWObjVersionTracker objv;
558 int ret = read_history(store, &state, &objv);
560 if (ret == -ENOENT) {
561 // initialize the mdlog history and write it
562 ldout(cct, 10) << "initializing mdlog history" << dendl;
563 auto cursor = find_oldest_period(store);
568 // write the initial history
569 state.oldest_realm_epoch = cursor.get_epoch();
570 state.oldest_period_id = cursor.get_period().get_id();
572 constexpr bool exclusive = true; // don't overwrite
573 int ret = write_history(store, state, &objv, exclusive);
574 if (ret < 0 && ret != -EEXIST) {
575 ldout(cct, 1) << "failed to write mdlog history: "
576 << cpp_strerror(ret) << dendl;
580 } else if (ret < 0) {
581 ldout(cct, 1) << "failed to read mdlog history: "
582 << cpp_strerror(ret) << dendl;
586 // if it's already in the history, return it
587 auto cursor = store->period_history->lookup(state.oldest_realm_epoch);
591 // pull the oldest period by id
593 ret = store->period_puller->pull(state.oldest_period_id, period);
595 ldout(cct, 1) << "failed to read period id=" << state.oldest_period_id
596 << " for mdlog history: " << cpp_strerror(ret) << dendl;
599 // verify its realm_epoch
600 if (period.get_realm_epoch() != state.oldest_realm_epoch) {
601 ldout(cct, 1) << "inconsistent mdlog history: read period id="
602 << period.get_id() << " with realm_epoch=" << period.get_realm_epoch()
603 << ", expected realm_epoch=" << state.oldest_realm_epoch << dendl;
604 return Cursor{-EINVAL};
606 // attach the period to our history
607 return store->period_history->attach(std::move(period));
610 Cursor RGWMetadataManager::read_oldest_log_period() const
612 RGWMetadataLogHistory state;
613 int ret = read_history(store, &state, nullptr);
615 ldout(store->ctx(), 1) << "failed to read mdlog history: "
616 << cpp_strerror(ret) << dendl;
620 ldout(store->ctx(), 10) << "read mdlog history with oldest period id="
621 << state.oldest_period_id << " realm_epoch="
622 << state.oldest_realm_epoch << dendl;
624 return store->period_history->lookup(state.oldest_realm_epoch);
627 RGWCoroutine* RGWMetadataManager::read_oldest_log_period_cr(Cursor *period,
628 RGWObjVersionTracker *objv) const
630 return new ReadHistoryCR(store, period, objv);
633 RGWCoroutine* RGWMetadataManager::trim_log_period_cr(Cursor period,
634 RGWObjVersionTracker *objv) const
636 return new TrimHistoryCR(store, period, objv);
639 int RGWMetadataManager::init(const std::string& current_period)
641 // open a log for the current period
642 current_log = get_log(current_period);
646 RGWMetadataLog* RGWMetadataManager::get_log(const std::string& period)
648 // construct the period's log in place if it doesn't exist
649 auto insert = md_logs.emplace(std::piecewise_construct,
650 std::forward_as_tuple(period),
651 std::forward_as_tuple(cct, store, period));
652 return &insert.first->second;
655 int RGWMetadataManager::register_handler(RGWMetadataHandler *handler)
657 string type = handler->get_type();
659 if (handlers.find(type) != handlers.end())
662 handlers[type] = handler;
667 RGWMetadataHandler *RGWMetadataManager::get_handler(const string& type)
669 map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
670 if (iter == handlers.end())
676 void RGWMetadataManager::parse_metadata_key(const string& metadata_key, string& type, string& entry)
678 auto pos = metadata_key.find(':');
679 if (pos == string::npos) {
682 type = metadata_key.substr(0, pos);
683 entry = metadata_key.substr(pos + 1);
687 int RGWMetadataManager::find_handler(const string& metadata_key, RGWMetadataHandler **handler, string& entry)
691 parse_metadata_key(metadata_key, type, entry);
694 *handler = &md_top_handler;
698 map<string, RGWMetadataHandler *>::iterator iter = handlers.find(type);
699 if (iter == handlers.end())
702 *handler = iter->second;
708 int RGWMetadataManager::get(string& metadata_key, Formatter *f)
710 RGWMetadataHandler *handler;
712 int ret = find_handler(metadata_key, &handler, entry);
717 RGWMetadataObject *obj;
719 ret = handler->get(store, entry, &obj);
724 f->open_object_section("metadata_info");
725 encode_json("key", metadata_key, f);
726 encode_json("ver", obj->get_version(), f);
727 real_time mtime = obj->get_mtime();
728 if (!real_clock::is_zero(mtime)) {
730 encode_json("mtime", ut, f);
732 encode_json("data", *obj, f);
740 int RGWMetadataManager::put(string& metadata_key, bufferlist& bl,
741 RGWMetadataHandler::sync_type_t sync_type,
742 obj_version *existing_version)
744 RGWMetadataHandler *handler;
747 int ret = find_handler(metadata_key, &handler, entry);
752 if (!parser.parse(bl.c_str(), bl.length())) {
756 RGWObjVersionTracker objv_tracker;
758 obj_version *objv = &objv_tracker.write_version;
763 JSONDecoder::decode_json("key", metadata_key, &parser);
764 JSONDecoder::decode_json("ver", *objv, &parser);
765 JSONDecoder::decode_json("mtime", mtime, &parser);
766 } catch (JSONDecoder::err& e) {
770 JSONObj *jo = parser.find_obj("data");
775 ret = handler->put(store, entry, objv_tracker, mtime.to_real_time(), jo, sync_type);
776 if (existing_version) {
777 *existing_version = objv_tracker.read_version;
782 int RGWMetadataManager::remove(string& metadata_key)
784 RGWMetadataHandler *handler;
787 int ret = find_handler(metadata_key, &handler, entry);
791 RGWMetadataObject *obj;
793 ret = handler->get(store, entry, &obj);
798 RGWObjVersionTracker objv_tracker;
800 objv_tracker.read_version = obj->get_version();
804 return handler->remove(store, entry, objv_tracker);
807 int RGWMetadataManager::lock_exclusive(string& metadata_key, timespan duration, string& owner_id) {
808 RGWMetadataHandler *handler;
812 int ret = find_handler(metadata_key, &handler, entry);
819 handler->get_pool_and_oid(store, entry, pool, oid);
821 return store->lock_exclusive(pool, oid, duration, zone_id, owner_id);
824 int RGWMetadataManager::unlock(string& metadata_key, string& owner_id) {
825 librados::IoCtx io_ctx;
826 RGWMetadataHandler *handler;
830 int ret = find_handler(metadata_key, &handler, entry);
837 handler->get_pool_and_oid(store, entry, pool, oid);
839 return store->unlock(pool, oid, zone_id, owner_id);
842 struct list_keys_handle {
844 RGWMetadataHandler *handler;
847 int RGWMetadataManager::list_keys_init(string& section, void **handle)
849 return list_keys_init(section, string(), handle);
852 int RGWMetadataManager::list_keys_init(string& section, const string& marker, void **handle)
855 RGWMetadataHandler *handler;
859 ret = find_handler(section, &handler, entry);
864 list_keys_handle *h = new list_keys_handle;
865 h->handler = handler;
866 ret = handler->list_keys_init(store, marker, &h->handle);
877 int RGWMetadataManager::list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)
879 list_keys_handle *h = static_cast<list_keys_handle *>(handle);
881 RGWMetadataHandler *handler = h->handler;
883 return handler->list_keys_next(h->handle, max, keys, truncated);
886 void RGWMetadataManager::list_keys_complete(void *handle)
888 list_keys_handle *h = static_cast<list_keys_handle *>(handle);
890 RGWMetadataHandler *handler = h->handler;
892 handler->list_keys_complete(h->handle);
896 string RGWMetadataManager::get_marker(void *handle)
898 list_keys_handle *h = static_cast<list_keys_handle *>(handle);
900 return h->handler->get_marker(h->handle);
903 void RGWMetadataManager::dump_log_entry(cls_log_entry& entry, Formatter *f)
905 f->open_object_section("entry");
906 f->dump_string("id", entry.id);
907 f->dump_string("section", entry.section);
908 f->dump_string("name", entry.name);
909 entry.timestamp.gmtime_nsec(f->dump_stream("timestamp"));
912 RGWMetadataLogData log_data;
913 bufferlist::iterator iter = entry.data.begin();
914 ::decode(log_data, iter);
916 encode_json("data", log_data, f);
917 } catch (buffer::error& err) {
918 lderr(cct) << "failed to decode log entry: " << entry.section << ":" << entry.name<< " ts=" << entry.timestamp << dendl;
923 void RGWMetadataManager::get_sections(list<string>& sections)
925 for (map<string, RGWMetadataHandler *>::iterator iter = handlers.begin(); iter != handlers.end(); ++iter) {
926 sections.push_back(iter->first);
930 int RGWMetadataManager::pre_modify(RGWMetadataHandler *handler, string& section, const string& key,
931 RGWMetadataLogData& log_data, RGWObjVersionTracker *objv_tracker,
932 RGWMDLogStatus op_type)
934 section = handler->get_type();
936 /* if write version has not been set, and there's a read version, set it so that we can
940 if (objv_tracker->read_version.ver && !objv_tracker->write_version.ver) {
941 objv_tracker->write_version = objv_tracker->read_version;
942 objv_tracker->write_version.ver++;
944 log_data.read_version = objv_tracker->read_version;
945 log_data.write_version = objv_tracker->write_version;
948 log_data.status = op_type;
951 ::encode(log_data, logbl);
953 assert(current_log); // must have called init()
954 int ret = current_log->add_entry(handler, section, key, logbl);
961 int RGWMetadataManager::post_modify(RGWMetadataHandler *handler, const string& section, const string& key, RGWMetadataLogData& log_data,
962 RGWObjVersionTracker *objv_tracker, int ret)
965 log_data.status = MDLOG_STATUS_COMPLETE;
967 log_data.status = MDLOG_STATUS_ABORT;
970 ::encode(log_data, logbl);
972 assert(current_log); // must have called init()
973 int r = current_log->add_entry(handler, section, key, logbl);
983 string RGWMetadataManager::heap_oid(RGWMetadataHandler *handler, const string& key, const obj_version& objv)
985 char buf[objv.tag.size() + 32];
986 snprintf(buf, sizeof(buf), "%s:%lld", objv.tag.c_str(), (long long)objv.ver);
987 return string(".meta:") + handler->get_type() + ":" + key + ":" + buf;
990 int RGWMetadataManager::store_in_heap(RGWMetadataHandler *handler, const string& key, bufferlist& bl,
991 RGWObjVersionTracker *objv_tracker, real_time mtime,
992 map<string, bufferlist> *pattrs)
998 rgw_pool heap_pool(store->get_zone_params().metadata_heap);
1000 if (heap_pool.empty()) {
1004 RGWObjVersionTracker otracker;
1005 otracker.write_version = objv_tracker->write_version;
1006 string oid = heap_oid(handler, key, objv_tracker->write_version);
1007 int ret = rgw_put_system_obj(store, heap_pool, oid,
1008 bl.c_str(), bl.length(), false,
1009 &otracker, mtime, pattrs);
1011 ldout(store->ctx(), 0) << "ERROR: rgw_put_system_obj() oid=" << oid << ") returned ret=" << ret << dendl;
1018 int RGWMetadataManager::remove_from_heap(RGWMetadataHandler *handler, const string& key, RGWObjVersionTracker *objv_tracker)
1020 if (!objv_tracker) {
1024 rgw_pool heap_pool(store->get_zone_params().metadata_heap);
1026 if (heap_pool.empty()) {
1030 string oid = heap_oid(handler, key, objv_tracker->write_version);
1031 rgw_raw_obj obj(heap_pool, oid);
1032 int ret = store->delete_system_obj(obj);
1034 ldout(store->ctx(), 0) << "ERROR: store->delete_system_obj()=" << oid << ") returned ret=" << ret << dendl;
1041 int RGWMetadataManager::put_entry(RGWMetadataHandler *handler, const string& key, bufferlist& bl, bool exclusive,
1042 RGWObjVersionTracker *objv_tracker, real_time mtime, map<string, bufferlist> *pattrs)
1045 RGWMetadataLogData log_data;
1046 int ret = pre_modify(handler, section, key, log_data, objv_tracker, MDLOG_STATUS_WRITE);
1053 handler->get_pool_and_oid(store, key, pool, oid);
1055 ret = store_in_heap(handler, key, bl, objv_tracker, mtime, pattrs);
1057 ldout(store->ctx(), 0) << "ERROR: " << __func__ << ": store_in_heap() key=" << key << " returned ret=" << ret << dendl;
1061 ret = rgw_put_system_obj(store, pool, oid,
1062 bl.c_str(), bl.length(), exclusive,
1063 objv_tracker, mtime, pattrs);
1066 int r = remove_from_heap(handler, key, objv_tracker);
1068 ldout(store->ctx(), 0) << "ERROR: " << __func__ << ": remove_from_heap() key=" << key << " returned ret=" << r << dendl;
1072 /* cascading ret into post_modify() */
1074 ret = post_modify(handler, section, key, log_data, objv_tracker, ret);
1081 int RGWMetadataManager::remove_entry(RGWMetadataHandler *handler, string& key, RGWObjVersionTracker *objv_tracker)
1084 RGWMetadataLogData log_data;
1085 int ret = pre_modify(handler, section, key, log_data, objv_tracker, MDLOG_STATUS_REMOVE);
1092 handler->get_pool_and_oid(store, key, pool, oid);
1094 rgw_raw_obj obj(pool, oid);
1096 ret = store->delete_system_obj(obj, objv_tracker);
1097 /* cascading ret into post_modify() */
1099 ret = post_modify(handler, section, key, log_data, objv_tracker, ret);
1106 int RGWMetadataManager::get_log_shard_id(const string& section,
1107 const string& key, int *shard_id)
1109 RGWMetadataHandler *handler = get_handler(section);
1114 handler->get_hash_key(section, key, hash_key);
1115 *shard_id = store->key_to_shard_id(hash_key, cct->_conf->rgw_md_log_max_shards);