Add flush_threshold/flush_interval logic to GraphiteWriter

Author: jschmidt-icinga

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).

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<void> 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;
 	}
 }
 

lib/perfdata/graphitewriter.hpp

@@ -36,13 +36,18 @@ class GraphiteWriter final : public ObjectImpl<GraphiteWriter>
 
 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);

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; }}}