[v24.1.x] cloud_storage: add error_outcome::operation_not_supported

src/v/cloud_storage/remote.cc

@@ -329,6 +329,8 @@ ss::future<download_result> remote::do_download_manifest(
  retry_permit.delay, fib.root_abort_source());
  retry_permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = download_result::failed;
  vlog(
@@ -427,6 +429,8 @@ ss::future<upload_result> remote::upload_manifest(
  co_await ss::sleep_abortable(permit.delay, fib.root_abort_source());
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::key_not_found:
  // not expected during upload
  [[fallthrough]];
@@ -595,6 +599,8 @@ ss::future<upload_result> remote::upload_stream(
  }
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::key_not_found:
  // not expected during upload
  [[fallthrough]];
@@ -775,6 +781,8 @@ ss::future<download_result> remote::download_stream(
  co_await ss::sleep_abortable(permit.delay, fib.root_abort_source());
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = download_result::failed;
  break;
@@ -878,6 +886,8 @@ remote::download_object(cloud_storage::download_request download_request) {
  co_await ss::sleep_abortable(permit.delay, fib.root_abort_source());
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = download_result::failed;
  break;
@@ -949,6 +959,8 @@ ss::future<download_result> remote::object_exists(
  co_await ss::sleep_abortable(permit.delay, fib.root_abort_source());
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = download_result::failed;
  break;
@@ -1065,6 +1077,8 @@ ss::future<upload_result> remote::delete_object(
  co_await ss::sleep_abortable(permit.delay, fib.root_abort_source());
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = upload_result::failed;
  break;
@@ -1224,6 +1238,8 @@ ss::future<upload_result> remote::delete_object_batch(
  co_await ss::sleep_abortable(permit.delay, _as);
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = upload_result::failed;
  break;
@@ -1436,6 +1452,8 @@ ss::future<remote::list_result> remote::list_objects(
  co_await ss::sleep_abortable(permit.delay, _as);
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:
  result = cloud_storage_clients::error_outcome::fail;
  break;
@@ -1518,6 +1536,8 @@ ss::future<upload_result> remote::upload_object(upload_request upload_request) {
  co_await ss::sleep_abortable(permit.delay, _as);
  permit = fib.retry();
  break;
+ case cloud_storage_clients::error_outcome::operation_not_supported:
+ [[fallthrough]];
  case cloud_storage_clients::error_outcome::key_not_found:
  [[fallthrough]];
  case cloud_storage_clients::error_outcome::fail:

src/v/cloud_storage_clients/abs_client.cc

@@ -541,6 +541,14 @@ ss::future<result<T, error_outcome>> abs_client::send_request(
  vlog(abs_log.debug, "BlobNotFound response received {}", key);
  outcome = error_outcome::key_not_found;
  _probe->register_failure(err.code());
+ } else if (
+ err.code() == abs_error_code::operation_not_supported_on_directory) {
+ vlog(
+ abs_log.debug,
+ "OperationNotSupportedOnDirectory response received {}",
+ key);
+ outcome = error_outcome::operation_not_supported;
+ _probe->register_failure(err.code());
  } else {
  vlog(
  abs_log.error,
@@ -731,20 +739,34 @@ abs_client::delete_object(
  }),
  key)
  .then([&name, &key](const ret_t& result) {
- // ABS returns a 404 for attempts to delete a blob that doesn't
- // exist. The remote doesn't expect this, so we map 404s to a
- // successful response.
- if (!result && result.error() == error_outcome::key_not_found) {
- vlog(
- abs_log.debug,
- "Object to be deleted was not found in cloud storage: "
- "object={}, bucket={}. Ignoring ...",
- name,
- key);
- return ss::make_ready_future<ret_t>(no_response{});
- } else {
- return ss::make_ready_future<ret_t>(result);
+ if (!result) {
+ if (result.error() == error_outcome::key_not_found) {
+ // ABS returns a 404 for attempts to delete a blob that
+ // doesn't exist. The remote doesn't expect this, so we
+ // map 404s to a successful response.
+ vlog(
+ abs_log.debug,
+ "Object to be deleted was not found in cloud storage: "
+ "object={}, bucket={}. Ignoring ...",
+ name,
+ key);
+ return ss::make_ready_future<ret_t>(no_response{});
+ } else if (
+ result.error() == error_outcome::operation_not_supported) {
+ // ABS does not allow for deletion of directories when HNS
+ // is disabled. The "folder" is "removed" when all blobs
+ // inside of it are deleted. Map this to a successful
+ // response.
+ vlog(
+ abs_log.warn,
+ "Cannot delete a directory in ABS cloud storage: "
+ "object={}, bucket={}. Ignoring ...",
+ name,
+ key);
+ return ss::make_ready_future<ret_t>(no_response{});
+ }
  }
+ return ss::make_ready_future<ret_t>(result);
  });
  } else {
  return delete_path(name, key, timeout);

src/v/cloud_storage_clients/abs_error.cc

@@ -46,6 +46,9 @@ std::istream& operator>>(std::istream& i, abs_error_code& code) {
  .match("ContainerNotFound", abs_error_code::container_not_found)
  .match("DirectoryNotEmpty", abs_error_code::directory_not_empty)
  .match("PathNotFound", abs_error_code::path_not_found)
+ .match(
+ "OperationNotSupportedOnDirectory",
+ abs_error_code::operation_not_supported_on_directory)
  .default_match(abs_error_code::_unknown);
 
  return i;

src/v/cloud_storage_clients/abs_error.h

@@ -40,7 +40,8 @@ enum class abs_error_code {
  container_being_deleted,
  container_not_found,
  directory_not_empty,
- path_not_found
+ path_not_found,
+ operation_not_supported_on_directory
 };
 
 /// Operators to use with lexical_cast

src/v/cloud_storage_clients/types.h

@@ -32,6 +32,9 @@ enum class error_outcome {
  fail,
  /// Missing key API error (only suitable for downloads and deletions)
  key_not_found,
+ /// Currently used for directory deletion errors in ABS, typically treated
+ /// as regular failure outcomes.
+ operation_not_supported
 };
 
 struct error_outcome_category final : public std::error_category {
@@ -47,6 +50,8 @@ struct error_outcome_category final : public std::error_category {
  return "Non retriable error";
  case error_outcome::key_not_found:
  return "Key not found error";
+ case error_outcome::operation_not_supported:
+ return "Operation not supported error";
  default:
  return "Undefined error_outcome encountered";
  }