Files
scylladb/test/lib/gcs_fixture.cc
2026-03-11 12:32:02 +01:00

253 lines
7.2 KiB
C++

/*
* Copyright (C) 2025-present ScyllaDB
*/
/*
* SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
*/
#include <string>
#include <memory>
#include <seastar/core/with_timeout.hh>
#include <seastar/core/reactor.hh>
#include <seastar/core/sleep.hh>
#include "gcs_fixture.hh"
#include "tmpdir.hh"
#include "proc_utils.hh"
#include "test_utils.hh"
#include "utils/gcp/gcp_credentials.hh"
#include "utils/UUID_gen.hh"
namespace fs = std::filesystem;
namespace tp = tests::proc;
using namespace utils::gcp;
using namespace tests;
using namespace std::chrono_literals;
using namespace std::string_view_literals;
static future<std::optional<google_credentials>> credentials(const std::string& source) {
try {
if (!source.empty()) {
BOOST_TEST_MESSAGE(fmt::format("Loading credentials from {}", source));
co_return co_await google_credentials::from_file(source);
}
BOOST_TEST_MESSAGE("Loading default credentials");
co_return co_await google_credentials::get_default_credentials();
} catch (...) {
BOOST_TEST_MESSAGE(fmt::format("Warning: could not load {} credentials: {}", source, std::current_exception()));
}
co_return std::nullopt; // empty, useless. will work on fake server
}
static future<std::tuple<tp::process_fixture, int>> start_fake_gcs_server(const tmpdir& tmp) {
return tp::start_docker_service("local-kms"
, "docker.io/fsouza/fake-gcs-server:1.52.3"
, {}
, [](std::string_view line) {
if (line.find("server started at") != std::string::npos) {
return tp::service_parse_state::success;
}
if (line.find("address already in use") != std::string::npos) {
return tp::service_parse_state::failed;
}
return tp::service_parse_state::cont;
}
, {}
, { "-scheme", "http", "-log-level", "debug", "--port", "4443", "-public-host", "127.0.0.1" } // image args
, 4443
);
}
class gcs_fixture::impl {
public:
std::optional<tp::process_fixture> fake_gcs_server;
std::optional<google_credentials> creds;
std::vector<std::string> objects_to_delete;
std::string endpoint;
std::string project;
std::string bucket;
std::string user_1_creds;
std::string user_2_creds;
tmpdir tmp;
bool created_bucket = false;
std::unique_ptr<storage::client> client;
std::vector<tmp_set_env> variables;
impl();
seastar::future<> setup();
seastar::future<> teardown();
};
gcs_fixture::impl::impl()
: endpoint(getenv_or_default({"GCP_STORAGE_ENDPOINT"sv, "GS_SERVER_ADDRESS_FOR_TEST"sv}))
, project(getenv_or_default("GCP_STORAGE_PROJECT"))
, bucket(getenv_or_default({"GCP_STORAGE_BUCKET"sv, "GS_BUCKET_FOR_TEST"sv}))
, user_1_creds(getenv_or_default({"GCP_STORAGE_USER_1_CREDENTIALS"sv, "GS_CREDENTIALS_FILE"sv}))
, user_2_creds(getenv_or_default("GCP_STORAGE_USER_2_CREDENTIALS"))
{}
seastar::future<> gcs_fixture::impl::setup() {
if (!bucket.empty() && endpoint.empty()) {
endpoint = storage::client::DEFAULT_ENDPOINT;
}
if (endpoint.empty()) {
auto [proc, port] = co_await start_fake_gcs_server(tmp);
fake_gcs_server.emplace(std::move(proc));
endpoint = "http://127.0.0.1:" + std::to_string(port);
BOOST_TEST_MESSAGE(fmt::format("Test server endpoint {}", endpoint));
user_1_creds = "none";
} else {
creds = co_await credentials(user_1_creds);
}
client = std::make_unique<storage::client>(endpoint, std::move(creds));
std::exception_ptr p;
try {
if (bucket.empty()) {
bucket = "test-" + fmt::format("{}", utils::UUID_gen::get_time_UUID());
co_await client->create_bucket(project, bucket);
created_bucket = true;
BOOST_TEST_MESSAGE(fmt::format("Created test bucket {}", bucket));
}
} catch (...) {
p = std::current_exception();
}
if (p) {
try {
co_await teardown();
} catch (...) {
}
std::rethrow_exception(p);
}
variables.emplace_back("GS_SERVER_ADDRESS_FOR_TEST", endpoint);
variables.emplace_back("GS_BUCKET_FOR_TEST", bucket);
variables.emplace_back("GS_CREDENTIALS_FILE", user_1_creds);
}
seastar::future<> gcs_fixture::impl::teardown() {
variables.clear();
if (client) {
for (auto& name : objects_to_delete) {
try {
co_await client->delete_object(bucket, name);
} catch (...) {
BOOST_TEST_MESSAGE(fmt::format("Warning: could not delete object: {}", name, std::current_exception()));
}
}
if (created_bucket) {
try {
auto objects = co_await client->list_objects(bucket);
for (auto& o : objects) {
co_await client->delete_object(bucket, o.name);
}
co_await client->delete_bucket(bucket);
} catch (...) {
BOOST_TEST_MESSAGE(fmt::format("Warning: could not delete bucket: {}", bucket, std::current_exception()));
}
}
}
if (fake_gcs_server) {
fake_gcs_server->terminate();
co_await fake_gcs_server->wait();
}
if (client) {
co_await client->close();
client = {};
}
}
static thread_local gcs_fixture* active_gcs_fixture = nullptr;
gcs_fixture::gcs_fixture()
: _impl(std::make_unique<impl>())
{}
gcs_fixture::~gcs_fixture() = default;
utils::gcp::storage::client& gcs_fixture::client() const {
return *_impl->client;
}
const std::string& gcs_fixture::endpoint() const {
return _impl->endpoint;
}
const std::string& gcs_fixture::project() const {
return _impl->project;
}
const std::string& gcs_fixture::bucket() const {
return _impl->bucket;
}
void gcs_fixture::add_object_to_delete(const std::string& name) {
_impl->objects_to_delete.emplace_back(name);
}
seastar::future<> gcs_fixture::setup() {
co_await _impl->setup();
active_gcs_fixture = this;
}
seastar::future<> gcs_fixture::teardown() {
active_gcs_fixture = nullptr;
return _impl->teardown();
}
gcs_fixture* gcs_fixture::active() {
return active_gcs_fixture;
}
local_gcs_wrapper::local_gcs_wrapper() = default;
local_gcs_wrapper::~local_gcs_wrapper() = default;
utils::gcp::storage::client& local_gcs_wrapper::client() const {
return gcs_fixture::active()->client();
}
seastar::future<> local_gcs_wrapper::setup() {
auto f = gcs_fixture::active();
if (!f) {
_local = std::make_unique<gcs_fixture>();
co_await _local->setup();
f = gcs_fixture::active();;
}
endpoint = f->endpoint();
project = f->project();
bucket = f->bucket();
}
seastar::future<> local_gcs_wrapper::teardown() {
auto f = gcs_fixture::active();
assert(f);
auto& c = client();
for (auto& name : objects_to_delete) {
try {
co_await c.delete_object(bucket, name);
} catch (...) {
BOOST_TEST_MESSAGE(fmt::format("Warning: could not delete object: {}", name, std::current_exception()));
}
}
if (_local) {
co_await _local->teardown();
_local = {};
}
}