diff options
Diffstat (limited to 'src/kv/RocksDBStore.cc')
| -rw-r--r-- | src/kv/RocksDBStore.cc | 3453 |
1 files changed, 3453 insertions, 0 deletions
diff --git a/src/kv/RocksDBStore.cc b/src/kv/RocksDBStore.cc new file mode 100644 index 000000000..8e8983c18 --- /dev/null +++ b/src/kv/RocksDBStore.cc @@ -0,0 +1,3453 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <set> +#include <map> +#include <string> +#include <memory> +#if __has_include(<filesystem>) +#include <filesystem> +namespace fs = std::filesystem; +#elif __has_include(<experimental/filesystem>) +#include <experimental/filesystem> +namespace fs = std::experimental::filesystem; +#endif +#include <errno.h> +#include <unistd.h> +#include <sys/types.h> +#include <sys/stat.h> + +#include "rocksdb/db.h" +#include "rocksdb/table.h" +#include "rocksdb/env.h" +#include "rocksdb/slice.h" +#include "rocksdb/cache.h" +#include "rocksdb/filter_policy.h" +#include "rocksdb/utilities/convenience.h" +#include "rocksdb/merge_operator.h" + +#include "common/perf_counters.h" +#include "common/PriorityCache.h" +#include "include/common_fwd.h" +#include "include/scope_guard.h" +#include "include/str_list.h" +#include "include/stringify.h" +#include "include/str_map.h" +#include "KeyValueDB.h" +#include "RocksDBStore.h" + +#include "common/debug.h" + +#define dout_context cct +#define dout_subsys ceph_subsys_rocksdb +#undef dout_prefix +#define dout_prefix *_dout << "rocksdb: " + +using std::function; +using std::list; +using std::map; +using std::ostream; +using std::pair; +using std::set; +using std::string; +using std::unique_ptr; +using std::vector; + +using ceph::bufferlist; +using ceph::bufferptr; +using ceph::Formatter; + +static const char* sharding_def_dir = "sharding"; +static const char* sharding_def_file = "sharding/def"; +static const char* sharding_recreate = "sharding/recreate_columns"; +static const char* resharding_column_lock = "reshardingXcommencingXlocked"; + +static bufferlist to_bufferlist(rocksdb::Slice in) { + bufferlist bl; + bl.append(bufferptr(in.data(), in.size())); + return bl; +} + +static rocksdb::SliceParts prepare_sliceparts(const bufferlist &bl, + vector<rocksdb::Slice> *slices) +{ + unsigned n = 0; + for (auto& buf : bl.buffers()) { + (*slices)[n].data_ = buf.c_str(); + (*slices)[n].size_ = buf.length(); + n++; + } + return rocksdb::SliceParts(slices->data(), slices->size()); +} + + +// +// One of these for the default rocksdb column family, routing each prefix +// to the appropriate MergeOperator. +// +class RocksDBStore::MergeOperatorRouter + : public rocksdb::AssociativeMergeOperator +{ + RocksDBStore& store; +public: + const char *Name() const override { + // Construct a name that rocksDB will validate against. We want to + // do this in a way that doesn't constrain the ordering of calls + // to set_merge_operator, so sort the merge operators and then + // construct a name from all of those parts. + store.assoc_name.clear(); + map<std::string,std::string> names; + + for (auto& p : store.merge_ops) { + names[p.first] = p.second->name(); + } + for (auto& p : names) { + store.assoc_name += '.'; + store.assoc_name += p.first; + store.assoc_name += ':'; + store.assoc_name += p.second; + } + return store.assoc_name.c_str(); + } + + explicit MergeOperatorRouter(RocksDBStore &_store) : store(_store) {} + + bool Merge(const rocksdb::Slice& key, + const rocksdb::Slice* existing_value, + const rocksdb::Slice& value, + std::string* new_value, + rocksdb::Logger* logger) const override { + // for default column family + // extract prefix from key and compare against each registered merge op; + // even though merge operator for explicit CF is included in merge_ops, + // it won't be picked up, since it won't match. + for (auto& p : store.merge_ops) { + if (p.first.compare(0, p.first.length(), + key.data(), p.first.length()) == 0 && + key.data()[p.first.length()] == 0) { + if (existing_value) { + p.second->merge(existing_value->data(), existing_value->size(), + value.data(), value.size(), + new_value); + } else { + p.second->merge_nonexistent(value.data(), value.size(), new_value); + } + break; + } + } + return true; // OK :) + } +}; + +// +// One of these per non-default column family, linked directly to the +// merge operator for that CF/prefix (if any). +// +class RocksDBStore::MergeOperatorLinker + : public rocksdb::AssociativeMergeOperator +{ +private: + std::shared_ptr<KeyValueDB::MergeOperator> mop; +public: + explicit MergeOperatorLinker(const std::shared_ptr<KeyValueDB::MergeOperator> &o) : mop(o) {} + + const char *Name() const override { + return mop->name(); + } + + bool Merge(const rocksdb::Slice& key, + const rocksdb::Slice* existing_value, + const rocksdb::Slice& value, + std::string* new_value, + rocksdb::Logger* logger) const override { + if (existing_value) { + mop->merge(existing_value->data(), existing_value->size(), + value.data(), value.size(), + new_value); + } else { + mop->merge_nonexistent(value.data(), value.size(), new_value); + } + return true; + } +}; + +int RocksDBStore::set_merge_operator( + const string& prefix, + std::shared_ptr<KeyValueDB::MergeOperator> mop) +{ + // If you fail here, it's because you can't do this on an open database + ceph_assert(db == nullptr); + merge_ops.push_back(std::make_pair(prefix,mop)); + return 0; +} + +class CephRocksdbLogger : public rocksdb::Logger { + CephContext *cct; +public: + explicit CephRocksdbLogger(CephContext *c) : cct(c) { + cct->get(); + } + ~CephRocksdbLogger() override { + cct->put(); + } + + // Write an entry to the log file with the specified format. + void Logv(const char* format, va_list ap) override { + Logv(rocksdb::INFO_LEVEL, format, ap); + } + + // Write an entry to the log file with the specified log level + // and format. Any log with level under the internal log level + // of *this (see @SetInfoLogLevel and @GetInfoLogLevel) will not be + // printed. + void Logv(const rocksdb::InfoLogLevel log_level, const char* format, + va_list ap) override { + int v = rocksdb::NUM_INFO_LOG_LEVELS - log_level - 1; + dout(ceph::dout::need_dynamic(v)); + char buf[65536]; + vsnprintf(buf, sizeof(buf), format, ap); + *_dout << buf << dendl; + } +}; + +rocksdb::Logger *create_rocksdb_ceph_logger() +{ + return new CephRocksdbLogger(g_ceph_context); +} + +static int string2bool(const string &val, bool &b_val) +{ + if (strcasecmp(val.c_str(), "false") == 0) { + b_val = false; + return 0; + } else if (strcasecmp(val.c_str(), "true") == 0) { + b_val = true; + return 0; + } else { + std::string err; + int b = strict_strtol(val.c_str(), 10, &err); + if (!err.empty()) + return -EINVAL; + b_val = !!b; + return 0; + } +} + +namespace rocksdb { +extern std::string trim(const std::string& str); +} + +// this function is a modification of rocksdb's StringToMap: +// 1) accepts ' \n ; as separators +// 2) leaves compound options with enclosing { and } +rocksdb::Status StringToMap(const std::string& opts_str, + std::unordered_map<std::string, std::string>* opts_map) +{ + using rocksdb::Status; + using rocksdb::trim; + assert(opts_map); + // Example: + // opts_str = "write_buffer_size=1024;max_write_buffer_number=2;" + // "nested_opt={opt1=1;opt2=2};max_bytes_for_level_base=100" + size_t pos = 0; + std::string opts = trim(opts_str); + while (pos < opts.size()) { + size_t eq_pos = opts.find('=', pos); + if (eq_pos == std::string::npos) { + return Status::InvalidArgument("Mismatched key value pair, '=' expected"); + } + std::string key = trim(opts.substr(pos, eq_pos - pos)); + if (key.empty()) { + return Status::InvalidArgument("Empty key found"); + } + + // skip space after '=' and look for '{' for possible nested options + pos = eq_pos + 1; + while (pos < opts.size() && isspace(opts[pos])) { + ++pos; + } + // Empty value at the end + if (pos >= opts.size()) { + (*opts_map)[key] = ""; + break; + } + if (opts[pos] == '{') { + int count = 1; + size_t brace_pos = pos + 1; + while (brace_pos < opts.size()) { + if (opts[brace_pos] == '{') { + ++count; + } else if (opts[brace_pos] == '}') { + --count; + if (count == 0) { + break; + } + } + ++brace_pos; + } + // found the matching closing brace + if (count == 0) { + //include both '{' and '}' + (*opts_map)[key] = trim(opts.substr(pos, brace_pos - pos + 1)); + // skip all whitespace and move to the next ';,' + // brace_pos points to the matching '}' + pos = brace_pos + 1; + while (pos < opts.size() && isspace(opts[pos])) { + ++pos; + } + if (pos < opts.size() && opts[pos] != ';' && opts[pos] != ',') { + return Status::InvalidArgument( + "Unexpected chars after nested options"); + } + ++pos; + } else { + return Status::InvalidArgument( + "Mismatched curly braces for nested options"); + } + } else { + size_t sc_pos = opts.find_first_of(",;", pos); + if (sc_pos == std::string::npos) { + (*opts_map)[key] = trim(opts.substr(pos)); + // It either ends with a trailing , ; or the last key-value pair + break; + } else { + (*opts_map)[key] = trim(opts.substr(pos, sc_pos - pos)); + } + pos = sc_pos + 1; + } + } + return Status::OK(); +} + +int RocksDBStore::tryInterpret(const string &key, const string &val, rocksdb::Options &opt) +{ + if (key == "compaction_threads") { + std::string err; + int f = strict_iecstrtoll(val.c_str(), &err); + if (!err.empty()) + return -EINVAL; + //Low priority threadpool is used for compaction + opt.env->SetBackgroundThreads(f, rocksdb::Env::Priority::LOW); + } else if (key == "flusher_threads") { + std::string err; + int f = strict_iecstrtoll(val.c_str(), &err); + if (!err.empty()) + return -EINVAL; + //High priority threadpool is used for flusher + opt.env->SetBackgroundThreads(f, rocksdb::Env::Priority::HIGH); + } else if (key == "compact_on_mount") { + int ret = string2bool(val, compact_on_mount); + if (ret != 0) + return ret; + } else if (key == "disableWAL") { + int ret = string2bool(val, disableWAL); + if (ret != 0) + return ret; + } else { + //unrecognize config options. + return -EINVAL; + } + return 0; +} + +int RocksDBStore::ParseOptionsFromString(const string &opt_str, rocksdb::Options &opt) +{ + return ParseOptionsFromStringStatic(cct, opt_str, opt, + [&](const string& k, const string& v, rocksdb::Options& o) { + return tryInterpret(k, v, o); + } + ); +} + +int RocksDBStore::ParseOptionsFromStringStatic( + CephContext *cct, + const string& opt_str, + rocksdb::Options& opt, + function<int(const string&, const string&, rocksdb::Options&)> interp) +{ + // keep aligned with func tryInterpret + const set<string> need_interp_keys = {"compaction_threads", "flusher_threads", "compact_on_mount", "disableWAL"}; + int r; + rocksdb::Status status; + std::unordered_map<std::string, std::string> str_map; + status = StringToMap(opt_str, &str_map); + if (!status.ok()) { + dout(5) << __func__ << " error '" << status.getState() << + "' while parsing options '" << opt_str << "'" << dendl; + return -EINVAL; + } + + for (auto it = str_map.begin(); it != str_map.end(); ++it) { + string this_opt = it->first + "=" + it->second; + rocksdb::Status status = + rocksdb::GetOptionsFromString(opt, this_opt, &opt); + if (!status.ok()) { + if (interp != nullptr) { + r = interp(it->first, it->second, opt); + } else if (!need_interp_keys.count(it->first)) { + r = -1; + } + if (r < 0) { + derr << status.ToString() << dendl; + return -EINVAL; + } + } + lgeneric_dout(cct, 1) << " set rocksdb option " << it->first + << " = " << it->second << dendl; + } + return 0; +} + +int RocksDBStore::init(string _options_str) +{ + options_str = _options_str; + rocksdb::Options opt; + //try parse options + if (options_str.length()) { + int r = ParseOptionsFromString(options_str, opt); + if (r != 0) { + return -EINVAL; + } + } + return 0; +} + +int RocksDBStore::create_db_dir() +{ + if (env) { + unique_ptr<rocksdb::Directory> dir; + env->NewDirectory(path, &dir); + } else { + if (!fs::exists(path)) { + std::error_code ec; + if (!fs::create_directory(path, ec)) { + derr << __func__ << " failed to create " << path + << ": " << ec.message() << dendl; + return -ec.value(); + } + fs::permissions(path, + fs::perms::owner_all | + fs::perms::group_read | fs::perms::group_exec | + fs::perms::others_read | fs::perms::others_exec); + } + } + return 0; +} + +int RocksDBStore::install_cf_mergeop( + const string &key_prefix, + rocksdb::ColumnFamilyOptions *cf_opt) +{ + ceph_assert(cf_opt != nullptr); + cf_opt->merge_operator.reset(); + for (auto& i : merge_ops) { + if (i.first == key_prefix) { + cf_opt->merge_operator.reset(new MergeOperatorLinker(i.second)); + } + } + return 0; +} + +int RocksDBStore::create_and_open(ostream &out, + const std::string& cfs) +{ + int r = create_db_dir(); + if (r < 0) + return r; + return do_open(out, true, false, cfs); +} + +std::shared_ptr<rocksdb::Cache> RocksDBStore::create_block_cache( + const std::string& cache_type, size_t cache_size, double cache_prio_high) { + std::shared_ptr<rocksdb::Cache> cache; + auto shard_bits = cct->_conf->rocksdb_cache_shard_bits; + if (cache_type == "binned_lru") { + cache = rocksdb_cache::NewBinnedLRUCache(cct, cache_size, shard_bits, false, cache_prio_high); + } else if (cache_type == "lru") { + cache = rocksdb::NewLRUCache(cache_size, shard_bits); + } else if (cache_type == "clock") { + cache = rocksdb::NewClockCache(cache_size, shard_bits); + if (!cache) { + derr << "rocksdb_cache_type '" << cache + << "' chosen, but RocksDB not compiled with LibTBB. " + << dendl; + } + } else { + derr << "unrecognized rocksdb_cache_type '" << cache_type << "'" << dendl; + } + return cache; +} + +int RocksDBStore::load_rocksdb_options(bool create_if_missing, rocksdb::Options& opt) +{ + rocksdb::Status status; + + if (options_str.length()) { + int r = ParseOptionsFromString(options_str, opt); + if (r != 0) { + return -EINVAL; + } + } + + if (cct->_conf->rocksdb_perf) { + dbstats = rocksdb::CreateDBStatistics(); + opt.statistics = dbstats; + } + + opt.create_if_missing = create_if_missing; + if (kv_options.count("separate_wal_dir")) { + opt.wal_dir = path + ".wal"; + } + + // Since ceph::for_each_substr doesn't return a value and + // std::stoull does throw, we may as well just catch everything here. + try { + if (kv_options.count("db_paths")) { + list<string> paths; + get_str_list(kv_options["db_paths"], "; \t", paths); + for (auto& p : paths) { + size_t pos = p.find(','); + if (pos == std::string::npos) { + derr << __func__ << " invalid db path item " << p << " in " + << kv_options["db_paths"] << dendl; + return -EINVAL; + } + string path = p.substr(0, pos); + string size_str = p.substr(pos + 1); + uint64_t size = atoll(size_str.c_str()); + if (!size) { + derr << __func__ << " invalid db path item " << p << " in " + << kv_options["db_paths"] << dendl; + return -EINVAL; + } + opt.db_paths.push_back(rocksdb::DbPath(path, size)); + dout(10) << __func__ << " db_path " << path << " size " << size << dendl; + } + } + } catch (const std::system_error& e) { + return -e.code().value(); + } + + if (cct->_conf->rocksdb_log_to_ceph_log) { + opt.info_log.reset(new CephRocksdbLogger(cct)); + } + + if (priv) { + dout(10) << __func__ << " using custom Env " << priv << dendl; + opt.env = static_cast<rocksdb::Env*>(priv); + } else { + env = opt.env; + } + + opt.env->SetAllowNonOwnerAccess(false); + + // caches + if (!set_cache_flag) { + cache_size = cct->_conf->rocksdb_cache_size; + } + uint64_t row_cache_size = cache_size * cct->_conf->rocksdb_cache_row_ratio; + uint64_t block_cache_size = cache_size - row_cache_size; + + bbt_opts.block_cache = create_block_cache(cct->_conf->rocksdb_cache_type, block_cache_size); + if (!bbt_opts.block_cache) { + return -EINVAL; + } + bbt_opts.block_size = cct->_conf->rocksdb_block_size; + + if (row_cache_size > 0) + opt.row_cache = rocksdb::NewLRUCache(row_cache_size, + cct->_conf->rocksdb_cache_shard_bits); + uint64_t bloom_bits = cct->_conf.get_val<uint64_t>("rocksdb_bloom_bits_per_key"); + if (bloom_bits > 0) { + dout(10) << __func__ << " set bloom filter bits per key to " + << bloom_bits << dendl; + bbt_opts.filter_policy.reset(rocksdb::NewBloomFilterPolicy(bloom_bits)); + } + using std::placeholders::_1; + if (cct->_conf.with_val<std::string>("rocksdb_index_type", + std::bind(std::equal_to<std::string>(), _1, + "binary_search"))) + bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kBinarySearch; + if (cct->_conf.with_val<std::string>("rocksdb_index_type", + std::bind(std::equal_to<std::string>(), _1, + "hash_search"))) + bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kHashSearch; + if (cct->_conf.with_val<std::string>("rocksdb_index_type", + std::bind(std::equal_to<std::string>(), _1, + "two_level"))) + bbt_opts.index_type = rocksdb::BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch; + if (!bbt_opts.no_block_cache) { + bbt_opts.cache_index_and_filter_blocks = + cct->_conf.get_val<bool>("rocksdb_cache_index_and_filter_blocks"); + bbt_opts.cache_index_and_filter_blocks_with_high_priority = + cct->_conf.get_val<bool>("rocksdb_cache_index_and_filter_blocks_with_high_priority"); + bbt_opts.pin_l0_filter_and_index_blocks_in_cache = + cct->_conf.get_val<bool>("rocksdb_pin_l0_filter_and_index_blocks_in_cache"); + } + bbt_opts.partition_filters = cct->_conf.get_val<bool>("rocksdb_partition_filters"); + if (cct->_conf.get_val<Option::size_t>("rocksdb_metadata_block_size") > 0) + bbt_opts.metadata_block_size = cct->_conf.get_val<Option::size_t>("rocksdb_metadata_block_size"); + + opt.table_factory.reset(rocksdb::NewBlockBasedTableFactory(bbt_opts)); + dout(10) << __func__ << " block size " << cct->_conf->rocksdb_block_size + << ", block_cache size " << byte_u_t(block_cache_size) + << ", row_cache size " << byte_u_t(row_cache_size) + << "; shards " + << (1 << cct->_conf->rocksdb_cache_shard_bits) + << ", type " << cct->_conf->rocksdb_cache_type + << dendl; + + opt.merge_operator.reset(new MergeOperatorRouter(*this)); + comparator = opt.comparator; + return 0; +} + +void RocksDBStore::add_column_family(const std::string& cf_name, uint32_t hash_l, uint32_t hash_h, + size_t shard_idx, rocksdb::ColumnFamilyHandle *handle) { + dout(10) << __func__ << " column_name=" << cf_name << " shard_idx=" << shard_idx << + " hash_l=" << hash_l << " hash_h=" << hash_h << " handle=" << (void*) handle << dendl; + bool exists = cf_handles.count(cf_name) > 0; + auto& column = cf_handles[cf_name]; + if (exists) { + ceph_assert(hash_l == column.hash_l); + ceph_assert(hash_h == column.hash_h); + } else { + ceph_assert(hash_l < hash_h); + column.hash_l = hash_l; + column.hash_h = hash_h; + } + if (column.handles.size() <= shard_idx) + column.handles.resize(shard_idx + 1); + column.handles[shard_idx] = handle; + cf_ids_to_prefix.emplace(handle->GetID(), cf_name); +} + +bool RocksDBStore::is_column_family(const std::string& prefix) { + return cf_handles.count(prefix); +} + +std::string_view RocksDBStore::get_key_hash_view(const prefix_shards& shards, const char* key, const size_t keylen) { + uint32_t hash_l = std::min<uint32_t>(shards.hash_l, keylen); + uint32_t hash_h = std::min<uint32_t>(shards.hash_h, keylen); + return { key + hash_l, hash_h - hash_l }; +} + +rocksdb::ColumnFamilyHandle *RocksDBStore::get_key_cf(const prefix_shards& shards, const char* key, const size_t keylen) { + auto sv = get_key_hash_view(shards, key, keylen); + uint32_t hash = ceph_str_hash_rjenkins(sv.data(), sv.size()); + return shards.handles[hash % shards.handles.size()]; +} + +rocksdb::ColumnFamilyHandle *RocksDBStore::get_cf_handle(const std::string& prefix, const std::string& key) { + auto iter = cf_handles.find(prefix); + if (iter == cf_handles.end()) { + return nullptr; + } else { + if (iter->second.handles.size() == 1) { + return iter->second.handles[0]; + } else { + return get_key_cf(iter->second, key.data(), key.size()); + } + } +} + +rocksdb::ColumnFamilyHandle *RocksDBStore::get_cf_handle(const std::string& prefix, const char* key, size_t keylen) { + auto iter = cf_handles.find(prefix); + if (iter == cf_handles.end()) { + return nullptr; + } else { + if (iter->second.handles.size() == 1) { + return iter->second.handles[0]; + } else { + return get_key_cf(iter->second, key, keylen); + } + } +} + +/** + * If the specified IteratorBounds arg has both an upper and a lower bound defined, and they have equal placement hash + * strings, we can be sure that the entire iteration range exists in a single CF. In that case, we return the relevant + * CF handle. In all other cases, we return a nullptr to indicate that the specified bounds cannot necessarily be mapped + * to a single CF. + */ +rocksdb::ColumnFamilyHandle *RocksDBStore::get_cf_handle(const std::string& prefix, const IteratorBounds& bounds) { + if (!bounds.lower_bound || !bounds.upper_bound) { + return nullptr; + } + auto iter = cf_handles.find(prefix); + ceph_assert(iter != cf_handles.end()); + ceph_assert(iter->second.handles.size() != 1); + if (iter->second.hash_l != 0) { + return nullptr; + } + auto lower_bound_hash_str = get_key_hash_view(iter->second, bounds.lower_bound->data(), bounds.lower_bound->size()); + auto upper_bound_hash_str = get_key_hash_view(iter->second, bounds.upper_bound->data(), bounds.upper_bound->size()); + if (lower_bound_hash_str == upper_bound_hash_str) { + auto key = *bounds.lower_bound; + return get_key_cf(iter->second, key.data(), key.size()); + } else { + return nullptr; + } +} + +/** + * Definition of sharding: + * space-separated list of: column_def [ '=' options ] + * column_def := column_name '(' shard_count ')' + * column_def := column_name '(' shard_count ',' hash_begin '-' ')' + * column_def := column_name '(' shard_count ',' hash_begin '-' hash_end ')' + * I=write_buffer_size=1048576 O(6) m(7,10-) prefix(4,0-10)=disable_auto_compactions=true,max_bytes_for_level_base=1048576 + */ +bool RocksDBStore::parse_sharding_def(const std::string_view text_def_in, + std::vector<ColumnFamily>& sharding_def, + char const* *error_position, + std::string *error_msg) +{ + std::string_view text_def = text_def_in; + char const* error_position_local = nullptr; + std::string error_msg_local; + if (error_position == nullptr) { + error_position = &error_position_local; + } + *error_position = nullptr; + if (error_msg == nullptr) { + error_msg = &error_msg_local; + error_msg->clear(); + } + + sharding_def.clear(); + while (!text_def.empty()) { + std::string_view options; + std::string_view name; + size_t shard_cnt = 1; + uint32_t l_bound = 0; + uint32_t h_bound = std::numeric_limits<uint32_t>::max(); + + std::string_view column_def; + size_t spos = text_def.find(' '); + if (spos == std::string_view::npos) { + column_def = text_def; + text_def = std::string_view(text_def.end(), 0); + } else { + column_def = text_def.substr(0, spos); + text_def = text_def.substr(spos + 1); + } + size_t eqpos = column_def.find('='); + if (eqpos != std::string_view::npos) { + options = column_def.substr(eqpos + 1); + column_def = column_def.substr(0, eqpos); + } + + size_t bpos = column_def.find('('); + if (bpos != std::string_view::npos) { + name = column_def.substr(0, bpos); + const char* nptr = &column_def[bpos + 1]; + char* endptr; + shard_cnt = strtol(nptr, &endptr, 10); + if (nptr == endptr) { + *error_position = nptr; + *error_msg = "expecting integer"; + break; + } + nptr = endptr; + if (*nptr == ',') { + nptr++; + l_bound = strtol(nptr, &endptr, 10); + if (nptr == endptr) { + *error_position = nptr; + *error_msg = "expecting integer"; + break; + } + nptr = endptr; + if (*nptr != '-') { + *error_position = nptr; + *error_msg = "expecting '-'"; + break; + } + nptr++; + h_bound = strtol(nptr, &endptr, 10); + if (nptr == endptr) { + h_bound = std::numeric_limits<uint32_t>::max(); + } + nptr = endptr; + } + if (*nptr != ')') { + *error_position = nptr; + *error_msg = "expecting ')'"; + break; + } + } else { + name = column_def; + } + sharding_def.emplace_back(std::string(name), shard_cnt, std::string(options), l_bound, h_bound); + } + return *error_position == nullptr; +} + +void RocksDBStore::sharding_def_to_columns(const std::vector<ColumnFamily>& sharding_def, + std::vector<std::string>& columns) +{ + columns.clear(); + for (size_t i = 0; i < sharding_def.size(); i++) { + if (sharding_def[i].shard_cnt == 1) { + columns.push_back(sharding_def[i].name); + } else { + for (size_t j = 0; j < sharding_def[i].shard_cnt; j++) { + columns.push_back(sharding_def[i].name + "-" + to_string(j)); + } + } + } +} + +int RocksDBStore::create_shards(const rocksdb::Options& opt, + const std::vector<ColumnFamily>& sharding_def) +{ + for (auto& p : sharding_def) { + // copy default CF settings, block cache, merge operators as + // the base for new CF + rocksdb::ColumnFamilyOptions cf_opt(opt); + rocksdb::Status status; + // apply options to column family + int r = update_column_family_options(p.name, p.options, &cf_opt); + if (r != 0) { + return r; + } + for (size_t idx = 0; idx < p.shard_cnt; idx++) { + std::string cf_name; + if (p.shard_cnt == 1) + cf_name = p.name; + else + cf_name = p.name + "-" + to_string(idx); + rocksdb::ColumnFamilyHandle *cf; + status = db->CreateColumnFamily(cf_opt, cf_name, &cf); + if (!status.ok()) { + derr << __func__ << " Failed to create rocksdb column family: " + << cf_name << dendl; + return -EINVAL; + } + // store the new CF handle + add_column_family(p.name, p.hash_l, p.hash_h, idx, cf); + } + } + return 0; +} + +int RocksDBStore::apply_sharding(const rocksdb::Options& opt, + const std::string& sharding_text) +{ + // create and open column families + if (!sharding_text.empty()) { + bool b; + int r; + rocksdb::Status status; + std::vector<ColumnFamily> sharding_def; + char const* error_position; + std::string error_msg; + b = parse_sharding_def(sharding_text, sharding_def, &error_position, &error_msg); + if (!b) { + dout(1) << __func__ << " bad sharding: " << dendl; + dout(1) << __func__ << sharding_text << dendl; + dout(1) << __func__ << std::string(error_position - &sharding_text[0], ' ') << "^" << error_msg << dendl; + return -EINVAL; + } + r = create_shards(opt, sharding_def); + if (r != 0 ) { + derr << __func__ << " create_shards failed error=" << r << dendl; + return r; + } + opt.env->CreateDir(sharding_def_dir); + status = rocksdb::WriteStringToFile(opt.env, sharding_text, + sharding_def_file, true); + if (!status.ok()) { + derr << __func__ << " cannot write to " << sharding_def_file << dendl; + return -EIO; + } + } else { + opt.env->DeleteFile(sharding_def_file); + } + return 0; +} + +// linking to rocksdb function defined in options_helper.cc +// it can parse nested params like "nested_opt={opt1=1;opt2=2}" +extern rocksdb::Status rocksdb::StringToMap(const std::string& opts_str, + std::unordered_map<std::string, std::string>* opts_map); + +// Splits column family options from single string into name->value column_opts_map. +// The split is done using RocksDB parser that understands "{" and "}", so it +// properly extracts compound options. +// If non-RocksDB option "block_cache" is defined it is extracted to block_cache_opt. +int RocksDBStore::split_column_family_options(const std::string& options, + std::unordered_map<std::string, std::string>* opt_map, + std::string* block_cache_opt) +{ + dout(20) << __func__ << " options=" << options << dendl; + rocksdb::Status status = rocksdb::StringToMap(options, opt_map); + if (!status.ok()) { + dout(5) << __func__ << " error '" << status.getState() + << "' while parsing options '" << options << "'" << dendl; + return -EINVAL; + } + // if "block_cache" option exists, then move it to separate string + if (auto it = opt_map->find("block_cache"); it != opt_map->end()) { + *block_cache_opt = it->second; + opt_map->erase(it); + } else { + block_cache_opt->clear(); + } + return 0; +} + +// Updates column family options. +// Take options from more_options and apply them to cf_opt. +// Allowed options are exactly the same as allowed for column families in RocksDB. +// Ceph addition is "block_cache" option that is translated to block_cache and +// allows to specialize separate block cache for O column family. +// +// base_name - name of column without shard suffix: "-"+number +// options - additional options to apply +// cf_opt - column family options to update +int RocksDBStore::update_column_family_options(const std::string& base_name, + const std::string& more_options, + rocksdb::ColumnFamilyOptions* cf_opt) +{ + std::unordered_map<std::string, std::string> options_map; + std::string block_cache_opt; + rocksdb::Status status; + int r = split_column_family_options(more_options, &options_map, &block_cache_opt); + if (r != 0) { + dout(5) << __func__ << " failed to parse options; column family=" << base_name + << " options=" << more_options << dendl; + return r; + } + status = rocksdb::GetColumnFamilyOptionsFromMap(*cf_opt, options_map, cf_opt); + if (!status.ok()) { + dout(5) << __func__ << " invalid column family optionsp; column family=" + << base_name << " options=" << more_options << dendl; + dout(5) << __func__ << " RocksDB error='" << status.getState() << "'" << dendl; + return -EINVAL; + } + if (base_name != rocksdb::kDefaultColumnFamilyName) { + // default cf has its merge operator defined in load_rocksdb_options, should not override it + install_cf_mergeop(base_name, cf_opt); + } + if (!block_cache_opt.empty()) { + r = apply_block_cache_options(base_name, block_cache_opt, cf_opt); + if (r != 0) { + // apply_block_cache_options already does all necessary douts + return r; + } + } + return 0; +} + +int RocksDBStore::apply_block_cache_options(const std::string& column_name, + const std::string& block_cache_opt, + rocksdb::ColumnFamilyOptions* cf_opt) +{ + rocksdb::Status status; + std::unordered_map<std::string, std::string> cache_options_map; + status = rocksdb::StringToMap(block_cache_opt, &cache_options_map); + if (!status.ok()) { + dout(5) << __func__ << " invalid block cache options; column=" << column_name + << " options=" << block_cache_opt << dendl; + dout(5) << __func__ << " RocksDB error='" << status.getState() << "'" << dendl; + return -EINVAL; + } + bool require_new_block_cache = false; + std::string cache_type = cct->_conf->rocksdb_cache_type; + if (const auto it = cache_options_map.find("type"); it != cache_options_map.end()) { + cache_type = it->second; + cache_options_map.erase(it); + require_new_block_cache = true; + } + size_t cache_size = cct->_conf->rocksdb_cache_size; + if (auto it = cache_options_map.find("size"); it != cache_options_map.end()) { + std::string error; + cache_size = strict_iecstrtoll(it->second.c_str(), &error); + if (!error.empty()) { + dout(10) << __func__ << " invalid size: '" << it->second << "'" << dendl; + return -EINVAL; + } + cache_options_map.erase(it); + require_new_block_cache = true; + } + double high_pri_pool_ratio = 0.0; + if (auto it = cache_options_map.find("high_ratio"); it != cache_options_map.end()) { + std::string error; + high_pri_pool_ratio = strict_strtod(it->second.c_str(), &error); + if (!error.empty()) { + dout(10) << __func__ << " invalid high_pri (float): '" << it->second << "'" << dendl; + return -EINVAL; + } + cache_options_map.erase(it); + require_new_block_cache = true; + } + + rocksdb::BlockBasedTableOptions column_bbt_opts; + status = GetBlockBasedTableOptionsFromMap(bbt_opts, cache_options_map, &column_bbt_opts); + if (!status.ok()) { + dout(5) << __func__ << " invalid block cache options; column=" << column_name + << " options=" << block_cache_opt << dendl; + dout(5) << __func__ << " RocksDB error='" << status.getState() << "'" << dendl; + return -EINVAL; + } + std::shared_ptr<rocksdb::Cache> block_cache; + if (column_bbt_opts.no_block_cache) { + // clear all settings except no_block_cache + // rocksdb does not like then + column_bbt_opts = rocksdb::BlockBasedTableOptions(); + column_bbt_opts.no_block_cache = true; + } else { + if (require_new_block_cache) { + block_cache = create_block_cache(cache_type, cache_size, high_pri_pool_ratio); + if (!block_cache) { + dout(5) << __func__ << " failed to create block cache for params: " << block_cache_opt << dendl; + return -EINVAL; + } + } else { + block_cache = bbt_opts.block_cache; + } + } + column_bbt_opts.block_cache = block_cache; + cf_bbt_opts[column_name] = column_bbt_opts; + cf_opt->table_factory.reset(NewBlockBasedTableFactory(cf_bbt_opts[column_name])); + return 0; +} + +int RocksDBStore::verify_sharding(const rocksdb::Options& opt, + std::vector<rocksdb::ColumnFamilyDescriptor>& existing_cfs, + std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> >& existing_cfs_shard, + std::vector<rocksdb::ColumnFamilyDescriptor>& missing_cfs, + std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> >& missing_cfs_shard) +{ + rocksdb::Status status; + std::string stored_sharding_text; + status = opt.env->FileExists(sharding_def_file); + if (status.ok()) { + status = rocksdb::ReadFileToString(opt.env, + sharding_def_file, + &stored_sharding_text); + if(!status.ok()) { + derr << __func__ << " cannot read from " << sharding_def_file << dendl; + return -EIO; + } + dout(20) << __func__ << " sharding=" << stored_sharding_text << dendl; + } else { + dout(30) << __func__ << " no sharding" << dendl; + //no "sharding_def" present + } + //check if sharding_def matches stored_sharding_def + std::vector<ColumnFamily> stored_sharding_def; + parse_sharding_def(stored_sharding_text, stored_sharding_def); + + std::sort(stored_sharding_def.begin(), stored_sharding_def.end(), + [](ColumnFamily& a, ColumnFamily& b) { return a.name < b.name; } ); + + std::vector<string> rocksdb_cfs; + status = rocksdb::DB::ListColumnFamilies(rocksdb::DBOptions(opt), + path, &rocksdb_cfs); + if (!status.ok()) { + derr << __func__ << " unable to list column families: " << status.ToString() << dendl; + return -EIO; + } + dout(5) << __func__ << " column families from rocksdb: " << rocksdb_cfs << dendl; + + auto emplace_cf = [&] (const RocksDBStore::ColumnFamily& column, + int32_t shard_id, + const std::string& shard_name, + const rocksdb::ColumnFamilyOptions& opt) { + if (std::find(rocksdb_cfs.begin(), rocksdb_cfs.end(), shard_name) != rocksdb_cfs.end()) { + existing_cfs.emplace_back(shard_name, opt); + existing_cfs_shard.emplace_back(shard_id, column); + } else { + missing_cfs.emplace_back(shard_name, opt); + missing_cfs_shard.emplace_back(shard_id, column); + } + }; + + for (auto& column : stored_sharding_def) { + rocksdb::ColumnFamilyOptions cf_opt(opt); + int r = update_column_family_options(column.name, column.options, &cf_opt); + if (r != 0) { + return r; + } + if (column.shard_cnt == 1) { + emplace_cf(column, 0, column.name, cf_opt); + } else { + for (size_t i = 0; i < column.shard_cnt; i++) { + std::string cf_name = column.name + "-" + to_string(i); + emplace_cf(column, i, cf_name, cf_opt); + } + } + } + existing_cfs.emplace_back("default", opt); + + if (existing_cfs.size() != rocksdb_cfs.size()) { + std::vector<std::string> columns_from_stored; + sharding_def_to_columns(stored_sharding_def, columns_from_stored); + derr << __func__ << " extra columns in rocksdb. rocksdb columns = " << rocksdb_cfs + << " target columns = " << columns_from_stored << dendl; + return -EIO; + } + return 0; +} + +std::ostream& operator<<(std::ostream& out, const RocksDBStore::ColumnFamily& cf) +{ + out << "("; + out << cf.name; + out << ","; + out << cf.shard_cnt; + out << ","; + out << cf.hash_l; + out << "-"; + if (cf.hash_h != std::numeric_limits<uint32_t>::max()) { + out << cf.hash_h; + } + out << ","; + out << cf.options; + out << ")"; + return out; +} + +int RocksDBStore::do_open(ostream &out, + bool create_if_missing, + bool open_readonly, + const std::string& sharding_text) +{ + ceph_assert(!(create_if_missing && open_readonly)); + rocksdb::Options opt; + int r = load_rocksdb_options(create_if_missing, opt); + if (r) { + dout(1) << __func__ << " load rocksdb options failed" << dendl; + return r; + } + rocksdb::Status status; + if (create_if_missing) { + status = rocksdb::DB::Open(opt, path, &db); + if (!status.ok()) { + derr << status.ToString() << dendl; + return -EINVAL; + } + r = apply_sharding(opt, sharding_text); + if (r < 0) { + return r; + } + default_cf = db->DefaultColumnFamily(); + } else { + std::vector<rocksdb::ColumnFamilyDescriptor> existing_cfs; + std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> > existing_cfs_shard; + std::vector<rocksdb::ColumnFamilyDescriptor> missing_cfs; + std::vector<std::pair<size_t, RocksDBStore::ColumnFamily> > missing_cfs_shard; + + r = verify_sharding(opt, + existing_cfs, existing_cfs_shard, + missing_cfs, missing_cfs_shard); + if (r < 0) { + return r; + } + std::string sharding_recreate_text; + status = rocksdb::ReadFileToString(opt.env, + sharding_recreate, + &sharding_recreate_text); + bool recreate_mode = status.ok() && sharding_recreate_text == "1"; + + ceph_assert(!recreate_mode || !open_readonly); + if (recreate_mode == false && missing_cfs.size() != 0) { + // We do not accept when there are missing column families, except case that we are during resharding. + // We can get into this case if resharding was interrupted. It gives a chance to continue. + // Opening DB is only allowed in read-only mode. + if (open_readonly == false && + std::find_if(missing_cfs.begin(), missing_cfs.end(), + [](const rocksdb::ColumnFamilyDescriptor& c) { return c.name == resharding_column_lock; } + ) != missing_cfs.end()) { + derr << __func__ << " missing column families: " << missing_cfs_shard << dendl; + return -EIO; + } + } + + if (existing_cfs.empty()) { + // no column families + if (open_readonly) { + status = rocksdb::DB::OpenForReadOnly(opt, path, &db); + } else { + status = rocksdb::DB::Open(opt, path, &db); + } + if (!status.ok()) { + derr << status.ToString() << dendl; + return -EINVAL; + } + default_cf = db->DefaultColumnFamily(); + } else { + std::vector<rocksdb::ColumnFamilyHandle*> handles; + if (open_readonly) { + status = rocksdb::DB::OpenForReadOnly(rocksdb::DBOptions(opt), + path, existing_cfs, + &handles, &db); + } else { + status = rocksdb::DB::Open(rocksdb::DBOptions(opt), + path, existing_cfs, &handles, &db); + } + if (!status.ok()) { + derr << status.ToString() << dendl; + return -EINVAL; + } + ceph_assert(existing_cfs.size() == existing_cfs_shard.size() + 1); + ceph_assert(handles.size() == existing_cfs.size()); + dout(10) << __func__ << " existing_cfs=" << existing_cfs.size() << dendl; + for (size_t i = 0; i < existing_cfs_shard.size(); i++) { + add_column_family(existing_cfs_shard[i].second.name, + existing_cfs_shard[i].second.hash_l, + existing_cfs_shard[i].second.hash_h, + existing_cfs_shard[i].first, + handles[i]); + } + default_cf = handles[handles.size() - 1]; + must_close_default_cf = true; + + if (missing_cfs.size() > 0 && + std::find_if(missing_cfs.begin(), missing_cfs.end(), + [](const rocksdb::ColumnFamilyDescriptor& c) { return c.name == resharding_column_lock; } + ) == missing_cfs.end()) + { + dout(10) << __func__ << " missing_cfs=" << missing_cfs.size() << dendl; + ceph_assert(recreate_mode); + ceph_assert(missing_cfs.size() == missing_cfs_shard.size()); + for (size_t i = 0; i < missing_cfs.size(); i++) { + rocksdb::ColumnFamilyHandle *cf; + status = db->CreateColumnFamily(missing_cfs[i].options, missing_cfs[i].name, &cf); + if (!status.ok()) { + derr << __func__ << " Failed to create rocksdb column family: " + << missing_cfs[i].name << dendl; + return -EINVAL; + } + add_column_family(missing_cfs_shard[i].second.name, + missing_cfs_shard[i].second.hash_l, + missing_cfs_shard[i].second.hash_h, + missing_cfs_shard[i].first, + cf); + } + opt.env->DeleteFile(sharding_recreate); + } + } + } + ceph_assert(default_cf != nullptr); + + PerfCountersBuilder plb(cct, "rocksdb", l_rocksdb_first, l_rocksdb_last); + plb.add_u64_counter(l_rocksdb_gets, "get", "Gets"); + plb.add_time_avg(l_rocksdb_get_latency, "get_latency", "Get latency"); + plb.add_time_avg(l_rocksdb_submit_latency, "submit_latency", "Submit Latency"); + plb.add_time_avg(l_rocksdb_submit_sync_latency, "submit_sync_latency", "Submit Sync Latency"); + plb.add_u64_counter(l_rocksdb_compact, "compact", "Compactions"); + plb.add_u64_counter(l_rocksdb_compact_range, "compact_range", "Compactions by range"); + plb.add_u64_counter(l_rocksdb_compact_queue_merge, "compact_queue_merge", "Mergings of ranges in compaction queue"); + plb.add_u64(l_rocksdb_compact_queue_len, "compact_queue_len", "Length of compaction queue"); + plb.add_time_avg(l_rocksdb_write_wal_time, "rocksdb_write_wal_time", "Rocksdb write wal time"); + plb.add_time_avg(l_rocksdb_write_memtable_time, "rocksdb_write_memtable_time", "Rocksdb write memtable time"); + plb.add_time_avg(l_rocksdb_write_delay_time, "rocksdb_write_delay_time", "Rocksdb write delay time"); + plb.add_time_avg(l_rocksdb_write_pre_and_post_process_time, + "rocksdb_write_pre_and_post_time", "total time spent on writing a record, excluding write process"); + logger = plb.create_perf_counters(); + cct->get_perfcounters_collection()->add(logger); + + if (compact_on_mount) { + derr << "Compacting rocksdb store..." << dendl; + compact(); + derr << "Finished compacting rocksdb store" << dendl; + } + return 0; +} + +int RocksDBStore::_test_init(const string& dir) +{ + rocksdb::Options options; + options.create_if_missing = true; + rocksdb::DB *db; + rocksdb::Status status = rocksdb::DB::Open(options, dir, &db); + delete db; + db = nullptr; + return status.ok() ? 0 : -EIO; +} + +RocksDBStore::~RocksDBStore() +{ + close(); + if (priv) { + delete static_cast<rocksdb::Env*>(priv); + } +} + +void RocksDBStore::close() +{ + // stop compaction thread + compact_queue_lock.lock(); + if (compact_thread.is_started()) { + dout(1) << __func__ << " waiting for compaction thread to stop" << dendl; + compact_queue_stop = true; + compact_queue_cond.notify_all(); + compact_queue_lock.unlock(); + compact_thread.join(); + dout(1) << __func__ << " compaction thread to stopped" << dendl; + } else { + compact_queue_lock.unlock(); + } + + if (logger) { + cct->get_perfcounters_collection()->remove(logger); + delete logger; + logger = nullptr; + } + + // Ensure db is destroyed before dependent db_cache and filterpolicy + for (auto& p : cf_handles) { + for (size_t i = 0; i < p.second.handles.size(); i++) { + db->DestroyColumnFamilyHandle(p.second.handles[i]); + } + } + cf_handles.clear(); + if (must_close_default_cf) { + db->DestroyColumnFamilyHandle(default_cf); + must_close_default_cf = false; + } + default_cf = nullptr; + delete db; + db = nullptr; +} + +int RocksDBStore::repair(std::ostream &out) +{ + rocksdb::Status status; + rocksdb::Options opt; + int r = load_rocksdb_options(false, opt); + if (r) { + dout(1) << __func__ << " load rocksdb options failed" << dendl; + out << "load rocksdb options failed" << std::endl; + return r; + } + //need to save sharding definition, repairDB will delete files it does not know + std::string stored_sharding_text; + status = opt.env->FileExists(sharding_def_file); + if (status.ok()) { + status = rocksdb::ReadFileToString(opt.env, + sharding_def_file, + &stored_sharding_text); + if (!status.ok()) { + stored_sharding_text.clear(); + } + } + dout(10) << __func__ << " stored_sharding: " << stored_sharding_text << dendl; + status = rocksdb::RepairDB(path, opt); + bool repaired = status.ok(); + if (!stored_sharding_text.empty()) { + //recreate markers even if repair failed + opt.env->CreateDir(sharding_def_dir); + status = rocksdb::WriteStringToFile(opt.env, stored_sharding_text, + sharding_def_file, true); + if (!status.ok()) { + derr << __func__ << " cannot write to " << sharding_def_file << dendl; + return -1; + } + status = rocksdb::WriteStringToFile(opt.env, "1", + sharding_recreate, true); + if (!status.ok()) { + derr << __func__ << " cannot write to " << sharding_recreate << dendl; + return -1; + } + // fiinalize sharding recreate + if (do_open(out, false, false)) { + derr << __func__ << " cannot finalize repair" << dendl; + return -1; + } + close(); + } + + if (repaired && status.ok()) { + return 0; + } else { + out << "repair rocksdb failed : " << status.ToString() << std::endl; + return -1; + } +} + +void RocksDBStore::split_stats(const std::string &s, char delim, std::vector<std::string> &elems) { + std::stringstream ss; + ss.str(s); + std::string item; + while (std::getline(ss, item, delim)) { + elems.push_back(item); + } +} + +bool RocksDBStore::get_property( + const std::string &property, + uint64_t *out) +{ + return db->GetIntProperty(property, out); +} + +int64_t RocksDBStore::estimate_prefix_size(const string& prefix, + const string& key_prefix) +{ + uint64_t size = 0; + uint8_t flags = + //rocksdb::DB::INCLUDE_MEMTABLES | // do not include memtables... + rocksdb::DB::INCLUDE_FILES; + auto p_iter = cf_handles.find(prefix); + if (p_iter != cf_handles.end()) { + for (auto cf : p_iter->second.handles) { + uint64_t s = 0; + string start = key_prefix + string(1, '\x00'); + string limit = key_prefix + string("\xff\xff\xff\xff"); + rocksdb::Range r(start, limit); + db->GetApproximateSizes(cf, &r, 1, &s, flags); + size += s; + } + } else { + string start = combine_strings(prefix , key_prefix); + string limit = combine_strings(prefix , key_prefix + "\xff\xff\xff\xff"); + rocksdb::Range r(start, limit); + db->GetApproximateSizes(default_cf, &r, 1, &size, flags); + } + return size; +} + +void RocksDBStore::get_statistics(Formatter *f) +{ + if (!cct->_conf->rocksdb_perf) { + dout(20) << __func__ << " RocksDB perf is disabled, can't probe for stats" + << dendl; + return; + } + + if (cct->_conf->rocksdb_collect_compaction_stats) { + std::string stat_str; + bool status = db->GetProperty("rocksdb.stats", &stat_str); + if (status) { + f->open_object_section("rocksdb_statistics"); + f->dump_string("rocksdb_compaction_statistics", ""); + vector<string> stats; + split_stats(stat_str, '\n', stats); + for (auto st :stats) { + f->dump_string("", st); + } + f->close_section(); + } + } + if (cct->_conf->rocksdb_collect_extended_stats) { + if (dbstats) { + f->open_object_section("rocksdb_extended_statistics"); + string stat_str = dbstats->ToString(); + vector<string> stats; + split_stats(stat_str, '\n', stats); + f->dump_string("rocksdb_extended_statistics", ""); + for (auto st :stats) { + f->dump_string(".", st); + } + f->close_section(); + } + f->open_object_section("rocksdbstore_perf_counters"); + logger->dump_formatted(f,0); + f->close_section(); + } + if (cct->_conf->rocksdb_collect_memory_stats) { + f->open_object_section("rocksdb_memtable_statistics"); + std::string str; + if (!bbt_opts.no_block_cache) { + str.append(stringify(bbt_opts.block_cache->GetUsage())); + f->dump_string("block_cache_usage", str.data()); + str.clear(); + str.append(stringify(bbt_opts.block_cache->GetPinnedUsage())); + f->dump_string("block_cache_pinned_blocks_usage", str); + str.clear(); + } + db->GetProperty("rocksdb.cur-size-all-mem-tables", &str); + f->dump_string("rocksdb_memtable_usage", str); + str.clear(); + db->GetProperty("rocksdb.estimate-table-readers-mem", &str); + f->dump_string("rocksdb_index_filter_blocks_usage", str); + f->close_section(); + } +} + +struct RocksDBStore::RocksWBHandler: public rocksdb::WriteBatch::Handler { + RocksWBHandler(const RocksDBStore& db) : db(db) {} + const RocksDBStore& db; + std::stringstream seen; + int num_seen = 0; + + void dump(const char* op_name, + uint32_t column_family_id, + const rocksdb::Slice& key_in, + const rocksdb::Slice* value = nullptr) { + string prefix; + string key; + ssize_t size = value ? value->size() : -1; + seen << std::endl << op_name << "("; + + if (column_family_id == 0) { + db.split_key(key_in, &prefix, &key); + } else { + auto it = db.cf_ids_to_prefix.find(column_family_id); + ceph_assert(it != db.cf_ids_to_prefix.end()); + prefix = it->second; + key = key_in.ToString(); + } + seen << " prefix = " << prefix; + seen << " key = " << pretty_binary_string(key); + if (size != -1) + seen << " value size = " << std::to_string(size); + seen << ")"; + num_seen++; + } + void Put(const rocksdb::Slice& key, + const rocksdb::Slice& value) override { + dump("Put", 0, key, &value); + } + rocksdb::Status PutCF(uint32_t column_family_id, const rocksdb::Slice& key, + const rocksdb::Slice& value) override { + dump("PutCF", column_family_id, key, &value); + return rocksdb::Status::OK(); + } + void SingleDelete(const rocksdb::Slice& key) override { + dump("SingleDelete", 0, key); + } + rocksdb::Status SingleDeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override { + dump("SingleDeleteCF", column_family_id, key); + return rocksdb::Status::OK(); + } + void Delete(const rocksdb::Slice& key) override { + dump("Delete", 0, key); + } + rocksdb::Status DeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override { + dump("DeleteCF", column_family_id, key); + return rocksdb::Status::OK(); + } + void Merge(const rocksdb::Slice& key, + const rocksdb::Slice& value) override { + dump("Merge", 0, key, &value); + } + rocksdb::Status MergeCF(uint32_t column_family_id, const rocksdb::Slice& key, + const rocksdb::Slice& value) override { + dump("MergeCF", column_family_id, key, &value); + return rocksdb::Status::OK(); + } + bool Continue() override { return num_seen < 50; } +}; + +int RocksDBStore::submit_common(rocksdb::WriteOptions& woptions, KeyValueDB::Transaction t) +{ + // enable rocksdb breakdown + // considering performance overhead, default is disabled + if (cct->_conf->rocksdb_perf) { + rocksdb::SetPerfLevel(rocksdb::PerfLevel::kEnableTimeExceptForMutex); + rocksdb::get_perf_context()->Reset(); + } + + RocksDBTransactionImpl * _t = + static_cast<RocksDBTransactionImpl *>(t.get()); + woptions.disableWAL = disableWAL; + lgeneric_subdout(cct, rocksdb, 30) << __func__; + RocksWBHandler bat_txc(*this); + _t->bat.Iterate(&bat_txc); + *_dout << " Rocksdb transaction: " << bat_txc.seen.str() << dendl; + + rocksdb::Status s = db->Write(woptions, &_t->bat); + if (!s.ok()) { + RocksWBHandler rocks_txc(*this); + _t->bat.Iterate(&rocks_txc); + derr << __func__ << " error: " << s.ToString() << " code = " << s.code() + << " Rocksdb transaction: " << rocks_txc.seen.str() << dendl; + } + + if (cct->_conf->rocksdb_perf) { + utime_t write_memtable_time; + utime_t write_delay_time; + utime_t write_wal_time; + utime_t write_pre_and_post_process_time; + write_wal_time.set_from_double( + static_cast<double>(rocksdb::get_perf_context()->write_wal_time)/1000000000); + write_memtable_time.set_from_double( + static_cast<double>(rocksdb::get_perf_context()->write_memtable_time)/1000000000); + write_delay_time.set_from_double( + static_cast<double>(rocksdb::get_perf_context()->write_delay_time)/1000000000); + write_pre_and_post_process_time.set_from_double( + static_cast<double>(rocksdb::get_perf_context()->write_pre_and_post_process_time)/1000000000); + logger->tinc(l_rocksdb_write_memtable_time, write_memtable_time); + logger->tinc(l_rocksdb_write_delay_time, write_delay_time); + logger->tinc(l_rocksdb_write_wal_time, write_wal_time); + logger->tinc(l_rocksdb_write_pre_and_post_process_time, write_pre_and_post_process_time); + } + + return s.ok() ? 0 : -1; +} + +int RocksDBStore::submit_transaction(KeyValueDB::Transaction t) +{ + utime_t start = ceph_clock_now(); + rocksdb::WriteOptions woptions; + woptions.sync = false; + + int result = submit_common(woptions, t); + + utime_t lat = ceph_clock_now() - start; + logger->tinc(l_rocksdb_submit_latency, lat); + + return result; +} + +int RocksDBStore::submit_transaction_sync(KeyValueDB::Transaction t) +{ + utime_t start = ceph_clock_now(); + rocksdb::WriteOptions woptions; + // if disableWAL, sync can't set + woptions.sync = !disableWAL; + + int result = submit_common(woptions, t); + + utime_t lat = ceph_clock_now() - start; + logger->tinc(l_rocksdb_submit_sync_latency, lat); + + return result; +} + +RocksDBStore::RocksDBTransactionImpl::RocksDBTransactionImpl(RocksDBStore *_db) +{ + db = _db; +} + +void RocksDBStore::RocksDBTransactionImpl::put_bat( + rocksdb::WriteBatch& bat, + rocksdb::ColumnFamilyHandle *cf, + const string &key, + const bufferlist &to_set_bl) +{ + // bufferlist::c_str() is non-constant, so we can't call c_str() + if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) { + bat.Put(cf, + rocksdb::Slice(key), + rocksdb::Slice(to_set_bl.buffers().front().c_str(), + to_set_bl.length())); + } else { + rocksdb::Slice key_slice(key); + vector<rocksdb::Slice> value_slices(to_set_bl.get_num_buffers()); + bat.Put(cf, + rocksdb::SliceParts(&key_slice, 1), + prepare_sliceparts(to_set_bl, &value_slices)); + } +} + +void RocksDBStore::RocksDBTransactionImpl::set( + const string &prefix, + const string &k, + const bufferlist &to_set_bl) +{ + auto cf = db->get_cf_handle(prefix, k); + if (cf) { + put_bat(bat, cf, k, to_set_bl); + } else { + string key = combine_strings(prefix, k); + put_bat(bat, db->default_cf, key, to_set_bl); + } +} + +void RocksDBStore::RocksDBTransactionImpl::set( + const string &prefix, + const char *k, size_t keylen, + const bufferlist &to_set_bl) +{ + auto cf = db->get_cf_handle(prefix, k, keylen); + if (cf) { + string key(k, keylen); // fixme? + put_bat(bat, cf, key, to_set_bl); + } else { + string key; + combine_strings(prefix, k, keylen, &key); + put_bat(bat, cf, key, to_set_bl); + } +} + +void RocksDBStore::RocksDBTransactionImpl::rmkey(const string &prefix, + const string &k) +{ + auto cf = db->get_cf_handle(prefix, k); + if (cf) { + bat.Delete(cf, rocksdb::Slice(k)); + } else { + bat.Delete(db->default_cf, combine_strings(prefix, k)); + } +} + +void RocksDBStore::RocksDBTransactionImpl::rmkey(const string &prefix, + const char *k, + size_t keylen) +{ + auto cf = db->get_cf_handle(prefix, k, keylen); + if (cf) { + bat.Delete(cf, rocksdb::Slice(k, keylen)); + } else { + string key; + combine_strings(prefix, k, keylen, &key); + bat.Delete(db->default_cf, rocksdb::Slice(key)); + } +} + +void RocksDBStore::RocksDBTransactionImpl::rm_single_key(const string &prefix, + const string &k) +{ + auto cf = db->get_cf_handle(prefix, k); + if (cf) { + bat.SingleDelete(cf, k); + } else { + bat.SingleDelete(db->default_cf, combine_strings(prefix, k)); + } +} + +void RocksDBStore::RocksDBTransactionImpl::rmkeys_by_prefix(const string &prefix) +{ + auto p_iter = db->cf_handles.find(prefix); + if (p_iter == db->cf_handles.end()) { + uint64_t cnt = db->delete_range_threshold; + bat.SetSavePoint(); + auto it = db->get_iterator(prefix); + for (it->seek_to_first(); it->valid() && (--cnt) != 0; it->next()) { + bat.Delete(db->default_cf, combine_strings(prefix, it->key())); + } + if (cnt == 0) { + bat.RollbackToSavePoint(); + string endprefix = prefix; + endprefix.push_back('\x01'); + bat.DeleteRange(db->default_cf, + combine_strings(prefix, string()), + combine_strings(endprefix, string())); + } else { + bat.PopSavePoint(); + } + } else { + ceph_assert(p_iter->second.handles.size() >= 1); + for (auto cf : p_iter->second.handles) { + uint64_t cnt = db->delete_range_threshold; + bat.SetSavePoint(); + auto it = db->new_shard_iterator(cf); + for (it->SeekToFirst(); it->Valid() && (--cnt) != 0; it->Next()) { + bat.Delete(cf, it->key()); + } + if (cnt == 0) { + bat.RollbackToSavePoint(); + string endprefix = "\xff\xff\xff\xff"; // FIXME: this is cheating... + bat.DeleteRange(cf, string(), endprefix); + } else { + bat.PopSavePoint(); + } + } + } +} + +void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix, + const string &start, + const string &end) +{ + ldout(db->cct, 10) << __func__ << " enter start=" << start + << " end=" << end << dendl; + auto p_iter = db->cf_handles.find(prefix); + if (p_iter == db->cf_handles.end()) { + uint64_t cnt = db->delete_range_threshold; + bat.SetSavePoint(); + auto it = db->get_iterator(prefix); + for (it->lower_bound(start); + it->valid() && db->comparator->Compare(it->key(), end) < 0 && (--cnt) != 0; + it->next()) { + bat.Delete(db->default_cf, combine_strings(prefix, it->key())); + } + if (cnt == 0) { + ldout(db->cct, 10) << __func__ << " p_iter == end(), resorting to DeleteRange" + << dendl; + bat.RollbackToSavePoint(); + bat.DeleteRange(db->default_cf, + rocksdb::Slice(combine_strings(prefix, start)), + rocksdb::Slice(combine_strings(prefix, end))); + } else { + bat.PopSavePoint(); + } + } else { + ceph_assert(p_iter->second.handles.size() >= 1); + for (auto cf : p_iter->second.handles) { + uint64_t cnt = db->delete_range_threshold; + bat.SetSavePoint(); + rocksdb::Iterator* it = db->new_shard_iterator(cf); + ceph_assert(it != nullptr); + for (it->Seek(start); + it->Valid() && db->comparator->Compare(it->key(), end) < 0 && (--cnt) != 0; + it->Next()) { + bat.Delete(cf, it->key()); + } + if (cnt == 0) { + ldout(db->cct, 10) << __func__ << " p_iter != end(), resorting to DeleteRange" + << dendl; + bat.RollbackToSavePoint(); + bat.DeleteRange(cf, rocksdb::Slice(start), rocksdb::Slice(end)); + } else { + bat.PopSavePoint(); + } + delete it; + } + } + ldout(db->cct, 10) << __func__ << " end" << dendl; +} + +void RocksDBStore::RocksDBTransactionImpl::merge( + const string &prefix, + const string &k, + const bufferlist &to_set_bl) +{ + auto cf = db->get_cf_handle(prefix, k); + if (cf) { + // bufferlist::c_str() is non-constant, so we can't call c_str() + if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) { + bat.Merge( + cf, + rocksdb::Slice(k), + rocksdb::Slice(to_set_bl.buffers().front().c_str(), to_set_bl.length())); + } else { + // make a copy + rocksdb::Slice key_slice(k); + vector<rocksdb::Slice> value_slices(to_set_bl.get_num_buffers()); + bat.Merge(cf, rocksdb::SliceParts(&key_slice, 1), + prepare_sliceparts(to_set_bl, &value_slices)); + } + } else { + string key = combine_strings(prefix, k); + // bufferlist::c_str() is non-constant, so we can't call c_str() + if (to_set_bl.is_contiguous() && to_set_bl.length() > 0) { + bat.Merge( + db->default_cf, + rocksdb::Slice(key), + rocksdb::Slice(to_set_bl.buffers().front().c_str(), to_set_bl.length())); + } else { + // make a copy + rocksdb::Slice key_slice(key); + vector<rocksdb::Slice> value_slices(to_set_bl.get_num_buffers()); + bat.Merge( + db->default_cf, + rocksdb::SliceParts(&key_slice, 1), + prepare_sliceparts(to_set_bl, &value_slices)); + } + } +} + +int RocksDBStore::get( + const string &prefix, + const std::set<string> &keys, + std::map<string, bufferlist> *out) +{ + rocksdb::PinnableSlice value; + utime_t start = ceph_clock_now(); + if (cf_handles.count(prefix) > 0) { + for (auto& key : keys) { + auto cf_handle = get_cf_handle(prefix, key); + auto status = db->Get(rocksdb::ReadOptions(), + cf_handle, + rocksdb::Slice(key), + &value); + if (status.ok()) { + (*out)[key].append(value.data(), value.size()); + } else if (status.IsIOError()) { + ceph_abort_msg(status.getState()); + } + value.Reset(); + } + } else { + for (auto& key : keys) { + string k = combine_strings(prefix, key); + auto status = db->Get(rocksdb::ReadOptions(), + default_cf, + rocksdb::Slice(k), + &value); + if (status.ok()) { + (*out)[key].append(value.data(), value.size()); + } else if (status.IsIOError()) { + ceph_abort_msg(status.getState()); + } + value.Reset(); + } + } + utime_t lat = ceph_clock_now() - start; + logger->inc(l_rocksdb_gets); + logger->tinc(l_rocksdb_get_latency, lat); + return 0; +} + +int RocksDBStore::get( + const string &prefix, + const string &key, + bufferlist *out) +{ + ceph_assert(out && (out->length() == 0)); + utime_t start = ceph_clock_now(); + int r = 0; + rocksdb::PinnableSlice value; + rocksdb::Status s; + auto cf = get_cf_handle(prefix, key); + if (cf) { + s = db->Get(rocksdb::ReadOptions(), + cf, + rocksdb::Slice(key), + &value); + } else { + string k = combine_strings(prefix, key); + s = db->Get(rocksdb::ReadOptions(), + default_cf, + rocksdb::Slice(k), + &value); + } + if (s.ok()) { + out->append(value.data(), value.size()); + } else if (s.IsNotFound()) { + r = -ENOENT; + } else { + ceph_abort_msg(s.getState()); + } + utime_t lat = ceph_clock_now() - start; + logger->inc(l_rocksdb_gets); + logger->tinc(l_rocksdb_get_latency, lat); + return r; +} + +int RocksDBStore::get( + const string& prefix, + const char *key, + size_t keylen, + bufferlist *out) +{ + ceph_assert(out && (out->length() == 0)); + utime_t start = ceph_clock_now(); + int r = 0; + rocksdb::PinnableSlice value; + rocksdb::Status s; + auto cf = get_cf_handle(prefix, key, keylen); + if (cf) { + s = db->Get(rocksdb::ReadOptions(), + cf, + rocksdb::Slice(key, keylen), + &value); + } else { + string k; + combine_strings(prefix, key, keylen, &k); + s = db->Get(rocksdb::ReadOptions(), + default_cf, + rocksdb::Slice(k), + &value); + } + if (s.ok()) { + out->append(value.data(), value.size()); + } else if (s.IsNotFound()) { + r = -ENOENT; + } else { + ceph_abort_msg(s.getState()); + } + utime_t lat = ceph_clock_now() - start; + logger->inc(l_rocksdb_gets); + logger->tinc(l_rocksdb_get_latency, lat); + return r; +} + +int RocksDBStore::split_key(rocksdb::Slice in, string *prefix, string *key) +{ + size_t prefix_len = 0; + + // Find separator inside Slice + char* separator = (char*) memchr(in.data(), 0, in.size()); + if (separator == NULL) + return -EINVAL; + prefix_len = size_t(separator - in.data()); + if (prefix_len >= in.size()) + return -EINVAL; + + // Fetch prefix and/or key directly from Slice + if (prefix) + *prefix = string(in.data(), prefix_len); + if (key) + *key = string(separator+1, in.size()-prefix_len-1); + return 0; +} + +void RocksDBStore::compact() +{ + logger->inc(l_rocksdb_compact); + rocksdb::CompactRangeOptions options; + db->CompactRange(options, default_cf, nullptr, nullptr); + for (auto cf : cf_handles) { + for (auto shard_cf : cf.second.handles) { + db->CompactRange( + options, + shard_cf, + nullptr, nullptr); + } + } +} + +void RocksDBStore::compact_thread_entry() +{ + std::unique_lock l{compact_queue_lock}; + dout(10) << __func__ << " enter" << dendl; + while (!compact_queue_stop) { + if (!compact_queue.empty()) { + auto range = compact_queue.front(); + compact_queue.pop_front(); + logger->set(l_rocksdb_compact_queue_len, compact_queue.size()); + l.unlock(); + logger->inc(l_rocksdb_compact_range); + if (range.first.empty() && range.second.empty()) { + compact(); + } else { + compact_range(range.first, range.second); + } + l.lock(); + continue; + } + dout(10) << __func__ << " waiting" << dendl; + compact_queue_cond.wait(l); + } + dout(10) << __func__ << " exit" << dendl; +} + +void RocksDBStore::compact_range_async(const string& start, const string& end) +{ + std::lock_guard l(compact_queue_lock); + + // try to merge adjacent ranges. this is O(n), but the queue should + // be short. note that we do not cover all overlap cases and merge + // opportunities here, but we capture the ones we currently need. + list< pair<string,string> >::iterator p = compact_queue.begin(); + while (p != compact_queue.end()) { + if (p->first == start && p->second == end) { + // dup; no-op + return; + } + if (start <= p->first && p->first <= end) { + // new region crosses start of existing range + // select right bound that is bigger + compact_queue.push_back(make_pair(start, end > p->second ? end : p->second)); + compact_queue.erase(p); + logger->inc(l_rocksdb_compact_queue_merge); + break; + } + if (start <= p->second && p->second <= end) { + // new region crosses end of existing range + //p->first < p->second and p->second <= end, so p->first <= end. + //But we break if previous condition, so start > p->first. + compact_queue.push_back(make_pair(p->first, end)); + compact_queue.erase(p); + logger->inc(l_rocksdb_compact_queue_merge); + break; + } + ++p; + } + if (p == compact_queue.end()) { + // no merge, new entry. + compact_queue.push_back(make_pair(start, end)); + logger->set(l_rocksdb_compact_queue_len, compact_queue.size()); + } + compact_queue_cond.notify_all(); + if (!compact_thread.is_started()) { + compact_thread.create("rstore_compact"); + } +} +bool RocksDBStore::check_omap_dir(string &omap_dir) +{ + rocksdb::Options options; + options.create_if_missing = true; + rocksdb::DB *db; + rocksdb::Status status = rocksdb::DB::Open(options, omap_dir, &db); + delete db; + db = nullptr; + return status.ok(); +} + +void RocksDBStore::compact_range(const string& start, const string& end) +{ + rocksdb::CompactRangeOptions options; + rocksdb::Slice cstart(start); + rocksdb::Slice cend(end); + string prefix_start, key_start; + string prefix_end, key_end; + string key_highest = "\xff\xff\xff\xff"; //cheating + string key_lowest = ""; + + auto compact_range = [&] (const decltype(cf_handles)::iterator column_it, + const std::string& start, + const std::string& end) { + rocksdb::Slice cstart(start); + rocksdb::Slice cend(end); + for (const auto& shard_it : column_it->second.handles) { + db->CompactRange(options, shard_it, &cstart, &cend); + } + }; + db->CompactRange(options, default_cf, &cstart, &cend); + split_key(cstart, &prefix_start, &key_start); + split_key(cend, &prefix_end, &key_end); + if (prefix_start == prefix_end) { + const auto& column = cf_handles.find(prefix_start); + if (column != cf_handles.end()) { + compact_range(column, key_start, key_end); + } + } else { + auto column = cf_handles.find(prefix_start); + if (column != cf_handles.end()) { + compact_range(column, key_start, key_highest); + ++column; + } + const auto& column_end = cf_handles.find(prefix_end); + while (column != column_end) { + compact_range(column, key_lowest, key_highest); + column++; + } + if (column != cf_handles.end()) { + compact_range(column, key_lowest, key_end); + } + } +} + +RocksDBStore::RocksDBWholeSpaceIteratorImpl::~RocksDBWholeSpaceIteratorImpl() +{ + delete dbiter; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_first() +{ + dbiter->SeekToFirst(); + ceph_assert(!dbiter->status().IsIOError()); + return dbiter->status().ok() ? 0 : -1; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_first(const string &prefix) +{ + rocksdb::Slice slice_prefix(prefix); + dbiter->Seek(slice_prefix); + ceph_assert(!dbiter->status().IsIOError()); + return dbiter->status().ok() ? 0 : -1; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_last() +{ + dbiter->SeekToLast(); + ceph_assert(!dbiter->status().IsIOError()); + return dbiter->status().ok() ? 0 : -1; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::seek_to_last(const string &prefix) +{ + string limit = past_prefix(prefix); + rocksdb::Slice slice_limit(limit); + dbiter->Seek(slice_limit); + + if (!dbiter->Valid()) { + dbiter->SeekToLast(); + } else { + dbiter->Prev(); + } + return dbiter->status().ok() ? 0 : -1; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::upper_bound(const string &prefix, const string &after) +{ + lower_bound(prefix, after); + if (valid()) { + pair<string,string> key = raw_key(); + if (key.first == prefix && key.second == after) + next(); + } + return dbiter->status().ok() ? 0 : -1; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::lower_bound(const string &prefix, const string &to) +{ + string bound = combine_strings(prefix, to); + rocksdb::Slice slice_bound(bound); + dbiter->Seek(slice_bound); + return dbiter->status().ok() ? 0 : -1; +} +bool RocksDBStore::RocksDBWholeSpaceIteratorImpl::valid() +{ + return dbiter->Valid(); +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::next() +{ + if (valid()) { + dbiter->Next(); + } + ceph_assert(!dbiter->status().IsIOError()); + return dbiter->status().ok() ? 0 : -1; +} +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::prev() +{ + if (valid()) { + dbiter->Prev(); + } + ceph_assert(!dbiter->status().IsIOError()); + return dbiter->status().ok() ? 0 : -1; +} +string RocksDBStore::RocksDBWholeSpaceIteratorImpl::key() +{ + string out_key; + split_key(dbiter->key(), 0, &out_key); + return out_key; +} +pair<string,string> RocksDBStore::RocksDBWholeSpaceIteratorImpl::raw_key() +{ + string prefix, key; + split_key(dbiter->key(), &prefix, &key); + return make_pair(prefix, key); +} + +bool RocksDBStore::RocksDBWholeSpaceIteratorImpl::raw_key_is_prefixed(const string &prefix) { + // Look for "prefix\0" right in rocksb::Slice + rocksdb::Slice key = dbiter->key(); + if ((key.size() > prefix.length()) && (key[prefix.length()] == '\0')) { + return memcmp(key.data(), prefix.c_str(), prefix.length()) == 0; + } else { + return false; + } +} + +bufferlist RocksDBStore::RocksDBWholeSpaceIteratorImpl::value() +{ + return to_bufferlist(dbiter->value()); +} + +size_t RocksDBStore::RocksDBWholeSpaceIteratorImpl::key_size() +{ + return dbiter->key().size(); +} + +size_t RocksDBStore::RocksDBWholeSpaceIteratorImpl::value_size() +{ + return dbiter->value().size(); +} + +bufferptr RocksDBStore::RocksDBWholeSpaceIteratorImpl::value_as_ptr() +{ + rocksdb::Slice val = dbiter->value(); + return bufferptr(val.data(), val.size()); +} + +int RocksDBStore::RocksDBWholeSpaceIteratorImpl::status() +{ + return dbiter->status().ok() ? 0 : -1; +} + +string RocksDBStore::past_prefix(const string &prefix) +{ + string limit = prefix; + limit.push_back(1); + return limit; +} + +class CFIteratorImpl : public KeyValueDB::IteratorImpl { +protected: + string prefix; + rocksdb::Iterator *dbiter; + const KeyValueDB::IteratorBounds bounds; + const rocksdb::Slice iterate_lower_bound; + const rocksdb::Slice iterate_upper_bound; +public: + explicit CFIteratorImpl(const RocksDBStore* db, + const std::string& p, + rocksdb::ColumnFamilyHandle* cf, + KeyValueDB::IteratorBounds bounds_) + : prefix(p), bounds(std::move(bounds_)), + iterate_lower_bound(make_slice(bounds.lower_bound)), + iterate_upper_bound(make_slice(bounds.upper_bound)) + { + auto options = rocksdb::ReadOptions(); + if (db->cct->_conf->osd_rocksdb_iterator_bounds_enabled) { + if (bounds.lower_bound) { + options.iterate_lower_bound = &iterate_lower_bound; + } + if (bounds.upper_bound) { + options.iterate_upper_bound = &iterate_upper_bound; + } + } + dbiter = db->db->NewIterator(options, cf); + } + ~CFIteratorImpl() { + delete dbiter; + } + + int seek_to_first() override { + dbiter->SeekToFirst(); + return dbiter->status().ok() ? 0 : -1; + } + int seek_to_last() override { + dbiter->SeekToLast(); + return dbiter->status().ok() ? 0 : -1; + } + int upper_bound(const string &after) override { + lower_bound(after); + if (valid() && (key() == after)) { + next(); + } + return dbiter->status().ok() ? 0 : -1; + } + int lower_bound(const string &to) override { + rocksdb::Slice slice_bound(to); + dbiter->Seek(slice_bound); + return dbiter->status().ok() ? 0 : -1; + } + int next() override { + if (valid()) { + dbiter->Next(); + } + return dbiter->status().ok() ? 0 : -1; + } + int prev() override { + if (valid()) { + dbiter->Prev(); + } + return dbiter->status().ok() ? 0 : -1; + } + bool valid() override { + return dbiter->Valid(); + } + string key() override { + return dbiter->key().ToString(); + } + std::pair<std::string, std::string> raw_key() override { + return make_pair(prefix, key()); + } + bufferlist value() override { + return to_bufferlist(dbiter->value()); + } + bufferptr value_as_ptr() override { + rocksdb::Slice val = dbiter->value(); + return bufferptr(val.data(), val.size()); + } + int status() override { + return dbiter->status().ok() ? 0 : -1; + } +}; + + +//merge column iterators and rest iterator +class WholeMergeIteratorImpl : public KeyValueDB::WholeSpaceIteratorImpl { +private: + RocksDBStore* db; + KeyValueDB::WholeSpaceIterator main; + std::map<std::string, KeyValueDB::Iterator> shards; + std::map<std::string, KeyValueDB::Iterator>::iterator current_shard; + enum {on_main, on_shard} smaller; + +public: + WholeMergeIteratorImpl(RocksDBStore* db) + : db(db) + , main(db->get_default_cf_iterator()) + { + for (auto& e : db->cf_handles) { + shards.emplace(e.first, db->get_iterator(e.first)); + } + } + + // returns true if value in main is smaller then in shards + // invalid is larger then actual value + bool is_main_smaller() { + if (main->valid()) { + if (current_shard != shards.end()) { + auto main_rk = main->raw_key(); + ceph_assert(current_shard->second->valid()); + auto shards_rk = current_shard->second->raw_key(); + if (main_rk.first < shards_rk.first) + return true; + if (main_rk.first > shards_rk.first) + return false; + return main_rk.second < shards_rk.second; + } else { + return true; + } + } else { + if (current_shard != shards.end()) { + return false; + } else { + //this means that neither is valid + //we select main to be smaller, so valid() will signal properly + return true; + } + } + } + + int seek_to_first() override { + int r0 = main->seek_to_first(); + int r1 = 0; + // find first shard that has some data + current_shard = shards.begin(); + while (current_shard != shards.end()) { + r1 = current_shard->second->seek_to_first(); + if (r1 != 0 || current_shard->second->valid()) { + //this is the first shard that will yield some keys + break; + } + ++current_shard; + } + smaller = is_main_smaller() ? on_main : on_shard; + return r0 == 0 && r1 == 0 ? 0 : -1; + } + + int seek_to_first(const std::string &prefix) override { + int r0 = main->seek_to_first(prefix); + int r1 = 0; + // find first shard that has some data + current_shard = shards.lower_bound(prefix); + while (current_shard != shards.end()) { + r1 = current_shard->second->seek_to_first(); + if (r1 != 0 || current_shard->second->valid()) { + //this is the first shard that will yield some keys + break; + } + ++current_shard; + } + smaller = is_main_smaller() ? on_main : on_shard; + return r0 == 0 && r1 == 0 ? 0 : -1; + }; + + int seek_to_last() override { + int r0 = main->seek_to_last(); + int r1 = 0; + r1 = shards_seek_to_last(); + //if we have 2 candidates, we need to select + if (main->valid()) { + if (shards_valid()) { + if (is_main_smaller()) { + smaller = on_shard; + main->next(); + } else { + smaller = on_main; + shards_next(); + } + } else { + smaller = on_main; + } + } else { + if (shards_valid()) { + smaller = on_shard; + } else { + smaller = on_main; + } + } + return r0 == 0 && r1 == 0 ? 0 : -1; + } + + int seek_to_last(const std::string &prefix) override { + int r0 = main->seek_to_last(prefix); + int r1 = 0; + // find last shard that has some data + bool found = false; + current_shard = shards.lower_bound(prefix); + while (current_shard != shards.begin()) { + r1 = current_shard->second->seek_to_last(); + if (r1 != 0) + break; + if (current_shard->second->valid()) { + found = true; + break; + } + } + //if we have 2 candidates, we need to select + if (main->valid() && found) { + if (is_main_smaller()) { + main->next(); + } else { + shards_next(); + } + } + if (!found) { + //set shards state that properly represents eof + current_shard = shards.end(); + } + smaller = is_main_smaller() ? on_main : on_shard; + return r0 == 0 && r1 == 0 ? 0 : -1; + } + + int upper_bound(const std::string &prefix, const std::string &after) override { + int r0 = main->upper_bound(prefix, after); + int r1 = 0; + if (r0 != 0) + return r0; + current_shard = shards.lower_bound(prefix); + if (current_shard != shards.end()) { + bool located = false; + if (current_shard->first == prefix) { + r1 = current_shard->second->upper_bound(after); + if (r1 != 0) + return r1; + if (current_shard->second->valid()) { + located = true; + } + } + if (!located) { + while (current_shard != shards.end()) { + r1 = current_shard->second->seek_to_first(); + if (r1 != 0) + return r1; + if (current_shard->second->valid()) + break; + ++current_shard; + } + } + } + smaller = is_main_smaller() ? on_main : on_shard; + return 0; + } + + int lower_bound(const std::string &prefix, const std::string &to) override { + int r0 = main->lower_bound(prefix, to); + int r1 = 0; + if (r0 != 0) + return r0; + current_shard = shards.lower_bound(prefix); + if (current_shard != shards.end()) { + bool located = false; + if (current_shard->first == prefix) { + r1 = current_shard->second->lower_bound(to); + if (r1 != 0) + return r1; + if (current_shard->second->valid()) { + located = true; + } + } + if (!located) { + while (current_shard != shards.end()) { + r1 = current_shard->second->seek_to_first(); + if (r1 != 0) + return r1; + if (current_shard->second->valid()) + break; + ++current_shard; + } + } + } + smaller = is_main_smaller() ? on_main : on_shard; + return 0; + } + + bool valid() override { + if (smaller == on_main) { + return main->valid(); + } else { + if (current_shard == shards.end()) + return false; + return current_shard->second->valid(); + } + }; + + int next() override { + int r; + if (smaller == on_main) { + r = main->next(); + } else { + r = shards_next(); + } + if (r != 0) + return r; + smaller = is_main_smaller() ? on_main : on_shard; + return 0; + } + + int prev() override { + int r; + bool main_was_valid = false; + if (main->valid()) { + main_was_valid = true; + r = main->prev(); + } else { + r = main->seek_to_last(); + } + if (r != 0) + return r; + + bool shards_was_valid = false; + if (shards_valid()) { + shards_was_valid = true; + r = shards_prev(); + } else { + r = shards_seek_to_last(); + } + if (r != 0) + return r; + + if (!main->valid() && !shards_valid()) { + //end, no previous. set marker so valid() can work + smaller = on_main; + return 0; + } + + //if 1 is valid, select it + //if 2 are valid select larger and advance the other + if (main->valid()) { + if (shards_valid()) { + if (is_main_smaller()) { + smaller = on_shard; + if (main_was_valid) { + if (main->valid()) { + r = main->next(); + } else { + r = main->seek_to_first(); + } + } else { + //if we have resurrected main, kill it + if (main->valid()) { + main->next(); + } + } + } else { + smaller = on_main; + if (shards_was_valid) { + if (shards_valid()) { + r = shards_next(); + } else { + r = shards_seek_to_first(); + } + } else { + //if we have resurected shards, kill it + if (shards_valid()) { + shards_next(); + } + } + } + } else { + smaller = on_main; + r = shards_seek_to_first(); + } + } else { + smaller = on_shard; + r = main->seek_to_first(); + } + return r; + } + + std::string key() override + { + if (smaller == on_main) { + return main->key(); + } else { + return current_shard->second->key(); + } + } + + std::pair<std::string,std::string> raw_key() override + { + if (smaller == on_main) { + return main->raw_key(); + } else { + return { current_shard->first, current_shard->second->key() }; + } + } + + bool raw_key_is_prefixed(const std::string &prefix) override + { + if (smaller == on_main) { + return main->raw_key_is_prefixed(prefix); + } else { + return current_shard->first == prefix; + } + } + + ceph::buffer::list value() override + { + if (smaller == on_main) { + return main->value(); + } else { + return current_shard->second->value(); + } + } + + int status() override + { + //because we already had to inspect key, it must be ok + return 0; + } + + size_t key_size() override + { + if (smaller == on_main) { + return main->key_size(); + } else { + return current_shard->second->key().size(); + } + } + size_t value_size() override + { + if (smaller == on_main) { + return main->value_size(); + } else { + return current_shard->second->value().length(); + } + } + + int shards_valid() { + if (current_shard == shards.end()) + return false; + return current_shard->second->valid(); + } + + int shards_next() { + if (current_shard == shards.end()) { + //illegal to next() on !valid() + return -1; + } + int r = 0; + r = current_shard->second->next(); + if (r != 0) + return r; + if (current_shard->second->valid()) + return 0; + //current shard exhaused, search for key + ++current_shard; + while (current_shard != shards.end()) { + r = current_shard->second->seek_to_first(); + if (r != 0) + return r; + if (current_shard->second->valid()) + break; + ++current_shard; + } + //either we found key or not, but it is success + return 0; + } + + int shards_prev() { + if (current_shard == shards.end()) { + //illegal to prev() on !valid() + return -1; + } + int r = current_shard->second->prev(); + while (r == 0) { + if (current_shard->second->valid()) { + break; + } + if (current_shard == shards.begin()) { + //we have reached pre-first element + //this makes it !valid(), but guarantees next() moves to first element + break; + } + --current_shard; + r = current_shard->second->seek_to_last(); + } + return r; + } + + int shards_seek_to_last() { + int r = 0; + current_shard = shards.end(); + if (current_shard == shards.begin()) { + //no shards at all + return 0; + } + while (current_shard != shards.begin()) { + --current_shard; + r = current_shard->second->seek_to_last(); + if (r != 0) + return r; + if (current_shard->second->valid()) { + return 0; + } + } + //no keys at all + current_shard = shards.end(); + return r; + } + + int shards_seek_to_first() { + int r = 0; + current_shard = shards.begin(); + while (current_shard != shards.end()) { + r = current_shard->second->seek_to_first(); + if (r != 0) + break; + if (current_shard->second->valid()) { + //this is the first shard that will yield some keys + break; + } + ++current_shard; + } + return r; + } +}; + +class ShardMergeIteratorImpl : public KeyValueDB::IteratorImpl { +private: + struct KeyLess { + private: + const rocksdb::Comparator* comparator; + public: + KeyLess(const rocksdb::Comparator* comparator) : comparator(comparator) { }; + + bool operator()(rocksdb::Iterator* a, rocksdb::Iterator* b) const + { + if (a->Valid()) { + if (b->Valid()) { + return comparator->Compare(a->key(), b->key()) < 0; + } else { + return true; + } + } else { + if (b->Valid()) { + return false; + } else { + return false; + } + } + } + }; + + const RocksDBStore* db; + KeyLess keyless; + string prefix; + const KeyValueDB::IteratorBounds bounds; + const rocksdb::Slice iterate_lower_bound; + const rocksdb::Slice iterate_upper_bound; + std::vector<rocksdb::Iterator*> iters; +public: + explicit ShardMergeIteratorImpl(const RocksDBStore* db, + const std::string& prefix, + const std::vector<rocksdb::ColumnFamilyHandle*>& shards, + KeyValueDB::IteratorBounds bounds_) + : db(db), keyless(db->comparator), prefix(prefix), bounds(std::move(bounds_)), + iterate_lower_bound(make_slice(bounds.lower_bound)), + iterate_upper_bound(make_slice(bounds.upper_bound)) + { + iters.reserve(shards.size()); + auto options = rocksdb::ReadOptions(); + if (db->cct->_conf->osd_rocksdb_iterator_bounds_enabled) { + if (bounds.lower_bound) { + options.iterate_lower_bound = &iterate_lower_bound; + } + if (bounds.upper_bound) { + options.iterate_upper_bound = &iterate_upper_bound; + } + } + for (auto& s : shards) { + iters.push_back(db->db->NewIterator(options, s)); + } + } + ~ShardMergeIteratorImpl() { + for (auto& it : iters) { + delete it; + } + } + int seek_to_first() override { + for (auto& it : iters) { + it->SeekToFirst(); + if (!it->status().ok()) { + return -1; + } + } + //all iterators seeked, sort + std::sort(iters.begin(), iters.end(), keyless); + return 0; + } + int seek_to_last() override { + for (auto& it : iters) { + it->SeekToLast(); + if (!it->status().ok()) { + return -1; + } + } + for (size_t i = 1; i < iters.size(); i++) { + if (iters[0]->Valid()) { + if (iters[i]->Valid()) { + if (keyless(iters[0], iters[i])) { + swap(iters[0], iters[i]); + } + } else { + //iters[i] empty + } + } else { + if (iters[i]->Valid()) { + swap(iters[0], iters[i]); + } + } + //it might happen that cf was empty + if (iters[i]->Valid()) { + iters[i]->Next(); + } + } + //no need to sort, as at most 1 iterator is valid now + return 0; + } + int upper_bound(const string &after) override { + rocksdb::Slice slice_bound(after); + for (auto& it : iters) { + it->Seek(slice_bound); + if (it->Valid() && it->key() == after) { + it->Next(); + } + if (!it->status().ok()) { + return -1; + } + } + std::sort(iters.begin(), iters.end(), keyless); + return 0; + } + int lower_bound(const string &to) override { + rocksdb::Slice slice_bound(to); + for (auto& it : iters) { + it->Seek(slice_bound); + if (!it->status().ok()) { + return -1; + } + } + std::sort(iters.begin(), iters.end(), keyless); + return 0; + } + int next() override { + int r = -1; + if (iters[0]->Valid()) { + iters[0]->Next(); + if (iters[0]->status().ok()) { + r = 0; + //bubble up + for (size_t i = 0; i < iters.size() - 1; i++) { + if (keyless(iters[i], iters[i + 1])) { + //matches, fixed + break; + } + std::swap(iters[i], iters[i + 1]); + } + } + } + return r; + } + // iters are sorted, so + // a[0] < b[0] < c[0] < d[0] + // a[0] > a[-1], a[0] > b[-1], a[0] > c[-1], a[0] > d[-1] + // so, prev() will be one of: + // a[-1], b[-1], c[-1], d[-1] + // prev() will be the one that is *largest* of them + // + // alg: + // 1. go prev() on each iterator we can + // 2. select largest key from those iterators + // 3. go next() on all iterators except (2) + // 4. sort + int prev() override { + std::vector<rocksdb::Iterator*> prev_done; + //1 + for (auto it: iters) { + if (it->Valid()) { + it->Prev(); + if (it->Valid()) { + prev_done.push_back(it); + } else { + it->SeekToFirst(); + } + } else { + it->SeekToLast(); + if (it->Valid()) { + prev_done.push_back(it); + } + } + } + if (prev_done.size() == 0) { + /* there is no previous element */ + if (iters[0]->Valid()) { + iters[0]->Prev(); + ceph_assert(!iters[0]->Valid()); + } + return 0; + } + //2,3 + rocksdb::Iterator* highest = prev_done[0]; + for (size_t i = 1; i < prev_done.size(); i++) { + if (keyless(highest, prev_done[i])) { + highest->Next(); + highest = prev_done[i]; + } else { + prev_done[i]->Next(); + } + } + //4 + //insert highest in the beginning, and shift values until we pick highest + //untouched rest is sorted - we just prev()/next() them + rocksdb::Iterator* hold = highest; + for (size_t i = 0; i < iters.size(); i++) { + std::swap(hold, iters[i]); + if (hold == highest) break; + } + ceph_assert(hold == highest); + return 0; + } + bool valid() override { + return iters[0]->Valid(); + } + string key() override { + return iters[0]->key().ToString(); + } + std::pair<std::string, std::string> raw_key() override { + return make_pair(prefix, key()); + } + bufferlist value() override { + return to_bufferlist(iters[0]->value()); + } + bufferptr value_as_ptr() override { + rocksdb::Slice val = iters[0]->value(); + return bufferptr(val.data(), val.size()); + } + int status() override { + return iters[0]->status().ok() ? 0 : -1; + } +}; + +KeyValueDB::Iterator RocksDBStore::get_iterator(const std::string& prefix, IteratorOpts opts, IteratorBounds bounds) +{ + auto cf_it = cf_handles.find(prefix); + if (cf_it != cf_handles.end()) { + rocksdb::ColumnFamilyHandle* cf = nullptr; + if (cf_it->second.handles.size() == 1) { + cf = cf_it->second.handles[0]; + } else if (cct->_conf->osd_rocksdb_iterator_bounds_enabled) { + cf = get_cf_handle(prefix, bounds); + } + if (cf) { + return std::make_shared<CFIteratorImpl>( + this, + prefix, + cf, + std::move(bounds)); + } else { + return std::make_shared<ShardMergeIteratorImpl>( + this, + prefix, + cf_it->second.handles, + std::move(bounds)); + } + } else { + return KeyValueDB::get_iterator(prefix, opts); + } +} + +rocksdb::Iterator* RocksDBStore::new_shard_iterator(rocksdb::ColumnFamilyHandle* cf) +{ + return db->NewIterator(rocksdb::ReadOptions(), cf); +} + +RocksDBStore::WholeSpaceIterator RocksDBStore::get_wholespace_iterator(IteratorOpts opts) +{ + if (cf_handles.size() == 0) { + return std::make_shared<RocksDBWholeSpaceIteratorImpl>( + this, default_cf, opts); + } else { + return std::make_shared<WholeMergeIteratorImpl>(this); + } +} + +RocksDBStore::WholeSpaceIterator RocksDBStore::get_default_cf_iterator() +{ + return std::make_shared<RocksDBWholeSpaceIteratorImpl>(this, default_cf, 0); +} + +int RocksDBStore::prepare_for_reshard(const std::string& new_sharding, + RocksDBStore::columns_t& to_process_columns) +{ + //0. lock db from opening + //1. list existing columns + //2. apply merge operator to (main + columns) opts + //3. prepare std::vector<rocksdb::ColumnFamilyDescriptor> existing_cfs + //4. open db, acquire existing column handles + //5. calculate missing columns + //6. create missing columns + //7. construct cf_handles according to new sharding + //8. check is all cf_handles are filled + + bool b; + std::vector<ColumnFamily> new_sharding_def; + char const* error_position; + std::string error_msg; + b = parse_sharding_def(new_sharding, new_sharding_def, &error_position, &error_msg); + if (!b) { + dout(1) << __func__ << " bad sharding: " << dendl; + dout(1) << __func__ << new_sharding << dendl; + dout(1) << __func__ << std::string(error_position - &new_sharding[0], ' ') << "^" << error_msg << dendl; + return -EINVAL; + } + + //0. lock db from opening + std::string stored_sharding_text; + rocksdb::ReadFileToString(env, + sharding_def_file, + &stored_sharding_text); + if (stored_sharding_text.find(resharding_column_lock) == string::npos) { + rocksdb::Status status; + if (stored_sharding_text.size() != 0) + stored_sharding_text += " "; + stored_sharding_text += resharding_column_lock; + env->CreateDir(sharding_def_dir); + status = rocksdb::WriteStringToFile(env, stored_sharding_text, + sharding_def_file, true); + if (!status.ok()) { + derr << __func__ << " cannot write to " << sharding_def_file << dendl; + return -EIO; + } + } + + //1. list existing columns + + rocksdb::Status status; + std::vector<std::string> existing_columns; + rocksdb::Options opt; + int r = load_rocksdb_options(false, opt); + if (r) { + dout(1) << __func__ << " load rocksdb options failed" << dendl; + return r; + } + status = rocksdb::DB::ListColumnFamilies(rocksdb::DBOptions(opt), path, &existing_columns); + if (!status.ok()) { + derr << "Unable to list column families: " << status.ToString() << dendl; + return -EINVAL; + } + dout(5) << "existing columns = " << existing_columns << dendl; + + //2. apply merge operator to (main + columns) opts + //3. prepare std::vector<rocksdb::ColumnFamilyDescriptor> cfs_to_open + + std::vector<rocksdb::ColumnFamilyDescriptor> cfs_to_open; + for (const auto& full_name : existing_columns) { + //split col_name to <prefix>-<number> + std::string base_name; + size_t pos = full_name.find('-'); + if (std::string::npos == pos) + base_name = full_name; + else + base_name = full_name.substr(0,pos); + + rocksdb::ColumnFamilyOptions cf_opt(opt); + // search if we have options for this column + std::string options; + for (const auto& nsd : new_sharding_def) { + if (nsd.name == base_name) { + options = nsd.options; + break; + } + } + int r = update_column_family_options(base_name, options, &cf_opt); + if (r != 0) { + return r; + } + cfs_to_open.emplace_back(full_name, cf_opt); + } + + //4. open db, acquire existing column handles + std::vector<rocksdb::ColumnFamilyHandle*> handles; + status = rocksdb::DB::Open(rocksdb::DBOptions(opt), + path, cfs_to_open, &handles, &db); + if (!status.ok()) { + derr << status.ToString() << dendl; + return -EINVAL; + } + for (size_t i = 0; i < cfs_to_open.size(); i++) { + dout(10) << "column " << cfs_to_open[i].name << " handle " << (void*)handles[i] << dendl; + } + + //5. calculate missing columns + std::vector<std::string> new_sharding_columns; + std::vector<std::string> missing_columns; + sharding_def_to_columns(new_sharding_def, + new_sharding_columns); + dout(5) << "target columns = " << new_sharding_columns << dendl; + for (const auto& n : new_sharding_columns) { + bool found = false; + for (const auto& e : existing_columns) { + if (n == e) { + found = true; + break; + } + } + if (!found) { + missing_columns.push_back(n); + } + } + dout(5) << "missing columns = " << missing_columns << dendl; + + //6. create missing columns + for (const auto& full_name : missing_columns) { + std::string base_name; + size_t pos = full_name.find('-'); + if (std::string::npos == pos) + base_name = full_name; + else + base_name = full_name.substr(0,pos); + + rocksdb::ColumnFamilyOptions cf_opt(opt); + // search if we have options for this column + std::string options; + for (const auto& nsd : new_sharding_def) { + if (nsd.name == base_name) { + options = nsd.options; + break; + } + } + int r = update_column_family_options(base_name, options, &cf_opt); + if (r != 0) { + return r; + } + rocksdb::ColumnFamilyHandle *cf; + status = db->CreateColumnFamily(cf_opt, full_name, &cf); + if (!status.ok()) { + derr << __func__ << " Failed to create rocksdb column family: " + << full_name << dendl; + return -EINVAL; + } + dout(10) << "created column " << full_name << " handle = " << (void*)cf << dendl; + existing_columns.push_back(full_name); + handles.push_back(cf); + } + + //7. construct cf_handles according to new sharding + for (size_t i = 0; i < existing_columns.size(); i++) { + std::string full_name = existing_columns[i]; + rocksdb::ColumnFamilyHandle *cf = handles[i]; + std::string base_name; + size_t shard_idx = 0; + size_t pos = full_name.find('-'); + dout(10) << "processing column " << full_name << dendl; + if (std::string::npos == pos) { + base_name = full_name; + } else { + base_name = full_name.substr(0,pos); + shard_idx = atoi(full_name.substr(pos+1).c_str()); + } + if (rocksdb::kDefaultColumnFamilyName == base_name) { + default_cf = handles[i]; + must_close_default_cf = true; + std::unique_ptr<rocksdb::ColumnFamilyHandle, cf_deleter_t> ptr{ + cf, [](rocksdb::ColumnFamilyHandle*) {}}; + to_process_columns.emplace(full_name, std::move(ptr)); + } else { + for (const auto& nsd : new_sharding_def) { + if (nsd.name == base_name) { + if (shard_idx < nsd.shard_cnt) { + add_column_family(base_name, nsd.hash_l, nsd.hash_h, shard_idx, cf); + } else { + //ignore columns with index larger then shard count + } + break; + } + } + std::unique_ptr<rocksdb::ColumnFamilyHandle, cf_deleter_t> ptr{ + cf, [this](rocksdb::ColumnFamilyHandle* handle) { + db->DestroyColumnFamilyHandle(handle); + }}; + to_process_columns.emplace(full_name, std::move(ptr)); + } + } + + //8. check if all cf_handles are filled + for (const auto& col : cf_handles) { + for (size_t i = 0; i < col.second.handles.size(); i++) { + if (col.second.handles[i] == nullptr) { + derr << "missing handle for column " << col.first << " shard " << i << dendl; + return -EIO; + } + } + } + return 0; +} + +int RocksDBStore::reshard_cleanup(const RocksDBStore::columns_t& current_columns) +{ + std::vector<std::string> new_sharding_columns; + for (const auto& [name, handle] : cf_handles) { + if (handle.handles.size() == 1) { + new_sharding_columns.push_back(name); + } else { + for (size_t i = 0; i < handle.handles.size(); i++) { + new_sharding_columns.push_back(name + "-" + to_string(i)); + } + } + } + + for (auto& [name, handle] : current_columns) { + auto found = std::find(new_sharding_columns.begin(), + new_sharding_columns.end(), + name) != new_sharding_columns.end(); + if (found || name == rocksdb::kDefaultColumnFamilyName) { + dout(5) << "Column " << name << " is part of new sharding." << dendl; + continue; + } + dout(5) << "Column " << name << " not part of new sharding. Deleting." << dendl; + + // verify that column is empty + std::unique_ptr<rocksdb::Iterator> it{ + db->NewIterator(rocksdb::ReadOptions(), handle.get())}; + ceph_assert(it); + it->SeekToFirst(); + ceph_assert(!it->Valid()); + + if (rocksdb::Status status = db->DropColumnFamily(handle.get()); !status.ok()) { + derr << __func__ << " Failed to drop column: " << name << dendl; + return -EINVAL; + } + } + return 0; +} + +int RocksDBStore::reshard(const std::string& new_sharding, const RocksDBStore::resharding_ctrl* ctrl_in) +{ + + resharding_ctrl ctrl = ctrl_in ? *ctrl_in : resharding_ctrl(); + size_t bytes_in_batch = 0; + size_t keys_in_batch = 0; + size_t bytes_per_iterator = 0; + size_t keys_per_iterator = 0; + size_t keys_processed = 0; + size_t keys_moved = 0; + + auto flush_batch = [&](rocksdb::WriteBatch* batch) { + dout(10) << "flushing batch, " << keys_in_batch << " keys, for " + << bytes_in_batch << " bytes" << dendl; + rocksdb::WriteOptions woptions; + woptions.sync = true; + rocksdb::Status s = db->Write(woptions, batch); + ceph_assert(s.ok()); + bytes_in_batch = 0; + keys_in_batch = 0; + batch->Clear(); + }; + + auto process_column = [&](rocksdb::ColumnFamilyHandle* handle, + const std::string& fixed_prefix) + { + dout(5) << " column=" << (void*)handle << " prefix=" << fixed_prefix << dendl; + std::unique_ptr<rocksdb::Iterator> it{ + db->NewIterator(rocksdb::ReadOptions(), handle)}; + ceph_assert(it); + + rocksdb::WriteBatch bat; + for (it->SeekToFirst(); it->Valid(); it->Next()) { + rocksdb::Slice raw_key = it->key(); + dout(30) << "key=" << pretty_binary_string(raw_key.ToString()) << dendl; + //check if need to refresh iterator + if (bytes_per_iterator >= ctrl.bytes_per_iterator || + keys_per_iterator >= ctrl.keys_per_iterator) { + dout(8) << "refreshing iterator" << dendl; + bytes_per_iterator = 0; + keys_per_iterator = 0; + std::string raw_key_str = raw_key.ToString(); + it.reset(db->NewIterator(rocksdb::ReadOptions(), handle)); + ceph_assert(it); + it->Seek(raw_key_str); + ceph_assert(it->Valid()); + raw_key = it->key(); + } + rocksdb::Slice value = it->value(); + std::string prefix, key; + if (fixed_prefix.size() == 0) { + split_key(raw_key, &prefix, &key); + } else { + prefix = fixed_prefix; + key = raw_key.ToString(); + } + keys_processed++; + if ((keys_processed % 10000) == 0) { + dout(10) << "processed " << keys_processed << " keys, moved " << keys_moved << dendl; + } + rocksdb::ColumnFamilyHandle* new_handle = get_cf_handle(prefix, key); + if (new_handle == nullptr) { + new_handle = default_cf; + } + if (handle == new_handle) { + continue; + } + std::string new_raw_key; + if (new_handle == default_cf) { + new_raw_key = combine_strings(prefix, key); + } else { + new_raw_key = key; + } + bat.Delete(handle, raw_key); + bat.Put(new_handle, new_raw_key, value); + dout(25) << "moving " << (void*)handle << "/" << pretty_binary_string(raw_key.ToString()) << + " to " << (void*)new_handle << "/" << pretty_binary_string(new_raw_key) << + " size " << value.size() << dendl; + keys_moved++; + bytes_in_batch += new_raw_key.size() * 2 + value.size(); + keys_in_batch++; + bytes_per_iterator += new_raw_key.size() * 2 + value.size(); + keys_per_iterator++; + + //check if need to write batch + if (bytes_in_batch >= ctrl.bytes_per_batch || + keys_in_batch >= ctrl.keys_per_batch) { + flush_batch(&bat); + if (ctrl.unittest_fail_after_first_batch) { + return -1000; + } + } + } + if (bat.Count() > 0) { + flush_batch(&bat); + } + return 0; + }; + + auto close_column_handles = make_scope_guard([this] { + cf_handles.clear(); + close(); + }); + columns_t to_process_columns; + int r = prepare_for_reshard(new_sharding, to_process_columns); + if (r != 0) { + dout(1) << "failed to prepare db for reshard" << dendl; + return r; + } + + for (auto& [name, handle] : to_process_columns) { + dout(5) << "Processing column=" << name + << " handle=" << handle.get() << dendl; + if (name == rocksdb::kDefaultColumnFamilyName) { + ceph_assert(handle.get() == default_cf); + r = process_column(default_cf, std::string()); + } else { + std::string fixed_prefix = name.substr(0, name.find('-')); + dout(10) << "Prefix: " << fixed_prefix << dendl; + r = process_column(handle.get(), fixed_prefix); + } + if (r != 0) { + derr << "Error processing column " << name << dendl; + return r; + } + if (ctrl.unittest_fail_after_processing_column) { + return -1001; + } + } + + r = reshard_cleanup(to_process_columns); + if (r != 0) { + dout(5) << "failed to cleanup after reshard" << dendl; + return r; + } + + if (ctrl.unittest_fail_after_successful_processing) { + return -1002; + } + env->CreateDir(sharding_def_dir); + if (auto status = rocksdb::WriteStringToFile(env, new_sharding, + sharding_def_file, true); + !status.ok()) { + derr << __func__ << " cannot write to " << sharding_def_file << dendl; + return -EIO; + } + + return r; +} + +bool RocksDBStore::get_sharding(std::string& sharding) { + rocksdb::Status status; + std::string stored_sharding_text; + bool result = false; + sharding.clear(); + + status = env->FileExists(sharding_def_file); + if (status.ok()) { + status = rocksdb::ReadFileToString(env, + sharding_def_file, + &stored_sharding_text); + if(status.ok()) { + result = true; + sharding = stored_sharding_text; + } + } + return result; +} |