mirror of
https://github.com/Icinga/icinga2.git
synced 2026-07-09 00:50:59 -04:00
503e23e723
Since perfdata is set once when a check result is created and never changed again, locking this is unnecessary. This avoids components unnecessarily waiting on each other when processing perfdata. This fixes the locking cascade observed sometimes when the perfdata writer work queue blocks, where it extends to a lock on the entire check result eventually, affecting even more components.
649 lines
21 KiB
C++
649 lines
21 KiB
C++
// SPDX-FileCopyrightText: 2012 Icinga GmbH <https://icinga.com>
|
|
// SPDX-License-Identifier: GPL-2.0-or-later
|
|
|
|
#include "perfdata/elasticsearchwriter.hpp"
|
|
#include "base/defer.hpp"
|
|
#include "perfdata/elasticsearchwriter-ti.cpp"
|
|
#include "remote/url.hpp"
|
|
#include "icinga/compatutility.hpp"
|
|
#include "icinga/service.hpp"
|
|
#include "icinga/macroprocessor.hpp"
|
|
#include "icinga/checkcommand.hpp"
|
|
#include "base/application.hpp"
|
|
#include "base/stream.hpp"
|
|
#include "base/base64.hpp"
|
|
#include "base/json.hpp"
|
|
#include "base/utility.hpp"
|
|
#include "base/perfdatavalue.hpp"
|
|
#include "base/exception.hpp"
|
|
#include "base/statsfunction.hpp"
|
|
#include <boost/algorithm/string.hpp>
|
|
#include <string>
|
|
#include <utility>
|
|
|
|
using namespace icinga;
|
|
|
|
REGISTER_TYPE(ElasticsearchWriter);
|
|
|
|
REGISTER_STATSFUNCTION(ElasticsearchWriter, &ElasticsearchWriter::StatsFunc);
|
|
|
|
void ElasticsearchWriter::OnConfigLoaded()
|
|
{
|
|
ObjectImpl<ElasticsearchWriter>::OnConfigLoaded();
|
|
|
|
m_WorkQueue.SetName("ElasticsearchWriter, " + GetName());
|
|
|
|
if (!GetEnableHa()) {
|
|
Log(LogDebug, "ElasticsearchWriter")
|
|
<< "HA functionality disabled. Won't pause connection: " << GetName();
|
|
|
|
SetHAMode(HARunEverywhere);
|
|
} else {
|
|
SetHAMode(HARunOnce);
|
|
}
|
|
}
|
|
|
|
void ElasticsearchWriter::OnAllConfigLoaded()
|
|
{
|
|
ObjectImpl<ElasticsearchWriter>::OnAllConfigLoaded();
|
|
|
|
Log(LogWarning, "ElasticsearchWriter")
|
|
<< "This feature is DEPRECATED and will be removed in v2.18.";
|
|
}
|
|
|
|
void ElasticsearchWriter::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
|
|
{
|
|
DictionaryData nodes;
|
|
|
|
for (const ElasticsearchWriter::Ptr& elasticsearchwriter : ConfigType::GetObjectsByType<ElasticsearchWriter>()) {
|
|
size_t workQueueItems = elasticsearchwriter->m_WorkQueue.GetLength();
|
|
double workQueueItemRate = elasticsearchwriter->m_WorkQueue.GetTaskCount(60) / 60.0;
|
|
|
|
nodes.emplace_back(elasticsearchwriter->GetName(), new Dictionary({
|
|
{ "work_queue_items", workQueueItems },
|
|
{ "work_queue_item_rate", workQueueItemRate }
|
|
}));
|
|
|
|
perfdata->Add(new PerfdataValue("elasticsearchwriter_" + elasticsearchwriter->GetName() + "_work_queue_items", workQueueItems));
|
|
perfdata->Add(new PerfdataValue("elasticsearchwriter_" + elasticsearchwriter->GetName() + "_work_queue_item_rate", workQueueItemRate));
|
|
}
|
|
|
|
status->Set("elasticsearchwriter", new Dictionary(std::move(nodes)));
|
|
}
|
|
|
|
void ElasticsearchWriter::Start(bool runtimeCreated)
|
|
{
|
|
ObjectImpl::Start(runtimeCreated);
|
|
|
|
if (GetEnableTls()) {
|
|
try {
|
|
m_SslContext = MakeAsioSslContext(GetCertPath(), GetKeyPath(), GetCaPath());
|
|
} catch (const std::exception& ex) {
|
|
Log(LogCritical, "ElasticsearchWriter")
|
|
<< "Unable to create SSL context: " << ex.what();
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
|
|
void ElasticsearchWriter::Resume()
|
|
{
|
|
ObjectImpl<ElasticsearchWriter>::Resume();
|
|
|
|
Log(LogInformation, "ElasticsearchWriter")
|
|
<< "'" << GetName() << "' resumed.";
|
|
|
|
m_WorkQueue.SetExceptionCallback([this](boost::exception_ptr exp) { ExceptionHandler(std::move(exp)); });
|
|
|
|
/* Setup timer for periodically flushing m_DataBuffer */
|
|
m_FlushTimer = Timer::Create();
|
|
m_FlushTimer->SetInterval(GetFlushInterval());
|
|
m_FlushTimer->OnTimerExpired.connect([this](const Timer * const&) { FlushTimeout(); });
|
|
m_FlushTimer->Start();
|
|
m_FlushTimer->Reschedule(0);
|
|
|
|
m_Connection = new PerfdataWriterConnection{this, GetHost(), GetPort(), m_SslContext, !GetInsecureNoverify()};
|
|
|
|
/* Register for new metrics. */
|
|
m_HandleCheckResults = Checkable::OnNewCheckResult.connect([this](const Checkable::Ptr& checkable,
|
|
const CheckResult::Ptr& cr, const MessageOrigin::Ptr&) {
|
|
CheckResultHandler(checkable, cr);
|
|
});
|
|
m_HandleStateChanges = Checkable::OnStateChange.connect([this](const Checkable::Ptr& checkable,
|
|
const CheckResult::Ptr& cr, StateType, const MessageOrigin::Ptr&) {
|
|
StateChangeHandler(checkable, cr);
|
|
});
|
|
m_HandleNotifications = Checkable::OnNotificationSentToAllUsers.connect([this](const Notification::Ptr&,
|
|
const Checkable::Ptr& checkable, const std::set<User::Ptr>& users, const NotificationType& type,
|
|
const CheckResult::Ptr& cr, const String& author, const String& text, const MessageOrigin::Ptr&) {
|
|
NotificationSentToAllUsersHandler(checkable, users, type, cr, author, text);
|
|
});
|
|
}
|
|
|
|
/* Pause is equivalent to Stop, but with HA capabilities to resume at runtime. */
|
|
void ElasticsearchWriter::Pause()
|
|
{
|
|
m_HandleCheckResults.disconnect();
|
|
m_HandleStateChanges.disconnect();
|
|
m_HandleNotifications.disconnect();
|
|
|
|
m_FlushTimer->Stop(true);
|
|
|
|
std::promise<void> queueDonePromise;
|
|
m_WorkQueue.Enqueue([&]() {
|
|
Flush();
|
|
queueDonePromise.set_value();
|
|
}, PriorityLow);
|
|
|
|
auto timeout = std::chrono::duration<double>{GetDisconnectTimeout()};
|
|
m_Connection->CancelAfterTimeout(queueDonePromise.get_future(), timeout);
|
|
|
|
m_WorkQueue.Join();
|
|
|
|
Log(LogInformation, "ElasticsearchWriter")
|
|
<< "'" << GetName() << "' paused.";
|
|
|
|
ObjectImpl<ElasticsearchWriter>::Pause();
|
|
}
|
|
|
|
void ElasticsearchWriter::AddTemplateTags(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
|
|
{
|
|
Host::Ptr host;
|
|
Service::Ptr service;
|
|
tie(host, service) = GetHostService(checkable);
|
|
|
|
Dictionary::Ptr tmpl = service ? GetServiceTagsTemplate() : GetHostTagsTemplate();
|
|
|
|
if (tmpl) {
|
|
MacroProcessor::ResolverList resolvers;
|
|
resolvers.emplace_back("host", host);
|
|
if (service) {
|
|
resolvers.emplace_back("service", service);
|
|
}
|
|
|
|
ObjectLock olock(tmpl);
|
|
for (const Dictionary::Pair& pair : tmpl) {
|
|
String missingMacro;
|
|
Value value = MacroProcessor::ResolveMacros(pair.second, resolvers, cr, &missingMacro);
|
|
|
|
if (missingMacro.IsEmpty()) {
|
|
fields->Set(pair.first, value);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void ElasticsearchWriter::AddCheckResult(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
|
|
{
|
|
String prefix = "check_result.";
|
|
|
|
fields->Set(prefix + "output", cr->GetOutput());
|
|
fields->Set(prefix + "check_source", cr->GetCheckSource());
|
|
fields->Set(prefix + "exit_status", cr->GetExitStatus());
|
|
fields->Set(prefix + "command", cr->GetCommand());
|
|
fields->Set(prefix + "state", cr->GetState());
|
|
fields->Set(prefix + "vars_before", cr->GetVarsBefore());
|
|
fields->Set(prefix + "vars_after", cr->GetVarsAfter());
|
|
|
|
fields->Set(prefix + "execution_start", FormatTimestamp(cr->GetExecutionStart()));
|
|
fields->Set(prefix + "execution_end", FormatTimestamp(cr->GetExecutionEnd()));
|
|
fields->Set(prefix + "schedule_start", FormatTimestamp(cr->GetScheduleStart()));
|
|
fields->Set(prefix + "schedule_end", FormatTimestamp(cr->GetScheduleEnd()));
|
|
|
|
/* Add extra calculated field. */
|
|
fields->Set(prefix + "latency", cr->CalculateLatency());
|
|
fields->Set(prefix + "execution_time", cr->CalculateExecutionTime());
|
|
|
|
if (!GetEnableSendPerfdata())
|
|
return;
|
|
|
|
Array::Ptr perfdata = cr->GetPerformanceData();
|
|
|
|
CheckCommand::Ptr checkCommand = checkable->GetCheckCommand();
|
|
|
|
if (perfdata) {
|
|
for (const Value& val : perfdata) {
|
|
PerfdataValue::Ptr pdv;
|
|
|
|
if (val.IsObjectType<PerfdataValue>())
|
|
pdv = val;
|
|
else {
|
|
try {
|
|
pdv = PerfdataValue::Parse(val);
|
|
} catch (const std::exception&) {
|
|
Log(LogWarning, "ElasticsearchWriter")
|
|
<< "Ignoring invalid perfdata for checkable '"
|
|
<< checkable->GetName() << "' and command '"
|
|
<< checkCommand->GetName() << "' with value: " << val;
|
|
continue;
|
|
}
|
|
}
|
|
|
|
String escapedKey = pdv->GetLabel();
|
|
boost::replace_all(escapedKey, " ", "_");
|
|
boost::replace_all(escapedKey, ".", "_");
|
|
boost::replace_all(escapedKey, "\\", "_");
|
|
boost::algorithm::replace_all(escapedKey, "::", ".");
|
|
|
|
String perfdataPrefix = prefix + "perfdata." + escapedKey;
|
|
|
|
fields->Set(perfdataPrefix + ".value", pdv->GetValue());
|
|
|
|
if (!pdv->GetMin().IsEmpty())
|
|
fields->Set(perfdataPrefix + ".min", pdv->GetMin());
|
|
if (!pdv->GetMax().IsEmpty())
|
|
fields->Set(perfdataPrefix + ".max", pdv->GetMax());
|
|
if (!pdv->GetWarn().IsEmpty())
|
|
fields->Set(perfdataPrefix + ".warn", pdv->GetWarn());
|
|
if (!pdv->GetCrit().IsEmpty())
|
|
fields->Set(perfdataPrefix + ".crit", pdv->GetCrit());
|
|
|
|
if (!pdv->GetUnit().IsEmpty())
|
|
fields->Set(perfdataPrefix + ".unit", pdv->GetUnit());
|
|
}
|
|
}
|
|
}
|
|
|
|
void ElasticsearchWriter::CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
|
|
{
|
|
if (IsPaused())
|
|
return;
|
|
|
|
if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata())
|
|
return;
|
|
|
|
Host::Ptr host;
|
|
Service::Ptr service;
|
|
tie(host, service) = GetHostService(checkable);
|
|
|
|
Dictionary::Ptr fields = new Dictionary();
|
|
|
|
if (service) {
|
|
fields->Set("service", service->GetShortName());
|
|
fields->Set("state", service->GetState());
|
|
fields->Set("last_state", service->GetLastState());
|
|
fields->Set("last_hard_state", service->GetLastHardState());
|
|
} else {
|
|
fields->Set("state", host->GetState());
|
|
fields->Set("last_state", host->GetLastState());
|
|
fields->Set("last_hard_state", host->GetLastHardState());
|
|
}
|
|
|
|
fields->Set("host", host->GetName());
|
|
fields->Set("state_type", checkable->GetStateType());
|
|
|
|
fields->Set("current_check_attempt", checkable->GetCheckAttempt());
|
|
fields->Set("max_check_attempts", checkable->GetMaxCheckAttempts());
|
|
|
|
fields->Set("reachable", checkable->IsReachable());
|
|
fields->Set("check_command", checkable->GetCheckCommand()->GetName());
|
|
|
|
AddTemplateTags(fields, checkable, cr);
|
|
|
|
m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
|
|
if (m_Connection->IsStopped()) {
|
|
return;
|
|
}
|
|
|
|
CONTEXT("Elasticwriter processing check result for '" << checkable->GetName() << "'");
|
|
|
|
AddCheckResult(fields, checkable, cr);
|
|
|
|
Enqueue(checkable, "checkresult", fields, cr->GetExecutionEnd());
|
|
});
|
|
}
|
|
|
|
void ElasticsearchWriter::StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
|
|
{
|
|
if (IsPaused())
|
|
return;
|
|
|
|
Host::Ptr host;
|
|
Service::Ptr service;
|
|
tie(host, service) = GetHostService(checkable);
|
|
|
|
Dictionary::Ptr fields = new Dictionary();
|
|
|
|
fields->Set("current_check_attempt", checkable->GetCheckAttempt());
|
|
fields->Set("max_check_attempts", checkable->GetMaxCheckAttempts());
|
|
fields->Set("host", host->GetName());
|
|
|
|
if (service) {
|
|
fields->Set("service", service->GetShortName());
|
|
fields->Set("state", service->GetState());
|
|
fields->Set("last_state", service->GetLastState());
|
|
fields->Set("last_hard_state", service->GetLastHardState());
|
|
} else {
|
|
fields->Set("state", host->GetState());
|
|
fields->Set("last_state", host->GetLastState());
|
|
fields->Set("last_hard_state", host->GetLastHardState());
|
|
}
|
|
|
|
fields->Set("check_command", checkable->GetCheckCommand()->GetName());
|
|
|
|
AddTemplateTags(fields, checkable, cr);
|
|
|
|
m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
|
|
if (m_Connection->IsStopped()) {
|
|
return;
|
|
}
|
|
|
|
CONTEXT("Elasticwriter processing state change '" << checkable->GetName() << "'");
|
|
|
|
AddCheckResult(fields, checkable, cr);
|
|
|
|
Enqueue(checkable, "statechange", fields, cr->GetExecutionEnd());
|
|
});
|
|
}
|
|
|
|
void ElasticsearchWriter::NotificationSentToAllUsersHandler(const Checkable::Ptr& checkable, const std::set<User::Ptr>& users,
|
|
NotificationType type, const CheckResult::Ptr& cr, const String& author, const String& text)
|
|
{
|
|
if (IsPaused())
|
|
return;
|
|
|
|
Host::Ptr host;
|
|
Service::Ptr service;
|
|
tie(host, service) = GetHostService(checkable);
|
|
|
|
String notificationTypeString = Notification::NotificationTypeToStringCompat(type); //TODO: Change that to our own types.
|
|
|
|
Dictionary::Ptr fields = new Dictionary();
|
|
|
|
if (service) {
|
|
fields->Set("service", service->GetShortName());
|
|
fields->Set("state", service->GetState());
|
|
fields->Set("last_state", service->GetLastState());
|
|
fields->Set("last_hard_state", service->GetLastHardState());
|
|
} else {
|
|
fields->Set("state", host->GetState());
|
|
fields->Set("last_state", host->GetLastState());
|
|
fields->Set("last_hard_state", host->GetLastHardState());
|
|
}
|
|
|
|
fields->Set("host", host->GetName());
|
|
|
|
ArrayData userNames;
|
|
|
|
for (const User::Ptr& user : users) {
|
|
userNames.push_back(user->GetName());
|
|
}
|
|
|
|
fields->Set("users", new Array(std::move(userNames)));
|
|
fields->Set("notification_type", notificationTypeString);
|
|
fields->Set("author", author);
|
|
fields->Set("text", text);
|
|
fields->Set("check_command", checkable->GetCheckCommand()->GetName());
|
|
|
|
AddTemplateTags(fields, checkable, cr);
|
|
|
|
m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
|
|
if (m_Connection->IsStopped()) {
|
|
return;
|
|
}
|
|
|
|
CONTEXT("Elasticwriter processing notification to all users '" << checkable->GetName() << "'");
|
|
|
|
Log(LogDebug, "ElasticsearchWriter")
|
|
<< "Processing notification for '" << checkable->GetName() << "'";
|
|
|
|
double ts = Utility::GetTime();
|
|
|
|
if (cr) {
|
|
AddCheckResult(fields, checkable, cr);
|
|
ts = cr->GetExecutionEnd();
|
|
}
|
|
|
|
Enqueue(checkable, "notification", fields, ts);
|
|
});
|
|
}
|
|
|
|
void ElasticsearchWriter::Enqueue(const Checkable::Ptr& checkable, const String& type,
|
|
const Dictionary::Ptr& fields, double ts)
|
|
{
|
|
AssertOnWorkQueue();
|
|
|
|
/* Format the timestamps to dynamically select the date datatype inside the index. */
|
|
fields->Set("@timestamp", FormatTimestamp(ts));
|
|
fields->Set("timestamp", FormatTimestamp(ts));
|
|
fields->Set("type", "icinga2.event." + type);
|
|
|
|
/* Every payload needs a line describing the index.
|
|
* We do it this way to avoid problems with a near full queue.
|
|
*/
|
|
String indexBody = "{\"index\": {} }\n";
|
|
String fieldsBody = JsonEncode(fields);
|
|
|
|
Log(LogDebug, "ElasticsearchWriter")
|
|
<< "Checkable '" << checkable->GetName() << "' adds to metric list: '" << fieldsBody << "'.";
|
|
|
|
m_DataBuffer.emplace_back(indexBody + fieldsBody);
|
|
|
|
/* Flush if we've buffered too much to prevent excessive memory use. */
|
|
if (static_cast<int>(m_DataBuffer.size()) >= GetFlushThreshold()) {
|
|
Log(LogDebug, "ElasticsearchWriter")
|
|
<< "Data buffer overflow writing " << m_DataBuffer.size() << " data points";
|
|
Flush();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Queues a Flush on the work-queue if there isn't one queued already.
|
|
*/
|
|
void ElasticsearchWriter::FlushTimeout()
|
|
{
|
|
if (m_FlushTimerInQueue.exchange(true, std::memory_order_relaxed)) {
|
|
return;
|
|
}
|
|
|
|
m_WorkQueue.Enqueue([&]() {
|
|
Defer resetFlushTimer{
|
|
[&]() { m_FlushTimerInQueue.store(false, std::memory_order_relaxed); }
|
|
};
|
|
Flush();
|
|
});
|
|
}
|
|
|
|
void ElasticsearchWriter::Flush()
|
|
{
|
|
/* Flush can be called from 1) Timeout 2) Threshold 3) on shutdown/reload. */
|
|
if (m_DataBuffer.empty())
|
|
return;
|
|
|
|
/* Ensure you hold a lock against m_DataBuffer so that things
|
|
* don't go missing after creating the body and clearing the buffer.
|
|
*/
|
|
String body = boost::algorithm::join(m_DataBuffer, "\n");
|
|
m_DataBuffer.clear();
|
|
|
|
/* Elasticsearch 6.x requires a new line. This is compatible to 5.x.
|
|
* Tested with 6.0.0 and 5.6.4.
|
|
*/
|
|
body += "\n";
|
|
|
|
SendRequest(body);
|
|
}
|
|
|
|
void ElasticsearchWriter::SendRequest(const String& body)
|
|
{
|
|
namespace beast = boost::beast;
|
|
namespace http = beast::http;
|
|
|
|
Url::Ptr url = new Url();
|
|
|
|
url->SetScheme(GetEnableTls() ? "https" : "http");
|
|
url->SetHost(GetHost());
|
|
url->SetPort(GetPort());
|
|
|
|
std::vector<String> path;
|
|
|
|
/* Specify the index path. Best practice is a daily rotation.
|
|
* Example: http://localhost:9200/icinga2-2017.09.11?pretty=1
|
|
*/
|
|
path.emplace_back(GetIndex() + "-" + Utility::FormatDateTime("%Y.%m.%d", Utility::GetTime()));
|
|
|
|
/* Use the bulk message format. */
|
|
path.emplace_back("_bulk");
|
|
|
|
url->SetPath(path);
|
|
|
|
http::request<http::string_body> request (http::verb::post, std::string(url->Format(true)), 10);
|
|
|
|
request.set(http::field::user_agent, "Icinga/" + Application::GetAppVersion());
|
|
request.set(http::field::host, url->GetHost() + ":" + url->GetPort());
|
|
|
|
/* Specify required headers by Elasticsearch. */
|
|
request.set(http::field::accept, "application/json");
|
|
|
|
/* Use application/x-ndjson for bulk streams. While ES
|
|
* is able to handle application/json, the newline separator
|
|
* causes problems with Logstash (#6609).
|
|
*/
|
|
request.set(http::field::content_type, "application/x-ndjson");
|
|
|
|
/* Send authentication if configured. */
|
|
String username = GetUsername();
|
|
String password = GetPassword();
|
|
|
|
if (!username.IsEmpty() && !password.IsEmpty())
|
|
request.set(http::field::authorization, "Basic " + Base64::Encode(username + ":" + password));
|
|
|
|
request.body() = body;
|
|
request.content_length(request.body().size());
|
|
|
|
/* Don't log the request body to debug log, this is already done above. */
|
|
Log(LogDebug, "ElasticsearchWriter")
|
|
<< "Sending " << request.method_string() << " request" << ((!username.IsEmpty() && !password.IsEmpty()) ? " with basic auth" : "" )
|
|
<< " to '" << url->Format() << "'.";
|
|
|
|
decltype(m_Connection->Send(request)) response;
|
|
try {
|
|
response = m_Connection->Send(request);
|
|
} catch (const PerfdataWriterConnection::Stopped& ex) {
|
|
Log(LogDebug, "ElasticsearchWriter") << ex.what();
|
|
return;
|
|
}
|
|
|
|
if (response.result_int() > 299) {
|
|
if (response.result() == http::status::unauthorized) {
|
|
/* More verbose error logging with Elasticsearch is hidden behind a proxy. */
|
|
if (!username.IsEmpty() && !password.IsEmpty()) {
|
|
Log(LogCritical, "ElasticsearchWriter")
|
|
<< "401 Unauthorized. Please ensure that the user '" << username
|
|
<< "' is able to authenticate against the HTTP API/Proxy.";
|
|
} else {
|
|
Log(LogCritical, "ElasticsearchWriter")
|
|
<< "401 Unauthorized. The HTTP API requires authentication but no username/password has been configured.";
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
std::ostringstream msgbuf;
|
|
msgbuf << "Unexpected response code " << response.result_int() << " from URL '" << url->Format() << "'";
|
|
|
|
auto& contentType (response[http::field::content_type]);
|
|
|
|
if (contentType != "application/json" && contentType != "application/json; charset=utf-8") {
|
|
msgbuf << "; Unexpected Content-Type: '" << contentType << "'";
|
|
}
|
|
|
|
auto& body (response.body());
|
|
|
|
#ifdef I2_DEBUG
|
|
msgbuf << "; Response body: '" << body << "'";
|
|
#endif /* I2_DEBUG */
|
|
|
|
Dictionary::Ptr jsonResponse;
|
|
|
|
try {
|
|
jsonResponse = JsonDecode(body);
|
|
} catch (...) {
|
|
Log(LogWarning, "ElasticsearchWriter")
|
|
<< "Unable to parse JSON response:\n" << body;
|
|
return;
|
|
}
|
|
|
|
String error = jsonResponse->Get("error");
|
|
|
|
Log(LogCritical, "ElasticsearchWriter")
|
|
<< "Error: '" << error << "'. " << msgbuf.str();
|
|
}
|
|
}
|
|
|
|
void ElasticsearchWriter::AssertOnWorkQueue()
|
|
{
|
|
ASSERT(m_WorkQueue.IsWorkerThread());
|
|
}
|
|
|
|
void ElasticsearchWriter::ExceptionHandler(boost::exception_ptr exp)
|
|
{
|
|
Log(LogCritical, "ElasticsearchWriter", "Exception during Elastic operation: Verify that your backend is operational!");
|
|
|
|
Log(LogDebug, "ElasticsearchWriter")
|
|
<< "Exception during Elasticsearch operation: " << DiagnosticInformation(std::move(exp));
|
|
}
|
|
|
|
String ElasticsearchWriter::FormatTimestamp(double ts)
|
|
{
|
|
/* The date format must match the default dynamic date detection
|
|
* pattern in indexes. This enables applications like Kibana to
|
|
* detect a qualified timestamp index for time-series data.
|
|
*
|
|
* Example: 2017-09-11T10:56:21.463+0200
|
|
*
|
|
* References:
|
|
* https://www.elastic.co/guide/en/elasticsearch/reference/current/dynamic-field-mapping.html#date-detection
|
|
* https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-date-format.html
|
|
* https://www.elastic.co/guide/en/elasticsearch/reference/current/date.html
|
|
*/
|
|
auto milliSeconds = static_cast<int>((ts - static_cast<int>(ts)) * 1000);
|
|
|
|
return Utility::FormatDateTime("%Y-%m-%dT%H:%M:%S", ts) + "." + Convert::ToString(milliSeconds) + Utility::FormatDateTime("%z", ts);
|
|
}
|
|
|
|
void ElasticsearchWriter::ValidateHostTagsTemplate(const Lazy<Dictionary::Ptr>& lvalue, const ValidationUtils& utils)
|
|
{
|
|
ObjectImpl<ElasticsearchWriter>::ValidateHostTagsTemplate(lvalue, utils);
|
|
|
|
Dictionary::Ptr tags = lvalue();
|
|
if (tags) {
|
|
ObjectLock olock(tags);
|
|
for (const Dictionary::Pair& pair : tags) {
|
|
if (pair.second.IsObjectType<Array>()) {
|
|
Array::Ptr arrObject = pair.second;
|
|
ObjectLock arrLock(arrObject);
|
|
for (const Value& arrValue : arrObject) {
|
|
if (!MacroProcessor::ValidateMacroString(arrValue)) {
|
|
BOOST_THROW_EXCEPTION(ValidationError(this, { "host_tags_template", pair.first }, "Closing $ not found in macro format string '" + arrValue + "'."));
|
|
}
|
|
}
|
|
} else if (!MacroProcessor::ValidateMacroString(pair.second)) {
|
|
BOOST_THROW_EXCEPTION(ValidationError(this, { "host_tags_template", pair.first }, "Closing $ not found in macro format string '" + pair.second + "'."));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void ElasticsearchWriter::ValidateServiceTagsTemplate(const Lazy<Dictionary::Ptr>& lvalue, const ValidationUtils& utils)
|
|
{
|
|
ObjectImpl<ElasticsearchWriter>::ValidateServiceTagsTemplate(lvalue, utils);
|
|
|
|
Dictionary::Ptr tags = lvalue();
|
|
if (tags) {
|
|
ObjectLock olock(tags);
|
|
for (const Dictionary::Pair& pair : tags) {
|
|
if (pair.second.IsObjectType<Array>()) {
|
|
Array::Ptr arrObject = pair.second;
|
|
ObjectLock arrLock(arrObject);
|
|
for (const Value& arrValue : arrObject) {
|
|
if (!MacroProcessor::ValidateMacroString(arrValue)) {
|
|
BOOST_THROW_EXCEPTION(ValidationError(this, { "service_tags_template", pair.first }, "Closing $ not found in macro format string '" + arrValue + "'."));
|
|
}
|
|
}
|
|
} else if (!MacroProcessor::ValidateMacroString(pair.second)) {
|
|
BOOST_THROW_EXCEPTION(ValidationError(this, { "service_tags_template", pair.first }, "Closing $ not found in macro format string '" + pair.second + "'."));
|
|
}
|
|
}
|
|
}
|
|
}
|