Merge branch 'wip-cls-version' into wip-rgw-zone-2

9 files changed, 857 insertions, 1 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index 448fd116759..9e4d7053951 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -382,7 +382,7 @@ noinst_LIBRARIES += librgw.a
 
 my_radosgw_ldadd = \
        libglobal.la librgw.a librados.la libcls_rgw_client.a \
-       libcls_lock_client.a libcls_refcount_client.a -lcurl -lexpat \
+       libcls_lock_client.a libcls_refcount_client.a libcls_version_client.a -lcurl -lexpat \
        $(PTHREAD_LIBS) -lm $(CRYPTO_LIBS) $(EXTRALIBS)
 
 radosgw_SOURCES = \
@@ -524,6 +524,15 @@ libcls_refcount_la_LDFLAGS = ${AM_LDFLAGS} -version-info 1:0:0 -export-symbols-r
 
 radoslib_LTLIBRARIES += libcls_refcount.la
 
+# version class
+libcls_version_la_SOURCES = cls/version/cls_version.cc
+libcls_version_la_CFLAGS = ${AM_CFLAGS}
+libcls_version_la_CXXFLAGS= ${AM_CXXFLAGS}
+libcls_version_la_LIBADD = $(PTHREAD_LIBS) $(EXTRALIBS)
+libcls_version_la_LDFLAGS = ${AM_LDFLAGS} -version-info 1:0:0 -export-symbols-regex '.*__cls_.*'
+
+radoslib_LTLIBRARIES += libcls_version.la
+
 
 if WITH_RADOSGW
 # rgw: rados gateway
@@ -546,6 +555,10 @@ libcls_refcount_client_a_SOURCES =  \
 	cls/refcount/cls_refcount_client.cc
 noinst_LIBRARIES += libcls_refcount_client.a
 
+libcls_version_client_a_SOURCES =  \
+	cls/version/cls_version_client.cc
+noinst_LIBRARIES += libcls_version_client.a
+
 libcls_rgw_client_a_SOURCES =  \
 	cls/rgw/cls_rgw_client.cc \
 	cls/rgw/cls_rgw_types.cc \
