diff --git a/simplexmq.cabal b/simplexmq.cabal index bbe7583fa..b21ce714b 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -95,6 +95,7 @@ library Simplex.Messaging.Agent.Protocol Simplex.Messaging.Agent.QueryString Simplex.Messaging.Agent.RetryInterval + Simplex.Messaging.Agent.Stats Simplex.Messaging.Agent.Store Simplex.Messaging.Agent.Store.SQLite Simplex.Messaging.Agent.Store.SQLite.Common diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index af9eef919..9ae6ae357 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -170,7 +170,6 @@ import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import Data.Either (partitionEithers) import Data.Functor (($>)) -import Data.Int (Int64) import Data.List (deleteFirstsBy, foldl', partition, (\\)) import Data.List.NonEmpty (NonEmpty (..), (<|)) import qualified Data.List.NonEmpty as L @@ -198,6 +197,7 @@ import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Lock import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval +import Simplex.Messaging.Agent.Stats import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction) import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB @@ -317,7 +317,9 @@ data AgentClient = AgentClient agentStats :: TMap AgentStatsKey (TVar Int), msgCounts :: TMap ConnId (TVar (Int, Int)), -- (total, duplicates) clientId :: Int, - agentEnv :: Env + agentEnv :: Env, + smpServersStats :: TMap (UserId, SMPServer) AgentSMPServerStats, + xftpServersStats :: TMap (UserId, XFTPServer) AgentXFTPServerStats } data SMPConnectedClient = SMPConnectedClient @@ -484,6 +486,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = smpSubWorkers <- TM.empty agentStats <- TM.empty msgCounts <- TM.empty + smpServersStats <- TM.empty + xftpServersStats <- TM.empty return AgentClient { acThread, @@ -522,7 +526,9 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = agentStats, msgCounts, clientId, - agentEnv + agentEnv, + smpServersStats, + xftpServersStats } slowNetworkConfig :: NetworkConfig -> NetworkConfig @@ -1908,46 +1914,53 @@ withNextSrv c userId usedSrvs initUsed action = do writeTVar usedSrvs $! used' action srvAuth +-- - currently used servers - those that have state +-- - previously used servers - have stats but no state - they could be used earlier in session, +-- or in previous sessions and stats for them were restored +-- - proxied destination servers (onlyProxiedSMPServers) - based on smpProxiedRelays (key) minus smpClients; +-- don't have state and stats? since: state is based on smpClients and activeSubs/pendingSubs, +-- and stats are used to track real network activity (probably also via smpClients), +-- so they [stats] would be accounted for proxy (unless we want double counting messages as sent and proxied) data AgentServersSummary = AgentServersSummary - { usersServersSummary :: Map UserId ServersSummary - -- totalServersSummary :: ServersSummary -- compute in backend to not repeat logic in UI? + { smpServersState :: Map (UserId, SMPServer) SMPServerState, + smpServersStatsData :: Map (UserId, SMPServer) AgentSMPServerStatsData, + onlyProxiedSMPServers :: [SMPServer], + xftpServersState :: Map (UserId, XFTPServer) XFTPServerState, + xftpServersStatsData :: Map (UserId, XFTPServer) AgentXFTPServerStatsData } deriving (Show) -data ServersSummary = ServersSummary - { smpServersSummary :: [SMPServerSummary], - xftpServersSummary :: [XFTPServerSummary] +data SMPServerState = SMPServerState + { sessions :: Int, -- number of active sessions, based on smpClients - SMPClientVar has client + sessionErrs :: Int, -- number of sessions with errors, based on smpClients - SMPClientVar has error + activeSubscriptions :: Int, -- based on activeSubs + pendingSubscriptions :: Int -- based on pendingSubs } deriving (Show) -data SMPServerSummary = SMPServerSummary - { smpServer :: SMPServer, - usedForNewConnections :: Bool, - activeSubscriptions :: [SMPServerSubRcvId], -- or even simply count? - pendingSubscriptions :: [SMPServerSubRcvId], - rcvMsgs :: Int64, - duplicateRcvMsgs :: Int64, -- tracking it per server allows to easier detect it before checking detailed info - sndMsgs :: Int64 - } - deriving (Show) - -type SMPServerSubRcvId = Text - -data XFTPServerSummary = XFTPServerSummary - { xftpServer :: XFTPServer, - usedForNewFiles :: Bool, - rcvFilesCount :: Int64, - rcvInProgress :: Bool, - sndFilesCount :: Int64, - sndInProgress :: Bool, - delFilesCount :: Int64, - delInProgress :: Bool +data XFTPServerState = XFTPServerState + { sessions :: Int, -- number of active sessions, based on xftpClients - XFTPClientVar has client + sessionErrs :: Int, -- number of sessions with errors, based on xftpClients - XFTPClientVar has error + rcvInProgress :: Bool, -- based on xftpRcvWorkers, hasWork + sndInProgress :: Bool, -- based on xftpSndWorkers, hasWork + delInProgress :: Bool -- based on xftpDelWorkers, hasWork } deriving (Show) getAgentServersSummary :: AgentClient -> IO AgentServersSummary -getAgentServersSummary _c = do - pure AgentServersSummary {usersServersSummary = M.empty} +getAgentServersSummary AgentClient {smpServersStats, xftpServersStats} = do + sss <- readTVarIO smpServersStats + smpServersStatsData <- mapM (atomically . getAgentSMPServerStats) sss + xss <- readTVarIO xftpServersStats + xftpServersStatsData <- mapM (atomically . getAgentXFTPServerStats) xss + pure + AgentServersSummary + { smpServersState = M.empty, -- collect, see SMPServerState + smpServersStatsData, + onlyProxiedSMPServers = [], -- collect, smpProxiedRelays (key) minus smpClients + xftpServersState = M.empty, -- collect, see XFTPServerState + xftpServersStatsData + } data SubInfo = SubInfo {userId :: UserId, server :: Text, rcvId :: Text, subError :: Maybe String} deriving (Show) @@ -2139,11 +2152,9 @@ $(J.deriveJSON (enumJSON $ dropPrefix "TS") ''ProtocolTestStep) $(J.deriveJSON defaultJSON ''ProtocolTestFailure) -$(J.deriveJSON defaultJSON ''XFTPServerSummary) +$(J.deriveJSON defaultJSON ''XFTPServerState) -$(J.deriveJSON defaultJSON ''SMPServerSummary) - -$(J.deriveJSON defaultJSON ''ServersSummary) +$(J.deriveJSON defaultJSON ''SMPServerState) $(J.deriveJSON defaultJSON ''AgentServersSummary) diff --git a/src/Simplex/Messaging/Agent/Stats.hs b/src/Simplex/Messaging/Agent/Stats.hs new file mode 100644 index 000000000..f841d02df --- /dev/null +++ b/src/Simplex/Messaging/Agent/Stats.hs @@ -0,0 +1,300 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE TemplateHaskell #-} + +module Simplex.Messaging.Agent.Stats where + +import qualified Data.Aeson.TH as J +import Data.Time.Clock (UTCTime (..)) +import Simplex.Messaging.Parsers (defaultJSON) +import UnliftIO.STM + +data AgentSMPServerStats = AgentSMPServerStats + { fromTime :: TVar UTCTime, + msgSent :: TVar Int, -- total messages sent to server directly + msgSentRetries :: TVar Int, -- sending retries + msgSentSuccesses :: TVar Int, -- successful sends + msgSentAuth :: TVar Int, -- send AUTH errors + msgSentQuota :: TVar Int, -- send QUOTA errors + msgSentExpired :: TVar Int, -- send expired errors + msgSentErr :: TVar Int, -- other send errors (excluding above) + msgProx :: TVar Int, -- total messages sent to proxy server for forwarding + msgProxRetries :: TVar Int, -- proxy sending retries + msgProxSuccesses :: TVar Int, -- successful proxy sends + msgProxExpired :: TVar Int, -- proxy send expired errors + msgProxAuth :: TVar Int, -- proxy send AUTH errors (AUTH error at destination relay) + msgProxQuota :: TVar Int, -- proxy send QUOTA errors (QUOTA error at destination relay) + msgProxErr :: TVar Int, -- other proxy send errors (excluding above) + msgRecv :: TVar Int, -- total messages received + msgRecvDuplicate :: TVar Int, -- duplicate messages received + msgRecvErr :: TVar Int, -- receive errors + sub :: TVar Int, -- total subscriptions + subRetries :: TVar Int, -- subscription retries + subErr :: TVar Int -- subscription errors + } + +data AgentSMPServerStatsData = AgentSMPServerStatsData + { _fromTime :: UTCTime, + _msgSent :: Int, + _msgSentRetries :: Int, + _msgSentSuccesses :: Int, + _msgSentAuth :: Int, + _msgSentQuota :: Int, + _msgSentExpired :: Int, + _msgSentErr :: Int, + _msgProx :: Int, + _msgProxRetries :: Int, + _msgProxSuccesses :: Int, + _msgProxExpired :: Int, + _msgProxAuth :: Int, + _msgProxQuota :: Int, + _msgProxErr :: Int, + _msgRecv :: Int, + _msgRecvDuplicate :: Int, + _msgRecvErr :: Int, + _sub :: Int, + _subRetries :: Int, + _subErr :: Int + } + deriving (Show) + +newAgentSMPServerStats :: UTCTime -> STM AgentSMPServerStats +newAgentSMPServerStats ts = do + fromTime <- newTVar ts + msgSent <- newTVar 0 + msgSentRetries <- newTVar 0 + msgSentSuccesses <- newTVar 0 + msgSentAuth <- newTVar 0 + msgSentQuota <- newTVar 0 + msgSentExpired <- newTVar 0 + msgSentErr <- newTVar 0 + msgProx <- newTVar 0 + msgProxRetries <- newTVar 0 + msgProxSuccesses <- newTVar 0 + msgProxExpired <- newTVar 0 + msgProxAuth <- newTVar 0 + msgProxQuota <- newTVar 0 + msgProxErr <- newTVar 0 + msgRecv <- newTVar 0 + msgRecvDuplicate <- newTVar 0 + msgRecvErr <- newTVar 0 + sub <- newTVar 0 + subRetries <- newTVar 0 + subErr <- newTVar 0 + pure + AgentSMPServerStats + { fromTime, + msgSent, + msgSentRetries, + msgSentSuccesses, + msgSentAuth, + msgSentQuota, + msgSentExpired, + msgSentErr, + msgProx, + msgProxRetries, + msgProxSuccesses, + msgProxExpired, + msgProxAuth, + msgProxQuota, + msgProxErr, + msgRecv, + msgRecvDuplicate, + msgRecvErr, + sub, + subRetries, + subErr + } + +getAgentSMPServerStats :: AgentSMPServerStats -> STM AgentSMPServerStatsData +getAgentSMPServerStats s@AgentSMPServerStats {fromTime} = do + _fromTime <- readTVar fromTime + _msgSent <- readTVar $ msgSent s + _msgSentRetries <- readTVar $ msgSentRetries s + _msgSentSuccesses <- readTVar $ msgSentSuccesses s + _msgSentAuth <- readTVar $ msgSentAuth s + _msgSentQuota <- readTVar $ msgSentQuota s + _msgSentExpired <- readTVar $ msgSentExpired s + _msgSentErr <- readTVar $ msgSentErr s + _msgProx <- readTVar $ msgProx s + _msgProxRetries <- readTVar $ msgProxRetries s + _msgProxSuccesses <- readTVar $ msgProxSuccesses s + _msgProxExpired <- readTVar $ msgProxExpired s + _msgProxAuth <- readTVar $ msgProxAuth s + _msgProxQuota <- readTVar $ msgProxQuota s + _msgProxErr <- readTVar $ msgProxErr s + _msgRecv <- readTVar $ msgRecv s + _msgRecvDuplicate <- readTVar $ msgRecvDuplicate s + _msgRecvErr <- readTVar $ msgRecvErr s + _sub <- readTVar $ sub s + _subRetries <- readTVar $ subRetries s + _subErr <- readTVar $ subErr s + pure + AgentSMPServerStatsData + { _fromTime, + _msgSent, + _msgSentRetries, + _msgSentSuccesses, + _msgSentAuth, + _msgSentQuota, + _msgSentExpired, + _msgSentErr, + _msgProx, + _msgProxRetries, + _msgProxSuccesses, + _msgProxExpired, + _msgProxAuth, + _msgProxQuota, + _msgProxErr, + _msgRecv, + _msgRecvDuplicate, + _msgRecvErr, + _sub, + _subRetries, + _subErr + } + +setAgentSMPServerStats :: AgentSMPServerStats -> AgentSMPServerStatsData -> STM () +setAgentSMPServerStats s@AgentSMPServerStats {fromTime} d@AgentSMPServerStatsData {_fromTime} = do + writeTVar fromTime $! _fromTime + writeTVar (msgSent s) $! _msgSent d + writeTVar (msgSentRetries s) $! _msgSentRetries d + writeTVar (msgSentSuccesses s) $! _msgSentSuccesses d + writeTVar (msgSentAuth s) $! _msgSentAuth d + writeTVar (msgSentQuota s) $! _msgSentQuota d + writeTVar (msgSentExpired s) $! _msgSentExpired d + writeTVar (msgSentErr s) $! _msgSentErr d + writeTVar (msgProx s) $! _msgProx d + writeTVar (msgProxRetries s) $! _msgProxRetries d + writeTVar (msgProxSuccesses s) $! _msgProxSuccesses d + writeTVar (msgProxErr s) $! _msgProxErr d + writeTVar (msgRecv s) $! _msgRecv d + writeTVar (msgRecvDuplicate s) $! _msgRecvDuplicate d + writeTVar (msgRecvErr s) $! _msgRecvErr d + writeTVar (sub s) $! _sub d + writeTVar (subRetries s) $! _subRetries d + writeTVar (subErr s) $! _subErr d + +data AgentXFTPServerStats = AgentXFTPServerStats + { fromTime :: TVar UTCTime, + replUpload :: TVar Int, -- total replicas uploaded to server + replUploadRetries :: TVar Int, -- upload retries + replUploadSuccesses :: TVar Int, -- successful uploads + replUploadErr :: TVar Int, -- upload errors + replDownload :: TVar Int, -- total replicas downloaded from server + replDownloadRetries :: TVar Int, -- download retries + replDownloadSuccesses :: TVar Int, -- successful downloads + replDownloadAuth :: TVar Int, -- download AUTH errors + replDownloadErr :: TVar Int, -- other download errors (excluding above) + replDelete :: TVar Int, -- total replicas deleted from server + replDeleteRetries :: TVar Int, -- delete retries + replDeleteSuccesses :: TVar Int, -- successful deletes + replDeleteErr :: TVar Int -- delete errors + } + +data AgentXFTPServerStatsData = AgentXFTPServerStatsData + { _fromTime :: UTCTime, + _replUpload :: Int, + _replUploadRetries :: Int, + _replUploadSuccesses :: Int, + _replUploadErr :: Int, + _replDownload :: Int, + _replDownloadRetries :: Int, + _replDownloadSuccesses :: Int, + _replDownloadAuth :: Int, + _replDownloadErr :: Int, + _replDelete :: Int, + _replDeleteRetries :: Int, + _replDeleteSuccesses :: Int, + _replDeleteErr :: Int + } + deriving (Show) + +newAgentXFTPServerStats :: UTCTime -> STM AgentXFTPServerStats +newAgentXFTPServerStats ts = do + fromTime <- newTVar ts + replUpload <- newTVar 0 + replUploadRetries <- newTVar 0 + replUploadSuccesses <- newTVar 0 + replUploadErr <- newTVar 0 + replDownload <- newTVar 0 + replDownloadRetries <- newTVar 0 + replDownloadSuccesses <- newTVar 0 + replDownloadAuth <- newTVar 0 + replDownloadErr <- newTVar 0 + replDelete <- newTVar 0 + replDeleteRetries <- newTVar 0 + replDeleteSuccesses <- newTVar 0 + replDeleteErr <- newTVar 0 + pure + AgentXFTPServerStats + { fromTime, + replUpload, + replUploadRetries, + replUploadSuccesses, + replUploadErr, + replDownload, + replDownloadRetries, + replDownloadSuccesses, + replDownloadAuth, + replDownloadErr, + replDelete, + replDeleteRetries, + replDeleteSuccesses, + replDeleteErr + } + +getAgentXFTPServerStats :: AgentXFTPServerStats -> STM AgentXFTPServerStatsData +getAgentXFTPServerStats s@AgentXFTPServerStats {fromTime} = do + _fromTime <- readTVar fromTime + _replUpload <- readTVar $ replUpload s + _replUploadRetries <- readTVar $ replUpload s + _replUploadSuccesses <- readTVar $ replUpload s + _replUploadErr <- readTVar $ replUpload s + _replDownload <- readTVar $ replUpload s + _replDownloadRetries <- readTVar $ replUpload s + _replDownloadSuccesses <- readTVar $ replUpload s + _replDownloadAuth <- readTVar $ replUpload s + _replDownloadErr <- readTVar $ replUpload s + _replDelete <- readTVar $ replUpload s + _replDeleteRetries <- readTVar $ replUpload s + _replDeleteSuccesses <- readTVar $ replUpload s + _replDeleteErr <- readTVar $ replUpload s + pure + AgentXFTPServerStatsData + { _fromTime, + _replUpload, + _replUploadRetries, + _replUploadSuccesses, + _replUploadErr, + _replDownload, + _replDownloadRetries, + _replDownloadSuccesses, + _replDownloadAuth, + _replDownloadErr, + _replDelete, + _replDeleteRetries, + _replDeleteSuccesses, + _replDeleteErr + } + +setAgentXFTPServerStats :: AgentXFTPServerStats -> AgentXFTPServerStatsData -> STM () +setAgentXFTPServerStats s@AgentXFTPServerStats {fromTime} d@AgentXFTPServerStatsData {_fromTime} = do + writeTVar fromTime $! _fromTime + writeTVar (replUpload s) $! _replUpload d + writeTVar (replUploadRetries s) $! _replUploadRetries d + writeTVar (replUploadSuccesses s) $! _replUploadSuccesses d + writeTVar (replUploadErr s) $! _replUploadErr d + writeTVar (replDownload s) $! _replDownload d + writeTVar (replDownloadRetries s) $! _replDownloadRetries d + writeTVar (replDownloadSuccesses s) $! _replDownloadSuccesses d + writeTVar (replDownloadAuth s) $! _replDownloadAuth d + writeTVar (replDownloadErr s) $! _replDownloadErr d + writeTVar (replDelete s) $! _replDelete d + writeTVar (replDeleteRetries s) $! _replDeleteRetries d + writeTVar (replDeleteSuccesses s) $! _replDeleteSuccesses d + writeTVar (replDeleteErr s) $! _replDeleteErr d + +$(J.deriveJSON defaultJSON ''AgentSMPServerStatsData) + +$(J.deriveJSON defaultJSON ''AgentXFTPServerStatsData)