mon/MonitorDBStore: allow compaction of ranges

Allow a transaction to describe the compaction of a range of keys.  Do this
in a backward compatible say, such that older code will interpret the
compaction of a prefix + range as compaction of the entire prefix.  This
allows us to avoid introducing any new feature bits.

Signed-off-by: Sage Weil <sage@inktank.com>
(cherry picked from commit ab09f1e5c1305a64482ebbb5a6156a0bb12a63a4)

Conflicts:

	src/mon/MonitorDBStore.h

1 files changed, 30 insertions, 15 deletions

diff --git a/src/mon/MonitorDBStore.h b/src/mon/MonitorDBStore.h
index f496fa0e587..6fa7dcac354 100644
--- a/src/mon/MonitorDBStore.h
+++ b/src/mon/MonitorDBStore.h
@@ -38,7 +38,7 @@ class MonitorDBStore
   struct Op {
     uint8_t type;
     string prefix;
-    string key;
+    string key, endkey;
     bufferlist bl;
 
     Op() { }
@@ -46,22 +46,27 @@ class MonitorDBStore
       : 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(1, 1, encode_bl);
+      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(1, 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);
     }
   };
@@ -72,7 +77,7 @@ class MonitorDBStore
     enum {
       OP_PUT	= 1,
       OP_ERASE	= 2,
-      OP_COMPACT_PREFIX = 3,
+      OP_COMPACT = 3,
     };
 
     void put(string prefix, string key, bufferlist& bl) {
@@ -102,7 +107,11 @@ class MonitorDBStore
     }
 
     void compact_prefix(string prefix) {
-      ops.push_back(Op(OP_COMPACT_PREFIX, prefix, string()));
+      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 {
@@ -164,10 +173,12 @@ class MonitorDBStore
 	    f->dump_string("key", op.key);
 	  }
 	  break;
-	case OP_COMPACT_PREFIX:
+	case OP_COMPACT:
 	  {
-	    f->dump_string("type", "COMPACT_PREFIX");
+	    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:
@@ -193,9 +204,9 @@ class MonitorDBStore
       bl.write_fd(dump_fd);
     }
 
-    list<string> compact_prefixes;
-    for (list<Op>::iterator it = t.ops.begin(); it != t.ops.end(); ++it) {
-      Op& op = *it;
+    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);
@@ -203,8 +214,8 @@ class MonitorDBStore
       case Transaction::OP_ERASE:
 	dbt->rmkey(op.prefix, op.key);
 	break;
-      case Transaction::OP_COMPACT_PREFIX:
-	compact_prefixes.push_back(op.prefix);
+      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;
@@ -214,9 +225,13 @@ class MonitorDBStore
     }
     int r = db->submit_transaction_sync(dbt);
     if (r >= 0) {
-      while (!compact_prefixes.empty()) {
-	db->compact_prefix_async(compact_prefixes.front());
-	compact_prefixes.pop_front();
+      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();
       }
     }
     return r;