{-# LANGUAGE RecordWildCards #-}

-- | Prometheus exposition and trace-driven accounting. Counters describe this
-- process; state gauges are sampled independently of the processing queue.
module Hydra.Logging.Monitoring (
  withMonitoring,
  withMonitoringForBackend,
  prepareRegistry,
  isMetricSample,
) where

import Hydra.Prelude hiding (bool, handle, state)

import Control.Concurrent.Class.MonadSTM (putTMVar, takeTMVar)
import Control.Concurrent.MVar qualified as MVar
import Control.Monad.Class.MonadAsync (link)
import Control.Tracer (Tracer (Tracer), emit, traceWith)
import Data.EventSource.SQLite (SQLiteLog (..))
import Data.Map.Strict qualified as Map
import Data.Set qualified as Set
import Data.Time.Clock.POSIX (utcTimeToPOSIXSeconds)
import GHC.Stats (RTSStats (..), getRTSStats, getRTSStatsEnabled)
import Hydra.API.APIServerLog qualified as API
import Hydra.API.ClientInput qualified as Client
import Hydra.API.ServerOutput qualified as Output
import Hydra.Cardano.Api (ChainPoint (..), SlotNo (..))
import Hydra.Chain.Direct.Handlers qualified as ChainLog
import Hydra.HeadLogic (Input (..))
import Hydra.HeadLogic.Outcome (Effect (ClientEffect), Outcome (..), StateChanged (..), WaitReason (..))
import Hydra.HeadLogic.State qualified as Head
import Hydra.Logging.Messages (HydraLog (..))
import Hydra.Logging.Monitoring.Accounting qualified as Accounting
import Hydra.Logging.Monitoring.Metrics (Metrics (..), at, registerMetrics)
import Hydra.Logging.Monitoring.Registry qualified as Registry
import Hydra.Network (PortNumber, StallReason (..))
import Hydra.Network.Etcd qualified as Etcd
import Hydra.Network.Message (Message (ReqTx), NetworkEvent (..))
import Hydra.Node (HydraNodeLog (..))
import Hydra.Node.Network qualified as Network
import Hydra.Node.State qualified as NodeState
import Hydra.Node.Stats qualified as Stats
import Hydra.Tx (HeadId, IsTx (..), Snapshot (..), getSnapshot)
import Network.HTTP.Types (status200, status404)
import Network.Wai (pathInfo, responseLBS)
import Network.Wai.Handler.Warp (defaultSettings, runSettings, setBeforeMainLoop, setPort)
import System.Metrics.Prometheus.Metric.Counter qualified as Counter
import System.Metrics.Prometheus.Metric.Gauge qualified as Gauge
import System.Metrics.Prometheus.Metric.Histogram qualified as Histogram

withMonitoring :: IsTx tx => Maybe PortNumber -> Tracer IO (HydraLog tx) -> (Tracer IO (HydraLog tx) -> IO a) -> IO a
withMonitoring :: forall tx a.
IsTx tx =>
Maybe PortNumber
-> Tracer IO (HydraLog tx)
-> (Tracer IO (HydraLog tx) -> IO a)
-> IO a
withMonitoring = Text
-> Maybe PortNumber
-> Tracer IO (HydraLog tx)
-> (Tracer IO (HydraLog tx) -> IO a)
-> IO a
forall tx a.
IsTx tx =>
Text
-> Maybe PortNumber
-> Tracer IO (HydraLog tx)
-> (Tracer IO (HydraLog tx) -> IO a)
-> IO a
withMonitoringForBackend Text
"unknown"

withMonitoringForBackend :: IsTx tx => Text -> Maybe PortNumber -> Tracer IO (HydraLog tx) -> (Tracer IO (HydraLog tx) -> IO a) -> IO a
withMonitoringForBackend :: forall tx a.
IsTx tx =>
Text
-> Maybe PortNumber
-> Tracer IO (HydraLog tx)
-> (Tracer IO (HydraLog tx) -> IO a)
-> IO a
withMonitoringForBackend Text
_ Maybe PortNumber
Nothing Tracer IO (HydraLog tx)
tracer Tracer IO (HydraLog tx) -> IO a
action = Tracer IO (HydraLog tx) -> IO a
action Tracer IO (HydraLog tx)
tracer
withMonitoringForBackend Text
backend (Just PortNumber
port) Tracer IO (HydraLog tx)
tracer Tracer IO (HydraLog tx) -> IO a
action = do
  (HydraLog tx -> IO ()
traceMetric, Registry
registry, IO ()
refresh) <- Text -> IO (HydraLog tx -> IO (), Registry, IO ())
forall tx.
IsTx tx =>
Text -> IO (HydraLog tx -> IO (), Registry, IO ())
prepareRegistry Text
backend
  IO ()
refreshRuntime <- Registry -> IO (IO ())
registerRtsMetrics Registry
registry
  MVar ()
scrapeLock <- () -> IO (MVar ())
forall a. a -> IO (MVar a)
MVar.newMVar ()
  TMVar ()
listening <- String -> IO (TMVar IO ())
forall (m :: * -> *) a.
MonadLabelledSTM m =>
String -> m (TMVar m a)
newLabelledEmptyTMVarIO String
"monitoring-listening"
  let settings :: Settings
settings = Port -> Settings -> Settings
setPort (PortNumber -> Port
forall a b. (Integral a, Num b) => a -> b
fromIntegral PortNumber
port) (Settings -> Settings) -> Settings -> Settings
forall a b. (a -> b) -> a -> b
$ IO () -> Settings -> Settings
setBeforeMainLoop (STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TMVar IO () -> () -> STM IO ()
forall a. TMVar IO a -> a -> STM IO ()
forall (m :: * -> *) a. MonadSTM m => TMVar m a -> a -> STM m ()
putTMVar TMVar ()
TMVar IO ()
listening ()) Settings
defaultSettings
      app :: Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived
app Request
request Response -> IO ResponseReceived
respond
        | Request -> [Text]
pathInfo Request
request [Text] -> [Text] -> Bool
forall a. Eq a => a -> a -> Bool
== [Text
"metrics"] = do
            ByteString
body <- MVar () -> (() -> IO ByteString) -> IO ByteString
forall a b. MVar a -> (a -> IO b) -> IO b
MVar.withMVar MVar ()
scrapeLock ((() -> IO ByteString) -> IO ByteString)
-> (() -> IO ByteString) -> IO ByteString
forall a b. (a -> b) -> a -> b
$ \()
_ -> IO ()
refresh IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO ()
refreshRuntime IO () -> IO ByteString -> IO ByteString
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Registry -> IO ByteString
Registry.sample Registry
registry
            Response -> IO ResponseReceived
respond (Response -> IO ResponseReceived)
-> Response -> IO ResponseReceived
forall a b. (a -> b) -> a -> b
$ Status -> ResponseHeaders -> ByteString -> Response
responseLBS Status
status200 [(HeaderName
"Content-Type", ByteString
"text/plain; version=0.0.4; charset=utf-8")] ByteString
body
        | Bool
otherwise = Response -> IO ResponseReceived
respond (Response -> IO ResponseReceived)
-> Response -> IO ResponseReceived
forall a b. (a -> b) -> a -> b
$ Status -> ResponseHeaders -> ByteString -> Response
responseLBS Status
status404 [] ByteString
"Not found"
      wrapped :: Tracer IO (HydraLog tx)
wrapped = TracerA IO (HydraLog tx) () -> Tracer IO (HydraLog tx)
forall (m :: * -> *) a. TracerA m a () -> Tracer m a
Tracer (TracerA IO (HydraLog tx) () -> Tracer IO (HydraLog tx))
-> ((HydraLog tx -> IO ()) -> TracerA IO (HydraLog tx) ())
-> (HydraLog tx -> IO ())
-> Tracer IO (HydraLog tx)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (HydraLog tx -> IO ()) -> TracerA IO (HydraLog tx) ()
forall (m :: * -> *) a.
Applicative m =>
(a -> m ()) -> TracerA m a ()
emit ((HydraLog tx -> IO ()) -> Tracer IO (HydraLog tx))
-> (HydraLog tx -> IO ()) -> Tracer IO (HydraLog tx)
forall a b. (a -> b) -> a -> b
$ \HydraLog tx
msg -> HydraLog tx -> IO ()
traceMetric HydraLog tx
msg IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Tracer IO (HydraLog tx) -> HydraLog tx -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO (HydraLog tx)
tracer HydraLog tx
msg
  (String, IO ()) -> (Async IO () -> IO a) -> IO a
forall (m :: * -> *) a b.
MonadAsync m =>
(String, m a) -> (Async m a -> m b) -> m b
withAsyncLabelled (String
"monitoring-server", Settings
-> (Request
    -> (Response -> IO ResponseReceived) -> IO ResponseReceived)
-> IO ()
runSettings Settings
settings Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived
app) ((Async IO () -> IO a) -> IO a) -> (Async IO () -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Async IO ()
server -> do
    -- Bind errors and later server failures are fatal, and callers may scrape
    -- as soon as their continuation starts.
    Async IO () -> IO ()
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m) =>
Async m a -> m ()
link Async IO ()
server
    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TMVar IO () -> STM IO ()
forall a. TMVar IO a -> STM IO a
forall (m :: * -> *) a. MonadSTM m => TMVar m a -> STM m a
takeTMVar TMVar ()
TMVar IO ()
listening
    Tracer IO (HydraLog tx) -> IO a
action Tracer IO (HydraLog tx)
wrapped

-- | Samples bypass the JSON queue in production, so a blocked log output does
-- not freeze the health workers.
isMetricSample :: HydraLog tx -> Bool
isMetricSample :: forall tx. HydraLog tx -> Bool
isMetricSample = \case
  Node InputDequeued{} -> Bool
True
  Node ChainPostStarted{} -> Bool
True
  Node ChainPostFinished{} -> Bool
True
  Node NodeHealth{} -> Bool
True
  Node BroadcastHealth{} -> Bool
True
  Node BroadcastBacklog{} -> Bool
True
  Network (Network.Etcd Etcd.PersistentQueueStats{}) -> Bool
True
  Network (Network.Etcd Etcd.PersistentQueueStatFailed{}) -> Bool
True
  Network (Network.Etcd Etcd.BroadcastCompleted{}) -> Bool
True
  Network (Network.Etcd Etcd.NetworkConnectivity{}) -> Bool
True
  SQLite SQLiteLog
SQLiteWriteStarted -> Bool
True
  SQLite SQLiteWriteCompleted{} -> Bool
True
  SQLite SQLiteLog
SQLiteWriteFailed -> Bool
True
  APIServer API.APIL1Started{} -> Bool
True
  APIServer API.APIL1Finished{} -> Bool
True
  SQLite SQLiteStats{} -> Bool
True
  HydraLog tx
_ -> Bool
False

-- | The refresh action expires unused timers even when no messages arrive.
prepareRegistry :: forall tx. IsTx tx => Text -> IO (HydraLog tx -> IO (), Registry.Registry, IO ())
prepareRegistry :: forall tx.
IsTx tx =>
Text -> IO (HydraLog tx -> IO (), Registry, IO ())
prepareRegistry Text
backend = do
  Registry
registry <- IO Registry
Registry.new
  Metrics{Map Text Counter
Map Text Gauge
Map Text Histogram
Counter
Gauge
Histogram
headInputs :: Counter
headRequestedTx :: Counter
headConfirmedTx :: Counter
txConfirmationTime :: Histogram
snapshotConfirmationTime :: Histogram
snapshots :: Map Text Counter
evictedTimings :: Counter
peersConnected :: Gauge
chainDriftSeconds :: Gauge
chainLastBlockTimestampSeconds :: Gauge
chainBackendTimestamp :: Gauge
broadcastStalled :: Gauge
broadcastStalledNoProgress :: Gauge
broadcastStalledBacklogFull :: Gauge
pendingBroadcasts :: Gauge
broadcastNoProgressSeconds :: Gauge
inputsRefusedBroadcastStalled :: Counter
inputRetries :: Map Text Counter
inputDrops :: Map Text Counter
inputProcessing :: Map Text Histogram
inputQueueWait :: Histogram
queuedInputs :: Gauge
parkedInputs :: Gauge
synced :: Gauge
phase :: Map Text Gauge
networkConnected :: Gauge
snapshotNumber :: Gauge
snapshotPending :: Gauge
missingSignatures :: Gauge
outstandingTransactions :: Gauge
txOutcomes :: Map Text Counter
persistentDepth :: Gauge
persistentAge :: Gauge
putLatency :: Histogram
networkEvents :: Map Text Counter
chainFailures :: Map Text Counter
chainDuration :: Map Text Histogram
chainInflight :: Map Text Gauge
rollbackDepth :: Histogram
blockfrostRetries :: Counter
blockfrostRateLimits :: Counter
deposits :: Map Text Gauge
decommit :: Map Text Gauge
fanoutRemaining :: Gauge
fanoutMode :: Map Text Gauge
sqliteQueue :: Gauge
sqliteInflight :: Gauge
sqliteDuration :: 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:snapshots:Metrics :: Metrics -> Map Text Counter
$sel:evictedTimings:Metrics :: Metrics -> Counter
$sel:peersConnected:Metrics :: Metrics -> Gauge
$sel:chainDriftSeconds:Metrics :: Metrics -> Gauge
$sel:chainLastBlockTimestampSeconds:Metrics :: Metrics -> Gauge
$sel:chainBackendTimestamp: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
$sel:inputRetries:Metrics :: Metrics -> Map Text Counter
$sel:inputDrops:Metrics :: Metrics -> Map Text Counter
$sel:inputProcessing:Metrics :: Metrics -> Map Text Histogram
$sel:inputQueueWait:Metrics :: Metrics -> Histogram
$sel:queuedInputs:Metrics :: Metrics -> Gauge
$sel:parkedInputs:Metrics :: Metrics -> Gauge
$sel:synced:Metrics :: Metrics -> Gauge
$sel:phase:Metrics :: Metrics -> Map Text Gauge
$sel:networkConnected:Metrics :: Metrics -> Gauge
$sel:snapshotNumber:Metrics :: Metrics -> Gauge
$sel:snapshotPending:Metrics :: Metrics -> Gauge
$sel:missingSignatures:Metrics :: Metrics -> Gauge
$sel:outstandingTransactions:Metrics :: Metrics -> Gauge
$sel:txOutcomes:Metrics :: Metrics -> Map Text Counter
$sel:persistentDepth:Metrics :: Metrics -> Gauge
$sel:persistentAge:Metrics :: Metrics -> Gauge
$sel:putLatency:Metrics :: Metrics -> Histogram
$sel:networkEvents:Metrics :: Metrics -> Map Text Counter
$sel:chainFailures:Metrics :: Metrics -> Map Text Counter
$sel:chainDuration:Metrics :: Metrics -> Map Text Histogram
$sel:chainInflight:Metrics :: Metrics -> Map Text Gauge
$sel:rollbackDepth:Metrics :: Metrics -> Histogram
$sel:blockfrostRetries:Metrics :: Metrics -> Counter
$sel:blockfrostRateLimits:Metrics :: Metrics -> Counter
$sel:deposits:Metrics :: Metrics -> Map Text Gauge
$sel:decommit:Metrics :: Metrics -> Map Text Gauge
$sel:fanoutRemaining:Metrics :: Metrics -> Gauge
$sel:fanoutMode:Metrics :: Metrics -> Map Text Gauge
$sel:sqliteQueue:Metrics :: Metrics -> Gauge
$sel:sqliteInflight:Metrics :: Metrics -> Gauge
$sel:sqliteDuration:Metrics :: Metrics -> Histogram
..} <- Registry -> Text -> IO Metrics
registerMetrics Registry
registry Text
backend
  MVar (Accounting HeadId (TxIdType tx))
accounting <- Accounting HeadId (TxIdType tx)
-> IO (MVar (Accounting HeadId (TxIdType tx)))
forall a. a -> IO (MVar a)
MVar.newMVar Accounting HeadId (TxIdType tx)
forall head txid. Accounting head txid
Accounting.empty
  IORef (Maybe (Double, Text))
inputStarted <- Maybe (Double, Text) -> IO (IORef (Maybe (Double, Text)))
forall (m :: * -> *) a. MonadIO m => a -> m (IORef a)
newIORef Maybe (Double, Text)
forall a. Maybe a
Nothing
  IORef (Set Host)
peerSet <- Set Host -> IO (IORef (Set Host))
forall (m :: * -> *) a. MonadIO m => a -> m (IORef a)
newIORef Set Host
forall a. Set a
Set.empty
  IORef Double
backendSlot <- Double -> IO (IORef Double)
forall (m :: * -> *) a. MonadIO m => a -> m (IORef a)
newIORef Double
0
  let set :: Gauge -> Double -> IO ()
set = (Double -> Gauge -> IO ()) -> Gauge -> Double -> IO ()
forall a b c. (a -> b -> c) -> b -> a -> c
flip Double -> Gauge -> IO ()
Gauge.set
      tick :: Counter -> IO ()
tick = Counter -> IO ()
Counter.inc
      add :: Counter -> Port -> IO ()
add = (Port -> Counter -> IO ()) -> Counter -> Port -> IO ()
forall a b c. (a -> b -> c) -> b -> a -> c
flip Port -> Counter -> IO ()
Counter.add
      observe :: Histogram -> Double -> IO ()
observe = (Double -> Histogram -> IO ()) -> Histogram -> Double -> IO ()
forall a b c. (a -> b -> c) -> b -> a -> c
flip Double -> Histogram -> IO ()
Histogram.observe
      bool :: Gauge -> Bool -> IO ()
bool Gauge
gauge Bool
value = Gauge -> Double -> IO ()
set Gauge
gauge (if Bool
value then Double
1 else Double
0)
      timestamp :: Gauge -> IO ()
timestamp Gauge
gauge = IO UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime IO UTCTime -> (UTCTime -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Gauge -> Double -> IO ()
set Gauge
gauge (Double -> IO ()) -> (UTCTime -> Double) -> UTCTime -> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (NominalDiffTime -> Double)
-> (UTCTime -> NominalDiffTime) -> UTCTime -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. UTCTime -> NominalDiffTime
utcTimeToPOSIXSeconds
      mono :: IO Double
mono = DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (DiffTime -> Double) -> (Time -> DiffTime) -> Time -> Double
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (\(Time DiffTime
t) -> DiffTime
t) (Time -> Double) -> IO Time -> IO Double
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
      oneHot :: Map Text Gauge.Gauge -> Text -> IO ()
      oneHot :: Map Text Gauge -> Text -> IO ()
oneHot Map Text Gauge
gauges Text
value = [(Text, Gauge)] -> ((Text, Gauge) -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ (Map Text Gauge -> [(Text, Gauge)]
forall k a. Map k a -> [(k, a)]
Map.toList Map Text Gauge
gauges) (((Text, Gauge) -> IO ()) -> IO ())
-> ((Text, Gauge) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \(Text
key, Gauge
gauge) -> Gauge -> Bool -> IO ()
bool Gauge
gauge (Text
key Text -> Text -> Bool
forall a. Eq a => a -> a -> Bool
== Text
value)
      refresh :: IO ()
refresh = do
        Double
now <- IO Double
mono
        MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx)))
 -> IO ())
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a b. (a -> b) -> a -> b
$ \Accounting HeadId (TxIdType tx)
state -> do
          let (Port
removed, Accounting HeadId (TxIdType tx)
next) = Port
-> Double
-> Double
-> Accounting HeadId (TxIdType tx)
-> (Port, Accounting HeadId (TxIdType tx))
forall txid head.
Ord txid =>
Port
-> Double
-> Double
-> Accounting head txid
-> (Port, Accounting head txid)
Accounting.prune Port
100000 Double
3600 Double
now Accounting HeadId (TxIdType tx)
state
          Counter -> Port -> IO ()
