ntf server: additional statistics (#1558)

epoberezkin · web-flow · commit 46ff37c36205 · 2025-06-09T14:14:27.000+01:00
* ntf server: additional statistics

* version

* fix stats

* add stats to track notifications without active token

* refactor

* fix stats parser

* version
diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs
@@ -74,7 +74,6 @@ import Simplex.Messaging.Protocol
     subscriberParty,
     subscriberServiceRole
   )
-import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Session
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
diff --git a/src/Simplex/Messaging/Encoding/String.hs b/src/Simplex/Messaging/Encoding/String.hs
@@ -43,7 +43,7 @@ import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (parseAll)
-import Simplex.Messaging.Util (bshow, (<$?>))
+import Simplex.Messaging.Util (bshow, safeDecodeUtf8, (<$?>))
 
 class TextEncoding a where
   textEncode :: a -> Text
@@ -91,6 +91,10 @@ instance StrEncoding String where
   strEncode = strEncode . B.pack
   strP = B.unpack <$> strP
 
+instance StrEncoding Text where
+  strEncode = encodeUtf8
+  strP = safeDecodeUtf8 <$> A.takeTill (\c -> c == ' ' || c == '\n')
+
 instance ToJSON Str where
   toJSON (Str s) = strToJSON s
   toEncoding (Str s) = strToJEncoding s
diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs
@@ -281,7 +281,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
         getSMPServiceSubMetrics a sel subQueueCount = getSubMetrics_ a sel countSubs
           where
             countSubs :: (NtfSMPSubMetrics, S.Set Text) -> (SMPServer, TVar (Maybe sub)) -> IO (NtfSMPSubMetrics, S.Set Text)
-            countSubs acc (srv, serviceSubs) = maybe acc (subMetricsResult a acc srv . fromIntegral . subQueueCount) <$> readTVarIO serviceSubs
+            countSubs acc (srv, serviceSubs) = subMetricsResult a acc srv . fromIntegral . maybe 0 subQueueCount <$> readTVarIO serviceSubs
 
         getSMPSubMetrics :: SMPClientAgent 'Notifier -> (SMPClientAgent 'Notifier -> TMap SMPServer (TMap NotifierId a)) -> IO NtfSMPSubMetrics
         getSMPSubMetrics a sel = getSubMetrics_ a sel countSubs
@@ -305,11 +305,11 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
           | isOwnServer a srv =
               let !ownSrvSubs' = M.alter (Just . maybe cnt (+ cnt)) host ownSrvSubs
                   metrics' = metrics {ownSrvSubs = ownSrvSubs'} :: NtfSMPSubMetrics
