Count incoming messages per type and endpoint

Author: Al2Klimov

lib/remote/endpoint.hpp

@@ -5,12 +5,17 @@
 
 #include "remote/i2-remote.hpp"
 #include "remote/endpoint-ti.hpp"
+#include "base/atomic.hpp"
 #include "base/ringbuffer.hpp"
+#include <cstdint>
 #include <set>
+#include <shared_mutex>
+#include <unordered_map>
 
 namespace icinga
 {
 
+class ApiFunction;
 class JsonRpcConnection;
 class Zone;
 
@@ -21,6 +26,8 @@ class Zone;
  */
 class Endpoint final : public ObjectImpl<Endpoint>
 {
+	friend JsonRpcConnection;
+
 public:
 	DECLARE_OBJECT(Endpoint);
 	DECLARE_OBJECTNAME(Endpoint);
@@ -61,6 +68,9 @@ class Endpoint final : public ObjectImpl<Endpoint>
 	mutable RingBuffer m_MessagesReceived{60};
 	mutable RingBuffer m_BytesSent{60};
 	mutable RingBuffer m_BytesReceived{60};
+
+	mutable std::shared_mutex m_MessageCountersMutex;
+	std::unordered_map<intrusive_ptr<ApiFunction>, Atomic<uint_fast64_t>> m_MessageCounters;
 };
 
 }

lib/remote/jsonrpcconnection.cpp

@@ -351,6 +351,25 @@ void JsonRpcConnection::MessageHandler(const Dictionary::Ptr& message)
 			Log(LogNotice, "JsonRpcConnection")
 				<< "Call to non-existent function '" << method << "' from endpoint '" << m_Identity << "'.";
 		} else {
+			if (m_Endpoint) {
+				std::shared_lock sLock (m_Endpoint->m_MessageCountersMutex);
+				auto& mc (m_Endpoint->m_MessageCounters);
+				auto it (mc.find(afunc.get())); // Lookup by pointer is faster than by string
+
+				if (it == mc.end()) {
+					sLock.unlock();
+					std::unique_lock uLock (m_Endpoint->m_MessageCountersMutex);
+
+					if (it = mc.find(afunc.get()); it == mc.end()) {
+						mc.emplace(afunc.get(), 1);
+					} else {
+						it->second.fetch_add(1, std::memory_order_relaxed);
+					}
+				} else {
+					it->second.fetch_add(1, std::memory_order_relaxed);
+				}
+			}
+
 			Dictionary::Ptr params = message->Get("params");
 			if (params)
 				resultMessage->Set("result", afunc->Invoke(origin, params));