ClickHouse/dbms/programs/server/InterserverIOHTTPHandler.cpp

141 lines
4.7 KiB
C++
Raw Normal View History

#include "InterserverIOHTTPHandler.h"
#include <Poco/Net/HTTPBasicCredentials.h>
2017-08-09 14:33:07 +00:00
#include <Poco/Net/HTTPServerRequest.h>
#include <Poco/Net/HTTPServerResponse.h>
#include <common/logger_useful.h>
#include <Common/HTMLForm.h>
2018-08-31 00:59:48 +00:00
#include <Common/setThreadName.h>
2018-12-28 18:15:26 +00:00
#include <Compression/CompressedWriteBuffer.h>
#include <IO/ReadBufferFromIStream.h>
2017-08-09 14:33:07 +00:00
#include <IO/WriteBufferFromHTTPServerResponse.h>
#include <Interpreters/InterserverIOHandler.h>
#include "IServer.h"
2014-03-21 13:42:14 +00:00
namespace DB
{
2016-01-12 02:21:15 +00:00
namespace ErrorCodes
{
extern const int ABORTED;
2018-03-09 23:23:15 +00:00
extern const int TOO_MANY_SIMULTANEOUS_QUERIES;
2016-01-12 02:21:15 +00:00
}
std::pair<String, bool> InterserverIOHTTPHandler::checkAuthentication(Poco::Net::HTTPServerRequest & request) const
2014-03-21 13:42:14 +00:00
{
const auto & config = server.config();
2018-07-26 09:53:59 +00:00
if (config.has("interserver_http_credentials.user"))
{
if (!request.hasCredentials())
2020-01-11 09:50:41 +00:00
return {"Server requires HTTP Basic authentication, but client doesn't provide it", false};
String scheme, info;
request.getCredentials(scheme, info);
if (scheme != "Basic")
2020-01-11 09:50:41 +00:00
return {"Server requires HTTP Basic authentication but client provides another method", false};
String user = config.getString("interserver_http_credentials.user");
String password = config.getString("interserver_http_credentials.password", "");
Poco::Net::HTTPBasicCredentials credentials(info);
if (std::make_pair(user, password) != std::make_pair(credentials.getUsername(), credentials.getPassword()))
2020-01-11 09:50:41 +00:00
return {"Incorrect user or password in HTTP Basic authentication", false};
}
else if (request.hasCredentials())
{
2020-01-11 09:50:41 +00:00
return {"Client requires HTTP Basic authentication, but server doesn't provide it", false};
}
return {"", true};
}
void InterserverIOHTTPHandler::processQuery(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response, Output & used_output)
{
HTMLForm params(request);
LOG_TRACE(log, "Request URI: " << request.getURI());
String endpoint_name = params.get("endpoint");
bool compress = params.get("compress") == "true";
2014-03-21 13:42:14 +00:00
ReadBufferFromIStream body(request.stream());
2016-03-01 17:47:53 +00:00
2017-08-09 11:57:09 +00:00
auto endpoint = server.context().getInterserverIOHandler().getEndpoint(endpoint_name);
2020-01-14 14:27:48 +00:00
/// Locked for read while query processing
std::shared_lock lock(endpoint->rwlock);
if (endpoint->blocker.isCancelled())
throw Exception("Transferring part to replica was cancelled", ErrorCodes::ABORTED);
2014-03-21 13:42:14 +00:00
if (compress)
{
2020-03-09 03:38:43 +00:00
CompressedWriteBuffer compressed_out(*used_output.out);
endpoint->processQuery(params, body, compressed_out, response);
}
else
{
2020-03-09 03:38:43 +00:00
endpoint->processQuery(params, body, *used_output.out, response);
}
2014-03-21 13:42:14 +00:00
}
void InterserverIOHTTPHandler::handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response)
{
2018-08-31 00:59:48 +00:00
setThreadName("IntersrvHandler");
/// In order to work keep-alive.
if (request.getVersion() == Poco::Net::HTTPServerRequest::HTTP_1_1)
response.setChunkedTransferEncoding(true);
Output used_output;
const auto & config = server.config();
unsigned keep_alive_timeout = config.getUInt("keep_alive_timeout", 10);
used_output.out = std::make_shared<WriteBufferFromHTTPServerResponse>(request, response, keep_alive_timeout);
try
{
if (auto [message, success] = checkAuthentication(request); success)
{
processQuery(request, response, used_output);
LOG_INFO(log, "Done processing query");
}
else
{
response.setStatusAndReason(Poco::Net::HTTPServerResponse::HTTP_UNAUTHORIZED);
if (!response.sent())
writeString(message, *used_output.out);
2020-01-11 09:50:41 +00:00
LOG_WARNING(log, "Query processing failed request: '" << request.getURI() << "' authentication failed");
}
}
catch (Exception & e)
{
2018-03-09 23:23:15 +00:00
if (e.code() == ErrorCodes::TOO_MANY_SIMULTANEOUS_QUERIES)
return;
response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR);
/// Sending to remote server was cancelled due to server shutdown or drop table.
bool is_real_error = e.code() != ErrorCodes::ABORTED;
std::string message = getCurrentExceptionMessage(is_real_error);
if (!response.sent())
writeString(message, *used_output.out);
if (is_real_error)
LOG_ERROR(log, message);
else
LOG_INFO(log, message);
}
catch (...)
{
response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR);
std::string message = getCurrentExceptionMessage(false);
if (!response.sent())
writeString(message, *used_output.out);
LOG_ERROR(log, message);
}
2014-03-21 13:42:14 +00:00
}
}