feat: Add RsCumulativeTrafficStats for cumulative traffic tracking per peer/service

Author: ayushsingh658

src/retroshare/rsconfig.h

@@ -212,6 +212,48 @@ struct RSTrafficClue : RsSerializable
 	}
 };
 
+/*!
+ * \brief Cumulative traffic statistics for tracking all-time data transfer
+ * Used to persist and display per-peer and per-service data usage
+ */
+struct RsCumulativeTrafficStats : RsSerializable
+{
+    uint64_t bytesIn;      //< Total bytes received
+    uint64_t bytesOut;     //< Total bytes sent
+    uint32_t countIn;      //< Number of incoming packets
+    uint32_t countOut;     //< Number of outgoing packets
+    rstime_t firstSeen;    //< Timestamp of first recorded traffic
+    rstime_t lastSeen;     //< Timestamp of most recent traffic
+
+    RsCumulativeTrafficStats() : 
+        bytesIn(0), bytesOut(0), countIn(0), countOut(0), 
+        firstSeen(0), lastSeen(0) {}
+
+    RsCumulativeTrafficStats& operator+=(const RsCumulativeTrafficStats& other) {
+        bytesIn += other.bytesIn;
+        bytesOut += other.bytesOut;
+        countIn += other.countIn;
+        countOut += other.countOut;
+        if (firstSeen == 0 || (other.firstSeen != 0 && other.firstSeen < firstSeen))
+            firstSeen = other.firstSeen;
+        if (other.lastSeen > lastSeen)
+            lastSeen = other.lastSeen;
+        return *this;
+    }
+
+    void clear() { bytesIn = bytesOut = countIn = countOut = 0; firstSeen = lastSeen = 0; }
+
+    // RsSerializable interface
+    void serial_process(RsGenericSerializer::SerializeJob j, RsGenericSerializer::SerializeContext &ctx) {
+        RS_SERIAL_PROCESS(bytesIn);
+        RS_SERIAL_PROCESS(bytesOut);
+        RS_SERIAL_PROCESS(countIn);
+        RS_SERIAL_PROCESS(countOut);
+        RS_SERIAL_PROCESS(firstSeen);
+        RS_SERIAL_PROCESS(lastSeen);
+    }
+};
+
 struct RsConfigNetStatus : RsSerializable
 {
 	RsConfigNetStatus() : netLocalOk(true)
@@ -338,6 +380,39 @@ class RsServerConfig
 	 */
     virtual int getTrafficInfo(std::list<RSTrafficClue>& out_lst,std::list<RSTrafficClue>& in_lst) = 0 ;
 
+	/**
+	 * @brief getCumulativeTrafficByPeer returns cumulative traffic stats grouped by peer
+	 * @jsonapi{development}
+	 * @param[out] stats map of peer ID to cumulative traffic stats
+	 * @return returns true on success
+	 */
+    virtual bool getCumulativeTrafficByPeer(std::map<RsPeerId, RsCumulativeTrafficStats>& stats) = 0;
+
+	/**
+	 * @brief getCumulativeTrafficByService returns cumulative traffic stats grouped by service
+	 * @jsonapi{development}
+	 * @param[out] stats map of service ID to cumulative traffic stats
+	 * @return returns true on success
+	 */
+    virtual bool getCumulativeTrafficByService(std::map<uint16_t, RsCumulativeTrafficStats>& stats) = 0;
+
+	/**
+	 * @brief clearCumulativeTraffic clears all cumulative traffic statistics
+	 * @jsonapi{development}
+	 * @param[in] clearPeerStats if true, clears per-peer stats
+	 * @param[in] clearServiceStats if true, clears per-service stats
+	 * @return returns true on success
+	 */
+    virtual bool clearCumulativeTraffic(bool clearPeerStats = true, bool clearServiceStats = true) = 0;
+
+	/**
+	 * @brief getTotalCumulativeTraffic returns the total cumulative traffic across all peers/services
+	 * @jsonapi{development}
+	 * @param[out] stats total cumulative traffic stats
+	 * @return returns true on success
+	 */
+    virtual bool getTotalCumulativeTraffic(RsCumulativeTrafficStats& stats) = 0;
+
     /* From RsInit */
 
     // NOT IMPLEMENTED YET!

src/rsitems/rstrafficstatsitems.h

@@ -0,0 +1,85 @@
+/*******************************************************************************
+ * libretroshare/src/rsitems: rstrafficstatsitems.h                            *
+ *                                                                             *
+ * libretroshare: retroshare core library                                      *
+ *                                                                             *
+ * Copyright (C) 2024                                                          *
+ *                                                                             *
+ * This program is free software: you can redistribute it and/or modify        *
+ * it under the terms of the GNU Lesser General Public License as              *
+ * published by the Free Software Foundation, either version 3 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 Lesser General Public License for more details.                         *
+ *                                                                             *
+ * You should have received a copy of the GNU Lesser General Public License    *
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.       *
+ *                                                                             *
+ *******************************************************************************/
+#ifndef RS_TRAFFIC_STATS_ITEMS_H
+#define RS_TRAFFIC_STATS_ITEMS_H
+
+#include <map>
+
+#include "rsitems/rsitem.h"
+#include "rsitems/rsserviceids.h"
+#include "retroshare/rsconfig.h"
+#include "serialiser/rsserializer.h"
+#include "serialiser/rstypeserializer.h"
+
+// Use BANDWIDTH_CONTROL service type for config items
+const uint8_t RS_PKT_SUBTYPE_TRAFFIC_STATS_ITEM = 0x10;
+
+/**************************************************************************/
+
+class RsTrafficStatsConfigItem : public RsItem
+{
+public:
+    RsTrafficStatsConfigItem() : RsItem(RS_PKT_VERSION_SERVICE, RS_SERVICE_TYPE_BWCTRL, RS_PKT_SUBTYPE_TRAFFIC_STATS_ITEM)
+    {}
+    
+    virtual ~RsTrafficStatsConfigItem() {}
+    
+    virtual void clear() 
+    { 
+        peerStats.clear(); 
+        serviceStats.clear(); 
+    }
+
+    void serial_process(RsGenericSerializer::SerializeJob j, RsGenericSerializer::SerializeContext& ctx)
+    {
+        RsTypeSerializer::serial_process(j, ctx, peerStats, "peerStats");
+        RsTypeSerializer::serial_process(j, ctx, serviceStats, "serviceStats");
+    }
+
+    std::map<RsPeerId, RsCumulativeTrafficStats> peerStats;
+    std::map<uint16_t, RsCumulativeTrafficStats> serviceStats;
+};
+
+class RsTrafficStatsSerialiser : public RsServiceSerializer
+{
+public:
+    RsTrafficStatsSerialiser() : RsServiceSerializer(RS_SERVICE_TYPE_BWCTRL) {}
+    virtual ~RsTrafficStatsSerialiser() {}
+
+    RsItem *create_item(uint16_t service, uint8_t item_sub_id) const
+    {
+        if (service != RS_SERVICE_TYPE_BWCTRL)
+            return nullptr;
+
+        switch (item_sub_id)
+        {
+        case RS_PKT_SUBTYPE_TRAFFIC_STATS_ITEM:
+            return new RsTrafficStatsConfigItem();
+        default:
+            return nullptr;
+        }
+    }
+};
+
+/**************************************************************************/
+
+#endif /* RS_TRAFFIC_STATS_ITEMS_H */

src/rsserver/p3serverconfig.cc

@@ -22,6 +22,7 @@
 #include <retroshare/rsturtle.h>
 #include "rsserver/p3serverconfig.h"
 #include "services/p3bwctrl.h"
+#include "rsitems/rstrafficstatsitems.h"
 
 #include "pqi/authgpg.h"
 #include "pqi/authssl.h"
@@ -37,7 +38,7 @@ static constexpr float DEFAULT_UPLOAD_KB_RATE   = 10000.0;
 static constexpr float MIN_MINIMAL_RATE = 5.0;
 
 
-p3ServerConfig::p3ServerConfig(p3PeerMgr *peerMgr, p3LinkMgr *linkMgr, p3NetMgr *netMgr, pqihandler *pqih, p3GeneralConfig *genCfg)
+p3ServerConfig::p3ServerConfig(p3PeerMgr *peerMgr, p3LinkMgr *linkMgr, p3NetMgr *netMgr, pqihandler *pqih, p3GeneralConfig *genCfg, p3ConfigMgr *cfgMgr)
     :  mPeerMgr(peerMgr), mLinkMgr(linkMgr), mNetMgr(netMgr), mPqiHandler(pqih)
     , mGeneralConfig(genCfg)
     , configMtx("p3ServerConfig")
@@ -46,6 +47,9 @@ p3ServerConfig::p3ServerConfig(p3PeerMgr *peerMgr, p3LinkMgr *linkMgr, p3NetMgr
     , mRateDownloadWhenIdle(DEFAULT_DOWNLOAD_KB_RATE), mRateUploadWhenIdle(DEFAULT_UPLOAD_KB_RATE)
     , mIsIdle(false), mOpMode(RsOpMode::FULL)
 {
+    // Register with config manager for persistence
+    if (cfgMgr)
+        cfgMgr->addConfiguration("traffic_stats.cfg", this);
 }
 
 void p3ServerConfig::load_config()
@@ -184,6 +188,108 @@ int p3ServerConfig::getTrafficInfo(std::list<RSTrafficClue>& out_lst,std::list<R
         return 0 ;
 }
 
+bool p3ServerConfig::getCumulativeTrafficByPeer(std::map<RsPeerId, RsCumulativeTrafficStats>& stats)
+{
+    RsStackMutex stack(configMtx);
+
+    // First, update cumulative stats from current traffic clues
+    std::list<RSTrafficClue> out_lst, in_lst;
+    if (rsBandwidthControl)
+    {
+        rsBandwidthControl->ExtractTrafficInfo(out_lst, in_lst);
+        
+        rstime_t now = time(nullptr);
+        
+        // Accumulate outgoing traffic
+        for (const auto& clue : out_lst)
+        {
+            auto& peerStats = mCumulativeTrafficByPeer[clue.peer_id];
+            peerStats.bytesOut += clue.size;
+            peerStats.countOut += clue.count;
+            if (peerStats.firstSeen == 0) peerStats.firstSeen = now;
+            peerStats.lastSeen = now;
+        }
+        
+        // Accumulate incoming traffic
+        for (const auto& clue : in_lst)
+        {
+            auto& peerStats = mCumulativeTrafficByPeer[clue.peer_id];
+            peerStats.bytesIn += clue.size;
+            peerStats.countIn += clue.count;
+            if (peerStats.firstSeen == 0) peerStats.firstSeen = now;
+            peerStats.lastSeen = now;
+        }
+    }
+    
+    stats = mCumulativeTrafficByPeer;
+    return true;
+}
+
+bool p3ServerConfig::getCumulativeTrafficByService(std::map<uint16_t, RsCumulativeTrafficStats>& stats)
+{
+    RsStackMutex stack(configMtx);
+
+    // First, update cumulative stats from current traffic clues
+    std::list<RSTrafficClue> out_lst, in_lst;
+    if (rsBandwidthControl)
+    {
+        rsBandwidthControl->ExtractTrafficInfo(out_lst, in_lst);
+        
+        rstime_t now = time(nullptr);
+        
+        // Accumulate outgoing traffic by service
+        for (const auto& clue : out_lst)
+        {
+            auto& serviceStats = mCumulativeTrafficByService[clue.service_id];
+            serviceStats.bytesOut += clue.size;
+            serviceStats.countOut += clue.count;
+            if (serviceStats.firstSeen == 0) serviceStats.firstSeen = now;
+            serviceStats.lastSeen = now;
+        }
+        
+        // Accumulate incoming traffic by service
+        for (const auto& clue : in_lst)
+        {
+            auto& serviceStats = mCumulativeTrafficByService[clue.service_id];
+            serviceStats.bytesIn += clue.size;
+            serviceStats.countIn += clue.count;
+            if (serviceStats.firstSeen == 0) serviceStats.firstSeen = now;
+            serviceStats.lastSeen = now;
+        }
+    }
+    
+    stats = mCumulativeTrafficByService;
+    return true;
+}
+
+bool p3ServerConfig::clearCumulativeTraffic(bool clearPeerStats, bool clearServiceStats)
+{
+    RsStackMutex stack(configMtx);
+    
+    if (clearPeerStats)
+        mCumulativeTrafficByPeer.clear();
+    
+    if (clearServiceStats)
+        mCumulativeTrafficByService.clear();
+    
+    return true;
+}
+
+bool p3ServerConfig::getTotalCumulativeTraffic(RsCumulativeTrafficStats& stats)
+{
+    RsStackMutex stack(configMtx);
+    
+    stats.clear();
+    
+    // Sum up all peer stats
+    for (const auto& pair : mCumulativeTrafficByPeer)
+    {
+        stats += pair.second;
+    }
+    
+    return true;
+}
+
 int 	p3ServerConfig::getTotalBandwidthRates(RsConfigDataRates &rates)
 {
 	if (rsBandwidthControl)
@@ -562,3 +668,45 @@ void p3ServerConfig::setIsIdle(bool isIdle)
 	mIsIdle = isIdle;
 }
 
+/********************* p3Config persistence methods *******/
+
+RsSerialiser *p3ServerConfig::setupSerialiser()
+{
+    RsSerialiser *rss = new RsSerialiser();
+    rss->addSerialType(new RsTrafficStatsSerialiser());
+    return rss;
+}
+
+bool p3ServerConfig::saveList(bool &cleanup, std::list<RsItem *>& items)
+{
+    cleanup = true;
+
+    RsTrafficStatsConfigItem *item = new RsTrafficStatsConfigItem();
+    {
+        RsStackMutex stack(configMtx);
+        item->peerStats = mCumulativeTrafficByPeer;
+        item->serviceStats = mCumulativeTrafficByService;
+    }
+    items.push_back(item);
+    
+    return true;
+}
+
+bool p3ServerConfig::loadList(std::list<RsItem *>& load)
+{
+    RsStackMutex stack(configMtx);
+    
+    for (auto it = load.begin(); it != load.end(); ++it)
+    {
+        RsTrafficStatsConfigItem *item = dynamic_cast<RsTrafficStatsConfigItem*>(*it);
+        if (item)
+        {
+            mCumulativeTrafficByPeer = item->peerStats;
+            mCumulativeTrafficByService = item->serviceStats;
+        }
+        delete *it;
+    }
+    load.clear();
+    
+    return true;
+}