diff --git a/doc/09-object-types.md b/doc/09-object-types.md index 3fe29702f..de0991d9d 100644 --- a/doc/09-object-types.md +++ b/doc/09-object-types.md @@ -1350,6 +1350,8 @@ Configuration Attributes: service\_name\_template | String | **Optional.** Metric prefix for service name. Defaults to `icinga2.$host.name$.services.$service.name$.$service.check_command$`. enable\_send\_thresholds | Boolean | **Optional.** Send additional threshold metrics. Defaults to `false`. enable\_send\_metadata | Boolean | **Optional.** Send additional metadata metrics. Defaults to `false`. + flush\_interval | Duration | **Optional.** How long to buffer data points before sending. Defaults to `15s`. + flush\_threshold | Number | **Optional.** How many bytes to buffer before forcing a flush to the backend. Defaults to `2MiB`. enable\_ha | Boolean | **Optional.** Enable the high availability functionality. Only valid in a [cluster setup](06-distributed-monitoring.md#distributed-monitoring-high-availability-features). Defaults to `false`. Additional usage examples can be found [here](14-features.md#graphite-carbon-cache-writer). diff --git a/lib/perfdata/graphitewriter.cpp b/lib/perfdata/graphitewriter.cpp index e00cd9275..28c684db8 100644 --- a/lib/perfdata/graphitewriter.cpp +++ b/lib/perfdata/graphitewriter.cpp @@ -2,6 +2,7 @@ // SPDX-License-Identifier: GPL-2.0-or-later #include "perfdata/graphitewriter.hpp" +#include "base/defer.hpp" #include "perfdata/graphitewriter-ti.cpp" #include "icinga/service.hpp" #include "icinga/checkcommand.hpp" @@ -85,6 +86,13 @@ void GraphiteWriter::Resume() /* Register exception handler for WQ tasks. */ 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()}; /* Register event handlers. */ @@ -101,6 +109,8 @@ void GraphiteWriter::Pause() { m_HandleCheckResults.disconnect(); + m_FlushTimer->Stop(true); + std::promise queueDonePromise; m_WorkQueue.Enqueue([&]() { @@ -199,10 +209,10 @@ void GraphiteWriter::CheckResultHandler(const Checkable::Ptr& checkable, const C CONTEXT("Processing check result for '" << checkable->GetName() << "'"); for (auto& [name, val] : metadata) { - SendMetric(checkable, prefix + ".metadata", name, val, cr->GetExecutionEnd()); + AddMetric(checkable, prefix + ".metadata", name, val, cr->GetExecutionEnd()); } - SendPerfdata(checkable, prefix + ".perfdata", cr); + AddPerfdata(checkable, prefix + ".perfdata", cr); }); } @@ -213,7 +223,7 @@ void GraphiteWriter::CheckResultHandler(const Checkable::Ptr& checkable, const C * @param prefix Metric prefix string * @param cr Check result including performance data */ -void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr) +void GraphiteWriter::AddPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr) { AssertOnWorkQueue(); @@ -245,17 +255,17 @@ void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String& String escapedKey = EscapeMetricLabel(pdv->GetLabel()); double ts = cr->GetExecutionEnd(); - SendMetric(checkable, prefix, escapedKey + ".value", pdv->GetValue(), ts); + AddMetric(checkable, prefix, escapedKey + ".value", pdv->GetValue(), ts); if (GetEnableSendThresholds()) { if (!pdv->GetCrit().IsEmpty()) - SendMetric(checkable, prefix, escapedKey + ".crit", pdv->GetCrit(), ts); + AddMetric(checkable, prefix, escapedKey + ".crit", pdv->GetCrit(), ts); if (!pdv->GetWarn().IsEmpty()) - SendMetric(checkable, prefix, escapedKey + ".warn", pdv->GetWarn(), ts); + AddMetric(checkable, prefix, escapedKey + ".warn", pdv->GetWarn(), ts); if (!pdv->GetMin().IsEmpty()) - SendMetric(checkable, prefix, escapedKey + ".min", pdv->GetMin(), ts); + AddMetric(checkable, prefix, escapedKey + ".min", pdv->GetMin(), ts); if (!pdv->GetMax().IsEmpty()) - SendMetric(checkable, prefix, escapedKey + ".max", pdv->GetMax(), ts); + AddMetric(checkable, prefix, escapedKey + ".max", pdv->GetMax(), ts); } } } @@ -269,7 +279,7 @@ void GraphiteWriter::SendPerfdata(const Checkable::Ptr& checkable, const String& * @param value Metric value * @param ts Timestamp when the check result was created */ -void GraphiteWriter::SendMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts) +void GraphiteWriter::AddMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts) { AssertOnWorkQueue(); @@ -284,11 +294,34 @@ void GraphiteWriter::SendMetric(const Checkable::Ptr& checkable, const String& p // do not send \n to debug log msgbuf << "\n"; + m_MsgBuf += std::move(msgbuf).str(); + + if (GetFlushThreshold() <= m_MsgBuf.GetLength()) { + Flush(); + } +} + +/** + * Queues a Flush on the work-queue if none is queued yet. + */ +void GraphiteWriter::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 GraphiteWriter::Flush() +{ try { - m_Connection->Send(asio::buffer(msgbuf.str())); + m_Connection->Send(boost::asio::buffer(std::exchange(m_MsgBuf.GetData(), {}))); } catch (const PerfdataWriterConnection::Stopped& ex) { Log(LogDebug, "GraphiteWriter") << ex.what(); - return; } } diff --git a/lib/perfdata/graphitewriter.hpp b/lib/perfdata/graphitewriter.hpp index 470fcc07d..5f1eda9ce 100644 --- a/lib/perfdata/graphitewriter.hpp +++ b/lib/perfdata/graphitewriter.hpp @@ -36,13 +36,18 @@ protected: private: PerfdataWriterConnection::Ptr m_Connection; + Timer::Ptr m_FlushTimer; + std::atomic_bool m_FlushTimerInQueue{false}; + String m_MsgBuf; WorkQueue m_WorkQueue{10000000, 1}; boost::signals2::connection m_HandleCheckResults; void CheckResultHandler(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 SendPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr); + void AddMetric(const Checkable::Ptr& checkable, const String& prefix, const String& name, double value, double ts); + void AddPerfdata(const Checkable::Ptr& checkable, const String& prefix, const CheckResult::Ptr& cr); + void FlushTimeout(); + void Flush(); static String EscapeMetric(const String& str); static String EscapeMetricLabel(const String& str); static Value EscapeMacroMetric(const Value& value); diff --git a/lib/perfdata/graphitewriter.ti b/lib/perfdata/graphitewriter.ti index f0d9bfb80..2cafdf38d 100644 --- a/lib/perfdata/graphitewriter.ti +++ b/lib/perfdata/graphitewriter.ti @@ -24,8 +24,14 @@ class GraphiteWriter : ConfigObject [config] String service_name_template { default {{{ return "icinga2.$host.name$.services.$service.name$.$service.check_command$"; }}} }; - [config] bool enable_send_thresholds; - [config] bool enable_send_metadata; + [config] bool enable_send_thresholds; + [config] bool enable_send_metadata; + [config] int flush_interval { + default {{{ return 15; }}} + }; + [config] std::size_t flush_threshold { + default {{{ return 2 * 1024 * 1024; }}} + }; [config] double disconnect_timeout { default {{{ return 10; }}}