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

lib/remote/CMakeLists.txt

@@ -37,6 +37,7 @@ set(remote_SOURCES
   modifyobjecthandler.cpp modifyobjecthandler.hpp
   objectqueryhandler.cpp objectqueryhandler.hpp
   pkiutility.cpp pkiutility.hpp
+  statsreporter.cpp statsreporter.hpp
   statushandler.cpp statushandler.hpp
   templatequeryhandler.cpp templatequeryhandler.hpp
   typequeryhandler.cpp typequeryhandler.hpp

lib/remote/endpoint.cpp

@@ -5,10 +5,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;
 
@@ -17,6 +19,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();
@@ -101,6 +105,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,6 +5,8 @@
 
 #include "remote/i2-remote.hpp"
 #include "remote/endpoint-ti.hpp"
+#include "base/array.hpp"
+#include "base/dictionary.hpp"
 #include "base/ringbuffer.hpp"
 #include <set>
 
@@ -37,6 +39,7 @@ class Endpoint final : public ObjectImpl<Endpoint>
 	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/statsreporter.cpp

@@ -0,0 +1,195 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2018 Icinga Development Team (https://www.icinga.com/)  *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+#include "base/application.hpp"
+#include "base/array.hpp"
+#include "base/configtype.hpp"
+#include "base/dictionary.hpp"
+#include "base/function.hpp"
+#include "base/objectlock.hpp"
+#include "base/scriptglobal.hpp"
+#include "base/statsfunction.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/thread/mutex.hpp>
+#include <set>
+
+using namespace icinga;
+
+REGISTER_APIFUNCTION(ClusterStats, event, &StatsReporter::ClusterStatsAPIHandler);
+
+REGISTER_STATSFUNCTION(ClusterStats, &StatsReporter::StatsFunc);
+
+StatsReporter StatsReporter::m_Instance;
+
+StatsReporter::StatsReporter()
+{
+	Endpoint::OnConnected.connect([this](const Endpoint::Ptr& endpoint, const intrusive_ptr<JsonRpcConnection>&) {
+		OnConnected(endpoint);
+	});
+}
+
+void StatsReporter::OnConnected(const Endpoint::Ptr& endpoint)
+{
+	if (!m_HasBeenInitialized.test_and_set()) {
+		timer = Timer::Create();
+		timer->OnTimerExpired.connect([this](const Timer * const&) { ReportStats(); });
+		timer->SetInterval(10);
+		timer->Start();
+		timer->Reschedule(1);
+	}
+}
+
+void StatsReporter::ReportStats()
+{
+	auto parent (Zone::GetLocalZone()->GetParent());
+
+	if (parent) {
+		Dictionary::Ptr message;
+
+		for (auto& endpoint : parent->GetEndpoints()) {
+			for (auto& client : endpoint->GetClients()) {
+				if (!message) {
+					message = new Dictionary({
+						{ "jsonrpc", "2.0" },
+						{ "method", "event::ClusterStats" },
+						{ "params", new Dictionary({
+							{ "stats", GenerateStats() }
+						}) }
+					});
+				}
+
+				client->SendMessage(message);
+
+				break;
+			}
+		}
+	}
+}
+
+Dictionary::Ptr StatsReporter::GenerateStats()
+{
+	auto allStats (new Dictionary);
+
+	{
+		boost::mutex::scoped_lock lock (m_Mutex);
+
+		for (auto& endpointStats : m_SecondaryStats) {
+			allStats->Set(endpointStats.first, endpointStats.second);
+		}
+	}
+
+	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;
+	auto ourStatus (new Dictionary);
+	auto now (Utility::GetTime());
+
+	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(Endpoint::GetLocalEndpoint());
+
+	for (auto& endpoint : endpoints) {
+		ourStatus->Set(endpoint->GetName(), new Dictionary({
+			{"connected", endpoint->GetConnected()},
+			{"last_message_received", endpoint->GetLastMessageReceived()}
+		}));
+	}
+
+	allStats->Set(Endpoint::GetLocalEndpoint()->GetName(), new Dictionary({
+		{"mtime", now},
+		{"version", Application::GetAppVersion()},
+		{"uptime", now - Application::GetStartTime()},
+		{"endpoints", ourStatus}
+	}));
+
+	return allStats;
+}
+
+Value StatsReporter::ClusterStatsAPIHandler(const MessageOrigin::Ptr& origin, const Dictionary::Ptr& params)
+{
+	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) {
+				// Don't permit any child to speak in our zone's name
+				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;
+}
+
+void StatsReporter::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
+{
+	status->Set("cluster", m_Instance.GenerateStats());
+}

lib/remote/statsreporter.hpp

@@ -0,0 +1,65 @@
+/******************************************************************************
+ * Icinga 2                                                                   *
+ * Copyright (C) 2012-2018 Icinga Development Team (https://www.icinga.com/)  *
+ *                                                                            *
+ * This program is free software; you can redistribute it and/or              *
+ * modify it under the terms of the GNU General Public License                *
+ * as published by the Free Software Foundation; either version 2             *
+ * of the License, or (at your option) any later version.                     *
+ *                                                                            *
+ * This program is distributed in the hope that it will be useful,            *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
+ * GNU General Public License for more details.                               *
+ *                                                                            *
+ * You should have received a copy of the GNU General Public License          *
+ * along with this program; if not, write to the Free Software Foundation     *
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
+ ******************************************************************************/
+
+#ifndef STATSREPORTER_H
+#define STATSREPORTER_H
+
+#include "base/array.hpp"
+#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
+{
+
+/**
+* @ingroup remote
+*/
+class StatsReporter
+{
+public:
+	static Value ClusterStatsAPIHandler(const MessageOrigin::Ptr& origin, const Dictionary::Ptr& params);
+	static void StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata);
+
+private:
+	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 = ATOMIC_FLAG_INIT;
+	Timer::Ptr timer;
+
+	boost::mutex m_Mutex;
+	std::map<String, Dictionary::Ptr> m_SecondaryStats;
+};
+
+}
+
+#endif /* STATSREPORTER_H */