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 | |
11 | namespace DB |
12 | { |
13 | |
14 | InternalTextLogsQueue::InternalTextLogsQueue() |
15 | : ConcurrentBoundedQueue<MutableColumns>(std::numeric_limits<int>::max()), |
16 | max_priority(Poco::Message::Priority::PRIO_INFORMATION) {} |
17 | |
18 | |
19 | Block 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 | |
33 | MutableColumns InternalTextLogsQueue::getSampleColumns() |
34 | { |
35 | static Block sample_block = getSampleBlock(); |
36 | return sample_block.cloneEmptyColumns(); |
37 | } |
38 | |
39 | void 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 | |
49 | const 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 | |