ntf server: control port (#1354)

* ntf server: control port

* version .405

* control

* use own_server_domains from INI file

* fix subs by server in control port

* bigger queues

* ntf server: only print subscriptions per own server when they are > 0

* fix tests

* revert version change

* dont import listThreads in ghc 8.10.7

simplexmq.cabal

@@ -154,6 +154,7 @@ library
       Simplex.Messaging.Notifications.Client
       Simplex.Messaging.Notifications.Protocol
       Simplex.Messaging.Notifications.Server
+      Simplex.Messaging.Notifications.Server.Control
       Simplex.Messaging.Notifications.Server.Env
       Simplex.Messaging.Notifications.Server.Main
       Simplex.Messaging.Notifications.Server.Push.APNS

src/Simplex/FileTransfer/Server.hs

@@ -55,6 +55,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (CorrId (..), EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth, pattern NoEntity)
 import Simplex.Messaging.Server (dummyVerifyCmd, verifyCmdAuthorization)
+import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, getRoundedSystemTime)
 import Simplex.Messaging.Server.Stats

src/Simplex/FileTransfer/Server/Control.hs

@@ -8,8 +8,6 @@ import Simplex.FileTransfer.Protocol (XFTPFileId)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (BasicAuth)
 
-data CPClientRole = CPRNone | CPRUser | CPRAdmin
-
 data ControlProtocol
   = CPAuth BasicAuth
   | CPStatsRTS

src/Simplex/FileTransfer/Server/Main.hs

@@ -101,6 +101,7 @@ xftpServerCLI cfgPath logPath = do
                \\n\
                \# control_port_admin_password:\n\
                \# control_port_user_password:\n\
+               \\n\
                \[TRANSPORT]\n\
                \# host is only used to print server address on start\n"
             <> ("host: " <> T.pack host <> "\n")

src/Simplex/Messaging/Client/Agent.hs

@@ -84,8 +84,8 @@ defaultSMPClientAgentConfig =
             maxInterval = 10 * second
           },
       persistErrorInterval = 30, -- seconds
-      msgQSize = 1024,
-      agentQSize = 1024,
+      msgQSize = 2048,
+      agentQSize = 2048,
       agentSubsBatchSize = 1360,
       ownServerDomains = []
     }

src/Simplex/Messaging/Notifications/Server.hs

@@ -1,3 +1,5 @@
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
@@ -23,7 +25,8 @@ import Data.Either (partitionEithers)
 import Data.Functor (($>))
 import Data.IORef
 import Data.Int (Int64)
-import Data.List (intercalate, sort)
+import qualified Data.IntSet as IS
+import Data.List (intercalate, partition, sort)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
@@ -34,34 +37,42 @@ import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import GHC.IORef (atomicSwapIORef)
-import Network.Socket (ServiceName)
+import GHC.Stats (getRTSStats)
+import Network.Socket (ServiceName, Socket, socketToHandle)
 import Simplex.Messaging.Client (ProtocolClientError (..), SMPClientError, ServerTransmission (..))
 import Simplex.Messaging.Client.Agent
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
+import Simplex.Messaging.Notifications.Server.Control
 import Simplex.Messaging.Notifications.Server.Env
 import Simplex.Messaging.Notifications.Server.Push.APNS (PNMessageData (..), PushNotification (..), PushProviderError (..))
 import Simplex.Messaging.Notifications.Server.Stats
 import Simplex.Messaging.Notifications.Server.Store
 import Simplex.Messaging.Notifications.Server.StoreLog
 import Simplex.Messaging.Notifications.Transport
-import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), ProtocolServer (host), SMPServer, SignedTransmission, Transmission, pattern NoEntity, encodeTransmission, tGet, tPut)
+import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), ProtocolServer (host), SMPServer, SignedTransmission, Transmission, pattern NoEntity, pattern SMPServer, encodeTransmission, tGet, tPut)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server
-import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.Server.Control (CPClientRole (..))
+import Simplex.Messaging.Server.Stats (PeriodStats (..), PeriodStatCounts (..), periodStatCounts, updatePeriodStats)
+import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (ATransport (..), THandle (..), THandleAuth (..), THandleParams (..), TProxy, Transport (..), TransportPeer (..), defaultSupportedParams)
-import Simplex.Messaging.Transport.Server (AddHTTP, runTransportServer)
+import Simplex.Messaging.Transport.Buffer (trimCR)
+import Simplex.Messaging.Transport.Server (AddHTTP, runTransportServer, runLocalTCPServer)
 import Simplex.Messaging.Util
 import System.Exit (exitFailure)
-import System.IO (BufferMode (..), hPutStrLn, hSetBuffering)
+import System.IO (BufferMode (..), hClose, hPrint, hPutStrLn, hSetBuffering, hSetNewlineMode, universalNewlineMode)
 import System.Mem.Weak (deRefWeak)
