rados: implement rm --force option to force remove when full.

            librados extend remove interface, add flags parameter, and use
            this extended interface to implement force remove when cluster
            full.
Signed-off-by: Xiaowei Chen <chen.xiaowei@h3c.com>

qa/workunits/rados/test_rados_tool.sh

@@ -258,12 +258,22 @@ run_expect_nosignal "$RADOS_TOOL" --pool "$POOL" bench 1 seq
 set -e
 
 OBJ=test_rados_obj
+p=`uuidgen`
 
 expect_false()
 {
 	if "$@"; then return 1; else return 0; fi
 }
 
+delay_osd() {
+    MSGTYPE=$1
+    shift
+    $@ --rados-osd-op-timeout 5 --ms-inject-delay-type osd --ms-inject-delay-max 1 --ms-inject-delay-probability 1 --ms-inject-delay-msg-type $MSGTYPE
+    if [ $? -eq 0 ]; then
+        exit 2
+    fi
+}
+
 cleanup() {
     $RADOS_TOOL -p $POOL rm $OBJ || true
 }
@@ -304,9 +314,20 @@ test_xattr() {
     rm $V1 $V2
     cleanup
 }
+test_rmobj() {
+    ceph osd pool create $p 1
+    ceph osd pool set-quota $p max_objects 1
+    rados put $OBJ -p $p
+    sleep 30
+
+    delay_osd osd_op_reply rados -p $p rm $OBJ
+    rados -p $p rm $OBJ --force-full
+    rados rmpool $p $p --yes-i-really-really-mean-it
+}
 
 test_xattr
 test_omap
+test_rmobj
 
 echo "SUCCESS!"
 exit 0

src/include/rados/librados.hpp

@@ -677,6 +677,7 @@ namespace librados
                    size_t len);
     int read(const std::string& oid, bufferlist& bl, size_t len, uint64_t off);
     int remove(const std::string& oid);
+    int remove(const std::string& oid, int flags);
     int trunc(const std::string& oid, uint64_t size);
     int mapext(const std::string& o, uint64_t off, size_t len, std::map<uint64_t,uint64_t>& m);
     int sparse_read(const std::string& o, std::map<uint64_t,uint64_t>& m, bufferlist& bl, size_t len, uint64_t off);

src/librados/IoCtxImpl.cc

@@ -947,7 +947,13 @@ int librados::IoCtxImpl::remove(const object_t& oid)
   op.remove();
   return operate(oid, &op, NULL);
 }
-
+int librados::IoCtxImpl::remove(const object_t& oid, int flags)
+{
+  ::ObjectOperation op;
+  prepare_assert_ops(&op);
+  op.remove();
+  return operate(oid, &op, NULL, flags);
+}
 int librados::IoCtxImpl::trunc(const object_t& oid, uint64_t size)
 {
   ::ObjectOperation op;

src/librados/IoCtxImpl.h

@@ -127,6 +127,7 @@ struct librados::IoCtxImpl {
   int sparse_read(const object_t& oid, std::map<uint64_t,uint64_t>& m,
 		  bufferlist& bl, size_t len, uint64_t off);
   int remove(const object_t& oid);
+  int remove(const object_t& oid, int flags);
   int stat(const object_t& oid, uint64_t *psize, time_t *pmtime);
   int trunc(const object_t& oid, uint64_t size);
 

src/librados/librados.cc

@@ -1158,6 +1158,11 @@ int librados::IoCtx::remove(const std::string& oid)
   object_t obj(oid);
   return io_ctx_impl->remove(obj);
 }
+int librados::IoCtx::remove(const std::string& oid, int flags)
+{
+  object_t obj(oid);
+  return io_ctx_impl->remove(obj, flags); 
+}
 
 int librados::IoCtx::trunc(const std::string& oid, uint64_t size)
 {

src/tools/rados/rados.cc

@@ -84,7 +84,7 @@ void usage(ostream& out)
 "   put <obj-name> [infile]          write object\n"
 "   truncate <obj-name> length       truncate object\n"
 "   create <obj-name>                create object\n"
-"   rm <obj-name> ...                remove object(s)\n"
+"   rm <obj-name> ...[--force-full]  [force no matter full or not]remove object(s)\n"
 "   cp <obj-name> [target-obj]       copy object\n"
 "   clonedata <src-obj> <dst-obj>    clone object data\n"
 "   listxattr <obj-name>\n"
@@ -1239,7 +1239,7 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts,
 
   std::string run_name;
   std::string prefix;
-
+  std::string forcefull("");
   Formatter *formatter = NULL;
   bool pretty_format = false;
   const char *output = NULL;
@@ -1282,6 +1282,11 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts,
   if (i != opts.end()) {
     run_name = i->second;
   }
+
+  i = opts.find("force-full");
+  if (i != opts.end()) {
+    forcefull = i->second;
+  }
   i = opts.find("prefix");
   if (i != opts.end()) {
     prefix = i->second;
@@ -2146,7 +2151,11 @@ static int rados_tool_common(const std::map < std::string, std::string > &opts,
       if (use_striper) {
 	ret = striper.remove(oid);
       } else {
-	ret = io_ctx.remove(oid);
+	if ( forcefull == "true" ){
+	  ret = io_ctx.remove(oid, CEPH_OSD_FLAG_FULL_FORCE);
+	} else {
+	  ret = io_ctx.remove(oid);
+	}
       }
       if (ret < 0) {
         string name = (nspace.size() ? nspace + "/" : "" ) + oid;
@@ -2929,6 +2938,8 @@ int main(int argc, const char **argv)
       exit(0);
     } else if (ceph_argparse_flag(args, i, "-f", "--force", (char*)NULL)) {
       opts["force"] = "true";
+    } else if (ceph_argparse_flag(args, i, "--force-full", (char*)NULL)) {
+      opts["force-full"] = "true";
     } else if (ceph_argparse_flag(args, i, "-d", "--delete-after", (char*)NULL)) {
       opts["delete-after"] = "true";
     } else if (ceph_argparse_flag(args, i, "-C", "--create", "--create-pool",