diff --git a/simplexmq.cabal b/simplexmq.cabal index 1166ca4ff..2474cebd4 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 @@ -132,6 +133,7 @@ library Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240223_connections_wait_delivery Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240225_ratchet_kem Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240417_rcv_files_approved_relays + Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240518_servers_stats Simplex.Messaging.Agent.TRcvQueues Simplex.Messaging.Client Simplex.Messaging.Client.Agent diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 890966888..90dda5cbc 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -63,6 +63,7 @@ import Simplex.Messaging.Agent.Client import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval +import Simplex.Messaging.Agent.Stats import Simplex.Messaging.Agent.Store.SQLite import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB import qualified Simplex.Messaging.Crypto as C @@ -184,6 +185,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do liftIO $ waitForUserNetwork c + atomically $ incXFTPServerStat c userId srv downloadAttempts downloadFileChunk fc replica approvedRelays `catchAgentError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e where @@ -194,7 +196,11 @@ runXFTPRcvWorker c srv Worker {doWork} = do withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay atomically $ assertAgentForeground c loop - retryDone = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) + retryDone e = do + atomically . incXFTPServerStat c userId srv $ case e of + XFTP _ XFTP.AUTH -> downloadAuthErrs + _ -> downloadErrs + rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) e downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> Bool -> AM () downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica approvedRelays = do unlessM ((approvedRelays ||) <$> ipAddressProtected') $ throwE $ FILE NOT_APPROVED @@ -214,6 +220,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do Just RcvFileRedirect {redirectFileInfo = RedirectFileInfo {size = FileSize finalSize}, redirectEntityId} -> (redirectEntityId, finalSize) liftIO . when complete $ updateRcvFileStatus db rcvFileId RFSReceived pure (entityId, complete, RFPROG rcvd total) + atomically $ incXFTPServerStat c userId srv downloads notify c entityId progress when complete . lift . void $ getXFTPRcvWorker True c Nothing @@ -484,6 +491,7 @@ runXFTPSndWorker c srv Worker {doWork} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do liftIO $ waitForUserNetwork c + atomically $ incXFTPServerStat c userId srv uploadAttempts uploadFileChunk cfg fc replica `catchAgentError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e where @@ -494,7 +502,9 @@ runXFTPSndWorker c srv Worker {doWork} = do withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay atomically $ assertAgentForeground c loop - retryDone = sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) + retryDone e = do + atomically $ incXFTPServerStat c userId srv uploadErrs + sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) e uploadFileChunk :: AgentConfig -> SndFileChunk -> SndFileChunkReplica -> AM () uploadFileChunk AgentConfig {xftpMaxRecipientsPerRequest = maxRecipients} sndFileChunk@SndFileChunk {sndFileId, userId, chunkSpec = chunkSpec@XFTPChunkSpec {filePath}, digest = chunkDigest} replica = do replica'@SndFileChunkReplica {sndChunkReplicaId} <- addRecipients sndFileChunk replica @@ -510,6 +520,7 @@ runXFTPSndWorker c srv Worker {doWork} = do let uploaded = uploadedSize chunks total = totalSize chunks complete = all chunkUploaded chunks + atomically $ incXFTPServerStat c userId srv uploads notify c sndFileEntityId $ SFPROG uploaded total when complete $ do (sndDescr, rcvDescrs) <- sndFileToDescrs sf @@ -651,6 +662,7 @@ runXFTPDelWorker c srv Worker {doWork} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do liftIO $ waitForUserNetwork c + atomically $ incXFTPServerStat c userId srv deleteAttempts deleteChunkReplica `catchAgentError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e where @@ -661,10 +673,13 @@ runXFTPDelWorker c srv Worker {doWork} = do withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay atomically $ assertAgentForeground c loop - retryDone = delWorkerInternalError c deletedSndChunkReplicaId + retryDone e = do + atomically $ incXFTPServerStat c userId srv deleteErrs + delWorkerInternalError c deletedSndChunkReplicaId e deleteChunkReplica = do agentXFTPDeleteChunk c userId replica withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId + atomically $ incXFTPServerStat c userId srv deletions delWorkerInternalError :: AgentClient -> Int64 -> AgentErrorType -> AM () delWorkerInternalError c deletedSndChunkReplicaId e = do diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index f1ab78200..0c433cfed 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -104,6 +104,8 @@ module Simplex.Messaging.Agent rcConnectHost, rcConnectCtrl, rcDiscoverCtrl, + getAgentServersSummary, + resetAgentServersStats, foregroundAgent, suspendAgent, execAgentStoreSQL, @@ -157,6 +159,7 @@ import Simplex.Messaging.Agent.Lock (withLock, withLock') import Simplex.Messaging.Agent.NtfSubSupervisor 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 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB @@ -202,23 +205,57 @@ getSMPAgentClient_ clientId cfg initServers store backgroundMode = liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent where runAgent = do - c@AgentClient {acThread} <- atomically . newAgentClient clientId initServers =<< ask + currentTs <- liftIO getCurrentTime + c@AgentClient {acThread} <- atomically . newAgentClient clientId initServers currentTs =<< ask t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c) atomically . writeTVar acThread . Just =<< mkWeakThreadId t pure c runAgentThreads c | backgroundMode = run c "subscriber" $ subscriber c - | otherwise = + | otherwise = do + restoreServersStats c raceAny_ [ run c "subscriber" $ subscriber c, run c "runNtfSupervisor" $ runNtfSupervisor c, - run c "cleanupManager" $ cleanupManager c + run c "cleanupManager" $ cleanupManager c, + run c "logServersStats" $ logServersStats c ] + `E.finally` saveServersStats c run AgentClient {subQ, acThread} name a = a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do logError $ "Agent thread " <> name <> " crashed: " <> tshow e atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e) +logServersStats :: AgentClient -> AM' () +logServersStats c = do + delay <- asks (initialLogStatsDelay . config) + liftIO $ threadDelay' delay + int <- asks (logStatsInterval . config) + forever $ do + saveServersStats c + liftIO $ threadDelay' int + +saveServersStats :: AgentClient -> AM' () +saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats} = do + sss <- mapM (lift . getAgentSMPServerStats) =<< readTVarIO smpServersStats + xss <- mapM (lift . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats + let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss} + tryAgentError' (withStore' c (`updateServersStats` stats)) >>= \case + Left e -> atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e) + Right () -> pure () + +restoreServersStats :: AgentClient -> AM' () +restoreServersStats c@AgentClient {smpServersStats, xftpServersStats, srvStatsStartedAt} = do + tryAgentError' (withStore c getServersStats) >>= \case + Left e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e) + Right (startedAt, Nothing) -> atomically $ writeTVar srvStatsStartedAt startedAt + Right (startedAt, Just AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss}) -> do + atomically $ writeTVar srvStatsStartedAt startedAt + sss' <- mapM (atomically . newAgentSMPServerStats') sss + atomically $ writeTVar smpServersStats sss' + xss' <- mapM (atomically . newAgentXFTPServerStats') xss + atomically $ writeTVar xftpServersStats xss' + disconnectAgentClient :: AgentClient -> IO () disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do closeAgentClient c @@ -550,6 +587,10 @@ rcDiscoverCtrl :: AgentClient -> NonEmpty RCCtrlPairing -> AE (RCCtrlPairing, RC rcDiscoverCtrl AgentClient {agentEnv = Env {multicastSubscribers = subs}} = withExceptT RCP . discoverRCCtrl subs {-# INLINE rcDiscoverCtrl #-} +resetAgentServersStats :: AgentClient -> AE () +resetAgentServersStats c = withAgentEnv c $ resetAgentServersStats' c +{-# INLINE resetAgentServersStats #-} + getAgentStats :: AgentClient -> IO [(AgentStatsKey, Int)] getAgentStats c = readTVarIO (agentStats c) >>= mapM (\(k, cnt) -> (k,) <$> readTVarIO cnt) . M.assocs @@ -581,11 +622,15 @@ createUser' c smp xftp = do pure userId deleteUser' :: AgentClient -> UserId -> Bool -> AM () -deleteUser' c userId delSMPQueues = do +deleteUser' c@AgentClient {smpServersStats, xftpServersStats} userId delSMPQueues = do if delSMPQueues then withStore c (`setUserDeleted` userId) >>= deleteConnectionsAsync_ delUser c False else withStore c (`deleteUserRecord` userId) atomically $ TM.delete userId $ smpServers c + atomically $ TM.delete userId $ xftpServers c + atomically $ modifyTVar' smpServersStats $ M.filterWithKey (\(userId', _) _ -> userId' /= userId) + atomically $ modifyTVar' xftpServersStats $ M.filterWithKey (\(userId', _) _ -> userId' /= userId) + lift $ saveServersStats c where delUser = whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically $ @@ -701,12 +746,13 @@ newConnSrv c userId connId hasNewConn enableNtfs cMode clientData pqInitKeys sub newRcvConnSrv c userId connId' enableNtfs cMode clientData pqInitKeys subMode srv newRcvConnSrv :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe CRClientData -> CR.InitialKeys -> SubscriptionMode -> SMPServerWithAuth -> AM (ConnId, ConnectionRequestUri c) -newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv = do +newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srvWithAuth@(ProtoServerWithAuth srv _) = do case (cMode, pqInitKeys) of (SCMContact, CR.IKUsePQ) -> throwE $ CMD PROHIBITED "newRcvConnSrv" _ -> pure () AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config - (rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srv smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwE e + (rq, qUri, tSess, sessId) <- newRcvQueue c userId connId srvWithAuth smpClientVRange subMode `catchAgentError` \e -> liftIO (print e) >> throwE e + atomically $ incSMPServerStat c userId srv connCreated rq' <- withStore c $ \db -> updateNewConnRcv db connId rq lift . when (subMode == SMSubscribe) $ addNewQueueSubscription c rq' tSess sessId when enableNtfs $ do @@ -1121,6 +1167,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do ICDeleteRcvQueue rId -> withServer $ \srv -> tryWithLock "ICDeleteRcvQueue" $ do rq <- withStore c (\db -> getDeletedRcvQueue db connId srv rId) deleteQueue c rq + atomically $ incSMPServerStat c userId srv connDeleted withStore' c (`deleteConnRcvQueue` rq) ICQSecure rId senderKey -> withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) -> @@ -1129,6 +1176,9 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do case find ((replaceQId ==) . dbQId) rqs of Just rq1 -> when (status == Confirmed) $ do secureQueue c rq' senderKey + -- we may add more statistics special to queue rotation later on, + -- not accounting secure during rotation for now: + -- atomically $ incSMPServerStat c userId server connSecured withStore' c $ \db -> setRcvQueueStatus db rq' Secured void . enqueueMessages c cData sqs SMP.noMsgFlags $ QUSE [((server, sndId), True)] rq1' <- withStore' c $ \db -> setRcvSwitchStatus db rq1 $ Just RSSendingQUSE @@ -1164,8 +1214,9 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do rq <- withStore c $ \db -> getRcvQueue db connId srv rId ackQueueMessage c rq srvMsgId secure :: RcvQueue -> SMP.SndPublicAuthKey -> AM () - secure rq senderKey = do + secure rq@RcvQueue {server} senderKey = do secureQueue c rq senderKey + atomically $ incSMPServerStat c userId server connSecured withStore' c $ \db -> setRcvQueueStatus db rq Secured where withServer a = case server_ of @@ -1281,7 +1332,7 @@ submitPendingMsg c cData sq = do void $ getDeliveryWorker True c cData sq runSmpQueueMsgDelivery :: AgentClient -> ConnData -> SndQueue -> (Worker, TMVar ()) -> AM () -runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork}, qLock) = do +runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq@SndQueue {userId, server} (Worker {doWork}, qLock) = do AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config forever $ do atomically $ endAgentOperation c AOSndNetwork @@ -1304,36 +1355,40 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork Left e -> do let err = if msgType == AM_A_MSG_ then MERR mId e else ERR e case e of - SMP _ SMP.QUOTA -> case msgType of - AM_CONN_INFO -> connError msgId NOT_AVAILABLE - AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE - _ -> do - expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime - if internalTs < expireTs - then notifyDelMsgs msgId e expireTs - else do - notify $ MWARN (unId msgId) e - retrySndMsg RISlow - SMP _ SMP.AUTH -> case msgType of - AM_CONN_INFO -> connError msgId NOT_AVAILABLE - AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE - AM_RATCHET_INFO -> connError msgId NOT_AVAILABLE - -- in duplexHandshake mode (v2) HELLO is only sent once, without retrying, - -- because the queue must be secured by the time the confirmation or the first HELLO is received - AM_HELLO_ -> case rq_ of - -- party initiating connection - Just _ -> connError msgId NOT_AVAILABLE - -- party joining connection - _ -> connError msgId NOT_ACCEPTED - AM_REPLY_ -> notifyDel msgId err - AM_A_MSG_ -> notifyDel msgId err - AM_A_RCVD_ -> notifyDel msgId err - AM_QCONT_ -> notifyDel msgId err - AM_QADD_ -> qError msgId "QADD: AUTH" - AM_QKEY_ -> qError msgId "QKEY: AUTH" - AM_QUSE_ -> qError msgId "QUSE: AUTH" - AM_QTEST_ -> qError msgId "QTEST: AUTH" - AM_EREADY_ -> notifyDel msgId err + SMP _ SMP.QUOTA -> do + atomically $ incSMPServerStat c userId server sentQuotaErrs + case msgType of + AM_CONN_INFO -> connError msgId NOT_AVAILABLE + AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE + _ -> do + expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime + if internalTs < expireTs + then notifyDelMsgs msgId e expireTs + else do + notify $ MWARN (unId msgId) e + retrySndMsg RISlow + SMP _ SMP.AUTH -> do + atomically $ incSMPServerStat c userId server sentAuthErrs + case msgType of + AM_CONN_INFO -> connError msgId NOT_AVAILABLE + AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE + AM_RATCHET_INFO -> connError msgId NOT_AVAILABLE + -- in duplexHandshake mode (v2) HELLO is only sent once, without retrying, + -- because the queue must be secured by the time the confirmation or the first HELLO is received + AM_HELLO_ -> case rq_ of + -- party initiating connection + Just _ -> connError msgId NOT_AVAILABLE + -- party joining connection + _ -> connError msgId NOT_ACCEPTED + AM_REPLY_ -> notifyDel msgId err + AM_A_MSG_ -> notifyDel msgId err + AM_A_RCVD_ -> notifyDel msgId err + AM_QCONT_ -> notifyDel msgId err + AM_QADD_ -> qError msgId "QADD: AUTH" + AM_QKEY_ -> qError msgId "QKEY: AUTH" + AM_QUSE_ -> qError msgId "QUSE: AUTH" + AM_QTEST_ -> qError msgId "QTEST: AUTH" + AM_EREADY_ -> notifyDel msgId err _ -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server), -- the message sending would be retried @@ -1345,7 +1400,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork else do when (serverHostError e) $ notify $ MWARN (unId msgId) e retrySndMsg RIFast - | otherwise -> notifyDel msgId err + | otherwise -> do + atomically $ incSMPServerStat c userId server sentOtherErrs + notifyDel msgId err where retrySndMsg riMode = do withStore' c $ \db -> updatePendingMsgRIState db connId msgId riState @@ -1369,7 +1426,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork -- it would lead to the non-deterministic internal ID of the first sent message, at to some other race conditions, -- because it can be sent before HELLO is received -- With `status == Active` condition, CON is sent here only by the accepting party, that previously received HELLO - when (status == Active) $ notify $ CON pqEncryption + when (status == Active) $ do + atomically $ incSMPServerStat c userId server connCompleted + notify $ CON pqEncryption -- this branch should never be reached as receive queue is created before the confirmation, _ -> logError "HELLO sent without receive queue" AM_A_MSG_ -> notify $ SENT mId proxySrv_ @@ -1422,6 +1481,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork forM_ (L.nonEmpty msgIds_) $ \msgIds -> do notify $ MERRS (L.map unId msgIds) err withStore' c $ \db -> forM_ msgIds $ \msgId' -> deleteSndMsgDelivery db connId sq msgId' False `catchAll_` pure () + atomically $ incSMPServerStat' c userId server sentExpiredErrs (length msgIds_ + 1) delMsg :: InternalId -> AM () delMsg = delMsgKeep False delMsgKeep :: Bool -> InternalId -> AM () @@ -1940,6 +2000,12 @@ setNtfServers :: AgentClient -> [NtfServer] -> IO () setNtfServers c = atomically . writeTVar (ntfServers c) {-# INLINE setNtfServers #-} +resetAgentServersStats' :: AgentClient -> AM () +resetAgentServersStats' c@AgentClient {smpServersStats, xftpServersStats} = do + atomically $ TM.clear smpServersStats + atomically $ TM.clear xftpServersStats + withStore' c resetServersStats + -- | Activate operations foregroundAgent :: AgentClient -> IO () foregroundAgent c = do @@ -2108,13 +2174,15 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts) Left e -> notify' connId (ERR e) Right () -> pure () processSubOk :: RcvQueue -> TVar [ConnId] -> AM () - processSubOk rq@RcvQueue {connId} upConnIds = + processSubOk rq@RcvQueue {userId, connId} upConnIds = do atomically . whenM (isPendingSub connId) $ do addSubscription c rq modifyTVar' upConnIds (connId :) + atomically $ incSMPServerStat c userId srv connSubscribed processSubErr :: RcvQueue -> SMPClientError -> AM () - processSubErr rq@RcvQueue {connId} e = do + processSubErr rq@RcvQueue {userId, connId} e = do atomically . whenM (isPendingSub connId) $ failSubscription c rq e + atomically $ incSMPServerStat c userId srv connSubErrs lift $ notifyErr connId e isPendingSub connId = (&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId notify' :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m () @@ -2128,7 +2196,8 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts) cData@ConnData {userId, connId, connAgentVersion, ratchetSyncState = rss} smpMsg = withConnLock c connId "processSMP" $ case smpMsg of - SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> + SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> do + atomically $ incSMPServerStat c userId srv recvMsgs void . handleNotifyAck $ do msg' <- decryptSMPMessage rq msg ack' <- handleNotifyAck $ case msg' of @@ -2212,6 +2281,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts) Right Nothing -> prohibited "msg: bad agent msg" >> ack Left e@(AGENT A_DUPLICATE) -> do atomically updateDupMsgCount + atomically $ incSMPServerStat c userId srv recvDuplicates withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck} | userAck -> ackDel internalId @@ -2224,6 +2294,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts) _ -> ack _ -> checkDuplicateHash e encryptedMsgHash >> ack Left (AGENT (A_CRYPTO e)) -> do + atomically $ incSMPServerStat c userId srv recvCryptoErrs exists <- withStore' c $ \db -> checkRcvMsgHashExists db connId encryptedMsgHash unless exists notifySync ack @@ -2236,7 +2307,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts) conn'' = updateConnection cData'' connDuplex notify . RSYNC rss' (Just e) $ connectionStats conn'' withStore' c $ \db -> setConnRatchetSync db connId rss' - Left e -> checkDuplicateHash e encryptedMsgHash >> ack + Left e -> do + atomically $ incSMPServerStat c userId srv recvErrs + checkDuplicateHash e encryptedMsgHash >> ack where checkDuplicateHash :: AgentErrorType -> ByteString -> AM () checkDuplicateHash e encryptedMsgHash = @@ -2400,7 +2473,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts) -- `sndStatus == Active` when HELLO was previously sent, and this is the reply HELLO -- this branch is executed by the accepting party in duplexHandshake mode (v2) -- (was executed by initiating party in v1 that is no longer supported) - | sndStatus == Active -> notify $ CON pqEncryption + | sndStatus == Active -> do + atomically $ incSMPServerStat c userId srv connCompleted + notify $ CON pqEncryption | otherwise -> enqueueDuplexHello sq _ -> pure () where diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 2ffd6e3a4..ec8424745 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -1,4 +1,5 @@ {-# LANGUAGE AllowAmbiguousTypes #-} +{-# LANGUAGE BangPatterns #-} {-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveAnyClass #-} @@ -88,6 +89,10 @@ module Simplex.Messaging.Agent.Client activeClientSession, agentClientStore, agentDRG, + AgentServersSummary (..), + ServerSessions (..), + SMPServerSubs (..), + getAgentServersSummary, getAgentSubscriptions, slowNetworkConfig, protocolClientError, @@ -135,6 +140,9 @@ module Simplex.Messaging.Agent.Client getNextServer, withUserServers, withNextSrv, + incSMPServerStat, + incSMPServerStat', + incXFTPServerStat, AgentWorkersDetails (..), getAgentWorkersDetails, AgentWorkersSummary (..), @@ -174,7 +182,7 @@ import Data.List.NonEmpty (NonEmpty (..), (<|)) import qualified Data.List.NonEmpty as L import Data.Map.Strict (Map) import qualified Data.Map.Strict as M -import Data.Maybe (isJust, isNothing, listToMaybe) +import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, listToMaybe) import Data.Set (Set) import qualified Data.Set as S import Data.Text (Text) @@ -196,6 +204,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 @@ -315,7 +324,10 @@ 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, + srvStatsStartedAt :: TVar UTCTime } data SMPConnectedClient = SMPConnectedClient @@ -443,8 +455,8 @@ data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther deriving (Eq, Show) -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's. -newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient -newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do +newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Env -> STM AgentClient +newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs agentEnv = do let cfg = config agentEnv qSize = tbqSize cfg acThread <- newTVar Nothing @@ -482,6 +494,9 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = smpSubWorkers <- TM.empty agentStats <- TM.empty msgCounts <- TM.empty + smpServersStats <- TM.empty + xftpServersStats <- TM.empty + srvStatsStartedAt <- newTVar currentTs return AgentClient { acThread, @@ -520,7 +535,10 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = agentStats, msgCounts, clientId, - agentEnv + agentEnv, + smpServersStats, + xftpServersStats, + srvStatsStartedAt } slowNetworkConfig :: NetworkConfig -> NetworkConfig @@ -1060,7 +1078,11 @@ sendOrProxySMPMessage c userId destSrv cmdStr spKey_ senderId msgFlags msg = do unknownServer = maybe True (all ((destSrv /=) . protoServer)) <$> TM.lookup userId (userServers c) sendViaProxy destSess@(_, _, qId) = do r <- tryAgentError . withProxySession c destSess senderId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess) -> do - liftClient SMP (clientServer smp) (proxySMPMessage smp proxySess spKey_ senderId msgFlags msg) >>= \case + r' <- liftClient SMP (clientServer smp) $ do + atomically $ incSMPServerStat c userId destSrv sentViaProxyAttempts + atomically $ incSMPServerStat c userId (protocolClientServer' smp) sentProxiedAttempts + proxySMPMessage smp proxySess spKey_ senderId msgFlags msg + case r' of Right () -> pure . Just $ protocolClientServer' smp Left proxyErr -> do case proxyErr of @@ -1088,13 +1110,23 @@ sendOrProxySMPMessage c userId destSrv cmdStr spKey_ senderId msgFlags msg = do sameClient smp' = sessionId (thParams smp) == sessionId (thParams smp') sameProxiedRelay proxySess' = prSessionId proxySess == prSessionId proxySess' case r of - Right r' -> pure r' + Right r' -> do + atomically $ incSMPServerStat c userId destSrv sentViaProxy + forM_ r' $ \proxySrv -> atomically $ incSMPServerStat c userId proxySrv sentProxied + pure r' Left e | serverHostError e -> ifM (atomically directAllowed) (sendDirectly destSess $> Nothing) (throwE e) | otherwise -> throwE e sendDirectly tSess = - withLogClient_ c tSess senderId ("SEND " <> cmdStr) $ \(SMPConnectedClient smp _) -> - liftClient SMP (clientServer smp) $ sendSMPMessage smp spKey_ senderId msgFlags msg + withLogClient_ c tSess senderId ("SEND " <> cmdStr) $ \(SMPConnectedClient smp _) -> do + r <- + tryAgentError $ + liftClient SMP (clientServer smp) $ do + atomically $ incSMPServerStat c userId destSrv sentDirectAttempts + sendSMPMessage smp spKey_ senderId msgFlags msg + case r of + Right () -> atomically $ incSMPServerStat c userId destSrv sentDirect + Left e -> throwE e ipAddressProtected :: NetworkConfig -> ProtocolServer p -> Bool ipAddressProtected NetworkConfig {socksProxy, hostMode} (ProtocolServer _ hosts _ _) = do @@ -1369,6 +1401,8 @@ subscribeQueues c qs = do pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED "subscribeQueues") else Right rq subscribeQueues_ :: Env -> TVar (Maybe SessionId) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ()) subscribeQueues_ env session smp qs' = do + let (userId, srv, _) = transportSession' smp + atomically $ incSMPServerStat' c userId srv connSubAttempts (length qs') rs <- sendBatch subscribeSMPQueues smp qs' active <- atomically $ @@ -1916,6 +1950,103 @@ withNextSrv c userId usedSrvs initUsed action = do writeTVar usedSrvs $! used' action srvAuth +incSMPServerStat :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats -> TVar Int) -> STM () +incSMPServerStat c userId srv sel = incSMPServerStat' c userId srv sel 1 + +incSMPServerStat' :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats -> TVar Int) -> Int -> STM () +incSMPServerStat' AgentClient {smpServersStats} userId srv sel n = do + TM.lookup (userId, srv) smpServersStats >>= \case + Just v -> modifyTVar' (sel v) (+ n) + Nothing -> do + newStats <- newAgentSMPServerStats + modifyTVar' (sel newStats) (+ n) + TM.insert (userId, srv) newStats smpServersStats + +incXFTPServerStat :: AgentClient -> UserId -> XFTPServer -> (AgentXFTPServerStats -> TVar Int) -> STM () +incXFTPServerStat AgentClient {xftpServersStats} userId srv sel = do + TM.lookup (userId, srv) xftpServersStats >>= \case + Just v -> modifyTVar' (sel v) (+ 1) + Nothing -> do + newStats <- newAgentXFTPServerStats + modifyTVar' (sel newStats) (+ 1) + TM.insert (userId, srv) newStats xftpServersStats + +data AgentServersSummary = AgentServersSummary + { smpServersStats :: Map (UserId, SMPServer) AgentSMPServerStatsData, + xftpServersStats :: Map (UserId, XFTPServer) AgentXFTPServerStatsData, + statsStartedAt :: UTCTime, + smpServersSessions :: Map (UserId, SMPServer) ServerSessions, + smpServersSubs :: Map (UserId, SMPServer) SMPServerSubs, + xftpServersSessions :: Map (UserId, XFTPServer) ServerSessions, + xftpRcvInProgress :: [XFTPServer], + xftpSndInProgress :: [XFTPServer], + xftpDelInProgress :: [XFTPServer] + } + deriving (Show) + +data SMPServerSubs = SMPServerSubs + { ssActive :: Int, -- based on activeSubs + ssPending :: Int -- based on pendingSubs + } + deriving (Show) + +data ServerSessions = ServerSessions + { ssConnected :: Int, + ssErrors :: Int, + ssConnecting :: Int + } + deriving (Show) + +getAgentServersSummary :: AgentClient -> IO AgentServersSummary +getAgentServersSummary c@AgentClient {smpServersStats, xftpServersStats, srvStatsStartedAt, agentEnv} = do + sss <- mapM getAgentSMPServerStats =<< readTVarIO smpServersStats + xss <- mapM getAgentXFTPServerStats =<< readTVarIO xftpServersStats + statsStartedAt <- readTVarIO srvStatsStartedAt + smpServersSessions <- countSessions =<< readTVarIO (smpClients c) + smpServersSubs <- getServerSubs + xftpServersSessions <- countSessions =<< readTVarIO (xftpClients c) + xftpRcvInProgress <- catMaybes <$> getXFTPWorkerSrvs xftpRcvWorkers + xftpSndInProgress <- catMaybes <$> getXFTPWorkerSrvs xftpSndWorkers + xftpDelInProgress <- getXFTPWorkerSrvs xftpDelWorkers + pure + AgentServersSummary + { smpServersStats = sss, + xftpServersStats = xss, + statsStartedAt, + smpServersSessions, + smpServersSubs, + xftpServersSessions, + xftpRcvInProgress, + xftpSndInProgress, + xftpDelInProgress + } + where + getServerSubs = do + subs <- M.foldrWithKey' (addSub incActive) M.empty <$> readTVarIO (getRcvQueues $ activeSubs c) + M.foldrWithKey' (addSub incPending) subs <$> readTVarIO (getRcvQueues $ pendingSubs c) + where + addSub f (userId, srv, _) _ = M.alter (Just . f . fromMaybe SMPServerSubs {ssActive = 0, ssPending = 0}) (userId, srv) + incActive ss = ss {ssActive = ssActive ss + 1} + incPending ss = ss {ssPending = ssPending ss + 1} + Env {xftpAgent = XFTPAgent {xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers}} = agentEnv + getXFTPWorkerSrvs workers = foldM addSrv [] . M.toList =<< readTVarIO workers + where + addSrv acc (srv, Worker {doWork}) = do + hasWork <- atomically $ not <$> isEmptyTMVar doWork + pure $ if hasWork then srv : acc else acc + countSessions :: Map (TransportSession msg) (ClientVar msg) -> IO (Map (UserId, ProtoServer msg) ServerSessions) + countSessions = foldM addClient M.empty . M.toList + where + addClient !acc ((userId, srv, _), SessionVar {sessionVar}) = do + c_ <- atomically $ tryReadTMVar sessionVar + pure $ M.alter (Just . add c_) (userId, srv) acc + where + add c_ = modifySessions c_ . fromMaybe ServerSessions {ssConnected = 0, ssErrors = 0, ssConnecting = 0} + modifySessions c_ ss = case c_ of + Just (Right _) -> ss {ssConnected = ssConnected ss + 1} + Just (Left _) -> ss {ssErrors = ssErrors ss + 1} + Nothing -> ss {ssConnecting = ssConnecting ss + 1} + data SubInfo = SubInfo {userId :: UserId, server :: Text, rcvId :: Text, subError :: Maybe String} deriving (Show) @@ -2106,6 +2237,12 @@ $(J.deriveJSON (enumJSON $ dropPrefix "TS") ''ProtocolTestStep) $(J.deriveJSON defaultJSON ''ProtocolTestFailure) +$(J.deriveJSON defaultJSON ''ServerSessions) + +$(J.deriveJSON defaultJSON ''SMPServerSubs) + +$(J.deriveJSON defaultJSON ''AgentServersSummary) + $(J.deriveJSON defaultJSON ''SubInfo) $(J.deriveJSON defaultJSON ''SubscriptionsInfo) diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index ee2bb16cc..2ae2ad5c0 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -100,6 +100,8 @@ data AgentConfig = AgentConfig persistErrorInterval :: NominalDiffTime, initialCleanupDelay :: Int64, cleanupInterval :: Int64, + initialLogStatsDelay :: Int64, + logStatsInterval :: Int64, cleanupStepInterval :: Int, maxWorkerRestartsPerMin :: Int, storedMsgDataTTL :: NominalDiffTime, @@ -170,6 +172,8 @@ defaultAgentConfig = persistErrorInterval = 3, -- seconds initialCleanupDelay = 30 * 1000000, -- 30 seconds cleanupInterval = 30 * 60 * 1000000, -- 30 minutes + initialLogStatsDelay = 10 * 1000000, -- 10 seconds + logStatsInterval = 10 * 1000000, -- 10 seconds cleanupStepInterval = 200000, -- 200ms maxWorkerRestartsPerMin = 5, storedMsgDataTTL = 21 * nominalDay, diff --git a/src/Simplex/Messaging/Agent/Stats.hs b/src/Simplex/Messaging/Agent/Stats.hs new file mode 100644 index 000000000..c8f81a6aa --- /dev/null +++ b/src/Simplex/Messaging/Agent/Stats.hs @@ -0,0 +1,337 @@ +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE TemplateHaskell #-} + +module Simplex.Messaging.Agent.Stats where + +import qualified Data.Aeson.TH as J +import Data.Map (Map) +import Database.SQLite.Simple.FromField (FromField (..)) +import Database.SQLite.Simple.ToField (ToField (..)) +import Simplex.Messaging.Agent.Protocol (UserId) +import Simplex.Messaging.Parsers (defaultJSON, fromTextField_) +import Simplex.Messaging.Protocol (SMPServer, XFTPServer) +import Simplex.Messaging.Util (decodeJSON, encodeJSON) +import UnliftIO.STM + +data AgentSMPServerStats = AgentSMPServerStats + { sentDirect :: TVar Int, -- successfully sent messages + sentViaProxy :: TVar Int, -- successfully sent messages via proxy + sentProxied :: TVar Int, -- successfully sent messages to other destination server via this as proxy + sentDirectAttempts :: TVar Int, -- direct sending attempts (min 1 for each sent message) + sentViaProxyAttempts :: TVar Int, -- proxy sending attempts + sentProxiedAttempts :: TVar Int, -- attempts sending to other destination server via this as proxy + sentAuthErrs :: TVar Int, -- send AUTH errors + sentQuotaErrs :: TVar Int, -- send QUOTA permanent errors (message expired) + sentExpiredErrs :: TVar Int, -- send expired errors + sentOtherErrs :: TVar Int, -- other send permanent errors (excluding above) + recvMsgs :: TVar Int, -- total messages received + recvDuplicates :: TVar Int, -- duplicate messages received + recvCryptoErrs :: TVar Int, -- message decryption errors + recvErrs :: TVar Int, -- receive errors + connCreated :: TVar Int, + connSecured :: TVar Int, + connCompleted :: TVar Int, + connDeleted :: TVar Int, + connSubscribed :: TVar Int, -- total successful subscription + connSubAttempts :: TVar Int, -- subscription attempts + connSubErrs :: TVar Int -- permanent subscription errors (temporary accounted for in attempts) + } + +data AgentSMPServerStatsData = AgentSMPServerStatsData + { _sentDirect :: Int, + _sentViaProxy :: Int, + _sentProxied :: Int, + _sentDirectAttempts :: Int, + _sentViaProxyAttempts :: Int, + _sentProxiedAttempts :: Int, + _sentAuthErrs :: Int, + _sentQuotaErrs :: Int, + _sentExpiredErrs :: Int, + _sentOtherErrs :: Int, + _recvMsgs :: Int, + _recvDuplicates :: Int, + _recvCryptoErrs :: Int, + _recvErrs :: Int, + _connCreated :: Int, + _connSecured :: Int, + _connCompleted :: Int, + _connDeleted :: Int, + _connSubscribed :: Int, + _connSubAttempts :: Int, + _connSubErrs :: Int + } + deriving (Show) + +newAgentSMPServerStats :: STM AgentSMPServerStats +newAgentSMPServerStats = do + sentDirect <- newTVar 0 + sentViaProxy <- newTVar 0 + sentProxied <- newTVar 0 + sentDirectAttempts <- newTVar 0 + sentViaProxyAttempts <- newTVar 0 + sentProxiedAttempts <- newTVar 0 + sentAuthErrs <- newTVar 0 + sentQuotaErrs <- newTVar 0 + sentExpiredErrs <- newTVar 0 + sentOtherErrs <- newTVar 0 + recvMsgs <- newTVar 0 + recvDuplicates <- newTVar 0 + recvCryptoErrs <- newTVar 0 + recvErrs <- newTVar 0 + connCreated <- newTVar 0 + connSecured <- newTVar 0 + connCompleted <- newTVar 0 + connDeleted <- newTVar 0 + connSubscribed <- newTVar 0 + connSubAttempts <- newTVar 0 + connSubErrs <- newTVar 0 + pure + AgentSMPServerStats + { sentDirect, + sentViaProxy, + sentProxied, + sentDirectAttempts, + sentViaProxyAttempts, + sentProxiedAttempts, + sentAuthErrs, + sentQuotaErrs, + sentExpiredErrs, + sentOtherErrs, + recvMsgs, + recvDuplicates, + recvCryptoErrs, + recvErrs, + connCreated, + connSecured, + connCompleted, + connDeleted, + connSubscribed, + connSubAttempts, + connSubErrs + } + +newAgentSMPServerStats' :: AgentSMPServerStatsData -> STM AgentSMPServerStats +newAgentSMPServerStats' s = do + sentDirect <- newTVar $ _sentDirect s + sentViaProxy <- newTVar $ _sentViaProxy s + sentProxied <- newTVar $ _sentProxied s + sentDirectAttempts <- newTVar $ _sentDirectAttempts s + sentViaProxyAttempts <- newTVar $ _sentViaProxyAttempts s + sentProxiedAttempts <- newTVar $ _sentProxiedAttempts s + sentAuthErrs <- newTVar $ _sentAuthErrs s + sentQuotaErrs <- newTVar $ _sentQuotaErrs s + sentExpiredErrs <- newTVar $ _sentExpiredErrs s + sentOtherErrs <- newTVar $ _sentOtherErrs s + recvMsgs <- newTVar $ _recvMsgs s + recvDuplicates <- newTVar $ _recvDuplicates s + recvCryptoErrs <- newTVar $ _recvCryptoErrs s + recvErrs <- newTVar $ _recvErrs s + connCreated <- newTVar $ _connCreated s + connSecured <- newTVar $ _connSecured s + connCompleted <- newTVar $ _connCompleted s + connDeleted <- newTVar $ _connDeleted s + connSubscribed <- newTVar $ _connSubscribed s + connSubAttempts <- newTVar $ _connSubAttempts s + connSubErrs <- newTVar $ _connSubErrs s + pure + AgentSMPServerStats + { sentDirect, + sentViaProxy, + sentProxied, + sentDirectAttempts, + sentViaProxyAttempts, + sentProxiedAttempts, + sentAuthErrs, + sentQuotaErrs, + sentExpiredErrs, + sentOtherErrs, + recvMsgs, + recvDuplicates, + recvCryptoErrs, + recvErrs, + connCreated, + connSecured, + connCompleted, + connDeleted, + connSubscribed, + connSubAttempts, + connSubErrs + } + +-- as this is used to periodically update stats in db, +-- this is not STM to decrease contention with stats updates +getAgentSMPServerStats :: AgentSMPServerStats -> IO AgentSMPServerStatsData +getAgentSMPServerStats s = do + _sentDirect <- readTVarIO $ sentDirect s + _sentViaProxy <- readTVarIO $ sentViaProxy s + _sentProxied <- readTVarIO $ sentProxied s + _sentDirectAttempts <- readTVarIO $ sentDirectAttempts s + _sentViaProxyAttempts <- readTVarIO $ sentViaProxyAttempts s + _sentProxiedAttempts <- readTVarIO $ sentProxiedAttempts s + _sentAuthErrs <- readTVarIO $ sentAuthErrs s + _sentQuotaErrs <- readTVarIO $ sentQuotaErrs s + _sentExpiredErrs <- readTVarIO $ sentExpiredErrs s + _sentOtherErrs <- readTVarIO $ sentOtherErrs s + _recvMsgs <- readTVarIO $ recvMsgs s + _recvDuplicates <- readTVarIO $ recvDuplicates s + _recvCryptoErrs <- readTVarIO $ recvCryptoErrs s + _recvErrs <- readTVarIO $ recvErrs s + _connCreated <- readTVarIO $ connCreated s + _connSecured <- readTVarIO $ connSecured s + _connCompleted <- readTVarIO $ connCompleted s + _connDeleted <- readTVarIO $ connDeleted s + _connSubscribed <- readTVarIO $ connSubscribed s + _connSubAttempts <- readTVarIO $ connSubAttempts s + _connSubErrs <- readTVarIO $ connSubErrs s + pure + AgentSMPServerStatsData + { _sentDirect, + _sentViaProxy, + _sentProxied, + _sentDirectAttempts, + _sentViaProxyAttempts, + _sentProxiedAttempts, + _sentAuthErrs, + _sentQuotaErrs, + _sentExpiredErrs, + _sentOtherErrs, + _recvMsgs, + _recvDuplicates, + _recvCryptoErrs, + _recvErrs, + _connCreated, + _connSecured, + _connCompleted, + _connDeleted, + _connSubscribed, + _connSubAttempts, + _connSubErrs + } + +data AgentXFTPServerStats = AgentXFTPServerStats + { uploads :: TVar Int, -- total replicas uploaded to server + uploadAttempts :: TVar Int, -- upload attempts + uploadErrs :: TVar Int, -- upload errors + downloads :: TVar Int, -- total replicas downloaded from server + downloadAttempts :: TVar Int, -- download attempts + downloadAuthErrs :: TVar Int, -- download AUTH errors + downloadErrs :: TVar Int, -- other download errors (excluding above) + deletions :: TVar Int, -- total replicas deleted from server + deleteAttempts :: TVar Int, -- delete attempts + deleteErrs :: TVar Int -- delete errors + } + +data AgentXFTPServerStatsData = AgentXFTPServerStatsData + { _uploads :: Int, + _uploadAttempts :: Int, + _uploadErrs :: Int, + _downloads :: Int, + _downloadAttempts :: Int, + _downloadAuthErrs :: Int, + _downloadErrs :: Int, + _deletions :: Int, + _deleteAttempts :: Int, + _deleteErrs :: Int + } + deriving (Show) + +newAgentXFTPServerStats :: STM AgentXFTPServerStats +newAgentXFTPServerStats = do + uploads <- newTVar 0 + uploadAttempts <- newTVar 0 + uploadErrs <- newTVar 0 + downloads <- newTVar 0 + downloadAttempts <- newTVar 0 + downloadAuthErrs <- newTVar 0 + downloadErrs <- newTVar 0 + deletions <- newTVar 0 + deleteAttempts <- newTVar 0 + deleteErrs <- newTVar 0 + pure + AgentXFTPServerStats + { uploads, + uploadAttempts, + uploadErrs, + downloads, + downloadAttempts, + downloadAuthErrs, + downloadErrs, + deletions, + deleteAttempts, + deleteErrs + } + +newAgentXFTPServerStats' :: AgentXFTPServerStatsData -> STM AgentXFTPServerStats +newAgentXFTPServerStats' s = do + uploads <- newTVar $ _uploads s + uploadAttempts <- newTVar $ _uploadAttempts s + uploadErrs <- newTVar $ _uploadErrs s + downloads <- newTVar $ _downloads s + downloadAttempts <- newTVar $ _downloadAttempts s + downloadAuthErrs <- newTVar $ _downloadAuthErrs s + downloadErrs <- newTVar $ _downloadErrs s + deletions <- newTVar $ _deletions s + deleteAttempts <- newTVar $ _deleteAttempts s + deleteErrs <- newTVar $ _deleteErrs s + pure + AgentXFTPServerStats + { uploads, + uploadAttempts, + uploadErrs, + downloads, + downloadAttempts, + downloadAuthErrs, + downloadErrs, + deletions, + deleteAttempts, + deleteErrs + } + +-- as this is used to periodically update stats in db, +-- this is not STM to decrease contention with stats updates +getAgentXFTPServerStats :: AgentXFTPServerStats -> IO AgentXFTPServerStatsData +getAgentXFTPServerStats s = do + _uploads <- readTVarIO $ uploads s + _uploadAttempts <- readTVarIO $ uploadAttempts s + _uploadErrs <- readTVarIO $ uploadErrs s + _downloads <- readTVarIO $ downloads s + _downloadAttempts <- readTVarIO $ downloadAttempts s + _downloadAuthErrs <- readTVarIO $ downloadAuthErrs s + _downloadErrs <- readTVarIO $ downloadErrs s + _deletions <- readTVarIO $ deletions s + _deleteAttempts <- readTVarIO $ deleteAttempts s + _deleteErrs <- readTVarIO $ deleteErrs s + pure + AgentXFTPServerStatsData + { _uploads, + _uploadAttempts, + _uploadErrs, + _downloads, + _downloadAttempts, + _downloadAuthErrs, + _downloadErrs, + _deletions, + _deleteAttempts, + _deleteErrs + } + +-- Type for gathering both smp and xftp stats across all users and servers, +-- to then be persisted to db as a single json. +data AgentPersistedServerStats = AgentPersistedServerStats + { smpServersStats :: Map (UserId, SMPServer) AgentSMPServerStatsData, + xftpServersStats :: Map (UserId, XFTPServer) AgentXFTPServerStatsData + } + deriving (Show) + +$(J.deriveJSON defaultJSON ''AgentSMPServerStatsData) + +$(J.deriveJSON defaultJSON ''AgentXFTPServerStatsData) + +$(J.deriveJSON defaultJSON ''AgentPersistedServerStats) + +instance ToField AgentPersistedServerStats where + toField = toField . encodeJSON + +instance FromField AgentPersistedServerStats where + fromField = fromTextField_ decodeJSON diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 807ca223a..b9ffecbbd 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -635,4 +635,6 @@ data StoreError SEDeletedSndChunkReplicaNotFound | -- | Error when reading work item that suspends worker - do not use! SEWorkItemError ByteString + | -- | Servers stats not found. + SEServersStatsNotFound deriving (Eq, Show, Exception) diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index a8b18c5b7..434344c89 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -210,6 +210,10 @@ module Simplex.Messaging.Agent.Store.SQLite deleteDeletedSndChunkReplica, getPendingDelFilesServers, deleteDeletedSndChunkReplicasExpired, + -- Stats + updateServersStats, + getServersStats, + resetServersStats, -- * utilities withConnection, @@ -263,6 +267,7 @@ import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..)) import Simplex.FileTransfer.Types import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval (RI2State (..)) +import Simplex.Messaging.Agent.Stats import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.SQLite.Common import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB @@ -3017,6 +3022,21 @@ deleteDeletedSndChunkReplicasExpired db ttl = do cutoffTs <- addUTCTime (-ttl) <$> getCurrentTime DB.execute db "DELETE FROM deleted_snd_chunk_replicas WHERE created_at < ?" (Only cutoffTs) +updateServersStats :: DB.Connection -> AgentPersistedServerStats -> IO () +updateServersStats db stats = do + updatedAt <- getCurrentTime + DB.execute db "UPDATE servers_stats SET servers_stats = ?, updated_at = ? WHERE servers_stats_id = 1" (stats, updatedAt) + +getServersStats :: DB.Connection -> IO (Either StoreError (UTCTime, Maybe AgentPersistedServerStats)) +getServersStats db = + firstRow id SEServersStatsNotFound $ + DB.query_ db "SELECT started_at, servers_stats FROM servers_stats WHERE servers_stats_id = 1" + +resetServersStats :: DB.Connection -> IO () +resetServersStats db = do + currentTs <- getCurrentTime + DB.execute db "UPDATE servers_stats SET servers_stats = NULL, started_at = ?, updated_at = ? WHERE servers_stats_id = 1" (currentTs, currentTs) + $(J.deriveJSON defaultJSON ''UpMigration) $(J.deriveToJSON (sumTypeJSON $ dropPrefix "ME") ''MigrationError) diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs index 5a5ed5b5b..340063f5c 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs @@ -72,6 +72,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240124_file_redirect import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240223_connections_wait_delivery import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240225_ratchet_kem import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240417_rcv_files_approved_relays +import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240518_servers_stats import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON) import Simplex.Messaging.Transport.Client (TransportHost) @@ -112,7 +113,8 @@ schemaMigrations = ("m20240124_file_redirect", m20240124_file_redirect, Just down_m20240124_file_redirect), ("m20240223_connections_wait_delivery", m20240223_connections_wait_delivery, Just down_m20240223_connections_wait_delivery), ("m20240225_ratchet_kem", m20240225_ratchet_kem, Just down_m20240225_ratchet_kem), - ("m20240417_rcv_files_approved_relays", m20240417_rcv_files_approved_relays, Just down_m20240417_rcv_files_approved_relays) + ("m20240417_rcv_files_approved_relays", m20240417_rcv_files_approved_relays, Just down_m20240417_rcv_files_approved_relays), + ("m20240518_servers_stats", m20240518_servers_stats, Just down_m20240518_servers_stats) ] -- | The list of migrations in ascending order by date diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20240518_servers_stats.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20240518_servers_stats.hs new file mode 100644 index 000000000..fe017e233 --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20240518_servers_stats.hs @@ -0,0 +1,29 @@ +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240518_servers_stats where + +import Database.SQLite.Simple (Query) +import Database.SQLite.Simple.QQ (sql) + +-- servers_stats_id: dummy id, there should always only be one record with servers_stats_id = 1 +-- servers_stats: overall accumulated stats, past and session, reset to null on stats reset +-- started_at: starting point of tracking stats, reset on stats reset +m20240518_servers_stats :: Query +m20240518_servers_stats = + [sql| +CREATE TABLE servers_stats( + servers_stats_id INTEGER PRIMARY KEY, + servers_stats TEXT, + started_at TEXT NOT NULL DEFAULT(datetime('now')), + created_at TEXT NOT NULL DEFAULT(datetime('now')), + updated_at TEXT NOT NULL DEFAULT(datetime('now')) +); + +INSERT INTO servers_stats (servers_stats_id) VALUES (1); +|] + +down_m20240518_servers_stats :: Query +down_m20240518_servers_stats = + [sql| +DROP TABLE servers_stats; +|] diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index caf94418a..50cf6d74a 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -394,6 +394,13 @@ CREATE TABLE processed_ratchet_key_hashes( created_at TEXT NOT NULL DEFAULT(datetime('now')), updated_at TEXT NOT NULL DEFAULT(datetime('now')) ); +CREATE TABLE servers_stats( + servers_stats_id INTEGER PRIMARY KEY, + servers_stats TEXT, + started_at TEXT NOT NULL DEFAULT(datetime('now')), + created_at TEXT NOT NULL DEFAULT(datetime('now')), + updated_at TEXT NOT NULL DEFAULT(datetime('now')) +); CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id); CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(conn_id, rcv_queue_id); CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(conn_id, snd_queue_id); diff --git a/src/Simplex/Messaging/Util.hs b/src/Simplex/Messaging/Util.hs index 37557cd23..b023f460a 100644 --- a/src/Simplex/Messaging/Util.hs +++ b/src/Simplex/Messaging/Util.hs @@ -8,16 +8,19 @@ import Control.Monad import Control.Monad.Except import Control.Monad.IO.Unlift import Control.Monad.Trans.Except +import Data.Aeson (FromJSON, ToJSON) +import qualified Data.Aeson as J import Data.Bifunctor (first) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B +import qualified Data.ByteString.Lazy.Char8 as LB import Data.Int (Int64) import Data.List (groupBy, sortOn) import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as L import Data.Text (Text) import qualified Data.Text as T -import Data.Text.Encoding (decodeUtf8With) +import Data.Text.Encoding (decodeUtf8With, encodeUtf8) import Data.Time (NominalDiffTime) import GHC.Conc (labelThread, myThreadId, threadDelay) import UnliftIO @@ -170,3 +173,9 @@ diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer) labelMyThread :: MonadIO m => String -> m () labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label) + +encodeJSON :: ToJSON a => a -> Text +encodeJSON = safeDecodeUtf8 . LB.toStrict . J.encode + +decodeJSON :: FromJSON a => Text -> Maybe a +decodeJSON = J.decode . LB.fromStrict . encodeUtf8