[CAD-2547] Add Prometheus metrics for slot_no.

cardano-db-sync-extended/app/cardano-db-sync-extended.hs

@@ -7,6 +7,7 @@ import           Cardano.Config.Git.Rev (gitRev)
 
 import           Cardano.DbSync (runDbSyncNode)
 import           Cardano.DbSync.Plugin.Extended (extendedDbSyncNodePlugin)
+import           Cardano.DbSync.Metrics (withMetricsLayer)
 
 import           Cardano.Sync.Config
 import           Cardano.Sync.Config.Types
@@ -29,7 +30,14 @@ main = do
   cmd <- Opt.execParser opts
   case cmd of
     CmdVersion -> runVersionCommand
-    CmdRun params -> runDbSyncNode extendedDbSyncNodePlugin params
+    CmdRun params -> do
+        let configFile = enpConfigFile params
+        enc <- readSyncNodeConfig configFile
+        let prometheusPort = dncPrometheusPort enc
+
+        -- This enables us to be much more flexibile with what we actually measure.
+        withMetricsLayer prometheusPort $ \metricsLayer ->
+            runDbSyncNode metricsLayer extendedDbSyncNodePlugin params
 
 -- -------------------------------------------------------------------------------------------------
 

cardano-db-sync/app/cardano-db-sync.hs

@@ -6,6 +6,7 @@ import           Cardano.Prelude
 import           Cardano.Config.Git.Rev (gitRev)
 
 import           Cardano.DbSync (defDbSyncNodePlugin, runDbSyncNode)
+import           Cardano.DbSync.Metrics (withMetricsLayer)
 
 import           Cardano.Sync.Config
 import           Cardano.Sync.Config.Types
@@ -29,7 +30,14 @@ main = do
   cmd <- Opt.execParser opts
   case cmd of
     CmdVersion -> runVersionCommand
-    CmdRun params -> runDbSyncNode defDbSyncNodePlugin params
+    CmdRun params -> do
+        let configFile = enpConfigFile params
+        enc <- readSyncNodeConfig configFile
+        let prometheusPort = dncPrometheusPort enc
+
+        -- This enables us to be much more flexibile with what we actually measure.
+        withMetricsLayer prometheusPort $ \metricsLayer ->
+            runDbSyncNode metricsLayer defDbSyncNodePlugin params
 
 -- -------------------------------------------------------------------------------------------------
 

cardano-db-sync/src/Cardano/DbSync.hs

@@ -37,7 +37,7 @@ import           Cardano.DbSync.Plugin.Default (defDbSyncNodePlugin)
 import           Cardano.DbSync.Rollback (unsafeRollback)
 import           Cardano.Sync.Database (runDbThread)
 
