ClickHouse/dbms/include/DB/Interpreters/SystemLog.h
2016-11-24 23:25:02 +03:00

325 lines
8.6 KiB
C++

#pragma once
#include <thread>
#include <boost/noncopyable.hpp>
#include <DB/Core/Types.h>
#include <DB/Common/ConcurrentBoundedQueue.h>
#include <DB/Storages/IStorage.h>
#include <DB/Interpreters/Context.h>
#include <DB/Common/Stopwatch.h>
#include <DB/Parsers/ASTCreateQuery.h>
#include <DB/Parsers/parseQuery.h>
#include <DB/Parsers/ExpressionElementParsers.h>
#include <DB/Parsers/ASTRenameQuery.h>
#include <DB/Parsers/formatAST.h>
#include <DB/Parsers/ASTInsertQuery.h>
#include <DB/Interpreters/InterpreterCreateQuery.h>
#include <DB/Interpreters/InterpreterRenameQuery.h>
#include <DB/Interpreters/InterpreterInsertQuery.h>
#include <DB/Common/setThreadName.h>
namespace DB
{
/** Allow to store structured log in system table.
*
* Logging is asynchronous. Data is put into queue from where it will be read by separate thread.
* That thread inserts log into a table with no more than specified periodicity.
*/
/** Structure of log, template parameter.
* Structure could change on server version update.
* If on first write, existing table has different structure,
* then it get renamed (put aside) and new table is created.
*/
/* Example:
struct LogElement
{
/// default constructor must be available
/// fields
static std::string name();
static Block createBlock();
void appendToBlock(Block & block) const;
};
*/
#define DBMS_SYSTEM_LOG_QUEUE_SIZE 1024
class Context;
template <typename LogElement>
class SystemLog : private boost::noncopyable
{
public:
/** Parameter: table name where to write log.
* If table is not exists, then it get created with specified engine.
* If it already exists, then its structure is checked to be compatible with structure of log record.
* If it is compatible, then existing table will be used.
* If not - then existing table will be renamed to same name but with suffix '_N' at end,
* where N - is a minimal number from 1, for that table with corresponding name doesn't exist yet;
* and new table get created - as if previous table was not exist.
*/
SystemLog(
Context & context_,
const String & database_name_,
const String & table_name_,
const String & engine_,
size_t flush_interval_milliseconds_);
~SystemLog();
/** Append a record into log.
* Writing to table will be done asynchronously and in case of failure, record could be lost.
*/
void add(const LogElement & element)
{
/// We could lock here in case of queue overflow. Maybe better to throw an exception or even don't do logging in that case.
queue.push({false, element});
}
private:
Context & context;
const String database_name;
const String table_name;
StoragePtr table;
const String engine;
const size_t flush_interval_milliseconds;
using QueueItem = std::pair<bool, LogElement>; /// First element is shutdown flag for thread.
/// Queue is bounded. But its size is quite large to not block in all normal cases.
ConcurrentBoundedQueue<QueueItem> queue {DBMS_SYSTEM_LOG_QUEUE_SIZE};
/** Data that was pulled from queue. Data is accumulated here before enough time passed.
* It's possible to implement double-buffering, but we assume that insertion into table is faster
* than accumulation of large amount of log records (for example, for query log - processing of large amount of queries).
*/
std::vector<LogElement> data;
Logger * log;
/** In this thread, data is pulled from 'queue' and stored in 'data', and then written into table.
*/
std::thread saving_thread;
void threadFunction();
void flush();
/** Creates new table if it does not exist.
* Renames old table if its structure is not suitable.
* This cannot be done in constructor to avoid deadlock while renaming a table under locked Context when SystemLog object is created.
*/
bool is_prepared = false;
void prepareTable();
};
template <typename LogElement>
SystemLog<LogElement>::SystemLog(Context & context_,
const String & database_name_,
const String & table_name_,
const String & engine_,
size_t flush_interval_milliseconds_)
: context(context_),
database_name(database_name_), table_name(table_name_), engine(engine_),
flush_interval_milliseconds(flush_interval_milliseconds_)
{
log = &Logger::get("SystemLog (" + database_name + "." + table_name + ")");
data.reserve(DBMS_SYSTEM_LOG_QUEUE_SIZE);
saving_thread = std::thread([this] { threadFunction(); });
}
template <typename LogElement>
SystemLog<LogElement>::~SystemLog()
{
/// Tell thread to shutdown.
queue.push({true, {}});
saving_thread.join();
}
template <typename LogElement>
void SystemLog<LogElement>::threadFunction()
{
setThreadName("SystemLogFlush");
Stopwatch time_after_last_write;
bool first = true;
while (true)
{
try
{
if (first)
{
time_after_last_write.restart();
first = false;
}
QueueItem element;
bool has_element = false;
if (data.empty())
{
queue.pop(element);
has_element = true;
}
else
{
size_t milliseconds_elapsed = time_after_last_write.elapsed() / 1000000;
if (milliseconds_elapsed < flush_interval_milliseconds)
has_element = queue.tryPop(element, flush_interval_milliseconds - milliseconds_elapsed);
}
if (has_element)
{
if (element.first)
{
/// Shutdown.
flush();
break;
}
else
data.push_back(element.second);
}
size_t milliseconds_elapsed = time_after_last_write.elapsed() / 1000000;
if (milliseconds_elapsed >= flush_interval_milliseconds)
{
/// Write data to a table.
flush();
time_after_last_write.restart();
}
}
catch (...)
{
/// In case of exception we lost accumulated data - to avoid locking.
data.clear();
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
}
template <typename LogElement>
void SystemLog<LogElement>::flush()
{
try
{
LOG_TRACE(log, "Flushing query log");
if (!is_prepared) /// BTW, flush method is called from single thread.
prepareTable();
Block block = LogElement::createBlock();
for (const LogElement & elem : data)
elem.appendToBlock(block);
/// We write to table indirectly, using InterpreterInsertQuery.
/// This is needed to support DEFAULT-columns in table.
std::unique_ptr<ASTInsertQuery> insert = std::make_unique<ASTInsertQuery>();
insert->database = database_name;
insert->table = table_name;
ASTPtr query_ptr(insert.release());
InterpreterInsertQuery interpreter(query_ptr, context);
BlockIO io = interpreter.execute();
io.out->writePrefix();
io.out->write(block);
io.out->writeSuffix();
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
/// In case of exception, also clean accumulated data - to avoid locking.
data.clear();
}
template <typename LogElement>
void SystemLog<LogElement>::prepareTable()
{
String description = backQuoteIfNeed(database_name) + "." + backQuoteIfNeed(table_name);
table = context.tryGetTable(database_name, table_name);
if (table)
{
const Block expected = LogElement::createBlock();
const Block actual = table->getSampleBlockNonMaterialized();
if (!blocksHaveEqualStructure(actual, expected))
{
/// Переименовываем существующую таблицу.
int suffix = 0;
while (context.isTableExist(database_name, table_name + "_" + toString(suffix)))
++suffix;
auto rename = std::make_shared<ASTRenameQuery>();
ASTRenameQuery::Table from;
from.database = database_name;
from.table = table_name;
ASTRenameQuery::Table to;
to.database = database_name;
to.table = table_name + "_" + toString(suffix);
ASTRenameQuery::Element elem;
elem.from = from;
elem.to = to;
rename->elements.emplace_back(elem);
LOG_DEBUG(log, "Existing table " << description << " for system log has obsolete or different structure."
" Renaming it to " << backQuoteIfNeed(to.table));
InterpreterRenameQuery(rename, context).execute();
/// Нужная таблица будет создана.
table = nullptr;
}
else
LOG_DEBUG(log, "Will use existing table " << description << " for " + LogElement::name());
}
if (!table)
{
/// Создаём таблицу.
LOG_DEBUG(log, "Creating new table " << description << " for " + LogElement::name());
auto create = std::make_shared<ASTCreateQuery>();
create->database = database_name;
create->table = table_name;
Block sample = LogElement::createBlock();
create->columns = InterpreterCreateQuery::formatColumns(sample.getColumnsList());
ParserFunction engine_parser;
create->storage = parseQuery(engine_parser, engine.data(), engine.data() + engine.size(), "ENGINE to create table for" + LogElement::name());
InterpreterCreateQuery(create, context).execute();
table = context.getTable(database_name, table_name);
}
is_prepared = true;
}
}