mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-05-27 22:54:38 +00:00
smp server: prometheus histogram for message confirmation times (ACK) (#1575)
* time buckets * split max time metric * histogram * histogram for confirmed delivery times * gaugehistogram * fix created, _ in gauge_histogram * remove comments * fix metrics
This commit is contained in:
Evgeny
@@ -684,24 +684,29 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
|
||||
let threadsCount = 0
|
||||
#endif
|
||||
clientsCount <- IM.size <$> getServerClients srv
|
||||
deliveredSubs <- getDeliveredMetrics
|
||||
(deliveredSubs, deliveredTimes) <- getDeliveredMetrics =<< getSystemSeconds
|
||||
smpSubs <- getSubscribersMetrics subscribers
|
||||
ntfSubs <- getSubscribersMetrics ntfSubscribers
|
||||
loadedCounts <- loadedQueueCounts $ fromMsgStore ms
|
||||
pure RealTimeMetrics {socketStats, threadsCount, clientsCount, deliveredSubs, smpSubs, ntfSubs, loadedCounts}
|
||||
pure RealTimeMetrics {socketStats, threadsCount, clientsCount, deliveredSubs, deliveredTimes, smpSubs, ntfSubs, loadedCounts}
|
||||
where
|
||||
getSubscribersMetrics ServerSubscribers {queueSubscribers, serviceSubscribers, subClients} = do
|
||||
subsCount <- M.size <$> getSubscribedClients queueSubscribers
|
||||
subClientsCount <- IS.size <$> readTVarIO subClients
|
||||
subServicesCount <- M.size <$> getSubscribedClients serviceSubscribers
|
||||
pure RTSubscriberMetrics {subsCount, subClientsCount, subServicesCount}
|
||||
getDeliveredMetrics = foldM countClnt (RTSubscriberMetrics 0 0 0) =<< getServerClients srv
|
||||
countClnt metrics Client {subscriptions} = do
|
||||
cnt <- foldM countSubs 0 =<< readTVarIO subscriptions
|
||||
pure $ if cnt > 0
|
||||
then metrics {subsCount = subsCount metrics + cnt, subClientsCount = subClientsCount metrics + 1}
|
||||
else metrics
|
||||
countSubs !cnt Sub {delivered} = (\empty -> if empty then cnt else cnt + 1) <$> atomically (isEmptyTMVar delivered)
|
||||
getDeliveredMetrics ts' = foldM countClnt (RTSubscriberMetrics 0 0 0, emptyTimeBuckets) =<< getServerClients srv
|
||||
where
|
||||
countClnt acc@(metrics, times) Client {subscriptions} = do
|
||||
(cnt, times') <- foldM countSubs (0, times) =<< readTVarIO subscriptions
|
||||
pure $ if cnt > 0
|
||||
then (metrics {subsCount = subsCount metrics + cnt, subClientsCount = subClientsCount metrics + 1}, times')
|
||||
else acc
|
||||
countSubs acc@(!cnt, times) Sub {delivered} = do
|
||||
delivered_ <- atomically $ tryReadTMVar delivered
|
||||
pure $ case delivered_ of
|
||||
Nothing -> acc
|
||||
Just (_, ts) -> (cnt + 1, updateTimeBuckets ts ts' times)
|
||||
|
||||
runClient :: Transport c => X.CertificateChain -> C.APrivateSignKey -> TProxy c 'TServer -> c 'TServer -> M s ()
|
||||
runClient srvCert srvSignKey tp h = do
|
||||
@@ -1633,7 +1638,7 @@ client
|
||||
-- This is tracked as "subscription" in the client to prevent these
|
||||
-- clients from being able to subscribe.
|
||||
pure s
|
||||
getMessage_ :: Sub -> Maybe MsgId -> M s (Transmission BrokerMsg)
|
||||
getMessage_ :: Sub -> Maybe (MsgId, RoundedSystemTime) -> M s (Transmission BrokerMsg)
|
||||
getMessage_ s delivered_ = do
|
||||
stats <- asks serverStats
|
||||
fmap (either err id) $ liftIO $ runExceptT $
|
||||
@@ -1641,7 +1646,8 @@ client
|
||||
Just msg -> do
|
||||
let encMsg = encryptMsg qr msg
|
||||
incStat $ (if isJust delivered_ then msgGetDuplicate else msgGet) stats
|
||||
atomically $ setDelivered s msg $> (corrId, entId, MSG encMsg)
|
||||
ts <- liftIO getSystemSeconds
|
||||
atomically $ setDelivered s msg ts $> (corrId, entId, MSG encMsg)
|
||||
Nothing -> incStat (msgGetNoMsg stats) $> ok
|
||||
|
||||
withQueue :: (StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)) -> M s (Transmission BrokerMsg)
|
||||
@@ -1734,28 +1740,28 @@ client
|
||||
Nothing -> pure $ err NO_MSG
|
||||
Just sub ->
|
||||
atomically (getDelivered sub) >>= \case
|
||||
Just st -> do
|
||||
Just (st, ts) -> do
|
||||
stats <- asks serverStats
|
||||
fmap (either err id) $ liftIO $ runExceptT $ do
|
||||
case st of
|
||||
ProhibitSub -> do
|
||||
deletedMsg_ <- tryDelMsg ms q msgId
|
||||
liftIO $ mapM_ (updateStats stats True) deletedMsg_
|
||||
liftIO $ mapM_ (updateStats stats True ts) deletedMsg_
|
||||
pure ok
|
||||
_ -> do
|
||||
(deletedMsg_, msg_) <- tryDelPeekMsg ms q msgId
|
||||
liftIO $ mapM_ (updateStats stats False) deletedMsg_
|
||||
liftIO $ mapM_ (updateStats stats False ts) deletedMsg_
|
||||
liftIO $ deliverMessage "ACK" qr entId sub msg_
|
||||
_ -> pure $ err NO_MSG
|
||||
where
|
||||
getDelivered :: Sub -> STM (Maybe ServerSub)
|
||||
getDelivered :: Sub -> STM (Maybe (ServerSub, RoundedSystemTime))
|
||||
getDelivered Sub {delivered, subThread} = do
|
||||
tryTakeTMVar delivered $>>= \msgId' ->
|
||||
tryTakeTMVar delivered $>>= \v@(msgId', ts) ->
|
||||
if msgId == msgId' || B.null msgId
|
||||
then pure $ Just subThread
|
||||
else putTMVar delivered msgId' $> Nothing
|
||||
updateStats :: ServerStats -> Bool -> Message -> IO ()
|
||||
updateStats stats isGet = \case
|
||||
then pure $ Just (subThread, ts)
|
||||
else putTMVar delivered v $> Nothing
|
||||
updateStats :: ServerStats -> Bool -> RoundedSystemTime -> Message -> IO ()
|
||||
updateStats stats isGet deliveryTime = \case
|
||||
MessageQuota {} -> pure ()
|
||||
Message {msgFlags} -> do
|
||||
incStat $ msgRecv stats
|
||||
@@ -1772,6 +1778,8 @@ client
|
||||
when (notification msgFlags) $ do
|
||||
incStat $ msgRecvNtf stats
|
||||
updatePeriodStats (activeQueuesNtf stats) entId
|
||||
currTime <- getSystemSeconds
|
||||
atomicModifyIORef'_ (msgRecvAckTimes stats) $ updateTimeBuckets deliveryTime currTime
|
||||
|
||||
sendMessage :: MsgFlags -> MsgBody -> StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
|
||||
sendMessage msgFlags msgBody q qr
|
||||
@@ -1839,33 +1847,35 @@ client
|
||||
-- the subscribed client var is read outside of STM to avoid transaction cost
|
||||
-- in case no client is subscribed.
|
||||
getSubscribedClient rId (queueSubscribers subscribers)
|
||||
$>>= atomically . deliverToSub
|
||||
$>>= deliverToSub
|
||||
>>= mapM_ forkDeliver
|
||||
where
|
||||
rId = recipientId q
|
||||
deliverToSub rcv =
|
||||
deliverToSub rcv = do
|
||||
ts <- getSystemSeconds
|
||||
atomically $
|
||||
-- reading client TVar in the same transaction,
|
||||
-- so that if subscription ends, it re-evalutates
|
||||
-- and delivery is cancelled -
|
||||
-- the new client will receive message in response to SUB.
|
||||
readTVar rcv
|
||||
$>>= \rc@Client {subscriptions = subs, sndQ = sndQ'} -> TM.lookup rId subs
|
||||
$>>= \s@Sub {subThread, delivered} -> case subThread of
|
||||
ProhibitSub -> pure Nothing
|
||||
ServerSub st -> readTVar st >>= \case
|
||||
NoSub ->
|
||||
tryTakeTMVar delivered >>= \case
|
||||
Just _ -> pure Nothing -- if a message was already delivered, should not deliver more
|
||||
Nothing ->
|
||||
ifM
|
||||
(isFullTBQueue sndQ')
|
||||
(writeTVar st SubPending $> Just (rc, s, st))
|
||||
(deliver sndQ' s $> Nothing)
|
||||
_ -> pure Nothing
|
||||
deliver sndQ' s = do
|
||||
readTVar rcv
|
||||
$>>= \rc@Client {subscriptions = subs, sndQ = sndQ'} -> TM.lookup rId subs
|
||||
$>>= \s@Sub {subThread, delivered} -> case subThread of
|
||||
ProhibitSub -> pure Nothing
|
||||
ServerSub st -> readTVar st >>= \case
|
||||
NoSub ->
|
||||
tryTakeTMVar delivered >>= \case
|
||||
Just _ -> pure Nothing -- if a message was already delivered, should not deliver more
|
||||
Nothing ->
|
||||
ifM
|
||||
(isFullTBQueue sndQ')
|
||||
(writeTVar st SubPending $> Just (rc, s, st))
|
||||
(deliver sndQ' s ts $> Nothing)
|
||||
_ -> pure Nothing
|
||||
deliver sndQ' s ts = do
|
||||
let encMsg = encryptMsg qr msg
|
||||
writeTBQueue sndQ' [(CorrId "", rId, MSG encMsg)]
|
||||
void $ setDelivered s msg
|
||||
void $ setDelivered s msg ts
|
||||
forkDeliver (rc@Client {sndQ = sndQ'}, s@Sub {delivered}, st) = do
|
||||
t <- mkWeakThreadId =<< forkIO deliverThread
|
||||
atomically $ modifyTVar' st $ \case
|
||||
@@ -1878,13 +1888,14 @@ client
|
||||
-- lookup can be outside of STM transaction,
|
||||
-- as long as the check that it is the same client is inside.
|
||||
getSubscribedClient rId (queueSubscribers subscribers) >>= mapM_ deliverIfSame
|
||||
deliverIfSame rcv = time "deliver" . atomically $
|
||||
whenM (sameClient rc rcv) $
|
||||
deliverIfSame rcv = time "deliver" $ do
|
||||
ts <- getSystemSeconds
|
||||
atomically $ whenM (sameClient rc rcv) $
|
||||
tryTakeTMVar delivered >>= \case
|
||||
Just _ -> pure () -- if a message was already delivered, should not deliver more
|
||||
Nothing -> do
|
||||
-- a separate thread is needed because it blocks when client sndQ is full.
|
||||
deliver sndQ' s
|
||||
deliver sndQ' s ts
|
||||
writeTVar st NoSub
|
||||
|
||||
enqueueNotification :: NtfCreds -> Message -> M s ()
|
||||
@@ -1958,13 +1969,14 @@ client
|
||||
VRFailed e -> Left (corrId', entId', ERR e)
|
||||
|
||||
deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> IO (Transmission BrokerMsg)
|
||||
deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") . atomically $
|
||||
deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") $
|
||||
case subThread of
|
||||
ProhibitSub -> pure resp
|
||||
_ -> case msg_ of
|
||||
Just msg ->
|
||||
Just msg -> do
|
||||
ts <- getSystemSeconds
|
||||
let encMsg = encryptMsg qr msg
|
||||
in setDelivered s msg $> (corrId, rId, MSG encMsg)
|
||||
atomically (setDelivered s msg ts) $> (corrId, rId, MSG encMsg)
|
||||
_ -> pure resp
|
||||
where
|
||||
resp = (corrId, rId, OK)
|
||||
@@ -1982,8 +1994,10 @@ client
|
||||
msgId' = messageId msg
|
||||
msgTs' = messageTs msg
|
||||
|
||||
setDelivered :: Sub -> Message -> STM Bool
|
||||
setDelivered s msg = tryPutTMVar (delivered s) $! messageId msg
|
||||
setDelivered :: Sub -> Message -> RoundedSystemTime -> STM Bool
|
||||
setDelivered Sub {delivered} msg !ts = do
|
||||
let !msgId = messageId msg
|
||||
tryPutTMVar delivered (msgId, ts)
|
||||
|
||||
delQueueAndMsgs :: (StoreQueue s, QueueRec) -> M s (Transmission BrokerMsg)
|
||||
delQueueAndMsgs (q, QueueRec {rcvServiceId}) = do
|
||||
@@ -2024,7 +2038,7 @@ client
|
||||
SubPending -> QSubPending
|
||||
SubThread _ -> QSubThread
|
||||
ProhibitSub -> pure QProhibitSub
|
||||
qDelivered <- atomically $ decodeLatin1 . encode <$$> tryReadTMVar delivered
|
||||
qDelivered <- atomically $ decodeLatin1 . encode . fst <$$> tryReadTMVar delivered
|
||||
pure QSub {qSubThread, qDelivered}
|
||||
|
||||
ok :: Transmission BrokerMsg
|
||||
|
||||
Reference in New Issue
Block a user