CLICKHOUSE-3878: Better odbc-bridge with ability to handle custom logs and exceptions

This commit is contained in:
alesapin 2018-08-08 19:15:29 +03:00
parent dd01eb6b40
commit 92f3beb95b
7 changed files with 261 additions and 166 deletions

View File

@ -1,6 +1,7 @@
add_library (clickhouse-odbc-bridge-lib
Handlers.cpp
HandlerFactory.cpp
ODBCBridge.cpp
ODBCHandler.cpp
)
target_link_libraries (clickhouse-odbc-bridge-lib clickhouse_common_io daemon)

View File

@ -0,0 +1,65 @@
#include "HandlerFactory.h"
#include <Common/HTMLForm.h>
#include "Handlers.h"
#include <Dictionaries/validateODBCConnectionString.h>
#include <Poco/Ext/SessionPoolHelpers.h>
#include <Poco/Net/HTTPServerRequest.h>
#include <common/logger_useful.h>
namespace DB
{
namespace
{
std::string buildConnectionString(const std::string & DSN, const std::string & database)
{
std::stringstream ss;
ss << "DSN=" << DSN << ";DATABASE=" << database;
return ss.str();
}
}
Poco::Net::HTTPRequestHandler * HandlerFactory::createRequestHandler(const Poco::Net::HTTPServerRequest & request)
{
const auto & uri = request.getURI();
LOG_TRACE(log, "Request URI: " + uri);
if (uri == "/ping" && request.getMethod() == Poco::Net::HTTPRequest::HTTP_GET)
return new PingHandler(keep_alive_timeout);
HTMLForm params(request);
std::string DSN = params.get("DSN", "");
std::string database = params.get("database", "");
std::string max_block_size = params.get("max_block_size", "");
std::string format = params.get("format", "RowBinary");
std::string connection_string = buildConnectionString(DSN, database);
LOG_TRACE(log, "Connection string:" << connection_string);
std::shared_ptr<Poco::Data::SessionPool> pool = nullptr;
if (!pool_map.count(connection_string))
try
{
std::string validated = validateODBCConnectionString(connection_string);
pool
= createAndCheckResizePocoSessionPool([validated] { return std::make_shared<Poco::Data::SessionPool>("ODBC", validated); });
pool_map[connection_string] = pool;
}
catch (const Exception & ex)
{
LOG_WARNING(log, "Connection string validation failed: " + ex.message());
}
else
{
pool = pool_map[connection_string];
}
if (request.getMethod() == Poco::Net::HTTPRequest::HTTP_POST)
return new ODBCHandler(
pool, format, max_block_size == "" ? DEFAULT_BLOCK_SIZE : parse<size_t>(max_block_size), keep_alive_timeout, context);
return nullptr;
}
}

View File

@ -0,0 +1,32 @@
#pragma once
#include <Interpreters/Context.h>
#include <Poco/Logger.h>
#include <Poco/Net/HTTPRequestHandler.h>
#include <Poco/Net/HTTPRequestHandlerFactory.h>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-parameter"
#include <Poco/Data/SessionPool.h>
#pragma GCC diagnostic pop
namespace DB
{
class HandlerFactory : public Poco::Net::HTTPRequestHandlerFactory
{
public:
HandlerFactory(const std::string & name_, size_t keep_alive_timeout_, std::shared_ptr<Context> context_)
: log(&Poco::Logger::get(name_)), name(name_), keep_alive_timeout(keep_alive_timeout_), context(context_)
{
}
Poco::Net::HTTPRequestHandler * createRequestHandler(const Poco::Net::HTTPServerRequest & request) override;
private:
Poco::Logger * log;
std::string name;
size_t keep_alive_timeout;
std::shared_ptr<Context> context;
std::unordered_map<std::string, std::shared_ptr<Poco::Data::SessionPool>> pool_map;
};
}

View File

