|
| 1 | +/* Copyright 2023 Man Group Operations Limited |
| 2 | + * |
| 3 | + * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt. |
| 4 | + * |
| 5 | + * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0. |
| 6 | + */ |
| 7 | + |
| 8 | +#include <arcticdb/storage/s3/s3_client_interface.hpp> |
| 9 | +#include <arcticdb/log/log.hpp> |
| 10 | +#include <arcticdb/util/buffer_pool.hpp> |
| 11 | +#include <arcticdb/storage/storage_utils.hpp> |
| 12 | +#include <arcticdb/storage/s3/s3_client_wrapper.hpp> |
| 13 | + |
| 14 | +#include <aws/s3/S3Errors.h> |
| 15 | + |
| 16 | +namespace arcticdb::storage{ |
| 17 | + |
| 18 | +using namespace object_store_utils; |
| 19 | + |
| 20 | +namespace s3 { |
| 21 | + |
| 22 | +std::optional<Aws::S3::S3Error> S3ClientTestWrapper::has_failure_trigger(const std::string& bucket_name) const { |
| 23 | + bool static_failures_enabled = ConfigsMap::instance()->get_int("S3ClientTestWrapper.EnableFailures", 0) == 1; |
| 24 | + // Check if mock failures are enabled |
| 25 | + if (!static_failures_enabled) { |
| 26 | + return std::nullopt; |
| 27 | + } |
| 28 | + |
| 29 | + // Get target buckets (if not set or "all", affects all buckets) |
| 30 | + auto failure_buckets_str = ConfigsMap::instance()->get_string("S3ClientTestWrapper.FailureBucket", "all"); |
| 31 | + |
| 32 | + if (failure_buckets_str != "all") { |
| 33 | + // Split the comma-separated bucket names and check if current bucket is in the list |
| 34 | + std::istringstream bucket_stream(failure_buckets_str); |
| 35 | + std::string target_bucket; |
| 36 | + bool bucket_found = false; |
| 37 | + |
| 38 | + while (std::getline(bucket_stream, target_bucket, ',')) { |
| 39 | + // Trim whitespace |
| 40 | + target_bucket.erase(0, target_bucket.find_first_not_of(" \t")); |
| 41 | + target_bucket.erase(target_bucket.find_last_not_of(" \t") + 1); |
| 42 | + |
| 43 | + if (target_bucket == bucket_name) { |
| 44 | + bucket_found = true; |
| 45 | + break; |
| 46 | + } |
| 47 | + } |
| 48 | + |
| 49 | + if (!bucket_found) { |
| 50 | + return std::nullopt; |
| 51 | + } |
| 52 | + } |
| 53 | + |
| 54 | + // Get error configuration |
| 55 | + auto error_code = ConfigsMap::instance()->get_int("S3ClientTestWrapper.ErrorCode", static_cast<int>(Aws::S3::S3Errors::NETWORK_CONNECTION)); |
| 56 | + auto retryable = ConfigsMap::instance()->get_int("S3ClientTestWrapper.ErrorRetryable", 0) == 1; |
| 57 | + |
| 58 | + auto failure_error_ = Aws::S3::S3Error(Aws::Client::AWSError<Aws::S3::S3Errors>( |
| 59 | + static_cast<Aws::S3::S3Errors>(error_code), |
| 60 | + "SimulatedFailure", |
| 61 | + "Simulated failure from environment variables", |
| 62 | + retryable |
| 63 | + )); |
| 64 | + |
| 65 | + |
| 66 | + return failure_error_; |
| 67 | +} |
| 68 | + |
| 69 | +S3Result<std::monostate> S3ClientTestWrapper::head_object( |
| 70 | + const std::string& s3_object_name, |
| 71 | + const std::string &bucket_name) const { |
| 72 | + auto maybe_error = has_failure_trigger(bucket_name); |
| 73 | + if (maybe_error.has_value()) { |
| 74 | + return {*maybe_error}; |
| 75 | + } |
| 76 | + |
| 77 | + |
| 78 | + return actual_client_->head_object(s3_object_name, bucket_name); |
| 79 | +} |
| 80 | + |
| 81 | +S3Result<Segment> S3ClientTestWrapper::get_object( |
| 82 | + const std::string &s3_object_name, |
| 83 | + const std::string &bucket_name) const { |
| 84 | + auto maybe_error = has_failure_trigger(bucket_name); |
| 85 | + if (maybe_error.has_value()) { |
| 86 | + return {*maybe_error}; |
| 87 | + } |
| 88 | + |
| 89 | + return actual_client_->get_object(s3_object_name, bucket_name); |
| 90 | +} |
| 91 | + |
| 92 | +folly::Future<S3Result<Segment>> S3ClientTestWrapper::get_object_async( |
| 93 | + const std::string &s3_object_name, |
| 94 | + const std::string &bucket_name) const { |
| 95 | + auto maybe_error = has_failure_trigger(bucket_name); |
| 96 | + if (maybe_error.has_value()) { |
| 97 | + return folly::makeFuture<S3Result<Segment>>({*maybe_error}); |
| 98 | + } |
| 99 | + |
| 100 | + return actual_client_->get_object_async(s3_object_name, bucket_name); |
| 101 | +} |
| 102 | + |
| 103 | +S3Result<std::monostate> S3ClientTestWrapper::put_object( |
| 104 | + const std::string &s3_object_name, |
| 105 | + Segment &segment, |
| 106 | + const std::string &bucket_name, |
| 107 | + PutHeader header) { |
| 108 | + auto maybe_error = has_failure_trigger(bucket_name); |
| 109 | + if (maybe_error.has_value()) { |
| 110 | + return {*maybe_error}; |
| 111 | + } |
| 112 | + |
| 113 | + return actual_client_->put_object(s3_object_name, segment, bucket_name, header); |
| 114 | +} |
| 115 | + |
| 116 | +S3Result<DeleteOutput> S3ClientTestWrapper::delete_objects( |
| 117 | + const std::vector<std::string>& s3_object_names, |
| 118 | + const std::string& bucket_name) { |
| 119 | + auto maybe_error = has_failure_trigger(bucket_name); |
| 120 | + if (maybe_error.has_value()) { |
| 121 | + return {*maybe_error}; |
| 122 | + } |
| 123 | + |
| 124 | + |
| 125 | + return actual_client_->delete_objects(s3_object_names, bucket_name); |
| 126 | +} |
| 127 | + |
| 128 | +// Using a fixed page size since it's only being used for simple tests. |
| 129 | +// If we ever need to configure it we should move it to the s3 proto config instead. |
| 130 | +constexpr auto page_size = 10; |
| 131 | +S3Result<ListObjectsOutput> S3ClientTestWrapper::list_objects( |
| 132 | + const std::string& name_prefix, |
| 133 | + const std::string& bucket_name, |
| 134 | + const std::optional<std::string>& continuation_token) const { |
| 135 | + auto maybe_error = has_failure_trigger(bucket_name); |
| 136 | + if (maybe_error.has_value()) { |
| 137 | + return {*maybe_error}; |
| 138 | + } |
| 139 | + |
| 140 | + return actual_client_->list_objects(name_prefix, bucket_name, continuation_token); |
| 141 | +} |
| 142 | + |
| 143 | +} |
| 144 | + |
| 145 | +} |
0 commit comments