diff options
Diffstat (limited to '')
| -rw-r--r-- | lib/livestatus/livestatusquery.cpp | 648 |
1 files changed, 648 insertions, 0 deletions
diff --git a/lib/livestatus/livestatusquery.cpp b/lib/livestatus/livestatusquery.cpp new file mode 100644 index 0000000..0f9b3da --- /dev/null +++ b/lib/livestatus/livestatusquery.cpp @@ -0,0 +1,648 @@ +/* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */ + +#include "livestatus/livestatusquery.hpp" +#include "livestatus/countaggregator.hpp" +#include "livestatus/sumaggregator.hpp" +#include "livestatus/minaggregator.hpp" +#include "livestatus/maxaggregator.hpp" +#include "livestatus/avgaggregator.hpp" +#include "livestatus/stdaggregator.hpp" +#include "livestatus/invsumaggregator.hpp" +#include "livestatus/invavgaggregator.hpp" +#include "livestatus/attributefilter.hpp" +#include "livestatus/negatefilter.hpp" +#include "livestatus/orfilter.hpp" +#include "livestatus/andfilter.hpp" +#include "icinga/externalcommandprocessor.hpp" +#include "base/debug.hpp" +#include "base/convert.hpp" +#include "base/objectlock.hpp" +#include "base/logger.hpp" +#include "base/exception.hpp" +#include "base/utility.hpp" +#include "base/json.hpp" +#include "base/serializer.hpp" +#include "base/timer.hpp" +#include "base/initialize.hpp" +#include <boost/algorithm/string/replace.hpp> +#include <boost/algorithm/string/join.hpp> + +using namespace icinga; + +static int l_ExternalCommands = 0; +static std::mutex l_QueryMutex; + +LivestatusQuery::LivestatusQuery(const std::vector<String>& lines, const String& compat_log_path) + : m_KeepAlive(false), m_OutputFormat("csv"), m_ColumnHeaders(true), m_Limit(-1), m_ErrorCode(0), + m_LogTimeFrom(0), m_LogTimeUntil(static_cast<long>(Utility::GetTime())) +{ + if (lines.size() == 0) { + m_Verb = "ERROR"; + m_ErrorCode = LivestatusErrorQuery; + m_ErrorMessage = "Empty Query. Aborting."; + return; + } + + String msg; + for (const String& line : lines) { + msg += line + "\n"; + } + Log(LogDebug, "LivestatusQuery", msg); + + m_CompatLogPath = compat_log_path; + + /* default separators */ + m_Separators.emplace_back("\n"); + m_Separators.emplace_back(";"); + m_Separators.emplace_back(","); + m_Separators.emplace_back("|"); + + String line = lines[0]; + + size_t sp_index = line.FindFirstOf(" "); + + if (sp_index == String::NPos) + BOOST_THROW_EXCEPTION(std::runtime_error("Livestatus header must contain a verb.")); + + String verb = line.SubStr(0, sp_index); + String target = line.SubStr(sp_index + 1); + + m_Verb = verb; + + if (m_Verb == "COMMAND") { + m_KeepAlive = true; + m_Command = target; + } else if (m_Verb == "GET") { + m_Table = target; + } else { + m_Verb = "ERROR"; + m_ErrorCode = LivestatusErrorQuery; + m_ErrorMessage = "Unknown livestatus verb: " + m_Verb; + return; + } + + std::deque<Filter::Ptr> filters, stats; + std::deque<Aggregator::Ptr> aggregators; + + for (unsigned int i = 1; i < lines.size(); i++) { + line = lines[i]; + + size_t col_index = line.FindFirstOf(":"); + String header = line.SubStr(0, col_index); + String params; + + //OutputFormat:json or OutputFormat: json + if (line.GetLength() > col_index + 1) + params = line.SubStr(col_index + 1).Trim(); + + if (header == "ResponseHeader") + m_ResponseHeader = params; + else if (header == "OutputFormat") + m_OutputFormat = params; + else if (header == "KeepAlive") + m_KeepAlive = (params == "on"); + else if (header == "Columns") { + m_ColumnHeaders = false; // Might be explicitly re-enabled later on + m_Columns = params.Split(" "); + } else if (header == "Separators") { + std::vector<String> separators = params.Split(" "); + + /* ugly ascii long to char conversion, but works */ + if (separators.size() > 0) + m_Separators[0] = String(1, static_cast<char>(Convert::ToLong(separators[0]))); + if (separators.size() > 1) + m_Separators[1] = String(1, static_cast<char>(Convert::ToLong(separators[1]))); + if (separators.size() > 2) + m_Separators[2] = String(1, static_cast<char>(Convert::ToLong(separators[2]))); + if (separators.size() > 3) + m_Separators[3] = String(1, static_cast<char>(Convert::ToLong(separators[3]))); + } else if (header == "ColumnHeaders") + m_ColumnHeaders = (params == "on"); + else if (header == "Limit") + m_Limit = Convert::ToLong(params); + else if (header == "Filter") { + Filter::Ptr filter = ParseFilter(params, m_LogTimeFrom, m_LogTimeUntil); + + if (!filter) { + m_Verb = "ERROR"; + m_ErrorCode = LivestatusErrorQuery; + m_ErrorMessage = "Invalid filter specification: " + line; + return; + } + + filters.push_back(filter); + } else if (header == "Stats") { + m_ColumnHeaders = false; // Might be explicitly re-enabled later on + + std::vector<String> tokens = params.Split(" "); + + if (tokens.size() < 2) { + m_Verb = "ERROR"; + m_ErrorCode = LivestatusErrorQuery; + m_ErrorMessage = "Missing aggregator column name: " + line; + return; + } + + String aggregate_arg = tokens[0]; + String aggregate_attr = tokens[1]; + + Aggregator::Ptr aggregator; + Filter::Ptr filter; + + if (aggregate_arg == "sum") { + aggregator = new SumAggregator(aggregate_attr); + } else if (aggregate_arg == "min") { + aggregator = new MinAggregator(aggregate_attr); + } else if (aggregate_arg == "max") { + aggregator = new MaxAggregator(aggregate_attr); + } else if (aggregate_arg == "avg") { + aggregator = new AvgAggregator(aggregate_attr); + } else if (aggregate_arg == "std") { + aggregator = new StdAggregator(aggregate_attr); + } else if (aggregate_arg == "suminv") { + aggregator = new InvSumAggregator(aggregate_attr); + } else if (aggregate_arg == "avginv") { + aggregator = new InvAvgAggregator(aggregate_attr); + } else { + filter = ParseFilter(params, m_LogTimeFrom, m_LogTimeUntil); + + if (!filter) { + m_Verb = "ERROR"; + m_ErrorCode = LivestatusErrorQuery; + m_ErrorMessage = "Invalid filter specification: " + line; + return; + } + + aggregator = new CountAggregator(); + } + + aggregator->SetFilter(filter); + aggregators.push_back(aggregator); + + stats.push_back(filter); + } else if (header == "Or" || header == "And" || header == "StatsOr" || header == "StatsAnd") { + std::deque<Filter::Ptr>& deq = (header == "Or" || header == "And") ? filters : stats; + + unsigned int num = Convert::ToLong(params); + CombinerFilter::Ptr filter; + + if (header == "Or" || header == "StatsOr") { + filter = new OrFilter(); + Log(LogDebug, "LivestatusQuery") + << "Add OR filter for " << params << " column(s). " << deq.size() << " filters available."; + } else { + filter = new AndFilter(); + Log(LogDebug, "LivestatusQuery") + << "Add AND filter for " << params << " column(s). " << deq.size() << " filters available."; + } + + if (num > deq.size()) { + m_Verb = "ERROR"; + m_ErrorCode = 451; + m_ErrorMessage = "Or/StatsOr is referencing " + Convert::ToString(num) + " filters; stack only contains " + Convert::ToString(static_cast<long>(deq.size())) + " filters"; + return; + } + + while (num > 0 && num--) { + filter->AddSubFilter(deq.back()); + Log(LogDebug, "LivestatusQuery") + << "Add " << num << " filter."; + deq.pop_back(); + if (&deq == &stats) + aggregators.pop_back(); + } + + deq.emplace_back(filter); + if (&deq == &stats) { + Aggregator::Ptr aggregator = new CountAggregator(); + aggregator->SetFilter(filter); + aggregators.push_back(aggregator); + } + } else if (header == "Negate" || header == "StatsNegate") { + std::deque<Filter::Ptr>& deq = (header == "Negate") ? filters : stats; + + if (deq.empty()) { + m_Verb = "ERROR"; + m_ErrorCode = 451; + m_ErrorMessage = "Negate/StatsNegate used, however the filter stack is empty"; + return; + } + + Filter::Ptr filter = deq.back(); + deq.pop_back(); + + if (!filter) { + m_Verb = "ERROR"; + m_ErrorCode = 451; + m_ErrorMessage = "Negate/StatsNegate used, however last stats doesn't have a filter"; + return; + } + + deq.push_back(new NegateFilter(filter)); + + if (deq == stats) { + Aggregator::Ptr aggregator = aggregators.back(); + aggregator->SetFilter(filter); + } + } + } + + /* Combine all top-level filters into a single filter. */ + AndFilter::Ptr top_filter = new AndFilter(); + + for (const Filter::Ptr& filter : filters) { + top_filter->AddSubFilter(filter); + } + + m_Filter = top_filter; + m_Aggregators.swap(aggregators); +} + +int LivestatusQuery::GetExternalCommands() +{ + std::unique_lock<std::mutex> lock(l_QueryMutex); + + return l_ExternalCommands; +} + +Filter::Ptr LivestatusQuery::ParseFilter(const String& params, unsigned long& from, unsigned long& until) +{ + /* + * time >= 1382696656 + * type = SERVICE FLAPPING ALERT + */ + std::vector<String> tokens; + size_t sp_index; + String temp_buffer = params; + + /* extract attr and op */ + for (int i = 0; i < 2; i++) { + sp_index = temp_buffer.FindFirstOf(" "); + + /* check if this is the last argument */ + if (sp_index == String::NPos) { + /* 'attr op' or 'attr op val' is valid */ + if (i < 1) + BOOST_THROW_EXCEPTION(std::runtime_error("Livestatus filter '" + params + "' does not contain all required fields.")); + + break; + } + + tokens.emplace_back(temp_buffer.SubStr(0, sp_index)); + temp_buffer = temp_buffer.SubStr(sp_index + 1); + } + + /* add the rest as value */ + tokens.emplace_back(std::move(temp_buffer)); + + if (tokens.size() == 2) + tokens.emplace_back(""); + + if (tokens.size() < 3) + return nullptr; + + bool negate = false; + String attr = tokens[0]; + String op = tokens[1]; + String val = tokens[2]; + + if (op == "!=") { + op = "="; + negate = true; + } else if (op == "!~") { + op = "~"; + negate = true; + } else if (op == "!=~") { + op = "=~"; + negate = true; + } else if (op == "!~~") { + op = "~~"; + negate = true; + } + + Filter::Ptr filter = new AttributeFilter(attr, op, val); + + if (negate) + filter = new NegateFilter(filter); + + /* pre-filter log time duration */ + if (attr == "time") { + if (op == "<" || op == "<=") { + until = Convert::ToLong(val); + } else if (op == ">" || op == ">=") { + from = Convert::ToLong(val); + } + } + + Log(LogDebug, "LivestatusQuery") + << "Parsed filter with attr: '" << attr << "' op: '" << op << "' val: '" << val << "'."; + + return filter; +} + +void LivestatusQuery::BeginResultSet(std::ostream& fp) const +{ + if (m_OutputFormat == "json" || m_OutputFormat == "python") + fp << "["; +} + +void LivestatusQuery::EndResultSet(std::ostream& fp) const +{ + if (m_OutputFormat == "json" || m_OutputFormat == "python") + fp << "]"; +} + +void LivestatusQuery::AppendResultRow(std::ostream& fp, const Array::Ptr& row, bool& first_row) const +{ + if (m_OutputFormat == "csv") { + bool first = true; + + ObjectLock rlock(row); + for (const Value& value : row) { + if (first) + first = false; + else + fp << m_Separators[1]; + + if (value.IsObjectType<Array>()) + PrintCsvArray(fp, value, 0); + else + fp << value; + } + + fp << m_Separators[0]; + } else if (m_OutputFormat == "json") { + if (!first_row) + fp << ", "; + + fp << JsonEncode(row); + } else if (m_OutputFormat == "python") { + if (!first_row) + fp << ", "; + + PrintPythonArray(fp, row); + } + + first_row = false; +} + +void LivestatusQuery::PrintCsvArray(std::ostream& fp, const Array::Ptr& array, int level) const +{ + bool first = true; + + ObjectLock olock(array); + for (const Value& value : array) { + if (first) + first = false; + else + fp << ((level == 0) ? m_Separators[2] : m_Separators[3]); + + if (value.IsObjectType<Array>()) + PrintCsvArray(fp, value, level + 1); + else if (value.IsBoolean()) + fp << Convert::ToLong(value); + else + fp << value; + } +} + +void LivestatusQuery::PrintPythonArray(std::ostream& fp, const Array::Ptr& rs) const +{ + fp << "[ "; + + bool first = true; + + for (const Value& value : rs) { + if (first) + first = false; + else + fp << ", "; + + if (value.IsObjectType<Array>()) + PrintPythonArray(fp, value); + else if (value.IsNumber()) + fp << value; + else + fp << QuoteStringPython(value); + } + + fp << " ]"; +} + +String LivestatusQuery::QuoteStringPython(const String& str) { + String result = str; + boost::algorithm::replace_all(result, "\"", "\\\""); + return "r\"" + result + "\""; +} + +void LivestatusQuery::ExecuteGetHelper(const Stream::Ptr& stream) +{ + Log(LogNotice, "LivestatusQuery") + << "Table: " << m_Table; + + Table::Ptr table = Table::GetByName(m_Table, m_CompatLogPath, m_LogTimeFrom, m_LogTimeUntil); + + if (!table) { + SendResponse(stream, LivestatusErrorNotFound, "Table '" + m_Table + "' does not exist."); + + return; + } + + std::vector<LivestatusRowValue> objects = table->FilterRows(m_Filter, m_Limit); + std::vector<String> columns; + + if (m_Columns.size() > 0) + columns = m_Columns; + else + columns = table->GetColumnNames(); + + std::ostringstream result; + bool first_row = true; + BeginResultSet(result); + + if (m_Aggregators.empty()) { + typedef std::pair<String, Column> ColumnPair; + + std::vector<ColumnPair> column_objs; + column_objs.reserve(columns.size()); + + for (const String& columnName : columns) + column_objs.emplace_back(columnName, table->GetColumn(columnName)); + + ArrayData header; + + for (const LivestatusRowValue& object : objects) { + ArrayData row; + + row.reserve(column_objs.size()); + + for (const ColumnPair& cv : column_objs) { + if (m_ColumnHeaders) + header.push_back(cv.first); + + row.push_back(cv.second.ExtractValue(object.Row, object.GroupByType, object.GroupByObject)); + } + + if (m_ColumnHeaders) { + AppendResultRow(result, new Array(std::move(header)), first_row); + m_ColumnHeaders = false; + } + + AppendResultRow(result, new Array(std::move(row)), first_row); + } + } else { + std::map<std::vector<Value>, std::vector<AggregatorState *> > allStats; + + /* add aggregated stats */ + for (const LivestatusRowValue& object : objects) { + std::vector<Value> statsKey; + + for (const String& columnName : m_Columns) { + Column column = table->GetColumn(columnName); + statsKey.emplace_back(column.ExtractValue(object.Row, object.GroupByType, object.GroupByObject)); + } + + auto it = allStats.find(statsKey); + + if (it == allStats.end()) { + std::vector<AggregatorState *> newStats(m_Aggregators.size(), nullptr); + it = allStats.insert(std::make_pair(statsKey, newStats)).first; + } + + auto& stats = it->second; + + int index = 0; + + for (const Aggregator::Ptr& aggregator : m_Aggregators) { + aggregator->Apply(table, object.Row, &stats[index]); + index++; + } + } + + /* add column headers both for raw and aggregated data */ + if (m_ColumnHeaders) { + ArrayData header; + + for (const String& columnName : m_Columns) { + header.push_back(columnName); + } + + for (size_t i = 1; i <= m_Aggregators.size(); i++) { + header.push_back("stats_" + Convert::ToString(i)); + } + + AppendResultRow(result, new Array(std::move(header)), first_row); + } + + for (const auto& kv : allStats) { + ArrayData row; + + row.reserve(m_Columns.size() + m_Aggregators.size()); + + for (const Value& keyPart : kv.first) { + row.push_back(keyPart); + } + + auto& stats = kv.second; + + for (size_t i = 0; i < m_Aggregators.size(); i++) + row.push_back(m_Aggregators[i]->GetResultAndFreeState(stats[i])); + + AppendResultRow(result, new Array(std::move(row)), first_row); + } + + /* add a bogus zero value if aggregated is empty*/ + if (allStats.empty()) { + ArrayData row; + + row.reserve(m_Aggregators.size()); + + for (size_t i = 1; i <= m_Aggregators.size(); i++) { + row.push_back(0); + } + + AppendResultRow(result, new Array(std::move(row)), first_row); + } + } + + EndResultSet(result); + + SendResponse(stream, LivestatusErrorOK, result.str()); +} + +void LivestatusQuery::ExecuteCommandHelper(const Stream::Ptr& stream) +{ + { + std::unique_lock<std::mutex> lock(l_QueryMutex); + + l_ExternalCommands++; + } + + Log(LogNotice, "LivestatusQuery") + << "Executing command: " << m_Command; + ExternalCommandProcessor::Execute(m_Command); + SendResponse(stream, LivestatusErrorOK, ""); +} + +void LivestatusQuery::ExecuteErrorHelper(const Stream::Ptr& stream) +{ + Log(LogDebug, "LivestatusQuery") + << "ERROR: Code: '" << m_ErrorCode << "' Message: '" << m_ErrorMessage << "'."; + SendResponse(stream, m_ErrorCode, m_ErrorMessage); +} + +void LivestatusQuery::SendResponse(const Stream::Ptr& stream, int code, const String& data) +{ + if (m_ResponseHeader == "fixed16") + PrintFixed16(stream, code, data); + + if (m_ResponseHeader == "fixed16" || code == LivestatusErrorOK) { + try { + stream->Write(data.CStr(), data.GetLength()); + } catch (const std::exception&) { + Log(LogCritical, "LivestatusQuery", "Cannot write query response to socket."); + } + } +} + +void LivestatusQuery::PrintFixed16(const Stream::Ptr& stream, int code, const String& data) +{ + ASSERT(code >= 100 && code <= 999); + + String sCode = Convert::ToString(code); + String sLength = Convert::ToString(static_cast<long>(data.GetLength())); + + String header = sCode + String(16 - 3 - sLength.GetLength() - 1, ' ') + sLength + m_Separators[0]; + + try { + stream->Write(header.CStr(), header.GetLength()); + } catch (const std::exception&) { + Log(LogCritical, "LivestatusQuery", "Cannot write to TCP socket."); + } +} + +bool LivestatusQuery::Execute(const Stream::Ptr& stream) +{ + try { + Log(LogNotice, "LivestatusQuery") + << "Executing livestatus query: " << m_Verb; + + if (m_Verb == "GET") + ExecuteGetHelper(stream); + else if (m_Verb == "COMMAND") + ExecuteCommandHelper(stream); + else if (m_Verb == "ERROR") + ExecuteErrorHelper(stream); + else + BOOST_THROW_EXCEPTION(std::runtime_error("Invalid livestatus query verb.")); + } catch (const std::exception& ex) { + SendResponse(stream, LivestatusErrorQuery, DiagnosticInformation(ex)); + } + + if (!m_KeepAlive) { + stream->Close(); + return false; + } + + return true; +} |