@@ -884,6 +897,12 @@ ceph_test_cls_refcount_LDADD = librados.la libcls_refcount_client.a ${UNITTEST_S
 ceph_test_cls_refcount_CXXFLAGS = ${AM_CXXFLAGS} ${UNITTEST_CXXFLAGS}
 bin_DEBUGPROGRAMS += ceph_test_cls_refcount
 
+ceph_test_cls_version_SOURCES = test/cls_version/test_cls_version.cc \
+	test/librados/test.cc
+ceph_test_cls_version_LDADD = librados.la libcls_version_client.a ${UNITTEST_STATIC_LDADD}
+ceph_test_cls_version_CXXFLAGS = ${AM_CXXFLAGS} ${UNITTEST_CXXFLAGS}
+bin_DEBUGPROGRAMS += ceph_test_cls_version
+
 ceph_test_cls_lock_SOURCES = test/cls_lock/test_cls_lock.cc test/librados/test.cc
 ceph_test_cls_lock_LDFLAGS = ${AM_LDFLAGS}
 ceph_test_cls_lock_LDADD =  libcls_lock_client.a librados.la ${UNITTEST_STATIC_LDADD}
@@ -1511,6 +1530,9 @@ noinst_HEADERS = \
 	cls/rbd/cls_rbd_client.h\
 	cls/refcount/cls_refcount_ops.h\
 	cls/refcount/cls_refcount_client.h\
+	cls/version/cls_version_types.h\
+	cls/version/cls_version_ops.h\
+	cls/version/cls_version_client.h\
 	cls/rgw/cls_rgw_client.h\
 	cls/rgw/cls_rgw_ops.h\
 	cls/rgw/cls_rgw_types.h\
diff --git a/src/cls/version/cls_version.cc b/src/cls/version/cls_version.cc
new file mode 100644
index 00000000000..baecc248a22
--- /dev/null
+++ b/src/cls/version/cls_version.cc
@@ -0,0 +1,234 @@
+// -*- mode:C; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <iostream>
+
+#include <string.h>
+#include <stdlib.h>
+#include <errno.h>
+
+#include "include/types.h"
+#include "include/utime.h"
+#include "objclass/objclass.h"
+#include "cls/version/cls_version_types.h"
+#include "cls/version/cls_version_ops.h"
+#include "common/Clock.h"
+
+#include "global/global_context.h"
+
+CLS_VER(1,0)
+CLS_NAME(version)
+
+cls_handle_t h_class;
+cls_method_handle_t h_version_set;
+cls_method_handle_t h_version_inc;
+cls_method_handle_t h_version_inc_conds;
+cls_method_handle_t h_version_read;
+cls_method_handle_t h_version_check_conds;
+
+
+#define VERSION_ATTR "version"
+
+static int set_version(cls_method_context_t hctx, struct obj_version *objv)
+{
+  bufferlist bl;
+
+  ::encode(*objv, bl);
+
+  int ret = cls_cxx_setxattr(hctx, VERSION_ATTR, &bl);
+  if (ret < 0)
+    return ret;
+
+  return 0;
+}
+
+static int init_version(cls_method_context_t hctx, struct obj_version *objv)
+{
+#define TAG_LEN 24
+  char buf[TAG_LEN + 1];
+
+  int ret = cls_gen_rand_base64(buf, sizeof(buf));
+  if (ret < 0)
+    return ret;
+
+  objv->ver = 1;
+  objv->tag = buf;
+
+  return set_version(hctx, objv);
+}
+
+/* implicit create should be true only if called from a write operation (set, inc), never from a read operation (read, check) */
+static int read_version(cls_method_context_t hctx, obj_version *objv, bool implicit_create)
+{
+  bufferlist bl;
+  int ret = cls_cxx_getxattr(hctx, VERSION_ATTR, &bl);
+  if (ret == -ENOENT || ret == -ENODATA) {
+    objv->ver = 0;
+
+    if (implicit_create) {
+      return init_version(hctx, objv);
+    }
+    return 0;
+  }
+  if (ret < 0)
+    return ret;
+
+  try {
+    bufferlist::iterator iter = bl.begin();
+    ::decode(*objv, iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(0, "ERROR: read_version(): failed to decode version entry\n");
+    return -EIO;
+  }
+
+  return 0;
+}
+
+static int cls_version_set(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  bufferlist::iterator in_iter = in->begin();
+
+  cls_version_set_op op;
+  try {
+    ::decode(op, in_iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(1, "ERROR: cls_version_get(): failed to decode entry\n");
+    return -EINVAL;
+  }
+
+  int ret = set_version(hctx, &op.objv);
+  if (ret < 0)
+    return ret;
+
+  return 0;
+}
+
+static bool check_conds(list<obj_version_cond>& conds, obj_version& objv)
+{
+  if (conds.empty())
+    return true;
+
+  for (list<obj_version_cond>::iterator iter = conds.begin(); iter != conds.end(); ++iter) {
+    obj_version_cond& cond = *iter;
+    obj_version& v = cond.ver;
+
+    switch (cond.cond) {
+      case VER_COND_NONE:
+	break;
+      case VER_COND_EQ:
+	if (!objv.compare(&v))
+	  return false;
+	break;
+      case VER_COND_GT:
+	if (!(objv.ver > v.ver))
+	  return false;
+	break;
+      case VER_COND_GE:
+	if (!(objv.ver >= v.ver))
+	  return false;
+	break;
+      case VER_COND_LT:
+	if (!(objv.ver < v.ver))
+	  return false;
+	break;
+      case VER_COND_LE:
+	if (!(objv.ver <= v.ver))
+	  return false;
+	break;
+      case VER_COND_TAG_EQ:
+	if (objv.tag.compare(v.tag) != 0)
+	  return false;
+	break;
+      case VER_COND_TAG_NE:
+	if (objv.tag.compare(v.tag) == 0)
+	  return false;
+	break;
+    }
+  }
+
+  return true;
+}
+
+static int cls_version_inc(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  bufferlist::iterator in_iter = in->begin();
+
+  cls_version_inc_op op;
+  try {
+    ::decode(op, in_iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(1, "ERROR: cls_version_get(): failed to decode entry\n");
+    return -EINVAL;
+  }
+
+  obj_version objv;
+  int ret = read_version(hctx, &objv, true);
+  if (ret < 0)
+    return ret;
+  
+  if (!check_conds(op.conds, objv)) {
+    return -ECANCELED;
+  }
+  objv.inc();
+
+  ret = set_version(hctx, &objv);
+  if (ret < 0)
+    return ret;
+
+  return 0;
+}
+
+static int cls_version_check(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  bufferlist::iterator in_iter = in->begin();
+
+  cls_version_check_op op;
+  try {
+    ::decode(op, in_iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(1, "ERROR: cls_version_get(): failed to decode entry\n");
+    return -EINVAL;
+  }
+
+  obj_version objv;
+  int ret = read_version(hctx, &objv, false);
+  if (ret < 0)
+    return ret;
+  
+  if (!check_conds(op.conds, objv)) {
+    return -ECANCELED;
+  }
+
+  return 0;
+}
+
+static int cls_version_read(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  obj_version objv;
+
+  cls_version_read_ret read_ret;
+  int ret = read_version(hctx, &read_ret.objv, false);
+  if (ret < 0)
+    return ret;
+
+  ::encode(read_ret, *out);
+
+  return 0;
+}
+
+void __cls_init()
+{
+  CLS_LOG(1, "Loaded version class!");
+
+  cls_register("version", &h_class);
+
+  /* version */
+  cls_register_cxx_method(h_class, "set", CLS_METHOD_RD | CLS_METHOD_WR, cls_version_set, &h_version_set);
+  cls_register_cxx_method(h_class, "inc", CLS_METHOD_RD | CLS_METHOD_WR, cls_version_inc, &h_version_inc);
+  cls_register_cxx_method(h_class, "inc_conds", CLS_METHOD_RD | CLS_METHOD_WR, cls_version_inc, &h_version_inc_conds);
+  cls_register_cxx_method(h_class, "read", CLS_METHOD_RD, cls_version_read, &h_version_read);
+  cls_register_cxx_method(h_class, "check_conds", CLS_METHOD_RD, cls_version_check, &h_version_check_conds);
+
+  return;
+}
+
diff --git a/src/cls/version/cls_version_client.cc b/src/cls/version/cls_version_client.cc
new file mode 100644
index 00000000000..b0b81653f1f
--- /dev/null
+++ b/src/cls/version/cls_version_client.cc
@@ -0,0 +1,77 @@
+#include <errno.h>
+
+#include "include/types.h"
+#include "cls/version/cls_version_ops.h"
+#include "include/rados/librados.hpp"
+
+using namespace librados;
+
+
+void cls_version_set(librados::ObjectWriteOperation& op, obj_version& objv)
+{
+  bufferlist in;
+  cls_version_set_op call;
+  call.objv = objv;
+  ::encode(call, in);
+  op.exec("version", "set", in);
+}
+
+void cls_version_inc(librados::ObjectWriteOperation& op)
+{
+  bufferlist in;
+  cls_version_inc_op call;
+  ::encode(call, in);
+  op.exec("version", "inc", in);
+}
+
+void cls_version_inc(librados::ObjectWriteOperation& op, obj_version& objv, VersionCond cond)
+{
+  bufferlist in;
+  cls_version_inc_op call;
+  call.objv = objv;
+  
+  obj_version_cond c;
+  c.cond = cond;
+  c.ver = objv;
+
+  call.conds.push_back(c);
+
+  ::encode(call, in);
+  op.exec("version", "inc_conds", in);
+}
+
+void cls_version_check(librados::ObjectOperation& op, obj_version& objv, VersionCond cond)
+{
+  bufferlist in;
+  cls_version_inc_op call;
+  call.objv = objv;
+
+  obj_version_cond c;
+  c.cond = cond;
+  c.ver = objv;
+
+  call.conds.push_back(c);
+
+  ::encode(call, in);
+  op.exec("version", "check_conds", in);
+}
+
+int cls_version_read(librados::IoCtx& io_ctx, string& oid, obj_version *ver)
+{
+  bufferlist in, out;
+  int r = io_ctx.exec(oid, "version", "read", in, out);
+  if (r < 0)
+    return r;
+
+  cls_version_read_ret ret;
+  try {
+    bufferlist::iterator iter = out.begin();
+    ::decode(ret, iter);
+  } catch (buffer::error& err) {
+    return -EIO;
+  }
+
+  *ver = ret.objv;
+
+  return r;
+}
diff --git a/src/cls/version/cls_version_client.h b/src/cls/version/cls_version_client.h
new file mode 100644
index 00000000000..b11ea4fd37d
--- /dev/null
+++ b/src/cls/version/cls_version_client.h
@@ -0,0 +1,23 @@
+#ifndef CEPH_CLS_VERSION_CLIENT_H
+#define CEPH_CLS_VERSION_CLIENT_H
+
+#include "include/types.h"
+#include "include/rados/librados.hpp"
+
+/*
+ * version objclass
+ */
+
+void cls_version_set(librados::ObjectWriteOperation& op, obj_version& ver);
+
+/* increase anyway */
+void cls_version_inc(librados::ObjectWriteOperation& op);
+
+/* conditional increase, return -EAGAIN if condition fails */
+void cls_version_inc(librados::ObjectWriteOperation& op, obj_version& ver, VersionCond cond);
+
+int cls_version_read(librados::IoCtx& io_ctx, string& oid, obj_version *ver);
+
+void cls_version_check(librados::ObjectOperation& op, obj_version& ver, VersionCond cond);
+
+#endif
diff --git a/src/cls/version/cls_version_ops.h b/src/cls/version/cls_version_ops.h
new file mode 100644
index 00000000000..86036bc79d1
--- /dev/null
+++ b/src/cls/version/cls_version_ops.h
@@ -0,0 +1,95 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_CLS_VERSION_OPS_H
+#define CEPH_CLS_VERSION_OPS_H
+
+#include <map>
+
+#include "include/types.h"
+#include "cls_version_types.h"
+
+struct cls_version_set_op {
+  obj_version objv;
+
+  cls_version_set_op() {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(objv, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(objv, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_version_set_op)
+
+struct cls_version_inc_op {
+  obj_version objv;
+  list<obj_version_cond> conds;
+
+  cls_version_inc_op() {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(objv, bl);
+    ::encode(conds, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(objv, bl);
+    ::decode(conds, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_version_inc_op)
+
+struct cls_version_check_op {
+  obj_version objv;
+  list<obj_version_cond> conds;
+
+  cls_version_check_op() {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(objv, bl);
+    ::encode(conds, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(objv, bl);
+    ::decode(conds, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_version_check_op)
+
+struct cls_version_read_ret {
+  obj_version objv;
+
+  cls_version_read_ret() {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(objv, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(objv, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_version_read_ret)
+
+
+#endif
diff --git a/src/cls/version/cls_version_types.h b/src/cls/version/cls_version_types.h
new file mode 100644
index 00000000000..759ac25e298
--- /dev/null
+++ b/src/cls/version/cls_version_types.h
@@ -0,0 +1,81 @@
+#ifndef CEPH_CLS_VERSION_TYPES_H
+#define CEPH_CLS_VERSION_TYPES_H
+
+#include "include/encoding.h"
+#include "include/types.h"
+
+
+struct obj_version {
+  uint64_t ver;
+  string tag;
+
+  obj_version() : ver(0) {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(ver, bl);
+    ::encode(tag, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(ver, bl);
+    ::decode(tag, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void inc() {
+    ver++;
+  }
+
+  bool empty() {
+    return tag.empty();
+  }
+
+  bool compare(struct obj_version *v) {
+    return (ver == v->ver &&
+            tag.compare(v->tag) == 0);
+  }
+};
+WRITE_CLASS_ENCODER(obj_version)
+
+enum VersionCond {
+  VER_COND_NONE =      0,
+  VER_COND_EQ,  /* equal */
+  VER_COND_GT,  /* greater than */
+  VER_COND_GE,  /* greater or equal */
+  VER_COND_LT,  /* less than */
+  VER_COND_LE,  /* less or equal */
+  VER_COND_TAG_EQ,
+  VER_COND_TAG_NE,
+};
+
+struct obj_version_cond {
+  struct obj_version ver;
+  VersionCond cond;
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(ver, bl);
+    uint32_t c = (uint32_t)cond;
+    ::encode(c, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(ver, bl);
+    uint32_t c;
+    ::decode(c, bl);
+    cond = (VersionCond)c;
+    DECODE_FINISH(bl);
+  }
+
+};
+WRITE_CLASS_ENCODER(obj_version_cond)
+
+
+#endif
+
+
diff --git a/src/objclass/class_api.cc b/src/objclass/class_api.cc
index 79f5af70e98..d49890d2d4a 100644
--- a/src/objclass/class_api.cc
+++ b/src/objclass/class_api.cc
@@ -8,6 +8,9 @@
 
 #include "osd/ClassHandler.h"
 
+#include "auth/Crypto.h"
+#include "common/armor.h"
+
 static ClassHandler *ch;
 
 void cls_initialize(ClassHandler *h)
@@ -542,3 +545,33 @@ int cls_cxx_map_remove_key(cls_method_context_t hctx, const string &key)
   return (*pctx)->pg->do_osd_ops(*pctx, ops);
 }
 
+int cls_gen_random_bytes(char *buf, int size)
+{
+  return get_random_bytes(buf, size);
+}
+
+int cls_gen_rand_base64(char *dest, int size) /* size should be the required string size + 1 */
+{
+  char buf[size];
+  char tmp_dest[size + 4]; /* so that there's space for the extra '=' characters, and some */
+  int ret;
+
+  ret = cls_gen_random_bytes(buf, sizeof(buf));
+  if (ret < 0) {
+    generic_derr << "cannot get random bytes: " << ret << dendl;
+    return -1;
+  }
+
+  ret = ceph_armor(tmp_dest, &tmp_dest[sizeof(tmp_dest)],
+		   (const char *)buf, ((const char *)buf) + ((size - 1) * 3 + 4 - 1) / 4);
+  if (ret < 0) {
+    generic_derr << "ceph_armor failed" << dendl;
+    return -1;
+  }
+  tmp_dest[ret] = '\0';
+  memcpy(dest, tmp_dest, size);
+  dest[size] = '\0';
+
+  return 0;
+}
+
diff --git a/src/objclass/objclass.h b/src/objclass/objclass.h
index a9ecd2e9d98..1acbcfd9be3 100644
--- a/src/objclass/objclass.h
+++ b/src/objclass/objclass.h
@@ -133,6 +133,10 @@ extern int cls_cxx_map_write_header(cls_method_context_t hctx, bufferlist *inbl)
 extern int cls_cxx_map_remove_key(cls_method_context_t hctx, const string &key);
 extern int cls_cxx_map_update(cls_method_context_t hctx, bufferlist *inbl);
 
+/* utility functions */
+extern int cls_gen_random_bytes(char *buf, int size);
+extern int cls_gen_rand_base64(char *dest, int size); /* size should be the required string size + 1 */
+
 /* These are also defined in rados.h and librados.h. Keep them in sync! */
 #define CEPH_OSD_TMAP_HDR 'h'
 #define CEPH_OSD_TMAP_SET 's'
diff --git a/src/test/cls_version/test_cls_version.cc b/src/test/cls_version/test_cls_version.cc
new file mode 100644
index 00000000000..5012957f0ae
--- /dev/null
+++ b/src/test/cls_version/test_cls_version.cc
@@ -0,0 +1,287 @@
+// -*- mode:C; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "include/types.h"
+#include "cls/version/cls_version_types.h"
+#include "cls/version/cls_version_client.h"
+
+#include "gtest/gtest.h"
+#include "test/librados/test.h"
+
+#include <errno.h>
+#include <string>
+#include <vector>
+
+static librados::ObjectWriteOperation *new_op() {
+  return new librados::ObjectWriteOperation();
+}
+
+static librados::ObjectReadOperation *new_rop() {
+  return new librados::ObjectReadOperation();
+}
+
+TEST(cls_rgw, test_version_inc_read)
+{
+  librados::Rados rados;
+  librados::IoCtx ioctx;
+  string pool_name = get_temp_pool_name();
+
+  /* create pool */
+  ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+  ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+
+  /* add chains */
+  string oid = "obj";
+
+
+  /* create object */
+
+  ASSERT_EQ(0, ioctx.create(oid, true));
+
+  obj_version ver;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_EQ(0, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver.tag.size());
+  
+
+  /* inc version */
+  librados::ObjectWriteOperation *op = new_op();
+  cls_version_inc(*op);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_GT((long long)ver.ver, 0);
+  ASSERT_NE(0, (int)ver.tag.size());
+
+  /* inc version again! */
+  op = new_op();
+  cls_version_inc(*op);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  obj_version ver2;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver2));
+  ASSERT_GT((long long)ver2.ver, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver2.tag.compare(ver.tag));
+
+  delete op;
+}
+
+
+TEST(cls_rgw, test_version_set)
+{
+  librados::Rados rados;
+  librados::IoCtx ioctx;
+  string pool_name = get_temp_pool_name();
+
+  /* create pool */
+  ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+  ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+
+  /* add chains */
+  string oid = "obj";
+
+
+  /* create object */
+
+  ASSERT_EQ(0, ioctx.create(oid, true));
+
+  obj_version ver;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_EQ(0, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver.tag.size());
+
+
+  ver.ver = 123;
+  ver.tag = "foo";
+
+  /* set version */
+  librados::ObjectWriteOperation *op = new_op();
+  cls_version_set(*op, ver);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  /* read version */
+  obj_version ver2;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver2));
+  ASSERT_EQ((long long)ver2.ver, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver2.tag.compare(ver.tag));
+
+  delete op;
+}
+
+TEST(cls_rgw, test_version_inc_cond)
+{
+  librados::Rados rados;
+  librados::IoCtx ioctx;
+  string pool_name = get_temp_pool_name();
+
+  /* create pool */
+  ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+  ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+
+  /* add chains */
+  string oid = "obj";
+
+
+  /* create object */
+
+  ASSERT_EQ(0, ioctx.create(oid, true));
+
+  obj_version ver;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_EQ(0, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver.tag.size());
+  
+  /* inc version */
+  librados::ObjectWriteOperation *op = new_op();
+  cls_version_inc(*op);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_GT((long long)ver.ver, 0);
+  ASSERT_NE(0, (int)ver.tag.size());
+
+  obj_version cond_ver = ver;
+
+
+  /* inc version again! */
+  op = new_op();
+  cls_version_inc(*op);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  obj_version ver2;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver2));
+  ASSERT_GT((long long)ver2.ver, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver2.tag.compare(ver.tag));
+
+
+  /* now check various condition tests */
+  cls_version_inc(*op, cond_ver, VER_COND_NONE);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver2));
+  ASSERT_GT((long long)ver2.ver, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver2.tag.compare(ver.tag));
+
+  /* a bunch of conditions that should fail */
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_EQ);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, op));
+
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_LT);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, op));
+
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_LE);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, op));
+
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_TAG_NE);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, op));
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver2));
+  ASSERT_GT((long long)ver2.ver, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver2.tag.compare(ver.tag));
+
+  /* a bunch of conditions that should succeed */
+  op = new_op();
+  cls_version_inc(*op, ver2, VER_COND_EQ);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_GT);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_GE);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  op = new_op();
+  cls_version_inc(*op, cond_ver, VER_COND_TAG_EQ);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  delete op;
+}
+
+TEST(cls_rgw, test_version_inc_check)
+{
+  librados::Rados rados;
+  librados::IoCtx ioctx;
+  string pool_name = get_temp_pool_name();
+
+  /* create pool */
+  ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+  ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+
+  /* add chains */
+  string oid = "obj";
+
+
+  /* create object */
+
+  ASSERT_EQ(0, ioctx.create(oid, true));
+
+  obj_version ver;
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_EQ(0, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver.tag.size());
+  
+  /* inc version */
+  librados::ObjectWriteOperation *op = new_op();
+  cls_version_inc(*op);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver));
+  ASSERT_GT((long long)ver.ver, 0);
+  ASSERT_NE(0, (int)ver.tag.size());
+
+  obj_version cond_ver = ver;
+
+  /* a bunch of conditions that should succeed */
+  librados::ObjectReadOperation *rop = new_rop();
+  cls_version_check(*rop, cond_ver, VER_COND_EQ);
+  bufferlist bl;
+  ASSERT_EQ(0, ioctx.operate(oid, rop, &bl));
+
+  rop = new_rop();
+  cls_version_check(*rop, cond_ver, VER_COND_GE);
+  ASSERT_EQ(0, ioctx.operate(oid, rop, &bl));
+
+  rop = new_rop();
+  cls_version_check(*rop, cond_ver, VER_COND_LE);
+  ASSERT_EQ(0, ioctx.operate(oid, rop, &bl));
+
+  rop = new_rop();
+  cls_version_check(*rop, cond_ver, VER_COND_TAG_EQ);
+  ASSERT_EQ(0, ioctx.operate(oid, rop, &bl));
+
+  obj_version ver2;
+
+  op = new_op();
+  cls_version_inc(*op);
+  ASSERT_EQ(0, ioctx.operate(oid, op));
+
+  ASSERT_EQ(0, cls_version_read(ioctx, oid, &ver2));
+  ASSERT_GT((long long)ver2.ver, (long long)ver.ver);
+  ASSERT_EQ(0, (int)ver2.tag.compare(ver.tag));
+
+  /* a bunch of conditions that should fail */
+  rop = new_rop();
+  cls_version_check(*rop, ver, VER_COND_LT);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, rop, &bl));
+
+  rop = new_rop();
+  cls_version_check(*rop, cond_ver, VER_COND_LE);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, rop, &bl));
+
+  rop = new_rop();
+  cls_version_check(*rop, cond_ver, VER_COND_TAG_NE);
+  ASSERT_EQ(-ECANCELED, ioctx.operate(oid, rop, &bl));
+}