Skip to content

[fix](recycler) Fix two errors for recycler #53042

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dataroaring merged 1 commit into from
Jul 14, 2025
Merged

[fix](recycler) Fix two errors for recycler #53042

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion cloud/src/recycler/azure_obj_client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,14 +137,19 @@ class AzureListIterator final : public ObjectListIterator {
SystemClockEpoch)
.count()});
}
} catch (Azure::Storage::StorageException& e) {
} catch (Azure::Core::RequestFailedException& e) {
LOG_WARNING(
"Azure request failed because {}, http_code: {}, request_id: {}, url: {}, "
"prefix: {}",
e.Message, static_cast<int>(e.StatusCode), e.RequestId, client_->GetUrl(),
req_.Prefix.Value());
is_valid_ = false;
return false;
} catch (std::exception& e) {
LOG_WARNING("Azure request failed because {}, url: {}, prefix: {}", e.what(),
client_->GetUrl(), req_.Prefix.Value());
is_valid_ = false;
return false;
}

return !results_.empty();
Expand Down
1 change: 1 addition & 0 deletions cloud/src/recycler/recycler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,7 @@ void Recycler::check_recycle_tasks() {
int Recycler::start(brpc::Server* server) {
instance_filter_.reset(config::recycle_whitelist, config::recycle_blacklist);
g_bvar_recycler_task_max_concurrency.set_value(config::recycle_concurrency);
S3Environment::getInstance();

if (config::enable_checker) {
checker_ = std::make_unique<Checker>(txn_kv_);
Expand Down
83 changes: 42 additions & 41 deletions cloud/src/recycler/s3_accessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,50 +111,52 @@ int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_bur
return AccessorRateLimiter::instance().rate_limiter(type)->reset(max_speed, max_burst, limit);
}

class S3Environment {
public:
S3Environment() {
aws_options_ = Aws::SDKOptions {};
auto logLevel = static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);
aws_options_.loggingOptions.logLevel = logLevel;
aws_options_.loggingOptions.logger_create_fn = [logLevel] {
return std::make_shared<DorisAWSLogger>(logLevel);
};
Aws::InitAPI(aws_options_);
S3Environment::S3Environment() {
LOG(INFO) << "Initializing S3 environment";
aws_options_ = Aws::SDKOptions {};
auto logLevel = static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);
aws_options_.loggingOptions.logLevel = logLevel;
aws_options_.loggingOptions.logger_create_fn = [logLevel] {
return std::make_shared<DorisAWSLogger>(logLevel);
};
Aws::InitAPI(aws_options_);

#ifdef USE_AZURE
auto azureLogLevel =
static_cast<Azure::Core::Diagnostics::Logger::Level>(config::azure_log_level);
Azure::Core::Diagnostics::Logger::SetLevel(azureLogLevel);
Azure::Core::Diagnostics::Logger::SetListener(
[&](Azure::Core::Diagnostics::Logger::Level level, const std::string& message) {
switch (level) {
case Azure::Core::Diagnostics::Logger::Level::Verbose:
LOG(INFO) << message;
break;
case Azure::Core::Diagnostics::Logger::Level::Informational:
LOG(INFO) << message;
break;
case Azure::Core::Diagnostics::Logger::Level::Warning:
LOG(WARNING) << message;
break;
case Azure::Core::Diagnostics::Logger::Level::Error:
LOG(ERROR) << message;
break;
default:
LOG(WARNING) << "Unknown level: " << static_cast<int>(level)
<< ", message: " << message;
break;
}
});
auto azureLogLevel =
static_cast<Azure::Core::Diagnostics::Logger::Level>(config::azure_log_level);
Azure::Core::Diagnostics::Logger::SetLevel(azureLogLevel);
Azure::Core::Diagnostics::Logger::SetListener(
[&](Azure::Core::Diagnostics::Logger::Level level, const std::string& message) {
switch (level) {
case Azure::Core::Diagnostics::Logger::Level::Verbose:
LOG(INFO) << message;
break;
case Azure::Core::Diagnostics::Logger::Level::Informational:
LOG(INFO) << message;
break;
case Azure::Core::Diagnostics::Logger::Level::Warning:
LOG(WARNING) << message;
break;
case Azure::Core::Diagnostics::Logger::Level::Error:
LOG(ERROR) << message;
break;
default:
LOG(WARNING) << "Unknown level: " << static_cast<int>(level)
<< ", message: " << message;
break;
}
});
#endif
}
}

~S3Environment() { Aws::ShutdownAPI(aws_options_); }
S3Environment& S3Environment::getInstance() {
static S3Environment instance;
return instance;
}

private:
Aws::SDKOptions aws_options_;
};
S3Environment::~S3Environment() {
Aws::ShutdownAPI(aws_options_);
}

class S3ListIterator final : public ListIterator {
public:
Expand Down Expand Up @@ -316,6 +318,7 @@ int S3Accessor::init() {
std::make_shared<SimpleThreadPool>(config::recycle_pool_parallelism, "s3_accessor");
worker_pool->start();
});
S3Environment::getInstance();
Copy link
Contributor

@gavinchou gavinchou Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this call seems confusing and meaningless

Copy link
Contributor Author

@swjtu-zhanglei swjtu-zhanglei Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this call seems confusing and meaningless

image just a refact, don't change the original logicality

switch (conf_.provider) {
case S3Conf::AZURE: {
#ifdef USE_AZURE
Expand Down Expand Up @@ -355,8 +358,6 @@ int S3Accessor::init() {
uri_ = conf_.endpoint + '/' + conf_.bucket + '/' + conf_.prefix;
}

static S3Environment s3_env;

// S3Conf::S3
Aws::Client::ClientConfiguration aws_config;
aws_config.endpointOverride = conf_.endpoint;
Expand Down
13 changes: 13 additions & 0 deletions cloud/src/recycler/s3_accessor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,19 @@ extern bvar::LatencyRecorder s3_get_bucket_version_latency;
extern bvar::LatencyRecorder s3_copy_object_latency;
}; // namespace s3_bvar

class S3Environment {
public:
S3Environment(const S3Environment&) = delete;
S3Environment& operator=(const S3Environment&) = delete;

static S3Environment& getInstance();

~S3Environment();

private:
S3Environment();
Aws::SDKOptions aws_options_;
};
struct AccessorRateLimiter {
public:
~AccessorRateLimiter() = default;
Expand Down
2 changes: 1 addition & 1 deletion cloud/test/s3_accessor_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ int main(int argc, char** argv) {
std::cerr << "failed to init glog" << std::endl;
return -1;
}
doris::cloud::config::aws_log_level = 5;
LOG(INFO) << "s3_accessor_test starting";
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
Expand Down
Loading