Merge pull request #41980 from yuvalif/wip-yuval-fix-51320

rgw/notification: support version-id for all event types

doc/radosgw/notifications.rst

@@ -437,7 +437,9 @@ pushed or pulled using the pubsub sync module. For example:
 - s3.object.key: object key
 - s3.object.size: object size
 - s3.object.eTag: object etag
-- s3.object.version: object version in case of versioned bucket
+- s3.object.versionId: object version in case of versioned bucket. 
+  When doing a copy, it would include the version of the target object. 
+  When creating a delete marker, it would include the version of the delete marker.
 - s3.object.sequencer: monotonically increasing identifier of the change per object (hexadecimal format)
 - s3.object.metadata: any metadata set on the object sent as: ``x-amz-meta-`` (an extension to the S3 notification API)
 - s3.object.tags: any tags set on the object (an extension to the S3 notification API)

src/rgw/rgw_notify.cc

@@ -672,6 +672,7 @@ void populate_event_from_request(const reservation_t& res,
         uint64_t size,
         const ceph::real_time& mtime, 
         const std::string& etag, 
+        const std::string& version, 
         EventType event_type,
         rgw_pubsub_s3_event& event) {
   const auto s = res.s;
@@ -687,7 +688,7 @@ void populate_event_from_request(const reservation_t& res,
   event.object_key = res.object_name ? *res.object_name : obj->get_name();
   event.object_size = size;
   event.object_etag = etag;
-  event.object_versionId = obj->get_instance();
+  event.object_versionId = version;
   // use timestamp as per key sequence id (hex encoded)
   const utime_t ts(real_clock::now());
   boost::algorithm::hex((const char*)&ts, (const char*)&ts + sizeof(utime_t), 
@@ -816,6 +817,7 @@ int publish_commit(rgw::sal::Object* obj,
         uint64_t size,
         const ceph::real_time& mtime, 
         const std::string& etag, 
+        const std::string& version,
         EventType event_type,
         reservation_t& res,
         const DoutPrefixProvider *dpp) 
@@ -826,7 +828,7 @@ int publish_commit(rgw::sal::Object* obj,
       continue;
     }
     event_entry_t event_entry;
-    populate_event_from_request(res, obj, size, mtime, etag, event_type, event_entry.event);
+    populate_event_from_request(res, obj, size, mtime, etag, version, event_type, event_entry.event);
     event_entry.event.configurationId = topic.configurationId;
     event_entry.event.opaque_data = topic.cfg.opaque_data;
     if (topic.cfg.dest.persistent) { 

src/rgw/rgw_notify.h

@@ -81,6 +81,7 @@ int publish_commit(rgw::sal::Object* obj,
         uint64_t size,
         const ceph::real_time& mtime, 
         const std::string& etag, 
+        const std::string& version,
         EventType event_type,
         reservation_t& reservation,
         const DoutPrefixProvider *dpp);

src/rgw/rgw_op.cc

@@ -4090,7 +4090,7 @@ void RGWPutObj::execute(optional_yield y)
   }
 
   // send request to notification manager
-  int ret = res->publish_commit(this, s->obj_size, mtime, etag);
+  int ret = res->publish_commit(this, s->obj_size, mtime, etag, s->object->get_instance());
   if (ret < 0) {
     ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
     // too late to rollback operation, hence op_ret is not set here
@@ -4357,7 +4357,7 @@ void RGWPostObj::execute(optional_yield y)
   } while (is_next_file_to_upload());
 
   // send request to notification manager
-  int ret = res->publish_commit(this, ofs, ceph::real_clock::now(), etag);
+  int ret = res->publish_commit(this, ofs, s->object->get_mtime(), etag, s->object->get_instance());
   if (ret < 0) {
     ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
     // too late to rollback operation, hence op_ret is not set here
@@ -4917,7 +4917,7 @@ void RGWDeleteObj::execute(optional_yield y)
     const auto obj_state = obj_ctx->get_state(s->object->get_obj());
 
     // send request to notification manager
-    int ret = res->publish_commit(this, obj_state->size, obj_state->mtime, attrs[RGW_ATTR_ETAG].to_str());
+    int ret = res->publish_commit(this, obj_state->size, obj_state->mtime, attrs[RGW_ATTR_ETAG].to_str(), version_id);
     if (ret < 0) {
       ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
       // too late to rollback operation, hence op_ret is not set here
@@ -5325,7 +5325,7 @@ void RGWCopyObj::execute(optional_yield y)
 	   s->yield);
 
   // send request to notification manager
-  int ret = res->publish_commit(this, astate->size, mtime, etag);
+  int ret = res->publish_commit(this, astate->size, mtime, etag, dest_object->get_instance());
   if (ret < 0) {
     ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
     // too late to rollback operation, hence op_ret is not set here
@@ -6015,7 +6015,7 @@ void RGWInitMultipart::execute(optional_yield y)
   } while (op_ret == -EEXIST);
   
   // send request to notification manager
-  int ret = res->publish_commit(this, s->obj_size, ceph::real_clock::now(), attrs[RGW_ATTR_ETAG].to_str());
+  int ret = res->publish_commit(this, s->obj_size, ceph::real_clock::now(), attrs[RGW_ATTR_ETAG].to_str(), "");
   if (ret < 0) {
     ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
     // too late to rollback operation, hence op_ret is not set here
@@ -6369,7 +6369,7 @@ void RGWCompleteMultipart::execute(optional_yield y)
   }
 
   // send request to notification manager
-  int ret = res->publish_commit(this, ofs, ceph::real_clock::now(), final_etag_str);
+  int ret = res->publish_commit(this, ofs, target_obj->get_mtime(), final_etag_str,  target_obj->get_instance());
   if (ret < 0) {
     ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
     // too late to rollback operation, hence op_ret is not set here
@@ -6918,7 +6918,7 @@ void RGWDeleteMultiObj::execute(optional_yield y)
     const auto etag = obj_state->get_attr(RGW_ATTR_ETAG, etag_bl) ? etag_bl.to_str() : "";
 
     // send request to notification manager
-    int ret = res->publish_commit(this, obj_state->size, obj_state->mtime, etag);
+    int ret = res->publish_commit(this, obj_state->size, obj_state->mtime, etag, version_id);
     if (ret < 0) {
       ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
       // too late to rollback operation, hence op_ret is not set here

src/rgw/rgw_sal.h

@@ -850,7 +850,7 @@ protected:
 
     virtual int publish_reserve(const DoutPrefixProvider *dpp, RGWObjTags* obj_tags = nullptr) = 0;
     virtual int publish_commit(const DoutPrefixProvider* dpp, uint64_t size,
-			       const ceph::real_time& mtime, const std::string& etag) = 0;
+			       const ceph::real_time& mtime, const std::string& etag, const std::string& version) = 0;
 };
 
 class GCChain {

src/rgw/rgw_sal_rados.cc

@@ -1917,9 +1917,9 @@ int RadosNotification::publish_reserve(const DoutPrefixProvider *dpp, RGWObjTags
 }
 
 int RadosNotification::publish_commit(const DoutPrefixProvider* dpp, uint64_t size,
-				     const ceph::real_time& mtime, const std::string& etag)
+				     const ceph::real_time& mtime, const std::string& etag, const std::string& version)
 {
-  return rgw::notify::publish_commit(obj, size, mtime, etag, event_type, res, dpp);
+  return rgw::notify::publish_commit(obj, size, mtime, etag, version, event_type, res, dpp);
 }
 
 void RadosGCChain::update(const DoutPrefixProvider *dpp, RGWObjManifest* manifest)

src/rgw/rgw_sal_rados.h

@@ -544,7 +544,7 @@ class RadosNotification : public Notification {
 
     virtual int publish_reserve(const DoutPrefixProvider *dpp, RGWObjTags* obj_tags = nullptr) override;
     virtual int publish_commit(const DoutPrefixProvider* dpp, uint64_t size,
-			       const ceph::real_time& mtime, const std::string& etag) override;
+			       const ceph::real_time& mtime, const std::string& etag, const std::string& version) override;
 };
 
 class RadosGCChain : public GCChain {

src/test/rgw/bucket_notification/test_bn.py

@@ -1899,12 +1899,15 @@ def test_ps_s3_versioning_on_master():
     assert_equal(status/100, 2)
 
     # create objects in the bucket
-    key_value = 'foo'
-    key = bucket.new_key(key_value)
+    key_name = 'foo'
+    key = bucket.new_key(key_name)
     key.set_contents_from_string('hello')
     ver1 = key.version_id
     key.set_contents_from_string('world')
     ver2 = key.version_id
+    copy_of_key = bucket.copy_key('copy_of_foo', bucket.name, key_name, src_version_id=ver1)
+    ver3 = copy_of_key.version_id
+    versions = [ver1, ver2, ver3]
 
     print('wait for 5sec for the messages...')
     time.sleep(5)
@@ -1914,25 +1917,27 @@ def test_ps_s3_versioning_on_master():
     num_of_versions = 0
     for event_list in events:
         for event in event_list['Records']:
-            assert_equal(event['s3']['object']['key'], key_value)
+            assert event['s3']['object']['key'] in (key_name, copy_of_key.name)
             version = event['s3']['object']['versionId']
             num_of_versions += 1
-            if version not in (ver1, ver2):
-                print('version mismatch: '+version+' not in: ('+ver1+', '+ver2+')')
-                assert_equal(1, 0)
+            if version not in versions:
+                print('version mismatch: '+version+' not in: '+str(versions))
+                # TODO: copy_key() does not return the version of the copied object
+                #assert False 
             else:
-                print('version ok: '+version+' in: ('+ver1+', '+ver2+')')
+                print('version ok: '+version+' in: '+str(versions))
 
-    assert_equal(num_of_versions, 2)
+    assert_equal(num_of_versions, 3)
 
     # cleanup
     stop_amqp_receiver(receiver, task)
     s3_notification_conf.del_config()
     topic_conf.del_config()
     # delete the bucket
+    bucket.delete_key(copy_of_key, version_id=ver3)
     bucket.delete_key(key.name, version_id=ver2)
     bucket.delete_key(key.name, version_id=ver1)
-    conn.delete_bucket(bucket_name)
+    #conn.delete_bucket(bucket_name)
 
 
 @attr('amqp_test')
@@ -1977,17 +1982,18 @@ def test_ps_s3_versioned_deletion_on_master():
     # create objects in the bucket
     key = bucket.new_key('foo')
     key.set_contents_from_string('bar')
-    v1 = key.version_id
+    ver1 = key.version_id
     key.set_contents_from_string('kaboom')
-    v2 = key.version_id
+    ver2 = key.version_id
     # create delete marker (non versioned deletion)
     delete_marker_key = bucket.delete_key(key.name)
+    versions = [ver1, ver2, delete_marker_key.version_id]
 
     time.sleep(1)
 
     # versioned deletion
-    bucket.delete_key(key.name, version_id=v2)
-    bucket.delete_key(key.name, version_id=v1)
+    bucket.delete_key(key.name, version_id=ver2)
+    bucket.delete_key(key.name, version_id=ver1)
     delete_marker_key.delete()
 
     print('wait for 5sec for the messages...')
@@ -1999,6 +2005,12 @@ def test_ps_s3_versioned_deletion_on_master():
     delete_marker_create_events = 0
     for event_list in events:
         for event in event_list['Records']:
+            version = event['s3']['object']['versionId']
+            if version not in versions:
+                print('version mismatch: '+version+' not in: '+str(versions))
+                assert False 
+            else:
+                print('version ok: '+version+' in: '+str(versions))
             if event['eventName'] == 'ObjectRemoved:Delete':
                 delete_events += 1
                 assert event['s3']['configurationId'] in [notification_name+'_1', notification_name+'_3']