WIP: command rate monitoring

package.yaml

@@ -45,6 +45,7 @@ dependencies:
   - direct-sqlcipher == 2.3.*
   - directory == 1.3.*
   - filepath == 1.4.*
+  - hashable == 1.4.*
   - hourglass == 0.2.*
   - http-types == 0.12.*
   - http2 >= 4.2.2 && < 4.3
@@ -59,6 +60,7 @@ dependencies:
   - network-udp >= 0.0 && < 0.1
   - optparse-applicative >= 0.15 && < 0.17
   - process == 1.6.*
+  - psqueues == 0.2.8.*
   - random >= 1.1 && < 1.3
   - simple-logger == 0.1.*
   - socks == 0.6.*

simplexmq.cabal

@@ -204,6 +204,7 @@ library
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -218,6 +219,7 @@ library
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , socks ==0.6.*
@@ -278,6 +280,7 @@ executable ntf-server
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -292,6 +295,7 @@ executable ntf-server
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq
@@ -353,6 +357,7 @@ executable smp-agent
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -367,6 +372,7 @@ executable smp-agent
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq
@@ -428,6 +434,7 @@ executable smp-server
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -442,6 +449,7 @@ executable smp-server
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq
@@ -503,6 +511,7 @@ executable xftp
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -517,6 +526,7 @@ executable xftp
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq
@@ -578,6 +588,7 @@ executable xftp-server
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -592,6 +603,7 @@ executable xftp-server
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq
@@ -689,6 +701,7 @@ test-suite simplexmq-test
     , directory ==1.3.*
     , filepath ==1.4.*
     , generic-random ==1.5.*
+    , hashable ==1.4.*
     , hourglass ==0.2.*
     , hspec ==2.11.*
     , hspec-core ==2.11.*
@@ -706,6 +719,7 @@ test-suite simplexmq-test
     , network-udp ==0.0.*
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
+    , psqueues ==0.2.8.*
     , random >=1.1 && <1.3
     , silently ==1.2.*
     , simple-logger ==0.1.*

src/Simplex/Messaging/Server.hs

@@ -128,7 +128,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
   raceAny_
     ( serverThread s "server subscribedQ" subscribedQ subscribers subscriptions cancelSub
         : serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubscriptions (\_ -> pure ())
-        : map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
+        : map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> rateStatsThread_ cfg <> controlPortThread_ cfg
     )
     `finally` withLock' (savingLock s) "final" (saveServer False)
   where
@@ -205,6 +205,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
       [logServerStats logStatsStartTime interval serverStatsLogFile]
     serverStatsThread_ _ = []
 
+    rateStatsThread_ :: ServerConfig -> [M ()]
+    rateStatsThread_ ServerConfig {rateStatsInterval = Just bucketWidth, logStatsInterval = Just logInterval, logStatsStartTime, rateStatsLogFile} =
+      [ monitorServerRates bucketWidth, -- roll windows, collect counters, runs at a faster rate so the measurements can be used for online anomaly detection
+        logServerRates logStatsStartTime logInterval rateStatsLogFile -- log distributions once in a while
+      ]
+    rateStatsThread_ _ = []
+
     logServerStats :: Int64 -> Int64 -> FilePath -> M ()
     logServerStats startAt logInterval statsFilePath = do
       labelMyThread "logServerStats"
@@ -257,6 +264,25 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
               ]
         liftIO $ threadDelay' interval
 
+    monitorServerRates :: Int64 -> M ()
+    monitorServerRates bucketWidth = do
+      labelMyThread "monitorServerRates"
+      forever $ do
+        -- TODO: calculate delay for the next bucket closing time
+        liftIO $ threadDelay' bucketWidth
+        -- TODO: collect and reset buckets
+
+    logServerRates :: Int64 -> Int64 -> FilePath -> M ()
+    logServerRates startAt logInterval statsFilePath = do
+      labelMyThread "logServerStats"
+      initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
+      liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
+      liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
+      let interval = 1000000 * logInterval
+      forever $ do
+        -- write the thing
+        liftIO $ threadDelay' interval
+
     runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
     runClient signKey tp h = do
       kh <- asks serverIdentity
