rgw/cloud-tier: retry on 503 from remote endpoint

We were aborting the lifecycle transition or restore on the first 503
SlowDown back from the remote. Wrap both paths in a small retry helper
that backs off and tries again on -EBUSY, using the caller's yield
context to sleep.

Tunable via rgw_cloud_tier_retry_limit (8), rgw_cloud_tier_retry_delay_ms
(500), and rgw_cloud_tier_retry_max_ms (30000). On the streamed restore
GET, RGWRadosPutObj::handle_headers rejects 5xx so the response body
doesn't get streamed into the put-processor before we know to retry.

This also affects the multi-zone fetch path that uses the same
put-processor; 503 there now returns -EBUSY cleanly instead of
streaming the error body.

Fixes: https://tracker.ceph.com/issues/76406
Co-authored-by: Kevin Lott <klott@backblaze.com>
Signed-off-by: Matthew N. Heler <matthew.heler@hotmail.com>

doc/radosgw/cloud-transition.rst

@@ -472,6 +472,19 @@ The objects transitioned to cloud can now be restored. For more information, ref
 :ref:`Restoring Objects from Cloud <radosgw-cloud-restore>`.
 
 
+Retry Behavior
+--------------
+When the remote cloud endpoint reports a transient error, RGW retries the
+request with backoff rather than failing the lifecycle action immediately.
+
+If the retry budget is exhausted the operation is not completed and is
+retried on the next lifecycle pass.
+
+.. confval:: rgw_cloud_tier_retry_limit
+.. confval:: rgw_cloud_tier_retry_delay_ms
+.. confval:: rgw_cloud_tier_retry_max_ms
+
+
 Future Work
 -----------
 

src/common/options/rgw.yaml.in

@@ -557,6 +557,40 @@ options:
   see_also:
   - rgw_lc_counters_cache
   with_legacy: true
+- name: rgw_cloud_tier_retry_limit
+  type: int
+  level: advanced
+  desc: Cloud-tier retry attempt limit
+  long_desc: Maximum number of retry attempts when a remote cloud-tier endpoint
+    signals throttling.
+  default: 8
+  min: 1
+  services:
+  - rgw
+  with_legacy: true
+- name: rgw_cloud_tier_retry_delay_ms
+  type: int
+  level: advanced
+  desc: Cloud-tier retry initial delay
+  long_desc: Number of milliseconds to wait before the first retry of a throttled
+    cloud-tier request. Subsequent retries grow this delay exponentially up to
+    rgw_cloud_tier_retry_max_ms.
+  default: 500
+  min: 1
+  services:
+  - rgw
+  with_legacy: true
+- name: rgw_cloud_tier_retry_max_ms
+  type: int
+  level: advanced
+  desc: Cloud-tier retry maximum delay
+  long_desc: Maximum number of milliseconds to wait between retries of a throttled
+    cloud-tier request after exponential backoff.
+  default: 30000
+  min: 1
+  services:
+  - rgw
+  with_legacy: true
 - name: rgw_restore_debug_interval
   type: int
   level: dev

src/rgw/driver/rados/rgw_lc_tier.cc

@@ -2,8 +2,14 @@
 // vim: ts=8 sw=2 sts=2 expandtab ft=cpp
 
 #include <string.h>
+#include <algorithm>
+#include <chrono>
 #include <iostream>
 #include <map>
+#include <random>
+#include <thread>
+
+#include <boost/asio/steady_timer.hpp>
 
 #include "common/XMLFormatter.h"
 #include <common/errno.h>
@@ -25,6 +31,41 @@
 
 using namespace std;
 
