Merge branch 'master' into ep/disable-background

* smp server: log prometheus metrics

* save metrics

* diff

* lines

* version

* do not include Prometheus into client

* corrections

Co-authored-by: sh <[email protected]>

* corrections

Co-authored-by: sh <[email protected]>

* corrections

Co-authored-by: sh <[email protected]>

* add timestamp to metrics

* remove type

* remove version

---------

Co-authored-by: sh <[email protected]>

rfcs/2024-12-15-server-cluster.md

@@ -1,85 +0,0 @@
-# SMP server cluster
-
-## Problem
-
-Currently we can only scale servers on a given address vertically, which has 2 problems:
-
-1. There are some limits to vertical scalability, with lower limits for cost-efficient scalability.
-2. Hard to manage resource exhaustion attacks.
-3. Creating new queue can fail and require address change.
-
-## Proposed solution
-
-While we don't need to implement clusters right now, we could continue evolving this design, and take some of the considerations to the design of short links.
-
-Server cluster can allow:
-1. Balance load on a given address horizontally, by adding servers to the cluster.
-2. Manage attacks by adding servers to cluster.
-3. Possibly, there may be approach to server-side redundancy.
-
-To group servers into a cluster we need to map requests to specific servers. This can be done in one of two ways:
-- additional server ID in the cluster added to transmissions.
-- map the first two letters in base64 encoding of queue ID to the server ID.
-
-The second approach makes it easy to migrate  parts of the state between servers in the cluster, as message queues are already grouped in folder with the top level having 2 letters in folder name. This would allow to have up to 4096 servers in the cluster.
-
-The proxy would then choose a random server from the list of servers to create a queue, and the server would have to be configured to use specific 2 letters in base64 encoding of queue addresses. For existing queues, the server will be choosed based on the queue ID.
-
-A separate question is how to map other queue IDs (sender, notifier, link) to the recipient ID. Two approaches are possible:
-- additional protocol commands used only by load-balancing proxy to create references to servers that have the actual queue from other IDs.
-- use the same 2 letters for all IDs.
-
-The latter approach is simpler, but it it cannot be used if some of the IDs are generated client-side and some IDs are generated server-side - we would need to generate all IDs in one place. Alternatively, the client can generate some IDs with the same 2 letters in the ID, and the server in the cluster will be chosen to match this ID.
-
-If we decide on the second approach, and add client-generated IDs, we already may start rejecting IDs that contain different first 2 letters. It would effectively reduce ID entropy from 192 to 180 bits which could be a better tradeoff than additional protocol commands and requests to find the queue, that would add to the request latency.
-
-The advantage of the first approach is that it is more generic, and does not impose any restriction on IDs, and making additional requests within the operators network would add a small fraction to the latency, compared with much larger latency to the end user. The balancing proxy could cache the results of dereferencing requests.
-
-**The sequence of requests to create new queue**:
-
-- client -> proxy: NEW
-- proxy -> server 1, choosen randomly: NEW
-- proxy <- server 1 (recipient ID should be one of IDs allowed for this server, based on cluster mapping): QID
-- proxy -> server 2 chosen based on sender ID: SET_SREF
-- proxy <- server 2: OK
-- proxy -> server 3 chosen based on notifier ID: SET_NREF
-- proxy <- server 3: OK
-- client <- proxy: QID
-
-SET_SREF and SET_NREF are new protocol commands to create references from sender and notifier IDs to recipient ID, or it may be sufficient to have the reference to cluster ID, as the server with queue would have to dereference it again in its local storage.
-
-**The sequence of requests to subscribe to receive messages**:
-
-- client -> proxy: SUB
-- proxy -> mapped server based on queue ID: SUB
-- proxy <- server: OK or MSG
-- client <- proxy: OK or MSG
-
-For subscriptions to work both the socket with the client and the socket with the server should remain open, and proxy should route the messages based on this session mapping.
-
-There are two possible mappings of sockets:
-- for each client socket the proxy creates a separate socket with the server. In this case response and message routing is simpler, and based purely on sockets.
-- proxy reuses connection to server and mixes all client requests into one connection with server. In this case response and message routing is more complex, and should take queue ID into account.
-
-**The sequence of requests to send the message**:
-
-- client -> proxy: SEND
-- proxy -> mapped server 1 based on sender ID: GET_SREF
-- proxy <- server 1: SREF
-- proxy -> mapped server 2 based on recipient ID (or cluster ID) in SREF: SEND
-- proxy <- server 2: OK
-- client <- proxy: OK
-
-SREF is the response containing the reference to find the server with the queue.
-
-## Proxy implementation
-
-It needs to be based on some general purpose proxy allowing TLS on both sides so that it can do protocol inspection.
-
-We considered haproxy and we have POC with open-resty (a fork of nginx with Lua scripting): https://github.com/epoberezkin/openresty-request-inspection
-
-There may be some other proxies with scripting and TCP protocol inspection that can be used, e.g. the search suggested: nginx, varnish, envoy, mitmproxy, caddy, apache traffic server, squid, traefik.
-
-As we also encrypt traffic inside TLS using client and server session keys, this traffic decryption and encryption of responses should happen in proxy, so the proxy should support encryption via openssl.
-
-Out of all options Apache Traffic Server looks attractive, as it supports plugins written in C/C++.

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.2.0.7
+version:        6.2.1.100
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -204,6 +204,7 @@ library
           Simplex.Messaging.Server.MsgStore.STM
           Simplex.Messaging.Server.MsgStore.Types
           Simplex.Messaging.Server.NtfStore
