thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-19 16:20:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Revert "Fix AWS ECS""

Browse Source
This commit is contained in:
Kruglov Pavel 2024-06-17 23:43:03 +02:00 committed by GitHub
parent 856f5c395a
commit 4817657375
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 61 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
contrib/aws-crt-cpp vendored

@ -1 +1 @@
Subproject commit f532d6abc0d2b0d8b5d6fe9e7c51eaedbe4afbd0
Subproject commit 0217761556a7ba7ec537fe933d0ab1159096746e

35
src/Common/ProxyConfigurationResolverProvider.cpp
View File

@ -112,9 +112,8 @@ namespace
return configuration.has(config_prefix + ".uri");
}
/*
* New syntax requires protocol prefix "<http> or <https>"
* */
/* New syntax requires protocol prefix "<http> or <https>"
*/
std::optional<std::string> getProtocolPrefix(
ProxyConfiguration::Protocol request_protocol,
const String & config_prefix,
@ -130,22 +129,18 @@ namespace
return protocol_prefix;
}
template <bool new_syntax>
std::optional<std::string> calculatePrefixBasedOnSettingsSyntax(
bool new_syntax,
ProxyConfiguration::Protocol request_protocol,
const String & config_prefix,
const Poco::Util::AbstractConfiguration & configuration
)
{
if (!configuration.has(config_prefix))
{
return std::nullopt;
}
if constexpr (new_syntax)
{
if (new_syntax)
return getProtocolPrefix(request_protocol, config_prefix, configuration);
}
return config_prefix;
}
@ -155,24 +150,21 @@ std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::
Protocol request_protocol,
const Poco::Util::AbstractConfiguration & configuration)
{
if (auto resolver = getFromSettings(request_protocol, "proxy", configuration))
{
if (auto resolver = getFromSettings(true, request_protocol, "proxy", configuration))
return resolver;
}
return std::make_shared<EnvironmentProxyConfigurationResolver>(
request_protocol,
isTunnelingDisabledForHTTPSRequestsOverHTTPProxy(configuration));
}
template <bool is_new_syntax>
std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::getFromSettings(
bool new_syntax,
Protocol request_protocol,
const String & config_prefix,
const Poco::Util::AbstractConfiguration & configuration
)
const Poco::Util::AbstractConfiguration & configuration)
{
auto prefix_opt = calculatePrefixBasedOnSettingsSyntax<is_new_syntax>(request_protocol, config_prefix, configuration);
auto prefix_opt = calculatePrefixBasedOnSettingsSyntax(new_syntax, request_protocol, config_prefix, configuration);
if (!prefix_opt)
{
@ -195,20 +187,17 @@ std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::
std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::getFromOldSettingsFormat(
Protocol request_protocol,
const String & config_prefix,
const Poco::Util::AbstractConfiguration & configuration
)
const Poco::Util::AbstractConfiguration & configuration)
{
/*
* First try to get it from settings only using the combination of config_prefix and configuration.
/* First try to get it from settings only using the combination of config_prefix and configuration.
* This logic exists for backward compatibility with old S3 storage specific proxy configuration.
* */
if (auto resolver = ProxyConfigurationResolverProvider::getFromSettings<false>(request_protocol, config_prefix + ".proxy", configuration))
if (auto resolver = ProxyConfigurationResolverProvider::getFromSettings(false, request_protocol, config_prefix + ".proxy", configuration))
{
return resolver;
}
/*
* In case the combination of config_prefix and configuration does not provide a resolver, try to get it from general / new settings.
/* In case the combination of config_prefix and configuration does not provide a resolver, try to get it from general / new settings.
* Falls back to Environment resolver if no configuration is found.
* */
return ProxyConfigurationResolverProvider::get(request_protocol, configuration);

5
src/Common/ProxyConfigurationResolverProvider.h
View File

@ -33,12 +33,11 @@ public:
);
private:
template <bool is_new_syntax = true>
static std::shared_ptr<ProxyConfigurationResolver> getFromSettings(
bool is_new_syntax,
Protocol protocol,
const String & config_prefix,
const Poco::Util::AbstractConfiguration & configuration
);
const Poco::Util::AbstractConfiguration & configuration);
};
}

2
src/IO/ReadWriteBufferFromHTTP.cpp
View File

