Provide zones and endpoints (ex local) stats via /v1/status

lib/remote/endpoint.cpp

@@ -6,10 +6,12 @@
 #include "remote/apilistener.hpp"
 #include "remote/jsonrpcconnection.hpp"
 #include "remote/zone.hpp"
+#include "base/perfdatavalue.hpp"
 #include "base/configtype.hpp"
 #include "base/utility.hpp"
 #include "base/exception.hpp"
 #include "base/convert.hpp"
+#include "base/statsfunction.hpp"
 
 using namespace icinga;
 
@@ -18,6 +20,8 @@ REGISTER_TYPE(Endpoint);
 boost::signals2::signal<void(const Endpoint::Ptr&, const JsonRpcConnection::Ptr&)> Endpoint::OnConnected;
 boost::signals2::signal<void(const Endpoint::Ptr&, const JsonRpcConnection::Ptr&)> Endpoint::OnDisconnected;
 
+REGISTER_STATSFUNCTION(Endpoint, &Endpoint::StatsFunc);
+
 void Endpoint::OnAllConfigLoaded()
 {
 	ObjectImpl<Endpoint>::OnAllConfigLoaded();
@@ -109,6 +113,70 @@ Endpoint::Ptr Endpoint::GetLocalEndpoint()
 	return listener->GetLocalEndpoint();
 }
 
+void Endpoint::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
+{
+	auto localZone (Zone::GetLocalZone());
+	auto parentZone (localZone->GetParent());
+	auto unorderedZones (ConfigType::GetObjectsByType<Zone>());
+	std::set<Zone::Ptr> zones (unorderedZones.begin(), unorderedZones.end());
+	std::set<Endpoint::Ptr> endpoints;
+	Dictionary::Ptr ourStatus = new Dictionary;
+
+	unorderedZones.clear();
+
+	for (auto zone (zones.begin()); zone != zones.end();) {
+		if ((*zone)->GetParent() == localZone) {
+			++zone;
+		} else {
+			zones.erase(zone++);
+		}
+	}
+
+	zones.emplace(localZone);
+
+	if (parentZone)
+		zones.emplace(parentZone);
+
+	for (auto& zone : zones) {
+		auto zoneEndpoints (zone->GetEndpoints());
+		endpoints.insert(zoneEndpoints.begin(), zoneEndpoints.end());
+	}
+
+	endpoints.erase(GetLocalEndpoint());
+
+	for (auto& endpoint : endpoints) {
+		ourStatus->Set(endpoint->GetName(), new Dictionary({
+			{"local_log_position", endpoint->GetLocalLogPosition()},
+			{"remote_log_position", endpoint->GetRemoteLogPosition()},
+			{"connecting", endpoint->GetConnecting()},
+			{"syncing", endpoint->GetSyncing()},
+			{"connected", endpoint->GetConnected()},
+			{"last_message_sent", endpoint->GetLastMessageSent()},
+			{"last_message_received", endpoint->GetLastMessageReceived()},
+			{"messages_sent_per_second", endpoint->GetMessagesSentPerSecond()},
+			{"messages_received_per_second", endpoint->GetMessagesReceivedPerSecond()},
+			{"bytes_sent_per_second", endpoint->GetBytesSentPerSecond()},
+			{"bytes_received_per_second", endpoint->GetBytesReceivedPerSecond()}
+		}));
+	}
+
+	{
+		ObjectLock ourStatusLock (ourStatus);
+
+		for (auto& nameEndpointStatus : ourStatus) {
+			Dictionary::Ptr endpointStatus = nameEndpointStatus.second;
+			ObjectLock endpointStatusLock (endpointStatus);
+			auto labelPrefix ("endpoint_" + nameEndpointStatus.first + "_");
+
+			for (auto& labelValue : endpointStatus) {
+				perfdata->Add(new PerfdataValue(labelPrefix + labelValue.first, labelValue.second));
+			}
+		}
+	}
+
+	status->Set("endpoint", ourStatus);
+}
+
 void Endpoint::AddMessageSent(int bytes)
 {
 	double time = Utility::GetTime();

lib/remote/endpoint.hpp

@@ -5,7 +5,9 @@
 
 #include "remote/i2-remote.hpp"
 #include "remote/endpoint-ti.hpp"
+#include "base/array.hpp"
 #include "base/atomic.hpp"
+#include "base/dictionary.hpp"
 #include "base/ringbuffer.hpp"
 #include <cstdint>
 #include <set>
@@ -43,6 +45,7 @@ public:
 	bool GetConnected() const override;
 
 	static Endpoint::Ptr GetLocalEndpoint();
+	static void StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata);
 
 	void SetCachedZone(const intrusive_ptr<Zone>& zone);
 