@@ -411,13 +437,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                     hPutStrLn h "AUTH"
 
 runClientTransport :: Transport c => THandleSMP c 'TServer -> M ()
-runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} = do
+runClientTransport th@THandle {connection, params = THandleParams {thVersion, sessionId}} = do
   q <- asks $ tbqSize . config
   ts <- liftIO getSystemTime
   active <- asks clients
   nextClientId <- asks clientSeq
   c <- atomically $ do
-    new@Client {clientId} <- newClient nextClientId q thVersion sessionId ts
+    new@Client {clientId} <- newClient (getPeerId connection) nextClientId q thVersion sessionId ts
     modifyTVar' active $ IM.insert clientId new
     pure new
   s <- asks server
@@ -634,6 +660,10 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
       where
         createQueue :: QueueStore -> RcvPublicAuthKey -> RcvPublicDhKey -> SubscriptionMode -> M (Transmission BrokerMsg)
         createQueue st recipientKey dhKey subMode = time "NEW" $ do
+          -- TODO: read client Q rate
+          -- TODO: read server Q rate for peerId
+          -- TODO: read global server Q rate
+          -- TODO: add throttling delay/blackhole request if needed
           (rcvPublicDhKey, privDhKey) <- atomically . C.generateKeyPair =<< asks random
           let rcvDhSecret = C.dh' dhKey privDhKey
               qik (rcvId, sndId) = QIK {rcvId, sndId, rcvPublicDhKey}
@@ -664,6 +694,9 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                   stats <- asks serverStats
                   atomically $ modifyTVar' (qCreated stats) (+ 1)
                   atomically $ modifyTVar' (qCount stats) (+ 1)
+                  -- TODO: increment client Q counter
+                  -- TODO: increment current Q counter in IP timeline
+                  -- TODO: increment current Q counter in server timeline
                   case subMode of
                     SMOnlyCreate -> pure ()
                     SMSubscribe -> void $ subscribeQueue qr rId
@@ -826,6 +859,10 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                 case C.maxLenBS msgBody of
                   Left _ -> pure $ err LARGE_MSG
                   Right body -> do
+                    -- TODO: read client S rate
+                    -- TODO: read server S rate for peerId
+                    -- TODO: read global server S rate
+                    -- TODO: add throttling delay/blackhole request if needed
                     msg_ <- time "SEND" $ do
                       q <- getStoreMsgQueue "SEND" $ recipientId qr
                       expireMessages q
@@ -841,6 +878,9 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                         atomically $ modifyTVar' (msgSent stats) (+ 1)
                         atomically $ modifyTVar' (msgCount stats) (+ 1)
                         atomically $ updatePeriodStats (activeQueues stats) (recipientId qr)
+                        -- TODO: increment client S counter
+                        -- TODO: increment current S counter in IP timeline
+                        -- TODO: increment current S counter in server timeline
                         pure ok
           where
             mkMessage :: C.MaxLenBS MaxMessageLen -> M Message

src/Simplex/Messaging/Server/Env/STM.hs

@@ -16,6 +16,7 @@ import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Time.Clock (getCurrentTime)
+import Data.Time.Clock.POSIX (getPOSIXTime)
 import Data.Time.Clock.System (SystemTime)
 import Data.X509.Validation (Fingerprint (..))
 import Network.Socket (ServiceName)
@@ -33,7 +34,7 @@ import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Transport (ATransport, VersionRangeSMP, VersionSMP)
+import Simplex.Messaging.Transport (ATransport, PeerId, VersionRangeSMP, VersionSMP)
 import Simplex.Messaging.Transport.Server (SocketState, TransportServerConfig, alpn, loadFingerprint, loadTLSServerParams, newSocketState)
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
@@ -70,6 +71,10 @@ data ServerConfig = ServerConfig
     serverStatsLogFile :: FilePath,
     -- | file to save and restore stats
     serverStatsBackupFile :: Maybe FilePath,
