-- | Provides Prometheus-based metrics server based on `Tracer` collection.
--
-- To add a new metric, one needs to:
--
--  * Add a 'MetricDefinition' to the 'allMetrics' list, providing a unique 'Name', the
--    relevant constructor for the 'Metric' value and a registration function,
--  * Update the 'monitor' function to Handle relevant 'HydraLog' entries and update
--    underlying Prometheus metrics store. Nested helpers are provided to increase a
--    'Counter' by one (@tick@) and to 'observe' some value in an 'Histogram'.
module Hydra.Logging.Monitoring (
  withMonitoring,
) where

import Hydra.Prelude

import Control.Concurrent.Class.MonadSTM (modifyTVar', newTVarIO, readTVarIO)
import Control.Tracer (Tracer (Tracer))
import Data.Map.Strict as Map
import Hydra.HeadLogic (
  Input (NetworkInput),
 )
import Hydra.HeadLogic.Outcome (Outcome (..), StateChanged (..))
import Hydra.Logging.Messages (HydraLog (..))
import Hydra.Network (PortNumber)
import Hydra.Network.Message (Message (ReqTx), NetworkEvent (..))
import Hydra.Node (HydraNodeLog (..))
import Hydra.Tx (IsTx (TxIdType), Snapshot (..), txId)
import System.Metrics.Prometheus.Http.Scrape (serveMetrics)
import System.Metrics.Prometheus.Metric (Metric (CounterMetric, HistogramMetric))
import System.Metrics.Prometheus.Metric.Counter (add, inc)
import System.Metrics.Prometheus.Metric.Histogram (observe)
import System.Metrics.Prometheus.MetricId (Name (Name))
import System.Metrics.Prometheus.Registry (Registry, new, registerCounter, registerHistogram, sample)

-- | 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) =>
  Maybe PortNumber ->
  Tracer m (HydraLog tx) ->
  (Tracer m (HydraLog tx) -> m ()) ->
  m ()
withMonitoring :: forall (m :: * -> *) tx.
(MonadIO m, MonadAsync m, IsTx tx, MonadMonotonicTime 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, MonadSTM m, MonadMonotonicTime m, IsTx tx) =>
m (HydraLog tx -> m (), Registry)
prepareRegistry
  m () -> (Async m () -> m ()) -> m ()