lib/remote/zone.cpp

@@ -4,13 +4,19 @@
 #include "remote/zone-ti.cpp"
 #include "remote/jsonrpcconnection.hpp"
 #include "base/array.hpp"
+#include "base/perfdatavalue.hpp"
 #include "base/objectlock.hpp"
 #include "base/logger.hpp"
+#include "base/statsfunction.hpp"
+#include <algorithm>
+#include <limits>
 
 using namespace icinga;
 
 REGISTER_TYPE(Zone);
 
+REGISTER_STATSFUNCTION(Zone, &Zone::StatsFunc);
+
 void Zone::OnAllConfigLoaded()
 {
 	ObjectImpl<Zone>::OnAllConfigLoaded();
@@ -140,6 +146,163 @@ Zone::Ptr Zone::GetLocalZone()
 	return local->GetZone();
 }
 
+static std::set<String> l_StatsFuncAggregateSum ({
+	"messages_sent_per_second", "messages_received_per_second", "bytes_sent_per_second", "bytes_received_per_second"
+});
+
+static std::set<String> l_StatsFuncAggregateCount ({
+	"connecting", "syncing", "connected"
+});
+
+static std::set<String> l_StatsFuncAggregateMin ({
+	"last_message_sent", "last_message_received"
+});
+
+void Zone::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
+{
+	auto localZone (Zone::GetLocalZone());
+	auto parentZone (localZone->GetParent());
+	auto unorderedZones (ConfigType::GetObjectsByType<Zone>());
+	std::set<Zone::Ptr> zones (unorderedZones.begin(), unorderedZones.end());
+	Dictionary::Ptr ourStatus = new Dictionary;
+	auto localEndpoint (Endpoint::GetLocalEndpoint());
+
+	unorderedZones.clear();
+
+	for (auto zone (zones.begin()); zone != zones.end();) {
+		if ((*zone)->GetParent() == localZone) {
+			++zone;
+		} else {
+			zones.erase(zone++);
+		}
+	}
+
+	zones.emplace(localZone);
+
+	if (parentZone)
+		zones.emplace(parentZone);
+
+	for (auto& zone : zones) {
+		Dictionary::Ptr endpointStats = new Dictionary({
+			{"local_log_position", new Array},
+			{"remote_log_position", new Array},
+			{"connecting", new Array},
+			{"syncing", new Array},
+			{"connected", new Array},
+			{"last_message_sent", new Array},
+			{"last_message_received", new Array},
+			{"messages_sent_per_second", new Array},
+			{"messages_received_per_second", new Array},
+			{"bytes_sent_per_second", new Array},
+			{"bytes_received_per_second", new Array}
+		});
+
+		auto endpoints (zone->GetEndpoints());
+
+		std::remove(endpoints.begin(), endpoints.end(), localEndpoint);
+
+		if (endpoints.empty())
+			continue;
+
+		for (auto& endpoint : endpoints) {
+			((Array::Ptr)endpointStats->Get("local_log_position"))->Add(endpoint->GetLocalLogPosition());
+			((Array::Ptr)endpointStats->Get("remote_log_position"))->Add(endpoint->GetRemoteLogPosition());
+			((Array::Ptr)endpointStats->Get("connecting"))->Add(endpoint->GetConnecting());
+			((Array::Ptr)endpointStats->Get("syncing"))->Add(endpoint->GetSyncing());
+			((Array::Ptr)endpointStats->Get("connected"))->Add(endpoint->GetConnected());
+			((Array::Ptr)endpointStats->Get("last_message_sent"))->Add(endpoint->GetLastMessageSent());
+			((Array::Ptr)endpointStats->Get("last_message_received"))->Add(endpoint->GetLastMessageReceived());
+			((Array::Ptr)endpointStats->Get("messages_sent_per_second"))->Add(endpoint->GetMessagesSentPerSecond());
+			((Array::Ptr)endpointStats->Get("messages_received_per_second"))->Add(endpoint->GetMessagesReceivedPerSecond());
+			((Array::Ptr)endpointStats->Get("bytes_sent_per_second"))->Add(endpoint->GetBytesSentPerSecond());
+			((Array::Ptr)endpointStats->Get("bytes_received_per_second"))->Add(endpoint->GetBytesReceivedPerSecond());
+		}
+
+		for (auto& label : l_StatsFuncAggregateSum) {
+			auto sum (0.0);
+			Array::Ptr values = endpointStats->Get(label);
+			ObjectLock valuesLock (values);
+
+			for (auto& value : values) {
+				sum += value.Get<double>();
+			}
+
+			endpointStats->Set(label, sum);
+		}
+
+		for (auto& label : l_StatsFuncAggregateCount) {
+			uintmax_t count = 0;
+			Array::Ptr values = endpointStats->Get(label);
+			ObjectLock valuesLock (values);
+
+			for (auto& value : values) {
+				if (value.Get<bool>()) {
+					++count;
+				}
+			}
+
+			endpointStats->Set(label, count);
+		}
+
+		for (auto& label : l_StatsFuncAggregateMin) {
+			auto min (std::numeric_limits<double>::infinity());
+			Array::Ptr values = endpointStats->Get(label);
+			ObjectLock valuesLock (values);
+
+			for (auto& value : values) {
+				auto number (value.Get<double>());
+
+				if (number < min) {
+					min = number;
+				}
+			}
+
+			endpointStats->Set(label, min);
+		}
+
+		{
+			auto maxDiff (-std::numeric_limits<double>::infinity());
+			Array::Ptr remoteLogPositions = endpointStats->Get("remote_log_position");
+			ObjectLock remoteLogPositionLock (remoteLogPositions);
+			auto remoteLogPosition (begin(remoteLogPositions));
+			Array::Ptr localLogPositions = endpointStats->Get("local_log_position");
+			ObjectLock localLogPositionLock (localLogPositions);
+
+			for (auto& localLogPosition : localLogPositions) {
+				auto diff (localLogPosition - *remoteLogPosition);
+
+				if (diff > maxDiff) {
+					maxDiff = diff;
+				}
+
+				++remoteLogPosition;
+			}
+
+			endpointStats->Set("client_log_lag", maxDiff);
+			endpointStats->Remove("local_log_position");
+			endpointStats->Remove("remote_log_position");
+		}
+
+		ourStatus->Set(zone->GetName(), endpointStats);
+	}
+
+	{
+		ObjectLock ourStatusLock (ourStatus);
+
+		for (auto& nameZoneStatus : ourStatus) {
+			Dictionary::Ptr zoneStatus = nameZoneStatus.second;
+			ObjectLock zoneStatusLock (zoneStatus);
+			auto labelPrefix ("zone_" + nameZoneStatus.first + "_");
+
+			for (auto& labelValue : zoneStatus) {
+				perfdata->Add(new PerfdataValue(labelPrefix + labelValue.first, labelValue.second));
+			}
+		}
+	}
+
+	status->Set("zone", ourStatus);
+}
+
 void Zone::ValidateEndpointsRaw(const Lazy<Array::Ptr>& lvalue, const ValidationUtils& utils)
 {
 	ObjectImpl<Zone>::ValidateEndpointsRaw(lvalue, utils);

lib/remote/zone.hpp

@@ -3,6 +3,8 @@
 #ifndef ZONE_H
 #define ZONE_H
 
+#include "base/array.hpp"
+#include "base/dictionary.hpp"
 #include "remote/i2-remote.hpp"
 #include "remote/zone-ti.hpp"
 #include "remote/endpoint.hpp"
@@ -32,6 +34,7 @@ public:
 	bool IsHACluster() const;
 
 	static Zone::Ptr GetLocalZone();
+	static void StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata);
 
 protected:
 	void ValidateEndpointsRaw(const Lazy<Array::Ptr>& lvalue, const ValidationUtils& utils) override;