add Counter
evictedTimings Port
removed
          Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Accounting HeadId (TxIdType tx)
next
      clearPending :: IO ()
clearPending = MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting (Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
 -> IO (Accounting HeadId (TxIdType tx)))
-> (Accounting HeadId (TxIdType tx)
    -> Accounting HeadId (TxIdType tx))
-> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Accounting HeadId (TxIdType tx) -> Accounting HeadId (TxIdType tx)
forall head txid. Accounting head txid -> Accounting head txid
Accounting.clearPending)
      resetHead :: IO ()
resetHead =
        MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx)))
 -> IO ())
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a b. (a -> b) -> a -> b
$ \Accounting HeadId (TxIdType tx)
state -> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx) -> Accounting HeadId (TxIdType tx)
forall head txid. Accounting head txid -> Accounting head txid
Accounting.clearPending Accounting HeadId (TxIdType tx)
state){Accounting.confirmed = Nothing}
      seed :: NodeState tx -> IO ()
seed NodeState tx
ns = do
        let (Maybe (Snapshot tx)
confirmed, Maybe (Snapshot tx)
pending) = case NodeState tx -> HeadState tx
forall tx. NodeState tx -> HeadState tx
NodeState.headState NodeState tx
ns of
              Head.Open Head.OpenState{$sel:coordinatedHeadState:OpenState :: forall tx. OpenState tx -> CoordinatedHeadState tx
coordinatedHeadState = Head.CoordinatedHeadState{ConfirmedSnapshot tx
confirmedSnapshot :: ConfirmedSnapshot tx
$sel:confirmedSnapshot:CoordinatedHeadState :: forall tx. CoordinatedHeadState tx -> ConfirmedSnapshot tx
confirmedSnapshot, SeenSnapshot tx
seenSnapshot :: SeenSnapshot tx
$sel:seenSnapshot:CoordinatedHeadState :: forall tx. CoordinatedHeadState tx -> SeenSnapshot tx
seenSnapshot}} ->
                (Snapshot tx -> Maybe (Snapshot tx)
forall a. a -> Maybe a
Just (Snapshot tx -> Maybe (Snapshot tx))
-> Snapshot tx -> Maybe (Snapshot tx)
forall a b. (a -> b) -> a -> b
$ ConfirmedSnapshot tx -> Snapshot tx
forall tx. IsTx tx => ConfirmedSnapshot tx -> Snapshot tx
getSnapshot ConfirmedSnapshot tx
confirmedSnapshot, case SeenSnapshot tx
seenSnapshot of Head.SeenSnapshot{Snapshot tx
snapshot :: Snapshot tx
$sel:snapshot:NoSeenSnapshot :: forall tx. SeenSnapshot tx -> Snapshot tx
snapshot} -> Snapshot tx -> Maybe (Snapshot tx)
forall a. a -> Maybe a
Just Snapshot tx
snapshot; SeenSnapshot tx
_ -> Maybe (Snapshot tx)
forall a. Maybe a
Nothing)
              Head.Closed Head.ClosedState{ConfirmedSnapshot tx
confirmedSnapshot :: ConfirmedSnapshot tx
$sel:confirmedSnapshot:ClosedState :: forall tx. ClosedState tx -> ConfirmedSnapshot tx
confirmedSnapshot} -> (Snapshot tx -> Maybe (Snapshot tx)
forall a. a -> Maybe a
Just (Snapshot tx -> Maybe (Snapshot tx))
-> Snapshot tx -> Maybe (Snapshot tx)
forall a b. (a -> b) -> a -> b
$ ConfirmedSnapshot tx -> Snapshot tx
forall tx. IsTx tx => ConfirmedSnapshot tx -> Snapshot tx
getSnapshot ConfirmedSnapshot tx
confirmedSnapshot, Maybe (Snapshot tx)
forall a. Maybe a
Nothing)
              Head.FanoutProgress Head.PartialFanoutState{ConfirmedSnapshot tx
confirmedSnapshot :: ConfirmedSnapshot tx
$sel:confirmedSnapshot:PartialFanoutState :: forall tx. PartialFanoutState tx -> ConfirmedSnapshot tx
confirmedSnapshot} -> (Snapshot tx -> Maybe (Snapshot tx)
forall a. a -> Maybe a
Just (Snapshot tx -> Maybe (Snapshot tx))
-> Snapshot tx -> Maybe (Snapshot tx)
forall a b. (a -> b) -> a -> b
$ ConfirmedSnapshot tx -> Snapshot tx
forall tx. IsTx tx => ConfirmedSnapshot tx -> Snapshot tx
getSnapshot ConfirmedSnapshot tx
confirmedSnapshot, Maybe (Snapshot tx)
forall a. Maybe a
Nothing)
              Head.Idle{} -> (Maybe (Snapshot tx)
forall a. Maybe a
Nothing, Maybe (Snapshot tx)
forall a. Maybe a
Nothing)
        MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx)))
 -> IO ())
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a b. (a -> b) -> a -> b
$ \Accounting HeadId (TxIdType tx)
state ->
          Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
 -> IO (Accounting HeadId (TxIdType tx)))
-> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a b. (a -> b) -> a -> b
$
            (Accounting HeadId (TxIdType tx) -> Accounting HeadId (TxIdType tx)
forall head txid. Accounting head txid -> Accounting head txid
Accounting.clearPending Accounting HeadId (TxIdType tx)
state)
              { Accounting.confirmed = snapshotKey <$> confirmed
              , Accounting.inFlight = (\Snapshot tx
snapshot -> (Snapshot tx -> SnapshotKey HeadId
forall tx. Snapshot tx -> SnapshotKey HeadId
snapshotKey Snapshot tx
snapshot, Maybe Double
forall a. Maybe a
Nothing, 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
<$> Snapshot tx -> [tx]
forall tx. Snapshot tx -> [tx]
Hydra.Tx.confirmed Snapshot tx
snapshot)) <$> pending
              }
      confirm :: Maybe (Snapshot tx) -> IO Bool
confirm Maybe (Snapshot tx)
supplied = do
        Double
now <- IO Double
mono
        MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx), Bool))
-> IO Bool
forall a b. MVar a -> (a -> IO (a, b)) -> IO b
MVar.modifyMVar MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx), Bool))
 -> IO Bool)
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx), Bool))
-> IO Bool
forall a b. (a -> b) -> a -> b
$ \Accounting HeadId (TxIdType tx)
state -> do
          let candidate :: Maybe (SnapshotKey HeadId, [TxIdType tx], Text)
