From 50e4abf8369c2d6a70ff32b41d909f9a61ff0b4d Mon Sep 17 00:00:00 2001 From: Evgeny Date: Sun, 6 Oct 2024 12:33:14 +0100 Subject: [PATCH] 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 | 1 + src/Simplex/FileTransfer/Server.hs | 1 + src/Simplex/FileTransfer/Server/Control.hs | 2 - src/Simplex/FileTransfer/Server/Main.hs | 1 + src/Simplex/Messaging/Client/Agent.hs | 4 +- src/Simplex/Messaging/Notifications/Server.hs | 152 +++++++++++++++++- .../Messaging/Notifications/Server/Control.hs | 37 +++++ .../Messaging/Notifications/Server/Env.hs | 5 +- .../Messaging/Notifications/Server/Main.hs | 13 ++ tests/NtfClient.hs | 3 + 10 files changed, 206 insertions(+), 13 deletions(-) create mode 100644 src/Simplex/Messaging/Notifications/Server/Control.hs diff --git a/simplexmq.cabal b/simplexmq.cabal index 43e18ff5c..a77dfe9b0 100644 --- a/simplexmq.cabal +++ b/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 diff --git a/src/Simplex/FileTransfer/Server.hs b/src/Simplex/FileTransfer/Server.hs index 24ad4815e..bfe5b8dd0 100644 --- a/src/Simplex/FileTransfer/Server.hs +++ b/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 diff --git a/src/Simplex/FileTransfer/Server/Control.hs b/src/Simplex/FileTransfer/Server/Control.hs index 75e54e55a..3a84ace49 100644 --- a/src/Simplex/FileTransfer/Server/Control.hs +++ b/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 diff --git a/src/Simplex/FileTransfer/Server/Main.hs b/src/Simplex/FileTransfer/Server/Main.hs index 1f9246b1c..aca3b60df 100644 --- a/src/Simplex/FileTransfer/Server/Main.hs +++ b/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") diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index 378b470a3..1a7a67806 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/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 = [] } diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 42c24aeb8..4e9f75f9e 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/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..." diff --git a/src/Simplex/Messaging/Notifications/Server/Control.hs b/src/Simplex/Messaging/Notifications/Server/Control.hs new file mode 100644 index 000000000..234baede2 --- /dev/null +++ b/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" diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 8fd2a33ad..6ac250258 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/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, diff --git a/src/Simplex/Messaging/Notifications/Server/Main.hs b/src/Simplex/Messaging/Notifications/Server/Main.hs index 30b241166..0c5d5e54b 100644 --- a/src/Simplex/Messaging/Notifications/Server/Main.hs +++ b/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, diff --git a/tests/NtfClient.hs b/tests/NtfClient.hs index 6c38af721..05055fe45 100644 --- a/tests/NtfClient.hs +++ b/tests/NtfClient.hs @@ -87,6 +87,9 @@ ntfServerCfg :: NtfServerConfig ntfServerCfg = NtfServerConfig { transports = [], + controlPort = Nothing, + controlPortUserAuth = Nothing, + controlPortAdminAuth = Nothing, subIdBytes = 24, regCodeBytes = 32, clientQSize = 2,