-import           Cardano.Sync (Block (..), SyncDataLayer (..), SyncNodePlugin (..),
+import           Cardano.Sync (Block (..), SyncDataLayer (..), SyncNodePlugin (..), MetricsLayer,
                    configureLogging, runSyncNode)
 import           Cardano.Sync.Config.Types (ConfigFile (..), GenesisFile (..), LedgerStateDir (..),
                    MigrationDir (..), NetworkName (..), SocketPath (..), SyncCommand (..),
@@ -49,8 +49,8 @@ import           Database.Persist.Postgresql (withPostgresqlConn)
 import           Database.Persist.Sql (SqlBackend)
 
 
-runDbSyncNode :: (SqlBackend -> SyncNodePlugin) -> SyncNodeParams -> IO ()
-runDbSyncNode mkPlugin params = do
+runDbSyncNode :: MetricsLayer -> (SqlBackend -> SyncNodePlugin) -> SyncNodeParams -> IO ()
+runDbSyncNode metricsLayer mkPlugin params = do
 
     -- Read the PG connection info
     pgConfig <- DB.readPGPassFileEnv Nothing
@@ -69,7 +69,7 @@ runDbSyncNode mkPlugin params = do
           Just slotNo -> void $ unsafeRollback trce slotNo
           Nothing -> pure ()
 
-        runSyncNode (mkSyncDataLayer trce backend) trce (mkPlugin backend)
+        runSyncNode (mkSyncDataLayer trce backend) metricsLayer trce (mkPlugin backend)
             params (insertValidateGenesisDist backend) runDbThread
 
 -- -------------------------------------------------------------------------------------------------

cardano-db-sync/src/Cardano/DbSync/Metrics.hs

@@ -4,25 +4,48 @@
 module Cardano.DbSync.Metrics
   ( Metrics (..)
   , makeMetrics
+  , withMetricsLayer
   , withMetricsServer
   ) where
 
 import           Cardano.Prelude
 
+import           Cardano.Sync.Types (MetricsLayer (..))
+
 import           System.Metrics.Prometheus.Concurrent.RegistryT (RegistryT (..), registerGauge,
                    runRegistryT, unRegistryT)
 import           System.Metrics.Prometheus.Http.Scrape (serveMetricsT)
 import           System.Metrics.Prometheus.Metric.Gauge (Gauge)
+import qualified System.Metrics.Prometheus.Metric.Gauge as Gauge
 
 
 data Metrics = Metrics
-  { mDbHeight :: !Gauge
-  , mNodeHeight :: !Gauge
-  , mQueuePre :: !Gauge
-  , mQueuePost :: !Gauge
-  , mQueuePostWrite :: !Gauge
+  { mNodeBlockHeight :: !Gauge
+  , mDbQueueHeight :: !Gauge
+  , mLatestBlockNum :: !Gauge
+  , mLatestSlotNum :: !Gauge
   }
 
+-- This enables us to be much more flexibile with what we actually measure.
+withMetricsLayer :: Int -> (MetricsLayer -> IO a) -> IO a
+withMetricsLayer prometheusPort action =
+    withMetricsServer prometheusPort $ \metrics -> do
+
+        -- Metrics layer.
+        let metricsLayer =
+                MetricsLayer
+                    { mlSetNodeBlockHeight = \nodeHeight ->
+                        Gauge.set (fromIntegral nodeHeight) $ mNodeBlockHeight metrics
+                    , mlSetQueuePostWrite = \queuePostWrite ->
+                        Gauge.set (fromIntegral queuePostWrite) $ mDbQueueHeight metrics
+                    , mlSetLatestBlockNo = \blockNo ->
+                        Gauge.set (fromIntegral blockNo) $ mLatestBlockNum metrics
+                    , mlSetLatestSlotNo = \slotNo ->
+                        Gauge.set (fromIntegral slotNo) $ mLatestSlotNum metrics
+                    }
+
+        action metricsLayer
+
 withMetricsServer :: Int -> (Metrics -> IO a) -> IO a
 withMetricsServer port action = do
   (metrics, registry) <- runRegistryT $ (,) <$> makeMetrics <*> RegistryT ask
@@ -34,9 +57,8 @@ withMetricsServer port action = do
 makeMetrics :: RegistryT IO Metrics
 makeMetrics =
   Metrics
-    <$> registerGauge "db_block_height" mempty
-    <*> registerGauge "remote_tip_height" mempty
-    <*> registerGauge "action_queue_length_pre" mempty
-    <*> registerGauge "action_queue_length_post" mempty
-    <*> registerGauge "action_queue_length_post_write" mempty
+    <$> registerGauge "remote_block_tip_height" mempty
+    <*> registerGauge "database_queue_height" mempty
+    <*> registerGauge "local_block_number" mempty
+    <*> registerGauge "local_slot_number" mempty
 

cardano-sync/cardano-sync.cabal

@@ -45,7 +45,6 @@ library
                         Cardano.Sync.Error
 
                         Cardano.Sync.LedgerState
-                        Cardano.Sync.Metrics
 
                         Cardano.Sync.Era.Byron.Util
                         Cardano.Sync.Era.Cardano.Util
@@ -97,7 +96,6 @@ library
                       , ouroboros-consensus-shelley
                       , ouroboros-network
                       , ouroboros-network-framework
-                      , prometheus
                       , shelley-spec-ledger
                       , stm
                       , text

cardano-sync/src/Cardano/Sync.hs

@@ -19,6 +19,8 @@ module Cardano.Sync
   , SocketPath (..)
 
   , SyncDataLayer (..)
+  , MetricsLayer (..)
+  , emptyMetricsLayer
   , Block (..)
   , Meta (..)
   , SyncEnv (..)
@@ -44,7 +46,6 @@ import           Cardano.Sync.Config
 import           Cardano.Sync.Database (DbAction (..), DbActionQueue, lengthDbActionQueue,
                    mkDbApply, mkDbRollback, newDbActionQueue, runDbStartup, writeDbActionQueue)
 import           Cardano.Sync.Error
-import           Cardano.Sync.Metrics
 import           Cardano.Sync.Plugin (SyncNodePlugin (..))
 import           Cardano.Sync.StateQuery (StateQueryTMVar, getSlotDetails, localStateQueryHandler,
                    newStateQueryTMVar)
@@ -103,7 +104,6 @@ import qualified Ouroboros.Network.Snocket as Snocket
 import           Ouroboros.Network.Subscription (SubscriptionTrace)
 
 import           System.Directory (createDirectoryIfMissing)
-import qualified System.Metrics.Prometheus.Metric.Gauge as Gauge
 
 
 type InsertValidateGenesisFunction
@@ -113,22 +113,23 @@ type InsertValidateGenesisFunction
     -> ExceptT SyncNodeError IO ()
 
 type RunDBThreadFunction
-    =  Trace IO Text
+    =  MetricsLayer
+    -> Trace IO Text
     -> SyncEnv
     -> SyncNodePlugin
-    -> Metrics
     -> DbActionQueue
     -> IO ()
 
 runSyncNode
     :: SyncDataLayer
+    -> MetricsLayer
     -> Trace IO Text
     -> SyncNodePlugin
     -> SyncNodeParams
     -> InsertValidateGenesisFunction
     -> RunDBThreadFunction
     -> IO ()
-runSyncNode dataLayer trce plugin enp insertValidateGenesisDist runDBThreadFunction =
+runSyncNode dataLayer metricsLayer trce plugin enp insertValidateGenesisDist runDBThreadFunction =
   withIOManager $ \iomgr -> do
 
     let configFile = enpConfigFile enp
@@ -152,7 +153,7 @@ runSyncNode dataLayer trce plugin enp insertValidateGenesisDist runDBThreadFunct
       case genCfg of
           GenesisCardano _ bCfg _sCfg -> do
             syncEnv <- ExceptT $ mkSyncEnvFromConfig dataLayer (enpLedgerStateDir enp) genCfg
-            liftIO $ runSyncNodeNodeClient (dncPrometheusPort enc) syncEnv iomgr trce plugin
+            liftIO $ runSyncNodeNodeClient metricsLayer syncEnv iomgr trce plugin
                         runDBThreadFunction (cardanoCodecConfig bCfg) (enpSocketPath enp)
   where
     cardanoCodecConfig :: Byron.Config -> CodecConfig CardanoBlock
@@ -166,7 +167,7 @@ runSyncNode dataLayer trce plugin enp insertValidateGenesisDist runDBThreadFunct
 -- -------------------------------------------------------------------------------------------------
 
 runSyncNodeNodeClient
-    :: Int
+    :: MetricsLayer
     -> SyncEnv
     -> IOManager
     -> Trace IO Text
@@ -175,17 +176,16 @@ runSyncNodeNodeClient
     -> CodecConfig CardanoBlock
     -> SocketPath
     -> IO ()
-runSyncNodeNodeClient port env iomgr trce plugin runDBThreadFunction codecConfig (SocketPath socketPath) = do
+runSyncNodeNodeClient metricsLayer env iomgr trce plugin runDBThreadFunction codecConfig (SocketPath socketPath) = do
   queryVar <- newStateQueryTMVar
   logInfo trce $ "localInitiatorNetworkApplication: connecting to node via " <> textShow socketPath
-  withMetricsServer port $ \ metrics ->
-    void $ subscribe
-      (localSnocket iomgr socketPath)
-      codecConfig
-      (envNetworkMagic env)
-      networkSubscriptionTracers
-      clientSubscriptionParams
-      (dbSyncProtocols trce env plugin metrics queryVar runDBThreadFunction)
+  void $ subscribe
+    (localSnocket iomgr socketPath)
+    codecConfig
+    (envNetworkMagic env)
+    networkSubscriptionTracers
+    clientSubscriptionParams
+    (dbSyncProtocols metricsLayer trce env plugin queryVar runDBThreadFunction)
   where
     clientSubscriptionParams =
       ClientSubscriptionParams
@@ -217,17 +217,17 @@ runSyncNodeNodeClient port env iomgr trce plugin runDBThreadFunction codecConfig
     handshakeTracer = toLogObject $ appendName "Handshake" trce
 
 dbSyncProtocols
-    :: Trace IO Text
+    :: MetricsLayer
+    -> Trace IO Text
     -> SyncEnv
     -> SyncNodePlugin
-    -> Metrics
     -> StateQueryTMVar CardanoBlock (Interpreter (CardanoEras StandardCrypto))
     -> RunDBThreadFunction
     -> Network.NodeToClientVersion
     -> ClientCodecs CardanoBlock IO
     -> ConnectionId LocalAddress
     -> NodeToClientProtocols 'InitiatorMode BSL.ByteString IO () Void
-dbSyncProtocols trce env plugin metrics queryVar runDBThreadFunction version codecs _connectionId =
+dbSyncProtocols metricsLayer trce env plugin queryVar runDBThreadFunction version codecs _connectionId =
     NodeToClientProtocols
       { localChainSyncProtocol = localChainSyncPtcl
       , localTxSubmissionProtocol = dummylocalTxSubmit
@@ -252,13 +252,13 @@ dbSyncProtocols trce env plugin metrics queryVar runDBThreadFunction version cod
         actionQueue <- newDbActionQueue
 
         race_
-            (runDBThreadFunction trce env plugin metrics actionQueue)
+            (runDBThreadFunction metricsLayer trce env plugin actionQueue)
             (runPipelinedPeer
                 localChainSyncTracer
                 (cChainSyncCodec codecs)
                 channel
                 (chainSyncClientPeerPipelined
-                    $ chainSyncClient (envDataLayer env) trce env queryVar metrics latestPoints currentTip actionQueue)
+                    $ chainSyncClient (envDataLayer env) metricsLayer trce env queryVar latestPoints currentTip actionQueue)
             )
 
         atomically $ writeDbActionQueue actionQueue DbFinish
@@ -337,15 +337,15 @@ getCurrentTipBlockNo dataLayer = do
 --
 chainSyncClient
     :: SyncDataLayer
+    -> MetricsLayer
     -> Trace IO Text
     -> SyncEnv
     -> StateQueryTMVar CardanoBlock (Interpreter (CardanoEras StandardCrypto))
-    -> Metrics
     -> [Point CardanoBlock]
     -> WithOrigin BlockNo
     -> DbActionQueue
     -> ChainSyncClientPipelined CardanoBlock (Point CardanoBlock) (Tip CardanoBlock) IO ()
-chainSyncClient dataLayer trce env queryVar metrics latestPoints currentTip actionQueue = do
+chainSyncClient dataLayer metricsLayer trce env queryVar latestPoints currentTip actionQueue = do
     ChainSyncClientPipelined $ pure $
       -- Notify the core node about the our latest points at which we are
       -- synchronised.  This client is not persistent and thus it just
@@ -387,12 +387,20 @@ chainSyncClient dataLayer trce env queryVar metrics latestPoints currentTip acti
       ClientStNext
         { recvMsgRollForward = \blk tip ->
               logException trce "recvMsgRollForward: " $ do
-                Gauge.set (withOrigin 0 (fromIntegral . unBlockNo) (getTipBlockNo tip)) (mNodeHeight metrics)
+
+                -- Setting the value from network/node
+                let setNodeBlockHeight = mlSetNodeBlockHeight metricsLayer
+                setNodeBlockHeight (withOrigin 0 unBlockNo (getTipBlockNo tip))
+
                 details <- getSlotDetails trce env queryVar (cardanoBlockSlotNo blk)
                 newSize <- atomically $ do
                                 writeDbActionQueue actionQueue $ mkDbApply blk details
                                 lengthDbActionQueue actionQueue
-                Gauge.set (fromIntegral newSize) $ mQueuePostWrite metrics
+
+                -- The action queue size
+                let setQueuePostWrite = mlSetQueuePostWrite metricsLayer
+                setQueuePostWrite newSize
+
                 pure $ finish (At (blockNo blk)) tip
         , recvMsgRollBackward = \point tip ->
               logException trce "recvMsgRollBackward: " $ do

cardano-sync/src/Cardano/Sync/Database.hs

@@ -24,13 +24,10 @@ import           Cardano.Sync.Api
 import           Cardano.Sync.DbAction
 import           Cardano.Sync.Error
 import           Cardano.Sync.LedgerState
-import           Cardano.Sync.Metrics
 import           Cardano.Sync.Plugin
 import           Cardano.Sync.Types
 import           Cardano.Sync.Util
 
-import qualified System.Metrics.Prometheus.Metric.Gauge as Gauge
-
 data NextState
   = Continue
   | Done
@@ -41,10 +38,10 @@ runDbStartup trce plugin =
     mapM_ (\action -> action trce) $ plugOnStartup plugin
 
 runDbThread
-    :: Trace IO Text -> SyncEnv -> SyncNodePlugin -> Metrics
+    :: MetricsLayer -> Trace IO Text -> SyncEnv -> SyncNodePlugin
     -> DbActionQueue
     -> IO ()
-runDbThread trce env plugin metrics queue = do
+runDbThread metricsLayer trce env plugin queue = do
     logInfo trce "Running DB thread"
     logException trce "runDBThread: " loop
     logInfo trce "Shutting down DB thread"
@@ -61,9 +58,22 @@ runDbThread trce env plugin metrics queue = do
       mBlkNo <- getLatestBlock
 
       -- Chain Maybe's.
+      -- Here we are extracting information from the database and writing
+      -- block number into the metrics so we have that available.
       case bBlockNo =<< mBlkNo of
         Nothing -> pure ()
-        Just blkNo -> Gauge.set (fromIntegral blkNo) $ mDbHeight metrics
+        Just blkNo -> do
+            let setLatestBlockNo = mlSetLatestBlockNo metricsLayer
+            setLatestBlockNo blkNo
+
+      -- Chain Maybe's.
+      -- Here we are extracting information from the database and writing
+      -- slot number into the metrics so we have that available.
+      case bSlotNo =<< mBlkNo of
+        Nothing -> pure ()
+        Just slotNo -> do
+            let setLatestSlotNo = mlSetLatestSlotNo metricsLayer
+            setLatestSlotNo slotNo
 
       case eNextState of
         Left err -> logError trce $ renderSyncNodeError err