Add metrics about communication between endpoints

refs #5509

diff --git a/lib/base/ringbuffer.cpp b/lib/base/ringbuffer.cpp
index 4e023e7abd770f886726a8f9c5b91b11cc6162a6..660dc820b35bc1af9f7b91f57642788200a704db 100644 (file)
--- a/lib/base/ringbuffer.cpp
+++ b/lib/base/ringbuffer.cpp
@@ -68,10 +68,10 @@ void RingBuffer::InsertValue(RingBuffer::SizeType tv, int num)
 
 int RingBuffer::UpdateAndGetValues(RingBuffer::SizeType tv, RingBuffer::SizeType span)
 {
-       InsertValue(tv, 0);
-
        ObjectLock olock(this);
 
+       InsertValue(tv, 0);
+
        if (span > m_Slots.size())
                span = m_Slots.size();
 

diff --git a/lib/methods/clusterzonechecktask.cpp b/lib/methods/clusterzonechecktask.cpp
index 919e7fbedca82bac1f5b117fa5ea4cf8f6f159a7..b3bf66a5d31eab14d734247858142d524e1a53f5 100644 (file)
--- a/lib/methods/clusterzonechecktask.cpp
+++ b/lib/methods/clusterzonechecktask.cpp
@@ -91,6 +91,13 @@ void ClusterZoneCheckTask::ScriptFunc(const Checkable::Ptr& checkable, const Che
        bool connected = false;
        double zoneLag = 0;
 
+       double lastMessageSent = 0;
+       double lastMessageReceived = 0;
+       double messagesSentPerSecond = 0;
+       double messagesReceivedPerSecond = 0;
+       double bytesSentPerSecond = 0;
+       double bytesReceivedPerSecond = 0;
+
        for (const Endpoint::Ptr& endpoint : zone->GetEndpoints()) {
                if (endpoint->GetConnected())
                        connected = true;
@@ -99,6 +106,17 @@ void ClusterZoneCheckTask::ScriptFunc(const Checkable::Ptr& checkable, const Che
 
                if (eplag > 0 && eplag > zoneLag)
                        zoneLag = eplag;
+
+               if (endpoint->GetLastMessageSent() > lastMessageSent)
+                       lastMessageSent = endpoint->GetLastMessageSent();
+
+               if (endpoint->GetLastMessageReceived() > lastMessageReceived)
+                       lastMessageReceived = endpoint->GetLastMessageReceived();
+
+               messagesSentPerSecond += endpoint->GetMessagesSentPerSecond();
+               messagesReceivedPerSecond += endpoint->GetMessagesReceivedPerSecond();
+               bytesSentPerSecond += endpoint->GetBytesSentPerSecond();
+               bytesReceivedPerSecond += endpoint->GetBytesReceivedPerSecond();
        }
 
        if (!connected) {
@@ -122,6 +140,12 @@ void ClusterZoneCheckTask::ScriptFunc(const Checkable::Ptr& checkable, const Che
 
        Array::Ptr perfdata = new Array();
        perfdata->Add(new PerfdataValue("slave_lag", zoneLag, false, "s", lagWarning, lagCritical));
+       perfdata->Add(new PerfdataValue("last_messages_sent", lastMessageSent));
+       perfdata->Add(new PerfdataValue("last_messages_received", lastMessageReceived));
+       perfdata->Add(new PerfdataValue("sum_messages_sent_per_second", messagesSentPerSecond));
+       perfdata->Add(new PerfdataValue("sum_messages_received_per_second", messagesReceivedPerSecond));
+       perfdata->Add(new PerfdataValue("sum_bytes_sent_per_second", bytesSentPerSecond));
+       perfdata->Add(new PerfdataValue("sum_bytes_received_per_second", bytesReceivedPerSecond));
        cr->SetPerformanceData(perfdata);
 
        checkable->ProcessCheckResult(cr);

diff --git a/lib/methods/icingachecktask.cpp b/lib/methods/icingachecktask.cpp
index 3a4b668a9526ccbfd4fe6111021cf35533f5d719..a5180982844f4fc01ae8db91cb494e2e7201abd0 100644 (file)
--- a/lib/methods/icingachecktask.cpp
+++ b/lib/methods/icingachecktask.cpp
@@ -100,6 +100,36 @@ void IcingaCheckTask::ScriptFunc(const Checkable::Ptr& service, const CheckResul
        perfdata->Add(new PerfdataValue("num_hosts_in_downtime", hs.hosts_in_downtime));
        perfdata->Add(new PerfdataValue("num_hosts_acknowledged", hs.hosts_acknowledged));
 
+       std::vector<Endpoint::Ptr> endpoints = ConfigType::GetObjectsByType<Endpoint>();
+
+       double lastMessageSent = 0;
+       double lastMessageReceived = 0;
+       double messagesSentPerSecond = 0;
+       double messagesReceivedPerSecond = 0;
+       double bytesSentPerSecond = 0;
+       double bytesReceivedPerSecond = 0;
+
+       for (Endpoint::Ptr endpoint : endpoints)
+       {
+               if (endpoint->GetLastMessageSent() > lastMessageSent)
+                       lastMessageSent = endpoint->GetLastMessageSent();
+
+               if (endpoint->GetLastMessageReceived() > lastMessageReceived)
+                       lastMessageReceived = endpoint->GetLastMessageReceived();
+
+               messagesSentPerSecond += endpoint->GetMessagesSentPerSecond();
+               messagesReceivedPerSecond += endpoint->GetMessagesReceivedPerSecond();
+               bytesSentPerSecond += endpoint->GetBytesSentPerSecond();
+               bytesReceivedPerSecond += endpoint->GetBytesReceivedPerSecond();
+       }
+
+       perfdata->Add(new PerfdataValue("last_messages_sent", lastMessageSent));
+       perfdata->Add(new PerfdataValue("last_messages_received", lastMessageReceived));
+       perfdata->Add(new PerfdataValue("sum_messages_sent_per_second", messagesSentPerSecond));
+       perfdata->Add(new PerfdataValue("sum_messages_received_per_second", messagesReceivedPerSecond));
+       perfdata->Add(new PerfdataValue("sum_bytes_sent_per_second", bytesSentPerSecond));
+       perfdata->Add(new PerfdataValue("sum_bytes_received_per_second", bytesReceivedPerSecond));
+
        cr->SetOutput("Icinga 2 has been running for " + Utility::FormatDuration(uptime) +
            ". Version: " + Application::GetAppVersion());
        cr->SetPerformanceData(perfdata);

diff --git a/lib/remote/apilistener.cpp b/lib/remote/apilistener.cpp
index 86cc797594f30b0fa41c01591f660747d1572415..6743b6e70729423bea7e7fad41607f166b93f133 100644 (file)
--- a/lib/remote/apilistener.cpp
+++ b/lib/remote/apilistener.cpp
@@ -1144,7 +1144,8 @@ void ApiListener::ReplayLog(const JsonRpcConnection::Ptr& client)
                                }
 
                                try  {
-                                       NetString::WriteStringToStream(client->GetStream(), pmessage->Get("message"));
+                                       size_t bytesSent = NetString::WriteStringToStream(client->GetStream(), pmessage->Get("message"));
+                                       endpoint->AddMessageSent(bytesSent);
                                        count++;
                                } catch (const std::exception& ex) {
                                        Log(LogWarning, "ApiListener")
@@ -1169,7 +1170,8 @@ void ApiListener::ReplayLog(const JsonRpcConnection::Ptr& client)
                                        lmessage->Set("method", "log::SetLogPosition");
                                        lmessage->Set("params", lparams);
 
-                                       JsonRpc::SendMessage(client->GetStream(), lmessage);
+                                       size_t bytesSent = JsonRpc::SendMessage(client->GetStream(), lmessage);
+                                       endpoint->AddMessageSent(bytesSent);
                                }
                        }
 

diff --git a/lib/remote/endpoint.cpp b/lib/remote/endpoint.cpp
index e922cd09c592915c5c84743309022a32acdc8082..b231be1651a0a2ae44a41403cdce2e06c8acbe64 100644 (file)
--- a/lib/remote/endpoint.cpp
+++ b/lib/remote/endpoint.cpp
@@ -34,6 +34,10 @@ 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;
 
+Endpoint::Endpoint(void)
+    : m_MessagesSent(60), m_BytesSent(60), m_MessagesReceived(60), m_BytesReceived(60)
+{ }
+
 void Endpoint::OnAllConfigLoaded(void)
 {
        ObjectImpl<Endpoint>::OnAllConfigLoaded();
@@ -117,3 +121,39 @@ Endpoint::Ptr Endpoint::GetLocalEndpoint(void)
 
        return listener->GetLocalEndpoint();
 }
+
+void Endpoint::AddMessageSent(int bytes)
+{
+       double time = Utility::GetTime();
+       m_MessagesSent.InsertValue(time, 1);
+       m_BytesSent.InsertValue(time, bytes);
+       SetLastMessageSent(time);
+}
+
+void Endpoint::AddMessageReceived(int bytes)
+{
+       double time = Utility::GetTime();
+       m_MessagesReceived.InsertValue(time, 1);
+       m_BytesReceived.InsertValue(time, bytes);
+       SetLastMessageReceived(time);
+}
+
+double Endpoint::GetMessagesSentPerSecond(void) const
+{
+       return m_MessagesSent.CalculateRate(Utility::GetTime(), 60);
+}
+
+double Endpoint::GetMessagesReceivedPerSecond(void) const
+{
+       return m_MessagesReceived.CalculateRate(Utility::GetTime(), 60);
+}
+
+double Endpoint::GetBytesSentPerSecond(void) const
+{
+       return m_BytesSent.CalculateRate(Utility::GetTime(), 60);
+}
+
+double Endpoint::GetBytesReceivedPerSecond(void) const
+{
+       return m_BytesReceived.CalculateRate(Utility::GetTime(), 60);
+}

diff --git a/lib/remote/endpoint.hpp b/lib/remote/endpoint.hpp
index 380c1e5c276880f3e0de9a1596472afb10e80635..5570ea5b57c4dd9dc163fa5bd07f8ac950f744ea 100644 (file)
--- a/lib/remote/endpoint.hpp
+++ b/lib/remote/endpoint.hpp
@@ -22,6 +22,7 @@
 
 #include "remote/i2-remote.hpp"
 #include "remote/endpoint.thpp"
+#include "base/ringbuffer.hpp"
 #include <set>
 
 namespace icinga
@@ -41,6 +42,8 @@ public:
        DECLARE_OBJECT(Endpoint);
        DECLARE_OBJECTNAME(Endpoint);
 
+       Endpoint(void);
+
        static boost::signals2::signal<void(const Endpoint::Ptr&, const intrusive_ptr<JsonRpcConnection>&)> OnConnected;
        static boost::signals2::signal<void(const Endpoint::Ptr&, const intrusive_ptr<JsonRpcConnection>&)> OnDisconnected;
 
@@ -56,6 +59,15 @@ public:
 
        void SetCachedZone(const intrusive_ptr<Zone>& zone);
 
+       void AddMessageSent(int bytes);
+       void AddMessageReceived(int bytes);
+
+       double GetMessagesSentPerSecond(void) const override;
+       double GetMessagesReceivedPerSecond(void) const override;
+
+       double GetBytesSentPerSecond(void) const override;
+       double GetBytesReceivedPerSecond(void) const override;
+
 protected:
        virtual void OnAllConfigLoaded(void) override;
 
@@ -63,6 +75,11 @@ private:
        mutable boost::mutex m_ClientsLock;
        std::set<intrusive_ptr<JsonRpcConnection> > m_Clients;
        intrusive_ptr<Zone> m_Zone;
+
+       mutable RingBuffer m_MessagesSent;
+       mutable RingBuffer m_MessagesReceived;
+       mutable RingBuffer m_BytesSent;
+       mutable RingBuffer m_BytesReceived;
 };
 
 }

diff --git a/lib/remote/endpoint.ti b/lib/remote/endpoint.ti
index edf9a58f09a4ca1543c4b6ff04f206171561f45c..e6964200362dc4285ce03f77d9196b5aca64cec2 100644 (file)
--- a/lib/remote/endpoint.ti
+++ b/lib/remote/endpoint.ti
@@ -45,6 +45,25 @@ class Endpoint : ConfigObject
        [no_user_modify, no_storage] bool connected {
                get;
        };
+
+       Timestamp last_message_sent;
+       Timestamp last_message_received;
+
+       [no_user_modify, no_storage] double messages_sent_per_second {
+               get;
+       };
+
+       [no_user_modify, no_storage] double messages_received_per_second {
+               get;
+       };
+
+       [no_user_modify, no_storage] double bytes_sent_per_second {
+               get;
+       };
+
+       [no_user_modify, no_storage] double bytes_received_per_second {
+               get;
+       };
 };
 
 }

