rgw: when deleted obj removed in versioned bucket, extra del-marker added

After initial checks are complete, this will read the OLH earlier than
previously to check the delete-marker flag and under the bug's
conditions will return -ENOENT rather than create a spurious delete
marker.

Signed-off-by: J. Eric Ivancich <ivancich@redhat.com>
(cherry picked from commit 69d7589fb1305b7d202ffd126c3c835e7cd0dda3)

Conflicts:
	src/cls/rgw/cls_rgw_types.h
	src/rgw/rgw_rados.cc

Cherry-pick notes:
- RGWRados::apply_olh_log does not take DoutPrefixProvider in Octopus
- change to use some namespace-qualified names in cls_rgw_types

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 84792f7407f5c9ce3e45d194f3dbd1c7d5b90401..271528305bdb4f8dbc6a795e3dc6132ee450b4eb 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -1184,6 +1184,7 @@ class BIVerObjEntry {
 
 public:
   BIVerObjEntry(cls_method_context_t& _hctx, const cls_rgw_obj_key& _key) : hctx(_hctx), key(_key), initialized(false) {
+    // empty
   }
 
   int init(bool check_delete_marker = true) {
@@ -1521,11 +1522,20 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     return -EINVAL;
   }
 
-  BIVerObjEntry obj(hctx, op.key);
-  BIOLHEntry olh(hctx, op.key);
-
   /* read instance entry */
+  BIVerObjEntry obj(hctx, op.key);
   int ret = obj.init(op.delete_marker);
+
+  /* NOTE: When a delete is issued, a key instance is always provided,
+   * either the one for which the delete is requested or a new random
+   * one when no instance is specified. So we need to see which of
+   * these two cases we're dealing with. The variable `existed` will
+   * be true if the instance was specified and false if it was
+   * randomly generated. It might have been cleaner if the instance
+   * were empty and randomly generated here and returned in the reply,
+   * as that would better allow a typo in the instance id. This code
+   * should be audited and possibly cleaned up. */
+
   bool existed = (ret == 0);
   if (ret == -ENOENT && op.delete_marker) {
     ret = 0;
@@ -1534,6 +1544,28 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     return ret;
   }
 
+  BIOLHEntry olh(hctx, op.key);
+  bool olh_read_attempt = false;
+  bool olh_found = false;
+  if (!existed && op.delete_marker) {
+    /* read olh */
+    ret = olh.init(&olh_found);
+    if (ret < 0) {
+      return ret;
+    }
+    olh_read_attempt = true;
+
+    // if we're deleting (i.e., adding a delete marker, and the OLH
+    // indicates it already refers to a delete marker, error out)
+    if (olh_found && olh.get_entry().delete_marker) {
+      CLS_LOG(10,
+             "%s: delete marker received for \"%s\" although OLH"
+             " already refers to a delete marker\n",
+             __func__, escape_str(op.key.to_string()).c_str());
+      return -ENOENT;
+    }
+  }
+
   if (existed && !real_clock::is_zero(op.unmod_since)) {
     timespec mtime = ceph::real_clock::to_timespec(obj.mtime());
     timespec unmod = ceph::real_clock::to_timespec(op.unmod_since);
@@ -1586,11 +1618,14 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
   }
 
   /* read olh */
-  bool olh_found;
-  ret = olh.init(&olh_found);
-  if (ret < 0) {
-    return ret;
+  if (!olh_read_attempt) { // only read if we didn't attempt earlier
+    ret = olh.init(&olh_found);
+    if (ret < 0) {
+      return ret;
+    }
+    olh_read_attempt = true;
   }
+
   const uint64_t prev_epoch = olh.get_epoch();
 
   if (!olh.start_modify(op.olh_epoch)) {

diff --git a/src/cls/rgw/cls_rgw_types.h b/src/cls/rgw/cls_rgw_types.h
index 620811dbc4a25925a12561f531826749c2d4b0f7..3f5b41a4753f33a343fc7ceb28d5832d41ad9dbe 100644 (file)
--- a/src/cls/rgw/cls_rgw_types.h
+++ b/src/cls/rgw/cls_rgw_types.h
@@ -1,13 +1,16 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
 // vim: ts=8 sw=2 smarttab
 
-#ifndef CEPH_CLS_RGW_TYPES_H
-#define CEPH_CLS_RGW_TYPES_H
+#pragma once
 
 #include <boost/container/flat_map.hpp>
 #include "common/ceph_time.h"
 #include "common/Formatter.h"
 
+#undef FMT_HEADER_ONLY
+#define FMT_HEADER_ONLY 1
+#include <fmt/format.h>
+
 #include "rgw/rgw_basic_types.h"
 
 #define CEPH_RGW_REMOVE 'r'
@@ -343,7 +346,15 @@ struct cls_rgw_obj_key {
   cls_rgw_obj_key(const string &_name) : name(_name) {}
   cls_rgw_obj_key(const string& n, const string& i) : name(n), instance(i) {}
 
-  void set(const string& _name) {
+  std::string to_string() const {
+    return fmt::format("{}({})", name, instance);
+  }
+
+  bool empty() const {
+    return name.empty();
+  }
+
+  void set(const std::string& _name) {
     name = _name;
   }
 
@@ -351,6 +362,7 @@ struct cls_rgw_obj_key {
     return (name.compare(k.name) == 0) &&
            (instance.compare(k.instance) == 0);
   }
+
   bool operator<(const cls_rgw_obj_key& k) const {
     int r = name.compare(k.name);
     if (r == 0) {
@@ -358,13 +370,17 @@ struct cls_rgw_obj_key {
     }
     return (r < 0);
   }
+
   bool operator<=(const cls_rgw_obj_key& k) const {
     return !(k < *this);
   }
-  bool empty() const {
-    return name.empty();
+
+  std::ostream& operator<<(std::ostream& out) const {
+    out << to_string();
+    return out;
   }
-  void encode(bufferlist &bl) const {
+
+  void encode(ceph::buffer::list &bl) const {
     ENCODE_START(1, 1, bl);
     encode(name, bl);
     encode(instance, bl);
@@ -1288,5 +1304,3 @@ struct cls_rgw_reshard_entry
   void get_key(string *key) const;
 };
 WRITE_CLASS_ENCODER(cls_rgw_reshard_entry)
-
-#endif

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index aef7ea42f1f88209d10b4dab55c54e3d2ce12fcb..1e94e41662546e159df7ae9a9a019b905b37fbdb 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -7025,9 +7025,14 @@ static int decode_olh_info(CephContext* cct, const bufferlist& bl, RGWOLHInfo *o
   }
 }
 
-int RGWRados::apply_olh_log(RGWObjectCtx& obj_ctx, RGWObjState& state, const RGWBucketInfo& bucket_info, const rgw_obj& obj,
-                            bufferlist& olh_tag, map<uint64_t, vector<rgw_bucket_olh_log_entry> >& log,
-                            uint64_t *plast_ver, rgw_zone_set* zones_trace)
+int RGWRados::apply_olh_log(RGWObjectCtx& obj_ctx,
+                           RGWObjState& state,
+                           const RGWBucketInfo& bucket_info,
+                           const rgw_obj& obj,
+                           bufferlist& olh_tag,
+                           std::map<uint64_t, std::vector<rgw_bucket_olh_log_entry> >& log,
+                           uint64_t *plast_ver,
+                           rgw_zone_set* zones_trace)
 {
   if (log.empty()) {
     return 0;