{-# LANGUAGE RecordWildCards #-}

-- | Provides Prometheus-based metrics server based on `Tracer` collection.
--
-- To add a new metric, one needs to:
--
--  * Add a field to 'Metrics' and register it in 'registerMetrics',
--  * Update the 'monitor' function to handle relevant 'HydraLog' entries and
--    update the underlying Prometheus metric. Nested helpers are provided to
--    increase a 'Counter' by one (@tick@) and to 'observe' some value in a
--    'Histogram'.
--
-- The metric handles are typed, so a metric can only be updated in the way it
-- was registered and a name can only be referred to if it was registered.
module Hydra.Logging.Monitoring (
  withMonitoring,
) where

import Hydra.Prelude

import Control.Concurrent.Class.MonadSTM (modifyTVar', readTVarIO, writeTVar)
import Control.Tracer (Tracer (Tracer))
import Data.Map.Strict as Map
import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
import GHC.Stats (RTSStats (..), getRTSStats, getRTSStatsEnabled)
import Hydra.API.ServerOutput (ClientMessage (RejectedInputBecauseBroadcastStalled))
import Hydra.HeadLogic (
  Input (NetworkInput),
 )
import Hydra.HeadLogic.Outcome (Effect (ClientEffect), Outcome (..), StateChanged (..))
import Hydra.Logging.Messages (HydraLog (..))
import Hydra.Network (PortNumber, StallReason (..))
import Hydra.Network.Message (Message (ReqTx), NetworkEvent (..))
import Hydra.Node (HydraNodeLog (..))
import Hydra.Tx (IsTx (TxIdType), Snapshot (..), SnapshotNumber, txId)
import System.Metrics.Prometheus.Concurrent.Registry (
  Registry,
  new,
  registerCounter,
  registerGauge,
  registerHistogram,
  sample,
 )
import System.Metrics.Prometheus.Http.Scrape (serveMetrics)
import System.Metrics.Prometheus.Metric.Counter (Counter, add, inc)
import System.Metrics.Prometheus.Metric.Gauge (Gauge)
import System.Metrics.Prometheus.Metric.Gauge qualified as Gauge
import System.Metrics.Prometheus.Metric.Histogram (Histogram, observe)
import System.Metrics.Prometheus.MetricId (Name (Name))

-- | Handles to all metrics hydra-node exposes.
--
-- NOTE: The 'Name's below are a public interface: they are scraped by
-- Prometheus and referenced by the Grafana dashboards in @demo/grafana@, so
-- renaming one is a breaking change for operators.
data Metrics = Metrics
  { Metrics -> Counter
headInputs :: Counter
  , Metrics -> Counter
headRequestedTx :: Counter
  , Metrics -> Counter
headConfirmedTx :: Counter
  , Metrics -> Histogram
txConfirmationTime :: Histogram
  , Metrics -> Histogram
snapshotConfirmationTime :: Histogram
  , Metrics -> Gauge
peersConnected :: Gauge
  , Metrics -> Gauge
chainDriftSeconds :: Gauge
  , Metrics -> Gauge
chainLastBlockTimestampSeconds :: Gauge
  , Metrics -> Gauge
broadcastStalled :: Gauge
  , Metrics -> Gauge
broadcastStalledNoProgress :: Gauge
  , Metrics -> Gauge
broadcastStalledBacklogFull :: Gauge
  , Metrics -> Gauge
pendingBroadcasts :: Gauge
  , Metrics -> Gauge
broadcastNoProgressSeconds :: Gauge
  , Metrics -> Counter
inputsRefusedBroadcastStalled :: Counter
  }

-- | Wraps a monadic action using a `Tracer` and capture metrics based on traces.
-- Given a `portNumber`, this wrapper starts a Prometheus-compliant server on this port.
-- This is a no-op if given `Nothing`. This function is not polymorphic over the type of
-- messages because it needs to understand them in order to provide meaningful metrics.
withMonitoring ::
  (MonadIO m, MonadAsync m, IsTx tx, MonadMonotonicTime m, MonadTime m, MonadLabelledSTM m) =>
  Maybe PortNumber ->
  Tracer m (HydraLog tx) ->
  (Tracer m (HydraLog tx) -> m ()) ->
  m ()
withMonitoring :: forall (m :: * -> *) tx.
(MonadIO m, MonadAsync m, IsTx tx, MonadMonotonicTime m,
 MonadTime m, MonadLabelledSTM m) =>
Maybe PortNumber
-> Tracer m (HydraLog tx)
-> (Tracer m (HydraLog tx) -> m ())
-> m ()
withMonitoring Maybe PortNumber
Nothing Tracer m (HydraLog tx)
tracer Tracer m (HydraLog tx) -> m ()
action = Tracer m (HydraLog tx) -> m ()
action Tracer m (HydraLog tx)
tracer
withMonitoring (Just PortNumber
monitoringPort) (Tracer HydraLog tx -> m ()
tracer) Tracer m (HydraLog tx) -> m ()
action = do
  (HydraLog tx -> m ()
traceMetric, Registry
registry) <- m (HydraLog tx -> m (), Registry)
forall (m :: * -> *) tx.
(MonadIO m, MonadMonotonicTime m, MonadTime m, IsTx tx,
 MonadLabelledSTM m) =>
m (HydraLog tx -> m (), Registry)
prepareRegistry
  IO ()
refreshRts <- IO (IO ()) -> m (IO ())
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (IO ()) -> m (IO ())) -> IO (IO ()) -> m (IO ())
forall a b. (a -> b) -> a -> b
$ Registry -> IO (IO ())
registerRtsMetrics Registry
registry
  (String, m ()) -> (Async m () -> m ()) -> m ()
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (Async m a -> m b) -> m b
withAsyncLabelled
    (String
"monitoring-serveMetrics", Port -> Path -> IO RegistrySample -> m ()
forall (m :: * -> *).
MonadIO m =>
Port -> Path -> IO RegistrySample -> m ()
serveMetrics (PortNumber -> Port
forall a b. (Integral a, Num b) => a -> b
fromIntegral PortNumber
monitoringPort) [Text
"metrics"] (IO ()
refreshRts IO () -> IO RegistrySample -> IO RegistrySample
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Registry -> IO RegistrySample
sample Registry
registry))
    ((Async m () -> m ()) -> m ()) -> (Async m () -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Async m ()
_ ->
      let wrappedTracer :: Tracer m (HydraLog tx)
wrappedTracer = (HydraLog tx -> m ()) -> Tracer m (HydraLog tx)
forall (m :: * -> *) a. (a -> m ()) -> Tracer m a
Tracer ((HydraLog tx -> m ()) -> Tracer m (HydraLog tx))
-> (HydraLog tx -> m ()) -> Tracer m (HydraLog tx)
forall a b. (a -> b) -> a -> b
$ \HydraLog tx
msg -> do
            HydraLog tx -> m ()
traceMetric HydraLog tx
msg
            HydraLog tx -> m ()
tracer HydraLog tx
msg
       in Tracer m (HydraLog tx) -> m ()
action Tracer m (HydraLog tx)
wrappedTracer

-- | Register GHC RTS work counters when the runtime collects them (process
-- started with '+RTS -T'); returns a refresh action run before each scrape.
-- Without -T nothing is registered, so the endpoint output is identical to
-- before these metrics existed.
registerRtsMetrics :: Registry -> IO (IO ())
registerRtsMetrics :: Registry -> IO (IO ())
registerRtsMetrics Registry
registry = do
  Bool
enabled <- IO Bool
getRTSStatsEnabled
  if Bool -> Bool
not Bool
enabled
    then IO () -> IO (IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    else do
      Gauge
allocated <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_allocated_bytes"
      Gauge
mutatorCpu <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_mutator_cpu_seconds"
      Gauge
gcCpu <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_gc_cpu_seconds"
      Gauge
maxLive <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_max_live_bytes"
      -- Sum of live bytes measured at every major GC; divided by the major GC
      -- count over the same window this gives the mean live heap of that
      -- window, which a since-start maximum cannot provide.
      Gauge
cumulativeLive <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_cumulative_live_bytes"
      Gauge
majorGcs <- Text -> IO Gauge
rtsGauge Text
"hydra_rts_major_gcs"
      IO () -> IO (IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (IO () -> IO (IO ())) -> IO () -> IO (IO ())
forall a b. (a -> b) -> a -> b
$ do
        RTSStats
stats <- IO RTSStats
getRTSStats
        Double -> Gauge -> IO ()
Gauge.set (Word64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word64
allocated_bytes RTSStats
stats)) Gauge
allocated
        Double -> Gauge -> IO ()
Gauge.set (RtsTime -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> RtsTime
mutator_cpu_ns RTSStats
stats) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
1.0e9) Gauge
mutatorCpu
        Double -> Gauge -> IO ()
Gauge.set (RtsTime -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> RtsTime
gc_cpu_ns RTSStats
stats) Double -> Double -> Double
forall a. Fractional a => a -> a -> a
/ Double
1.0e9) Gauge
gcCpu
        Double -> Gauge -> IO ()
Gauge.set (Word64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word64
max_live_bytes RTSStats
stats)) Gauge
maxLive
        Double -> Gauge -> IO ()
