tyler.durden/icinga2
1
0
Fork 0
mirror of https://github.com/Icinga/icinga2.git synced 2025-07-26 23:24:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #10420 from Icinga/bundled-perfdata-writers-fix

Browse Source 
Serialize fields before queueing them to the workqueue
This commit is contained in:
Julian Brost 2025-06-17 10:17:27 +02:00 committed by GitHub
commit 1aa62d4bb9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 204 additions and 317 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
lib/perfdata/elasticsearchwriter.cpp
View File

@ -101,13 +101,13 @@ void ElasticsearchWriter::Resume()
CheckResultHandler(checkable, cr); CheckResultHandler(checkable, cr);
}); });
m_HandleStateChanges = Checkable::OnStateChange.connect([this](const Checkable::Ptr& checkable, m_HandleStateChanges = Checkable::OnStateChange.connect([this](const Checkable::Ptr& checkable,
const CheckResult::Ptr& cr, StateType type, const MessageOrigin::Ptr&) { const CheckResult::Ptr& cr, StateType, const MessageOrigin::Ptr&) {
StateChangeHandler(checkable, cr, type); StateChangeHandler(checkable, cr);
}); });
m_HandleNotifications = Checkable::OnNotificationSentToAllUsers.connect([this](const Notification::Ptr& notification, m_HandleNotifications = Checkable::OnNotificationSentToAllUsers.connect([this](const Notification::Ptr&,
const Checkable::Ptr& checkable, const std::set<User::Ptr>& users, const NotificationType& type, 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&) { const CheckResult::Ptr& cr, const String& author, const String& text, const MessageOrigin::Ptr&) {
NotificationSentToAllUsersHandler(notification, checkable, users, type, cr, author, text); NotificationSentToAllUsersHandler(checkable, users, type, cr, author, text);
}); });
} }
@ -236,15 +236,6 @@ void ElasticsearchWriter::CheckResultHandler(const Checkable::Ptr& checkable, co
if (IsPaused()) if (IsPaused())
return; return;
m_WorkQueue.Enqueue([this, checkable, cr]() { InternalCheckResultHandler(checkable, cr); });
}
void ElasticsearchWriter::InternalCheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
{
AssertOnWorkQueue();
CONTEXT("Elasticwriter processing check result for '" << checkable->GetName() << "'");
if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata()) if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata())
return; return;
@ -272,38 +263,24 @@ void ElasticsearchWriter::InternalCheckResultHandler(const Checkable::Ptr& check
fields->Set("max_check_attempts", checkable->GetMaxCheckAttempts()); fields->Set("max_check_attempts", checkable->GetMaxCheckAttempts());
fields->Set("reachable", checkable->IsReachable()); fields->Set("reachable", checkable->IsReachable());
fields->Set("check_command", checkable->GetCheckCommand()->GetName());
CheckCommand::Ptr commandObj = checkable->GetCheckCommand();
if (commandObj)
fields->Set("check_command", commandObj->GetName());
double ts = Utility::GetTime();
if (cr) {
AddCheckResult(fields, checkable, cr);
ts = cr->GetExecutionEnd();
}
AddTemplateTags(fields, checkable, cr); AddTemplateTags(fields, checkable, cr);
Enqueue(checkable, "checkresult", fields, ts); m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
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, StateType type) void ElasticsearchWriter::StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
{ {
if (IsPaused()) if (IsPaused())
return; return;
m_WorkQueue.Enqueue([this, checkable, cr, type]() { StateChangeHandlerInternal(checkable, cr, type); });
}
void ElasticsearchWriter::StateChangeHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type)
{
AssertOnWorkQueue();
CONTEXT("Elasticwriter processing state change '" << checkable->GetName() << "'");
Host::Ptr host; Host::Ptr host;
Service::Ptr service; Service::Ptr service;
tie(host, service) = GetHostService(checkable); tie(host, service) = GetHostService(checkable);
@ -325,45 +302,24 @@ void ElasticsearchWriter::StateChangeHandlerInternal(const Checkable::Ptr& check
fields->Set("last_hard_state", host->GetLastHardState()); fields->Set("last_hard_state", host->GetLastHardState());
} }
CheckCommand::Ptr commandObj = checkable->GetCheckCommand(); fields->Set("check_command", checkable->GetCheckCommand()->GetName());
if (commandObj)
fields->Set("check_command", commandObj->GetName());
double ts = Utility::GetTime();
if (cr) {
AddCheckResult(fields, checkable, cr);
ts = cr->GetExecutionEnd();
}
AddTemplateTags(fields, checkable, cr); AddTemplateTags(fields, checkable, cr);
Enqueue(checkable, "statechange", fields, ts); m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
} CONTEXT("Elasticwriter processing state change '" << checkable->GetName() << "'");
void ElasticsearchWriter::NotificationSentToAllUsersHandler(const Notification::Ptr& notification, AddCheckResult(fields, checkable, cr);
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;
m_WorkQueue.Enqueue([this, notification, checkable, users, type, cr, author, text]() { Enqueue(checkable, "statechange", fields, cr->GetExecutionEnd());
NotificationSentToAllUsersHandlerInternal(notification, checkable, users, type, cr, author, text);
}); });
} }
void ElasticsearchWriter::NotificationSentToAllUsersHandlerInternal(const Notification::Ptr& notification, void ElasticsearchWriter::NotificationSentToAllUsersHandler(const Checkable::Ptr& checkable, const std::set<User::Ptr>& users,
const Checkable::Ptr& checkable, const std::set<User::Ptr>& users, NotificationType type, NotificationType type, const CheckResult::Ptr& cr, const String& author, const String& text)
const CheckResult::Ptr& cr, const String& author, const String& text)
{ {
AssertOnWorkQueue(); if (IsPaused())
return;
CONTEXT("Elasticwriter processing notification to all users '" << checkable->GetName() << "'");
Log(LogDebug, "ElasticsearchWriter")
<< "Processing notification for '" << checkable->GetName() << "'";
Host::Ptr host; Host::Ptr host;
Service::Ptr service; Service::Ptr service;
@ -396,11 +352,15 @@ void ElasticsearchWriter::NotificationSentToAllUsersHandlerInternal(const Notifi
fields->Set("notification_type", notificationTypeString); fields->Set("notification_type", notificationTypeString);
fields->Set("author", author); fields->Set("author", author);
fields->Set("text", text); fields->Set("text", text);
fields->Set("check_command", checkable->GetCheckCommand()->GetName());
CheckCommand::Ptr commandObj = checkable->GetCheckCommand(); AddTemplateTags(fields, checkable, cr);
if (commandObj) m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
fields->Set("check_command", commandObj->GetName()); CONTEXT("Elasticwriter processing notification to all users '" << checkable->GetName() << "'");
Log(LogDebug, "ElasticsearchWriter")
<< "Processing notification for '" << checkable->GetName() << "'";
double ts = Utility::GetTime(); double ts = Utility::GetTime();
@ -409,14 +369,15 @@ void ElasticsearchWriter::NotificationSentToAllUsersHandlerInternal(const Notifi
ts = cr->GetExecutionEnd(); ts = cr->GetExecutionEnd();
} }
AddTemplateTags(fields, checkable, cr);
Enqueue(checkable, "notification", fields, ts); Enqueue(checkable, "notification", fields, ts);
});
} }
void ElasticsearchWriter::Enqueue(const Checkable::Ptr& checkable, const String& type, void ElasticsearchWriter::Enqueue(const Checkable::Ptr& checkable, const String& type,
const Dictionary::Ptr& fields, double ts) const Dictionary::Ptr& fields, double ts)
{ {
AssertOnWorkQueue();
/* Atomically buffer the data point. */ /* Atomically buffer the data point. */
std::unique_lock<std::mutex> lock(m_DataBufferMutex); std::unique_lock<std::mutex> lock(m_DataBufferMutex);

12
lib/perfdata/elasticsearchwriter.hpp
View File

@ -42,16 +42,10 @@ private:
void AddCheckResult(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void AddCheckResult(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void AddTemplateTags(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void AddTemplateTags(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type); void StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void StateChangeHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type);
void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void InternalCheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void NotificationSentToAllUsersHandler(const Checkable::Ptr& checkable, const std::set<User::Ptr>& users,
void NotificationSentToAllUsersHandler(const Notification::Ptr& notification, NotificationType type, const CheckResult::Ptr& cr, const String& author, const String& text);
const Checkable::Ptr& checkable, const std::set<User::Ptr>& users, NotificationType type,
const CheckResult::Ptr& cr, const String& author, const String& text);
void NotificationSentToAllUsersHandlerInternal(const Notification::Ptr& notification,
const Checkable::Ptr& checkable, const std::set<User::Ptr>& users, NotificationType type,
const CheckResult::Ptr& cr, const String& author, const String& text);
void Enqueue(const Checkable::Ptr& checkable, const String& type, void Enqueue(const Checkable::Ptr& checkable, const String& type,
const Dictionary::Ptr& fields, double ts); const Dictionary::Ptr& fields, double ts);

106
lib/perfdata/gelfwriter.cpp
View File

@ -94,14 +94,14 @@ void GelfWriter::Resume()
const CheckResult::Ptr& cr, const MessageOrigin::Ptr&) { const CheckResult::Ptr& cr, const MessageOrigin::Ptr&) {
CheckResultHandler(checkable, cr); CheckResultHandler(checkable, cr);
}); });
m_HandleNotifications = Checkable::OnNotificationSentToUser.connect([this](const Notification::Ptr& notification, m_HandleNotifications = Checkable::OnNotificationSentToUser.connect([this](const Notification::Ptr&,
const Checkable::Ptr& checkable, const User::Ptr& user, const NotificationType& type, const CheckResult::Ptr& cr, const Checkable::Ptr& checkable, const User::Ptr&, const NotificationType& type, const CheckResult::Ptr& cr,
const String& author, const String& commentText, const String& commandName, const MessageOrigin::Ptr&) { const String& author, const String& commentText, const String& commandName, const MessageOrigin::Ptr&) {
NotificationToUserHandler(notification, checkable, user, type, cr, author, commentText, commandName); NotificationToUserHandler(checkable, type, cr, author, commentText, commandName);
}); });
m_HandleStateChanges = Checkable::OnStateChange.connect([this](const Checkable::Ptr& checkable, m_HandleStateChanges = Checkable::OnStateChange.connect([this](const Checkable::Ptr& checkable,
const CheckResult::Ptr& cr, StateType type, const MessageOrigin::Ptr&) { const CheckResult::Ptr& cr, StateType, const MessageOrigin::Ptr&) {
StateChangeHandler(checkable, cr, type); StateChangeHandler(checkable, cr);
}); });
} }
@ -268,18 +268,6 @@ void GelfWriter::CheckResultHandler(const Checkable::Ptr& checkable, const Check
if (IsPaused()) if (IsPaused())
return; return;
m_WorkQueue.Enqueue([this, checkable, cr]() { CheckResultHandlerInternal(checkable, cr); });
}
void GelfWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
{
AssertOnWorkQueue();
CONTEXT("GELF Processing check result for '" << checkable->GetName() << "'");
Log(LogDebug, "GelfWriter")
<< "Processing check result for '" << checkable->GetName() << "'";
Host::Ptr host; Host::Ptr host;
Service::Ptr service; Service::Ptr service;
tie(host, service) = GetHostService(checkable); tie(host, service) = GetHostService(checkable);
@ -306,22 +294,21 @@ void GelfWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable, con
fields->Set("_reachable", checkable->IsReachable()); fields->Set("_reachable", checkable->IsReachable());
CheckCommand::Ptr checkCommand = checkable->GetCheckCommand(); CheckCommand::Ptr checkCommand = checkable->GetCheckCommand();
if (checkCommand)
fields->Set("_check_command", checkCommand->GetName()); fields->Set("_check_command", checkCommand->GetName());
double ts = Utility::GetTime(); m_WorkQueue.Enqueue([this, checkable, cr, fields = std::move(fields)]() {
CONTEXT("GELF Processing check result for '" << checkable->GetName() << "'");
Log(LogDebug, "GelfWriter")
<< "Processing check result for '" << checkable->GetName() << "'";
if (cr) {
fields->Set("_latency", cr->CalculateLatency()); fields->Set("_latency", cr->CalculateLatency());
fields->Set("_execution_time", cr->CalculateExecutionTime()); fields->Set("_execution_time", cr->CalculateExecutionTime());
fields->Set("short_message", CompatUtility::GetCheckResultOutput(cr)); fields->Set("short_message", CompatUtility::GetCheckResultOutput(cr));
fields->Set("full_message", cr->GetOutput()); fields->Set("full_message", cr->GetOutput());
fields->Set("_check_source", cr->GetCheckSource()); fields->Set("_check_source", cr->GetCheckSource());
ts = cr->GetExecutionEnd();
}
if (cr && GetEnableSendPerfdata()) { if (GetEnableSendPerfdata()) {
Array::Ptr perfdata = cr->GetPerformanceData(); Array::Ptr perfdata = cr->GetPerformanceData();
if (perfdata) { if (perfdata) {
@ -338,7 +325,7 @@ void GelfWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable, con
Log(LogWarning, "GelfWriter") Log(LogWarning, "GelfWriter")
<< "Ignoring invalid perfdata for checkable '" << "Ignoring invalid perfdata for checkable '"
<< checkable->GetName() << "' and command '" << checkable->GetName() << "' and command '"
<< checkCommand->GetName() << "' with value: " << val; << checkable->GetCheckCommand()->GetName() << "' with value: " << val;
continue; continue;
} }
} }
@ -366,31 +353,15 @@ void GelfWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable, con
} }
} }
SendLogMessage(checkable, ComposeGelfMessage(fields, GetSource(), ts)); SendLogMessage(checkable, ComposeGelfMessage(fields, GetSource(), cr->GetExecutionEnd()));
}
void GelfWriter::NotificationToUserHandler(const Notification::Ptr& notification, const Checkable::Ptr& checkable,
const User::Ptr& user, NotificationType notificationType, CheckResult::Ptr const& cr,
const String& author, const String& commentText, const String& commandName)
{
if (IsPaused())
return;
m_WorkQueue.Enqueue([this, notification, checkable, user, notificationType, cr, author, commentText, commandName]() {
NotificationToUserHandlerInternal(notification, checkable, user, notificationType, cr, author, commentText, commandName);
}); });
} }
void GelfWriter::NotificationToUserHandlerInternal(const Notification::Ptr& notification, const Checkable::Ptr& checkable, void GelfWriter::NotificationToUserHandler(const Checkable::Ptr& checkable, NotificationType notificationType,
const User::Ptr& user, NotificationType notificationType, CheckResult::Ptr const& cr, const CheckResult::Ptr& cr, const String& author, const String& commentText, const String& commandName)
const String& author, const String& commentText, const String& commandName)
{ {
AssertOnWorkQueue(); if (IsPaused())
return;
CONTEXT("GELF Processing notification to all users '" << checkable->GetName() << "'");
Log(LogDebug, "GelfWriter")
<< "Processing notification for '" << checkable->GetName() << "'";
Host::Ptr host; Host::Ptr host;
Service::Ptr service; Service::Ptr service;
@ -430,32 +401,23 @@ void GelfWriter::NotificationToUserHandlerInternal(const Notification::Ptr& noti
fields->Set("_command", commandName); fields->Set("_command", commandName);
fields->Set("_notification_type", notificationTypeString); fields->Set("_notification_type", notificationTypeString);
fields->Set("_comment", authorComment); fields->Set("_comment", authorComment);
fields->Set("_check_command", checkable->GetCheckCommand()->GetName());
CheckCommand::Ptr commandObj = checkable->GetCheckCommand(); m_WorkQueue.Enqueue([this, checkable, ts, fields = std::move(fields)]() {
CONTEXT("GELF Processing notification to all users '" << checkable->GetName() << "'");
if (commandObj) Log(LogDebug, "GelfWriter")
fields->Set("_check_command", commandObj->GetName()); << "Processing notification for '" << checkable->GetName() << "'";
SendLogMessage(checkable, ComposeGelfMessage(fields, GetSource(), ts)); SendLogMessage(checkable, ComposeGelfMessage(fields, GetSource(), ts));
});
} }
void GelfWriter::StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type) void GelfWriter::StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
{ {
if (IsPaused()) if (IsPaused())
return; return;
m_WorkQueue.Enqueue([this, checkable, cr, type]() { StateChangeHandlerInternal(checkable, cr, type); });
}
void GelfWriter::StateChangeHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type)
{
AssertOnWorkQueue();
CONTEXT("GELF Processing state change '" << checkable->GetName() << "'");
Log(LogDebug, "GelfWriter")
<< "Processing state change for '" << checkable->GetName() << "'";
Host::Ptr host; Host::Ptr host;
Service::Ptr service; Service::Ptr service;
tie(host, service) = GetHostService(checkable); tie(host, service) = GetHostService(checkable);
@ -478,21 +440,19 @@ void GelfWriter::StateChangeHandlerInternal(const Checkable::Ptr& checkable, con
fields->Set("_last_hard_state", host->GetLastHardState()); fields->Set("_last_hard_state", host->GetLastHardState());
} }
CheckCommand::Ptr commandObj = checkable->GetCheckCommand(); fields->Set("_check_command", checkable->GetCheckCommand()->GetName());
if (commandObj)
fields->Set("_check_command", commandObj->GetName());
double ts = Utility::GetTime();
if (cr) {
fields->Set("short_message", CompatUtility::GetCheckResultOutput(cr)); fields->Set("short_message", CompatUtility::GetCheckResultOutput(cr));
fields->Set("full_message", cr->GetOutput()); fields->Set("full_message", cr->GetOutput());
fields->Set("_check_source", cr->GetCheckSource()); fields->Set("_check_source", cr->GetCheckSource());
ts = cr->GetExecutionEnd();
} m_WorkQueue.Enqueue([this, checkable, fields = std::move(fields), ts = cr->GetExecutionEnd()]() {
CONTEXT("GELF Processing state change '" << checkable->GetName() << "'");
Log(LogDebug, "GelfWriter")
<< "Processing state change for '" << checkable->GetName() << "'";
SendLogMessage(checkable, ComposeGelfMessage(fields, GetSource(), ts)); SendLogMessage(checkable, ComposeGelfMessage(fields, GetSource(), ts));
});
} }
String GelfWriter::ComposeGelfMessage(const Dictionary::Ptr& fields, const String& source, double ts) String GelfWriter::ComposeGelfMessage(const Dictionary::Ptr& fields, const String& source, double ts)
@ -506,6 +466,8 @@ String GelfWriter::ComposeGelfMessage(const Dictionary::Ptr& fields, const Strin
void GelfWriter::SendLogMessage(const Checkable::Ptr& checkable, const String& gelfMessage) void GelfWriter::SendLogMessage(const Checkable::Ptr& checkable, const String& gelfMessage)
{ {
AssertOnWorkQueue();
std::ostringstream msgbuf; std::ostringstream msgbuf;
msgbuf << gelfMessage; msgbuf << gelfMessage;
msgbuf << '\0'; msgbuf << '\0';

10
lib/perfdata/gelfwriter.hpp
View File

@ -40,15 +40,9 @@ private:
Timer::Ptr m_ReconnectTimer; Timer::Ptr m_ReconnectTimer;
void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void CheckResultHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void NotificationToUserHandler(const Checkable::Ptr& checkable, NotificationType notificationType, const CheckResult::Ptr& cr,
void NotificationToUserHandler(const Notification::Ptr& notification, const Checkable::Ptr& checkable,
const User::Ptr& user, NotificationType notificationType, const CheckResult::Ptr& cr,
const String& author, const String& commentText, const String& commandName); const String& author, const String& commentText, const String& commandName);
void NotificationToUserHandlerInternal(const Notification::Ptr& notification, const Checkable::Ptr& checkable, void StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
const User::Ptr& user, NotificationType notification_type, const CheckResult::Ptr& cr,
const String& author, const String& comment_text, const String& command_name);
void StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type);
void StateChangeHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type);
String ComposeGelfMessage(const Dictionary::Ptr& fields, const String& source, double ts); String ComposeGelfMessage(const Dictionary::Ptr& fields, const String& source, double ts);
void SendLogMessage(const Checkable::Ptr& checkable, const String& gelfMessage); void SendLogMessage(const Checkable::Ptr& checkable, const String& gelfMessage);

70
lib/perfdata/graphitewriter.cpp
View File

@ -261,27 +261,6 @@ void GraphiteWriter::CheckResultHandler(const Checkable::Ptr& checkable, const C
if (IsPaused()) if (IsPaused())
return; return;
m_WorkQueue.Enqueue([this, checkable, cr]() { CheckResultHandlerInternal(checkable, cr); });
}
/**
* Check result event handler, prepares metadata and perfdata values and calls Send*()
*
* Called inside the WQ.
*
* @param checkable Host/Service object
* @param cr Check result including performance data
*/
void GraphiteWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
{
AssertOnWorkQueue();
CONTEXT("Processing check result for '" << checkable->GetName() << "'");
/* TODO: Deal with missing connection here. Needs refactoring
* into parsing the actual performance data and then putting it
* into a queue for re-inserting. */
if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata()) if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata())
return; return;
@ -306,29 +285,34 @@ void GraphiteWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable,
}); });
} }
String prefixPerfdata = prefix + ".perfdata"; std::vector<std::pair<String, double>> metadata;
String prefixMetadata = prefix + ".metadata";
double ts = cr->GetExecutionEnd();
if (GetEnableSendMetadata()) { if (GetEnableSendMetadata()) {
if (service) { metadata = {
SendMetric(checkable, prefixMetadata, "state", service->GetState(), ts); {"state", service ? service->GetState() : host->GetState()},
} else { {"current_attempt", checkable->GetCheckAttempt()},
SendMetric(checkable, prefixMetadata, "state", host->GetState(), ts); {"max_check_attempts", checkable->GetMaxCheckAttempts()},
{"state_type", checkable->GetStateType()},
{"reachable", checkable->IsReachable()},
{"downtime_depth", checkable->GetDowntimeDepth()},
{"acknowledgement", checkable->GetAcknowledgement()},
{"latency", cr->CalculateLatency()},
{"execution_time", cr->CalculateExecutionTime()}
};
} }
SendMetric(checkable, prefixMetadata, "current_attempt", checkable->GetCheckAttempt(), ts); m_WorkQueue.Enqueue([this, checkable, cr, prefix = std::move(prefix), metadata = std::move(metadata)]() {
SendMetric(checkable, prefixMetadata, "max_check_attempts", checkable->GetMaxCheckAttempts(), ts); CONTEXT("Processing check result for '" << checkable->GetName() << "'");
SendMetric(checkable, prefixMetadata, "state_type", checkable->GetStateType(), ts);
SendMetric(checkable, prefixMetadata, "reachable", checkable->IsReachable(), ts); /* TODO: Deal with missing connection here. Needs refactoring
SendMetric(checkable, prefixMetadata, "downtime_depth", checkable->GetDowntimeDepth(), ts); * into parsing the actual performance data and then putting it
SendMetric(checkable, prefixMetadata, "acknowledgement", checkable->GetAcknowledgement(), ts); * into a queue for re-inserting. */
SendMetric(checkable, prefixMetadata, "latency", cr->CalculateLatency(), ts);
SendMetric(checkable, prefixMetadata, "execution_time", cr->CalculateExecutionTime(), ts); for (auto& [name, val] : metadata) {
SendMetric(checkable, prefix + ".metadata", name, val, cr->GetExecutionEnd());
} }
SendPerfdata(checkable, prefixPerfdata, cr, ts); SendPerfdata(checkable, prefix + ".perfdata", cr);
});
} }
/** /**
@ -337,10 +321,11 @@ void GraphiteWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable,
* @param checkable Host/service object * @param checkable Host/service object
* @param prefix Metric prefix string * @param prefix Metric prefix string
* @param cr Check result including performance data * @param cr Check result including performance data
* @param ts Timestamp when the check result was created
*/ */
void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr, double ts) void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr)
{ {
AssertOnWorkQueue();
Array::Ptr perfdata = cr->GetPerformanceData(); Array::Ptr perfdata = cr->GetPerformanceData();
if (!perfdata) if (!perfdata)
@ -367,6 +352,7 @@ void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String&
} }
String escapedKey = EscapeMetricLabel(pdv->GetLabel()); String escapedKey = EscapeMetricLabel(pdv->GetLabel());
double ts = cr->GetExecutionEnd();
SendMetric(checkable, prefix, escapedKey + ".value", pdv->GetValue(), ts); SendMetric(checkable, prefix, escapedKey + ".value", pdv->GetValue(), ts);
@ -394,6 +380,8 @@ void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String&
*/ */
void GraphiteWriter::SendMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts) void GraphiteWriter::SendMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts)
{ {
AssertOnWorkQueue();
namespace asio = boost::asio; namespace asio = boost::asio;
std::ostringstream msgbuf; std::ostringstream msgbuf;

3
lib/perfdata/graphitewriter.hpp
View File

@ -45,9 +45,8 @@ private:
Timer::Ptr m_ReconnectTimer; Timer::Ptr m_ReconnectTimer;
void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void CheckResultHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void SendMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts); void SendMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts);
void SendPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr, double ts); void SendPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr);
static String EscapeMetric(const String& str); static String EscapeMetric(const String& str);
static String EscapeMetricLabel(const String& str); static String EscapeMetricLabel(const String& str);
static Value EscapeMacroMetric(const Value& value); static Value EscapeMacroMetric(const Value& value);