+          Simplex.Messaging.Server.Prometheus
           Simplex.Messaging.Server.QueueStore
           Simplex.Messaging.Server.QueueStore.STM
           Simplex.Messaging.Server.Stats

src/Simplex/Messaging/Agent.hs

@@ -221,13 +221,14 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp} stor
       | backgroundMode = run c "subscriber" $ subscriber c
       | otherwise = do
           restoreServersStats c
-          raceAny_
-            [ run c "subscriber" $ subscriber c,
-              run c "runNtfSupervisor" $ runNtfSupervisor c,
-              run c "cleanupManager" $ cleanupManager c,
-              run c "logServersStats" $ logServersStats c
-            ]
-            `E.finally` saveServersStats c
+          run c "subscriber" $ subscriber c
+          -- raceAny_
+          --   [ run c "subscriber" $ subscriber c
+          --     run c "runNtfSupervisor" $ runNtfSupervisor 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

src/Simplex/Messaging/Server.hs

@@ -71,6 +71,7 @@ import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
 import Data.Semigroup (Sum (..))
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
+import qualified Data.Text.IO as T
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
@@ -98,6 +99,7 @@ import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue, closeMsgQueue)
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
+import Simplex.Messaging.Server.Prometheus
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.QueueStore.STM
@@ -176,7 +178,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         : receiveFromProxyAgent pa
         : expireNtfsThread cfg
         : sigIntHandlerThread
-        : map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
+        : map runServer transports
+            <> expireMessagesThread_ cfg
+            <> serverStatsThread_ cfg
+            <> prometheusMetricsThread_ cfg
+            <> controlPortThread_ cfg
     )
     `finally` stopServer s
   where
@@ -555,6 +561,50 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         showProxyStats ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} =
           [show _pRequests, show _pSuccesses, show _pErrorsConnect, show _pErrorsCompat, show _pErrorsOther]
 
+    prometheusMetricsThread_ :: ServerConfig -> [M ()]
+    prometheusMetricsThread_ ServerConfig {prometheusInterval = Just interval, prometheusMetricsFile} =
+      [savePrometheusMetrics interval prometheusMetricsFile]
+    prometheusMetricsThread_ _ = []
+
+    savePrometheusMetrics :: Int -> FilePath -> M ()
+    savePrometheusMetrics saveInterval metricsFile = do
+      labelMyThread "savePrometheusMetrics"
+      liftIO $ putStrLn $ "Prometheus metrics saved every " <> show saveInterval <> " seconds to " <> metricsFile
+      AMS _ st <- asks msgStore
+      ss <- asks serverStats
+      env <- ask
+      let interval = 1000000 * saveInterval
+      liftIO $ forever $ do
+        threadDelay interval
+        ts <- getCurrentTime
+        sm <- getServerMetrics st ss
+        rtm <- getRealTimeMetrics env
+        T.writeFile metricsFile $ prometheusMetrics sm rtm ts
+
+    getServerMetrics :: STMQueueStore s => s -> ServerStats -> IO ServerMetrics
+    getServerMetrics st ss = do
+      d <- getServerStatsData ss
+      let ps = periodStatDataCounts $ _activeQueues d
+          psNtf = periodStatDataCounts $ _activeQueuesNtf d
+      queueCount <- M.size <$> readTVarIO (activeMsgQueues st)
+      notifierCount <- M.size <$> readTVarIO (notifiers' st)
+      pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount}
+
+    getRealTimeMetrics :: Env -> IO RealTimeMetrics
+    getRealTimeMetrics Env {clients, sockets, server = Server {subscribers, notifiers, subClients, ntfSubClients}} = do
+      socketStats <- mapM (traverse getSocketStats) =<< readTVarIO sockets
+#if MIN_VERSION_base(4,18,0)
+      threadsCount <- length <$> listThreads
+#else
+      let threadsCount = 0
+#endif
+      clientsCount <- IM.size <$> readTVarIO clients
+      smpSubsCount <- M.size <$> readTVarIO subscribers
+      smpSubClientsCount <- IM.size <$> readTVarIO subClients
+      ntfSubsCount <- M.size <$> readTVarIO notifiers
+      ntfSubClientsCount <- IM.size <$> readTVarIO ntfSubClients
+      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, smpSubsCount, smpSubClientsCount, ntfSubsCount, ntfSubClientsCount}
+
     runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
     runClient signKey tp h = do
       kh <- asks serverIdentity
