ClickHouse/src/Common/XDBCBridgeHelper.h

352 lines
11 KiB
C++
Raw Normal View History

2018-09-25 14:29:44 +00:00
#pragma once
#include <IO/ReadHelpers.h>
#include <IO/ReadWriteBufferFromHTTP.h>
#include <Interpreters/Context.h>
#include <Access/AccessType.h>
#include <Parsers/IdentifierQuotingStyle.h>
2018-09-25 14:29:44 +00:00
#include <Poco/File.h>
#include <Poco/Logger.h>
2018-09-25 14:29:44 +00:00
#include <Poco/Net/HTTPRequest.h>
#include <Poco/Path.h>
#include <Poco/URI.h>
2018-09-25 14:29:44 +00:00
#include <Poco/Util/AbstractConfiguration.h>
#include <Common/ShellCommand.h>
#include <IO/ConnectionTimeoutsContext.h>
2018-09-25 14:29:44 +00:00
#include <common/logger_useful.h>
#include <ext/range.h>
#if !defined(ARCADIA_BUILD)
# include <Common/config.h>
#endif
2018-09-25 14:29:44 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int EXTERNAL_SERVER_IS_NOT_RESPONDING;
extern const int ILLEGAL_TYPE_OF_ARGUMENT;
2018-09-25 14:29:44 +00:00
}
/**
* Class for Helpers for Xdbc-bridges, provide utility methods, not main request
*/
class IXDBCBridgeHelper
{
public:
static constexpr inline auto DEFAULT_FORMAT = "RowBinary";
2019-02-10 16:55:12 +00:00
virtual std::vector<std::pair<std::string, std::string>> getURLParams(const std::string & cols, UInt64 max_block_size) const = 0;
2018-09-25 14:29:44 +00:00
virtual void startBridgeSync() const = 0;
virtual Poco::URI getMainURI() const = 0;
virtual Poco::URI getColumnsInfoURI() const = 0;
2018-09-27 15:23:42 +00:00
virtual IdentifierQuotingStyle getIdentifierQuotingStyle() = 0;
virtual bool isSchemaAllowed() = 0;
virtual String getName() const = 0;
2018-09-25 14:29:44 +00:00
virtual ~IXDBCBridgeHelper() = default;
2018-09-25 14:29:44 +00:00
};
using BridgeHelperPtr = std::shared_ptr<IXDBCBridgeHelper>;
template <typename BridgeHelperMixin>
class XDBCBridgeHelper : public IXDBCBridgeHelper
{
private:
Poco::Timespan http_timeout;
std::string connection_string;
Poco::URI ping_url;
Poco::Logger * log = &Poco::Logger::get(BridgeHelperMixin::getName() + "BridgeHelper");
2018-09-25 14:29:44 +00:00
2018-09-27 15:23:42 +00:00
std::optional<IdentifierQuotingStyle> quote_style;
std::optional<bool> is_schema_allowed;
2018-09-27 15:23:42 +00:00
2018-09-25 14:29:44 +00:00
protected:
auto getConnectionString() const
{
return connection_string;
}
public:
using Configuration = Poco::Util::AbstractConfiguration;
2019-10-10 20:47:47 +00:00
const Context & context;
2018-09-25 14:29:44 +00:00
const Configuration & config;
static constexpr inline auto DEFAULT_HOST = "localhost";
static constexpr inline auto DEFAULT_PORT = BridgeHelperMixin::DEFAULT_PORT;
static constexpr inline auto PING_HANDLER = "/ping";
static constexpr inline auto MAIN_HANDLER = "/";
static constexpr inline auto COL_INFO_HANDLER = "/columns_info";
2018-09-27 15:23:42 +00:00
static constexpr inline auto IDENTIFIER_QUOTE_HANDLER = "/identifier_quote";
static constexpr inline auto SCHEMA_ALLOWED_HANDLER = "/schema_allowed";
2018-09-25 14:29:44 +00:00
static constexpr inline auto PING_OK_ANSWER = "Ok.";
2019-10-10 20:47:47 +00:00
XDBCBridgeHelper(const Context & global_context_, const Poco::Timespan & http_timeout_, const std::string & connection_string_)
: http_timeout(http_timeout_), connection_string(connection_string_), context(global_context_), config(context.getConfigRef())
2018-09-25 14:29:44 +00:00
{
size_t bridge_port = config.getUInt(BridgeHelperMixin::configPrefix() + ".port", DEFAULT_PORT);
std::string bridge_host = config.getString(BridgeHelperMixin::configPrefix() + ".host", DEFAULT_HOST);
ping_url.setHost(bridge_host);
ping_url.setPort(bridge_port);
ping_url.setScheme("http");
ping_url.setPath(PING_HANDLER);
}
2018-10-10 08:38:54 +00:00
String getName() const override
{
return BridgeHelperMixin::getName();
}
2018-09-27 15:23:42 +00:00
IdentifierQuotingStyle getIdentifierQuotingStyle() override
2018-09-25 14:29:44 +00:00
{
2018-09-27 15:23:42 +00:00
if (!quote_style.has_value())
{
startBridgeSync();
2018-09-27 15:23:42 +00:00
auto uri = createBaseURI();
uri.setPath(IDENTIFIER_QUOTE_HANDLER);
uri.addQueryParameter("connection_string", getConnectionString());
ReadWriteBufferFromHTTP buf(
2020-06-16 19:14:58 +00:00
uri, Poco::Net::HTTPRequest::HTTP_POST, {}, ConnectionTimeouts::getHTTPTimeouts(context));
2018-09-27 15:23:42 +00:00
std::string character;
readStringBinary(character, buf);
if (character.length() > 1)
throw Exception("Failed to parse quoting style from '" + character + "' for service " + BridgeHelperMixin::serviceAlias(), ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
else if (character.length() == 0)
2018-09-27 15:23:42 +00:00
quote_style = IdentifierQuotingStyle::None;
else if (character[0] == '`')
2018-09-27 15:23:42 +00:00
quote_style = IdentifierQuotingStyle::Backticks;
else if (character[0] == '"')
2018-09-27 15:23:42 +00:00
quote_style = IdentifierQuotingStyle::DoubleQuotes;
else
throw Exception("Can not map quote identifier '" + character + "' to enum value", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
2018-09-27 15:23:42 +00:00
}
return *quote_style;
2018-09-25 14:29:44 +00:00
}
bool isSchemaAllowed() override
{
if (!is_schema_allowed.has_value())
{
startBridgeSync();
auto uri = createBaseURI();
uri.setPath(SCHEMA_ALLOWED_HANDLER);
uri.addQueryParameter("connection_string", getConnectionString());
ReadWriteBufferFromHTTP buf(
uri, Poco::Net::HTTPRequest::HTTP_POST, {}, ConnectionTimeouts::getHTTPTimeouts(context));
bool res;
readBoolText(res, buf);
is_schema_allowed = res;
}
return *is_schema_allowed;
}
2018-09-25 14:29:44 +00:00
/**
* @todo leaky abstraction - used by external API's
*/
2019-02-10 16:55:12 +00:00
std::vector<std::pair<std::string, std::string>> getURLParams(const std::string & cols, UInt64 max_block_size) const override
2018-09-25 14:29:44 +00:00
{
std::vector<std::pair<std::string, std::string>> result;
result.emplace_back("connection_string", connection_string); /// already validated
result.emplace_back("columns", cols);
result.emplace_back("max_block_size", std::to_string(max_block_size));
return result;
}
/**
* Performs spawn of external daemon
*/
void startBridgeSync() const override
{
if (!checkBridgeIsRunning())
{
2020-05-23 22:24:01 +00:00
LOG_TRACE(log, "{} is not running, will try to start it", BridgeHelperMixin::serviceAlias());
2018-09-25 14:29:44 +00:00
startBridge();
bool started = false;
uint64_t milliseconds_to_wait = 10; /// Exponential backoff
uint64_t counter = 0;
while (milliseconds_to_wait < 10000)
2018-09-25 14:29:44 +00:00
{
++counter;
2020-05-23 22:24:01 +00:00
LOG_TRACE(log, "Checking {} is running, try {}", BridgeHelperMixin::serviceAlias(), counter);
2018-09-25 14:29:44 +00:00
if (checkBridgeIsRunning())
{
started = true;
break;
}
std::this_thread::sleep_for(std::chrono::milliseconds(milliseconds_to_wait));
milliseconds_to_wait *= 2;
2018-09-25 14:29:44 +00:00
}
2018-09-25 14:29:44 +00:00
if (!started)
throw Exception(BridgeHelperMixin::getName() + "BridgeHelper: " + BridgeHelperMixin::serviceAlias() + " is not responding",
ErrorCodes::EXTERNAL_SERVER_IS_NOT_RESPONDING);
2018-09-25 14:29:44 +00:00
}
}
/**
* URI to fetch the data from external service
*/
Poco::URI getMainURI() const override
{
auto uri = createBaseURI();
uri.setPath(MAIN_HANDLER);
return uri;
}
/**
* URI to retrieve column description from external service
*/
Poco::URI getColumnsInfoURI() const override
{
auto uri = createBaseURI();
uri.setPath(COL_INFO_HANDLER);
return uri;
}
protected:
Poco::URI createBaseURI() const
{
Poco::URI uri;
uri.setHost(ping_url.getHost());
uri.setPort(ping_url.getPort());
2018-09-25 14:29:44 +00:00
uri.setScheme("http");
return uri;
}
private:
bool checkBridgeIsRunning() const
{
try
{
ReadWriteBufferFromHTTP buf(
2020-06-16 19:14:58 +00:00
ping_url, Poco::Net::HTTPRequest::HTTP_GET, {}, ConnectionTimeouts::getHTTPTimeouts(context));
2018-09-25 14:29:44 +00:00
return checkString(XDBCBridgeHelper::PING_OK_ANSWER, buf);
}
catch (...)
{
return false;
}
}
/* Contains logic for instantiation of the bridge instance */
void startBridge() const
{
auto cmd = BridgeHelperMixin::startBridge(config, log, http_timeout);
context.addXDBCBridgeCommand(std::move(cmd));
2018-09-25 14:29:44 +00:00
}
};
struct JDBCBridgeMixin
{
static constexpr inline auto DEFAULT_PORT = 9019;
static const String configPrefix()
{
return "jdbc_bridge";
}
static const String serviceAlias()
{
return "clickhouse-jdbc-bridge";
}
static const String getName()
{
return "JDBC";
}
static AccessType getSourceAccessType()
{
return AccessType::JDBC;
}
2018-09-25 14:29:44 +00:00
static std::unique_ptr<ShellCommand> startBridge(const Poco::Util::AbstractConfiguration &, const Poco::Logger *, const Poco::Timespan &)
2018-09-25 14:29:44 +00:00
{
throw Exception("jdbc-bridge is not running. Please, start it manually", ErrorCodes::EXTERNAL_SERVER_IS_NOT_RESPONDING);
2018-09-25 14:29:44 +00:00
}
};
struct ODBCBridgeMixin
{
2018-09-25 14:29:44 +00:00
static constexpr inline auto DEFAULT_PORT = 9018;
static const String configPrefix()
{
return "odbc_bridge";
}
static const String serviceAlias()
{
return "clickhouse-odbc-bridge";
}
static const String getName()
{
return "ODBC";
}
static AccessType getSourceAccessType()
{
return AccessType::ODBC;
}
2018-09-25 14:29:44 +00:00
2020-06-15 22:23:13 +00:00
static std::unique_ptr<ShellCommand> startBridge(
const Poco::Util::AbstractConfiguration & config, Poco::Logger * log, const Poco::Timespan & http_timeout)
2018-09-25 14:29:44 +00:00
{
/// Path to executable folder
Poco::Path path{config.getString("application.dir", "/usr/bin")};
2018-09-25 14:29:44 +00:00
std::vector<std::string> cmd_args;
path.setFileName("clickhouse-odbc-bridge");
2018-09-25 14:29:44 +00:00
2018-11-22 16:01:17 +00:00
#if !CLICKHOUSE_SPLIT_BINARY
cmd_args.push_back("odbc-bridge");
2018-09-25 14:29:44 +00:00
#endif
cmd_args.push_back("--http-port");
cmd_args.push_back(std::to_string(config.getUInt(configPrefix() + ".port", DEFAULT_PORT)));
cmd_args.push_back("--listen-host");
cmd_args.push_back(config.getString(configPrefix() + ".listen_host", XDBCBridgeHelper<ODBCBridgeMixin>::DEFAULT_HOST));
cmd_args.push_back("--http-timeout");
cmd_args.push_back(std::to_string(http_timeout.totalMicroseconds()));
if (config.has("logger." + configPrefix() + "_log"))
{
cmd_args.push_back("--log-path");
cmd_args.push_back(config.getString("logger." + configPrefix() + "_log"));
}
if (config.has("logger." + configPrefix() + "_errlog"))
{
cmd_args.push_back("--err-log-path");
cmd_args.push_back(config.getString("logger." + configPrefix() + "_errlog"));
}
if (config.has("logger." + configPrefix() + "_stdout"))
{
cmd_args.push_back("--stdout-path");
cmd_args.push_back(config.getString("logger." + configPrefix() + "_stdout"));
}
if (config.has("logger." + configPrefix() + "_stderr"))
{
cmd_args.push_back("--stderr-path");
cmd_args.push_back(config.getString("logger." + configPrefix() + "_stderr"));
}
if (config.has("logger." + configPrefix() + "_level"))
{
cmd_args.push_back("--log-level");
cmd_args.push_back(config.getString("logger." + configPrefix() + "_level"));
}
2020-05-23 22:24:01 +00:00
LOG_TRACE(log, "Starting {}", serviceAlias());
2018-09-25 14:29:44 +00:00
return ShellCommand::executeDirect(path.toString(), cmd_args, true);
2018-09-25 14:29:44 +00:00
}
};
2018-10-10 08:38:54 +00:00
}