Add stats about connection count in $SYS

This required adding a global stats object. It also contains a bit of refactor to make a type out of the derived counters.

Add stats about connection count in $SYS
This required adding a global stats object. It also contains a bit of refactor to make a type out of the derived counters.
Wiebe Cazemier · Wiebe Cazemier
1 parent bf2193f9
Showing 11 changed files with 134 additions and 70 deletions
CMakeLists.txt
FlashMQTests/FlashMQTests.pro
client.cpp
derivablecounter.cpp
derivablecounter.h
globalstats.cpp
globalstats.h
mainapp.cpp
mqttpacket.cpp
threaddata.cpp
threaddata.h
@@ -61,6 +61,8 @@ add_executable(FlashMQ
     publishcopyfactory.h
     variablebyteint.h
     mqtt5properties.h
+    globalstats.h
+    derivablecounter.h
     mainapp.cpp
     main.cpp
@@ -101,6 +103,8 @@ add_executable(FlashMQ
     publishcopyfactory.cpp
     variablebyteint.cpp
     mqtt5properties.cpp
+    globalstats.cpp
+    derivablecounter.cpp
     )
@@ -52,6 +52,8 @@ SOURCES +=  tst_maintests.cpp \
     ../publishcopyfactory.cpp \
     ../variablebyteint.cpp \
     ../mqtt5properties.cpp \
+    ../globalstats.cpp \
+    ../derivablecounter.cpp \
     mainappthread.cpp \
     twoclienttestcontext.cpp
@@ -96,6 +98,8 @@ HEADERS += \
     ../publishcopyfactory.h \
     ../variablebyteint.h \
     ../mqtt5properties.h \
+    ../globalstats.h \
+    ../derivablecounter.h \
     mainappthread.h \
     twoclienttestcontext.h
@@ -223,7 +223,7 @@ void Client::writeMqttPacket(const MqttPacket &amp;packet)
     if (packet.packetType == PacketType::PUBLISH)
     {
         ThreadData *td = ThreadGlobals::getThreadData();
-        td->incrementSentMessageCount(1);
+        td->sentMessageCounter.inc();
     }
     else if (packet.packetType == PacketType::DISCONNECT)
         setReadyForDisconnect();
+#include "derivablecounter.h"
+
+void DerivableCounter::inc(uint64_t n)
+{
+    val += n;
+}
+
+uint64_t DerivableCounter::get() const
+{
+    return val;
+}
+
+/**
+ * @brief DerivableCounter::getPerSecond returns the amount per second since last time this method was called.
+ * @return
+ *
+ * Even though the class it not meant to be thread-safe, this method does use a mutex, because obtaining the value can be
+ * scheduled in different threads.
+ */
+uint64_t DerivableCounter::getPerSecond()
+{
+    std::lock_guard<std::mutex> locker(timeMutex);
+
+    std::chrono::time_point<std::chrono::steady_clock> now = std::chrono::steady_clock::now();
+    std::chrono::milliseconds msSinceLastTime = std::chrono::duration_cast<std::chrono::milliseconds>(now - timeOfPrevious);
+    uint64_t messagesTimes1000 = (val - valPrevious) * 1000;
+    uint64_t result = messagesTimes1000 / (msSinceLastTime.count() + 1); // branchless avoidance of div by 0;
+    timeOfPrevious = now;
+    valPrevious = val;
+    return result;
+}
+#ifndef DERIVABLECOUNTER_H
+#define DERIVABLECOUNTER_H
+
+#include <chrono>
+#include <mutex>
+
+/**
+ * @brief The DerivableCounter is a counter which can derive val/dt.
+ *
+ * It's not thread-safe, to avoid unnecessary locking. You should have counters per thread.
+ */
+class DerivableCounter
+{
+    uint64_t val = 0;
+    uint64_t valPrevious = 0;
+    std::chrono::time_point<std::chrono::steady_clock> timeOfPrevious = std::chrono::steady_clock::now();
+    std::mutex timeMutex;
+
+public:
+
+    void inc(uint64_t n = 1);
+    uint64_t get() const;
+    uint64_t getPerSecond();
+};
+
+#endif // DERIVABLECOUNTER_H
+#include "globalstats.h"
+
+GlobalStats *GlobalStats::instance = nullptr;
+
+GlobalStats::GlobalStats()
+{
+
+}
+
+GlobalStats *GlobalStats::getInstance()
+{
+    if (GlobalStats::instance == nullptr)
+        GlobalStats::instance = new GlobalStats();
+
+    return GlobalStats::instance;
+}
+
+#ifndef GLOBALSTATS_H
+#define GLOBALSTATS_H
+
+#include "stdint.h"
+#include "derivablecounter.h"
+
+class GlobalStats
+{
+    static GlobalStats *instance;
+
+    GlobalStats();
+public:
+    static GlobalStats *getInstance();
+
+    DerivableCounter socketConnects;
+};
+
+#endif // GLOBALSTATS_H
@@ -33,6 +33,7 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include "threadloop.h"
 #include "authplugin.h"
 #include "threadglobals.h"
+#include "globalstats.h"
 MainApp *MainApp::instance = nullptr;
@@ -529,6 +530,8 @@ void MainApp::start()
     }
 #endif