@ -221,7 +221,7 @@ ReadWriteBufferFromHTTP::ReadWriteBufferFromHTTP(
if (iter == http_header_entries.end())
{
http_header_entries.emplace_back(user_agent, fmt::format("ClickHouse/{}", VERSION_STRING));
http_header_entries.emplace_back(user_agent, fmt::format("ClickHouse/{}{}", VERSION_STRING, VERSION_OFFICIAL));
}
if (!delay_initialization && use_external_buffer)

6
src/IO/S3/Client.cpp
View File

@ -972,10 +972,10 @@ PocoHTTPClientConfiguration ClientFactory::createClientConfiguration( // NOLINT
{
auto context = Context::getGlobalContextInstance();
chassert(context);
auto proxy_configuration_resolver = DB::ProxyConfigurationResolverProvider::get(DB::ProxyConfiguration::protocolFromString(protocol), context->getConfigRef());
auto proxy_configuration_resolver = ProxyConfigurationResolverProvider::get(ProxyConfiguration::protocolFromString(protocol), context->getConfigRef());
auto per_request_configuration = [=] () { return proxy_configuration_resolver->resolve(); };
auto error_report = [=] (const DB::ProxyConfiguration & req) { proxy_configuration_resolver->errorReport(req); };
auto per_request_configuration = [=]{ return proxy_configuration_resolver->resolve(); };
auto error_report = [=](const ProxyConfiguration & req) { proxy_configuration_resolver->errorReport(req); };
auto config = PocoHTTPClientConfiguration(
per_request_configuration,

20
src/IO/S3/PocoHTTPClient.cpp
View File

@ -1,4 +1,5 @@
#include <Poco/Timespan.h>
#include <Common/config_version.h>
#include "config.h"
#if USE_AWS_S3
@ -17,6 +18,7 @@
#include <IO/WriteBufferFromString.h>
#include <IO/Operators.h>
#include <IO/S3/ProviderType.h>
#include <Interpreters/Context.h>
#include <aws/core/http/HttpRequest.h>
#include <aws/core/http/HttpResponse.h>
@ -29,6 +31,7 @@
#include <boost/algorithm/string.hpp>
static const int SUCCESS_RESPONSE_MIN = 200;
static const int SUCCESS_RESPONSE_MAX = 299;
@ -84,7 +87,7 @@ namespace DB::S3
{
PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
std::function<DB::ProxyConfiguration()> per_request_configuration_,
std::function<ProxyConfiguration()> per_request_configuration_,
const String & force_region_,
const RemoteHostFilter & remote_host_filter_,
unsigned int s3_max_redirects_,
@ -94,7 +97,7 @@ PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
bool s3_use_adaptive_timeouts_,
const ThrottlerPtr & get_request_throttler_,
const ThrottlerPtr & put_request_throttler_,
std::function<void(const DB::ProxyConfiguration &)> error_report_)
std::function<void(const ProxyConfiguration &)> error_report_)
: per_request_configuration(per_request_configuration_)
, force_region(force_region_)
, remote_host_filter(remote_host_filter_)
@ -107,6 +110,8 @@ PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
, s3_use_adaptive_timeouts(s3_use_adaptive_timeouts_)
, error_report(error_report_)
{
/// This is used to identify configurations created by us.
userAgent = std::string(VERSION_FULL) + VERSION_OFFICIAL;
}
void PocoHTTPClientConfiguration::updateSchemeAndRegion()
@ -166,6 +171,17 @@ PocoHTTPClient::PocoHTTPClient(const PocoHTTPClientConfiguration & client_config
{
}
PocoHTTPClient::PocoHTTPClient(const Aws::Client::ClientConfiguration & client_configuration)
: timeouts(ConnectionTimeouts()
.withConnectionTimeout(Poco::Timespan(client_configuration.connectTimeoutMs * 1000))
.withSendTimeout(Poco::Timespan(client_configuration.requestTimeoutMs * 1000))
.withReceiveTimeout(Poco::Timespan(client_configuration.requestTimeoutMs * 1000))
.withTCPKeepAliveTimeout(Poco::Timespan(
client_configuration.enableTcpKeepAlive ? client_configuration.tcpKeepAliveIntervalMs * 1000 : 0))),
remote_host_filter(Context::getGlobalContextInstance()->getRemoteHostFilter())
{
}
std::shared_ptr<Aws::Http::HttpResponse> PocoHTTPClient::MakeRequest(
const std::shared_ptr<Aws::Http::HttpRequest> & request,
Aws::Utils::RateLimits::RateLimiterInterface * readLimiter,

14
src/IO/S3/PocoHTTPClient.h
View File

@ -38,7 +38,7 @@ class PocoHTTPClient;
struct PocoHTTPClientConfiguration : public Aws::Client::ClientConfiguration
{
std::function<DB::ProxyConfiguration()> per_request_configuration;
std::function<ProxyConfiguration()> per_request_configuration;
String force_region;
const RemoteHostFilter & remote_host_filter;
unsigned int s3_max_redirects;
@ -54,13 +54,13 @@ struct PocoHTTPClientConfiguration : public Aws::Client::ClientConfiguration
size_t http_keep_alive_timeout = DEFAULT_HTTP_KEEP_ALIVE_TIMEOUT;
size_t http_keep_alive_max_requests = DEFAULT_HTTP_KEEP_ALIVE_MAX_REQUEST;
std::function<void(const DB::ProxyConfiguration &)> error_report;
std::function<void(const ProxyConfiguration &)> error_report;
void updateSchemeAndRegion();
private:
PocoHTTPClientConfiguration(
std::function<DB::ProxyConfiguration()> per_request_configuration_,
std::function<ProxyConfiguration()> per_request_configuration_,
const String & force_region_,
const RemoteHostFilter & remote_host_filter_,
unsigned int s3_max_redirects_,
@ -70,8 +70,7 @@ private:
bool s3_use_adaptive_timeouts_,
const ThrottlerPtr & get_request_throttler_,
const ThrottlerPtr & put_request_throttler_,
std::function<void(const DB::ProxyConfiguration &)> error_report_
);
std::function<void(const ProxyConfiguration &)> error_report_);
/// Constructor of Aws::Client::ClientConfiguration must be called after AWS SDK initialization.
friend ClientFactory;
@ -120,6 +119,7 @@ class PocoHTTPClient : public Aws::Http::HttpClient
{
public:
explicit PocoHTTPClient(const PocoHTTPClientConfiguration & client_configuration);
explicit PocoHTTPClient(const Aws::Client::ClientConfiguration & client_configuration);
~PocoHTTPClient() override = default;
std::shared_ptr<Aws::Http::HttpResponse> MakeRequest(
@ -166,8 +166,8 @@ protected:
static S3MetricKind getMetricKind(const Aws::Http::HttpRequest & request);
void addMetric(const Aws::Http::HttpRequest & request, S3MetricType type, ProfileEvents::Count amount = 1) const;
std::function<DB::ProxyConfiguration()> per_request_configuration;
std::function<void(const DB::ProxyConfiguration &)> error_report;
std::function<ProxyConfiguration()> per_request_configuration;
std::function<void(const ProxyConfiguration &)> error_report;
ConnectionTimeouts timeouts;
const RemoteHostFilter & remote_host_filter;
unsigned int s3_max_redirects;

5
src/IO/S3/PocoHTTPClientFactory.cpp
View File

@ -15,7 +15,10 @@ namespace DB::S3
std::shared_ptr<Aws::Http::HttpClient>
PocoHTTPClientFactory::CreateHttpClient(const Aws::Client::ClientConfiguration & client_configuration) const
{
return std::make_shared<PocoHTTPClient>(static_cast<const PocoHTTPClientConfiguration &>(client_configuration));
if (client_configuration.userAgent.starts_with("ClickHouse"))
return std::make_shared<PocoHTTPClient>(static_cast<const PocoHTTPClientConfiguration &>(client_configuration));
else /// This client is created inside the AWS SDK with default settings to obtain ECS credentials from localhost.
return std::make_shared<PocoHTTPClient>(client_configuration);
}
std::shared_ptr<Aws::Http::HttpRequest> PocoHTTPClientFactory::CreateHttpRequest(

4
tests/queries/0_stateless/03170_ecs_crash.reference Normal file
View File

@ -0,0 +1,4 @@
1 2 3
4 5 6
7 8 9
0 0 0

9
tests/queries/0_stateless/03170_ecs_crash.sh Executable file
View File

@ -0,0 +1,9 @@
#!/usr/bin/env bash
# Tags: no-fasttest
CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CUR_DIR"/../shell_config.sh
# Previous versions crashed in attempt to use this authentication method (regardless of whether it was able to authenticate):
AWS_CONTAINER_CREDENTIALS_FULL_URI=http://localhost:1338/latest/meta-data/container/security-credentials $CLICKHOUSE_LOCAL -q "select * from s3('http://localhost:11111/test/a.tsv')"