@@ -695,13 +745,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
 #endif
               CPSockets -> withUserRole $ unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSockets
                 where
-                  putSockets (tcpPort, (accepted', closed', active')) = do
-                    (accepted, closed, active) <- (,,) <$> readTVarIO accepted' <*> readTVarIO closed' <*> readTVarIO active'
+                  putSockets (tcpPort, socketsState) = do
+                    ss <- getSocketStats socketsState
                     hPutStrLn h $ "Sockets for port " <> tcpPort <> ":"
-                    hPutStrLn h $ "accepted: " <> show accepted
-                    hPutStrLn h $ "closed: " <> show closed
-                    hPutStrLn h $ "active: " <> show (IM.size active)
-                    hPutStrLn h $ "leaked: " <> show (accepted - closed - IM.size active)
+                    hPutStrLn h $ "accepted: " <> show (socketsAccepted ss)
+                    hPutStrLn h $ "closed: " <> show (socketsClosed ss)
+                    hPutStrLn h $ "active: " <> show (socketsActive ss)
+                    hPutStrLn h $ "leaked: " <> show (socketsLeaked ss)
               CPSocketThreads -> withAdminRole $ do
 #if MIN_VERSION_base(4,18,0)
                 unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSocketThreads

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

@@ -96,6 +96,9 @@ data ServerConfig = ServerConfig
     serverStatsLogFile :: FilePath,
     -- | file to save and restore stats
     serverStatsBackupFile :: Maybe FilePath,
+    -- | interval and file to save prometheus metrics
+    prometheusInterval :: Maybe Int,
+    prometheusMetricsFile :: FilePath,
     -- | notification delivery interval
     ntfDeliveryInterval :: Int,
     -- | interval between sending pending END events to unsubscribed clients, seconds

src/Simplex/Messaging/Server/Main.hs

@@ -253,7 +253,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
                 <> "# Log daily server statistics to CSV file\n"
                 <> ("log_stats: " <> onOff logStats <> "\n\n")
-                <> "[AUTH]\n\
+                <> "# Log interval for real-time Prometheus metrics\n\
+                   \# prometheus_interval: 300\n\n\
+                   \[AUTH]\n\
                    \# Set new_queues option to off to completely prohibit creating new messaging queues.\n\
                    \# This can be useful when you want to decommission the server, but not all connections are switched yet.\n\
                    \new_queues: on\n\n\
@@ -431,6 +433,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles 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",
+              prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
+              prometheusMetricsFile = combine logPath "smp-server-metrics.txt",
               pendingENDInterval = 15000000, -- 15 seconds
               ntfDeliveryInterval = 3000000, -- 3 seconds
               smpServerVRange = supportedServerSMPRelayVRange,

src/Simplex/Messaging/Server/Prometheus.hs

@@ -0,0 +1,379 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
+{-# OPTIONS_GHC -fno-warn-unrecognised-pragmas #-}
+
+module Simplex.Messaging.Server.Prometheus where
+
+import Data.Int (Int64)
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Time.Clock (UTCTime (..), diffUTCTime)
+import Data.Time.Clock.System (systemEpochDay)
+import Data.Time.Format.ISO8601 (iso8601Show)
+import Network.Socket (ServiceName)
+import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.Transport.Server (SocketStats (..))
+
+data ServerMetrics = ServerMetrics
+  { statsData :: ServerStatsData,
+    activeQueueCounts :: PeriodStatCounts,
+    activeNtfCounts :: PeriodStatCounts,
+    queueCount :: Int,
+    notifierCount :: Int
+  }
+
+data RealTimeMetrics = RealTimeMetrics
+  { socketStats :: [(ServiceName, SocketStats)],
+    threadsCount :: Int,
+    clientsCount :: Int,
+    smpSubsCount :: Int,
+    smpSubClientsCount :: Int,
+    ntfSubsCount :: Int,
+    ntfSubClientsCount :: Int
+  }
+
+{-# FOURMOLU_DISABLE\n#-}
+prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
+prometheusMetrics sm rtm ts =
+  time <> queues <> subscriptions <> messages <> ntfMessages <> ntfs <> relays <> info
+  where
+    ServerMetrics {statsData, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount} = sm
+    RealTimeMetrics
+      { socketStats,
+        threadsCount,
+        clientsCount,
+        smpSubsCount,
+        smpSubClientsCount,
+        ntfSubsCount,
+        ntfSubClientsCount
+      } = rtm
+    ServerStatsData
+      { _fromTime,
+        _qCreated,
+        _qSecured,
+        _qDeletedAll,
+        _qDeletedAllB,
+        _qDeletedNew,
+        _qDeletedSecured,
+        _qSub,
+        _qSubAllB,
+        _qSubAuth,
+        _qSubDuplicate,
+        _qSubProhibited,
+        _qSubEnd,
+        _qSubEndB,
+        _ntfCreated,
+        _ntfDeleted,
+        _ntfDeletedB,
+        _ntfSub,
+        _ntfSubB,
+        _ntfSubAuth,
+        _ntfSubDuplicate,
+        _msgSent,
+        _msgSentAuth,
+        _msgSentQuota,
+        _msgSentLarge,
+        _msgRecv,
+        _msgRecvGet,
+        _msgGet,
+        _msgGetNoMsg,
+        _msgGetAuth,
+        _msgGetDuplicate,
+        _msgGetProhibited,
+        _msgExpired,
+        _activeQueues,
+        _msgSentNtf,
+        _msgRecvNtf,
+        _activeQueuesNtf,
+        _msgNtfs,
+        _msgNtfsB,
+        _msgNtfNoSub,
+        _msgNtfLost,
+        _msgNtfExpired,
+        _pRelays,
+        _pRelaysOwn,
+        _pMsgFwds,
+        _pMsgFwdsOwn,
+        _pMsgFwdsRecv,
+        _qCount,
+        _msgCount,
+        _ntfCount
+      } = statsData
+    time =
+      "# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
+      \# Stats from: " <> T.pack (iso8601Show _fromTime) <> "\n\
+      \\n"
+    queues =
+      "# Queues\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_queues_created Created queues\n\
+      \# TYPE simplex_smp_queues_created counter\n\
+      \simplex_smp_queues_created " <> mshow _qCreated <> "\n# qCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_secured Secured queues\n\
+      \# TYPE simplex_smp_queues_secured counter\n\
+      \simplex_smp_queues_secured " <> mshow _qSecured <> "\n# qSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted Deleted queues\n\
+      \# TYPE simplex_smp_queues_deleted counter\n\
+      \simplex_smp_queues_deleted{type=\"all\"} " <> mshow _qDeletedAll <> "\n# qDeleted\n\
+      \simplex_smp_queues_deleted{type=\"new\"} " <> mshow _qDeletedNew <> "\n# qDeletedNew\n\
+      \simplex_smp_queues_deleted{type=\"secured\"} " <> mshow _qDeletedSecured <> "\n# qDeletedSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted_batch Batched requests to delete queues\n\
+      \# TYPE simplex_smp_queues_deleted_batch counter\n\
+      \simplex_smp_queues_deleted_batch " <> mshow _qDeletedAllB <> "\n# qDeletedAllB\n\
+      \\n\
+      \# HELP simplex_smp_queues_total1 Total number of stored queues (first type of count).\n\
+      \# TYPE simplex_smp_queues_total1 gauge\n\
+      \simplex_smp_queues_total1 " <> mshow _qCount <> "\n# qCount\n\
+      \\n\
+      \# HELP simplex_smp_queues_total2 Total number of stored queues (second type of count).\n\
+      \# TYPE simplex_smp_queues_total2 gauge\n\
+      \simplex_smp_queues_total2 " <> mshow queueCount <> "\n# qCount2\n\
+      \\n\
+      \# HELP simplex_smp_queues_daily Daily active queues.\n\
+      \# TYPE simplex_smp_queues_daily gauge\n\
+      \simplex_smp_queues_daily " <> mstr (dayCount ps) <> "\n# dayMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_weekly Weekly active queues.\n\
+      \# TYPE simplex_smp_queues_weekly gauge\n\
+      \simplex_smp_queues_weekly " <> mstr (weekCount ps) <> "\n# weekMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_monthly Monthly active queues.\n\
+      \# TYPE simplex_smp_queues_monthly gauge\n\
+      \simplex_smp_queues_monthly " <> mstr (monthCount ps) <> "\n# monthMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_daily Daily active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_daily gauge\n\
+      \simplex_smp_queues_notify_daily " <> mstr (dayCount psNtf) <> "\n# dayCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_weekly Weekly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_weekly gauge\n\
+      \simplex_smp_queues_notify_weekly " <> mstr (weekCount psNtf) <> "\n# weekCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_monthly Monthly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_monthly gauge\n\
+      \simplex_smp_queues_notify_monthly " <> mstr (monthCount psNtf) <> "\n# monthCountNtf\n\
+      \\n"
+    subscriptions =
+      "# Subscriptions\n\
+      \# -------------\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes Successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes counter\n\
+      \simplex_smp_subscribtion_successes " <> mshow _qSub <> "\n# qSub\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes_batch Batched successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes_batch counter\n\
+      \simplex_smp_subscribtion_successes_batch " <> mshow _qSubAllB <> "\n# qSubAllB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end Ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end counter\n\
+      \simplex_smp_subscribtion_end " <> mshow _qSubEnd <> "\n# qSubEnd\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end_batch Batched ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end_batch counter\n\
+      \simplex_smp_subscribtion_end_batch " <> mshow _qSubEndB <> "\n# qSubEndB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_errors Subscription errors.\n\
+      \# TYPE simplex_smp_subscribtion_errors counter\n\
+      \simplex_smp_subscribtion_errors{type=\"auth\"} " <> mshow _qSubAuth <> "\n# qSubAuth\n\
+      \simplex_smp_subscribtion_errors{type=\"duplicate\"} " <> mshow _qSubDuplicate <> "\n# qSubDuplicate\n\
+      \simplex_smp_subscribtion_errors{type=\"prohibited\"} " <> mshow _qSubProhibited <> "\n# qSubProhibited\n\
+      \\n"
+    messages =
+      "# Messages\n\
+      \# --------\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent Sent messages.\n\
+      \# TYPE simplex_smp_messages_sent counter\n\
+      \simplex_smp_messages_sent " <> mshow _msgSent <> "\n# msgSent\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent_errors Total number of messages errors by type.\n\
+      \# TYPE simplex_smp_messages_sent_errors counter\n\
+      \simplex_smp_messages_sent_errors{type=\"auth\"} " <> mshow _msgSentAuth <> "\n# msgSentAuth\n\
+      \simplex_smp_messages_sent_errors{type=\"quota\"} " <> mshow _msgSentQuota <> "\n# msgSentQuota\n\
+      \simplex_smp_messages_sent_errors{type=\"large\"} " <> mshow _msgSentLarge <> "\n# msgSentLarge\n\
+      \\n\
+      \# HELP simplex_smp_messages_received Received messages.\n\
+      \# TYPE simplex_smp_messages_received counter\n\
+      \simplex_smp_messages_received " <> mshow _msgRecv <> "\n# msgRecv\n\
+      \\n\
+      \# HELP simplex_smp_messages_expired Expired messages.\n\
+      \# TYPE simplex_smp_messages_expired counter\n\
+      \simplex_smp_messages_expired " <> mshow _msgExpired <> "\n# msgExpired\n\
+      \\n\
+      \# HELP simplex_smp_messages_total Total number of messages stored.\n\
+      \# TYPE simplex_smp_messages_total gauge\n\
+      \simplex_smp_messages_total " <> mshow _msgCount <> "\n# msgCount\n\
+      \\n"
+    ntfMessages =
+      "# Notification messages (client)\n\
+      \# ------------------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_sent Sent messages with notification flag (cleint).\n\
+      \# TYPE simplex_smp_messages_notify_sent counter\n\
+      \simplex_smp_messages_notify_sent " <> mshow _msgSentNtf <> "\n# msgSentNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_received Received messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_received counter\n\
+      \simplex_smp_messages_notify_received " <> mshow _msgRecvNtf <> "\n# msgRecvNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_sent Requests to get messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_sent counter\n\
+      \simplex_smp_messages_notify_get_sent " <> mshow _msgGet <> "\n# msgGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_received Succesfully received get requests messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_received counter\n\
+      \simplex_smp_messages_notify_get_received " <> mshow _msgRecvGet <> "\n# msgRecvGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_errors Error events with messages with notification flag (client). \n\
+      \# TYPE simplex_smp_messages_notify_get_errors counter\n\
+      \simplex_smp_messages_notify_get_errors{type=\"nomsg\"} " <> mshow _msgGetNoMsg <> "\n# msgGetNoMsg\n\
+      \simplex_smp_messages_notify_get_errors{type=\"auth\"} " <> mshow _msgGetAuth <> "\n# msgGetAuth\n\
+      \simplex_smp_messages_notify_get_errors{type=\"duplicate\"} " <> mshow _msgGetDuplicate <> "\n# msgGetDuplicate\n\
+      \simplex_smp_messages_notify_get_errors{type=\"prohibited\"} " <> mshow _msgGetProhibited <> "\n# msgGetProhibited\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_created Created queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_created counter\n\
+      \simplex_smp_queues_notify_created " <> mshow _ntfCreated <> "\n# ntfCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted Deleted queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted counter\n\
+      \simplex_smp_queues_notify_deleted " <> mshow _ntfDeleted <> "\n# ntfDeleted\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted_batch Deleted batched queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted_batch counter\n\
+      \simplex_smp_queues_notify_deleted_batch " <> mshow _ntfDeletedB <> "\n# ntfDeletedB\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total1 Total number of stored queues with notification flag (first type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total1 gauge\n\
+      \simplex_smp_queues_notify_total1 " <> mshow _ntfCount <> "\n# ntfCount1\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total2 Total number of stored queues with notification flag (second type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total2 gauge\n\
+      \simplex_smp_queues_notify_total2 " <> mshow notifierCount <> "\n# ntfCount2\n\
+      \\n"
+    ntfs =
+      "# Notifications (server)\n\
+      \# ----------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes Successful events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes counter\n\
+      \simplex_smp_messages_ntf_successes " <> mshow _msgNtfs <> "\n# msgNtfs\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes_batch Successful batched events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes_batch counter\n\
+      \simplex_smp_messages_ntf_successes_batch " <> mshow _msgNtfsB <> "\n# msgNtfsB\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_errors Error events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_errors counter\n\
+      \simplex_smp_messages_ntf_errors{type=\"nosub\"} " <> mshow _msgNtfNoSub <> "\n# msgNtfNoSub\n\
+      \simplex_smp_messages_ntf_errors{type=\"lost\"} " <> mshow _msgNtfLost <> "\n# msgNtfLost\n\
+      \simplex_smp_messages_ntf_errors{type=\"expired\"} " <> mshow _msgNtfExpired <> "\n# msgNtfExpired\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests Subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests counter\n\
+      \simplex_smp_subscription_ntf_requests " <> mshow _ntfSub <> "\n# ntfSub\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests_batch Batched subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests_batch counter\n\
+      \simplex_smp_subscription_ntf_requests_batch " <> mshow _ntfSubB <> "\n# ntfSubB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_ntf_errors Subscription errors with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscribtion_ntf_errors counter\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"auth\"} " <> mshow _ntfSubAuth <> "\n# ntfSubAuth\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"duplicate\"} " <> mshow _ntfSubDuplicate <> "\n# ntfSubDuplicate\n\
+      \\n"
+    relays =
+      "# Relays\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_requests Session requests through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_requests counter\n\
+      \simplex_smp_relay_sessions_requests{source=\"all\"} " <> mshow (_pRequests _pRelays) <> "\n# pRelays_pRequests\n\
+      \simplex_smp_relay_sessions_requests{source=\"own\"} " <> mshow (_pRequests _pRelaysOwn) <> "\n# pRelaysOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_successes Successful session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_successes counter\n\
+      \simplex_smp_relay_sessions_successes{source=\"all\"} " <> mshow (_pSuccesses _pRelays) <> "\n# pRelays_pSuccesses\n\
+      \simplex_smp_relay_sessions_successes{source=\"own\"} " <> mshow (_pSuccesses _pRelaysOwn) <> "\n# pRelaysOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_errors Error session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_errors counter\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelays) <> "\n# pRelays_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelays) <> "\n# pRelays_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pRelays) <> "\n# pRelays_pErrorsOther\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_requests Message requests sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_requests counter\n\
+      \simplex_smp_relay_messages_requests{source=\"all\"} " <> mshow (_pRequests _pMsgFwds) <> "\n# pMsgFwds_pRequests\n\
+      \simplex_smp_relay_messages_requests{source=\"own\"} " <> mshow (_pRequests _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_successes Successful messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_successes counter\n\
+      \simplex_smp_relay_messages_successes{source=\"all\"} " <> mshow (_pSuccesses _pMsgFwds) <> "\n# pMsgFwds_pSuccesses\n\
+      \simplex_smp_relay_messages_successes{source=\"own\"} " <> mshow (_pSuccesses _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_errors Error events with messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_errors counter\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwds) <> "\n# pMsgFwds_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwds) <> "\n# pMsgFwds_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwds) <> "\n# pMsgFwds_pErrorsOther\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_received Relay messages statistics.\n\
+      \# TYPE simplex_smp_relay_messages_received counter\n\
+      \simplex_smp_relay_messages_received " <> mshow _pMsgFwdsRecv <> "\n# pMsgFwdsRecv\n\
+      \\n"
+    info =
+      "# Info\n\
+      \# ----\n\
+      \\n"
+      <> socketsMetric socketsAccepted "simplex_smp_sockets_accepted" "Accepted sockets"
+      <> socketsMetric socketsClosed "simplex_smp_sockets_closed" "Closed sockets"
+      <> socketsMetric socketsActive "simplex_smp_sockets_active" "Active sockets"
+      <> socketsMetric socketsLeaked "simplex_smp_sockets_leaked" "Leaked sockets"
+      <> "# HELP simplex_smp_threads_total Threads\n\
+      \# TYPE simplex_smp_threads_total gauge\n\
+      \simplex_smp_threads_total " <> mshow threadsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_clients_total Clients\n\
+      \# TYPE simplex_smp_clients_total gauge\n\
+      \simplex_smp_clients_total " <> mshow clientsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_total Total subscriptions\n\
+      \# TYPE simplex_smp_subscribtion_total gauge\n\
+      \simplex_smp_subscribtion_total " <> mshow smpSubsCount <> "\n# smpSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_clients_total Subscribed clients, first counting method\n\
+      \# TYPE simplex_smp_subscribtion_clients_total gauge\n\
+      \simplex_smp_subscribtion_clients_total " <> mshow smpSubClientsCount <> "\n# smpSubClients\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_total Total notification subscripbtions (from ntf server)\n\
+      \# TYPE simplex_smp_subscription_ntf_total gauge\n\
+      \simplex_smp_subscription_ntf_total " <> mshow ntfSubsCount <> "\n# ntfSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers, first counting method\n\
+      \# TYPE simplex_smp_subscription_ntf_clients_total gauge\n\
+      \simplex_smp_subscription_ntf_clients_total " <> mshow ntfSubClientsCount <> "\n# ntfSubClients\n"
+    socketsMetric :: (SocketStats -> Int) -> Text -> Text -> Text
+    socketsMetric sel metric descr =
+      "# HELP " <> metric <> " " <> descr <> "\n"
+        <> "# TYPE " <> metric <> " gauge\n"
+        <> T.concat (map (\(port, ss) -> metric <> "{port=\"" <> T.pack port <> "\"} " <> mshow (sel ss) <> "\n") socketStats)
+        <> "\n"
+    mstr a = T.pack a <> " " <> tsEpoch
+    mshow :: Show a => a -> Text
+    mshow = mstr . show
+    tsEpoch = T.pack $ show @Int64 $ floor @Double $ realToFrac (ts `diffUTCTime` epoch) * 1000
+    epoch = UTCTime systemEpochDay 0
+{-# FOURMOLU_ENABLE\n#-}

src/Simplex/Messaging/Server/Stats.hs

@@ -638,6 +638,14 @@ data PeriodStatCounts = PeriodStatCounts
     monthCount :: String
   }
 
+periodStatDataCounts :: PeriodStatsData -> PeriodStatCounts
+periodStatDataCounts PeriodStatsData {_day, _week, _month} =
+  PeriodStatCounts
+    { dayCount = show $ IS.size _day,
+      weekCount = show $ IS.size _week,
+      monthCount = show $ IS.size _month
+    }
+
 periodStatCounts :: PeriodStats -> UTCTime -> IO PeriodStatCounts
 periodStatCounts ps ts = do
   let d = utctDay ts

src/Simplex/Messaging/Transport/Server.hs

@@ -12,7 +12,9 @@ module Simplex.Messaging.Transport.Server
     runTransportServerState,
     runTransportServerState_,
     SocketState,
+    SocketStats (..),
     newSocketState,
+    getSocketStats,
     runTransportServer,
     runTransportServerSocket,
     runLocalTCPServer,
@@ -163,9 +165,24 @@ safeAccept sock =
 
 type SocketState = (TVar Int, TVar Int, TVar (IntMap (Weak ThreadId)))
 
+data SocketStats = SocketStats
+  { socketsAccepted :: Int,
+    socketsClosed :: Int,
+    socketsActive :: Int,
+    socketsLeaked :: Int
+  }
+
 newSocketState :: IO SocketState
 newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty
 
+getSocketStats :: SocketState -> IO SocketStats
+getSocketStats (accepted, closed, active) = do
+  socketsAccepted <- readTVarIO accepted
+  socketsClosed <- readTVarIO closed
+  socketsActive <- IM.size <$> readTVarIO active
+  let socketsLeaked = socketsAccepted - socketsClosed - socketsActive
+  pure SocketStats {socketsAccepted, socketsClosed, socketsActive, socketsLeaked}
+
 closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
   close sock

tests/SMPClient.hs

@@ -79,6 +79,9 @@ testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
 testStoreNtfsFile2 :: FilePath
 testStoreNtfsFile2 = "tests/tmp/smp-server-ntfs.log.2"
 
+testPrometheusMetricsFile :: FilePath
+testPrometheusMetricsFile = "tests/tmp/smp-server-metrics.txt"
+
 testServerStatsBackupFile :: FilePath
 testServerStatsBackupFile = "tests/tmp/smp-server-stats.log"
 
@@ -141,8 +144,10 @@ cfgMS msType =
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       logStatsInterval = Nothing,
       logStatsStartTime = 0,
-      serverStatsLogFile = "tests/smp-server-stats.daily.log",
+      serverStatsLogFile = "tests/tmp/smp-server-stats.daily.log",
       serverStatsBackupFile = Nothing,
+      prometheusInterval = Nothing,
+      prometheusMetricsFile = testPrometheusMetricsFile,
       pendingENDInterval = 500000,
       ntfDeliveryInterval = 200000,
       smpCredentials =

tests/ServerTests.hs

@@ -46,7 +46,7 @@ import Simplex.Messaging.Server.StoreLog (closeStoreLog)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (whenM)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (doesDirectoryExist, removeDirectoryRecursive, removeFile)
+import System.Directory (doesDirectoryExist, doesFileExist, removeDirectoryRecursive, removeFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
@@ -71,6 +71,7 @@ serverTests  = do
   describe "Store log" testWithStoreLog
   describe "Restore messages" testRestoreMessages
   describe "Restore messages (old / v2)" testRestoreExpireMessages
+  describe "Save prometheus metrics" testPrometheusMetrics
   describe "Timing of AUTH error" testTiming
   describe "Message notifications" testMessageNotifications
   describe "Message expiration" $ do
@@ -825,6 +826,13 @@ testRestoreExpireMessages =
     runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
+testPrometheusMetrics :: SpecWith (ATransport, AMSType)
+testPrometheusMetrics =
+  it "should save Prometheus metrics" $ \(at, msType) -> do
+    let cfg' = (cfgMS msType) {prometheusInterval = Just 1}
+    withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1000000
+    doesFileExist testPrometheusMetricsFile `shouldReturn` True
+
 createAndSecureQueue :: Transport c => THandleSMP c 'TClient -> SndPublicAuthKey -> IO (SenderId, RecipientId, RcvPrivateAuthKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
   g <- C.newRandom