forall a b. m a -> (Async m a -> m b) -> m b
forall (m :: * -> *) a b.
MonadAsync m =>
m a -> (Async m a -> m b) -> m b
withAsync (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"] (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 all relevant metrics.
-- Returns an updated `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 :: (MonadIO m, MonadSTM m, MonadMonotonicTime m, IsTx tx) => m (HydraLog tx -> m (), Registry)
prepareRegistry :: forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, IsTx tx) =>
m (HydraLog tx -> m (), Registry)
prepareRegistry = do
  TVar m (Map (TxIdType tx) Time)
transactionsMap <- Map (TxIdType tx) Time -> m (TVar m (Map (TxIdType tx) Time))
forall a. a -> m (TVar m a)
forall (m :: * -> *) a. MonadSTM m => a -> m (TVar m a)
newTVarIO Map (TxIdType tx) Time
forall a. Monoid a => a
mempty
  (Map Name Metric -> HydraLog tx -> m ())
-> (Map Name Metric, Registry) -> (HydraLog tx -> m (), Registry)
forall a b c. (a -> b) -> (a, c) -> (b, c)
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first (TVar m (Map (TxIdType tx) Time)
-> Map Name Metric -> HydraLog tx -> m ()
forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, IsTx tx) =>
TVar m (Map (TxIdType tx) Time)
-> Map Name Metric -> HydraLog tx -> m ()
monitor TVar m (Map (TxIdType tx) Time)
transactionsMap) ((Map Name Metric, Registry) -> (HydraLog tx -> m (), Registry))
-> m (Map Name Metric, Registry)
-> m (HydraLog tx -> m (), Registry)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m (Map Name Metric, Registry)
registerMetrics
 where
  registerMetrics :: m (Map Name Metric, Registry)
registerMetrics = ((Map Name Metric, Registry)
 -> MetricDefinition -> m (Map Name Metric, Registry))
-> (Map Name Metric, Registry)
-> [MetricDefinition]
-> m (Map Name Metric, Registry)
forall (t :: * -> *) (m :: * -> *) b a.
(Foldable t, Monad m) =>
(b -> a -> m b) -> b -> t a -> m b
foldlM (Map Name Metric, Registry)
-> MetricDefinition -> m (Map Name Metric, Registry)
forall {m :: * -> *}.
MonadIO m =>
(Map Name Metric, Registry)
-> MetricDefinition -> m (Map Name Metric, Registry)
registerMetric (Map Name Metric
forall a. Monoid a => a
mempty, Registry
new) [MetricDefinition]
allMetrics

  registerMetric :: (Map Name Metric, Registry)
-> MetricDefinition -> m (Map Name Metric, Registry)
registerMetric (Map Name Metric
metricsMap, Registry
registry) (MetricDefinition Name
name a -> Metric
ctor Name -> Registry -> IO (a, Registry)
registration) = do
    (a
metric, Registry
registry') <- IO (a, Registry) -> m (a, Registry)
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO (a, Registry) -> m (a, Registry))
-> IO (a, Registry) -> m (a, Registry)
forall a b. (a -> b) -> a -> b
$ Name -> Registry -> IO (a, Registry)
registration Name
name Registry
registry
    (Map Name Metric, Registry) -> m (Map Name Metric, Registry)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Name -> Metric -> Map Name Metric -> Map Name Metric
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert Name
name (a -> Metric
ctor a
metric) Map Name Metric
metricsMap, Registry
registry')

-- | Existential wrapper around different kind of metrics construction logic.
data MetricDefinition where
  MetricDefinition :: forall a. Name -> (a -> Metric) -> (Name -> Registry -> IO (a, Registry)) -> MetricDefinition

-- | All custom 'MetricDefinition's for Hydra
allMetrics :: [MetricDefinition]
allMetrics :: [MetricDefinition]
allMetrics =
  [ Name
-> (Counter -> Metric)
-> (Name -> Registry -> IO (Counter, Registry))
-> MetricDefinition
forall a.
Name
-> (a -> Metric)
-> (Name -> Registry -> IO (a, Registry))
-> MetricDefinition
MetricDefinition (Text -> Name
Name Text
"hydra_head_inputs") Counter -> Metric
CounterMetric ((Name -> Registry -> IO (Counter, Registry)) -> MetricDefinition)
-> (Name -> Registry -> IO (Counter, Registry)) -> MetricDefinition
forall a b. (a -> b) -> a -> b
$ (Name -> Labels -> Registry -> IO (Counter, Registry))
-> Labels -> Name -> Registry -> IO (Counter, Registry)
forall a b c. (a -> b -> c) -> b -> a -> c
flip Name -> Labels -> Registry -> IO (Counter, Registry)
registerCounter Labels
forall a. Monoid a => a
mempty
  , Name
-> (Counter -> Metric)
-> (Name -> Registry -> IO (Counter, Registry))
-> MetricDefinition
forall a.
Name
-> (a -> Metric)
-> (Name -> Registry -> IO (a, Registry))
-> MetricDefinition
MetricDefinition (Text -> Name
Name Text
"hydra_head_requested_tx") Counter -> Metric
CounterMetric ((Name -> Registry -> IO (Counter, Registry)) -> MetricDefinition)
-> (Name -> Registry -> IO (Counter, Registry)) -> MetricDefinition
forall a b. (a -> b) -> a -> b
$ (Name -> Labels -> Registry -> IO (Counter, Registry))
-> Labels -> Name -> Registry -> IO (Counter, Registry)
forall a b c. (a -> b -> c) -> b -> a -> c
flip Name -> Labels -> Registry -> IO (Counter, Registry)
registerCounter Labels
forall a. Monoid a => a
mempty
  , Name
-> (Counter -> Metric)
-> (Name -> Registry -> IO (Counter, Registry))
-> MetricDefinition
forall a.
Name
-> (a -> Metric)
-> (Name -> Registry -> IO (a, Registry))
-> MetricDefinition
MetricDefinition (Text -> Name
Name Text
"hydra_head_confirmed_tx") Counter -> Metric
CounterMetric ((Name -> Registry -> IO (Counter, Registry)) -> MetricDefinition)
-> (Name -> Registry -> IO (Counter, Registry)) -> MetricDefinition
forall a b. (a -> b) -> a -> b
$ (Name -> Labels -> Registry -> IO (Counter, Registry))
-> Labels -> Name -> Registry -> IO (Counter, Registry)
forall a b c. (a -> b -> c) -> b -> a -> c
flip Name -> Labels -> Registry -> IO (Counter, Registry)
registerCounter Labels
forall a. Monoid a => a
mempty
  , Name
-> (Histogram -> Metric)
-> (Name -> Registry -> IO (Histogram, Registry))
-> MetricDefinition
forall a.
Name
-> (a -> Metric)
-> (Name -> Registry -> IO (a, Registry))
-> MetricDefinition
MetricDefinition (Text -> Name
Name Text
"hydra_head_tx_confirmation_time_ms") Histogram -> Metric
HistogramMetric ((Name -> Registry -> IO (Histogram, Registry))
 -> MetricDefinition)
-> (Name -> Registry -> IO (Histogram, Registry))
-> MetricDefinition
forall a b. (a -> b) -> a -> b
$ \Name
n -> Name
-> Labels -> [UpperBound] -> Registry -> IO (Histogram, Registry)
registerHistogram Name
n Labels
forall a. Monoid a => a
mempty [UpperBound
5, UpperBound
10, UpperBound
50, UpperBound
100, UpperBound
1000]
  ]

-- | Main monitoring function that updates metrics store given some log entries.
monitor ::
  (MonadIO m, MonadSTM m, MonadMonotonicTime m, IsTx tx) =>
  TVar m (Map (TxIdType tx) Time) ->
  Map Name Metric ->
  HydraLog tx ->
  m ()
monitor :: forall (m :: * -> *) tx.
(MonadIO m, MonadSTM m, MonadMonotonicTime m, IsTx tx) =>
TVar m (Map (TxIdType tx) Time)
-> Map Name Metric -> HydraLog tx -> m ()
monitor TVar m (Map (TxIdType tx) Time)
transactionsMap Map Name Metric
metricsMap = \case
  (Node BeginInput{$sel:input:BeginInput :: forall tx. HydraNodeLog tx -> Input tx
input = NetworkInput TTL
_ (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)
    Name -> m ()
tick Name
"hydra_head_requested_tx"
  (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}}) -> do
    [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
      SnapshotConfirmed{$sel:snapshot:NetworkConnected :: forall tx. StateChanged tx -> Snapshot tx
snapshot = Snapshot{[tx]
confirmed :: [tx]
$sel:confirmed:Snapshot :: forall tx. Snapshot tx -> [tx]
confirmed}} -> do
        Name -> Port -> m ()
tickN Name
"hydra_head_confirmed_tx" ([tx] -> Port
forall a. [a] -> Port
forall (t :: * -> *) a. Foldable t => t a -> Port
length [tx]
confirmed)
        [tx] -> (tx -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ [tx]
confirmed ((tx -> m ()) -> m ()) -> (tx -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \tx
tx -> do
          Time
t <- m Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
          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 (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
tx) 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 (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
tx)
              Name -> DiffTime -> m ()
histo Name
"hydra_head_tx_confirmation_time_ms" (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
_)) ->
    Name -> m ()
tick Name
"hydra_head_inputs"
  HydraLog tx
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
 where
  tick :: Name -> m ()
tick Name
metricName =
    case Name -> Map Name Metric -> Maybe Metric
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Name
metricName Map Name Metric
metricsMap of
      (Just (CounterMetric 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
      Maybe Metric
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

  tickN :: Name -> Port -> m ()
tickN Name
metricName Port
num =
    case Name -> Map Name Metric -> Maybe Metric
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Name
metricName Map Name Metric
metricsMap of
      (Just (CounterMetric 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
$ Port -> Counter -> IO ()
add Port
num Counter
c
      Maybe Metric
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()

  histo :: Name -> DiffTime -> m ()
histo Name
metricName DiffTime
time =
    case Name -> Map Name Metric -> Maybe Metric
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup Name
metricName Map Name Metric
metricsMap of
      (Just (HistogramMetric Histogram
h)) -> 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
$ UpperBound -> Histogram -> IO ()
observe (Rational -> UpperBound
forall a. Fractional a => Rational -> a
fromRational (Rational -> UpperBound) -> Rational -> UpperBound
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
      Maybe Metric
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()