candidate = case Maybe (Snapshot tx)
supplied of
                Just Snapshot tx
snapshot -> (SnapshotKey HeadId, [TxIdType tx], Text)
-> Maybe (SnapshotKey HeadId, [TxIdType tx], Text)
forall a. a -> Maybe a
Just (Snapshot tx -> SnapshotKey HeadId
forall tx. Snapshot tx -> SnapshotKey HeadId
snapshotKey Snapshot tx
snapshot, 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
<$> Snapshot tx -> [tx]
forall tx. Snapshot tx -> [tx]
Hydra.Tx.confirmed Snapshot tx
snapshot, Text
"adopted")
                Maybe (Snapshot tx)
Nothing -> (\(SnapshotKey HeadId
key, Maybe Double
_, [TxIdType tx]
ids) -> (SnapshotKey HeadId
key, [TxIdType tx]
ids, Text
"signed")) ((SnapshotKey HeadId, Maybe Double, [TxIdType tx])
 -> (SnapshotKey HeadId, [TxIdType tx], Text))
-> Maybe (SnapshotKey HeadId, Maybe Double, [TxIdType tx])
-> Maybe (SnapshotKey HeadId, [TxIdType tx], Text)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Accounting HeadId (TxIdType tx)
-> Maybe (SnapshotKey HeadId, Maybe Double, [TxIdType tx])
forall head txid.
Accounting head txid
-> Maybe (SnapshotKey head, Maybe Double, [txid])
Accounting.inFlight Accounting HeadId (TxIdType tx)
state
          case Maybe (SnapshotKey HeadId, [TxIdType tx], Text)
candidate of
            Maybe (SnapshotKey HeadId, [TxIdType tx], Text)
Nothing -> (Accounting HeadId (TxIdType tx), Bool)
-> IO (Accounting HeadId (TxIdType tx), Bool)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
state, Bool
False)
            Just (SnapshotKey HeadId
key, [TxIdType tx]
ids, Text
source) -> do
              let (Bool
newNumber, Maybe Double
duration, [Double]
txDurations, Accounting HeadId (TxIdType tx)
next) = Double
-> SnapshotKey HeadId
-> [TxIdType tx]
-> Accounting HeadId (TxIdType tx)
-> (Bool, Maybe Double, [Double], Accounting HeadId (TxIdType tx))
forall head txid.
(Eq head, Ord txid) =>
Double
-> SnapshotKey head
-> [txid]
-> Accounting head txid
-> (Bool, Maybe Double, [Double], Accounting head txid)
Accounting.confirmSnapshot Double
now SnapshotKey HeadId
key [TxIdType tx]
ids Accounting HeadId (TxIdType tx)
state
              Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
newNumber (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                Counter -> Port -> IO ()
add Counter
headConfirmedTx ([TxIdType tx] -> Port
forall a. [a] -> Port
forall (t :: * -> *) a. Foldable t => t a -> Port
length [TxIdType tx]
ids)
                Counter -> IO ()
tick (Counter -> IO ()) -> Counter -> IO ()
forall a b. (a -> b) -> a -> b
$ Map Text Counter
snapshots Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
source
                (Double -> IO ()) -> Maybe Double -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Histogram -> Double -> IO ()
observe Histogram
snapshotConfirmationTime) Maybe Double
duration
                (Double -> IO ()) -> [Double] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (Histogram -> Double -> IO ()
observe Histogram
txConfirmationTime) [Double]
txDurations
              (Accounting HeadId (TxIdType tx), Bool)
-> IO (Accounting HeadId (TxIdType tx), Bool)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
next, Bool
newNumber)
      stateChange :: StateChanged tx -> IO ()
stateChange = \case
        PeerConnected{Host
peer :: Host
$sel:peer:NetworkConnected :: forall tx. StateChanged tx -> Host
peer} -> do
          Set Host
peers <- IORef (Set Host)
-> (Set Host -> (Set Host, Set Host)) -> IO (Set Host)
forall (m :: * -> *) a b.
MonadIO m =>
IORef a -> (a -> (a, b)) -> m b
atomicModifyIORef' IORef (Set Host)
peerSet ((Set Host -> (Set Host, Set Host)) -> IO (Set Host))
-> (Set Host -> (Set Host, Set Host)) -> IO (Set Host)
forall a b. (a -> b) -> a -> b
$ \Set Host
peers -> let next :: Set Host
next = Host -> Set Host -> Set Host
forall a. Ord a => a -> Set a -> Set a
Set.insert Host
peer Set Host
peers in (Set Host
next, Set Host
next)
          Gauge -> Double -> IO ()
set Gauge
peersConnected (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ Set Host -> Port
forall a. Set a -> Port
Set.size Set Host
peers)
        PeerDisconnected{Host
$sel:peer:NetworkConnected :: forall tx. StateChanged tx -> Host
peer :: Host
peer} -> do
          Set Host