+int retry_on_busy(optional_yield y, const DoutPrefixProvider *dpp,
+                  CephContext *cct, const char *op_name,
+                  std::function<int()> op)
+{
+  const int max_attempts = cct->_conf.get_val<int64_t>("rgw_cloud_tier_retry_limit");
+  const int64_t initial_ms = cct->_conf.get_val<int64_t>("rgw_cloud_tier_retry_delay_ms");
+  const int64_t max_ms = cct->_conf.get_val<int64_t>("rgw_cloud_tier_retry_max_ms");
+  thread_local std::mt19937 rng{std::random_device{}()};
+
+  int ret = 0;
+  for (int i = 0; i < max_attempts; i++) {
+    ret = op();
+    if (ret != -EBUSY || i == max_attempts - 1) return ret;
+
+    int64_t base = std::min(initial_ms << std::min(i, 30), max_ms);
+    int delay_ms = base - std::uniform_int_distribution<int>(0, base / 10)(rng);
+
+    ldpp_dout(dpp, 1) << op_name << ": -EBUSY, attempt " << (i + 1) << "/"
+                      << max_attempts << "; retrying after " << delay_ms
+                      << "ms" << dendl;
+
+    if (y) {
+      auto& yc = y.get_yield_context();
+      boost::asio::steady_timer t(yc.get_executor());
+      t.expires_after(std::chrono::milliseconds(delay_ms));
+      boost::system::error_code ec;
+      t.async_wait(yc[ec]);
+      if (ec) return ret;
+    } else {
+      std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
+    }
+  }
+  return ret;
+}
+
 struct rgw_lc_multipart_part_info {
   int part_num{0};
   uint64_t ofs{0};
@@ -283,44 +324,44 @@ int rgw_cloud_tier_restore_object(RGWLCCloudTierCtx& tier_ctx,
   dest_bucket.name = tier_ctx.target_bucket_name;
   target_obj_name = make_target_obj_name(tier_ctx);
 
-  if (!in_progress) { // first time. Send RESTORE req.
+  rgw_obj dest_obj(dest_bucket, rgw_obj_key(target_obj_name));
 
-    rgw_obj dest_obj(dest_bucket, rgw_obj_key(target_obj_name));
-    ret = cloud_tier_restore(tier_ctx.dpp, tier_ctx.conn, dest_obj, days, glacier_params, tier_ctx.y);
-
-    ldpp_dout(tier_ctx.dpp, 20) << __func__ << "Restoring object=" << target_obj_name << "returned ret = " << ret << dendl;
-
-    if (ret < 0 ) {
-      ldpp_dout(tier_ctx.dpp, -1) << __func__ << "ERROR: failed to restore object=" << dest_obj << "; ret = " << ret << dendl;
-      return ret;
-    }
-    in_progress = true;
-  }
-
-  // now send HEAD request and verify if restore is complete on glacier/tape endpoint
-  static constexpr int MAX_RETRIES = 2;
-  uint32_t retries = 0;
-  do {
-    ret = rgw_cloud_tier_get_object(tier_ctx, true, headers, nullptr, etag,
-                                    accounted_size, attrs, nullptr);
-
-    if (ret < 0) {
-      ldpp_dout(tier_ctx.dpp, 0) << __func__ << "ERROR: failed to fetch HEAD from cloud for obj=" << tier_ctx.obj << " , ret = " << ret << dendl;
-      return ret;
+  ret = retry_on_busy(tier_ctx.y, tier_ctx.dpp, tier_ctx.cct, __func__, [&]() -> int {
+    if (!in_progress) { // first time. Send RESTORE req.
+      ret = cloud_tier_restore(tier_ctx.dpp, tier_ctx.conn, dest_obj, days, glacier_params, tier_ctx.y);
+      ldpp_dout(tier_ctx.dpp, 20) << __func__ << "Restoring object=" << target_obj_name << "returned ret = " << ret << dendl;
+      if (ret < 0 ) {
+        ldpp_dout(tier_ctx.dpp, -1) << __func__ << "ERROR: failed to restore object=" << dest_obj << "; ret = " << ret << dendl;
+        return ret;
+      }
+      in_progress = true;
     }
 
-    in_progress = is_restore_in_progress(tier_ctx.dpp, headers);
-
-  } while(retries++ < MAX_RETRIES && in_progress);
+    // now send HEAD request and verify if restore is complete on glacier/tape endpoint
+    static constexpr int MAX_RETRIES = 2;
+    uint32_t retries = 0;
+    do {
+      ret = rgw_cloud_tier_get_object(tier_ctx, true, headers, nullptr, etag,
+                                      accounted_size, attrs, nullptr);
+      if (ret < 0) {
+        ldpp_dout(tier_ctx.dpp, 0) << __func__ << "ERROR: failed to fetch HEAD from cloud for obj=" << tier_ctx.obj << " , ret = " << ret << dendl;
+        return ret;
+      }
+      in_progress = is_restore_in_progress(tier_ctx.dpp, headers);
+    } while(retries++ < MAX_RETRIES && in_progress);
+    return 0;
+  });
+  if (ret < 0) return ret;
 
   if (in_progress) {
     ldpp_dout(tier_ctx.dpp, 20) << __func__ << "Restoring object=" << target_obj_name << " still in progress; returning " << dendl;
     return 0;
-  } 
+  }
 
-  // now do the actual GET
-  ret = rgw_cloud_tier_get_object(tier_ctx, false, headers, pset_mtime, etag,
-                                  accounted_size, attrs, cb);
+  ret = retry_on_busy(tier_ctx.y, tier_ctx.dpp, tier_ctx.cct, __func__, [&]() {
+    return rgw_cloud_tier_get_object(tier_ctx, false, headers, pset_mtime,
+                                     etag, accounted_size, attrs, cb);
+  });
 
   ldpp_dout(tier_ctx.dpp, 20) << __func__ << "(): fetching object from cloud bucket:" << dest_bucket << ", object: " << target_obj_name << " returned ret:" << ret << dendl;
 
@@ -987,29 +1028,20 @@ static int cloud_tier_send_multipart_part(RGWLCCloudTierCtx& tier_ctx,
 
   tier_ctx.obj->set_atomic(true);
 
-  /* TODO: Define readf, writef as stack variables. For some reason,
-   * when used as stack variables (esp., readf), the transition seems to
-   * be taking lot of time eventually erroring out at times. */
-  std::shared_ptr<RGWLCStreamRead> readf;
-  readf.reset(new RGWLCStreamRead(tier_ctx.cct, tier_ctx.dpp,
-        tier_ctx.obj, tier_ctx.o.meta.mtime, tier_ctx.y));
-
-  std::shared_ptr<RGWLCCloudStreamPut> writef;
-  writef.reset(new RGWLCCloudStreamPut(tier_ctx.dpp, obj_properties, tier_ctx.conn,
-               dest_obj, tier_ctx.y));
-
-  /* Prepare Read from source */
   end = part_info.ofs + part_info.size - 1;
-  readf->set_multipart(part_info.size, part_info.ofs, end);
-
-  /* Prepare write */
-  writef->set_multipart(upload_id, part_info.part_num, part_info.size);
-
-  /* actual Read & Write */
-  ret = cloud_tier_transfer_object(tier_ctx.dpp, readf.get(), writef.get());
-  if (ret < 0) {
-    return ret;
-  }
+  std::shared_ptr<RGWLCCloudStreamPut> writef;
+  ret = retry_on_busy(tier_ctx.y, tier_ctx.dpp, tier_ctx.cct, __func__, [&]() {
+    auto rf = std::make_shared<RGWLCStreamRead>(tier_ctx.cct, tier_ctx.dpp,
+          tier_ctx.obj, tier_ctx.o.meta.mtime, tier_ctx.y);
+    auto wf = std::make_shared<RGWLCCloudStreamPut>(tier_ctx.dpp,
+          obj_properties, tier_ctx.conn, dest_obj, tier_ctx.y);
+    rf->set_multipart(part_info.size, part_info.ofs, end);
+    wf->set_multipart(upload_id, part_info.part_num, part_info.size);
+    int r = cloud_tier_transfer_object(tier_ctx.dpp, rf.get(), wf.get());
+    if (r == 0) writef = wf;
+    return r;
+  });
+  if (ret < 0) return ret;
 
   if (!(writef->get_etag(petag))) {
     ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to get etag from PUT request" << dendl;
@@ -1071,6 +1103,7 @@ int cloud_tier_restore(const DoutPrefixProvider *dpp, RGWRESTConn& dest_conn,
   ret = dest_conn.send_resource(dpp, "POST", resource, params, nullptr,
                                 out_bl, &bl, nullptr, y);
 
+  if (ret == -EBUSY) return ret;
   if (ret < 0) {
     ldpp_dout(dpp, 0) << __func__ << "ERROR: failed to send Restore request to cloud for obj=" << dest_obj << " , ret = " << ret << dendl;
   } else {
@@ -1413,6 +1446,7 @@ static int cloud_tier_multipart_transfer(RGWLCCloudTierCtx& tier_ctx) {
             cur_part_info,
             &cur_part_info.etag);
 
+    if (ret == -EBUSY) return ret;
     if (ret < 0) {
       ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to send multipart part of obj=" << tier_ctx.obj << ", sync via multipart upload, upload_id=" << status.upload_id << " part number " << cur_part << " (error: " << cpp_strerror(-ret) << ")" << dendl;
       cloud_tier_abort_multipart_upload(tier_ctx, dest_obj, status_obj, status.upload_id);
@@ -1422,6 +1456,7 @@ static int cloud_tier_multipart_transfer(RGWLCCloudTierCtx& tier_ctx) {
   }
 
   ret = cloud_tier_complete_multipart(tier_ctx.dpp, tier_ctx.conn, dest_obj, status.upload_id, parts, tier_ctx.y);
+  if (ret == -EBUSY) return ret;
   if (ret < 0) {
     ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to complete multipart upload of obj=" << tier_ctx.obj << " (error: " << cpp_strerror(-ret) << ")" << dendl;
     cloud_tier_abort_multipart_upload(tier_ctx, dest_obj, status_obj, status.upload_id);
@@ -1525,7 +1560,8 @@ static int cloud_tier_create_bucket(RGWLCCloudTierCtx& tier_ctx) {
   ret = tier_ctx.conn.send_resource(tier_ctx.dpp, "PUT", resource, nullptr, nullptr,
                                     out_bl, &bl, nullptr, tier_ctx.y);
 
-  if (ret < 0 ) {
+  if (ret == -EBUSY) return ret;
+  if (ret < 0) {
     ldpp_dout(tier_ctx.dpp, 0) << "create target bucket : " << tier_ctx.target_bucket_name << " returned ret:" << ret << dendl;
   }
   if (out_bl.length() > 0) {
@@ -1558,7 +1594,7 @@ static int cloud_tier_create_bucket(RGWLCCloudTierCtx& tier_ctx) {
   return 0;
 }
 
-int rgw_cloud_tier_transfer_object(RGWLCCloudTierCtx& tier_ctx, std::set<std::string>& cloud_targets) {
+static int do_cloud_tier_transfer_object(RGWLCCloudTierCtx& tier_ctx, std::set<std::string>& cloud_targets) {
   int ret = 0;
 
   // check if target bucket is in local cache
@@ -1589,6 +1625,7 @@ int rgw_cloud_tier_transfer_object(RGWLCCloudTierCtx& tier_ctx, std::set<std::st
   bool already_tiered = false;
   ret = cloud_tier_check_object(tier_ctx, already_tiered);
 
+  if (ret == -EBUSY) return ret;
   if (ret < 0) {
     ldpp_dout(tier_ctx.dpp, 0) << "ERROR: failed to check object on the cloud endpoint ret=" << ret << dendl;
   }
@@ -1618,3 +1655,8 @@ int rgw_cloud_tier_transfer_object(RGWLCCloudTierCtx& tier_ctx, std::set<std::st
 
   return ret;
 }
+
+int rgw_cloud_tier_transfer_object(RGWLCCloudTierCtx& tier_ctx, std::set<std::string>& cloud_targets) {
+  return retry_on_busy(tier_ctx.y, tier_ctx.dpp, tier_ctx.cct, __func__,
+      [&]() { return do_cloud_tier_transfer_object(tier_ctx, cloud_targets); });
+}

src/rgw/driver/rados/rgw_lc_tier.h

@@ -3,6 +3,8 @@
 
 #pragma once
 
+#include <functional>
+
 #include "rgw_lc.h"
 #include "rgw_rest_conn.h"
 #include "rgw_rados.h"
@@ -80,3 +82,6 @@ int cloud_tier_restore(const DoutPrefixProvider *dpp,
 
 bool is_restore_in_progress(const DoutPrefixProvider *dpp,
                             std::map<std::string, std::string>& headers);
+
+int retry_on_busy(optional_yield y, const DoutPrefixProvider *dpp, CephContext *cct,
+                  const char *op_name, std::function<int()> op);

src/rgw/driver/rados/rgw_rados.cc

@@ -3852,6 +3852,10 @@ public:
   }
 
   int handle_headers(const map<string, string>& headers, int http_status) override {
+    if (http_status == 503) {
+      return -EBUSY;
+    }
+
     if (src_bucket_perms && http_status != 403 && http_status != 401) {
       auto iter = headers.find("RGWX_PERM_CHECKED");
       // if the header is not present, we need to check the ACL
@@ -5649,9 +5653,11 @@ int RGWRados::restore_obj_from_cloud(RGWLCCloudTierCtx& tier_ctx,
                                 attrs, days, glacier_params, in_progress, &cb);
   } else {
     ldpp_dout(dpp, 20) << "Fetching  object:" << dest_obj << "from the cloud" <<  dendl;
-    ret = rgw_cloud_tier_get_object(tier_ctx, false,  headers,
-                                &set_mtime, etag, accounted_size,
-                                attrs, &cb);
+    ret = retry_on_busy(tier_ctx.y, dpp, cct, __func__, [&]() {
+      return rgw_cloud_tier_get_object(tier_ctx, false, headers,
+                                       &set_mtime, etag, accounted_size,
+                                       attrs, &cb);
+    });
     in_progress = false;
   }