+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
- * Ceph - scalable distributed file system
- *
- * This is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License version 2.1, as published by the Free Software
- * Foundation. See file COPYING.
- * Copyright 2013 Inktank
- */
-
-#include "common/ceph_json.h"
-
-#include "rgw_replica_log.h"
-#include "cls/replica_log/cls_replica_log_client.h"
-#include "cls/rgw/cls_rgw_client.h"
-#include "rgw_rados.h"
-
-#define dout_subsys ceph_subsys_rgw
-
-void RGWReplicaBounds::dump(Formatter *f) const
-{
- encode_json("marker", marker, f);
- encode_json("oldest_time", oldest_time, f);
- encode_json("markers", markers, f);
-}
-
-void RGWReplicaBounds::decode_json(JSONObj *obj) {
- JSONDecoder::decode_json("marker", marker, obj);
- JSONDecoder::decode_json("oldest_time", oldest_time, obj);
- JSONDecoder::decode_json("markers", markers, obj);
-}
-
-RGWReplicaLogger::RGWReplicaLogger(RGWRados *_store) :
- cct(_store->cct), store(_store) {}
-
-int RGWReplicaLogger::open_ioctx(librados::IoCtx& ctx, const rgw_pool& pool)
-{
- int r = rgw_init_ioctx(store->get_rados_handle(), pool, ctx, true);
- if (r < 0) {
- lderr(cct) << "ERROR: could not open rados pool " << pool << dendl;
- }
- return r;
-}
-
-int RGWReplicaLogger::update_bound(const string& oid, const rgw_pool& pool,
- const string& daemon_id,
- const string& marker, const utime_t& time,
- const list<RGWReplicaItemMarker> *entries,
- bool need_to_exist)
-{
- cls_replica_log_progress_marker progress;
- progress.entity_id = daemon_id;
- progress.position_marker = marker;
- progress.position_time = time;
- progress.items = *entries;
-
- librados::IoCtx ioctx;
- int r = open_ioctx(ioctx, pool);
- if (r < 0) {
- return r;
- }
-
- librados::ObjectWriteOperation opw;
- if (need_to_exist) {
- opw.assert_exists();
- }
- cls_replica_log_update_bound(opw, progress);
- return ioctx.operate(oid, &opw);
-}
-
-int RGWReplicaLogger::write_bounds(const string& oid, const rgw_pool& pool,
- RGWReplicaBounds& bounds)
-{
- librados::IoCtx ioctx;
- int r = open_ioctx(ioctx, pool);
- if (r < 0) {
- return r;
- }
-
- librados::ObjectWriteOperation opw;
- list<RGWReplicaProgressMarker>::iterator iter = bounds.markers.begin();
- for (; iter != bounds.markers.end(); ++iter) {
- RGWReplicaProgressMarker& progress = *iter;
- cls_replica_log_update_bound(opw, progress);
- }
-
- r = ioctx.operate(oid, &opw);
- if (r < 0) {
- return r;
- }
-
- return 0;
-}
-
-int RGWReplicaLogger::delete_bound(const string& oid, const rgw_pool& pool,
- const string& daemon_id, bool purge_all,
- bool need_to_exist)
-{
- librados::IoCtx ioctx;
- int r = open_ioctx(ioctx, pool);
- if (r < 0) {
- return r;
- }
-
- librados::ObjectWriteOperation opw;
- if (need_to_exist) {
- opw.assert_exists();
- }
- if (purge_all) {
- opw.remove();
- } else {
- cls_replica_log_delete_bound(opw, daemon_id);
- }
- return ioctx.operate(oid, &opw);
-}
-
-int RGWReplicaLogger::get_bounds(const string& oid, const rgw_pool& pool,
- RGWReplicaBounds& bounds)
-{
- librados::IoCtx ioctx;
- int r = open_ioctx(ioctx, pool);
- if (r < 0) {
- return r;
- }
-
- return cls_replica_log_get_bounds(ioctx, oid, bounds.marker, bounds.oldest_time, bounds.markers);
-}
-
-RGWReplicaObjectLogger::
-RGWReplicaObjectLogger(RGWRados *_store,
- const rgw_pool& _pool,
- const string& _prefix) : RGWReplicaLogger(_store),
- pool(_pool), prefix(_prefix) {
- if (pool.empty())
- store->get_log_pool(pool);
-}
-
-int RGWReplicaObjectLogger::create_log_objects(int shards)
-{
- librados::IoCtx ioctx;
- int r = open_ioctx(ioctx, pool);
- if (r < 0) {
- return r;
- }
- for (int i = 0; i < shards; ++i) {
- string oid;
- get_shard_oid(i, oid);
- r = ioctx.create(oid, false);
- if (r < 0)
- return r;
- }
- return r;
-}
-
-RGWReplicaBucketLogger::RGWReplicaBucketLogger(RGWRados *_store) :
- RGWReplicaLogger(_store)
-{
- store->get_log_pool(pool);
- prefix = _store->ctx()->_conf->rgw_replica_log_obj_prefix;
- prefix.append(".");
-}
-
-string RGWReplicaBucketLogger::obj_name(const rgw_bucket& bucket, int shard_id, bool index_by_instance)
-{
- string s;
-
- if (index_by_instance) {
- s = prefix + bucket.name + ":" + bucket.bucket_id;
- } else {
- s = prefix + bucket.name;
- }
-
- if (shard_id >= 0) {
- char buf[16];
- snprintf(buf, sizeof(buf), ".%d", shard_id);
- s += buf;
- }
- return s;
-}
-
-int RGWReplicaBucketLogger::update_bound(const rgw_bucket& bucket, int shard_id, const string& daemon_id,
- const string& marker, const utime_t& time,
- const list<RGWReplicaItemMarker> *entries)
-{
- if (shard_id >= 0 ||
- !BucketIndexShardsManager::is_shards_marker(marker)) {
- return RGWReplicaLogger::update_bound(obj_name(bucket, shard_id, true), pool,
- daemon_id, marker, time, entries,
- false);
- }
-
- BucketIndexShardsManager sm;
- int ret = sm.from_string(marker, shard_id);
- if (ret < 0) {
- ldout(cct, 0) << "ERROR: could not parse shards marker: " << marker << dendl;
- return ret;
- }
-
- map<int, string>& vals = sm.get();
-
- ret = 0;
-
- map<int, string>::iterator iter;
- for (iter = vals.begin(); iter != vals.end(); ++iter) {
- ldout(cct, 20) << "updating bound: bucket=" << bucket << " shard=" << iter->first << " marker=" << marker << dendl;
- int r = RGWReplicaLogger::update_bound(obj_name(bucket, iter->first, true), pool,
- daemon_id, iter->second, time, entries,
- true /* need to exist */);
-
- if (r == -ENOENT) {
- RGWReplicaBounds bounds;
- r = convert_old_bounds(bucket, -1, bounds);
- if (r < 0 && r != -ENOENT) {
- return r;
- }
- r = RGWReplicaLogger::update_bound(obj_name(bucket, iter->first, true), pool,
- daemon_id, marker, time, entries, false);
- }
- if (r < 0) {
- ldout(cct, 0) << "failed to update bound: bucket=" << bucket << " shard=" << iter->first << " marker=" << marker << dendl;
- ret = r;
- }
- }
-
- return ret;
-}
-
-int RGWReplicaBucketLogger::delete_bound(const rgw_bucket& bucket, int shard_id, const string& daemon_id, bool purge_all)
-{
- int r = RGWReplicaLogger::delete_bound(obj_name(bucket, shard_id, true), pool, daemon_id, purge_all, true /* need to exist */);
- if (r != -ENOENT) {
- return r;
- }
- /*
- * can only get here if need_to_exist == true,
- * entry is not found, let's convert old entry if exists
- */
- RGWReplicaBounds bounds;
- r = convert_old_bounds(bucket, shard_id, bounds);
- if (r < 0 && r != -ENOENT) {
- return r;
- }
- return RGWReplicaLogger::delete_bound(obj_name(bucket, shard_id, true), pool, daemon_id, purge_all, false);
-}
-
-int RGWReplicaBucketLogger::get_bounds(const rgw_bucket& bucket, int shard_id, RGWReplicaBounds& bounds) {
- int r = RGWReplicaLogger::get_bounds(obj_name(bucket, shard_id, true), pool, bounds);
- if (r != -ENOENT) {
- return r;
- }
-
- r = convert_old_bounds(bucket, shard_id, bounds);
- if (r < 0) {
- return r;
- }
-
- return RGWReplicaLogger::get_bounds(obj_name(bucket, shard_id, true), pool, bounds);
-}
-
-int RGWReplicaBucketLogger::convert_old_bounds(const rgw_bucket& bucket, int shard_id, RGWReplicaBounds& bounds) {
- string old_key = obj_name(bucket, shard_id, false);
- string new_key = obj_name(bucket, shard_id, true);
-
- /* couldn't find when indexed by instance, retry with old key by bucket name only */
- int r = RGWReplicaLogger::get_bounds(old_key, pool, bounds);
- if (r < 0) {
- return r;
- }
- /* convert to new keys */
- r = RGWReplicaLogger::write_bounds(new_key, pool, bounds);
- if (r < 0) {
- return r;
- }
-
- string daemon_id;
- r = RGWReplicaLogger::delete_bound(old_key, pool, daemon_id, true, false); /* purge all */
- if (r < 0) {
- return r;
- }
- return 0;
-}
Code Review / stor4nfv.git / blobdiff