PS-11001: Enable purge_binlogs on the s3 storage backend

mtr/binlog_streaming/include/set_up_binsrv_environment.inc

@@ -42,7 +42,7 @@ if ($storage_backend == s3)
   --let $aws_s3_bucket = $MTR_BINSRV_AWS_S3_BUCKET
   if ($MTR_BINSRV_AWS_S3_ENDPOINT != '')
   {
-    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/$MTR_BINSRV_AWS_S3_BUCKET', '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', REPLACE('$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '/', '%2F'), '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/', '$MTR_BINSRV_AWS_S3_BUCKET', '$binsrv_storage_path');
   }
   if ($MTR_BINSRV_AWS_S3_ENDPOINT == '')
   {
@@ -51,7 +51,7 @@ if ($storage_backend == s3)
     {
       --let $qualified_bucket = $qualified_bucket.$MTR_BINSRV_AWS_S3_REGION
     }
-    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$qualified_bucket',  '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', REPLACE('$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '/', '%2F'), '@', '$qualified_bucket', '$binsrv_storage_path');
   }
 }
 

mtr/binlog_streaming/t/purge_binlogs.test

@@ -2,15 +2,8 @@
 
 --source ../include/v80_v84_compatibility_defines.inc
 
-# identifying backend storage type ('file' or 's3') early - 'purge_binlogs'
-# is supported only on the local filesystem backend
 --source ../include/identify_storage_backend.inc
 
-if ($storage_backend == s3)
-{
-  --skip purge_binlogs is supported only on the local filesystem backend
-}
-
 # in case of --repeat=N, we need to start from a fresh binary log to make
 # this test deterministic
 --echo *** Resetting replication at the very beginning of the test.

src/app.cpp

@@ -1058,13 +1058,6 @@ bool handle_purge_binlogs(std::string_view config_file_path,
     const auto &replication_config = config.root().get<"replication">();
     const auto replication_mode{replication_config.get<"mode">()};
 
-    // for now, only file backend supported
-    if (storage_config.get<"backend">() != binsrv::storage_backend_type::file) {
-      throw std::runtime_error(
-          "purge_binlogs is only supported on the local filesystem storage "
-          "backend");
-    }
-
     binsrv::storage storage{storage_config,
                             binsrv::storage_construction_mode_type::purging,
                             replication_mode};

src/binsrv/s3_storage_backend.cpp

@@ -53,6 +53,7 @@
 #include <aws/s3-crt/S3CrtClient.h>
 
 #include <aws/s3-crt/model/BucketLocationConstraint.h>
+#include <aws/s3-crt/model/DeleteObjectRequest.h>
 #include <aws/s3-crt/model/GetBucketLocationRequest.h>
 #include <aws/s3-crt/model/GetBucketLocationResult.h>
 #include <aws/s3-crt/model/GetObjectRequest.h>
@@ -171,6 +172,8 @@ class s3_storage_backend::aws_context : private aws_context_base {
   void put_object_from_span(const qualified_object_path &dest,
                             util::const_byte_span content) const;
 
+  void delete_object(const qualified_object_path &target) const;
+
   [[nodiscard]] storage_object_name_container
   list_objects(const qualified_object_path &prefix);
 
@@ -346,6 +349,23 @@ void s3_storage_backend::aws_context::put_object_from_span(
   put_object_from_stream(dest, content_stream);
 }
 
+// TODO: Consider deleting several objects in one request by using
+// `DeleteObjects()` AWS SDK C++ API call
+void s3_storage_backend::aws_context::delete_object(
+    const qualified_object_path &target) const {
+  Aws::S3Crt::Model::DeleteObjectRequest delete_object_request;
+  delete_object_request.SetBucket(target.bucket);
+  delete_object_request.SetKey(target.object_path.generic_string());
+
+  const auto delete_object_outcome{
+      client_->DeleteObject(delete_object_request)};
+
+  if (!delete_object_outcome.IsSuccess()) {
+    raise_s3_error_from_outcome("cannot delete object from S3 bucket",
+                                delete_object_outcome.GetError());
+  }
+}
+
 [[nodiscard]] storage_object_name_container
 s3_storage_backend::aws_context::list_objects(
     const qualified_object_path &prefix) {
@@ -679,10 +699,9 @@ void s3_storage_backend::do_put_object(std::string_view name,
       {.bucket = bucket_, .object_path = get_object_path(name)}, content);
 }
 
-void s3_storage_backend::do_remove_object(
-    [[maybe_unused]] std::string_view name) {
-  util::exception_location().raise<std::logic_error>(
-      "remove_object is not supported on the S3 storage backend");
+void s3_storage_backend::do_remove_object(std::string_view name) {
+  impl_->delete_object(
+      {.bucket = bucket_, .object_path = get_object_path(name)});
 }
 
 void s3_storage_backend::do_fsync() {

src/binsrv/storage.cpp

@@ -33,7 +33,6 @@
 #include "binsrv/binlog_file_metadata.hpp"
 #include "binsrv/replication_mode_type.hpp"
 #include "binsrv/storage_backend_factory.hpp"
-#include "binsrv/storage_backend_type.hpp"
 #include "binsrv/storage_config.hpp"
 #include "binsrv/storage_metadata.hpp"
 
@@ -54,13 +53,6 @@ storage::storage(const storage_config &config,
                  replication_mode_type replication_mode)
     : construction_mode_{construction_mode}, backend_{},
       replication_mode_{replication_mode} {
-  if (construction_mode_ == storage_construction_mode_type::purging &&
-      config.get<"backend">() != storage_backend_type::file) {
-    util::exception_location().raise<std::runtime_error>(
-        "purge_binlogs is only supported on the local filesystem storage "
-        "backend");
-  }
-
   const auto &checkpoint_size_opt{config.get<"checkpoint_size">()};
   if (checkpoint_size_opt.has_value()) {
     checkpoint_size_bytes_ = checkpoint_size_opt->get_value();