+    GlobalStats *globalStats = GlobalStats::getInstance();
+
     for (int i = 0; i < num_threads; i++)
     {
         std::shared_ptr<ThreadData> t = std::make_shared<ThreadData>(i, subscriptionStore, settings);
@@ -592,6 +595,8 @@ void MainApp::start()
                     std::shared_ptr<Client> client = std::make_shared<Client>(fd, thread_data, clientSSL, listener->websocket, addr, settings.get());
                     thread_data->giveClient(client);
+
+                    globalStats->socketConnects.inc();
                 }
                 else
                 {
@@ -328,6 +328,8 @@ void MqttPacket::handleConnect()
     std::shared_ptr<SubscriptionStore> subscriptionStore = sender->getThreadData()->getSubscriptionStore();
+    sender->getThreadData()->mqttConnectCounter.inc();
+
     uint16_t variable_header_length = readTwoBytesToUInt16();
     const Settings &settings = *ThreadGlobals::getSettings();
@@ -1129,7 +1131,7 @@ void MqttPacket::handlePublish()
     ReasonCodes ackCode = ReasonCodes::Success;
-    sender->getThreadData()->incrementReceivedMessageCount();
+    sender->getThreadData()->receivedMessageCounter.inc();
     Authentication &authentication = *ThreadGlobals::getAuth();
@@ -20,6 +20,8 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include <sstream>
 #include <cassert>
+#include "globalstats.h"
+
 KeepAliveCheck::KeepAliveCheck(const std::shared_ptr<Client> client) :
     client(client)
 {
@@ -143,17 +145,31 @@ void ThreadData::publishStatsOnDollarTopic(std::vector&lt;std::shared_ptr&lt;ThreadDat
     uint64_t sentMessageCountPerSecond = 0;
     uint64_t sentMessageCount = 0;
+    uint64_t mqttConnectCountPerSecond = 0;
+    uint64_t mqttConnectCount = 0;
+
     for (const std::shared_ptr<ThreadData> &thread : threads)
     {
         nrOfClients += thread->getNrOfClients();
-        receivedMessageCountPerSecond += thread->getReceivedMessagePerSecond();
-        receivedMessageCount += thread->getReceivedMessageCount();
+        receivedMessageCountPerSecond += thread->receivedMessageCounter.getPerSecond();
+        receivedMessageCount += thread->receivedMessageCounter.get();
-        sentMessageCountPerSecond += thread->getSentMessagePerSecond();
-        sentMessageCount += thread->getSentMessageCount();
+        sentMessageCountPerSecond += thread->sentMessageCounter.getPerSecond();
+        sentMessageCount += thread->sentMessageCounter.get();
+
+        mqttConnectCountPerSecond += thread->mqttConnectCounter.getPerSecond();
+        mqttConnectCount += thread->mqttConnectCounter.get();
     }
+    GlobalStats *globalStats = GlobalStats::getInstance();
+
+    publishStat("$SYS/broker/network/socketconnects/total", globalStats->socketConnects.get());
+    publishStat("$SYS/broker/network/socketconnects/persecond", globalStats->socketConnects.getPerSecond());
+
+    publishStat("$SYS/broker/clients/mqttconnects/total", mqttConnectCount);
+    publishStat("$SYS/broker/clients/mqttconnects/persecond", mqttConnectCountPerSecond);
+
     publishStat("$SYS/broker/clients/total", nrOfClients);
     publishStat("$SYS/broker/load/messages/received/total", receivedMessageCount);
@@ -391,54 +407,6 @@ int ThreadData::getNrOfClients() const
     return clients_by_fd.size();
 }
-void ThreadData::incrementReceivedMessageCount()
-{
-    receivedMessageCount++;
-}
-
-uint64_t ThreadData::getReceivedMessageCount() const
-{
-    return receivedMessageCount;
-}
-
-/**
- * @brief ThreadData::getReceivedMessagePerSecond gets the amount of seconds received, averaged over the last time this was called.
- * @return
- *
- * Locking is not required, because the counter is not written to from here.
- */
-uint64_t ThreadData::getReceivedMessagePerSecond()
-{
-    std::chrono::time_point<std::chrono::steady_clock> now = std::chrono::steady_clock::now();
-    std::chrono::milliseconds msSinceLastTime = std::chrono::duration_cast<std::chrono::milliseconds>(now - receivedMessagePreviousTime);
-    uint64_t messagesTimes1000 = (receivedMessageCount - receivedMessageCountPrevious) * 1000;
-    uint64_t result = messagesTimes1000 / (msSinceLastTime.count() + 1); // branchless avoidance of div by 0;
-    receivedMessagePreviousTime = now;
-    receivedMessageCountPrevious = receivedMessageCount;
-    return result;
-}
-
-void ThreadData::incrementSentMessageCount(uint64_t n)
-{
-    sentMessageCount += n;
-}
-
-uint64_t ThreadData::getSentMessageCount() const
-{
-    return sentMessageCount;
-}
-
-uint64_t ThreadData::getSentMessagePerSecond()
-{
-    std::chrono::time_point<std::chrono::steady_clock> now = std::chrono::steady_clock::now();
-    std::chrono::milliseconds msSinceLastTime = std::chrono::duration_cast<std::chrono::milliseconds>(now - sentMessagePreviousTime);
-    uint64_t messagesTimes1000 = (sentMessageCount - sentMessageCountPrevious) * 1000;
-    uint64_t result = messagesTimes1000 / (msSinceLastTime.count() + 1); // branchless avoidance of div by 0;
-    sentMessagePreviousTime = now;
-    sentMessageCountPrevious = sentMessageCount;
-    return result;
-}
-
 void ThreadData::queueAuthPluginPeriodicEvent()
 {
     std::lock_guard<std::mutex> locker(taskQueueMutex);
@@ -37,6 +37,7 @@ License along with FlashMQ. If not, see &lt;https://www.gnu.org/licenses/&gt;.
 #include "configfileparser.h"
 #include "authplugin.h"
 #include "logger.h"
+#include "derivablecounter.h"
 typedef void (*thread_f)(ThreadData *);
@@ -55,14 +56,6 @@ class ThreadData
     std::shared_ptr<SubscriptionStore> subscriptionStore;
     Logger *logger;
-    uint64_t receivedMessageCount = 0;
-    uint64_t receivedMessageCountPrevious = 0;
-    std::chrono::time_point<std::chrono::steady_clock> receivedMessagePreviousTime = std::chrono::steady_clock::now();
-
-    uint64_t sentMessageCount = 0;
-    uint64_t sentMessageCountPrevious = 0;
-    std::chrono::time_point<std::chrono::steady_clock> sentMessagePreviousTime = std::chrono::steady_clock::now();
-
     std::mutex clientsToRemoveMutex;
     std::forward_list<std::weak_ptr<Client>> clientsQueuedForRemoving;
@@ -97,6 +90,10 @@ public:
     std::mutex taskQueueMutex;
     std::forward_list<std::function<void()>> taskQueue;
+    DerivableCounter receivedMessageCounter;
+    DerivableCounter sentMessageCounter;
+    DerivableCounter mqttConnectCounter;
+
     ThreadData(int threadnr, std::shared_ptr<SubscriptionStore> &subscriptionStore, std::shared_ptr<Settings> settings);
     ThreadData(const ThreadData &other) = delete;
     ThreadData(ThreadData &&other) = delete;
@@ -124,14 +121,6 @@ public:
     int getNrOfClients() const;
-    void incrementReceivedMessageCount();
-    uint64_t getReceivedMessageCount() const;
-    uint64_t getReceivedMessagePerSecond();
-
-    void incrementSentMessageCount(uint64_t n);
-    uint64_t getSentMessageCount() const;
-    uint64_t getSentMessagePerSecond();
-
     void queueAuthPluginPeriodicEvent();
     void authPluginPeriodicEvent();