peers <- IORef (Set Host)
-> (Set Host -> (Set Host, Set Host)) -> IO (Set Host)
forall (m :: * -> *) a b.
MonadIO m =>
IORef a -> (a -> (a, b)) -> m b
atomicModifyIORef' IORef (Set Host)
peerSet ((Set Host -> (Set Host, Set Host)) -> IO (Set Host))
-> (Set Host -> (Set Host, Set Host)) -> IO (Set Host)
forall a b. (a -> b) -> a -> b
$ \Set Host
peers -> let next :: Set Host
next = Host -> Set Host -> Set Host
forall a. Ord a => a -> Set a -> Set a
Set.delete Host
peer Set Host
peers in (Set Host
next, Set Host
next)
          Gauge -> Double -> IO ()
set Gauge
peersConnected (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ Set Host -> Port
forall a. Set a -> Port
Set.size Set Host
peers)
        StateChanged tx
NetworkDisconnected -> IORef (Set Host) -> Set Host -> IO ()
forall (m :: * -> *) a. MonadIO m => IORef a -> a -> m ()
writeIORef IORef (Set Host)
peerSet Set Host
forall a. Monoid a => a
mempty IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Gauge -> Double -> IO ()
set Gauge
peersConnected Double
0
        TickObserved{UTCTime
chainTime :: UTCTime
$sel:chainTime:NetworkConnected :: forall tx. StateChanged tx -> UTCTime
chainTime} -> do
          UTCTime
now <- IO UTCTime
forall (m :: * -> *). MonadTime m => m UTCTime
getCurrentTime
          Gauge -> Double -> IO ()
set Gauge
chainDriftSeconds (NominalDiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac (NominalDiffTime -> Double) -> NominalDiffTime -> Double
forall a b. (a -> b) -> a -> b
$ UTCTime
now UTCTime -> UTCTime -> NominalDiffTime
`diffUTCTime` UTCTime
chainTime)
          Gauge -> IO ()
timestamp Gauge
chainLastBlockTimestampSeconds
        HeadOpened{HeadId
headId :: HeadId
$sel:headId:NetworkConnected :: forall tx. StateChanged tx -> HeadId
headId} -> do
          IO ()
resetHead
          MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx)))
 -> IO ())
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a b. (a -> b) -> a -> b
$ \Accounting HeadId (TxIdType tx)
state -> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Accounting HeadId (TxIdType tx)
state{Accounting.confirmed = Just $ Accounting.SnapshotKey headId 0 0}
        HeadClosed{} -> IO ()
clearPending
        HeadFannedOut{} -> IO ()
resetHead
        LocalStateCleared{} -> IO ()
clearPending
        Checkpoint{NodeState tx
state :: NodeState tx
$sel:state:NetworkConnected :: forall tx. StateChanged tx -> NodeState tx
state} -> NodeState tx -> IO ()
seed NodeState tx
state
        SnapshotRequested{Snapshot tx
requestedSnapshot :: Snapshot tx
$sel:requestedSnapshot:NetworkConnected :: forall tx. StateChanged tx -> Snapshot tx
requestedSnapshot} -> do
          Double
now <- IO Double
mono
          MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx)))
 -> IO ())
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a b. (a -> b) -> a -> b
$ Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
 -> IO (Accounting HeadId (TxIdType tx)))
-> (Accounting HeadId (TxIdType tx)
    -> Accounting HeadId (TxIdType tx))
-> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SnapshotKey HeadId
-> Maybe Double
-> [TxIdType tx]
-> Accounting HeadId (TxIdType tx)
-> Accounting HeadId (TxIdType tx)
forall head txid.
Eq head =>
SnapshotKey head
-> Maybe Double
-> [txid]
-> Accounting head txid
-> Accounting head txid
Accounting.startSnapshot (Snapshot tx -> SnapshotKey HeadId
forall tx. Snapshot tx -> SnapshotKey HeadId
snapshotKey Snapshot tx
requestedSnapshot) (Double -> Maybe Double
forall a. a -> Maybe a
Just Double
now) (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
<$> Snapshot tx -> [tx]
forall tx. Snapshot tx -> [tx]
Hydra.Tx.confirmed Snapshot tx
requestedSnapshot)
        SnapshotConfirmed{Maybe (Snapshot tx)
snapshot :: Maybe (Snapshot tx)
$sel:snapshot:NetworkConnected :: forall tx. StateChanged tx -> Maybe (Snapshot tx)
snapshot} -> IO Bool -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ Maybe (Snapshot tx) -> IO Bool
confirm Maybe (Snapshot tx)
snapshot
        TxInvalid{tx
transaction :: tx
$sel:transaction:NetworkConnected :: forall tx. StateChanged tx -> tx
transaction} -> do
          MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting (Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
 -> IO (Accounting HeadId (TxIdType tx)))
-> (Accounting HeadId (TxIdType tx)
    -> Accounting HeadId (TxIdType tx))
-> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TxIdType tx
-> Accounting HeadId (TxIdType tx)
-> Accounting HeadId (TxIdType tx)
forall txid head.
Ord txid =>
txid -> Accounting head txid -> Accounting head txid
Accounting.forget (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
transaction))
          Counter -> IO ()
tick (Counter -> IO ()) -> Counter -> IO ()
forall a b. (a -> b) -> a -> b
$ Map Text Counter
txOutcomes Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
"invalid"
        TransactionAppliedToLocalUTxO{} -> Counter -> IO ()
tick (Counter -> IO ()) -> Counter -> IO ()
forall a b. (a -> b) -> a -> b
$ Map Text Counter
txOutcomes Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
"applied"
        DecommitInvalid{} -> Counter -> IO ()
tick (Map Text Counter
txOutcomes Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
"decommit_invalid")
        StateChanged tx
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      transactionRefused :: Text -> Client.ClientInput tx -> IO ()
      transactionRefused :: Text -> ClientInput tx -> IO ()
transactionRefused Text
outcome ClientInput tx
input = Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (ClientInput tx -> Bool
forall tx. ClientInput tx -> Bool
isTransactionInput ClientInput tx
input) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Counter -> IO ()
tick (Map Text Counter
txOutcomes Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
outcome)
      effect :: Effect tx -> IO ()
effect = \case
        ClientEffect Output.RejectedInputBecauseBroadcastStalled{ClientInput tx
clientInput :: ClientInput tx
$sel:clientInput:CommandFailed :: forall tx. ClientMessage tx -> ClientInput tx
clientInput} -> Counter -> IO ()
tick Counter
inputsRefusedBroadcastStalled IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Text -> ClientInput tx -> IO ()
transactionRefused Text
"refused_stalled" ClientInput tx
clientInput
        ClientEffect Output.RejectedInputBecauseUnsynced{ClientInput tx
$sel:clientInput:CommandFailed :: forall tx. ClientMessage tx -> ClientInput tx
clientInput :: ClientInput tx
clientInput} -> Text -> ClientInput tx -> IO ()
transactionRefused Text
"refused_unsynced" ClientInput tx
clientInput
        ClientEffect Output.CommandFailed{ClientInput tx
$sel:clientInput:CommandFailed :: forall tx. ClientMessage tx -> ClientInput tx
clientInput :: ClientInput tx
clientInput} -> Text -> ClientInput tx -> IO ()
transactionRefused Text
"rejected_state" ClientInput tx
clientInput
        Effect tx
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      networkEvent :: Text -> IO ()
networkEvent Text
event = Counter -> IO ()
tick (Counter -> IO ()) -> Counter -> IO ()
forall a b. (a -> b) -> a -> b
$ Map Text Counter
networkEvents Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
event
      handle :: HydraLog tx -> IO ()
