Merge pull request ceph#65023 from soumyakoduri/wip-skoduri-restore-notification

yuvalif · web-flow · commit b56fccc902d8 · 2025-11-11T09:12:55.000+02:00
rgw/restore: Add notifications for restore events
diff --git a/doc/radosgw/s3-notification-compatibility.rst b/doc/radosgw/s3-notification-compatibility.rst
@@ -142,9 +142,13 @@ Event Types
 +--------------------------------------------------------+-------------------------------------------+
 | ``s3:Replication:DeletionMarkerCreated``               | Defined, Supported (not generated)        |
 +--------------------------------------------------------+-------------------------------------------+
-| ``s3:ObjectRestore:Post``                              | Not applicable                            |
+| ``s3:ObjectRestore:*``                                 | Supported                                 |
 +--------------------------------------------------------+-------------------------------------------+
-| ``s3:ObjectRestore:Complete``                          | Not applicable                            |
+| ``s3:ObjectRestore:Post``                              | Supported                                 |
++--------------------------------------------------------+-------------------------------------------+
+| ``s3:ObjectRestore:Completed``                         | Supported                                 |
++--------------------------------------------------------+-------------------------------------------+
+| ``s3:ObjectRestore:Delete``                            | Supported                                 |
 +--------------------------------------------------------+-------------------------------------------+
 | ``s3:ReducedRedundancyLostObject``                     | Not applicable                            |
 +--------------------------------------------------------+-------------------------------------------+
