Forward all collected cluster stats to all connected parents

Author: Al2Klimov

lib/remote/statsreporter.cpp

@@ -22,15 +22,16 @@
 #include "base/function.hpp"
 #include "base/objectlock.hpp"
 #include "base/scriptglobal.hpp"
-#include "base/utility.hpp"
 #include "base/value.hpp"
 #include "remote/apifunction.hpp"
 #include "remote/endpoint.hpp"
 #include "remote/messageorigin.hpp"
 #include "remote/statsreporter.hpp"
 #include "remote/zone.hpp"
 #include <boost/bind.hpp>
+#include <boost/thread/mutex.hpp>
 #include <atomic>
+#include <set>
 
 using namespace icinga;
 
@@ -56,20 +57,40 @@ void StatsReporter::OnConnected(const Endpoint::Ptr& endpoint)
 
 void StatsReporter::ReportStats()
 {
-	Dictionary::Ptr stats;
+	Dictionary::Ptr message;
 
 	for (auto& zone : Zone::GetLocalZone()->GetAllParents()) {
 		for (auto& endpoint : zone->GetEndpoints()) {
 			for (auto& client : endpoint->GetClients()) {
-				if (!stats) {
-					stats = GenerateStats();
+				if (!message) {
+					Dictionary::Ptr allStats = new Dictionary;
 
-					if (!stats) {
-						return;
+					auto stats (GenerateStats());
+
+					if (stats)
+						allStats->Set(Endpoint::GetLocalEndpoint()->GetName(), stats);
+
+					{
+						boost::mutex::scoped_lock lock (m_Mutex);
+
+						for (auto& endpointStats : m_SecondaryStats) {
+							allStats->Set(endpointStats.first, endpointStats.second);
+						}
 					}
+
+					if (!allStats->GetLength())
+						return;
+
+					message = new Dictionary({
+						{ "jsonrpc", "2.0" },
+						{ "method", "event::ClusterStats" },
+						{ "params", new Dictionary({
+							{ "stats", allStats }
+						}) }
+					});
 				}
 
-				client->SendMessage(stats);
+				client->SendMessage(message);
 
 				break;
 			}
@@ -93,21 +114,50 @@ Dictionary::Ptr StatsReporter::GenerateStats()
 			}
 		}
 
-		return new Dictionary({
-			{ "jsonrpc", "2.0" },
-			{ "method", "event::ClusterStats" },
-			{ "params", new Dictionary({
-				{ "ctime", Utility::GetTime() },
-				{ "stats", status }
-			}) }
-		});
+		return status;
 	}
 
 	return nullptr;
 }
 
 Value StatsReporter::ClusterStatsAPIHandler(const MessageOrigin::Ptr& origin, const Dictionary::Ptr& params)
 {
-	// TODO
+	auto endpoint (origin->FromClient->GetEndpoint());
+
+	if (endpoint && endpoint->GetZone()->IsChildOf(Zone::GetLocalZone())) {
+		auto rawStats (params->Get("stats"));
+
+		if (rawStats.IsObject()) {
+			Dictionary::Ptr allStats (rawStats);
+
+			if (allStats) {
+				// For security reasons
+				std::set<String> neighborhood;
+
+				for (auto& endpoint : Zone::GetLocalZone()->GetEndpoints()) {
+					neighborhood.emplace(endpoint->GetName());
+				}
+
+				ObjectLock lock (allStats);
+
+				for (auto& endpointStats : allStats) {
+					if (endpointStats.second.IsObject()) {
+						Dictionary::Ptr stats (endpointStats.second);
+
+						if (stats && neighborhood.find(endpointStats.first) == neighborhood.end()) {
+							m_Instance.ClusterStatsHandler(endpointStats.first, endpointStats.second);
+						}
+					}
+				}
+			}
+		}
+	}
+
 	return Empty;
 }
+
+void StatsReporter::ClusterStatsHandler(const String& endpoint, const Dictionary::Ptr& stats)
+{
+	boost::mutex::scoped_lock lock (m_Mutex);
+	m_SecondaryStats[endpoint] = stats;
+}

lib/remote/statsreporter.hpp

@@ -21,11 +21,14 @@
 #define STATSREPORTER_H
 
 #include "base/dictionary.hpp"
+#include "base/string.hpp"
 #include "base/timer.hpp"
 #include "base/value.hpp"
 #include "remote/endpoint.hpp"
 #include "remote/messageorigin.hpp"
 #include <atomic>
+#include <boost/thread/mutex.hpp>
+#include <map>
 
 namespace icinga
 {
@@ -44,11 +47,15 @@ class StatsReporter
 	void OnConnected(const Endpoint::Ptr& endpoint);
 	void ReportStats();
 	Dictionary::Ptr GenerateStats();
+	void ClusterStatsHandler(const String& endpoint, const Dictionary::Ptr& stats);
 
 	static StatsReporter m_Instance;
 
 	std::atomic_flag m_HasBeenInitialized;
 	Timer::Ptr timer;
+
+	boost::mutex m_Mutex;
+	std::map<String, Dictionary::Ptr> m_SecondaryStats;
 };
 
 }