Gauge.set (Word64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word64
cumulative_live_bytes RTSStats
stats)) Gauge
cumulativeLive
        Double -> Gauge -> IO ()
Gauge.set (Word32 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (RTSStats -> Word32
major_gcs RTSStats
stats)) Gauge
majorGcs
 where
  rtsGauge :: Text -> IO Gauge
rtsGauge Text
name = Name -> Labels -> Registry -> IO Gauge
registerGauge (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty Registry
registry

-- | Register all relevant metrics.
-- Returns the `Registry` which is needed to `serveMetrics` or any other form of publication
-- of metrics, whether push or pull, and a function for updating metrics given some trace event.
prepareRegistry :: forall m tx. (MonadIO m, MonadMonotonicTime m, MonadTime m, IsTx tx, MonadLabelledSTM m) => m (HydraLog tx -> m (), Registry)
prepareRegistry :: forall (m :: * -> *) tx.
(MonadIO m, MonadMonotonicTime m, MonadTime m, IsTx tx,
 MonadLabelledSTM m) =>
m (HydraLog tx -> m (), Registry)
prepareRegistry = do
  TVar m (Map (TxIdType tx) Time)
transactionsMap <- String
-> Map (TxIdType tx) Time -> m (TVar m (Map (TxIdType tx) Time))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"monitoring-txs-map-registry" Map (TxIdType tx) Time
forall a. Monoid a => a
mempty
  TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap <- String
-> Map SnapshotNumber (Time, [TxIdType tx])
-> m (TVar m (Map SnapshotNumber (Time, [TxIdType tx])))
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> a -> m (TVar m a)
newLabelledTVarIO String
"monitoring-snapshots-map-registry" Map SnapshotNumber (Time, [TxIdType tx])
forall a. Monoid a => a
mempty
  Registry
registry <- IO Registry -> m Registry
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO Registry
new
  Metrics
metrics <- Registry -> m Metrics
forall (m :: * -> *). MonadIO m => Registry -> m Metrics
registerMetrics Registry
registry
  (HydraLog tx -> m (), Registry)
-> m (HydraLog tx -> m (), Registry)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (TVar m (Map (TxIdType tx) Time)
-> TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Metrics
-> HydraLog tx
-> m ()
forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, MonadTime m,
 IsTx tx) =>
TVar m (Map (TxIdType tx) Time)
-> TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Metrics
-> HydraLog tx
-> m ()
monitor TVar m (Map (TxIdType tx) Time)
transactionsMap TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap Metrics
metrics, Registry
registry)

