This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch branch-4.0-preview
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 717c6ce1ffbc1355e402e125dc1885958874088b
Author: huanghg1994 <519500...@qq.com>
AuthorDate: Thu Apr 25 20:05:03 2024 +0800
[fix](resource)fix check available fail when s3 aws_token is set and reset
as, sk faild on be. (#34057)
---
be/src/agent/task_worker_pool.cpp | 2 ++
be/src/io/fs/s3_file_system.cpp | 12 ++++++++----
.../main/java/org/apache/doris/catalog/S3Resource.java | 16 +++++++++++++---
.../datasource/property/constants/S3Properties.java | 1 +
gensrc/thrift/AgentService.thrift | 1 +
.../suites/cold_heat_separation/policy/alter.groovy | 11 +++++++++++
6 files changed, 36 insertions(+), 7 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 862c357c34a..07ef10ed333 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -1365,6 +1365,7 @@ void update_s3_resource(const TStorageResource& param,
io::RemoteFileSystemSPtr
.region = param.s3_storage_param.region,
.ak = param.s3_storage_param.ak,
.sk = param.s3_storage_param.sk,
+ .token = param.s3_storage_param.token,
.max_connections = param.s3_storage_param.max_conn,
.request_timeout_ms =
param.s3_storage_param.request_timeout_ms,
.connect_timeout_ms =
param.s3_storage_param.conn_timeout_ms,
@@ -1384,6 +1385,7 @@ void update_s3_resource(const TStorageResource& param,
io::RemoteFileSystemSPtr
S3ClientConf conf {
.ak = param.s3_storage_param.ak,
.sk = param.s3_storage_param.sk,
+ .token = param.s3_storage_param.token,
};
st = client->reset(conf);
fs = std::move(existed_fs);
diff --git a/be/src/io/fs/s3_file_system.cpp b/be/src/io/fs/s3_file_system.cpp
index 3d9f25686a8..dea32793509 100644
--- a/be/src/io/fs/s3_file_system.cpp
+++ b/be/src/io/fs/s3_file_system.cpp
@@ -115,16 +115,21 @@ Status S3ClientHolder::init() {
}
Status S3ClientHolder::reset(const S3ClientConf& conf) {
+ S3ClientConf reset_conf;
{
std::shared_lock lock(_mtx);
- if (conf.ak == _conf.ak && conf.sk == _conf.sk) {
+ if (conf.ak == _conf.ak && conf.sk == _conf.sk && conf.token ==
_conf.token) {
return Status::OK(); // Same conf
}
+ reset_conf = _conf;
+ reset_conf.ak = conf.ak;
+ reset_conf.sk = conf.sk;
+ reset_conf.token = conf.token;
// Should check endpoint here?
}
- auto client = S3ClientFactory::instance().create(conf);
+ auto client = S3ClientFactory::instance().create(reset_conf);
if (!client) {
return Status::InternalError("failed to init s3 client with conf {}",
conf.to_string());
}
@@ -134,8 +139,7 @@ Status S3ClientHolder::reset(const S3ClientConf& conf) {
{
std::lock_guard lock(_mtx);
_client = std::move(client);
- _conf.ak = conf.ak;
- _conf.sk = conf.sk;
+ _conf = std::move(reset_conf);
}
return Status::OK();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/S3Resource.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/S3Resource.java
index 5bcb5123c64..a2603897047 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/S3Resource.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/S3Resource.java
@@ -105,7 +105,8 @@ public class S3Resource extends Resource {
properties.putIfAbsent(S3Properties.REGION, region);
String ak = properties.get(S3Properties.ACCESS_KEY);
String sk = properties.get(S3Properties.SECRET_KEY);
- CloudCredentialWithEndpoint credential = new
CloudCredentialWithEndpoint(pingEndpoint, region, ak, sk);
+ String token = properties.get(S3Properties.SESSION_TOKEN);
+ CloudCredentialWithEndpoint credential = new
CloudCredentialWithEndpoint(pingEndpoint, region, ak, sk, token);
if (needCheck) {
String bucketName = properties.get(S3Properties.BUCKET);
@@ -123,6 +124,7 @@ public class S3Resource extends Resource {
Map<String, String> propertiesPing = new HashMap<>();
propertiesPing.put(S3Properties.Env.ACCESS_KEY,
credential.getAccessKey());
propertiesPing.put(S3Properties.Env.SECRET_KEY,
credential.getSecretKey());
+ propertiesPing.put(S3Properties.Env.TOKEN,
credential.getSessionToken());
propertiesPing.put(S3Properties.Env.ENDPOINT,
credential.getEndpoint());
propertiesPing.put(S3Properties.Env.REGION, credential.getRegion());
propertiesPing.put(PropertyConverter.USE_PATH_STYLE,
@@ -188,6 +190,10 @@ public class S3Resource extends Resource {
writeLock();
for (Map.Entry<String, String> kv : properties.entrySet()) {
replaceIfEffectiveValue(this.properties, kv.getKey(),
kv.getValue());
+ if (kv.getKey().equals(S3Properties.Env.TOKEN)
+ || kv.getKey().equals(S3Properties.SESSION_TOKEN)) {
+ this.properties.put(kv.getKey(), kv.getValue());
+ }
}
++version;
writeUnlock();
@@ -197,11 +203,13 @@ public class S3Resource extends Resource {
private CloudCredentialWithEndpoint getS3PingCredentials(Map<String,
String> properties) {
String ak = properties.getOrDefault(S3Properties.ACCESS_KEY,
this.properties.get(S3Properties.ACCESS_KEY));
String sk = properties.getOrDefault(S3Properties.SECRET_KEY,
this.properties.get(S3Properties.SECRET_KEY));
+ String token = properties.getOrDefault(S3Properties.SESSION_TOKEN,
+ this.properties.get(S3Properties.SESSION_TOKEN));
String endpoint = properties.getOrDefault(S3Properties.ENDPOINT,
this.properties.get(S3Properties.ENDPOINT));
String pingEndpoint = "http://"; + endpoint;
String region = S3Properties.getRegionOfEndpoint(pingEndpoint);
properties.putIfAbsent(S3Properties.REGION, region);
- return new CloudCredentialWithEndpoint(pingEndpoint, region, ak, sk);
+ return new CloudCredentialWithEndpoint(pingEndpoint, region, ak, sk,
token);
}
private boolean isNeedCheck(Map<String, String> newProperties) {
@@ -231,7 +239,9 @@ public class S3Resource extends Resource {
// it's dangerous to show password in show odbc resource,
// so we use empty string to replace the real password
if (entry.getKey().equals(S3Properties.Env.SECRET_KEY)
- || entry.getKey().equals(S3Properties.SECRET_KEY)) {
+ || entry.getKey().equals(S3Properties.SECRET_KEY)
+ || entry.getKey().equals(S3Properties.Env.TOKEN)
+ || entry.getKey().equals(S3Properties.SESSION_TOKEN)) {
result.addRow(Lists.newArrayList(name, lowerCaseType,
entry.getKey(), "******"));
} else {
result.addRow(Lists.newArrayList(name, lowerCaseType,
entry.getKey(), entry.getValue()));
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/S3Properties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/S3Properties.java
index 947174f86ef..26bd37b0cac 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/S3Properties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/S3Properties.java
@@ -265,6 +265,7 @@ public class S3Properties extends BaseProperties {
s3Info.setRegion(properties.get(S3Properties.REGION));
s3Info.setAk(properties.get(S3Properties.ACCESS_KEY));
s3Info.setSk(properties.get(S3Properties.SECRET_KEY));
+ s3Info.setToken(properties.get(S3Properties.SESSION_TOKEN));
s3Info.setRootPath(properties.get(S3Properties.ROOT_PATH));
s3Info.setBucket(properties.get(S3Properties.BUCKET));
diff --git a/gensrc/thrift/AgentService.thrift
b/gensrc/thrift/AgentService.thrift
index df478a75a1b..4e381b16805 100644
--- a/gensrc/thrift/AgentService.thrift
+++ b/gensrc/thrift/AgentService.thrift
@@ -73,6 +73,7 @@ struct TS3StorageParam {
8: optional string root_path
9: optional string bucket
10: optional bool use_path_style = false
+ 11: optional string token
}
struct TStoragePolicy {
diff --git a/regression-test/suites/cold_heat_separation/policy/alter.groovy
b/regression-test/suites/cold_heat_separation/policy/alter.groovy
index 672f7942763..cfd8e0da6fc 100644
--- a/regression-test/suites/cold_heat_separation/policy/alter.groovy
+++ b/regression-test/suites/cold_heat_separation/policy/alter.groovy
@@ -39,6 +39,7 @@ suite("alter_policy") {
"AWS_ROOT_PATH" = "path/to/rootaaaa",
"AWS_ACCESS_KEY" = "bbba",
"AWS_SECRET_KEY" = "aaaa",
+ "AWS_TOKEN" = "session_token",
"AWS_MAX_CONNECTIONS" = "50",
"AWS_REQUEST_TIMEOUT_MS" = "3000",
"AWS_CONNECTION_TIMEOUT_MS" = "1000",
@@ -70,6 +71,10 @@ suite("alter_policy") {
ALTER RESOURCE "${resource_name}"
PROPERTIES("AWS_REQUEST_TIMEOUT_MS" = "7777");
"""
+ def alter_result_succ_8 = try_sql """
+ ALTER RESOURCE "${resource_name}" PROPERTIES("AWS_TOKEN" =
"new_session_token");
+ """
+
// errCode = 2, detailMessage = current not support modify property :
AWS_REGION
def alter_result_fail_1 = try_sql """
ALTER RESOURCE "${resource_name}" PROPERTIES("AWS_REGION" =
"8888");
@@ -112,6 +117,7 @@ suite("alter_policy") {
// [has_resouce_policy_alter, s3, AWS_REQUEST_TIMEOUT_MS, 7777],
// [has_resouce_policy_alter, s3, AWS_ROOT_PATH, path/to/rootaaaa],
// [has_resouce_policy_alter, s3, AWS_SECRET_KEY, ******],
+ // [has_resouce_policy_alter, s3, AWS_TOKEN, ******],
// [has_resouce_policy_alter, s3, id, {id}],
// [has_resouce_policy_alter, s3, type, s3]
// [has_resouce_policy_alter, s3, version, {version}]]
@@ -133,6 +139,8 @@ suite("alter_policy") {
assertEquals(show_alter_result[8][3], "10101010")
// AWS_SECRET_KEY
assertEquals(show_alter_result[9][3], "******")
+ // AWS_SECRET_KEY
+ assertEquals(show_alter_result[10][3], "******")
}
def check_alter_resource_result_with_policy = { resource_name ->
@@ -151,6 +159,7 @@ suite("alter_policy") {
// [has_resouce_policy_alter, s3, AWS_REQUEST_TIMEOUT_MS, 7777],
// [has_resouce_policy_alter, s3, AWS_ROOT_PATH, path/to/rootaaaa],
// [has_resouce_policy_alter, s3, AWS_SECRET_KEY, ******],
+ // [has_resouce_policy_alter, s3, AWS_TOKEN, ******],
// [has_resouce_policy_alter, s3, id, {id}],
// [has_resouce_policy_alter, s3, type, s3]
// [has_resouce_policy_alter, s3, version, {version}]]
@@ -172,6 +181,8 @@ suite("alter_policy") {
assertEquals(show_alter_result[8][3], "path/to/rootaaaa")
// AWS_SECRET_KEY
assertEquals(show_alter_result[9][3], "******")
+ // AWS_SECRET_KEY
+ assertEquals(show_alter_result[10][3], "******")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
