+++ /dev/null
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-/*
-* Ceph - scalable distributed file system
-*
-* Copyright (C) 2012 Inktank, Inc.
-*
-* This is free software; you can redistribute it and/or
-* modify it under the terms of the GNU Lesser General Public
-* License version 2.1, as published by the Free Software
-* Foundation. See file COPYING.
-*/
-#ifndef CEPH_MONITOR_DB_STORE_H
-#define CEPH_MONITOR_DB_STORE_H
-
-#include "include/types.h"
-#include "include/buffer.h"
-#include <set>
-#include <map>
-#include <string>
-#include <boost/scoped_ptr.hpp>
-#include <sstream>
-#include <fstream>
-#include "kv/KeyValueDB.h"
-
-#include "include/assert.h"
-#include "common/Formatter.h"
-#include "common/Finisher.h"
-#include "common/errno.h"
-#include "common/debug.h"
-#include "common/safe_io.h"
-
-#define dout_context g_ceph_context
-
-class MonitorDBStore
-{
- string path;
- boost::scoped_ptr<KeyValueDB> db;
- bool do_dump;
- int dump_fd_binary;
- std::ofstream dump_fd_json;
- JSONFormatter dump_fmt;
-
-
- Finisher io_work;
-
- bool is_open;
-
- public:
-
- struct Op {
- uint8_t type;
- string prefix;
- string key, endkey;
- bufferlist bl;
-
- Op()
- : type(0) { }
- Op(int t, string p, string k)
- : type(t), prefix(p), key(k) { }
- Op(int t, const string& p, string k, bufferlist& b)
- : type(t), prefix(p), key(k), bl(b) { }
- Op(int t, const string& p, string start, string end)
- : type(t), prefix(p), key(start), endkey(end) { }
-
- void encode(bufferlist& encode_bl) const {
- ENCODE_START(2, 1, encode_bl);
- ::encode(type, encode_bl);
- ::encode(prefix, encode_bl);
- ::encode(key, encode_bl);
- ::encode(bl, encode_bl);
- ::encode(endkey, encode_bl);
- ENCODE_FINISH(encode_bl);
- }
-
- void decode(bufferlist::iterator& decode_bl) {
- DECODE_START(2, decode_bl);
- ::decode(type, decode_bl);
- ::decode(prefix, decode_bl);
- ::decode(key, decode_bl);
- ::decode(bl, decode_bl);
- if (struct_v >= 2)
- ::decode(endkey, decode_bl);
- DECODE_FINISH(decode_bl);
- }
-
- void dump(Formatter *f) const {
- f->dump_int("type", type);
- f->dump_string("prefix", prefix);
- f->dump_string("key", key);
- if (endkey.length())
- f->dump_string("endkey", endkey);
- }
-
- static void generate_test_instances(list<Op*>& ls) {
- ls.push_back(new Op);
- // we get coverage here from the Transaction instances
- }
- };
-
- struct Transaction;
- typedef ceph::shared_ptr<Transaction> TransactionRef;
- struct Transaction {
- list<Op> ops;
- uint64_t bytes, keys;
-
- Transaction() : bytes(0), keys(0) {}
-
- enum {
- OP_PUT = 1,
- OP_ERASE = 2,
- OP_COMPACT = 3,
- };
-
- void put(string prefix, string key, bufferlist& bl) {
- ops.push_back(Op(OP_PUT, prefix, key, bl));
- ++keys;
- bytes += prefix.length() + key.length() + bl.length();
- }
-
- void put(string prefix, version_t ver, bufferlist& bl) {
- ostringstream os;
- os << ver;
- put(prefix, os.str(), bl);
- }
-
- void put(string prefix, string key, version_t ver) {
- bufferlist bl;
- ::encode(ver, bl);
- put(prefix, key, bl);
- }
-
- void erase(string prefix, string key) {
- ops.push_back(Op(OP_ERASE, prefix, key));
- ++keys;
- bytes += prefix.length() + key.length();
- }
-
- void erase(string prefix, version_t ver) {
- ostringstream os;
- os << ver;
- erase(prefix, os.str());
- }
-
- void compact_prefix(string prefix) {
- ops.push_back(Op(OP_COMPACT, prefix, string()));
- }
-
- void compact_range(string prefix, string start, string end) {
- ops.push_back(Op(OP_COMPACT, prefix, start, end));
- }
-
- void encode(bufferlist& bl) const {
- ENCODE_START(2, 1, bl);
- ::encode(ops, bl);
- ::encode(bytes, bl);
- ::encode(keys, bl);
- ENCODE_FINISH(bl);
- }
-
- void decode(bufferlist::iterator& bl) {
- DECODE_START(2, bl);
- ::decode(ops, bl);
- if (struct_v >= 2) {
- ::decode(bytes, bl);
- ::decode(keys, bl);
- }
- DECODE_FINISH(bl);
- }
-
- static void generate_test_instances(list<Transaction*>& ls) {
- ls.push_back(new Transaction);
- ls.push_back(new Transaction);
- bufferlist bl;
- bl.append("value");
- ls.back()->put("prefix", "key", bl);
- ls.back()->erase("prefix2", "key2");
- ls.back()->compact_prefix("prefix3");
- ls.back()->compact_range("prefix4", "from", "to");
- }
-
- void append(TransactionRef other) {
- ops.splice(ops.end(), other->ops);
- keys += other->keys;
- bytes += other->bytes;
- }
-
- void append_from_encoded(bufferlist& bl) {
- auto other(std::make_shared<Transaction>());
- bufferlist::iterator it = bl.begin();
- other->decode(it);
- append(other);
- }
-
- bool empty() {
- return (size() == 0);
- }
-
- size_t size() const {
- return ops.size();
- }
- uint64_t get_keys() const {
- return keys;
- }
- uint64_t get_bytes() const {
- return bytes;
- }
-
- void dump(ceph::Formatter *f, bool dump_val=false) const {
- f->open_object_section("transaction");
- f->open_array_section("ops");
- list<Op>::const_iterator it;
- int op_num = 0;
- for (it = ops.begin(); it != ops.end(); ++it) {
- const Op& op = *it;
- f->open_object_section("op");
- f->dump_int("op_num", op_num++);
- switch (op.type) {
- case OP_PUT:
- {
- f->dump_string("type", "PUT");
- f->dump_string("prefix", op.prefix);
- f->dump_string("key", op.key);
- f->dump_unsigned("length", op.bl.length());
- if (dump_val) {
- ostringstream os;
- op.bl.hexdump(os);
- f->dump_string("bl", os.str());
- }
- }
- break;
- case OP_ERASE:
- {
- f->dump_string("type", "ERASE");
- f->dump_string("prefix", op.prefix);
- f->dump_string("key", op.key);
- }
- break;
- case OP_COMPACT:
- {
- f->dump_string("type", "COMPACT");
- f->dump_string("prefix", op.prefix);
- f->dump_string("start", op.key);
- f->dump_string("end", op.endkey);
- }
- break;
- default:
- {
- f->dump_string("type", "unknown");
- f->dump_unsigned("op_code", op.type);
- break;
- }
- }
- f->close_section();
- }
- f->close_section();
- f->dump_unsigned("num_keys", keys);
- f->dump_unsigned("num_bytes", bytes);
- f->close_section();
- }
- };
-
- int apply_transaction(MonitorDBStore::TransactionRef t) {
- KeyValueDB::Transaction dbt = db->get_transaction();
-
- if (do_dump) {
- if (!g_conf->mon_debug_dump_json) {
- bufferlist bl;
- t->encode(bl);
- bl.write_fd(dump_fd_binary);
- } else {
- t->dump(&dump_fmt, true);
- dump_fmt.flush(dump_fd_json);
- dump_fd_json.flush();
- }
- }
-
- list<pair<string, pair<string,string> > > compact;
- for (list<Op>::const_iterator it = t->ops.begin();
- it != t->ops.end();
- ++it) {
- const Op& op = *it;
- switch (op.type) {
- case Transaction::OP_PUT:
- dbt->set(op.prefix, op.key, op.bl);
- break;
- case Transaction::OP_ERASE:
- dbt->rmkey(op.prefix, op.key);
- break;
- case Transaction::OP_COMPACT:
- compact.push_back(make_pair(op.prefix, make_pair(op.key, op.endkey)));
- break;
- default:
- derr << __func__ << " unknown op type " << op.type << dendl;
- ceph_abort();
- break;
- }
- }
- int r = db->submit_transaction_sync(dbt);
- if (r >= 0) {
- while (!compact.empty()) {
- if (compact.front().second.first == string() &&
- compact.front().second.second == string())
- db->compact_prefix_async(compact.front().first);
- else
- db->compact_range_async(compact.front().first, compact.front().second.first, compact.front().second.second);
- compact.pop_front();
- }
- } else {
- assert(0 == "failed to write to db");
- }
- return r;
- }
-
- struct C_DoTransaction : public Context {
- MonitorDBStore *store;
- MonitorDBStore::TransactionRef t;
- Context *oncommit;
- C_DoTransaction(MonitorDBStore *s, MonitorDBStore::TransactionRef t,
- Context *f)
- : store(s), t(t), oncommit(f)
- {}
- void finish(int r) override {
- /* The store serializes writes. Each transaction is handled
- * sequentially by the io_work Finisher. If a transaction takes longer
- * to apply its state to permanent storage, then no other transaction
- * will be handled meanwhile.
- *
- * We will now randomly inject random delays. We can safely sleep prior
- * to applying the transaction as it won't break the model.
- */
- double delay_prob = g_conf->mon_inject_transaction_delay_probability;
- if (delay_prob && (rand() % 10000 < delay_prob * 10000.0)) {
- utime_t delay;
- double delay_max = g_conf->mon_inject_transaction_delay_max;
- delay.set_from_double(delay_max * (double)(rand() % 10000) / 10000.0);
- lsubdout(g_ceph_context, mon, 1)
- << "apply_transaction will be delayed for " << delay
- << " seconds" << dendl;
- delay.sleep();
- }
- int ret = store->apply_transaction(t);
- oncommit->complete(ret);
- }
- };
-
- /**
- * queue transaction
- *
- * Queue a transaction to commit asynchronously. Trigger a context
- * on completion (without any locks held).
- */
- void queue_transaction(MonitorDBStore::TransactionRef t,
- Context *oncommit) {
- io_work.queue(new C_DoTransaction(this, t, oncommit));
- }
-
- /**
- * block and flush all io activity
- */
- void flush() {
- io_work.wait_for_empty();
- }
-
- class StoreIteratorImpl {
- protected:
- bool done;
- pair<string,string> last_key;
- bufferlist crc_bl;
-
- StoreIteratorImpl() : done(false) { }
- virtual ~StoreIteratorImpl() { }
-
- bool add_chunk_entry(TransactionRef tx,
- string &prefix,
- string &key,
- bufferlist &value,
- uint64_t max) {
- auto tmp(std::make_shared<Transaction>());
- bufferlist tmp_bl;
- tmp->put(prefix, key, value);
- tmp->encode(tmp_bl);
-
- bufferlist tx_bl;
- tx->encode(tx_bl);
-
- size_t len = tx_bl.length() + tmp_bl.length();
-
- if (!tx->empty() && (len > max)) {
- return false;
- }
-
- tx->append(tmp);
- last_key.first = prefix;
- last_key.second = key;
-
- if (g_conf->mon_sync_debug) {
- ::encode(prefix, crc_bl);
- ::encode(key, crc_bl);
- ::encode(value, crc_bl);
- }
-
- return true;
- }
-
- virtual bool _is_valid() = 0;
-
- public:
- __u32 crc() {
- if (g_conf->mon_sync_debug)
- return crc_bl.crc32c(0);
- return 0;
- }
- pair<string,string> get_last_key() {
- return last_key;
- }
- virtual bool has_next_chunk() {
- return !done && _is_valid();
- }
- virtual void get_chunk_tx(TransactionRef tx, uint64_t max) = 0;
- virtual pair<string,string> get_next_key() = 0;
- };
- typedef ceph::shared_ptr<StoreIteratorImpl> Synchronizer;
-
- class WholeStoreIteratorImpl : public StoreIteratorImpl {
- KeyValueDB::WholeSpaceIterator iter;
- set<string> sync_prefixes;
-
- public:
- WholeStoreIteratorImpl(KeyValueDB::WholeSpaceIterator iter,
- set<string> &prefixes)
- : StoreIteratorImpl(),
- iter(iter),
- sync_prefixes(prefixes)
- { }
-
- ~WholeStoreIteratorImpl() override { }
-
- /**
- * Obtain a chunk of the store
- *
- * @param bl Encoded transaction that will recreate the chunk
- * @param first_key Pair containing the first key to obtain, and that
- * will contain the first key in the chunk (that may
- * differ from the one passed on to the function)
- * @param last_key[out] Last key in the chunk
- */
- void get_chunk_tx(TransactionRef tx, uint64_t max) override {
- assert(done == false);
- assert(iter->valid() == true);
-
- while (iter->valid()) {
- string prefix(iter->raw_key().first);
- string key(iter->raw_key().second);
- if (sync_prefixes.count(prefix)) {
- bufferlist value = iter->value();
- if (!add_chunk_entry(tx, prefix, key, value, max))
- return;
- }
- iter->next();
- }
- assert(iter->valid() == false);
- done = true;
- }
-
- pair<string,string> get_next_key() override {
- assert(iter->valid());
-
- for (; iter->valid(); iter->next()) {
- pair<string,string> r = iter->raw_key();
- if (sync_prefixes.count(r.first) > 0) {
- iter->next();
- return r;
- }
- }
- return pair<string,string>();
- }
-
- bool _is_valid() override {
- return iter->valid();
- }
- };
-
- Synchronizer get_synchronizer(pair<string,string> &key,
- set<string> &prefixes) {
- KeyValueDB::WholeSpaceIterator iter;
- iter = db->get_iterator();
-
- if (!key.first.empty() && !key.second.empty())
- iter->upper_bound(key.first, key.second);
- else
- iter->seek_to_first();
-
- return ceph::shared_ptr<StoreIteratorImpl>(
- new WholeStoreIteratorImpl(iter, prefixes)
- );
- }
-
- KeyValueDB::Iterator get_iterator(const string &prefix) {
- assert(!prefix.empty());
- KeyValueDB::Iterator iter = db->get_iterator(prefix);
- iter->seek_to_first();
- return iter;
- }
-
- KeyValueDB::WholeSpaceIterator get_iterator() {
- KeyValueDB::WholeSpaceIterator iter;
- iter = db->get_iterator();
- iter->seek_to_first();
- return iter;
- }
-
- int get(const string& prefix, const string& key, bufferlist& bl) {
- assert(bl.length() == 0);
- return db->get(prefix, key, &bl);
- }
-
- int get(const string& prefix, const version_t ver, bufferlist& bl) {
- ostringstream os;
- os << ver;
- return get(prefix, os.str(), bl);
- }
-
- version_t get(const string& prefix, const string& key) {
- bufferlist bl;
- int err = get(prefix, key, bl);
- if (err < 0) {
- if (err == -ENOENT) // if key doesn't exist, assume its value is 0
- return 0;
- // we're not expecting any other negative return value, and we can't
- // just return a negative value if we're returning a version_t
- generic_dout(0) << "MonitorDBStore::get() error obtaining"
- << " (" << prefix << ":" << key << "): "
- << cpp_strerror(err) << dendl;
- assert(0 == "error obtaining key");
- }
-
- assert(bl.length());
- version_t ver;
- bufferlist::iterator p = bl.begin();
- ::decode(ver, p);
- return ver;
- }
-
- bool exists(const string& prefix, const string& key) {
- KeyValueDB::Iterator it = db->get_iterator(prefix);
- int err = it->lower_bound(key);
- if (err < 0)
- return false;
-
- return (it->valid() && it->key() == key);
- }
-
- bool exists(const string& prefix, version_t ver) {
- ostringstream os;
- os << ver;
- return exists(prefix, os.str());
- }
-
- string combine_strings(const string& prefix, const string& value) {
- string out = prefix;
- out.push_back('_');
- out.append(value);
- return out;
- }
-
- string combine_strings(const string& prefix, const version_t ver) {
- ostringstream os;
- os << ver;
- return combine_strings(prefix, os.str());
- }
-
- void clear(set<string>& prefixes) {
- set<string>::iterator iter;
- KeyValueDB::Transaction dbt = db->get_transaction();
-
- for (iter = prefixes.begin(); iter != prefixes.end(); ++iter) {
- dbt->rmkeys_by_prefix((*iter));
- }
- int r = db->submit_transaction_sync(dbt);
- assert(r >= 0);
- }
-
- void _open(string kv_type) {
- string::const_reverse_iterator rit;
- int pos = 0;
- for (rit = path.rbegin(); rit != path.rend(); ++rit, ++pos) {
- if (*rit != '/')
- break;
- }
- ostringstream os;
- os << path.substr(0, path.size() - pos) << "/store.db";
- string full_path = os.str();
-
- KeyValueDB *db_ptr = KeyValueDB::create(g_ceph_context,
- kv_type,
- full_path);
- if (!db_ptr) {
- derr << __func__ << " error initializing "
- << kv_type << " db back storage in "
- << full_path << dendl;
- assert(0 == "MonitorDBStore: error initializing keyvaluedb back storage");
- }
- db.reset(db_ptr);
-
- if (g_conf->mon_debug_dump_transactions) {
- if (!g_conf->mon_debug_dump_json) {
- dump_fd_binary = ::open(
- g_conf->mon_debug_dump_location.c_str(),
- O_CREAT|O_APPEND|O_WRONLY, 0644);
- if (dump_fd_binary < 0) {
- dump_fd_binary = -errno;
- derr << "Could not open log file, got "
- << cpp_strerror(dump_fd_binary) << dendl;
- }
- } else {
- dump_fmt.reset();
- dump_fmt.open_array_section("dump");
- dump_fd_json.open(g_conf->mon_debug_dump_location.c_str());
- }
- do_dump = true;
- }
- if (kv_type == "rocksdb")
- db->init(g_conf->mon_rocksdb_options);
- else
- db->init();
-
-
- }
-
- int open(ostream &out) {
- string kv_type;
- int r = read_meta("kv_backend", &kv_type);
- if (r < 0 || kv_type.empty()) {
- // assume old monitors that did not mark the type were leveldb.
- kv_type = "leveldb";
- r = write_meta("kv_backend", kv_type);
- if (r < 0)
- return r;
- }
- _open(kv_type);
- r = db->open(out);
- if (r < 0)
- return r;
-
- // Monitors are few in number, so the resource cost of exposing
- // very detailed stats is low: ramp up the priority of all the
- // KV store's perf counters. Do this after open, because backend may
- // not have constructed PerfCounters earlier.
- if (db->get_perf_counters()) {
- db->get_perf_counters()->set_prio_adjust(
- PerfCountersBuilder::PRIO_USEFUL - PerfCountersBuilder::PRIO_DEBUGONLY);
- }
-
- io_work.start();
- is_open = true;
- return 0;
- }
-
- int create_and_open(ostream &out) {
- // record the type before open
- string kv_type;
- int r = read_meta("kv_backend", &kv_type);
- if (r < 0) {
- kv_type = g_conf->mon_keyvaluedb;
- r = write_meta("kv_backend", kv_type);
- if (r < 0)
- return r;
- }
- _open(kv_type);
- r = db->create_and_open(out);
- if (r < 0)
- return r;
- io_work.start();
- is_open = true;
- return 0;
- }
-
- void close() {
- // there should be no work queued!
- io_work.stop();
- is_open = false;
- db.reset(NULL);
- }
-
- void compact() {
- db->compact();
- }
-
- void compact_prefix(const string& prefix) {
- db->compact_prefix(prefix);
- }
-
- uint64_t get_estimated_size(map<string, uint64_t> &extras) {
- return db->get_estimated_size(extras);
- }
-
- /**
- * write_meta - write a simple configuration key out-of-band
- *
- * Write a simple key/value pair for basic store configuration
- * (e.g., a uuid or magic number) to an unopened/unmounted store.
- * The default implementation writes this to a plaintext file in the
- * path.
- *
- * A newline is appended.
- *
- * @param key key name (e.g., "fsid")
- * @param value value (e.g., a uuid rendered as a string)
- * @returns 0 for success, or an error code
- */
- int write_meta(const std::string& key,
- const std::string& value) const {
- string v = value;
- v += "\n";
- int r = safe_write_file(path.c_str(), key.c_str(),
- v.c_str(), v.length());
- if (r < 0)
- return r;
- return 0;
- }
-
- /**
- * read_meta - read a simple configuration key out-of-band
- *
- * Read a simple key value to an unopened/mounted store.
- *
- * Trailing whitespace is stripped off.
- *
- * @param key key name
- * @param value pointer to value string
- * @returns 0 for success, or an error code
- */
- int read_meta(const std::string& key,
- std::string *value) const {
- char buf[4096];
- int r = safe_read_file(path.c_str(), key.c_str(),
- buf, sizeof(buf));
- if (r <= 0)
- return r;
- // drop trailing newlines
- while (r && isspace(buf[r-1])) {
- --r;
- }
- *value = string(buf, r);
- return 0;
- }
-
- explicit MonitorDBStore(const string& path)
- : path(path),
- db(0),
- do_dump(false),
- dump_fd_binary(-1),
- dump_fmt(true),
- io_work(g_ceph_context, "monstore", "fn_monstore"),
- is_open(false) {
- }
- ~MonitorDBStore() {
- assert(!is_open);
- if (do_dump) {
- if (!g_conf->mon_debug_dump_json) {
- ::close(dump_fd_binary);
- } else {
- dump_fmt.close_section();
- dump_fmt.flush(dump_fd_json);
- dump_fd_json.flush();
- dump_fd_json.close();
- }
- }
- }
-
-};
-
-WRITE_CLASS_ENCODER(MonitorDBStore::Op)
-WRITE_CLASS_ENCODER(MonitorDBStore::Transaction)
-
-#endif /* CEPH_MONITOR_DB_STORE_H */
Code Review / stor4nfv.git / blobdiff