handle = \case
        Node LoadedState{NodeState tx
nodeState :: NodeState tx
$sel:nodeState:BeginInput :: forall tx. HydraNodeLog tx -> NodeState tx
nodeState} -> NodeState tx -> IO ()
seed NodeState tx
nodeState
        Node BeginInput{Word64
inputId :: Word64
$sel:inputId:BeginInput :: forall tx. HydraNodeLog tx -> Word64
inputId, Input tx
input :: Input tx
$sel:input:BeginInput :: forall tx. HydraNodeLog tx -> Input tx
input} -> do
          Double
now <- IO Double
mono
          IORef (Maybe (Double, Text)) -> Maybe (Double, Text) -> IO ()
forall (m :: * -> *) a. MonadIO m => IORef a -> a -> m ()
writeIORef IORef (Maybe (Double, Text))
inputStarted (Maybe (Double, Text) -> IO ()) -> Maybe (Double, Text) -> IO ()
forall a b. (a -> b) -> a -> b
$ (Double, Text) -> Maybe (Double, Text)
forall a. a -> Maybe a
Just (Double
now, Input tx -> Text
forall tx. Input tx -> Text
inputSource Input tx
input)
          case Input tx
input of
            NetworkInput Natural
_ ReceivedMessage{$sel:msg:ConnectivityEvent :: forall msg. NetworkEvent msg -> msg
msg = ReqTx tx
tx} -> MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting ((Accounting HeadId (TxIdType tx)
  -> IO (Accounting HeadId (TxIdType tx)))
 -> IO ())
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a b. (a -> b) -> a -> b
$ \Accounting HeadId (TxIdType tx)
state -> do
              let (Bool
fresh, Port
removed, Accounting HeadId (TxIdType tx)
next) = Port
-> Double
-> Double
-> Word64
-> TxIdType tx
-> Accounting HeadId (TxIdType tx)
-> (Bool, Port, Accounting HeadId (TxIdType tx))
forall txid head.
Ord txid =>
Port
-> Double
-> Double
-> Word64
-> txid
-> Accounting head txid
-> (Bool, Port, Accounting head txid)
Accounting.request Port
100000 Double
3600 Double
now Word64
inputId (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
tx) Accounting HeadId (TxIdType tx)
state
              Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