diff --git a/lib/remote/jsonrpcconnection.cpp b/lib/remote/jsonrpcconnection.cpp
index 3d9cacaf0c0c55eaef6f6dfffcb15a5dab8b0eff..23e9e27ac5650f0e7d6b60c6f3e0421c4cced634 100644 (file)
--- a/lib/remote/jsonrpcconnection.cpp
+++ b/lib/remote/jsonrpcconnection.cpp
@@ -111,7 +111,8 @@ void JsonRpcConnection::SendMessage(const Dictionary::Ptr& message)
                ObjectLock olock(m_Stream);
                if (m_Stream->IsEof())
                        return;
-               JsonRpc::SendMessage(m_Stream, message);
+               size_t bytesSent = JsonRpc::SendMessage(m_Stream, message);
+               m_Endpoint->AddMessageSent(bytesSent);
        } catch (const std::exception& ex) {
                std::ostringstream info;
                info << "Error while sending JSON-RPC message for identity '" << m_Identity << "'";
@@ -182,6 +183,8 @@ void JsonRpcConnection::MessageHandler(const String& jsonString)
                        origin->FromZone = m_Endpoint->GetZone();
                else
                        origin->FromZone = Zone::GetByName(message->Get("originZone"));
+
+               m_Endpoint->AddMessageReceived(jsonString.GetLength());
        }
 
        Value vmethod;