rgw: pubsub: store event in bucket index

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>

diff --git a/src/rgw/rgw_cr_tools.cc b/src/rgw/rgw_cr_tools.cc
index 69eed3689b26432d2fad39574a10aafd4b01f5aa..a07e709a96b757700d142c261db3e5bb573c1a28 100644 (file)
--- a/src/rgw/rgw_cr_tools.cc
+++ b/src/rgw/rgw_cr_tools.cc
@@ -234,6 +234,10 @@ int RGWObjectSimplePutCR::Request::_send_request()
     return -ret;
   }
 
+  if (params.user_data) {
+    obj->set_user_data(*params.user_data);
+  }
+
   ret = obj->put(params.data, params.attrs);
   if (ret < 0) {
     cerr << "ERROR: put object returned error: " << cpp_strerror(-ret) << std::endl;

diff --git a/src/rgw/rgw_cr_tools.h b/src/rgw/rgw_cr_tools.h
index 635bbb9302fc6f02fcc37c64e13ce41d24c5e4a1..748fd94fd2c19fff76d800014b4af41466429c2c 100644 (file)
--- a/src/rgw/rgw_cr_tools.h
+++ b/src/rgw/rgw_cr_tools.h
@@ -56,6 +56,7 @@ struct rgw_object_simple_put_params {
   rgw_obj_key key;
   bufferlist data;
   map<string, bufferlist> attrs;
+  std::optional<string> user_data;
 };
 
 using RGWObjectSimplePutCR = RGWSimpleWriteOnlyAsyncCR<rgw_object_simple_put_params>;

diff --git a/src/rgw/rgw_sync_module_pubsub.cc b/src/rgw/rgw_sync_module_pubsub.cc
index 184a61be8f16460cc5aeac00df533caaa5f0234f..af087f4bdee3810ca10eb4af8a6a7e4d1db6132f 100644 (file)
--- a/src/rgw/rgw_sync_module_pubsub.cc
+++ b/src/rgw/rgw_sync_module_pubsub.cc
@@ -559,6 +559,12 @@ class PSSubscription {
         put_obj.key = rgw_obj_key(oid_prefix + pse.generate_message_id());
 
         pse.format(&put_obj.data);
+
+        {
+          bufferlist bl64;
+          put_obj.data.encode_base64(bl64);
+          put_obj.user_data = bl64.to_str();
+        }
         
         yield call(new RGWObjectSimplePutCR(sync_env->async_rados,
                                             sync_env->store,

diff --git a/src/rgw/rgw_tools.cc b/src/rgw/rgw_tools.cc
index 2883da6b16367d124abfe737a6685ea7de654d47..0888bab7672c35179ff7b4048d18b4144aebe38a 100644 (file)
--- a/src/rgw/rgw_tools.cc
+++ b/src/rgw/rgw_tools.cc
@@ -410,9 +410,16 @@ int RGWDataAccess::Object::put(bufferlist& data,
   }
   attrs[RGW_ATTR_ACL] = *aclbl;
 
+  string *puser_data = nullptr;
+  if (user_data) {
+    puser_data = &(*user_data);
+  }
+
   return processor.complete(obj_size, etag,
                            &mtime, mtime,
-                           attrs, delete_at);
+                           attrs, delete_at,
+                            nullptr, nullptr,
+                            puser_data);
 }
 
 void RGWDataAccess::Object::set_policy(const RGWAccessControlPolicy& policy)

diff --git a/src/rgw/rgw_tools.h b/src/rgw/rgw_tools.h
index 0741eb6fa6939e1a18aa17989f7b2856b44c4951..59afa8357bffaa67e76b5e60c616fa0c8a42b241 100644 (file)
--- a/src/rgw/rgw_tools.h
+++ b/src/rgw/rgw_tools.h
@@ -139,6 +139,7 @@ public:
     string etag;
     std::optional<uint64_t> olh_epoch;
     ceph::real_time delete_at;
+    std::optional<string> user_data;
 
     std::optional<bufferlist> aclbl;
 
@@ -166,6 +167,10 @@ public:
       delete_at = _delete_at;
     }
 
+    void set_user_data(const string& _user_data) {
+      user_data = _user_data;
+    }
+
     void set_policy(const RGWAccessControlPolicy& policy);
 
     friend class Bucket;