Add counters to Pablo's integration of the defensive mempool

cardano-node/src/Cardano/Node/TraceConstraints.hs

@@ -13,6 +13,7 @@ import           Cardano.Ledger.Keys
 import           Cardano.Logging (LogFormatting)
 import           Cardano.Node.Queries (ConvertTxId, GetKESInfo (..), HasKESInfo (..),
                    HasKESMetricsData (..), LedgerQueries)
+import qualified Cardano.Node.Tracing.Tracers.Consensus as ConsensusTracers
 import           Cardano.Protocol.Crypto (StandardCrypto)
 import           Cardano.Tracing.HasIssuer (HasIssuer)
 import           Ouroboros.Consensus.Block (BlockProtocol, CannotForge, ForgeStateUpdateError,
@@ -73,4 +74,6 @@ type TraceConstraints blk =
     , LogFormatting (ForgeStateUpdateError blk)
     , LogFormatting (Set (Credential 'Staking))
     , LogFormatting (NonEmpty.NonEmpty (KeyHash 'Staking))
+
+    , ConsensusTracers.MempoolTimeoutSoftPredicate blk
     )

cardano-node/src/Cardano/Node/Tracing/Tracers/Consensus.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE LambdaCase #-}
@@ -18,6 +19,10 @@
   , calculateBlockFetchClientMetrics
   , servedBlockLatest
   , ClientMetrics
+  , txsMempoolTimeoutSoftCounterName
+  , MempoolTimeoutSoftPredicate (..)
+  , EraMempoolTimeoutSoftPredicate (..)
+  , impliesMempoolTimeoutSoft
   ) where
 
 
@@ -78,12 +83,23 @@
 import           Data.IntPSQ (IntPSQ)
 import qualified Data.IntPSQ as Pq
 import qualified Data.List as List
+import qualified Data.List.NonEmpty as NE
 import qualified Data.Text as Text
 import           Data.Time (NominalDiffTime)
 import           Data.Word (Word32, Word64)
 import           Network.TypedProtocol.Core
 
 
+-- all for MempoolTimeoutSoftPredicate
+import qualified Cardano.Ledger.Conway.Rules as Conway
+import qualified Cardano.Ledger.Core as SL (EraRule)
+import qualified Cardano.Ledger.Shelley.API as SL (ApplyTxError (..))
+import qualified Data.SOP as SOP
+import           Ouroboros.Consensus.Byron.Ledger.Block as Consensus
+import           Ouroboros.Consensus.HardFork.Combinator as Consensus
+import           Ouroboros.Consensus.Shelley.Ledger.Block as Consensus
+import           Ouroboros.Consensus.TypeFamilyWrappers as Consensus
+
 instance (LogFormatting adr, Show adr) => LogFormatting (ConnectionId adr) where
   forMachine _dtal (ConnectionId local' remote) =
     mconcat [ "connectionId" .= String (showT local'
@@ -1241,12 +1257,59 @@
 -- Mempool Tracer
 --------------------------------------------------------------------------------
 
+txsMempoolTimeoutSoftCounterName :: Text.Text
+txsMempoolTimeoutSoftCounterName = "txsMempoolTimeoutSoft"
+
+impliesMempoolTimeoutSoft ::
+  forall blk. MempoolTimeoutSoftPredicate blk => TraceEventMempool blk -> Bool
+impliesMempoolTimeoutSoft = \case
+  TraceMempoolRejectedTx _tx txApplyErr _mpSz ->
+    errImpliesMempoolTimeoutSoft (Proxy @blk) txApplyErr
+  _ -> False
+
+class MempoolTimeoutSoftPredicate blk where
+  -- | Does the error indicate a mempool timeout
+  errImpliesMempoolTimeoutSoft :: proxy blk -> ApplyTxErr blk -> Bool
+
+instance SOP.All MempoolTimeoutSoftPredicate xs => MempoolTimeoutSoftPredicate (Consensus.HardForkBlock xs) where
+  errImpliesMempoolTimeoutSoft _prx = \case
+    Consensus.HardForkApplyTxErrWrongEra{} -> False
+    Consensus.HardForkApplyTxErrFromEra (Consensus.OneEraApplyTxErr ns) ->
+      SOP.hcollapse $ SOP.hcmap (Proxy @MempoolTimeoutSoftPredicate) f ns
+    where
+      f :: forall x. MempoolTimeoutSoftPredicate x => Consensus.WrapApplyTxErr x -> SOP.K Bool x
+      f (Consensus.WrapApplyTxErr err) = SOP.K $ errImpliesMempoolTimeoutSoft (Proxy @x) err
+
+instance MempoolTimeoutSoftPredicate Consensus.ByronBlock where
+  errImpliesMempoolTimeoutSoft = \_prx _err -> False
+
+instance EraMempoolTimeoutSoftPredicate era => MempoolTimeoutSoftPredicate (Consensus.ShelleyBlock proto era) where
+  errImpliesMempoolTimeoutSoft _prx = \case
+    SL.ApplyTxError (err NE.:| errs) ->
+      null errs && eraImpliesMempoolTimeoutSoft (Proxy @era) err
+
+class EraMempoolTimeoutSoftPredicate era where
+  -- | Does the error indicate a mempool timeout
+  eraImpliesMempoolTimeoutSoft :: proxy era -> Conway.PredicateFailure (SL.EraRule "LEDGER" era) -> Bool
+
+instance EraMempoolTimeoutSoftPredicate ShelleyEra  where eraImpliesMempoolTimeoutSoft _prx _err = False
+instance EraMempoolTimeoutSoftPredicate AllegraEra  where eraImpliesMempoolTimeoutSoft _prx _err = False
+instance EraMempoolTimeoutSoftPredicate MaryEra     where eraImpliesMempoolTimeoutSoft _prx _err = False
+instance EraMempoolTimeoutSoftPredicate AlonzoEra   where eraImpliesMempoolTimeoutSoft _prx _err = False
+instance EraMempoolTimeoutSoftPredicate BabbageEra  where eraImpliesMempoolTimeoutSoft _prx _err = False
+instance EraMempoolTimeoutSoftPredicate ConwayEra   where
+  eraImpliesMempoolTimeoutSoft _prx = \case
+    Conway.ConwayMempoolFailure txt -> Text.pack "MempoolTxTooSlow" `Text.isPrefixOf` txt
+    _ -> False
+instance EraMempoolTimeoutSoftPredicate DijkstraEra where eraImpliesMempoolTimeoutSoft _prx _err = False
+
 instance
   ( LogFormatting (ApplyTxErr blk)
   , LogFormatting (GenTx blk)
   , ToJSON (GenTxId blk)
   , LedgerSupportsMempool blk
   , ConvertRawHash blk
+  , MempoolTimeoutSoftPredicate blk
   ) => LogFormatting (TraceEventMempool blk) where
   forMachine dtal (TraceMempoolAddedTx tx _mpSzBefore mpSzAfter) =
     mconcat
@@ -1311,10 +1374,14 @@
     [ IntM "txsInMempool" (fromIntegral $ msNumTxs mpSz)
     , IntM "mempoolBytes" (fromIntegral . unByteSize32 . msNumBytes $ mpSz)
     ]
-  asMetrics (TraceMempoolRejectedTx _tx _txApplyErr _ mpSz) =
+  asMetrics ev@(TraceMempoolRejectedTx _tx _txApplyErr mpSz) =
     [ IntM "txsInMempool" (fromIntegral $ msNumTxs mpSz)
     , IntM "mempoolBytes" (fromIntegral . unByteSize32 . msNumBytes $ mpSz)
     ]
+    ++
+    [ CounterM txsMempoolTimeoutSoftCounterName Nothing
+    | impliesMempoolTimeoutSoft ev
+    ]
   asMetrics (TraceMempoolRemoveTxs txs mpSz) =
     [ IntM "txsInMempool" (fromIntegral $ msNumTxs mpSz)
     , IntM "mempoolBytes" (fromIntegral . unByteSize32 . msNumBytes $ mpSz)
@@ -1370,6 +1437,7 @@
     metricsDocFor (Namespace _ ["RejectedTx"]) =
       [ ("txsInMempool","Transactions in mempool")
       , ("mempoolBytes", "Byte size of the mempool")
+      , (txsMempoolTimeoutSoftCounterName, "Transactions that soft timed out in mempool")
       ]
     metricsDocFor (Namespace _ ["RemoveTxs"]) =
       [ ("txsInMempool","Transactions in mempool")

cardano-node/src/Cardano/Node/Tracing/Tracers/Diffusion.hs

@@ -3,20 +3,26 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 
 
 {-# OPTIONS_GHC -Wno-orphans  #-}
 
 module Cardano.Node.Tracing.Tracers.Diffusion
-  () where
+  ( txsMempoolTimeoutHardCounterName
+  , impliesMempoolTimeoutHard
+  ) where
 
 
 import           Cardano.Logging
 import           Cardano.Node.Configuration.TopologyP2P ()
+import           Control.Exception (fromException)
+import           Ouroboros.Consensus.Mempool.API (ExnMempoolTimeout)
 import qualified Ouroboros.Network.Diffusion.Types as Diff
 import           Ouroboros.Network.PeerSelection.LedgerPeers (NumberOfPeers (..), PoolStake (..),
                    TraceLedgerPeers (..))
@@ -356,6 +362,16 @@ instance MetaTrace Mux.ChannelTrace where
       , Namespace [] ["ChannelSendEnd"]
       ]
 
+txsMempoolTimeoutHardCounterName :: Text
+txsMempoolTimeoutHardCounterName = "txsMempoolTimeoutHard"
+
+impliesMempoolTimeoutHard :: Mux.Trace -> Bool
+impliesMempoolTimeoutHard = \case
+  Mux.TraceExceptionExit _mid _dir e
+    | Just _ <- fromException @ExnMempoolTimeout e
+      -> True
+  _ -> False
+
 instance LogFormatting Mux.Trace where
     forMachine _dtal (Mux.TraceState new) = mconcat
       [ "kind" .= String "Mux.TraceState"
@@ -435,6 +451,24 @@ instance LogFormatting Mux.Trace where
     forHuman Mux.TraceStopping = "Mux stopping"
     forHuman Mux.TraceStopped  = "Mux stoppped"
 
+    asMetrics = \case
+      Mux.TraceState{} -> []
+      Mux.TraceCleanExit{} -> []
+      ev@Mux.TraceExceptionExit{} ->
+        -- Somewhat awkward to "catch" this Consensus exception here, but
+        -- Diffusion Layer is indeed the ultimate manager of the per-peer
+        -- threads.
+        [ CounterM txsMempoolTimeoutHardCounterName Nothing
+        | impliesMempoolTimeoutHard ev
+        ]
+      Mux.TraceStartEagerly{} -> []
+      Mux.TraceStartOnDemand{} -> []
+      Mux.TraceStartOnDemandAny{} -> []
+      Mux.TraceStartedOnDemand{} -> []
+      Mux.TraceTerminating{} -> []
+      Mux.TraceStopping{} -> []
+      Mux.TraceStopped{} -> []
+
 instance MetaTrace Mux.Trace where
     namespaceFor Mux.TraceState {}                 =
       Namespace [] ["State"]
@@ -491,6 +525,20 @@ instance MetaTrace Mux.Trace where
       "Mux shutdown."
     documentFor _ = Nothing
 
+    metricsDocFor (Namespace _ ["State"])               = []
+    metricsDocFor (Namespace _ ["CleanExit"])           = []
+    metricsDocFor (Namespace _ ["ExceptionExit"])       =
+      [ (txsMempoolTimeoutHardCounterName, "Transactions that hard timed out in mempool")
+      ]
+    metricsDocFor (Namespace _ ["StartEagerly"])        = []
+    metricsDocFor (Namespace _ ["StartOnDemand"])       = []
+    metricsDocFor (Namespace _ ["StartedOnDemand"])     = []
+    metricsDocFor (Namespace _ ["StartOnDemandAny"])    = []
+    metricsDocFor (Namespace _ ["Terminating"])         = []
+    metricsDocFor (Namespace _ ["Stopping"])            = []
+    metricsDocFor (Namespace _ ["Stopped"])             = []
+    metricsDocFor _                                     = []
+
     allNamespaces = [
         Namespace [] ["State"]
       , Namespace [] ["CleanExit"]

cardano-node/src/Cardano/Tracing/Tracers.hs

@@ -43,6 +43,8 @@ import           Cardano.Node.Startup
 import qualified Cardano.Node.STM as STM
 import           Cardano.Node.TraceConstraints
 import           Cardano.Node.Tracing
+import qualified Cardano.Node.Tracing.Tracers.Consensus as ConsensusTracers
+import qualified Cardano.Node.Tracing.Tracers.Diffusion as DiffusionTracers
 import           Cardano.Node.Tracing.Tracers.NodeVersion
 import           Cardano.Network.Diffusion (CardanoPeerSelectionCounters)
 import           Cardano.Protocol.TPraos.OCert (KESPeriod (..))
@@ -80,6 +82,8 @@ import qualified Ouroboros.Consensus.Storage.ChainDB as ChainDB
 import qualified Ouroboros.Consensus.Storage.LedgerDB as LedgerDB
 import           Ouroboros.Consensus.Util.Enclose
 
+import qualified Network.Mux as Mux
+
 import qualified Cardano.Network.Diffusion.Types as Cardano.Diffusion
 import qualified Cardano.Network.PeerSelection.Governor.Types as Cardano
 
@@ -388,7 +392,7 @@ mkTracers blockConfig tOpts@(TracingOnLegacy trSel) tr nodeKern ekgDirect = do
 
    diffusionTracers :: Cardano.Diffusion.CardanoTracers IO
    diffusionTracers = Cardano.Diffusion.Tracers
-     { Diffusion.dtMuxTracer            = muxTracer
+     { Diffusion.dtMuxTracer            = muxTracer ekgDirect trSel tr
      , Diffusion.dtChannelTracer        = channelTracer
      , Diffusion.dtBearerTracer         = bearerTracer
      , Diffusion.dtHandshakeTracer      = handshakeTracer
@@ -464,8 +468,6 @@ mkTracers blockConfig tOpts@(TracingOnLegacy trSel) tr nodeKern ekgDirect = do
      }
    verb :: TracingVerbosity
    verb = traceVerbosity trSel
-   muxTracer =
-     tracerOnOff (traceMux trSel) verb "Mux" tr
    channelTracer =
      tracerOnOff (traceMux trSel) verb "MuxChannel" tr
    bearerTracer =
@@ -537,6 +539,32 @@ mkTracers _ _ _ _ _ =
     , ledgerMetricsTracer = nullTracer
     }
 
+--------------------------------------------------------------------------------
+-- Diffusion Layer Tracers
+--------------------------------------------------------------------------------
+
+notifyTxsMempoolTimeoutHard :: Maybe EKGDirect -> Tracer IO Mux.Trace
+notifyTxsMempoolTimeoutHard mbEKGDirect = case mbEKGDirect of
+  Nothing -> nullTracer
+  Just ekgDirect -> Tracer $ \ev -> do
+    when (DiffusionTracers.impliesMempoolTimeoutHard ev) $ do
+      sendEKGDirectCounter ekgDirect DiffusionTracers.txsMempoolTimeoutHardCounterName
+
+muxTracer
+  :: Maybe EKGDirect
+  -> TraceSelection
+  -> Trace IO Text
+  -> Tracer IO (Mux.WithBearer (ConnectionId RemoteAddress) Mux.Trace)
+muxTracer mbEKGDirect trSel tracer = Tracer $ \ev -> do
+  -- Update the EKG metric even when this tracer is turned off.
+  flip traceWith (Mux.wbEvent ev) $
+    notifyTxsMempoolTimeoutHard mbEKGDirect
+  whenOn (traceMux trSel) $ do
+    flip traceWith ev $
+      annotateSeverity $
+        toLogObject' (traceVerbosity trSel) $
+          appendName "Mux" tracer
+
 --------------------------------------------------------------------------------
 -- Chain DB Tracers
 --------------------------------------------------------------------------------
@@ -730,6 +758,7 @@ mkConsensusTracers
      , ToObject (ValidationErr (BlockProtocol blk))
      , ToObject (ForgeStateUpdateError blk)
      , Consensus.RunNode blk
+     , ConsensusTracers.MempoolTimeoutSoftPredicate blk
      , HasKESMetricsData blk
      , HasKESInfo blk
      )
@@ -796,7 +825,7 @@ mkConsensusTracers mbEKGDirect trSel verb tr nodeKern fStats = do
 
     , Consensus.txOutboundTracer = tracerOnOff (traceTxOutbound trSel) verb "TxOutbound" tr
     , Consensus.localTxSubmissionServerTracer = tracerOnOff (traceLocalTxSubmissionServer trSel) verb "LocalTxSubmissionServer" tr
-    , Consensus.mempoolTracer = tracerOnOff' (traceMempool trSel) $ mempoolTracer trSel tr fStats
+    , Consensus.mempoolTracer = mempoolTracer mbEKGDirect trSel tr fStats
     , Consensus.forgeTracer = tracerOnOff' (traceForge trSel) $
         Tracer $ \tlcev@Consensus.TraceLabelCreds{} -> do
           traceWith (annotateSeverity
@@ -1243,6 +1272,16 @@ notifyBlockForging fStats tr = Tracer $ \case
 -- Mempool Tracers
 --------------------------------------------------------------------------------
 
+notifyTxsMempoolTimeoutSoft ::
+     ConsensusTracers.MempoolTimeoutSoftPredicate blk
+  => Maybe EKGDirect
+  -> Tracer IO (TraceEventMempool blk)
+notifyTxsMempoolTimeoutSoft mbEKGDirect = case mbEKGDirect of
+  Nothing -> nullTracer
+  Just ekgDirect -> Tracer $ \ev -> do
+    when (ConsensusTracers.impliesMempoolTimeoutSoft ev) $ do
+      sendEKGDirectCounter ekgDirect ConsensusTracers.txsMempoolTimeoutSoftCounterName
+
 notifyTxsProcessed :: ForgingStats -> Trace IO Text -> Tracer IO (TraceEventMempool blk)
 notifyTxsProcessed fStats tr = Tracer $ \case
   TraceMempoolRemoveTxs [] _ -> return ()
@@ -1285,17 +1324,22 @@ mempoolTracer
      , ToObject (ApplyTxErr blk)
      , ToObject (GenTx blk)
      , LedgerSupportsMempool blk
+     , ConsensusTracers.MempoolTimeoutSoftPredicate blk
      , ConvertRawHash blk
      )
-  => TraceSelection
+  => Maybe EKGDirect
+  -> TraceSelection
   -> Trace IO Text
   -> ForgingStats
   -> Tracer IO (TraceEventMempool blk)
-mempoolTracer tc tracer fStats = Tracer $ \ev -> do
-    traceWith (mempoolMetricsTraceTransformer tracer) ev
-    traceWith (notifyTxsProcessed fStats tracer) ev
-    let tr = appendName "Mempool" tracer
-    traceWith (mpTracer tc tr) ev
+mempoolTracer mbEKGDirect tc tracer fStats = Tracer $ \ev -> do
+    -- Update the EKG metric even when this tracer is turned off.
+    traceWith (notifyTxsMempoolTimeoutSoft mbEKGDirect) ev
+    whenOn (traceMempool tc) $ do
+      traceWith (mempoolMetricsTraceTransformer tracer) ev
+      traceWith (notifyTxsProcessed fStats tracer) ev
+      let tr = appendName "Mempool" tracer
+      traceWith (mpTracer tc tr) ev
 
 mpTracer :: ( ToJSON (GenTxId blk)
             , ToObject (ApplyTxErr blk)
@@ -1788,6 +1832,9 @@ tracerOnOff'
 tracerOnOff' (OnOff False) _ = nullTracer
 tracerOnOff' (OnOff True) tr = tr
 
+whenOn :: Monad m => OnOff b -> m () -> m ()
+whenOn (OnOff b) = when b
+
 instance Show a => Show (WithSeverity a) where
   show (WithSeverity _sev a) = show a