@ -0,0 +1,140 @@
#include "Handlers.h"
#include <Common/HTMLForm.h>
#include <memory>
#include <DataStreams/IBlockOutputStream.h>
#include <DataTypes/DataTypeFactory.h>
#include <Dictionaries/ODBCBlockInputStream.h>
#include <Formats/BinaryRowInputStream.h>
#include <Formats/FormatFactory.h>
#include <IO/ReadBufferFromIStream.h>
#include <IO/WriteBufferFromHTTPServerResponse.h>
#include <IO/WriteHelpers.h>
#include <Interpreters/Context.h>
#include <boost/algorithm/string.hpp>
#include <boost/tokenizer.hpp>
#include <Poco/Net/HTTPServerRequest.h>
#include <Poco/Net/HTTPServerResponse.h>
#include <common/logger_useful.h>
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_REQUEST_PARAMETER;
}
namespace
{
std::optional<NamesAndTypesList> parseColumns(std::string && column_string, Poco::Logger * log)
{
const auto & factory_instance = DataTypeFactory::instance();
NamesAndTypesList result;
static boost::char_separator<char> sep(",");
boost::tokenizer<boost::char_separator<char>> tokens(column_string, sep);
for (const std::string & name_and_type_str : tokens)
{
std::vector<std::string> name_and_type;
boost::split(name_and_type, name_and_type_str, boost::is_any_of(":"));
if (name_and_type.size() != 2)
return std::nullopt;
try
{
result.emplace_back(name_and_type[0], factory_instance.get(name_and_type[1]));
}
catch (...)
{
tryLogCurrentException(log);
return std::nullopt;
}
}
return result;
}
}
void ODBCHandler::handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response)
{
Poco::Net::HTMLForm params(request, request.stream());
LOG_TRACE(log, "Request URI: " + request.getURI());
auto process_error = [&response, this](const std::string & message) {
response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR);
if (!response.sent())
response.send() << message << std::endl;
LOG_WARNING(log, message);
};
if (pool == nullptr)
{
process_error("ODBCBridge: DSN or database in URL params is not provided or incorrect");
return;
}
if (!params.has("query"))
{
process_error("ODBCBridge: No 'query' in request body");
return;
}
if (!params.has("columns"))
{
process_error("ODBCBridge: No 'columns' in request body");
return;
}
std::string query = params.get("query");
std::string columns = params.get("columns");
auto names_and_types = parseColumns(std::move(columns), log);
if (!names_and_types)
{
process_error("ODBCBridge: Invalid 'columns' parameter in request body");
return;
}
Block sample_block;
for (const NameAndTypePair & column_data : *names_and_types)
sample_block.insert({column_data.type, column_data.name});
WriteBufferFromHTTPServerResponse out(request, response, keep_alive_timeout);
std::shared_ptr<IBlockOutputStream> writer = FormatFactory::instance().getOutput(format, out, sample_block, *context);
try
{
ODBCBlockInputStream inp(pool->get(), query, sample_block, max_block_size);
writer->writePrefix();
while (auto block = inp.read())
writer->write(block);
writer->writeSuffix();
writer->flush();
}
catch (...)
{
auto message = "Communication with ODBC-driver failed: \n" + getCurrentExceptionMessage(true);
response.setStatusAndReason(
Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR); // can't call process_error, bacause of too soon response sending
writeStringBinary(message, out);
LOG_WARNING(log, message);
}
}
void PingHandler::handleRequest(Poco::Net::HTTPServerRequest & /*request*/, Poco::Net::HTTPServerResponse & response)
{
try
{
setResponseDefaultHeaders(response, keep_alive_timeout);
const char * data = "Ok.\n";
response.sendBuffer(data, strlen(data));
}
catch (...)
{
tryLogCurrentException("PingHandler");
}
}
}

View File

