Skip to content

Commit eda3373

Browse files
epoberezkinepoberezkin
committed
histogram for confirmed delivery times
1 parent 5aca0f3 commit eda3373

3 files changed

Lines changed: 87 additions & 43 deletions

File tree

‎src/Simplex/Messaging/Server.hs‎

Lines changed: 13 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -695,33 +695,18 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
695695
subClientsCount <- IS.size <$> readTVarIO subClients
696696
subServicesCount <- M.size <$> getSubscribedClients serviceSubscribers
697697
pure RTSubscriberMetrics {subsCount, subClientsCount, subServicesCount}
698-
getDeliveredMetrics (RoundedSystemTime ts') = foldM countClnt (RTSubscriberMetrics 0 0 0, TimeAggregations 0 0 IM.empty) =<< getServerClients srv
698+
getDeliveredMetrics ts' = foldM countClnt (RTSubscriberMetrics 0 0 0, TimeBuckets 0 0 IM.empty) =<< getServerClients srv
699699
where
700700
countClnt acc@(metrics, times) Client {subscriptions} = do
701701
(cnt, times') <- foldM countSubs (0, times) =<< readTVarIO subscriptions
702702
pure $ if cnt > 0
703703
then (metrics {subsCount = subsCount metrics + cnt, subClientsCount = subClientsCount metrics + 1}, times')
704704
else acc
705-
countSubs acc@(!cnt, TimeAggregations {sumTime, maxTime, timeBuckets}) Sub {delivered} = do
705+
countSubs acc@(!cnt, times) Sub {delivered} = do
706706
delivered_ <- atomically $ tryReadTMVar delivered
707707
pure $ case delivered_ of
708708
Nothing -> acc
709-
Just (_, RoundedSystemTime ts) ->
710-
let t = ts' - ts
711-
seconds
712-
| t <= 5 = fromIntegral t
713-
| t <= 30 = t `toBucket` 5
714-
| t <= 60 = t `toBucket` 10
715-
| t <= 180 = t `toBucket` 30
716-
| otherwise = t `toBucket` 60
717-
toBucket n m = - fromIntegral (((- n) `div` m) * m) -- round up
718-
times' =
719-
TimeAggregations
720-
{ sumTime = sumTime + t,
721-
maxTime = max maxTime t,
722-
timeBuckets = IM.alter (Just . maybe 1 (+ 1)) seconds timeBuckets
723-
}
724-
in (cnt + 1, times')
709+
Just (_, ts) -> (cnt + 1, updateTimeBuckets ts ts' times)
725710

726711
runClient :: Transport c => X.CertificateChain -> C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M s ()
727712
runClient srvCert srvSignKey tp h = do
@@ -1755,28 +1740,28 @@ client
17551740
Nothing -> pure $ err NO_MSG
17561741
Just sub ->
17571742
atomically (getDelivered sub) >>= \case
1758-
Just st -> do
1743+
Just (st, ts) -> do
17591744
stats <- asks serverStats
17601745
fmap (either err id) $ liftIO $ runExceptT $ do
17611746
case st of
17621747
ProhibitSub -> do
17631748
deletedMsg_ <- tryDelMsg ms q msgId
1764-
liftIO $ mapM_ (updateStats stats True) deletedMsg_
1749+
liftIO $ mapM_ (updateStats stats True ts) deletedMsg_
17651750
pure ok
17661751
_ -> do
17671752
(deletedMsg_, msg_) <- tryDelPeekMsg ms q msgId
1768-
liftIO $ mapM_ (updateStats stats False) deletedMsg_
1753+
liftIO $ mapM_ (updateStats stats False ts) deletedMsg_
17691754
liftIO $ deliverMessage "ACK" qr entId sub msg_
17701755
_ -> pure $ err NO_MSG
17711756
where
1772-
getDelivered :: Sub -> STM (Maybe ServerSub)
1757+
getDelivered :: Sub -> STM (Maybe (ServerSub, RoundedSystemTime))
17731758
getDelivered Sub {delivered, subThread} = do
1774-
tryTakeTMVar delivered $>>= \v@(msgId', _) ->
1759+
tryTakeTMVar delivered $>>= \v@(msgId', ts) ->
17751760
if msgId == msgId' || B.null msgId
1776-
then pure $ Just subThread
1761+
then pure $ Just (subThread, ts)
17771762
else putTMVar delivered v $> Nothing
1778-
updateStats :: ServerStats -> Bool -> Message -> IO ()
1779-
updateStats stats isGet = \case
1763+
updateStats :: ServerStats -> Bool -> RoundedSystemTime -> Message -> IO ()
1764+
updateStats stats isGet deliveryTime = \case
17801765
MessageQuota {} -> pure ()
17811766
Message {msgFlags} -> do
17821767
incStat $ msgRecv stats
@@ -1793,6 +1778,8 @@ client
17931778
when (notification msgFlags) $ do
17941779
incStat $ msgRecvNtf stats
17951780
updatePeriodStats (activeQueuesNtf stats) entId
1781+
currTime <- getSystemSeconds
1782+
atomicModifyIORef'_ (msgRecvAckTimes stats) $ updateTimeBuckets deliveryTime currTime
17961783

17971784
sendMessage :: MsgFlags -> MsgBody -> StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
17981785
sendMessage msgFlags msgBody q qr

‎src/Simplex/Messaging/Server/Prometheus.hs‎

Lines changed: 26 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
module Simplex.Messaging.Server.Prometheus where
77

88
import Data.Int (Int64)
9-
import qualified Data.IntMap as IM
9+
import qualified Data.IntMap.Strict as IM
1010
import Data.List (mapAccumL)
1111
import Data.Text (Text)
1212
import qualified Data.Text as T
@@ -37,18 +37,12 @@ data RealTimeMetrics = RealTimeMetrics
3737
threadsCount :: Int,
3838
clientsCount :: Int,
3939
deliveredSubs :: RTSubscriberMetrics,
40-
deliveredTimes :: TimeAggregations,
40+
deliveredTimes :: TimeBuckets,
4141
smpSubs :: RTSubscriberMetrics,
4242
ntfSubs :: RTSubscriberMetrics,
4343
loadedCounts :: LoadedQueueCounts
4444
}
4545

46-
data TimeAggregations = TimeAggregations
47-
{ sumTime :: Int64,
48-
maxTime :: Int64,
49-
timeBuckets :: IM.IntMap Int
50-
}
51-
5246
data RTSubscriberMetrics = RTSubscriberMetrics
5347
{ subsCount :: Int,
5448
subClientsCount :: Int,
@@ -100,6 +94,7 @@ prometheusMetrics sm rtm ts =
10094
_msgSentLarge,
10195
_msgSentBlock,
10296
_msgRecv,
97+
_msgRecvAckTimes,
10398
_msgRecvGet,
10499
_msgGet,
105100
_msgGetNoMsg,
@@ -446,12 +441,26 @@ prometheusMetrics sm rtm ts =
446441
\# TYPE simplex_smp_delivered_clients_total gauge\n\
447442
\simplex_smp_delivered_clients_total " <> mshow (subClientsCount deliveredSubs) <> "\n# delivered.subClientsCount\n\
448443
\\n\
449-
\# HELP simplex_smp_delivery_ack_time Times to confirm message delivery\n\
444+
\# HELP simplex_smp_delivery_ack_time Times to confirm message delivery, including pending confirmation\n\
450445
\# TYPE simplex_smp_delivery_ack_time histogram\n\
451-
\simplex_smp_delivery_ack_time_sum " <> mshow (sumTime deliveredTimes) <> "\n# delivered.sumTime\n\
452-
\simplex_smp_delivery_ack_time_count " <> mshow (subsCount deliveredSubs) <> "\n# delivered.subsCount\n"
453-
<> showTimeBuckets (timeBuckets deliveredTimes)
454-
<> showTimeBucket "+Inf" (subsCount deliveredSubs)
446+
\simplex_smp_delivery_ack_time_sum " <> mshow (sumTime _msgRecvAckTimes + sumTime deliveredTimes) <> "\n# delivered.sumTime\n\
447+
\simplex_smp_delivery_ack_time_count " <> mshow (_msgRecv + _msgRecvGet + subsCount deliveredSubs) <> "\n# delivered.subsCount\n"
448+
<> showTimeBuckets "simplex_smp_delivery_ack_time" (IM.unionWith (+) (timeBuckets _msgRecvAckTimes) (timeBuckets deliveredTimes))
449+
<> showTimeBucket "simplex_smp_delivery_ack_time" "+Inf" (_msgRecv + _msgRecvGet + subsCount deliveredSubs)
450+
<> "\n\
451+
\# HELP simplex_smp_delivery_ack_confirmed_time Times to confirm message delivery, only confirmed deliveries\n\
452+
\# TYPE simplex_smp_delivery_ack_confirmed_time histogram\n\
453+
\simplex_smp_delivery_ack_confirmed_time_sum " <> mshow (sumTime _msgRecvAckTimes) <> "\n# delivered.sumTime\n\
454+
\simplex_smp_delivery_ack_confirmed_time_count " <> mshow (_msgRecv + _msgRecvGet) <> "\n# delivered.subsCount\n"
455+
<> showTimeBuckets "simplex_smp_delivery_ack_confirmed_time" (timeBuckets _msgRecvAckTimes)
456+
<> showTimeBucket "simplex_smp_delivery_ack_confirmed_time" "+Inf" (_msgRecv + _msgRecvGet)
457+
<> "\n\
458+
\# HELP simplex_smp_delivery_ack_pending_time Times to confirm message delivery, only pending confirmations\n\
459+
\# TYPE simplex_smp_delivery_ack_pending_time histogram\n\
460+
\simplex_smp_delivery_ack_pending_time_sum " <> mshow (sumTime deliveredTimes) <> "\n# delivered.sumTime\n\
461+
\simplex_smp_delivery_ack_pending_time_count " <> mshow (subsCount deliveredSubs) <> "\n# delivered.subsCount\n"
462+
<> showTimeBuckets "simplex_smp_delivery_ack_pending_time" (timeBuckets deliveredTimes)
463+
<> showTimeBucket "simplex_smp_delivery_ack_pending_time" "+Inf" (subsCount deliveredSubs)
455464
<> "\n\
456465
\# HELP simplex_smp_delivery_ack_time_max Max time to confirm message delivery\n\
457466
\# TYPE simplex_smp_delivery_ack_time_max gauge\n\
@@ -501,10 +510,10 @@ prometheusMetrics sm rtm ts =
501510
\# TYPE simplex_smp_loaded_queues_ntf_lock_count gauge\n\
502511
\simplex_smp_loaded_queues_ntf_lock_count " <> mshow (notifierLockCount loadedCounts) <> "\n# loadedCounts.notifierLockCount\n"
503512

504-
showTimeBuckets :: IM.IntMap Int -> Text
505-
showTimeBuckets = T.concat . snd . mapAccumL (\total (sec, cnt) -> (total + cnt, showTimeBucket (tshow sec) (total + cnt))) 0 . IM.assocs
506-
showTimeBucket :: Text -> Int -> Text
507-
showTimeBucket sec count = "simplex_smp_delivery_ack_time_bucket{le=\"" <> sec <> "\"} " <> mshow count <> "\n# delivered.timeBuckets\n"
513+
showTimeBuckets :: Text -> IM.IntMap Int -> Text
514+
showTimeBuckets metric = T.concat . snd . mapAccumL (\total (sec, cnt) -> (total + cnt, showTimeBucket metric (tshow sec) (total + cnt))) 0 . IM.assocs
515+
showTimeBucket :: Text -> Text -> Int -> Text
516+
showTimeBucket metric sec count = metric <> "_bucket{le=\"" <> sec <> "\"} " <> mshow count <> "\n# delivered.timeBuckets\n"
508517
socketsMetric :: (SocketStats -> Int) -> Text -> Text -> Text
509518
socketsMetric sel metric descr =
510519
"# HELP " <> metric <> " " <> descr <> "\n"

‎src/Simplex/Messaging/Server/Stats.hs‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ import Data.ByteString.Char8 (ByteString)
1414
import qualified Data.ByteString.Char8 as B
1515
import Data.Hashable (hash)
1616
import Data.IORef
17+
import Data.Int (Int64)
18+
import qualified Data.IntMap.Strict as IM
1719
import Data.IntSet (IntSet)
1820
import qualified Data.IntSet as IS
1921
import Data.Set (Set)
@@ -25,6 +27,7 @@ import Data.Time.Clock (UTCTime (..))
2527
import GHC.IORef (atomicSwapIORef)
2628
import Simplex.Messaging.Encoding.String
2729
import Simplex.Messaging.Protocol (EntityId (..))
30+
import Simplex.Messaging.Server.QueueStore (RoundedSystemTime (..))
2831
import Simplex.Messaging.Util (atomicModifyIORef'_, tshow, unlessM)
2932

3033
data ServerStats = ServerStats
@@ -57,6 +60,7 @@ data ServerStats = ServerStats
5760
msgSentLarge :: IORef Int,
5861
msgSentBlock :: IORef Int,
5962
msgRecv :: IORef Int,
63+
msgRecvAckTimes :: IORef TimeBuckets,
6064
msgRecvGet :: IORef Int,
6165
msgGet :: IORef Int,
6266
msgGetNoMsg :: IORef Int,
@@ -115,6 +119,7 @@ data ServerStatsData = ServerStatsData
115119
_msgSentLarge :: Int,
116120
_msgSentBlock :: Int,
117121
_msgRecv :: Int,
122+
_msgRecvAckTimes :: TimeBuckets,
118123
_msgRecvGet :: Int,
119124
_msgGet :: Int,
120125
_msgGetNoMsg :: Int,
@@ -174,6 +179,7 @@ newServerStats ts = do
174179
msgSentLarge <- newIORef 0
175180
msgSentBlock <- newIORef 0
176181
msgRecv <- newIORef 0
182+
msgRecvAckTimes <- newIORef $ TimeBuckets 0 0 IM.empty
177183
msgRecvGet <- newIORef 0
178184
msgGet <- newIORef 0
179185
msgGetNoMsg <- newIORef 0
@@ -230,6 +236,7 @@ newServerStats ts = do
230236
msgSentLarge,
231237
msgSentBlock,
232238
msgRecv,
239+
msgRecvAckTimes,
233240
msgRecvGet,
234241
msgGet,
235242
msgGetNoMsg,
@@ -288,6 +295,7 @@ getServerStatsData s = do
288295
_msgSentLarge <- readIORef $ msgSentLarge s
289296
_msgSentBlock <- readIORef $ msgSentBlock s
290297
_msgRecv <- readIORef $ msgRecv s
298+
_msgRecvAckTimes <- readIORef $ msgRecvAckTimes s
291299
_msgRecvGet <- readIORef $ msgRecvGet s
292300
_msgGet <- readIORef $ msgGet s
293301
_msgGetNoMsg <- readIORef $ msgGetNoMsg s
@@ -344,6 +352,7 @@ getServerStatsData s = do
344352
_msgSentLarge,
345353
_msgSentBlock,
346354
_msgRecv,
355+
_msgRecvAckTimes,
347356
_msgRecvGet,
348357
_msgGet,
349358
_msgGetNoMsg,
@@ -403,6 +412,7 @@ setServerStats s d = do
403412
writeIORef (msgSentLarge s) $! _msgSentLarge d
404413
writeIORef (msgSentBlock s) $! _msgSentBlock d
405414
writeIORef (msgRecv s) $! _msgRecv d
415+
writeIORef (msgRecvAckTimes s) $! _msgRecvAckTimes d
406416
writeIORef (msgRecvGet s) $! _msgRecvGet d
407417
writeIORef (msgGet s) $! _msgGet d
408418
writeIORef (msgGetNoMsg s) $! _msgGetNoMsg d
@@ -462,6 +472,7 @@ instance StrEncoding ServerStatsData where
462472
"msgSentLarge=" <> strEncode (_msgSentLarge d),
463473
"msgSentBlock=" <> strEncode (_msgSentBlock d),
464474
"msgRecv=" <> strEncode (_msgRecv d),
475+
"msgRecvAckTimes=" <> strEncode (_msgRecvAckTimes d),
465476
"msgRecvGet=" <> strEncode (_msgRecvGet d),
466477
"msgGet=" <> strEncode (_msgGet d),
467478
"msgGetNoMsg=" <> strEncode (_msgGetNoMsg d),
@@ -525,6 +536,7 @@ instance StrEncoding ServerStatsData where
525536
_msgSentLarge <- opt "msgSentLarge="
526537
_msgSentBlock <- opt "msgSentBlock="
527538
_msgRecv <- "msgRecv=" *> strP <* A.endOfLine
539+
_msgRecvAckTimes <- "msgRecvAckTimes=" *> strP <* A.endOfLine <|> pure (TimeBuckets 0 0 IM.empty)
528540
_msgRecvGet <- opt "msgRecvGet="
529541
_msgGet <- opt "msgGet="
530542
_msgGetNoMsg <- opt "msgGetNoMsg="
@@ -592,6 +604,7 @@ instance StrEncoding ServerStatsData where
592604
_msgSentLarge,
593605
_msgSentBlock,
594606
_msgRecv,
607+
_msgRecvAckTimes,
595608
_msgRecvGet,
596609
_msgGet,
597610
_msgGetNoMsg,
@@ -944,3 +957,38 @@ instance StrEncoding ServiceStatsData where
944957
_srvSubQueues,
945958
_srvSubEnd
946959
}
960+
961+
data TimeBuckets = TimeBuckets
962+
{ sumTime :: Int64,
963+
maxTime :: Int64,
964+
timeBuckets :: IM.IntMap Int
965+
}
966+
deriving (Show)
967+
968+
updateTimeBuckets :: RoundedSystemTime -> RoundedSystemTime -> TimeBuckets -> TimeBuckets
969+
updateTimeBuckets
970+
(RoundedSystemTime deliveryTime)
971+
(RoundedSystemTime currTime)
972+
TimeBuckets {sumTime, maxTime, timeBuckets} =
973+
TimeBuckets
974+
{ sumTime = sumTime + t,
975+
maxTime = max maxTime t,
976+
timeBuckets = IM.alter (Just . maybe 1 (+ 1)) seconds timeBuckets
977+
}
978+
where
979+
t = currTime - deliveryTime
980+
seconds
981+
| t <= 5 = fromIntegral t
982+
| t <= 30 = t `toBucket` 5
983+
| t <= 60 = t `toBucket` 10
984+
| t <= 180 = t `toBucket` 30
985+
| otherwise = t `toBucket` 60
986+
toBucket n m = - fromIntegral (((- n) `div` m) * m) -- round up
987+
988+
instance StrEncoding TimeBuckets where
989+
strEncode TimeBuckets {sumTime, maxTime, timeBuckets} =
990+
strEncode (sumTime, maxTime) <> " " <> strEncodeList (IM.toList timeBuckets)
991+
strP = do
992+
(sumTime, maxTime) <- strP_
993+
tbs <- strListP
994+
pure TimeBuckets {sumTime, maxTime, timeBuckets = IM.fromList tbs}

0 commit comments

Comments
 (0)