diff --git a/src/rgw/driver/daos/rgw_sal_daos.cc b/src/rgw/driver/daos/rgw_sal_daos.cc
@@ -1035,6 +1035,7 @@ int DaosObject::restore_obj_from_cloud(Bucket* bucket,
           uint64_t olh_epoch,
           std::optional<uint64_t> days,
 	  bool& in_progress,
+	  uint64_t& size,
           const DoutPrefixProvider* dpp, 
           optional_yield y)
 {
diff --git a/src/rgw/driver/daos/rgw_sal_daos.h b/src/rgw/driver/daos/rgw_sal_daos.h
@@ -661,6 +661,7 @@ class DaosObject : public StoreObject {
 			   uint64_t olh_epoch,
 			   std::optional<uint64_t> days,
 		           bool& in_progress,
+		           uint64_t& size,
 			   const DoutPrefixProvider* dpp,
 			   optional_yield y) override;
   virtual bool placement_rules_match(rgw_placement_rule& r1,
diff --git a/src/rgw/driver/posix/rgw_sal_posix.cc b/src/rgw/driver/posix/rgw_sal_posix.cc
@@ -3243,6 +3243,7 @@ int POSIXObject::restore_obj_from_cloud(Bucket* bucket,
 	  CephContext* cct,
           std::optional<uint64_t> days,
           bool& in_progress,
+	  uint64_t& size,
           const DoutPrefixProvider* dpp, 
           optional_yield y)
 {
diff --git a/src/rgw/driver/posix/rgw_sal_posix.h b/src/rgw/driver/posix/rgw_sal_posix.h
@@ -1080,6 +1080,7 @@ class POSIXObject : public StoreObject {
 			   CephContext* cct,
 			   std::optional<uint64_t> days,
 		           bool& in_progress,
+			   uint64_t& size,
 			   const DoutPrefixProvider* dpp,
 			   optional_yield y) override;
   virtual bool placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2) override;
diff --git a/src/rgw/driver/rados/rgw_rados.cc b/src/rgw/driver/rados/rgw_rados.cc
@@ -5472,6 +5472,7 @@ int RGWRados::restore_obj_from_cloud(RGWLCCloudTierCtx& tier_ctx,
                              RGWObjTier& tier_config,
                              std::optional<uint64_t> days,
 			     bool& in_progress,
+			     uint64_t& size,
                              const DoutPrefixProvider *dpp,
                              optional_yield y) {
 
@@ -5693,7 +5694,9 @@ int RGWRados::restore_obj_from_cloud(RGWLCCloudTierCtx& tier_ctx,
     return ret;
   }
 
-  return ret; 
+  // set size to be used to send bucket notification
+  size = accounted_size;
+  return 0;
 }
 
 int RGWRados::check_bucket_empty(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_info, optional_yield y)
diff --git a/src/rgw/driver/rados/rgw_rados.h b/src/rgw/driver/rados/rgw_rados.h
@@ -1294,6 +1294,7 @@ int restore_obj_from_cloud(RGWLCCloudTierCtx& tier_ctx,
                              RGWObjTier& tier_config,
                              std::optional<uint64_t> days,
 			     bool& in_progress,
+			     uint64_t& size,
                              const DoutPrefixProvider *dpp,
                              optional_yield y);
 
diff --git a/src/rgw/driver/rados/rgw_sal_rados.cc b/src/rgw/driver/rados/rgw_sal_rados.cc
@@ -3079,6 +3079,7 @@ int RadosObject::restore_obj_from_cloud(Bucket* bucket,
                           	  CephContext* cct,
                                   std::optional<uint64_t> days,
 				  bool& in_progress,
+				  uint64_t& size,
                                   const DoutPrefixProvider* dpp, 
                                   optional_yield y)
 {
@@ -3154,7 +3155,7 @@ int RadosObject::restore_obj_from_cloud(Bucket* bucket,
    * avoid any races */
   ret = store->getRados()->restore_obj_from_cloud(tier_ctx, *rados_ctx,
                                 bucket->get_info(), get_obj(),
-                                tier_config, days, in_progress, dpp, y);
+                                tier_config, days, in_progress, size, dpp, y);
 
   if (ret < 0) { //failed to restore
     ldpp_dout(dpp, 0) << "Restoring object(" << get_key() << ") from the cloud endpoint(" << endpoint << ") failed, ret=" << ret << dendl;
@@ -3364,13 +3365,26 @@ int RadosObject::handle_obj_expiry(const DoutPrefixProvider* dpp, optional_yield
 	    attrs[RGW_ATTR_INTERNAL_MTIME] = std::move(bl);
 	  }
           const req_context rctx{dpp, y, nullptr};
-          return obj_op.write_meta(0, 0, attrs, rctx, head_obj->get_trace(), false);
+          ret = obj_op.write_meta(0, 0, attrs, rctx, head_obj->get_trace(), false);
+
+          // send notification in case the temporary copy of restored obj is expired
+          if (!ret) { //send notification
+            string etag;
+            attr_iter = attrs.find(RGW_ATTR_ETAG);
+            if (attr_iter != attrs.end()) {
+              etag = rgw_bl_str(attr_iter->second);
+            }
+            store->get_rgwrestore()->send_notification(dpp, store, this, bucket, etag, 0,
+                      get_obj().key.instance,
+                      {rgw::notify::ObjectRestoreExpired}, y);
+
+          }
         } catch (const buffer::end_of_buffer&) {
           // ignore empty manifest; it's not cloud-tiered
         } catch (const std::exception& e) {
         }
       }
-      return 0;
+      return ret;
     }
   }
   // object is not restored/temporary; go for regular deletion
diff --git a/src/rgw/driver/rados/rgw_sal_rados.h b/src/rgw/driver/rados/rgw_sal_rados.h
@@ -644,6 +644,7 @@ class RadosObject : public StoreObject {
 			   CephContext* cct,
   		           std::optional<uint64_t> days,
 			   bool& in_progress,
+			   uint64_t& size,
 			   const DoutPrefixProvider* dpp,
 			   optional_yield y) override;
     virtual bool placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2) override;
diff --git a/src/rgw/rgw_notify_event_type.cc b/src/rgw/rgw_notify_event_type.cc
@@ -66,6 +66,14 @@ namespace rgw::notify {
       return "s3:Replication:Delete";
     case ReplicationDeletionMarkerCreated:
       return "s3:Replication:DeletionMarkerCreated";
+    case ObjectRestore:
+      return "s3:ObjectRestore:*";
+    case ObjectRestoreInitiated:
+      return "s3:ObjectRestore:Post";
+    case ObjectRestoreCompleted:
+      return "s3:ObjectRestore:Completed";
+    case ObjectRestoreExpired:
+      return "s3:ObjectRestore:Delete";
     case UnknownEvent:
         return "s3:UnknownEvent";
     }
