Merge pull request ceph#56499 from adamemerson/wip-rgw-close-datalog-write-hole

rgw: Close datalog write hole

Reviewed-by: Casey Bodley <[email protected]>

Author: cbodley

qa/suites/rgw/verify/tasks/cls.yaml

@@ -16,7 +16,9 @@ tasks:
         - cls/test_cls_cmpomap.sh
         - cls/test_cls_2pc_queue.sh
         - cls/test_cls_user.sh
+        - cls/test_cls_sem_set.sh
         - rgw/test_rgw_gc_log.sh
         - rgw/test_rgw_obj.sh
+        - rgw/test_rgw_datalog.sh
         - rgw/test_librgw_file.sh
         - rgw/test_awssdkv4_sig.sh

qa/valgrind.supp

@@ -1123,3 +1123,19 @@
    obj:/usr/lib64/libicuuc.so.67.1
    ...
 }
+{
+   rgw unclean shutdown one-time leak
+   Memcheck:Leak
+   match-leak-kinds: definite
+   fun:_Znam
+   fun:_ZN8Objecter15linger_registerERK8object_tRK16object_locator_ti
+   fun:_ZN8neorados5RADOS6watch_ENS_6ObjectENS_9IOContextESt8optionalINSt6chrono8durationIlSt5ratioILl1ELl1EEEEEN3fu27abi_3106detail8functionINSC_6configILb1ELb0ELm16EEENSC_8propertyILb1ELb0EJFvN5boost6system10error_codeEmmmON4ceph6buffer7v15_2_04listEEEEEEENSH_4asio22any_completion_handlerIJFvSJ_mEEEE
+   fun:_ZZZN5boost4asio6detail20awaitable_frame_baseINS0_15any_io_executorEE15await_transformIZNS0_12async_resultINS0_15use_awaitable_tIS3_EEJFvNS_6system10error_codeEmEEE8initiateINS6_INS0_9consign_tIS8_JNS0_19executor_work_guardINS0_10io_context19basic_executor_typeISaIvELm0EEEvvEEEEEJSB_EE12init_wrapperIZN8neorados5RADOS5watchIRKS8_EEDaNSO_6ObjectENSO_9IOContextESt8optionalINSt6chrono8durationIlSt5ratioILl1ELl1EEEEEN3fu27abi_3106detail8functionINS14_6configILb1ELb0ELm16EEENS14_8propertyILb1ELb0EJFvSA_mmmON4ceph6buffer7v15_2_04listEEEEEEEOT_EUlS1I_ST_SU_S1G_E_EEJSt5tupleIJSK_EEST_SU_S1G_EEENS0_9awaitableImS3_EES1H_S8_DpT0_EUlPS1H_E_EEDaS1H_PNSt9enable_ifIXsrSt14is_convertibleINS0_9result_ofIFS1H_PS4_EE4typeEPNS1_16awaitable_threadIS3_EEE5valueEvE4typeEEN6result13await_suspendENSt7__n486116coroutine_handleIvEEENUlPvE_4_FUNES2B_
+   ...
+}
+{
+   co_compose bug manifesting under Ubuntu only
+   Memcheck:Free
+   fun:_ZdaPvm
+   ...
+}

qa/workunits/cls/test_cls_sem_set.sh

@@ -0,0 +1,5 @@
+#!/bin/sh -e
+
+ceph_test_cls_sem_set
+
+exit 0

qa/workunits/rgw/test_rgw_datalog.sh

@@ -0,0 +1,5 @@
+#!/bin/sh -e
+
+ceph_test_datalog
+
+exit 0

src/cls/CMakeLists.txt