registerMetrics :: MonadIO m => Registry -> m Metrics
registerMetrics :: forall (m :: * -> *). MonadIO m => Registry -> m Metrics
registerMetrics Registry
registry = IO Metrics -> m Metrics
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO Metrics -> m Metrics) -> IO Metrics -> m Metrics
forall a b. (a -> b) -> a -> b
$ do
  Counter
headInputs <- Text -> IO Counter
counter Text
"hydra_head_inputs"
  Counter
headRequestedTx <- Text -> IO Counter
counter Text
"hydra_head_requested_tx"
  Counter
headConfirmedTx <- Text -> IO Counter
counter Text
"hydra_head_confirmed_tx"
  Histogram
txConfirmationTime <-
    Text -> [Double] -> IO Histogram
histogram Text
"hydra_head_tx_confirmation_time_ms" [Double
5, Double
10, Double
50, Double
100, Double
1000]
  Histogram
snapshotConfirmationTime <-
    Text -> [Double] -> IO Histogram
histogram Text
"hydra_head_snapshot_confirmation_time_ms" [Double
5, Double
10, Double
50, Double
100, Double
500, Double
1000, Double
5000, Double
10000, Double
30000]
  Gauge
peersConnected <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_peers_connected"
  Gauge
chainDriftSeconds <- Text -> IO Gauge
gaugeMetric Text
"hydra_chain_drift_seconds"
  Gauge