-                in (metrics', otherSrvs)
+               in (metrics', otherSrvs)
           | cnt == 0 = acc
           | otherwise =
               let metrics' = metrics {otherSrvSubCount = otherSrvSubCount + cnt} :: NtfSMPSubMetrics
-                in (metrics', S.insert host otherSrvs)
+               in (metrics', S.insert host otherSrvs)
           where
             NtfSMPSubMetrics {ownSrvSubs, otherSrvSubCount} = metrics
             host = safeDecodeUtf8 $ strEncode h
@@ -527,7 +527,7 @@ ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} =
       NtfPushServer {pushQ} <- asks pushServer
       stats <- asks serverStats
       liftIO $ forever $ do
-        ((_, srv, _), _thVersion, sessionId, ts) <- atomically $ readTBQueue msgQ
+        ((_, srv@(SMPServer (h :| _) _ _), _), _thVersion, sessionId, ts) <- atomically $ readTBQueue msgQ
         forM ts $ \(ntfId, t) -> case t of
           STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen
           STResponse {} -> pure () -- it was already reported as timeout error
@@ -538,9 +538,16 @@ ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} =
                 ntfTs <- getSystemTime
                 updatePeriodStats (activeSubs stats) ntfId
                 let newNtf = PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}
-                ntfs_ <- addTokenLastNtf st newNtf
-                forM_ ntfs_ $ \(tkn, lastNtfs) -> atomically $ writeTBQueue pushQ (tkn, PNMessage lastNtfs)
-                incNtfStat_ stats ntfReceived
+                    srvHost_ = if isOwnServer ca srv then Just (safeDecodeUtf8 $ strEncode h) else Nothing
+                addTokenLastNtf st newNtf >>= \case
+                  Right (tkn, lastNtfs) -> do
+                    atomically $ writeTBQueue pushQ (srvHost_, tkn, PNMessage lastNtfs)
+                    incNtfStat_ stats ntfReceived
+                    mapM_ (`incServerStat` ntfReceivedOwn stats) srvHost_
+                  Left AUTH -> do
+                    incNtfStat_ stats ntfReceivedAuth
+                    mapM_ (`incServerStat` ntfReceivedAuthOwn stats) srvHost_
+                  Left _ -> pure ()
               Right SMP.END ->
                 whenM (atomically $ activeClientSession' ca sessionId srv) $
                   void $ updateSrvSubStatus st smpQueue NSEnd
@@ -625,7 +632,7 @@ showServer' = decodeLatin1 . strEncode . host
 
 ntfPush :: NtfPushServer -> M ()
 ntfPush s@NtfPushServer {pushQ} = forever $ do
-  (tkn@NtfTknRec {ntfTknId, token = t@(DeviceToken pp _), tknStatus}, ntf) <- atomically (readTBQueue pushQ)
+  (srvHost_, tkn@NtfTknRec {ntfTknId, token = t@(DeviceToken pp _), tknStatus}, ntf) <- atomically (readTBQueue pushQ)
   liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
   st <- asks store
   case ntf of
@@ -644,8 +651,14 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
     PNMessage {} -> checkActiveTkn tknStatus $ do
       stats <- asks serverStats
       liftIO $ updatePeriodStats (activeTokens stats) ntfTknId
-      liftIO (deliverNotification st pp tkn ntf)
-        >>= incNtfStatT t . (\case Left _ -> ntfFailed; Right () -> ntfDelivered)
+      liftIO (deliverNotification st pp tkn ntf) >>= \case
+        Left _ -> do
+          incNtfStatT t ntfFailed
+          liftIO $ mapM_ (`incServerStat` ntfFailedOwn stats) srvHost_
+        Right () -> do
+          incNtfStatT t ntfDelivered
+          liftIO $ mapM_ (`incServerStat` ntfDeliveredOwn stats) srvHost_
+
   where
     checkActiveTkn :: NtfTknStatus -> M () -> M ()
     checkActiveTkn status action
@@ -686,7 +699,7 @@ periodicNtfsThread NtfPushServer {pushQ} = do
   liftIO $ forever $ do
     threadDelay interval
     now <- systemSeconds <$> getSystemTime
-    cnt <- withPeriodicNtfTokens st now $ \tkn -> atomically $ writeTBQueue pushQ (tkn, PNCheckMessages)
+    cnt <- withPeriodicNtfTokens st now $ \tkn -> atomically $ writeTBQueue pushQ (Nothing, tkn, PNCheckMessages)
     logNote $ "Scheduled periodic notifications: " <> tshow cnt
 
 runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M ()
@@ -794,7 +807,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} NtfPushServ
         ts <- liftIO $ getSystemDate
         let tkn = mkNtfTknRec tknId newTkn srvDhPrivKey dhSecret regCode ts
         withNtfStore (`addNtfToken` tkn) $ \_ -> do
-          atomically $ writeTBQueue pushQ (tkn, PNVerification regCode)
+          atomically $ writeTBQueue pushQ (Nothing, tkn, PNVerification regCode)
           incNtfStatT token ntfVrfQueued
           incNtfStatT token tknCreated
           pure $ NRTknId tknId srvDhPubKey
@@ -810,7 +823,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} NtfPushServ
               | otherwise -> withNtfStore (\st -> updateTknStatus st tkn NTRegistered) $ \_ -> sendVerification
             where
               sendVerification = do
-                atomically $ writeTBQueue pushQ (tkn, PNVerification tknRegCode)
+                atomically $ writeTBQueue pushQ (Nothing, tkn, PNVerification tknRegCode)
                 incNtfStatT token ntfVrfQueued
                 pure $ NRTknId ntfTknId $ C.publicKey tknDhPrivKey
           TVFY code -- this allows repeated verification for cases when client connection dropped before server response