@@ -358,4 +358,12 @@ set_target_properties(cls_fifo PROPERTIES
 target_link_libraries(cls_fifo ${FMT_LIB})
 install(TARGETS cls_fifo DESTINATION ${cls_dir})
 
-
+# cls_sem_set
+set(cls_sem_set_srcs sem_set/module.cc)
+add_library(cls_sem_set SHARED ${cls_sem_set_srcs})
+set_target_properties(cls_sem_set PROPERTIES
+  VERSION "1.0.0"
+  SOVERSION "1"
+  INSTALL_RPATH ""
+  CXX_VISIBILITY_PRESET hidden)
+install(TARGETS cls_sem_set DESTINATION ${cls_dir})

src/cls/log/cls_log.cc

@@ -2,7 +2,9 @@
 // vim: ts=8 sw=2 smarttab
 
 #include "include/types.h"
-#include "include/utime.h"
+
+#include "common/ceph_time.h"
+
 #include "objclass/objclass.h"
 
 #include "cls_log_types.h"
@@ -15,14 +17,15 @@ using std::map;
 using std::string;
 
 using ceph::bufferlist;
+using namespace std::literals;
 
 CLS_VER(1,0)
 CLS_NAME(log)
 
 static string log_index_prefix = "1_";
 
 
-static int write_log_entry(cls_method_context_t hctx, string& index, cls_log_entry& entry)
+static int write_log_entry(cls_method_context_t hctx, string& index, cls::log::entry& entry)
 {
   bufferlist bl;
   encode(entry, bl);
@@ -34,15 +37,16 @@ static int write_log_entry(cls_method_context_t hctx, string& index, cls_log_ent
   return 0;
 }
 
-static void get_index_time_prefix(utime_t& ts, string& index)
+static void get_index_time_prefix(ceph::real_time ts, string& index)
 {
+  auto tv = ceph::real_clock::to_timeval(ts);
   char buf[32];
-  snprintf(buf, sizeof(buf), "%010ld.%06ld_", (long)ts.sec(), (long)ts.usec());
+  snprintf(buf, sizeof(buf), "%010ld.%06ld_", (long)tv.tv_sec, (long)tv.tv_usec);
 
   index = log_index_prefix + buf;
 }
 
-static int read_header(cls_method_context_t hctx, cls_log_header& header)
+static int read_header(cls_method_context_t hctx, cls::log::header& header)
 {
   bufferlist header_bl;
 
@@ -51,7 +55,7 @@ static int read_header(cls_method_context_t hctx, cls_log_header& header)
     return ret;
 
   if (header_bl.length() == 0) {
-    header = cls_log_header();
+    header = cls::log::header();
     return 0;
   }
 
@@ -65,7 +69,7 @@ static int read_header(cls_method_context_t hctx, cls_log_header& header)
   return 0;
 }
 
-static int write_header(cls_method_context_t hctx, cls_log_header& header)
+static int write_header(cls_method_context_t hctx, cls::log::header& header)
 {
   bufferlist header_bl;
   encode(header, header_bl);
@@ -77,7 +81,7 @@ static int write_header(cls_method_context_t hctx, cls_log_header& header)
   return 0;
 }
 
-static void get_index(cls_method_context_t hctx, utime_t& ts, string& index)
+static void get_index(cls_method_context_t hctx, ceph::real_time ts, string& index)
 {
   get_index_time_prefix(ts, index);
 
@@ -92,26 +96,26 @@ static int cls_log_add(cls_method_context_t hctx, bufferlist *in, bufferlist *ou
 {
   auto in_iter = in->cbegin();
 
-  cls_log_add_op op;
+  cls::log::ops::add_op op;
   try {
     decode(op, in_iter);
   } catch (ceph::buffer::error& err) {
-    CLS_LOG(1, "ERROR: cls_log_add_op(): failed to decode op");
+    CLS_LOG(1, "ERROR: cls::log::ops::add_op(): failed to decode op");
     return -EINVAL;
   }
 
-  cls_log_header header;
+  cls::log::header header;
 
   int ret = read_header(hctx, header);
   if (ret < 0)
     return ret;
 
   for (auto iter = op.entries.begin(); iter != op.entries.end(); ++iter) {
-    cls_log_entry& entry = *iter;
+    cls::log::entry& entry = *iter;
 
     string index;
 
-    utime_t timestamp = entry.timestamp;
+    auto timestamp = entry.timestamp;
     if (op.monotonic_inc && timestamp < header.max_time)
       timestamp = header.max_time;
     else if (timestamp > header.max_time)
@@ -146,11 +150,11 @@ static int cls_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist *o
 {
   auto in_iter = in->cbegin();
 
-  cls_log_list_op op;
+  cls::log::ops::list_op op;
   try {
     decode(op, in_iter);
   } catch (ceph::buffer::error& err) {
-    CLS_LOG(1, "ERROR: cls_log_list_op(): failed to decode op");
+    CLS_LOG(1, "ERROR: cls::log::ops::list_op(): failed to decode op");
     return -EINVAL;
   }
 
@@ -164,17 +168,17 @@ static int cls_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist *o
   } else {
     from_index = op.marker;
   }
-  bool use_time_boundary = (!op.from_time.is_zero() && (op.to_time >= op.from_time));
+  bool use_time_boundary = (!ceph::real_clock::is_zero(op.from_time) && (op.to_time >= op.from_time));
 
   if (use_time_boundary)
     get_index_time_prefix(op.to_time, to_index);
 
-#define MAX_ENTRIES 1000
+  static constexpr auto MAX_ENTRIES = 1000u;
   size_t max_entries = op.max_entries;
   if (!max_entries || max_entries > MAX_ENTRIES)
     max_entries = MAX_ENTRIES;
 
-  cls_log_list_ret ret;
+  cls::log::ops::list_ret ret;
 
   int rc = cls_cxx_map_get_vals(hctx, from_index, log_index_prefix, max_entries, &keys, &ret.truncated);
   if (rc < 0)
@@ -196,7 +200,7 @@ static int cls_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist *o
     bufferlist& bl = iter->second;
     auto biter = bl.cbegin();
     try {
-      cls_log_entry e;
+      cls::log::entry e;
       decode(e, biter);
       entries.push_back(e);
     } catch (ceph::buffer::error& err) {
@@ -216,7 +220,7 @@ static int cls_log_trim(cls_method_context_t hctx, bufferlist *in, bufferlist *o
 {
   auto in_iter = in->cbegin();
 
-  cls_log_trim_op op;
+  cls::log::ops::trim_op op;
   try {
     decode(op, in_iter);
   } catch (ceph::buffer::error& err) {
@@ -236,8 +240,7 @@ static int cls_log_trim(cls_method_context_t hctx, bufferlist *in, bufferlist *o
   // cls_cxx_map_remove_range() expects one-past-end
   if (op.to_marker.empty()) {
     auto t = op.to_time;
-    t.nsec_ref() += 1000; // equivalent to usec() += 1
-    t.normalize();
+    t += 1000us; // equivalent to usec() += 1
     get_index_time_prefix(t, to_index);
   } else {
     to_index = op.to_marker;
@@ -281,15 +284,15 @@ static int cls_log_info(cls_method_context_t hctx, bufferlist *in, bufferlist *o
 {
   auto in_iter = in->cbegin();
 
-  cls_log_info_op op;
+  cls::log::ops::info_op op;
   try {
     decode(op, in_iter);
   } catch (ceph::buffer::error& err) {
-    CLS_LOG(1, "ERROR: cls_log_add_op(): failed to decode op");
+    CLS_LOG(1, "ERROR: cls::log::ops::add_op(): failed to decode op");
     return -EINVAL;
   }
 
-  cls_log_info_ret ret;
+  cls::log::ops::info_ret ret;
 
   int rc = read_header(hctx, ret.header);
   if (rc < 0)

src/cls/log/cls_log_client.cc

@@ -7,7 +7,7 @@
 #include "include/compat.h"
 
 
-using std::list;
+using std::vector;
 using std::string;
 
 using ceph::bufferlist;
@@ -16,47 +16,47 @@ using namespace librados;
 
 
 
-void cls_log_add(librados::ObjectWriteOperation& op, list<cls_log_entry>& entries, bool monotonic_inc)
+void cls_log_add(librados::ObjectWriteOperation& op, vector<cls::log::entry>& entries, bool monotonic_inc)
 {
   bufferlist in;
-  cls_log_add_op call;
+  cls::log::ops::add_op call;
   call.entries = entries;
   encode(call, in);
   op.exec("log", "add", in);
 }
 
-void cls_log_add(librados::ObjectWriteOperation& op, cls_log_entry& entry)
+void cls_log_add(librados::ObjectWriteOperation& op, cls::log::entry& entry)
 {
   bufferlist in;
-  cls_log_add_op call;
+  cls::log::ops::add_op call;
   call.entries.push_back(entry);
   encode(call, in);
   op.exec("log", "add", in);
 }
 
-void cls_log_add_prepare_entry(cls_log_entry& entry, const utime_t& timestamp,
-                 const string& section, const string& name, bufferlist& bl)
+void cls_log_add_prepare_entry(cls::log::entry& entry, ceph::real_time timestamp,
+			       const string& section, const string& name, bufferlist& bl)
 {
   entry.timestamp = timestamp;
   entry.section = section;
   entry.name = name;
   entry.data = bl;
 }
 
-void cls_log_add(librados::ObjectWriteOperation& op, const utime_t& timestamp,
+void cls_log_add(librados::ObjectWriteOperation& op, ceph::real_time timestamp,
                  const string& section, const string& name, bufferlist& bl)
 {
-  cls_log_entry entry;
+  cls::log::entry entry;
 
   cls_log_add_prepare_entry(entry, timestamp, section, name, bl);
   cls_log_add(op, entry);
 }
 
-void cls_log_trim(librados::ObjectWriteOperation& op, const utime_t& from_time, const utime_t& to_time,
-                  const string& from_marker, const string& to_marker)
+void cls_log_trim(librados::ObjectWriteOperation& op, ceph::real_time from_time,
+		  ceph::real_time to_time, const string& from_marker, const string& to_marker)
 {
   bufferlist in;
-  cls_log_trim_op call;
+  cls::log::ops::trim_op call;
   call.from_time = from_time;
   call.to_time = to_time;
   call.from_marker = from_marker;
@@ -65,7 +65,8 @@ void cls_log_trim(librados::ObjectWriteOperation& op, const utime_t& from_time,
   op.exec("log", "trim", in);
 }
 
-int cls_log_trim(librados::IoCtx& io_ctx, const string& oid, const utime_t& from_time, const utime_t& to_time,
+int cls_log_trim(librados::IoCtx& io_ctx, const string& oid,
+		 ceph::real_time from_time, ceph::real_time to_time,
                  const string& from_marker, const string& to_marker)
 {
   bool done = false;
@@ -88,15 +89,15 @@ int cls_log_trim(librados::IoCtx& io_ctx, const string& oid, const utime_t& from
 }
 
 class LogListCtx : public ObjectOperationCompletion {
-  list<cls_log_entry> *entries;
+  vector<cls::log::entry>* entries;
   string *marker;
   bool *truncated;
 public:
-  LogListCtx(list<cls_log_entry> *_entries, string *_marker, bool *_truncated) :
-                                      entries(_entries), marker(_marker), truncated(_truncated) {}
+  LogListCtx(vector<cls::log::entry> *_entries, string *_marker, bool *_truncated) :
+    entries(_entries), marker(_marker), truncated(_truncated) {}
   void handle_completion(int r, bufferlist& outbl) override {
     if (r >= 0) {
-      cls_log_list_ret ret;
+      cls::log::ops::list_ret ret;
       try {
         auto iter = outbl.cbegin();
         decode(ret, iter);
@@ -113,13 +114,13 @@ class LogListCtx : public ObjectOperationCompletion {
   }
 };
 
-void cls_log_list(librados::ObjectReadOperation& op, const utime_t& from,
-		  const utime_t& to, const string& in_marker, int max_entries,
-		  list<cls_log_entry>& entries,
+void cls_log_list(librados::ObjectReadOperation& op, ceph::real_time from,
+		  ceph::real_time to, const string& in_marker, int max_entries,
+		  vector<cls::log::entry>& entries,
                   string *out_marker, bool *truncated)
 {
   bufferlist inbl;
-  cls_log_list_op call;
+  cls::log::ops::list_op call;
   call.from_time = from;
   call.to_time = to;
   call.marker = in_marker;
@@ -131,12 +132,12 @@ void cls_log_list(librados::ObjectReadOperation& op, const utime_t& from,
 }
 
 class LogInfoCtx : public ObjectOperationCompletion {
-  cls_log_header *header;
+  cls::log::header* header;
 public:
-  explicit LogInfoCtx(cls_log_header *_header) : header(_header) {}
+  explicit LogInfoCtx(cls::log::header *_header) : header(_header) {}
   void handle_completion(int r, bufferlist& outbl) override {
     if (r >= 0) {
-      cls_log_info_ret ret;
+      cls::log::ops::info_ret ret;
       try {
         auto iter = outbl.cbegin();
         decode(ret, iter);
@@ -149,10 +150,10 @@ class LogInfoCtx : public ObjectOperationCompletion {
   }
 };
 
-void cls_log_info(librados::ObjectReadOperation& op, cls_log_header *header)
+void cls_log_info(librados::ObjectReadOperation& op, cls::log::header *header)
 {
   bufferlist inbl;
-  cls_log_info_op call;
+  cls::log::ops::info_op call;
 
   encode(call, inbl);