-import UnliftIO (IOMode (..), async, uninterruptibleCancel, withFile)
+import UnliftIO (IOMode (..), UnliftIO, askUnliftIO, async, uninterruptibleCancel, unliftIO, withFile)
 import UnliftIO.Concurrent (forkIO, killThread, mkWeakThreadId)
 import UnliftIO.Directory (doesFileExist, renameFile)
 import UnliftIO.Exception
 import UnliftIO.STM
+#if MIN_VERSION_base(4,18,0)
+import GHC.Conc (listThreads)
+#endif
 
 runNtfServer :: NtfServerConfig -> IO ()
 runNtfServer cfg = do
@@ -79,7 +90,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
   s <- asks subscriber
   ps <- asks pushServer
   resubscribe s
-  raceAny_ (ntfSubscriber s : ntfPush ps : map runServer transports <> serverStatsThread_ cfg) `finally` stopServer
+  raceAny_ (ntfSubscriber s : ntfPush ps : map runServer transports <> serverStatsThread_ cfg <> controlPortThread_ cfg) `finally` stopServer
   where
     runServer :: (ServiceName, ATransport, AddHTTP) -> M ()
     runServer (tcpPort, ATransport t, _addHTTP) = do
@@ -152,6 +163,131 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
               ]
         liftIO $ threadDelay' interval
 
