From: Yehuda Sadeh Date: Sat, 25 Apr 2015 16:37:53 +0000 (-0700) Subject: rgw: initial work of orphan detection tool implementation X-Git-Tag: v0.94.4~49^2~19 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=c1b0e7a985cc9e742dcef4c8c427c57a2f58fa43;p=ceph.git rgw: initial work of orphan detection tool implementation So far doesn't do much, iterate through all objects in a specific pool data, store it in a sharded index. Signed-off-by: Yehuda Sadeh (cherry picked from commit 767fc29aa62a0a4e4dd8a78785576e8f1b44ee5b) --- diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index d523ff1ce2db..45b646e8da82 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -762,7 +762,8 @@ if(${WITH_RADOSGW}) rgw/rgw_main.cc) set(radosgw_admin_srcs - rgw/rgw_admin.cc) + rgw/rgw_admin.cc + rgw/rgw_orphan.cc) add_executable(radosgw ${radosgw_srcs} $) target_link_libraries(radosgw rgw_a librados diff --git a/src/rgw/Makefile.am b/src/rgw/Makefile.am index 316ae7620ce5..7620d73b053d 100644 --- a/src/rgw/Makefile.am +++ b/src/rgw/Makefile.am @@ -100,7 +100,7 @@ radosgw_CFLAGS = -I$(srcdir)/civetweb/include radosgw_LDADD = $(LIBRGW) $(LIBCIVETWEB) $(LIBRGW_DEPS) $(RESOLV_LIBS) $(CEPH_GLOBAL) bin_PROGRAMS += radosgw -radosgw_admin_SOURCES = rgw/rgw_admin.cc +radosgw_admin_SOURCES = rgw/rgw_admin.cc rgw/rgw_orphan.cc radosgw_admin_LDADD = $(LIBRGW) $(LIBRGW_DEPS) $(CEPH_GLOBAL) bin_PROGRAMS += radosgw-admin @@ -141,6 +141,7 @@ noinst_HEADERS += \ rgw/rgw_metadata.h \ rgw/rgw_multi_del.h \ rgw/rgw_op.h \ + rgw/rgw_orphan.h \ rgw/rgw_http_client.h \ rgw/rgw_swift.h \ rgw/rgw_swift_auth.h \ diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc index 5debefb4e572..2736faac3cd2 100644 --- a/src/rgw/rgw_admin.cc +++ b/src/rgw/rgw_admin.cc @@ -31,6 +31,7 @@ using namespace std; #include "rgw_formats.h" #include "rgw_usage.h" #include "rgw_replica_log.h" +#include "rgw_orphan.h" #define dout_subsys ceph_subsys_rgw @@ -232,6 +233,7 @@ enum { OPT_QUOTA_DISABLE, OPT_GC_LIST, OPT_GC_PROCESS, + OPT_ORPHANS_FIND, OPT_REGION_GET, OPT_REGION_LIST, OPT_REGION_SET, @@ -281,6 +283,7 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more) strcmp(cmd, "object") == 0 || strcmp(cmd, "olh") == 0 || strcmp(cmd, "opstate") == 0 || + strcmp(cmd, "orphans") == 0 || strcmp(cmd, "pool") == 0 || strcmp(cmd, "pools") == 0 || strcmp(cmd, "quota") == 0 || @@ -441,6 +444,9 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more) return OPT_GC_LIST; if (strcmp(cmd, "process") == 0) return OPT_GC_PROCESS; + } else if (strcmp(prev_cmd, "orphans") == 0) { + if (strcmp(cmd, "find") == 0) + return OPT_ORPHANS_FIND; } else if (strcmp(prev_cmd, "metadata") == 0) { if (strcmp(cmd, "get") == 0) return OPT_METADATA_GET; @@ -1059,6 +1065,7 @@ int do_check_object_locator(const string& bucket_name, bool fix, bool remove_bad return 0; } + int main(int argc, char **argv) { vector args; @@ -1140,6 +1147,10 @@ int main(int argc, char **argv) BIIndexType bi_index_type = PlainIdx; + string job_id; + int init_search = false; + int num_shards = 0; + std::string val; std::ostringstream errs; string err; @@ -1189,6 +1200,8 @@ int main(int argc, char **argv) cerr << "bad key type: " << key_type_str << std::endl; return usage(); } + } else if (ceph_argparse_witharg(args, i, &val, "--job-id", (char*)NULL)) { + job_id = val; } else if (ceph_argparse_binary_flag(args, i, &gen_access_key, NULL, "--gen-access-key", (char*)NULL)) { // do nothing } else if (ceph_argparse_binary_flag(args, i, &gen_secret_key, NULL, "--gen-secret", (char*)NULL)) { @@ -1238,6 +1251,8 @@ int main(int argc, char **argv) start_date = val; } else if (ceph_argparse_witharg(args, i, &val, "--end-date", "--end-time", (char*)NULL)) { end_date = val; + } else if (ceph_argparse_witharg(args, i, &val, "--num-shards", (char*)NULL)) { + num_shards = atoi(val.c_str()); } else if (ceph_argparse_witharg(args, i, &val, "--shard-id", (char*)NULL)) { shard_id = atoi(val.c_str()); specified_shard_id = true; @@ -1292,6 +1307,8 @@ int main(int argc, char **argv) // do nothing } else if (ceph_argparse_binary_flag(args, i, &include_all, NULL, "--include-all", (char*)NULL)) { // do nothing + } else if (ceph_argparse_binary_flag(args, i, &init_search, NULL, "--init-search", (char*)NULL)) { + // do nothing } else if (ceph_argparse_witharg(args, i, &val, "--caps", (char*)NULL)) { caps = val; } else if (ceph_argparse_witharg(args, i, &val, "-i", "--infile", (char*)NULL)) { @@ -2557,6 +2574,35 @@ next: } } + if (opt_cmd == OPT_ORPHANS_FIND) { + RGWOrphanSearch search(store); + + if (job_id.empty()) { + cerr << "ERROR: --job-id not specified" << std::endl; + return EINVAL; + } + RGWOrphanSearchInfo info, *pinfo = NULL; + if (init_search) { + if (pool_name.empty()) { + cerr << "ERROR: --pool-name not specified" << std::endl; + return EINVAL; + } + info.pool = pool_name; + info.job_name = job_id; + info.num_shards = num_shards; + pinfo = &info; + } + + int ret = search.init(job_id, pinfo); + if (ret < 0) { + return -ret; + } + ret = search.run(); + if (ret < 0) { + return -ret; + } + } + if (opt_cmd == OPT_USER_CHECK) { check_bad_user_bucket_mapping(store, user_id, fix); } diff --git a/src/rgw/rgw_orphan.cc b/src/rgw/rgw_orphan.cc new file mode 100644 index 000000000000..42e05d12d98c --- /dev/null +++ b/src/rgw/rgw_orphan.cc @@ -0,0 +1,273 @@ + + +#include + +using namespace std; + +#include "common/config.h" +#include "common/Formatter.h" +#include "common/errno.h" + +#include "rgw_rados.h" +#include "rgw_orphan.h" + +#define dout_subsys ceph_subsys_rgw + +#define DEFAULT_NUM_SHARDS 10 + +int RGWOrphanStore::read_job(const string& job_name, RGWOrphanSearchState& state) +{ + set keys; + map vals; + keys.insert(job_name); + int r = ioctx.omap_get_vals_by_keys(oid, keys, &vals); + if (r < 0) { + return r; + } + + map::iterator iter = vals.find(job_name); + if (iter == vals.end()) { + return -ENOENT; + } + + try { + bufferlist& bl = iter->second; + ::decode(state, bl); + } catch (buffer::error& err) { + lderr(store->ctx()) << "ERROR: could not decode buffer" << dendl; + return -EIO; + } + + return 0; +} + +int RGWOrphanStore::write_job(const string& job_name, const RGWOrphanSearchState& state) +{ + map vals; + bufferlist bl; + ::encode(state, bl); + vals[job_name] = bl; + int r = ioctx.omap_set(oid, vals); + if (r < 0) { + return r; + } + + return 0; +} + +int RGWOrphanStore::init() +{ + const char *log_pool = store->get_zone_params().log_pool.name.c_str(); + librados::Rados *rados = store->get_rados(); + int r = rados->ioctx_create(log_pool, ioctx); + if (r < 0) { + cerr << "ERROR: failed to open log pool ret=" << r << std::endl; + return r; + } + + + + return 0; +} + +int RGWOrphanStore::store_entries(const string& oid, map entries) +{ + librados::ObjectWriteOperation op; + op.omap_set(entries); + cout << "storing " << entries.size() << " entries at " << oid << std::endl; + int ret = ioctx.operate(oid, &op); + if (ret < 0) { + cerr << "ERROR: " << __func__ << "(" << oid << ") returned ret=" << ret << std::endl; + } + + return 0; +} + +int RGWOrphanSearch::init(const string& job_name, RGWOrphanSearchInfo *info) { + int r = orphan_store.init(); + if (r < 0) { + return r; + } + + if (info) { + search_info = *info; + search_info.job_name = job_name; + search_info.num_shards = (info->num_shards ? info->num_shards : DEFAULT_NUM_SHARDS); + search_state = ORPHAN_SEARCH_INIT; + r = save_state(); + if (r < 0) { + lderr(store->ctx()) << "ERROR: failed to write state ret=" << r << dendl; + return r; + } + } else { + RGWOrphanSearchState state; + r = orphan_store.read_job(job_name, state); + if (r < 0) { + lderr(store->ctx()) << "ERROR: failed to read state ret=" << r << dendl; + return r; + } + + search_info = state.info; + search_state = state.state; + } + + log_objs_prefix = RGW_ORPHAN_LOG_PREFIX + string("."); + log_objs_prefix += job_name; + + for (int i = 0; i < search_info.num_shards; i++) { + char buf[128]; + + snprintf(buf, sizeof(buf), "%s.%d", log_objs_prefix.c_str(), i); + orphan_objs_log[i] = buf; + } + return 0; +} + +int RGWOrphanSearch::log_oids(map >& oids) +{ + map >::iterator miter = oids.begin(); + + list liters; /* a list of iterator pairs for begin and end */ + + for (; miter != oids.end(); ++miter) { + log_iter_info info; + info.oid = orphan_objs_log[miter->first]; + info.cur = miter->second.begin(); + info.end = miter->second.end(); + liters.push_back(info); + } + + list::iterator list_iter; + while (!liters.empty()) { + list_iter = liters.begin(); + + while (list_iter != liters.end()) { + log_iter_info& cur_info = *list_iter; + + list::iterator& cur = cur_info.cur; + list::iterator& end = cur_info.end; + + map entries; +#define MAX_OMAP_SET_ENTRIES 100 + for (int j = 0; cur != end && j != MAX_OMAP_SET_ENTRIES; ++cur, ++j) { + ldout(store->ctx(), 20) << "adding obj: " << *cur << dendl; + entries[*cur] = bufferlist(); + } + + int ret = orphan_store.store_entries(cur_info.oid, entries); + if (ret < 0) { + return ret; + } + list::iterator tmp = list_iter; + ++list_iter; + if (cur == end) { + liters.erase(tmp); + } + } + } + return 0; +} + +int RGWOrphanSearch::build_all_oids_index() +{ + librados::Rados *rados = store->get_rados(); + + librados::IoCtx ioctx; + + int ret = rados->ioctx_create(search_info.pool.c_str(), ioctx); + if (ret < 0) { + lderr(store->ctx()) << __func__ << ": ioctx_create() returned ret=" << ret << dendl; + return ret; + } + + ioctx.set_namespace(librados::all_nspaces); + librados::NObjectIterator i = ioctx.nobjects_begin(); + librados::NObjectIterator i_end = ioctx.nobjects_end(); + + map > oids; + + int count = 0; + + cout << "logging all objects in the pool" << std::endl; + + for (; i != i_end; ++i) { + string nspace = i->get_nspace(); + string oid = i->get_oid(); + string locator = i->get_locator(); + + string name = oid; + if (locator.size()) + name += " (@" + locator + ")"; + + ssize_t pos = oid.find('_'); + if (pos < 0) { + cerr << "ERROR: object does not have a bucket marker: " << oid << std::endl; + } + string obj_marker = oid.substr(0, pos); + + int shard = orphan_shard(oid); + oids[shard].push_back(oid); + +#define COUNT_BEFORE_FLUSH 1000 + if (++count >= COUNT_BEFORE_FLUSH) { + ret = log_oids(oids); + if (ret < 0) { + cerr << __func__ << ": ERROR: log_oids() returned ret=" << ret << std::endl; + return ret; + } + count = 0; + oids.clear(); + } + } + ret = log_oids(oids); + if (ret < 0) { + cerr << __func__ << ": ERROR: log_oids() returned ret=" << ret << std::endl; + return ret; + } + + return 0; +} + +int RGWOrphanSearch::run() +{ + int r; + + switch (search_state) { + + case ORPHAN_SEARCH_INIT: + ldout(store->ctx(), 0) << __func__ << "(): initializing state" << dendl; + search_state = ORPHAN_SEARCH_LSPOOL; + r = save_state(); + if (r < 0) { + lderr(store->ctx()) << __func__ << ": ERROR: failed to save state, ret=" << r << dendl; + return r; + } + // fall through + case ORPHAN_SEARCH_LSPOOL: + ldout(store->ctx(), 0) << __func__ << "(): listing all objects in pool" << dendl; + r = build_all_oids_index(); + if (r < 0) { + lderr(store->ctx()) << __func__ << ": ERROR: build_all_objs_index returnr ret=" << r << dendl; + return r; + } + + search_state = ORPHAN_SEARCH_LSBUCKETS; + r = save_state(); + if (r < 0) { + lderr(store->ctx()) << __func__ << ": ERROR: failed to save state, ret=" << r << dendl; + return r; + } + // fall through + + case ORPHAN_SEARCH_LSBUCKETS: + case ORPHAN_SEARCH_DONE: + break; + + default: + assert(0); + }; + + return 0; +} + + diff --git a/src/rgw/rgw_orphan.h b/src/rgw/rgw_orphan.h new file mode 100644 index 000000000000..82de63574cac --- /dev/null +++ b/src/rgw/rgw_orphan.h @@ -0,0 +1,159 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab +/* + * Ceph - scalable distributed file system + * + * Copyright (C) 2015 Red Hat + * + * 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. + * + */ + +#ifndef CEPH_RGW_ORPHAN_H +#define CEPH_RGW_ORPHAN_H + +#include "common/config.h" +#include "common/Formatter.h" +#include "common/errno.h" + +#include "rgw_rados.h" + +#define dout_subsys ceph_subsys_rgw + +#define RGW_ORPHAN_INDEX_OID "orphan.index" +#define RGW_ORPHAN_LOG_PREFIX "orphan.scan" + + +enum OrphanSearchState { + ORPHAN_SEARCH_UNKNOWN = 0, + ORPHAN_SEARCH_INIT = 1, + ORPHAN_SEARCH_LSPOOL = 2, + ORPHAN_SEARCH_LSBUCKETS = 3, + ORPHAN_SEARCH_DONE = 4, +}; + + +struct RGWOrphanSearchInfo { + string job_name; + string pool; + uint16_t num_shards; + void encode(bufferlist& bl) const { + ENCODE_START(1, 1, bl); + ::encode(job_name, bl); + ::encode(pool, bl); + ::encode(num_shards, bl); + ENCODE_FINISH(bl); + } + + void decode(bufferlist::iterator& bl) { + DECODE_START(1, bl); + ::decode(job_name, bl); + ::decode(pool, bl); + ::decode(num_shards, bl); + DECODE_FINISH(bl); + } + + void dump(Formatter *f) const; +}; +WRITE_CLASS_ENCODER(RGWOrphanSearchInfo) + +struct RGWOrphanSearchState { + RGWOrphanSearchInfo info; + OrphanSearchState state; + bufferlist state_info; + + RGWOrphanSearchState() : state(ORPHAN_SEARCH_UNKNOWN) {} + + void encode(bufferlist& bl) const { + ENCODE_START(1, 1, bl); + ::encode(info, bl); + ::encode((int)state, bl); + ::encode(state_info, bl); + ENCODE_FINISH(bl); + } + + void decode(bufferlist::iterator& bl) { + DECODE_START(1, bl); + ::decode(info, bl); + int s; + ::decode(s, bl); + state = (OrphanSearchState)s; + ::decode(state_info, bl); + DECODE_FINISH(bl); + } + + void dump(Formatter *f) const; +}; +WRITE_CLASS_ENCODER(RGWOrphanSearchState) + +class RGWOrphanStore { + RGWRados *store; + librados::IoCtx ioctx; + + string oid; + +public: + RGWOrphanStore(RGWRados *_store) : store(_store) { + oid = RGW_ORPHAN_INDEX_OID; + } + + int init(); + + int read_job(const string& job_name, RGWOrphanSearchState& state); + int write_job(const string& job_name, const RGWOrphanSearchState& state); + + + int store_entries(const string& oid, map entries); +}; + + +class RGWOrphanSearch { + RGWRados *store; + librados::IoCtx log_ioctx; + + RGWOrphanStore orphan_store; + + RGWOrphanSearchInfo search_info; + OrphanSearchState search_state; + + map orphan_objs_log; + + string log_objs_prefix; + + struct log_iter_info { + string oid; + list::iterator cur; + list::iterator end; + }; + + int log_oids(map >& oids); + +#define RGW_ORPHANSEARCH_HASH_PRIME 7877 + int orphan_shard(const string& str) { + return ceph_str_hash_linux(str.c_str(), str.size()) % RGW_ORPHANSEARCH_HASH_PRIME % search_info.num_shards; + } + +public: + RGWOrphanSearch(RGWRados *_store) : store(_store), orphan_store(store) {} + + int save_state() { + RGWOrphanSearchState state; + state.info = search_info; + state.state = search_state; + return orphan_store.write_job(search_info.job_name, state); + } + + int init(const string& job_name, RGWOrphanSearchInfo *info); + + int create(const string& job_name, int num_shards); + + int build_all_oids_index(); + int run(); +}; + + + +#endif diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h index 07839be222d7..c5bd3933fefb 100644 --- a/src/rgw/rgw_rados.h +++ b/src/rgw/rgw_rados.h @@ -1271,6 +1271,8 @@ public: rest_master_conn(NULL), meta_mgr(NULL), data_log(NULL) {} + librados::Rados *get_rados() { return rados; } + uint64_t get_new_req_id() { return max_req_id.inc(); } @@ -1295,6 +1297,8 @@ public: map zone_conn_map; map region_conn_map; + RGWZoneParams& get_zone_params() { return zone; } + RGWMetadataManager *meta_mgr; RGWDataChangesLog *data_log;