@@ -139,6 +147,14 @@ namespace rgw::notify {
       return ReplicationDelete;
     if (s == "s3:Replication:DeletionMarkerCreated")
       return ReplicationDeletionMarkerCreated;
+    if (s =="s3:ObjectRestore:*")
+      return ObjectRestore;
+    if (s == "s3:ObjectRestore:Post")
+      return ObjectRestoreInitiated;
+    if (s == "s3:ObjectRestore:Completed")
+      return ObjectRestoreCompleted;
+    if (s == "s3:ObjectRestore:Delete")
+      return ObjectRestoreExpired;
     return UnknownEvent;
   }
 
diff --git a/src/rgw/rgw_notify_event_type.h b/src/rgw/rgw_notify_event_type.h
@@ -39,7 +39,11 @@ namespace rgw::notify {
     ReplicationCreate                      = 0x10000000,
     ReplicationDelete                      = 0x20000000,
     ReplicationDeletionMarkerCreated       = 0x40000000,
-    UnknownEvent                           = 0x100000000
+    ObjectRestore                          = 0xF00000000,
+    ObjectRestoreInitiated                 = 0x100000000,
+    ObjectRestoreCompleted                 = 0x200000000,
+    ObjectRestoreExpired                   = 0x400000000,
+    UnknownEvent                           = 0x1000000000
 };
 
   using EventTypeList = std::vector<EventType>;
diff --git a/src/rgw/rgw_restore.cc b/src/rgw/rgw_restore.cc
@@ -109,6 +109,43 @@ void RestoreEntry::generate_test_instances(std::list<RestoreEntry*>& l)
   l.push_back(new RestoreEntry);
 }
 
+static std::string restore_id = "rgw restore";
+static std::string restore_req_id = "0";
+
+void Restore::send_notification(const DoutPrefixProvider* dpp,
+                              rgw::sal::Driver* driver,
+                              rgw::sal::Object* obj,
+                              rgw::sal::Bucket* bucket,
+                              const std::string& etag,
+                              uint64_t size,
+                              const std::string& version_id,
+                              const rgw::notify::EventTypeList& event_types,
+                              optional_yield y) {
+  // notification supported only for RADOS driver for now
+  auto notify = driver->get_notification(
+      dpp, obj, nullptr, event_types, bucket, restore_id,
+      const_cast<std::string&>(bucket->get_tenant()), restore_req_id, y);
+
+  if (!notify) {
+    return;
+  }
+
+  int ret = notify->publish_reserve(dpp, nullptr);
+  if (ret < 0) {
+    ldpp_dout(dpp, 1) << "ERROR: notify publish_reserve failed with error: "
+                      << ret << " for restore object: " << obj->get_name()
+                      << " for event_types: " << event_types << dendl;
+    return;
+  }
+  ret = notify->publish_commit(dpp, size, ceph::real_clock::now(), etag,
+                               version_id);
+  if (ret < 0) {
+    ldpp_dout(dpp, 5) << "WARNING: notify publish_commit failed with error: "
+                      << ret << " for lc object: " << obj->get_name()
+                      << " for event_types: " << event_types << dendl;
+  }
+}
+
 int Restore::initialize(CephContext *_cct, rgw::sal::Driver* _driver) {
   int ret = 0;
   cct = _cct;
@@ -458,6 +495,7 @@ int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
   } else {
     ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: Attr RGW_ATTR_STORAGE_CLASS not found for object: " << obj->get_key() << dendl;
   }
+
   ret = driver->get_zone()->get_zonegroup().get_placement_tier(target_placement, &tier);
 
   if (ret < 0) {
@@ -476,9 +514,9 @@ int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
     goto done;
   }
 
+  uint64_t size;
   // now go ahead with restoring object