chainLastBlockTimestampSeconds <- Text -> IO Gauge
gaugeMetric Text
"hydra_chain_last_block_timestamp_seconds"
  -- Diagnosing a node whose outbound messages are backing up
  -- (GHSA-3mmr-q43p-g6p2): 'broadcastStalled' is the condition clients are
  -- told about and the one to alert on; the two beside it split it by cause,
  -- because only 'NoProgress' means the network cannot be reached, while
  -- 'BacklogFull' also fires on a client outrunning a network that is keeping
  -- up. The numbers below let an operator see a backlog building before it
  -- trips, and tell a slow network from a dead one.
  -- 'inputsRefusedBroadcastStalled' is the client-visible cost.
  Gauge
broadcastStalled <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_broadcast_stalled"
  Gauge
broadcastStalledNoProgress <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_broadcast_stalled_no_progress"
  Gauge
broadcastStalledBacklogFull <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_broadcast_stalled_backlog_full"
  Gauge
pendingBroadcasts <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_pending_broadcasts"
  Gauge
broadcastNoProgressSeconds <- Text -> IO Gauge
gaugeMetric Text
"hydra_head_broadcast_no_progress_seconds"
  Counter
inputsRefusedBroadcastStalled <- Text -> IO Counter
counter Text
"hydra_head_inputs_refused_broadcast_stalled"
  Metrics -> IO Metrics
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Metrics{Counter
Gauge
Histogram
$sel:headInputs:Metrics :: Counter
$sel:headRequestedTx:Metrics :: Counter
$sel:headConfirmedTx:Metrics :: Counter
$sel:txConfirmationTime:Metrics :: Histogram
$sel:snapshotConfirmationTime:Metrics :: Histogram
$sel:peersConnected:Metrics :: Gauge
$sel:chainDriftSeconds:Metrics :: Gauge
$sel:chainLastBlockTimestampSeconds:Metrics :: Gauge
$sel:broadcastStalled:Metrics :: Gauge
$sel:broadcastStalledNoProgress:Metrics :: Gauge
$sel:broadcastStalledBacklogFull:Metrics :: Gauge
$sel:pendingBroadcasts:Metrics :: Gauge
$sel:broadcastNoProgressSeconds:Metrics :: Gauge
$sel:inputsRefusedBroadcastStalled:Metrics :: Counter
headInputs :: Counter
headRequestedTx :: Counter
headConfirmedTx :: Counter
txConfirmationTime :: Histogram
snapshotConfirmationTime :: Histogram
peersConnected :: Gauge
chainDriftSeconds :: Gauge
chainLastBlockTimestampSeconds :: Gauge
broadcastStalled :: Gauge
broadcastStalledNoProgress :: Gauge
broadcastStalledBacklogFull :: Gauge
pendingBroadcasts :: Gauge
broadcastNoProgressSeconds :: Gauge
inputsRefusedBroadcastStalled :: Counter
..}
 where
  counter :: Text -> IO Counter