@ -3,7 +3,6 @@
#include <Interpreters/Context.h>
#include <Poco/Logger.h>
#include <Poco/Net/HTTPRequestHandler.h>
#include <Poco/Net/HTTPRequestHandlerFactory.h>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-parameter"
@ -12,15 +11,15 @@
namespace DB
{
class ODBCHTTPHandler : public Poco::Net::HTTPRequestHandler
class ODBCHandler : public Poco::Net::HTTPRequestHandler
{
public:
ODBCHTTPHandler(std::shared_ptr<Poco::Data::SessionPool> pool_,
ODBCHandler(std::shared_ptr<Poco::Data::SessionPool> pool_,
const std::string & format_,
size_t max_block_size_,
size_t keep_alive_timeout_,
std::shared_ptr<Context> context_)
: log(&Poco::Logger::get("ODBCHTTPHandler"))
: log(&Poco::Logger::get("ODBCHandler"))
, pool(pool_)
, format(format_)
, max_block_size(max_block_size_)
@ -49,22 +48,4 @@ public:
private:
size_t keep_alive_timeout;
};
class ODBCRequestHandlerFactory : public Poco::Net::HTTPRequestHandlerFactory
{
public:
ODBCRequestHandlerFactory(const std::string & name_, size_t keep_alive_timeout_, std::shared_ptr<Context> context_)
: log(&Poco::Logger::get(name_)), name(name_), keep_alive_timeout(keep_alive_timeout_), context(context_)
{
}
Poco::Net::HTTPRequestHandler * createRequestHandler(const Poco::Net::HTTPServerRequest & request) override;
private:
Poco::Logger * log;
std::string name;
size_t keep_alive_timeout;
std::shared_ptr<Context> context;
std::unordered_map<std::string, std::shared_ptr<Poco::Data::SessionPool>> pool_map;
};
}

View File

@ -1,5 +1,5 @@
#include "ODBCBridge.h"
#include "ODBCHandler.h"
#include "HandlerFactory.h"
#include <string>
#include <errno.h>
@ -101,7 +101,6 @@ void ODBCBridge::defineOptions(Poco::Util::OptionSet & options)
Poco::Util::Option("http-host", "", "hostname to listen, default localhost").argument("http-host").binding("http-host"));
options.addOption(
Poco::Util::Option("http-timeout", "", "http timout for socket, default 1800").argument("http-timeout").binding("http-timeout"));
options.addOption(Poco::Util::Option("log-level", "", "sets log level, default info").argument("log-level").binding("log-level"));
options.addOption(Poco::Util::Option("max-server-connections", "", "max connections to server, default 1024")
.argument("max-server-connections")
.binding("max-server-connections"));
@ -109,22 +108,36 @@ void ODBCBridge::defineOptions(Poco::Util::OptionSet & options)
.argument("keep-alive-timeout")
.binding("keep-alive-timeout"));
options.addOption(Poco::Util::Option("log-level", "", "sets log level, default info").argument("log-level").binding("logger.level"));
options.addOption(
Poco::Util::Option("log-path", "", "log path for all logs, default console").argument("log-path").binding("logger.log"));
options.addOption(Poco::Util::Option("err-log-path", "", "err log path for all logs, default no")
.argument("err-log-path")
.binding("logger.errorlog"));
using Me = std::decay_t<decltype(*this)>;
options.addOption(Poco::Util::Option("help", "", "produce this help message")
.binding("help")
.callback(Poco::Util::OptionCallback<Me>(this, &Me::handleHelp)));
ServerApplication::defineOptions(options); /// Don't need complex .xml config
ServerApplication::defineOptions(options); /// Don't need complex BaseDaemon's .xml config
}
void ODBCBridge::initialize(Application & self)
{
is_help = config().has("help");
if (is_help)
return;
log_level = config().getString("log-level", "info");
Poco::Logger::root().setLevel(log_level);
if (!config().has("logger.log"))
config().setBool("logger.console", true);
config().setString("logger", "ODBCBridge");
buildLoggers(config());
log = &logger();
hostname = config().getString("http-host", "localhost");
port = config().getUInt("http-port");
@ -165,7 +178,7 @@ int ODBCBridge::main(const std::vector<std::string> & /*args*/)
context->setGlobalContext(*context);
auto server = Poco::Net::HTTPServer(
new ODBCRequestHandlerFactory("ODBCRequestHandlerFactory-factory", keep_alive_timeout, context), server_pool, socket, http_params);
new HandlerFactory("ODBCRequestHandlerFactory-factory", keep_alive_timeout, context), server_pool, socket, http_params);
server.start();
LOG_INFO(log, "Listening http://" + address.toString());

View File