-  // XXX: first check if its already restored?
-  ret = obj->restore_obj_from_cloud(bucket.get(), tier.get(), cct, days, in_progress,
+  ret = obj->restore_obj_from_cloud(bucket.get(), tier.get(), cct, days, in_progress, size, 
 		  		      this, y);
   if (ret < 0) {
     ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": Restore of object(" << obj->get_key() << ") failed" << ret << dendl;	  
@@ -495,6 +533,17 @@ int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
   } else {
     ldpp_dout(this, 15) << __PRETTY_FUNCTION__ << ": Restore of object " << obj->get_key() << " succeeded" << dendl;
     entry.status = rgw::sal::RGWRestoreStatus::CloudRestored;
+    
+    string etag;
+    attr_iter = attrs.find(RGW_ATTR_ETAG);
+    if (attr_iter != attrs.end()) {
+      etag = rgw_bl_str(attr_iter->second);
+    }
+
+    // send notification in case the restore is successfully completed
+    send_notification(this, driver, obj.get(), bucket.get(), etag, size,
+                      obj->get_key().instance,
+                      {rgw::notify::ObjectRestoreCompleted}, y);
   }
 
 done:
@@ -612,11 +661,11 @@ int Restore::update_cloud_restore_exp_date(rgw::sal::Bucket* pbucket,
 }
 
 int Restore::restore_obj_from_cloud(rgw::sal::Bucket* pbucket,
-	       			       rgw::sal::Object* pobj,
-				       rgw::sal::PlacementTier* tier,
-				       std::optional<uint64_t> days,
-				       const DoutPrefixProvider* dpp,
-				       optional_yield y)
+	       			                      rgw::sal::Object* pobj,
+                       			        rgw::sal::PlacementTier* tier,
+                     				        std::optional<uint64_t> days,
+                    				        const DoutPrefixProvider* dpp,
+                    				        optional_yield y)
 {
   int ret = 0;
 
@@ -625,6 +674,22 @@ int Restore::restore_obj_from_cloud(rgw::sal::Bucket* pbucket,
     return -EINVAL;
   }
 
+  auto notify = driver->get_notification(
+      dpp, pobj, nullptr,
+      {rgw::notify::ObjectRestoreInitiated},
+      pbucket, restore_id,
+      const_cast<std::string&>(pbucket->get_tenant()), restore_req_id, y);
+
+  if (notify) {
+    int ret = notify->publish_reserve(dpp, nullptr);
+    if (ret < 0) {
+      ldpp_dout(dpp, 1) << "ERROR: notify publish_reserve failed with error: "
+      	                << ret << " for restore object: " << pobj->get_name()
+        	              << " for event_types: rgw::notify::ObjectRestoreInitiated" << dendl;
+      return ret;
+    }
+  }
+
   // set restore_status as RESTORE_ALREADY_IN_PROGRESS
   ret = set_cloud_restore_status(this, pobj, y, rgw::sal::RGWRestoreStatus::RestoreAlreadyInProgress);
   if (ret < 0) {
@@ -663,6 +728,24 @@ int Restore::restore_obj_from_cloud(rgw::sal::Bucket* pbucket,
   }
 
   ldpp_dout(this, 10) << __PRETTY_FUNCTION__ << ": Restore of object " << pobj->get_key() << " is in progress." << dendl;  
+
+  if (notify) {
+    auto& attrs = pobj->get_attrs();
+    string etag;
+    auto attr_iter = attrs.find(RGW_ATTR_ETAG);
+    if (attr_iter != attrs.end()) {
+      etag = rgw_bl_str(attr_iter->second);
+    }
+
+    ret = notify->publish_commit(dpp, pobj->get_size(), ceph::real_clock::now(), etag,
+		    		 pobj->get_key().instance);
+    if (ret < 0) {
+      ldpp_dout(dpp, 5) << "WARNING: notify publish_commit failed with error: "
+                        << ret << " for lc object: " << pobj->get_name()
+                        << " for event_types: rgw::notify::ObjectRestoreInitiated" << dendl;
+    }
+  }
+
   return ret;
 }
 
diff --git a/src/rgw/rgw_restore.h b/src/rgw/rgw_restore.h
@@ -19,6 +19,7 @@
 #include "rgw_common.h"
 #include "cls/rgw/cls_rgw_types.h"
 #include "rgw_sal.h"
+#include "rgw_notify.h"
 
 #include <atomic>
 #include <tuple>
@@ -150,6 +151,20 @@ class Restore : public DoutPrefixProvider {
 			     std::optional<uint64_t> days,
 			     const DoutPrefixProvider* dpp,
 			     optional_yield y);
+
+  /**
+   * Send notification incase of restore events
+   */
+
+  void send_notification(const DoutPrefixProvider* dpp,
+                              rgw::sal::Driver* driver,
+                              rgw::sal::Object* obj,
+                              rgw::sal::Bucket* bucket,
+                              const std::string& etag,
+                              uint64_t size,
+                              const std::string& version_id,
+                              const rgw::notify::EventTypeList& event_types,
+                              optional_yield y);
 };
 
 } // namespace rgw::restore
