rgw/logging: fix race condition when name update returns ECANCELED

* when we get ECANCELED indication from the name set operation we should
  bail out and not continue with the rollover
* this fix revealed a hidden bug where we do not check the existing temp
  name when we do conf change cleanup (rollover)

Fixes: https://tracker.ceph.com/issues/73434

Signed-off-by: Yuval Lifshitz <[email protected]>

Author: yuvalif

src/rgw/radosgw-admin/radosgw-admin.cc

@@ -7813,9 +7813,8 @@ int main(int argc, const char **argv)
     const auto region = driver->get_zone()->get_zonegroup().get_api_name();
     ret = rgw::bucketlogging::rollover_logging_object(configuration, target_bucket, obj_name, dpp(), region, bucket, null_yield, true, &objv_tracker, &old_obj);
     if (ret < 0) {
-      cerr << "ERROR: failed to flush pending logging object '" << obj_name << "' to target bucket '" << configuration.target_bucket << "'. "
-        << " last committed object is '" << old_obj <<
-        "'. error: " << cpp_strerror(-ret) << std::endl;
+      cerr << "ERROR: failed to flush pending logging object '" << obj_name << "' to target bucket '" << configuration.target_bucket
+        << "'. error: " << cpp_strerror(-ret) << std::endl;
       return -ret;
     }
     cout << "flushed pending logging object '" << old_obj

src/rgw/rgw_bucket_logging.cc

@@ -373,16 +373,19 @@ int rollover_logging_object(const configuration& conf,
   auto old_obj = obj_name.empty() ? std::nullopt : std::optional<std::string>(obj_name);
 
   auto handle_error = [&dpp, &old_obj, &target_bucket, err_message](int ret) {
-    if (ret == -ECANCELED) {
-      ldpp_dout(dpp, 20) << "INFO: rollover already performed for logging object '" << old_obj <<  "' to logging bucket '" <<
-        target_bucket->get_key() << "'. ret = " << ret << dendl;
-      return 0;
-    }
     if (ret < 0) {
-      ldpp_dout(dpp, 1) << "ERROR: failed to rollover logging object '" << old_obj << "' to logging bucket '" <<
-        target_bucket->get_key() << "'. ret = " << ret << dendl;
-      if (err_message) {
-        *err_message = fmt::format("Failed to rollover logging object of logging bucket '{}'", target_bucket->get_name());
+      if (ret == -ECANCELED) {
+        ldpp_dout(dpp, 20) << "INFO: rollover already performed for logging object '" << old_obj <<  "' to logging bucket '" <<
+          target_bucket->get_key() << "'. ret = " << ret << dendl;
+        if (err_message) {
+          *err_message = fmt::format("Rollover already performed on logging bucket '{}'", target_bucket->get_name());
+        }
+      } else {
+        ldpp_dout(dpp, 1) << "ERROR: failed to rollover logging object '" << old_obj << "' to logging bucket '" <<
+          target_bucket->get_key() << "'. ret = " << ret << dendl;
+        if (err_message) {
+          *err_message = fmt::format("Failed to rollover logging object of logging bucket '{}'", target_bucket->get_name());
+        }
       }
     }
     return ret;
@@ -515,7 +518,7 @@ int log_record(rgw::sal::Driver* driver,
     if (ceph::coarse_real_time::clock::now() > time_to_commit) {
       ldpp_dout(dpp, 20) << "INFO: logging object '" << obj_name << "' exceeded its time, will be committed to logging bucket '" <<
         target_bucket_id << "'" << dendl;
-      if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, false, &objv_tracker, nullptr, &err_message); ret < 0) {
+      if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, false, &objv_tracker, nullptr, &err_message); ret < 0 && ret != -ECANCELED) {
         set_journal_err(err_message);
         return ret;
       }
@@ -666,7 +669,7 @@ int log_record(rgw::sal::Driver* driver,
   if (ret == -EFBIG) {
     ldpp_dout(dpp, 5) << "WARNING: logging object '" << obj_name << "' is full, will be committed to logging bucket '" <<
       target_bucket->get_key() << "'" << dendl;
-    if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, true, &objv_tracker, nullptr, &err_message); ret < 0 ) {
+    if (ret = rollover_logging_object(conf, target_bucket, obj_name, dpp, region, s->bucket, y, true, &objv_tracker, nullptr, &err_message); ret < 0 && ret != -ECANCELED) {
       set_journal_err(err_message);
       return ret;
     }

src/rgw/rgw_rest_bucket_logging.cc

@@ -311,6 +311,12 @@ class RGWPutBucketLoggingOp : public RGWDefaultResponseOp {
       // conf changed - do cleanup
       RGWObjVersionTracker objv_tracker;
       std::string obj_name;
+      if (int ret = target_bucket->get_logging_object_name(obj_name, old_conf->target_prefix, y, this, nullptr); ret < 0 && ret != -ENOENT) {
+        ldpp_dout(this, 1) << "ERROR: failed to get name of logging object of logging bucket '" <<
+        target_bucket_id << "' and prefix '" << configuration.target_prefix << "', ret = " << ret << dendl;
+        op_ret = ret;
+        return;
+      }
       const auto region = driver->get_zone()->get_zonegroup().get_api_name();
       if (const auto ret = rollover_logging_object(*old_conf,
             target_bucket,
@@ -324,8 +330,7 @@ class RGWPutBucketLoggingOp : public RGWDefaultResponseOp {
             &old_obj); ret < 0) {
         ldpp_dout(this, 1) << "WARNING: failed to flush pending logging object '" << obj_name << "'"
             << " to target bucket '" << target_bucket_id << "'. "
-            << " last committed object is '" << old_obj <<
-            "' when updating logging configuration of bucket '" << src_bucket->get_key() << ". error: " << ret << dendl;
+            << "' when updating logging configuration of bucket '" << src_bucket->get_key() << ". error: " << ret << dendl;
       } else {
         ldpp_dout(this, 20) << "INFO: flushed pending logging object '" << old_obj
           << "' to target bucket '" << target_bucket_id << "' when updating logging configuration of bucket '"