1#include "InternalTextLogsQueue.h"
2#include <DataTypes/DataTypeDateTime.h>
3#include <DataTypes/DataTypeString.h>
4#include <DataTypes/DataTypeEnum.h>
5#include <DataTypes/DataTypesNumber.h>
6#include <common/logger_useful.h>
7
8#include <Poco/Message.h>
9
10
11namespace DB
12{
13
14InternalTextLogsQueue::InternalTextLogsQueue()
15 : ConcurrentBoundedQueue<MutableColumns>(std::numeric_limits<int>::max()),
16 max_priority(Poco::Message::Priority::PRIO_INFORMATION) {}
17
18
19Block InternalTextLogsQueue::getSampleBlock()
20{
21 return Block {
22 {std::make_shared<DataTypeDateTime>(), "event_time"},
23 {std::make_shared<DataTypeUInt32>(), "event_time_microseconds"},
24 {std::make_shared<DataTypeString>(), "host_name"},
25 {std::make_shared<DataTypeString>(), "query_id"},
26 {std::make_shared<DataTypeUInt32>(), "thread_number"},
27 {std::make_shared<DataTypeInt8>(), "priority"},
28 {std::make_shared<DataTypeString>(), "source"},
29 {std::make_shared<DataTypeString>(), "text"}
30 };
31}
32
33MutableColumns InternalTextLogsQueue::getSampleColumns()
34{
35 static Block sample_block = getSampleBlock();
36 return sample_block.cloneEmptyColumns();
37}
38
39void InternalTextLogsQueue::pushBlock(Block && log_block)
40{
41 static Block sample_block = getSampleBlock();
42
43 if (blocksHaveEqualStructure(sample_block, log_block))
44 emplace(log_block.mutateColumns());
45 else
46 LOG_WARNING(&Poco::Logger::get("InternalTextLogsQueue"), "Log block have different structure");
47}
48
49const char * InternalTextLogsQueue::getPriorityName(int priority)
50{
51 /// See Poco::Message::Priority
52
53 static const char * PRIORITIES [] = {
54 "Unknown",
55 "Fatal",
56 "Critical",
57 "Error",
58 "Warning",
59 "Notice",
60 "Information",
61 "Debug",
62 "Trace"
63 };
64
65 return (priority >= 1 && priority <= 8) ? PRIORITIES[priority] : PRIORITIES[0];
66}
67
68}
69