diff --git a/src/rgw/rgw_sal.h b/src/rgw/rgw_sal.h
@@ -1283,6 +1283,7 @@ class Object {
 			   CephContext* cct,
 		           std::optional<uint64_t> days,
    	 		   bool& in_progress,
+			   uint64_t& size,
 			   const DoutPrefixProvider* dpp,
 			   optional_yield y) = 0;
     /** Check to see if two placement rules match */
diff --git a/src/rgw/rgw_sal_filter.cc b/src/rgw/rgw_sal_filter.cc
@@ -1150,11 +1150,12 @@ int FilterObject::restore_obj_from_cloud(Bucket* bucket,
 		          CephContext* cct,
 		          std::optional<uint64_t> days,
 			  bool& in_progress,
+		          uint64_t& size,
 		          const DoutPrefixProvider* dpp,
 		          optional_yield y)
 {
   return next->restore_obj_from_cloud(nextBucket(bucket), nextPlacementTier(tier),
-           cct, days, in_progress, dpp, y);
+           cct, days, in_progress, size, dpp, y);
 }
 
 bool FilterObject::placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2)
diff --git a/src/rgw/rgw_sal_filter.h b/src/rgw/rgw_sal_filter.h
@@ -840,6 +840,7 @@ class FilterObject : public Object {
 			   CephContext* cct,
 		           std::optional<uint64_t> days,
 			   bool& in_progress,
+			   uint64_t& size,
 			   const DoutPrefixProvider* dpp,
 			   optional_yield y) override;
   virtual bool placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2) override;
diff --git a/src/rgw/rgw_sal_store.h b/src/rgw/rgw_sal_store.h
@@ -373,6 +373,7 @@ class StoreObject : public Object {
 			   CephContext* cct,
 		           std::optional<uint64_t> days,
 			   bool& in_progress,
+			   uint64_t& size,
 			   const DoutPrefixProvider* dpp,
 			   optional_yield y) override {
       return -1;

Original file line number	Diff line number	Diff line change
`@@ -1035,6 +1035,7 @@ int DaosObject::restore_obj_from_cloud(Bucket* bucket,`
`1035`	`1035`	`uint64_t olh_epoch,`
`1036`	`1036`	`std::optional<uint64_t> days,`
`1037`	`1037`	`bool& in_progress,`
	`1038`	`+ uint64_t& size,`
`1038`	`1039`	`const DoutPrefixProvider* dpp,`
`1039`	`1040`	`optional_yield y)`
`1040`	`1041`	`{`
Original file line number	Diff line number	Diff line change
`@@ -3243,6 +3243,7 @@ int POSIXObject::restore_obj_from_cloud(Bucket* bucket,`
`3243`	`3243`	`CephContext* cct,`
`3244`	`3244`	`std::optional<uint64_t> days,`
`3245`	`3245`	`bool& in_progress,`
	`3246`	`+ uint64_t& size,`
`3246`	`3247`	`const DoutPrefixProvider* dpp,`
`3247`	`3248`	`optional_yield y)`
`3248`	`3249`	`{`