68
lib/perfdata/influxdbcommonwriter.cpp
View File

@ -204,15 +204,6 @@ void InfluxdbCommonWriter::CheckResultHandler(const Checkable::Ptr& checkable, c
if (IsPaused()) if (IsPaused())
return; return;
m_WorkQueue.Enqueue([this, checkable, cr]() { CheckResultHandlerWQ(checkable, cr); }, PriorityLow);
}
void InfluxdbCommonWriter::CheckResultHandlerWQ(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
{
AssertOnWorkQueue();
CONTEXT("Processing check result for '" << checkable->GetName() << "'");
if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata()) if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata())
return; return;
@ -225,10 +216,6 @@ void InfluxdbCommonWriter::CheckResultHandlerWQ(const Checkable::Ptr& checkable,
resolvers.emplace_back("service", service); resolvers.emplace_back("service", service);
resolvers.emplace_back("host", host); resolvers.emplace_back("host", host);
String prefix;
double ts = cr->GetExecutionEnd();
// Clone the template and perform an in-place macro expansion of measurement and tag values // Clone the template and perform an in-place macro expansion of measurement and tag values
Dictionary::Ptr tmpl_clean = service ? GetServiceTemplate() : GetHostTemplate(); Dictionary::Ptr tmpl_clean = service ? GetServiceTemplate() : GetHostTemplate();
Dictionary::Ptr tmpl = static_pointer_cast<Dictionary>(tmpl_clean->ShallowClone()); Dictionary::Ptr tmpl = static_pointer_cast<Dictionary>(tmpl_clean->ShallowClone());
@ -253,11 +240,31 @@ void InfluxdbCommonWriter::CheckResultHandlerWQ(const Checkable::Ptr& checkable,
tmpl->Set("tags", tags); tmpl->Set("tags", tags);
} }
CheckCommand::Ptr checkCommand = checkable->GetCheckCommand(); Dictionary::Ptr fields;
if (GetEnableSendMetadata()) {
fields = new Dictionary();
Array::Ptr perfdata = cr->GetPerformanceData(); if (service)
fields->Set("state", new InfluxdbInteger(service->GetState()));
else
fields->Set("state", new InfluxdbInteger(host->GetState()));
if (perfdata) { fields->Set("current_attempt", new InfluxdbInteger(checkable->GetCheckAttempt()));
fields->Set("max_check_attempts", new InfluxdbInteger(checkable->GetMaxCheckAttempts()));
fields->Set("state_type", new InfluxdbInteger(checkable->GetStateType()));
fields->Set("reachable", checkable->IsReachable());
fields->Set("downtime_depth", new InfluxdbInteger(checkable->GetDowntimeDepth()));
fields->Set("acknowledgement", new InfluxdbInteger(checkable->GetAcknowledgement()));
fields->Set("latency", cr->CalculateLatency());
fields->Set("execution_time", cr->CalculateExecutionTime());
}
m_WorkQueue.Enqueue([this, checkable, cr, tmpl = std::move(tmpl), metadataFields = std::move(fields)]() {
CONTEXT("Processing check result for '" << checkable->GetName() << "'");
double ts = cr->GetExecutionEnd();
if (Array::Ptr perfdata = cr->GetPerformanceData()) {
ObjectLock olock(perfdata); ObjectLock olock(perfdata);
for (const Value& val : perfdata) { for (const Value& val : perfdata) {
PerfdataValue::Ptr pdv; PerfdataValue::Ptr pdv;
@ -271,7 +278,7 @@ void InfluxdbCommonWriter::CheckResultHandlerWQ(const Checkable::Ptr& checkable,
Log(LogWarning, GetReflectionType()->GetName()) Log(LogWarning, GetReflectionType()->GetName())
<< "Ignoring invalid perfdata for checkable '" << "Ignoring invalid perfdata for checkable '"
<< checkable->GetName() << "' and command '" << checkable->GetName() << "' and command '"
<< checkCommand->GetName() << "' with value: " << val; << checkable->GetCheckCommand()->GetName() << "' with value: " << val;
continue; continue;
} }
} }
@ -297,29 +304,10 @@ void InfluxdbCommonWriter::CheckResultHandlerWQ(const Checkable::Ptr& checkable,
} }
} }
if (GetEnableSendMetadata()) { if (metadataFields) {
Host::Ptr host; SendMetric(checkable, tmpl, Empty, metadataFields, ts);
Service::Ptr service;
tie(host, service) = GetHostService(checkable);
Dictionary::Ptr fields = new Dictionary();
if (service)
fields->Set("state", new InfluxdbInteger(service->GetState()));
else
fields->Set("state", new InfluxdbInteger(host->GetState()));
fields->Set("current_attempt", new InfluxdbInteger(checkable->GetCheckAttempt()));
fields->Set("max_check_attempts", new InfluxdbInteger(checkable->GetMaxCheckAttempts()));
fields->Set("state_type", new InfluxdbInteger(checkable->GetStateType()));
fields->Set("reachable", checkable->IsReachable());
fields->Set("downtime_depth", new InfluxdbInteger(checkable->GetDowntimeDepth()));
fields->Set("acknowledgement", new InfluxdbInteger(checkable->GetAcknowledgement()));
fields->Set("latency", cr->CalculateLatency());
fields->Set("execution_time", cr->CalculateExecutionTime());
SendMetric(checkable, tmpl, Empty, fields, ts);
} }
}, PriorityLow);
} }
String InfluxdbCommonWriter::EscapeKeyOrTagValue(const String& str) String InfluxdbCommonWriter::EscapeKeyOrTagValue(const String& str)
@ -364,6 +352,8 @@ String InfluxdbCommonWriter::EscapeValue(const Value& value)
void InfluxdbCommonWriter::SendMetric(const Checkable::Ptr& checkable, const Dictionary::Ptr& tmpl, void InfluxdbCommonWriter::SendMetric(const Checkable::Ptr& checkable, const Dictionary::Ptr& tmpl,
const String& label, const Dictionary::Ptr& fields, double ts) const String& label, const Dictionary::Ptr& fields, double ts)
{ {
AssertOnWorkQueue();
std::ostringstream msgbuf; std::ostringstream msgbuf;
msgbuf << EscapeKeyOrTagValue(tmpl->Get("measurement")); msgbuf << EscapeKeyOrTagValue(tmpl->Get("measurement"));

1
lib/perfdata/influxdbcommonwriter.hpp
View File

@ -54,7 +54,6 @@ private:
std::atomic_size_t m_DataBufferSize{0}; std::atomic_size_t m_DataBufferSize{0};
void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr); void CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void CheckResultHandlerWQ(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr);
void SendMetric(const Checkable::Ptr& checkable, const Dictionary::Ptr& tmpl, void SendMetric(const Checkable::Ptr& checkable, const Dictionary::Ptr& tmpl,
const String& label, const Dictionary::Ptr& fields, double ts); const String& label, const Dictionary::Ptr& fields, double ts);
void FlushTimeout(); void FlushTimeout();