fresh (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Counter -> IO ()
tick Counter
headRequestedTx
              Counter -> Port -> IO ()
add Counter
evictedTimings Port
removed
              Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Accounting HeadId (TxIdType tx)
next
            Input tx
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Node EndInput{} -> do
          Counter -> IO ()
tick Counter
headInputs
          Double
now <- IO Double
mono
          IORef (Maybe (Double, Text)) -> IO (Maybe (Double, Text))
forall (m :: * -> *) a. MonadIO m => IORef a -> m a
readIORef IORef (Maybe (Double, Text))
inputStarted IO (Maybe (Double, Text))
-> (Maybe (Double, Text) -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= ((Double, Text) -> IO ()) -> Maybe (Double, Text) -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (\(Double
start, Text
source) -> Histogram -> Double -> IO ()
observe (Map Text Histogram
inputProcessing Map Text Histogram -> Text -> Histogram
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
source) (Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (Double -> Double) -> Double -> Double
forall a b. (a -> b) -> a -> b
$ Double
now Double -> Double -> Double
forall a. Num a => a -> a -> a
- Double
start))
          IORef (Maybe (Double, Text)) -> Maybe (Double, Text) -> IO ()
forall (m :: * -> *) a. MonadIO m => IORef a -> a -> m ()
writeIORef IORef (Maybe (Double, Text))
inputStarted Maybe (Double, Text)
forall a. Maybe a
Nothing
        Node InputDequeued{DiffTime
queueWait :: DiffTime
$sel:queueWait:BeginInput :: forall tx. HydraNodeLog tx -> DiffTime
queueWait} -> Histogram -> Double -> IO ()
observe Histogram
inputQueueWait (DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
queueWait)
        Node LogicOutcome{Outcome tx
outcome :: Outcome tx
$sel:outcome:BeginInput :: forall tx. HydraNodeLog tx -> Outcome tx
outcome} -> case Outcome tx
outcome of
          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} -> (StateChanged tx -> IO ()) -> [StateChanged tx] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ StateChanged tx -> IO ()
stateChange [StateChanged tx]
stateChanges IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> (Effect tx -> IO ()) -> [Effect tx] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ Effect tx -> IO ()
effect [Effect tx]
effects
          Wait{WaitReason tx
reason :: WaitReason tx
$sel:reason:Continue :: forall tx. Outcome tx -> WaitReason tx
reason, [StateChanged tx]
$sel:stateChanges:Continue :: forall tx. Outcome tx -> [StateChanged tx]
stateChanges :: [StateChanged tx]
stateChanges} -> Counter -> IO ()
tick (Map Text Counter
inputRetries Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` WaitReason tx -> Text
forall tx. WaitReason tx -> Text
waitReason WaitReason tx
reason) IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> (StateChanged tx -> IO ()) -> [StateChanged tx] -> IO ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ StateChanged tx -> IO ()
stateChange [StateChanged tx]
stateChanges
          Error{} -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Node DroppedFromQueue{Input tx
$sel:input:BeginInput :: forall tx. HydraNodeLog tx -> Input tx
input :: Input tx
input} -> do
          Counter -> IO ()
tick (Counter -> IO ()) -> Counter -> IO ()
forall a b. (a -> b) -> a -> b
$ Map Text Counter
inputDrops Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Input tx -> Text
forall tx. Input tx -> Text
inputSource Input tx
input
          case Input tx
input of
            NetworkInput Natural
_ ReceivedMessage{$sel:msg:ConnectivityEvent :: forall msg. NetworkEvent msg -> msg
msg = ReqTx tx
tx} -> MVar (Accounting HeadId (TxIdType tx))
-> (Accounting HeadId (TxIdType tx)
    -> IO (Accounting HeadId (TxIdType tx)))
-> IO ()
forall a. MVar a -> (a -> IO a) -> IO ()
MVar.modifyMVar_ MVar (Accounting HeadId (TxIdType tx))
accounting (Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Accounting HeadId (TxIdType tx)
 -> IO (Accounting HeadId (TxIdType tx)))
-> (Accounting HeadId (TxIdType tx)
    -> Accounting HeadId (TxIdType tx))
-> Accounting HeadId (TxIdType tx)
-> IO (Accounting HeadId (TxIdType tx))
forall b c a. (b -> c) -> (a -> b) -> a -> c
. TxIdType tx
-> Accounting HeadId (TxIdType tx)
-> Accounting HeadId (TxIdType tx)
forall txid head.
Ord txid =>
txid -> Accounting head txid -> Accounting head txid
Accounting.forget (tx -> TxIdType tx
forall tx. IsTx tx => tx -> TxIdType tx
txId tx
tx))
            Input tx
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        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} -> Gauge -> Double -> IO ()
set Gauge
pendingBroadcasts (Natural -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Natural
pending) IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Gauge -> Double -> IO ()
set Gauge
broadcastNoProgressSeconds (DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
noProgressSeconds)
        Node BroadcastHealth{Maybe StallReason
stalledReason :: Maybe StallReason
$sel:stalledReason:BeginInput :: forall tx. HydraNodeLog tx -> Maybe StallReason
stalledReason} -> do
          Gauge -> Bool -> IO ()
bool Gauge
broadcastStalled (Maybe StallReason -> Bool
forall a. Maybe a -> Bool
isJust Maybe StallReason
stalledReason)
          Gauge -> Bool -> IO ()
bool Gauge
broadcastStalledNoProgress (Maybe StallReason
stalledReason Maybe StallReason -> Maybe StallReason -> Bool
forall a. Eq a => a -> a -> Bool
== StallReason -> Maybe StallReason
forall a. a -> Maybe a
Just StallReason
NoProgress)
          Gauge -> Bool -> IO ()
bool Gauge
broadcastStalledBacklogFull (Maybe StallReason
stalledReason Maybe StallReason -> Maybe StallReason -> Bool
forall a. Eq a => a -> a -> Bool
== StallReason -> Maybe StallReason
forall a. a -> Maybe a
Just StallReason
BacklogFull)
        Node NodeHealth{NodeStats
stats :: NodeStats
$sel:stats:BeginInput :: forall tx. HydraNodeLog tx -> NodeStats
stats} -> do
          Gauge -> Bool -> IO ()
bool Gauge
synced (NodeStats -> Bool
Stats.synchronized NodeStats
stats)
          Map Text Gauge -> Text -> IO ()
oneHot Map Text Gauge
phase (NodeStats -> Text
Stats.phase NodeStats
stats)
          Gauge -> Double -> IO ()
set Gauge
queuedInputs (Natural -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Natural -> Double) -> Natural -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Natural
Stats.queuedInputs NodeStats
stats)
          Gauge -> Double -> IO ()
set Gauge
parkedInputs (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.parkedInputs NodeStats
stats)
          Gauge -> Double -> IO ()
set Gauge
snapshotNumber (Natural -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Natural -> Double) -> Natural -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Natural
Stats.confirmedNumber NodeStats
stats)
          Gauge -> Bool -> IO ()
bool Gauge
snapshotPending (NodeStats -> Bool
Stats.snapshotPending NodeStats
stats)
          Gauge -> Double -> IO ()
set Gauge
missingSignatures (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.missingSignatures NodeStats
stats)
          Gauge -> Double -> IO ()
set Gauge
outstandingTransactions (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.pendingTransactions NodeStats
stats)
          Gauge -> Double -> IO ()
set (Map Text Gauge
deposits Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
"inactive") (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.inactiveDeposits NodeStats
stats)
          Gauge -> Double -> IO ()
set (Map Text Gauge
deposits Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
"active") (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.activeDeposits NodeStats
stats)
          Gauge -> Double -> IO ()
set (Map Text Gauge
deposits Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
"expired") (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.expiredDeposits NodeStats
stats)
          Map Text Gauge -> Text -> IO ()
oneHot Map Text Gauge
decommit (NodeStats -> Text
Stats.decommitState NodeStats
stats)
          Gauge -> Double -> IO ()
set Gauge
fanoutRemaining (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Port -> Double) -> Port -> Double
forall a b. (a -> b) -> a -> b
$ NodeStats -> Port
Stats.fanoutRemainingOutputs NodeStats
stats)
          Map Text Gauge -> Text -> IO ()
oneHot Map Text Gauge
fanoutMode (NodeStats -> Text
Stats.fanoutMode NodeStats
stats)
        Node ChainPostStarted{Text
operation :: Text
$sel:operation:BeginInput :: forall tx. HydraNodeLog tx -> Text
operation} -> Gauge -> IO ()
Gauge.inc (Map Text Gauge
chainInflight Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation)
        Node ChainPostFinished{Text
$sel:operation:BeginInput :: forall tx. HydraNodeLog tx -> Text
operation :: Text
operation, Bool
accepted :: Bool
$sel:accepted:BeginInput :: forall tx. HydraNodeLog tx -> Bool
accepted, DiffTime
duration :: DiffTime
$sel:duration:BeginInput :: forall tx. HydraNodeLog tx -> DiffTime
duration} -> do
          Gauge -> IO ()
Gauge.dec (Map Text Gauge
chainInflight Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation)
          Histogram -> Double -> IO ()
observe (Map Text Histogram
chainDuration Map Text Histogram -> Text -> Histogram
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation) (DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
duration)
          Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless Bool
accepted (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Counter -> IO ()
tick (Counter -> IO ()) -> Counter -> IO ()
forall a b. (a -> b) -> a -> b
$ Map Text Counter
chainFailures Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation
        Network (Network.Authenticate AuthLog
_) -> Text -> IO ()
networkEvent Text
"authentication_failure"
        Network (Network.Etcd EtcdLog
message) -> case EtcdLog
message of
          Etcd.PersistentQueueStats{Natural
depth :: Natural
$sel:depth:EtcdLog :: EtcdLog -> Natural
depth, Double
oldestAgeSeconds :: Double
$sel:oldestAgeSeconds:EtcdLog :: EtcdLog -> Double
oldestAgeSeconds} -> Gauge -> Double -> IO ()
set Gauge
persistentDepth (Natural -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Natural
depth) IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Gauge -> Double -> IO ()
set Gauge
persistentAge Double
oldestAgeSeconds
          Etcd.BroadcastCompleted{DiffTime
duration :: DiffTime
$sel:duration:EtcdLog :: EtcdLog -> DiffTime
duration} -> Histogram -> Double -> IO ()
observe Histogram
putLatency (DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
duration)
          Etcd.NetworkConnectivity{$sel:connected:EtcdLog :: EtcdLog -> Bool
connected = Bool
online} -> Gauge -> Bool -> IO ()
bool Gauge
networkConnected Bool
online
          EtcdLog
Etcd.Reconnecting -> Text -> IO ()
networkEvent Text
"reconnect"
          Etcd.BroadcastFailed{} -> Text -> IO ()
networkEvent Text
"broadcast_failure"
          Etcd.WatchFailed{} -> Text -> IO ()
networkEvent Text
"watch_failure"
          Etcd.FailedToDecodeValue{} -> Text -> IO ()
networkEvent Text
"decode_failure"
          Etcd.FailedToDecodeLog{} -> Text -> IO ()
networkEvent Text
"decode_failure"
          Etcd.PersistentQueueStatFailed{} -> Text -> IO ()
networkEvent Text
"queue_stat_failure"
          Etcd.PersistentQueueLoadFailed{} -> Text -> IO ()
networkEvent Text
"queue_storage_failure"
          Etcd.PersistentQueueWriteFailed{} -> Text -> IO ()
networkEvent Text
"queue_storage_failure"
          Etcd.PersistentQueueDeleteFailed{} -> Text -> IO ()
networkEvent Text
"queue_storage_failure"
          EtcdLog
Etcd.PersistentQueueFull -> Text -> IO ()
networkEvent Text
"queue_full"
          Etcd.LowLeaseTTL{} -> Text -> IO ()
networkEvent Text
"lease_warning"
          EtcdLog
Etcd.NoKeepAliveResponse -> Text -> IO ()
networkEvent Text
"lease_warning"
          EtcdLog
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        APIServer APIServerLog
message -> case APIServerLog
message of
          API.APIL1Started{Text
operation :: Text
$sel:operation:APIServerStarted :: APIServerLog -> Text
operation} -> Gauge -> IO ()
Gauge.inc (Map Text Gauge
chainInflight Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation)
          API.APIL1Finished{Text
$sel:operation:APIServerStarted :: APIServerLog -> Text
operation :: Text
operation, Bool
successful :: Bool
$sel:successful:APIServerStarted :: APIServerLog -> Bool
successful, DiffTime
duration :: DiffTime
$sel:duration:APIServerStarted :: APIServerLog -> DiffTime
duration} -> do
            Gauge -> IO ()
Gauge.dec (Map Text Gauge
chainInflight Map Text Gauge -> Text -> Gauge
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation)
            Histogram -> Double -> IO ()
observe (Map Text Histogram
chainDuration Map Text Histogram -> Text -> Histogram
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation) (DiffTime -> Double
forall a b. (Real a, Fractional b) => a -> b
realToFrac DiffTime
duration)
            Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless Bool
successful (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Counter -> IO ()
tick (Map Text Counter
chainFailures Map Text Counter -> Text -> Counter
forall key a. (Ord key, Show key) => Map key a -> key -> a
`at` Text
operation)
          APIServerLog
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        DirectChain CardanoChainLog
message -> case CardanoChainLog
message of
          ChainLog.RolledForward{ChainPoint
point :: ChainPoint
$sel:point:ToPost :: CardanoChainLog -> ChainPoint
point} -> Gauge -> IO ()
timestamp Gauge
chainBackendTimestamp IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IORef Double -> Double -> IO ()
forall (m :: * -> *) a. MonadIO m => IORef a -> a -> m ()
writeIORef IORef Double
backendSlot (ChainPoint -> Double
pointSlot ChainPoint
point)
          ChainLog.RolledBackward{ChainPoint
$sel:point:ToPost :: CardanoChainLog -> ChainPoint
point :: ChainPoint
point} -> do
            Double
previous <- IORef Double -> (Double -> (Double, Double)) -> IO Double
forall (m :: * -> *) a b.
MonadIO m =>
IORef a -> (a -> (a, b)) -> m b
atomicModifyIORef' IORef Double
backendSlot (ChainPoint -> Double
pointSlot ChainPoint
point,)
            Histogram -> Double -> IO ()
observe Histogram
rollbackDepth (Double -> Double -> Double
forall a. Ord a => a -> a -> a
max Double
0 (Double -> Double) -> Double -> Double
forall a b. (a -> b) -> a -> b
$ Double
previous Double -> Double -> Double
forall a. Num a => a -> a -> a
- ChainPoint -> Double
pointSlot ChainPoint
point)
          ChainLog.BlockfrostTransientError{} -> Counter -> IO ()
tick Counter
blockfrostRetries
          CardanoChainLog
ChainLog.BlockfrostRateLimit -> Counter -> IO ()
tick Counter
blockfrostRateLimits
          CardanoChainLog
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        SQLite SQLiteLog
message -> case SQLiteLog
message of
          SQLiteStats{Port
queueDepth :: Port
queueDepth :: SQLiteLog -> Port
queueDepth} -> Gauge -> Double -> IO ()
set Gauge
sqliteQueue (Port -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Port
queueDepth)
          SQLiteLog
SQLiteWriteStarted -> Gauge -> Double -> IO ()
set Gauge
sqliteInflight Double
1
          SQLiteWriteCompleted{Double
writeSeconds :: Double
writeSeconds :: SQLiteLog -> Double
writeSeconds} -> Gauge -> Double -> IO ()
set Gauge
sqliteInflight Double
0 IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Histogram -> Double -> IO ()
observe Histogram
sqliteDuration Double
writeSeconds
          SQLiteLog
SQLiteWriteFailed -> Gauge -> Double -> IO ()
set Gauge
sqliteInflight Double
0
          SQLiteLog
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        HydraLog tx
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  (HydraLog tx -> IO (), Registry, IO ())
-> IO (HydraLog tx -> IO (), Registry, IO ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (HydraLog tx -> IO ()
handle, Registry
registry, IO ()
refresh)

snapshotKey :: Snapshot tx -> Accounting.SnapshotKey Hydra.Tx.HeadId
snapshotKey :: forall tx. Snapshot tx -> SnapshotKey HeadId
snapshotKey Snapshot{HeadId
headId :: HeadId
$sel:headId:Snapshot :: forall tx. Snapshot tx -> HeadId
headId, SnapshotVersion
version :: SnapshotVersion
$sel:version:Snapshot :: forall tx. Snapshot tx -> SnapshotVersion
version, SnapshotNumber
number :: SnapshotNumber
$sel:number:Snapshot :: forall tx. Snapshot tx -> SnapshotNumber
number} = HeadId -> Natural -> Natural -> SnapshotKey HeadId
forall head. head -> Natural -> Natural -> SnapshotKey head
Accounting.SnapshotKey HeadId
headId (SnapshotVersion -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral SnapshotVersion
version) (SnapshotNumber -> Natural
forall a b. (Integral a, Num b) => a -> b
fromIntegral SnapshotNumber
number)

-- | Only inputs carrying a transaction contribute to transaction refusals.
isTransactionInput :: Client.ClientInput tx -> Bool
isTransactionInput :: forall tx. ClientInput tx -> Bool
isTransactionInput = \case
  Client.NewTx{} -> Bool
True
  Client.Decommit{} -> Bool
True
  ClientInput tx
_ -> Bool
False

inputSource :: Input tx -> Text
inputSource :: forall tx. Input tx -> Text
inputSource = \case ClientInput{} -> Text
"client"; NetworkInput{} -> Text
"network"; ChainInput{} -> Text
"chain"

waitReason :: WaitReason tx -> Text
waitReason :: forall tx. WaitReason tx -> Text
waitReason = \case
  WaitOnNotApplicableTx{} -> Text
"transaction"
  WaitOnSnapshotNumber{} -> Text
"snapshot_number"
  WaitOnSnapshotVersion{} -> Text
"snapshot_version"
  WaitReason tx
WaitOnSeenSnapshot -> Text
"seen_snapshot"
  WaitOnTxs{} -> Text
"transactions"
  WaitReason tx
WaitOnContestationDeadline -> Text
"contestation_deadline"
  WaitOnNotApplicableDecommitTx{} -> Text
"decommit"
  WaitOnUnresolvedCommit{} -> Text
"unresolved_commit"
  WaitOnUnresolvedDecommit{} -> Text
"unresolved_decommit"
  WaitOnDepositObserved{} -> Text
"deposit_observation"
  WaitOnDepositActivation{} -> Text
"deposit_activation"
  WaitOnNodeInSync{} -> Text
"sync"

pointSlot :: ChainPoint -> Double
pointSlot :: ChainPoint -> Double
pointSlot = \case ChainPoint
ChainPointAtGenesis -> Double
0; ChainPoint (SlotNo Word64
slot) Hash BlockHeader
_ -> Word64 -> Double
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word64
slot

registerRtsMetrics :: Registry.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 -> Text -> IO Gauge
counter Text
"hydra_rts_allocated_bytes" Text
"Cumulative bytes allocated by the GHC runtime."
      Gauge
mutatorCpu <- Text -> Text -> IO Gauge
counter Text
"hydra_rts_mutator_cpu_seconds" Text
"Cumulative GHC mutator CPU seconds."
      Gauge
gcCpu <- Text -> Text -> IO Gauge
counter Text
"hydra_rts_gc_cpu_seconds" Text
"Cumulative GHC garbage collector CPU seconds."
      Gauge
maxLive <- Registry -> Text -> Text -> [(Text, Text)] -> IO Gauge
Registry.gauge Registry
registry Text
"hydra_rts_max_live_bytes" Text
"Maximum live heap bytes observed at a major GC since process start." []
      Gauge
cumulativeLive <- Text -> Text -> IO Gauge
counter Text
"hydra_rts_cumulative_live_bytes" Text
"Sum of live bytes measured at each major GC; divide its rate by the major GC rate for mean live heap."
      Gauge
majorGcs <- Text -> Text -> IO Gauge
counter Text
"hydra_rts_major_gcs" Text
"Cumulative major GHC garbage collections."
      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 (Word64 -> Double) -> Word64 -> Double
forall a b. (a -> b) -> a -> b
$ 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 (Word64 -> Double) -> Word64 -> Double
forall a b. (a -> b) -> a -> b
$ 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 (Word64 -> Double) -> Word64 -> Double
forall a b. (a -> b) -> a -> b
$ 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 (Word32 -> Double) -> Word32 -> Double
forall a b. (a -> b) -> a -> b
$ RTSStats -> Word32
major_gcs RTSStats
stats) Gauge
majorGcs
 where
  counter :: Text -> Text -> IO Gauge
counter Text
legacy Text
help = do
    Gauge
handle <- Registry -> Text -> Text -> IO Gauge
Registry.sampledCounter Registry
registry (Text
legacy Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_total") Text
help
    Registry -> Text -> Text -> Double -> IO ()
Registry.alias Registry
registry (Text
legacy Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"_total") Text
legacy Double
1
    Gauge -> IO Gauge
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Gauge
handle