+    -- | rate limit monitoring interval / bucket width, seconds
+    rateStatsInterval :: Maybe Int64,
+    rateStatsLogFile :: FilePath,
+    rateStatsBackupFile :: Maybe FilePath,
     -- | CA certificate private key is not needed for initialization
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
@@ -109,6 +114,8 @@ data Env = Env
     storeLog :: Maybe (StoreLog 'WriteMode),
     tlsServerParams :: T.ServerParams,
     serverStats :: ServerStats,
+    qCreatedByIp :: Timeline,
+    msgSentByIp :: Timeline,
     sockets :: SocketState,
     clientSeq :: TVar Int,
     clients :: TVar (IntMap Client)
@@ -124,6 +131,8 @@ data Server = Server
 
 data Client = Client
   { clientId :: Int,
+    peerId :: PeerId, -- send updates for this Id to time series
+    clientStats :: ClientStats, -- capture final values on disconnect
     subscriptions :: TMap RecipientId (TVar Sub),
     ntfSubscriptions :: TMap NotifierId (),
     rcvQ :: TBQueue (NonEmpty (Maybe QueueRec, Transmission Cmd)),
@@ -154,8 +163,8 @@ newServer = do
   savingLock <- createLock
   return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, savingLock}
 
-newClient :: TVar Int -> Natural -> VersionSMP -> ByteString -> SystemTime -> STM Client
-newClient nextClientId qSize thVersion sessionId createdAt = do
+newClient :: PeerId -> TVar Int -> Natural -> VersionSMP -> ByteString -> SystemTime -> STM Client
+newClient peerId nextClientId qSize thVersion sessionId createdAt = do
   clientId <- stateTVar nextClientId $ \next -> (next, next + 1)
   subscriptions <- TM.empty
   ntfSubscriptions <- TM.empty
@@ -166,7 +175,8 @@ newClient nextClientId qSize thVersion sessionId createdAt = do
   connected <- newTVar True
   rcvActiveAt <- newTVar createdAt
   sndActiveAt <- newTVar createdAt
-  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt}
+  clientStats <- ClientStats <$> newTVar 0 <*> newTVar 0
+  return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt, peerId, clientStats}
 
 newSubscription :: SubscriptionThread -> STM Sub
 newSubscription subThread = do
@@ -187,7 +197,10 @@ newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile,
   sockets <- atomically newSocketState
   clientSeq <- newTVarIO 0
   clients <- newTVarIO mempty
-  return Env {config, server, serverIdentity, queueStore, msgStore, random, storeLog, tlsServerParams, serverStats, sockets, clientSeq, clients}
+  now <- getPOSIXTime
+  qCreatedByIp <- atomically $ newTimeline perMinute now
+  msgSentByIp <- atomically $ newTimeline perMinute now
+  return Env {config, server, serverIdentity, queueStore, msgStore, random, storeLog, tlsServerParams, serverStats, qCreatedByIp, msgSentByIp, sockets, clientSeq, clients}
   where
     restoreQueues :: QueueStore -> FilePath -> IO (StoreLog 'WriteMode)
     restoreQueues QueueStore {queues, senders, notifiers} f = do

src/Simplex/Messaging/Server/Main.hs

@@ -208,6 +208,9 @@ smpServerCLI cfgPath logPath =
               logStatsStartTime = 0, -- seconds from 00:00 UTC
               serverStatsLogFile = combine logPath "smp-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
+              rateStatsInterval = Just 60, -- TODO: add to options
+              rateStatsLogFile = combine logPath "smp-server-rates.daily.log",
+              rateStatsBackupFile = Just $ combine logPath "smp-server-rates.log",
               smpServerVRange = supportedServerSMPRelayVRange,
               transportConfig =
                 defaultTransportServerConfig