counter Text
name = Name -> Labels -> Registry -> IO Counter
registerCounter (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty Registry
registry
  gaugeMetric :: Text -> IO Gauge
gaugeMetric Text
name = Name -> Labels -> Registry -> IO Gauge
registerGauge (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty Registry
registry
  histogram :: Text -> [Double] -> IO Histogram
histogram Text
name [Double]
buckets = Name -> Labels -> [Double] -> Registry -> IO Histogram
registerHistogram (Text -> Name
Name Text
name) Labels
forall a. Monoid a => a
mempty [Double]
buckets Registry
registry

-- | Main monitoring function that updates metrics store given some log entries.
monitor ::
  forall m tx.
  (MonadIO m, MonadSTM m, MonadMonotonicTime m, MonadTime m, IsTx tx) =>
  TVar m (Map (TxIdType tx) Time) ->
  TVar m (Map SnapshotNumber (Time, [TxIdType tx])) ->
  Metrics ->
  HydraLog tx ->
  m ()
monitor :: forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, MonadTime m,
 IsTx tx) =>
TVar m (Map (TxIdType tx) Time)
-> TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Metrics
-> HydraLog tx
-> m ()
monitor TVar m (Map (TxIdType tx) Time)
transactionsMap TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap Metrics{Counter
Gauge
Histogram
$sel:headInputs:Metrics :: Metrics -> Counter
$sel:headRequestedTx:Metrics :: Metrics -> Counter
$sel:headConfirmedTx:Metrics :: Metrics -> Counter
$sel:txConfirmationTime:Metrics :: Metrics -> Histogram
$sel:snapshotConfirmationTime:Metrics :: Metrics -> Histogram
$sel:peersConnected:Metrics :: Metrics -> Gauge
$sel:chainDriftSeconds:Metrics :: Metrics -> Gauge
$sel:chainLastBlockTimestampSeconds:Metrics :: Metrics -> Gauge
$sel:broadcastStalled:Metrics :: Metrics -> Gauge
$sel:broadcastStalledNoProgress:Metrics :: Metrics -> Gauge
$sel:broadcastStalledBacklogFull:Metrics :: Metrics -> Gauge
$sel:pendingBroadcasts:Metrics :: Metrics -> Gauge
$sel:broadcastNoProgressSeconds:Metrics :: Metrics -> Gauge
$sel:inputsRefusedBroadcastStalled:Metrics :: Metrics -> Counter
headInputs :: Counter
headRequestedTx :: Counter
headConfirmedTx :: Counter
txConfirmationTime :: Histogram
snapshotConfirmationTime :: Histogram
peersConnected :: Gauge
chainDriftSeconds :: Gauge
chainLastBlockTimestampSeconds :: Gauge
broadcastStalled :: Gauge
broadcastStalledNoProgress :: Gauge
broadcastStalledBacklogFull :: Gauge
pendingBroadcasts :: Gauge
broadcastNoProgressSeconds :: Gauge
inputsRefusedBroadcastStalled :: Counter
..} = \case
  (Node BeginInput{$sel:input:BeginInput :: forall tx. HydraNodeLog tx -> Input tx
input = NetworkInput Natural
_ (ReceivedMessage{$sel:msg:ConnectivityEvent :: forall msg. NetworkEvent msg -> msg
msg = ReqTx tx
tx})}) -> do
    Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
    -- NOTE: If a requested transaction never gets confirmed, it might stick
    -- forever in the transactions map which could lead to unbounded growth and
    -- memory leak. We might want to have a 'cleaner' thread run that will remove
    -- transactions after some timeout expires
    STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Map (TxIdType tx) Time)
-> (Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m (Map (TxIdType tx) Time)
transactionsMap (TxIdType tx
-> Time -> Map (TxIdType tx) Time -> Map (TxIdType tx) Time
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
tx) Time
t)
    Counter -> m ()
tick Counter
headRequestedTx
  (Node BroadcastBacklog{$sel:pendingBroadcasts:BeginInput :: forall tx. HydraNodeLog tx -> Natural
pendingBroadcasts = Natural
pending, DiffTime
noProgressSeconds :: DiffTime
$sel:noProgressSeconds:BeginInput :: forall tx. HydraNodeLog tx -> DiffTime
noProgressSeconds}) -> do
    Gauge -> Double -> m ()