@ -1,137 +0,0 @@
#include "ODBCHandler.h"
#include <Common/HTMLForm.h>
#include <memory>
#include <DataStreams/IBlockOutputStream.h>
#include <DataTypes/DataTypeFactory.h>
#include <Dictionaries/ODBCBlockInputStream.h>
#include <Dictionaries/validateODBCConnectionString.h>
#include <Formats/BinaryRowInputStream.h>
#include <Formats/FormatFactory.h>
#include <IO/ReadBufferFromIStream.h>
#include <IO/WriteBufferFromHTTPServerResponse.h>
#include <IO/WriteHelpers.h>
#include <Interpreters/Context.h>
#include <boost/algorithm/string.hpp>
#include <boost/tokenizer.hpp>
#include <Poco/Ext/SessionPoolHelpers.h>
#include <Poco/Net/HTTPBasicCredentials.h>
#include <Poco/Net/HTTPServerRequest.h>
#include <Poco/Net/HTTPServerResponse.h>
#include <common/logger_useful.h>
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_REQUEST_PARAMETER;
}
namespace
{
std::string buildConnectionString(const std::string & DSN, const std::string & database)
{
std::stringstream ss;
ss << "DSN=" << DSN << ";DATABASE=" << database;
return ss.str();
}
NamesAndTypesList parseColumns(std::string && column_string)
{
const auto & factory_instance = DataTypeFactory::instance();
NamesAndTypesList result;
static boost::char_separator<char> sep(",");
boost::tokenizer<boost::char_separator<char>> tokens(column_string, sep);
for (const std::string & name_and_type_str : tokens)
{
std::vector<std::string> name_and_type;
boost::split(name_and_type, name_and_type_str, boost::is_any_of(":"));
if (name_and_type.size() != 2)
throw Exception("ODBCBridge: Invalid columns parameter '" + column_string + "'", ErrorCodes::BAD_REQUEST_PARAMETER);
result.emplace_back(name_and_type[0], factory_instance.get(name_and_type[1]));
}
return result;
}
}
void ODBCHTTPHandler::handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response)
{
Poco::Net::HTMLForm params(request, request.stream());
LOG_TRACE(log, "Request URI: " + request.getURI());
if (!params.has("query"))
throw Exception("ODBCBridge: No 'query' in request body", ErrorCodes::BAD_REQUEST_PARAMETER);
std::string query = params.get("query");
if (!params.has("columns"))
throw Exception("ODBCBridge: No 'columns' in request body", ErrorCodes::BAD_REQUEST_PARAMETER);
std::string columns = params.get("columns");
auto names_and_types = parseColumns(std::move(columns));
Block sample_block;
for (const NameAndTypePair & column_data : names_and_types)
{
sample_block.insert({column_data.type, column_data.name});
}
ODBCBlockInputStream inp(pool->get(), query, sample_block, max_block_size);
WriteBufferFromHTTPServerResponse out(request, response, keep_alive_timeout);
std::shared_ptr<IBlockOutputStream> writer = FormatFactory::instance().getOutput(format, out, sample_block, *context);
writer->writePrefix();
while (auto block = inp.read())
writer->write(block);
writer->writeSuffix();
writer->flush();
}
void PingHandler::handleRequest(Poco::Net::HTTPServerRequest & /*request*/, Poco::Net::HTTPServerResponse & response)
{
try
{
setResponseDefaultHeaders(response, keep_alive_timeout);
const char * data = "Ok.\n";
response.sendBuffer(data, strlen(data));
}
catch (...)
{
tryLogCurrentException("PingHandler");
}
}
Poco::Net::HTTPRequestHandler * ODBCRequestHandlerFactory::createRequestHandler(const Poco::Net::HTTPServerRequest & request)
{
const auto & uri = request.getURI();
LOG_TRACE(log, "Request URI: " + uri);
if (uri == "/ping" && request.getMethod() == Poco::Net::HTTPRequest::HTTP_GET)
return new PingHandler(keep_alive_timeout);
HTMLForm params(request);
std::string DSN = params.get("DSN");
std::string database = params.get("database");
std::string max_block_size = params.get("max_block_size", "");
std::string format = params.get("format", "RowBinary");
std::string connection_string = buildConnectionString(DSN, database);
LOG_TRACE(log, "Connection string:" << connection_string);
if (!pool_map.count(connection_string))
pool_map[connection_string] = createAndCheckResizePocoSessionPool(
[&] { return std::make_shared<Poco::Data::SessionPool>("ODBC", validateODBCConnectionString(connection_string)); });
if (request.getMethod() == Poco::Net::HTTPRequest::HTTP_POST)
return new ODBCHTTPHandler(pool_map.at(connection_string),
format,
max_block_size == "" ? DEFAULT_BLOCK_SIZE : parse<size_t>(max_block_size),
keep_alive_timeout,
context);
return nullptr;
}
}