+    controlPortThread_ :: NtfServerConfig -> [M ()]
+    controlPortThread_ NtfServerConfig {controlPort = Just port} = [runCPServer port]
+    controlPortThread_ _ = []
+
+    runCPServer :: ServiceName -> M ()
+    runCPServer port = do
+      cpStarted <- newEmptyTMVarIO
+      u <- askUnliftIO
+      liftIO $ do
+        labelMyThread "control port server"
+        runLocalTCPServer cpStarted port $ runCPClient u
+      where
+        runCPClient :: UnliftIO (ReaderT NtfEnv IO) -> Socket -> IO ()
+        runCPClient u sock = do
+          labelMyThread "control port client"
+          h <- socketToHandle sock ReadWriteMode
+          hSetBuffering h LineBuffering
+          hSetNewlineMode h universalNewlineMode
+          hPutStrLn h "Ntf server control port\n'help' for supported commands"
+          role <- newTVarIO CPRNone
+          cpLoop h role
+          where
+            cpLoop h role = do
+              s <- trimCR <$> B.hGetLine h
+              case strDecode s of
+                Right CPQuit -> hClose h
+                Right cmd -> logCmd s cmd >> processCP h role cmd >> cpLoop h role
+                Left err -> hPutStrLn h ("error: " <> err) >> cpLoop h role
+            logCmd s cmd = when shouldLog $ logWarn $ "ControlPort: " <> tshow s
+              where
+                shouldLog = case cmd of
+                  CPAuth _ -> False
+                  CPHelp -> False
+                  CPQuit -> False
+                  CPSkip -> False
+                  _ -> True
+            processCP h role = \case
+              CPAuth auth -> atomically $ writeTVar role $! newRole cfg
+                where
+                  newRole NtfServerConfig {controlPortUserAuth = user, controlPortAdminAuth = admin}
+                    | Just auth == admin = CPRAdmin
+                    | Just auth == user = CPRUser
+                    | otherwise = CPRNone
+              CPStats -> withUserRole $ do
+                ss <- unliftIO u $ asks serverStats
+                let getStat :: (NtfServerStats -> IORef a) -> IO a
+                    getStat var = readIORef (var ss)
+                    putStat :: Show a => String -> (NtfServerStats -> IORef a) -> IO ()
+                    putStat label var = getStat var >>= \v -> hPutStrLn h $ label <> ": " <> show v
+                putStat "fromTime" fromTime
+                putStat "tknCreated" tknCreated
+                putStat "tknVerified" tknVerified
+                putStat "tknDeleted" tknDeleted
+                putStat "subCreated" subCreated
+                putStat "subDeleted" subDeleted
+                putStat "ntfReceived" ntfReceived
+                getStat (day . activeTokens) >>= \v -> hPutStrLn h $ "daily active tokens: " <> show (IS.size v)
+                getStat (day . activeSubs) >>= \v -> hPutStrLn h $ "daily active subscriptions: " <> show (IS.size v)
+              CPStatsRTS -> tryAny getRTSStats >>= either (hPrint h) (hPrint h)
+              CPServerInfo -> readTVarIO role >>= \case
+                CPRNone -> do
+                  logError "Unauthorized control port command"
+                  hPutStrLn h "AUTH"
+                r -> do
+#if MIN_VERSION_base(4,18,0)
+                  threads <- liftIO listThreads
+                  hPutStrLn h $ "Threads: " <> show (length threads)
+#else
+                  hPutStrLn h "Threads: not available on GHC 8.10"
+#endif
+                  NtfSubscriber {smpSubscribers, smpAgent = a} <- unliftIO u $ asks subscriber
+                  putSMPWorkers a "SMP subcscribers" smpSubscribers
+                  let SMPClientAgent {smpClients, smpSessions, srvSubs, pendingSrvSubs, smpSubWorkers} = a
+                  putSMPWorkers a "SMP clients" smpClients
+                  putSMPWorkers a "SMP subscription workers" smpSubWorkers
+                  sessions <- readTVarIO smpSessions
+                  hPutStrLn h $ "SMP sessions count: " <> show (M.size sessions)
+                  putSMPSubs a "SMP subscriptions" srvSubs
+                  putSMPSubs a "Pending SMP subscriptions" pendingSrvSubs
+                  where
+                    putSMPSubs :: SMPClientAgent -> String -> TMap SMPServer (TMap SMPSub a) -> IO ()
+                    putSMPSubs a name v = do
+                      subs <- readTVarIO v
+                      (totalCnt, ownCount, otherCnt, servers, ownByServer) <- foldM countSubs (0, 0, 0, [], M.empty) $ M.assocs subs
+                      showServers a name servers
+                      hPutStrLn h $ name <> " total: " <> show totalCnt
+                      hPutStrLn h $ name <> " on own servers: " <> show ownCount
+                      when (r == CPRAdmin && not (null ownByServer)) $
+                        forM_ (M.assocs ownByServer) $ \(SMPServer (host :| _) _ _, cnt) ->
+                          hPutStrLn h $ name <> " on " <> B.unpack (strEncode host) <> ": " <> show cnt
+                      hPutStrLn h $ name <> " on other servers: " <> show otherCnt
+                      where
+                        countSubs :: (Int, Int, Int, [SMPServer], M.Map SMPServer Int) -> (SMPServer, TMap SMPSub a) -> IO (Int, Int, Int, [SMPServer], M.Map SMPServer Int)
+                        countSubs (!totalCnt, !ownCount, !otherCnt, !servers, !ownByServer) (srv, srvSubs) = do
+                          cnt <- M.size <$> readTVarIO srvSubs
+                          let totalCnt' = totalCnt + cnt
+                              ownServer = isOwnServer a srv
+                              (ownCount', otherCnt')
+                                | ownServer = (ownCount + cnt, otherCnt)
+                                | otherwise = (ownCount, otherCnt + cnt)
+                              servers' = if cnt > 0 then srv : servers else servers
+                              ownByServer'
+                                | r == CPRAdmin && ownServer && cnt > 0 = M.alter (Just . maybe cnt (+ cnt)) srv ownByServer
+                                | otherwise = ownByServer
+                          pure (totalCnt', ownCount', otherCnt', servers', ownByServer')
+                    putSMPWorkers :: SMPClientAgent -> String -> TMap SMPServer a -> IO ()
+                    putSMPWorkers a name v = readTVarIO v >>= showServers a name . M.keys
+                    showServers :: SMPClientAgent -> String -> [SMPServer] -> IO ()
+                    showServers a name srvs = do
+                      let (ownSrvs, otherSrvs) = partition (isOwnServer a) srvs
+                      hPutStrLn h $ name <> " own servers count: " <> show (length ownSrvs)
+                      when (r == CPRAdmin) $ hPutStrLn h $ name <> " own servers: " <> intercalate "," (sort $ map (\(SMPServer (host :| _) _ _) -> B.unpack $ strEncode host) ownSrvs)
+                      hPutStrLn h $ name <> " other servers count: " <> show (length otherSrvs)
+              CPHelp -> hPutStrLn h "commands: stats, stats-rts, server-info, help, quit"
+              CPQuit -> pure ()
+              CPSkip -> pure ()
+              where
+                withUserRole action =
+                  readTVarIO role >>= \case
+                    CPRAdmin -> action
+                    CPRUser -> action
+                    _ -> do
+                      logError "Unauthorized control port command"
+                      hPutStrLn h "AUTH"
+
 resubscribe :: NtfSubscriber -> M ()
 resubscribe NtfSubscriber {newSubQ} = do
   logInfo "Preparing SMP resubscriptions..."

src/Simplex/Messaging/Notifications/Server/Control.hs

@@ -0,0 +1,37 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Notifications.Server.Control where
+
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Protocol (BasicAuth)
+
+data ControlProtocol
+  = CPAuth BasicAuth
+  | CPStats
+  | CPStatsRTS
+  | CPServerInfo
+  | CPHelp
+  | CPQuit
+  | CPSkip
+
+instance StrEncoding ControlProtocol where
+  strEncode = \case
+    CPAuth tok -> "auth " <> strEncode tok
+    CPStats -> "stats"
+    CPStatsRTS -> "stats-rts"
+    CPServerInfo -> "server-info"
+    CPHelp -> "help"
+    CPQuit -> "quit"
+    CPSkip -> ""
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "auth" -> CPAuth <$> _strP
+      "stats" -> pure CPStats
+      "stats-rts" -> pure CPStatsRTS
+      "server-info" -> pure CPServerInfo
+      "help" -> pure CPHelp
+      "quit" -> pure CPQuit
+      "" -> pure CPSkip
+      _ -> fail "bad ControlProtocol command"

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

@@ -28,7 +28,7 @@ import Simplex.Messaging.Notifications.Server.Stats
 import Simplex.Messaging.Notifications.Server.Store
 import Simplex.Messaging.Notifications.Server.StoreLog
 import Simplex.Messaging.Notifications.Transport (NTFVersion, VersionRangeNTF)
-import Simplex.Messaging.Protocol (CorrId, SMPServer, Transmission)
+import Simplex.Messaging.Protocol (BasicAuth, CorrId, SMPServer, Transmission)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
@@ -40,6 +40,9 @@ import UnliftIO.STM
 
 data NtfServerConfig = NtfServerConfig
   { transports :: [(ServiceName, ATransport, AddHTTP)],
+    controlPort :: Maybe ServiceName,
+    controlPortUserAuth :: Maybe BasicAuth,
+    controlPortAdminAuth :: Maybe BasicAuth,
     subIdBytes :: Int,
     regCodeBytes :: Int,
     clientQSize :: Natural,

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -14,6 +14,7 @@ import Data.Functor (($>))
 import Data.Ini (lookupValue, readIniFile)
 import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
+import Data.Text.Encoding (encodeUtf8)
 import qualified Data.Text.IO as T
 import Network.Socket (HostName)
 import Options.Applicative
@@ -85,6 +86,10 @@ ntfServerCLI cfgPath logPath =
           \# Log is compacted on start (deleted objects are removed).\n"
             <> ("enable: " <> onOff enableStoreLog <> "\n\n")
             <> "log_stats: off\n\n\
+               \[AUTH]\n\
+               \# control_port_admin_password:\n\
+               \# control_port_user_password:\n\
+               \\n\
                \[TRANSPORT]\n\
                \# Host is only used to print server address on start.\n\
                \# You can specify multiple server ports.\n"
@@ -93,6 +98,8 @@ ntfServerCLI cfgPath logPath =
             <> "log_tls_errors: off\n\n\
                \# Use `websockets: 443` to run websockets server in addition to plain TLS.\n\
                \websockets: off\n\n\
+               \# control_port: 5227\n\
+               \\n\
                \[SUBSCRIBER]\n\
                \# Network configuration for notification server client.\n\
                \# `host_mode` can be 'public' (default) or 'onion'.\n\
@@ -105,6 +112,8 @@ ntfServerCLI cfgPath logPath =
                \# `socks_mode` can be 'onion' for SOCKS proxy to be used for .onion destination hosts only (default)\n\
                \# or 'always' to be used for all destination hosts (can be used if it is an .onion server).\n\
                \# socks_mode: onion\n\n\
+               \# The domain suffixes of the relays you operate (space-separated) to count as separate proxy statistics.\n\
+               \# own_server_domains: \n\n\
                \[INACTIVE_CLIENTS]\n\
                \# TTL and interval to check inactive clients\n\
                \disconnect: off\n"
@@ -128,6 +137,9 @@ ntfServerCLI cfgPath logPath =
         serverConfig =
           NtfServerConfig
             { transports = iniTransports ini,
+              controlPort = either (const Nothing) (Just . T.unpack) $ lookupValue "TRANSPORT" "control_port" ini,
+              controlPortAdminAuth = either error id <$!> strDecodeIni "AUTH" "control_port_admin_password" ini,
+              controlPortUserAuth = either error id <$!> strDecodeIni "AUTH" "control_port_user_password" ini,
               subIdBytes = 24,
               regCodeBytes = 32,
               clientQSize = 64,
@@ -146,6 +158,7 @@ ntfServerCLI cfgPath logPath =
                                 smpPingInterval = 60_000_000 -- 1 minute
                               }
                         },
+                    ownServerDomains = either (const []) (map encodeUtf8 . T.words) $ lookupValue "SUBSCRIBER" "own_server_domains" ini,
                     persistErrorInterval = 0 -- seconds
                   },
               apnsConfig = defaultAPNSPushClientConfig,

tests/NtfClient.hs

@@ -87,6 +87,9 @@ ntfServerCfg :: NtfServerConfig
 ntfServerCfg =
   NtfServerConfig
     { transports = [],
+      controlPort = Nothing,
+      controlPortUserAuth = Nothing,
+      controlPortAdminAuth = Nothing,
       subIdBytes = 24,
       regCodeBytes = 32,
       clientQSize = 2,