gaugeN Gauge
pendingBroadcasts (Natural -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Natural
pending)
    Gauge -> Double -> m ()
gaugeN Gauge
broadcastNoProgressSeconds (DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
noProgressSeconds)
  (Node LogicOutcome{$sel:outcome:BeginInput :: forall tx. HydraNodeLog tx -> Outcome tx
outcome = Continue{[StateChanged tx]
stateChanges :: [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges, [Effect tx]
effects :: [Effect tx]
$sel:effects:Continue :: forall tx. Outcome tx -> [Effect tx]
effects}}) -> do
    [Effect tx] -> (Effect tx -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [Effect tx]
effects ((Effect tx -> m ()) -> m ()) -> (Effect tx -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \case
      ClientEffect RejectedInputBecauseBroadcastStalled{} ->
        Counter -> m ()
tick Counter
inputsRefusedBroadcastStalled
      Effect tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    [StateChanged tx] -> (StateChanged tx -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [StateChanged tx]
stateChanges ((StateChanged tx -> m ()) -> m ())
-> (StateChanged tx -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \case
      PeerConnected{} -> (Gauge -> IO ()) -> Gauge -> m ()
gauge Gauge -> IO ()
Gauge.inc Gauge
peersConnected
      PeerDisconnected{} -> (Gauge -> IO ()) -> Gauge -> m ()
gauge Gauge -> IO ()
Gauge.dec Gauge
peersConnected
      NetworkDisconnected{} -> Gauge -> Double -> m ()
gaugeN Gauge
peersConnected Double
0
      -- Set from the state changes rather than from the poll above, so the
      -- gauges say exactly what clients were told - including which of the
      -- two conditions it was.
      NetworkBroadcastStalled{StallReason
stallReason :: StallReason
$sel:stallReason:NetworkConnected :: forall tx. StateChanged tx -> StallReason
stallReason} -> do
        Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalled Double
1
        case StallReason
stallReason of
          StallReason
NoProgress -> do
            Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalledNoProgress Double
1
            Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalledBacklogFull Double
0
          StallReason
BacklogFull -> do
            Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalledNoProgress Double
0
            Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalledBacklogFull Double
1
      StateChanged tx
NetworkBroadcastResumed -> do
        Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalled Double
0
        Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalledNoProgress Double
0
        Gauge -> Double -> m ()
gaugeN Gauge
broadcastStalledBacklogFull Double
0
      -- On every observed tick, report how far behind the chain we are and when
      -- we last heard from the backend. The latter is a wall-clock timestamp, so
      -- monitoring can alert on `time() - hydra_chain_last_block_timestamp_seconds`
      -- and detect a stalled backend even while the drift gauge is frozen (#2749).
      TickObserved{UTCTime
chainTime :: UTCTime
$sel:chainTime:NetworkConnected :: forall tx. StateChanged tx -> UTCTime
chainTime} -> do
        UTCTime
now <- m UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
        Gauge -> Double -> m ()
gaugeN Gauge
chainDriftSeconds (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (UTCTime
now UTCTime -> UTCTime -> NominalDiffTime
`diffUTCTime` UTCTime
chainTime))
        Gauge -> Double -> m ()
gaugeN Gauge
chainLastBlockTimestampSeconds (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (UTCTime -> NominalDiffTime
utcTimeToPOSIXSeconds UTCTime
now))
      SnapshotRequested{$sel:requestedSnapshot:NetworkConnected :: forall tx. StateChanged tx -> Snapshot tx
requestedSnapshot = Snapshot{SnapshotNumber
number :: SnapshotNumber
$sel:number:Snapshot :: forall tx. Snapshot tx -> SnapshotNumber
number, [tx]
confirmed :: [tx]
$sel:confirmed:Snapshot :: forall tx. Snapshot tx -> [tx]
confirmed}} -> do
        Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> (Map SnapshotNumber (Time, [TxIdType tx])
    -> Map SnapshotNumber (Time, [TxIdType tx]))
-> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap (SnapshotNumber
-> (Time, [TxIdType tx])
-> Map SnapshotNumber (Time, [TxIdType tx])
-> Map SnapshotNumber (Time, [TxIdType tx])
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert SnapshotNumber
number (Time
t, tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId (tx -> TxIdType tx) -> [tx] -> [TxIdType tx]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [tx]
confirmed))
      SnapshotConfirmed{$sel:snapshot:NetworkConnected :: forall tx. StateChanged tx -> Maybe (Snapshot tx)
snapshot = Maybe (Snapshot tx)
mSnapshot} -> do
        Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
        -- On the normal signing path the event carries no snapshot; the
        -- protocol confirms snapshots strictly in order, so the oldest
        -- in-flight request is the confirmed one.
        Maybe (Time, [TxIdType tx])
mEntry <- STM m (Maybe (Time, [TxIdType tx]))
-> m (Maybe (Time, [TxIdType tx]))
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (Maybe (Time, [TxIdType tx]))
 -> m (Maybe (Time, [TxIdType tx])))
-> STM m (Maybe (Time, [TxIdType tx]))
-> m (Maybe (Time, [TxIdType tx]))
forall a b. (a -> b) -> a -> b
$ do
          Map SnapshotNumber (Time, [TxIdType tx])
inFlight <- TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> STM m (Map SnapshotNumber (Time, [TxIdType tx]))
forall a. TVar m a -> STM m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> STM m a
readTVar TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap
          Maybe (SnapshotNumber, (Time, [TxIdType tx]))
-> ((SnapshotNumber, (Time, [TxIdType tx]))
    -> STM m (Time, [TxIdType tx]))
-> STM m (Maybe (Time, [TxIdType tx]))
forall (t :: * -> *) (m :: * -> *) a b.
(Traversable t, Monad m) =>
t a -> (a -> m b) -> m (t b)
forM (Map SnapshotNumber (Time, [TxIdType tx])
-> Maybe (SnapshotNumber, (Time, [TxIdType tx]))
forall k a. Map k a -> Maybe (k, a)
Map.lookupMin Map SnapshotNumber (Time, [TxIdType tx])
inFlight) (((SnapshotNumber, (Time, [TxIdType tx]))
  -> STM m (Time, [TxIdType tx]))
 -> STM m (Maybe (Time, [TxIdType tx])))
-> ((SnapshotNumber, (Time, [TxIdType tx]))
    -> STM m (Time, [TxIdType tx]))
-> STM m (Maybe (Time, [TxIdType tx]))
forall a b. (a -> b) -> a -> b
$ \(SnapshotNumber
number, (Time, [TxIdType tx])
entry) -> do
            TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
-> Map SnapshotNumber (Time, [TxIdType tx]) -> STM m ()
forall a. TVar m a -> a -> STM m ()
forall (m :: * -> *) a. MonadSTM m => TVar m a -> a -> STM m ()
writeTVar TVar m (Map SnapshotNumber (Time, [TxIdType tx]))
snapshotsMap (SnapshotNumber
-> Map SnapshotNumber (Time, [TxIdType tx])
-> Map SnapshotNumber (Time, [TxIdType tx])
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete SnapshotNumber
number Map SnapshotNumber (Time, [TxIdType tx])
inFlight)
            (Time, [TxIdType tx]) -> STM m (Time, [TxIdType tx])
forall a. a -> STM m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Time, [TxIdType tx])
entry
        [TxIdType tx]
confirmedIds <- case (Maybe (Snapshot tx)
mSnapshot, Maybe (Time, [TxIdType tx])
mEntry) of
          -- Side-load path: no preceding SnapshotRequested, but the event
          -- carries the snapshot.
          (Just Snapshot{[tx]
$sel:confirmed:Snapshot :: forall tx. Snapshot tx -> [tx]
confirmed :: [tx]
confirmed}, Maybe (Time, [TxIdType tx])
_) -> [TxIdType tx] -> m [TxIdType tx]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId (tx -> TxIdType tx) -> [tx] -> [TxIdType tx]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [tx]
confirmed)
          (Maybe (Snapshot tx)
Nothing, Just (Time
_, [TxIdType tx]
txIds)) -> [TxIdType tx] -> m [TxIdType tx]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure [TxIdType tx]
txIds
          (Maybe (Snapshot tx)
Nothing, Maybe (Time, [TxIdType tx])
Nothing) -> [TxIdType tx] -> m [TxIdType tx]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
        Maybe (Time, [TxIdType tx])
-> ((Time, [TxIdType tx]) -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Time, [TxIdType tx])
mEntry (((Time, [TxIdType tx]) -> m ()) -> m ())
-> ((Time, [TxIdType tx]) -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \(Time
start, [TxIdType tx]
_) ->
          Histogram -> DiffTime -> m ()
histo Histogram
snapshotConfirmationTime (Time -> Time -> DiffTime
diffTime Time
t Time
start)
        Counter -> Port -> m ()
tickN Counter
headConfirmedTx ([TxIdType tx] -> Port
forall a. [a] -> Port
forall (t :: * -> *) a. Foldable t => t a -> Port
length [TxIdType tx]
confirmedIds)
        [TxIdType tx] -> (TxIdType tx -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [TxIdType tx]
confirmedIds ((TxIdType tx -> m ()) -> m ()) -> (TxIdType tx -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \TxIdType tx
i -> do
          Map (TxIdType tx) Time
txsStartTime <- TVar m (Map (TxIdType tx) Time) -> m (Map (TxIdType tx) Time)
forall a. TVar m a -> m a
forall (m :: * -> *) a. MonadSTM m => TVar m a -> m a
readTVarIO TVar m (Map (TxIdType tx) Time)
transactionsMap
          case TxIdType tx -> Map (TxIdType tx) Time -> Maybe Time
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup TxIdType tx
i Map (TxIdType tx) Time
txsStartTime of
            Just Time
start -> do
              STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar m (Map (TxIdType tx) Time)
-> (Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ()
forall a. TVar m a -> (a -> a) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
TVar m a -> (a -> a) -> STM m ()
modifyTVar' TVar m (Map (TxIdType tx) Time)
transactionsMap ((Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ())
-> (Map (TxIdType tx) Time -> Map (TxIdType tx) Time) -> STM m ()
forall a b. (a -> b) -> a -> b
$ TxIdType tx -> Map (TxIdType tx) Time -> Map (TxIdType tx) Time
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete TxIdType tx
i
              Histogram -> DiffTime -> m ()
histo Histogram
txConfirmationTime (Time -> Time -> DiffTime
diffTime Time
t Time
start)
            Maybe Time
Nothing -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      StateChanged tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  (Node (EndInput Party
_ Word64
_)) ->
    Counter -> m ()
tick Counter
headInputs
  HydraLog tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
 where
  gaugeN :: Gauge -> Double -> m ()
  gaugeN :: Gauge -> Double -> m ()
gaugeN Gauge
g Double
num = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Double -> Gauge -> IO ()
Gauge.set Double
num Gauge
g

  gauge :: (Gauge -> IO ()) -> Gauge -> m ()
  gauge :: (Gauge -> IO ()) -> Gauge -> m ()
gauge Gauge -> IO ()
f Gauge
g = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Gauge -> IO ()
f Gauge
g

  tick :: Counter -> m ()
  tick :: Counter -> m ()
tick Counter
c = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Counter -> IO ()
inc Counter
c

  tickN :: Counter -> Int -> m ()
  tickN :: Counter -> Port -> m ()
tickN Counter
c Port
num = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Port -> Counter -> IO ()
add Port
num Counter
c

  histo :: Histogram -> DiffTime -> m ()
  histo :: Histogram -> DiffTime -> m ()
histo Histogram
h DiffTime
time = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Double -> Histogram -> IO ()
observe (Rational -> Double
forall a. Fractional a => Rational -> a
fromRational (Rational -> Double) -> Rational -> Double
forall a b. (a -> b) -> a -> b
$ DiffTime -> Rational
forall a. Real a => a -> Rational
toRational (DiffTime -> Rational) -> DiffTime -> Rational
forall a b. (a -> b) -> a -> b
$ DiffTime
time DiffTime -> DiffTime -> DiffTime
forall a. Num a => a -> a -> a
* DiffTime
1000) Histogram
h