@@ -828,7 +841,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} NtfPushServ
             regCode <- getRegCode
             let tkn' = tkn {token = token', tknStatus = NTRegistered, tknRegCode = regCode}
             withNtfStore (`replaceNtfToken` tkn') $ \_ -> do
-              atomically $ writeTBQueue pushQ (tkn', PNVerification regCode)
+              atomically $ writeTBQueue pushQ (Nothing, tkn', PNVerification regCode)
               incNtfStatT token ntfVrfQueued
               incNtfStatT token tknReplaced
               pure NROk
diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs
@@ -146,7 +146,7 @@ data SMPSubscriber = SMPSubscriber
   }
 
 data NtfPushServer = NtfPushServer
-  { pushQ :: TBQueue (NtfTknRec, PushNotification),
+  { pushQ :: TBQueue (Maybe T.Text, NtfTknRec, PushNotification), -- Maybe Text is a hostname of "own" server
     pushClients :: TMap PushProvider PushProviderClient,
     apnsConfig :: APNSPushClientConfig
   }
diff --git a/src/Simplex/Messaging/Notifications/Server/Prometheus.hs b/src/Simplex/Messaging/Notifications/Server/Prometheus.hs
@@ -76,8 +76,13 @@ ntfPrometheusMetrics sm rtm ts =
         _subCreated,
         _subDeleted,
         _ntfReceived,
+        _ntfReceivedAuth,
         _ntfDelivered,
         _ntfFailed,
+        _ntfReceivedOwn,
+        _ntfReceivedAuthOwn,
+        _ntfDeliveredOwn,
+        _ntfFailedOwn,
         _ntfCronDelivered,
         _ntfCronFailed,
         _ntfVrfQueued,
@@ -165,6 +170,10 @@ ntfPrometheusMetrics sm rtm ts =
       \# TYPE simplex_ntf_notifications_received counter\n\
       \simplex_ntf_notifications_received " <> mshow _ntfReceived <> "\n# ntfReceived\n\
       \\n\
+      \# HELP simplex_ntf_notifications_received_auth Received notifications without token or subscription (AUTH error)\n\
+      \# TYPE simplex_ntf_notifications_received_auth counter\n\
+      \simplex_ntf_notifications_received_auth " <> mshow _ntfReceivedAuth <> "\n# ntfReceivedAuth\n\
+      \\n\
       \# HELP simplex_ntf_notifications_delivered Delivered notifications\n\
       \# TYPE simplex_ntf_notifications_delivered counter\n\
       \simplex_ntf_notifications_delivered " <> mshow _ntfDelivered <> "\n# ntfDelivered\n\
@@ -201,6 +210,10 @@ ntfPrometheusMetrics sm rtm ts =
       \# TYPE simplex_ntf_notifications_total gauge\n\
       \simplex_ntf_notifications_total " <> mshow lastNtfCount <> "\n# lastNtfCount\n\
       \\n"
+      <> showNtfsByServer _ntfReceivedOwn "simplex_ntf_notifications_received_own" "Received notifications" "ntfReceivedOwn"
+      <> showNtfsByServer _ntfReceivedAuthOwn "simplex_ntf_notifications_received_auth_own" "Received notifications without token or subscription (AUTH error)" "ntfReceivedAuthOwn"
+      <> showNtfsByServer _ntfDeliveredOwn "simplex_ntf_notifications_delivered_own" "Delivered notifications" "ntfDeliveredOwn"
+      <> showNtfsByServer _ntfFailedOwn "simplex_ntf_notifications_failed_own" "Failed notifications" "ntfFailedOwn"
     info =
       "# Info\n\
       \# ----\n\
@@ -228,28 +241,41 @@ ntfPrometheusMetrics sm rtm ts =
       showOwnSrvSubs <> showOtherSrvSubs
       where
         showOwnSrvSubs
-          | M.null ownSrvSubs = showOwn_ "" 0 0
-          | otherwise = T.concat $ map (\(host, cnt) -> showOwn_ (metricHost host) 1 cnt) $ M.assocs ownSrvSubs
-        showOwn_ param srvCnt subCnt =
-          gaugeMetric (mPfx <> "server_count_own") param srvCnt (descrPfx <> " SMP subscriptions, own server count") "ownSrvSubs server"
-            <> gaugeMetric (mPfx <> "sub_count_own") param subCnt (descrPfx <> " SMP subscriptions count for own servers") "ownSrvSubs count"
+          | M.null ownSrvSubs = ""
+          | otherwise =
+              gaugeMetrics (mPfx <> "server_count_own") srvMetrics (descrPfx <> " SMP subscriptions, own server count") "ownSrvSubs server"
+                <> gaugeMetrics (mPfx <> "sub_count_own") subMetrics (descrPfx <> " SMP subscriptions count for own servers") "ownSrvSubs count"
+          where
+            subs = M.assocs ownSrvSubs
+            srvMetrics = map (\(host, _) -> (metricHost host, 1)) subs
+            subMetrics = map (\(host, cnt) -> (metricHost host, cnt)) subs
         showOtherSrvSubs =
-          gaugeMetric (mPfx <> "server_count_other") "" otherServers (descrPfx <> " SMP subscriptions, other server count") "otherServers"
-            <> gaugeMetric (mPfx <> "sub_count_other") "" otherSrvSubCount (descrPfx <> " SMP subscriptions count for other servers") "otherSrvSubCount"
+          gaugeMetrics (mPfx <> "server_count_other") [("", otherServers)] (descrPfx <> " SMP subscriptions, other server count") "otherServers"
+            <> gaugeMetrics (mPfx <> "sub_count_other") [("", otherSrvSubCount)] (descrPfx <> " SMP subscriptions count for other servers") "otherSrvSubCount"
+    showNtfsByServer (StatsByServerData srvNtfs) mName descr varName
+      | null srvNtfs = ""
+      | otherwise =
+          "# HELP " <> mName <> " " <> descr <> "\n\
+          \# TYPE " <> mName <> " counter\n"
+          <> showNtfMetrics
+          <> "# " <> varName <> "\n\n"
+      where
+        showNtfMetrics = T.concat $ map (\(host, value) -> mName <> metricHost host <> " " <> mshow value <> "\n") srvNtfs
     showWorkerMetric NtfSMPWorkerMetrics {ownServers, otherServers} mPfx descrPfx =
       showOwnServers <> showOtherServers
       where
         showOwnServers
-          | null ownServers = showOwn_ "" 0
-          | otherwise = T.concat $ map (\host -> showOwn_ (metricHost host) 1) ownServers
-        showOwn_ param cnt = gaugeMetric (mPfx <> "count_own") param cnt (descrPfx <> " count for own servers") "ownServers"
-        showOtherServers = gaugeMetric (mPfx <> "count_other") "" otherServers (descrPfx <> " count for other servers") "otherServers"
-    gaugeMetric :: Text -> Text -> Int -> Text -> Text -> Text
-    gaugeMetric name param value descr codeRef =
+          | null ownServers = ""
+          | otherwise = gaugeMetrics (mPfx <> "count_own") subMetrics (descrPfx <> " count for own servers") "ownServers"
+          where
+            subMetrics = map (\host -> (metricHost host, 1)) ownServers
+        showOtherServers = gaugeMetrics (mPfx <> "count_other") [("", otherServers)] (descrPfx <> " count for other servers") "otherServers"
+    gaugeMetrics :: Text -> [(Text, Int)] -> Text -> Text -> Text
+    gaugeMetrics name subMetrics descr codeRef =
       "# HELP " <> name <> " " <> descr <> "\n\
-      \# TYPE " <> name <> " gauge\n\
-      \" <> name <> param <> " " <> mshow value <> "\n# " <> codeRef <> "\n\
-      \\n"
+      \# TYPE " <> name <> " gauge\n"
+      <> T.concat (map (\(param, value) -> name <> param <> " " <> mshow value <> "\n") subMetrics)
+      <> "# " <> codeRef <> "\n\n"
     metricHost host = "{server=\"" <> host <> "\"}"
     mstr a = a <> " " <> tsEpoch
     mshow :: Show a => a -> Text
diff --git a/src/Simplex/Messaging/Notifications/Server/Stats.hs b/src/Simplex/Messaging/Notifications/Server/Stats.hs

Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,6 @@ import Simplex.Messaging.Protocol`
`74`	`74`	`subscriberParty,`
`75`	`75`	`subscriberServiceRole`
`76`	`76`	`)`
`77`		`-import qualified Simplex.Messaging.Protocol as SMP`
`78`	`77`	`import Simplex.Messaging.Session`
`79`	`78`	`import Simplex.Messaging.TMap (TMap)`
`80`	`79`	`import qualified Simplex.Messaging.TMap as TM`
Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ data SMPSubscriber = SMPSubscriber`
`146`	`146`	`}`
`147`	`147`
`148`	`148`	`data NtfPushServer = NtfPushServer`
`149`		`- { pushQ :: TBQueue (NtfTknRec, PushNotification),`
	`149`	`+ { pushQ :: TBQueue (Maybe T.Text, NtfTknRec, PushNotification), -- Maybe Text is a hostname of "own" server`
`150`	`150`	`pushClients :: TMap PushProvider PushProviderClient,`
`151`	`151`	`apnsConfig :: APNSPushClientConfig`
`152`	`152`	`}`