Feature/monitoring

Author: Cmdv

.gitignore

@@ -15,6 +15,7 @@ launch_*
 gen/
 cardano-chain-gen/test/testfiles/temp/
 /secp256k1/
+monitoring/data/
 
 # Vim
 *.swp
@@ -31,4 +32,4 @@ result*
 /.vscode
 
 # MacOS
-.DS_Store
+.DS_Store

cardano-chain-gen/test/Test/Cardano/Db/Mock/Config.hs

@@ -410,6 +410,9 @@ emptyMetricsSetters =
     , metricsSetDbSlotHeight = \_ -> pure ()
     , metricsSetDbEpochSyncDuration = \_ -> pure ()
     , metricsSetDbEpochSyncNumber = \_ -> pure ()
+    , metricsSetDbBlocksPerSecond = \_ -> pure ()
+    , metricsSetInsertDuration = \_ -> pure ()
+    , metricsSetCacheHitRate = \_ _ -> pure ()
     }
 
 withFullConfig ::

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

@@ -23,6 +23,7 @@ import Cardano.Prelude hiding (atomically)
 import Cardano.Slotting.Slot (SlotNo (..), WithOrigin (..))
 import Control.Concurrent.Class.MonadSTM.Strict
 import Control.Monad.Extra (whenJust)
+import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Ouroboros.Network.Block (BlockNo (..), Point (..))
 import Ouroboros.Network.Point (blockPointHash, blockPointSlot)
 
@@ -126,7 +127,21 @@ runActions syncEnv actions = do
           lift $ atomically $ putTMVar resultVar (points, blockNo)
           dbEvent Continue ys
         (ys, zs) -> do
+          -- Record start time and block count for performance metrics
+          startTime <- liftIO getCurrentTime
+          let blockCount = length ys
+
+          -- Process blocks
           ExceptT $ insertListBlocks syncEnv ys
+
+          -- Calculate and record blocks per second
+          when (blockCount > 0) $ do
+            endTime <- liftIO getCurrentTime
+            let duration = realToFrac $ diffUTCTime endTime startTime
+            when (duration > 0) $ do
+              let blocksPerSecond = fromIntegral blockCount / duration
+              liftIO $ setDbBlocksPerSecond (envMetricSetters syncEnv) blocksPerSecond
+
           if null zs
             then pure Continue
             else dbEvent Continue zs

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

@@ -21,6 +21,7 @@ import qualified Data.ByteString.Short as SBS
 import Data.List (span)
 import qualified Data.Set as Set
 import qualified Data.Strict.Maybe as Strict
+import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Ouroboros.Consensus.Cardano.Block (HardForkBlock (..))
 import qualified Ouroboros.Consensus.HardFork.Combinator as Consensus
 import Ouroboros.Network.Block (blockHash, blockNo, getHeaderFields, headerFieldBlockNo, unBlockNo)
@@ -42,6 +43,7 @@ import Cardano.DbSync.Error (SyncNodeError (..), mkSyncNodeCallStack)
 import Cardano.DbSync.Ledger.State (applyBlockAndSnapshot, defaultApplyResult)
 import Cardano.DbSync.Ledger.Types (ApplyResult (..))
 import Cardano.DbSync.LocalStateQuery
+import Cardano.DbSync.Metrics (setInsertDuration)
 import Cardano.DbSync.Rollback
 import Cardano.DbSync.Types
 import Cardano.DbSync.Util
@@ -151,6 +153,9 @@ insertBlock ::
   Bool ->
   ExceptT SyncNodeError DB.DbM ()
 insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
+  -- Start timing for insert duration metric
+  startTime <- liftIO getCurrentTime
+
   !epochEvents <- liftIO $ atomically $ generateNewEpochEvents syncEnv (apSlotDetails applyRes)
   let !applyResult = applyRes {apEvents = sort $ epochEvents <> apEvents applyRes}
   let !details = apSlotDetails applyResult
@@ -206,6 +211,11 @@ insertBlock syncEnv cblk applyRes firstAfterRollback tookSnapshot = do
       do
         lift $ DB.deleteConsumedTxOut tracer txOutVariantType (getSafeBlockNoDiff syncEnv)
   commitOrIndexes withinTwoMin withinHalfHour
+
+  -- Record insert duration metric
+  endTime <- liftIO getCurrentTime
+  let duration = realToFrac $ diffUTCTime endTime startTime
+  liftIO $ setInsertDuration (envMetricSetters syncEnv) duration
   where
     tracer = getTrace syncEnv
     txOutVariantType = getTxOutVariantType syncEnv

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

@@ -35,13 +35,15 @@ insertEpochSyncTime ::
   UTCTime ->
   ExceptT SyncNodeError DB.DbM ()
 insertEpochSyncTime epochNo syncState epochStats endTime = do
+  currentTime <- liftIO Time.getCurrentTime
   void
     . lift
     $ DB.insertEpochSyncTime
     $ DB.EpochSyncTime
       { DB.epochSyncTimeNo = unEpochNo epochNo - 1
       , DB.epochSyncTimeSeconds = ceiling (realToFrac (Time.diffUTCTime endTime (elsStartTime epochStats)) :: Double)
       , DB.epochSyncTimeState = syncState
+      , DB.epochSyncTimeSyncedAt = Just currentTime
       }
 
 initEpochStatistics :: MonadIO m => m (StrictTVar IO EpochStatistics)

cardano-db-sync/src/Cardano/DbSync/Era/Universal/Insert/LedgerEvent.hs

@@ -19,7 +19,7 @@ import Cardano.Slotting.Slot (EpochNo (..))
 
 import Cardano.DbSync.Api
 import Cardano.DbSync.Api.Types (EpochStatistics (..), InsertOptions (..), SyncEnv (..), UnicodeNullSource, formatUnicodeNullSource)
-import Cardano.DbSync.Cache.Types (textShowCacheStats)
+import Cardano.DbSync.Cache.Types (CacheStatistics (..), textShowCacheStats)
 import Cardano.DbSync.Era.Cardano.Util (insertEpochSyncTime, resetEpochStatistics)
 import qualified Cardano.DbSync.Era.Shelley.Generic as Generic
 import Cardano.DbSync.Era.Universal.Adjust (adjustEpochRewards)
@@ -31,7 +31,7 @@ import Cardano.DbSync.Types
 
 import Cardano.DbSync.Error (SyncNodeError)
 import Cardano.DbSync.Ledger.Types
-import Cardano.DbSync.Metrics (setDbEpochSyncDuration, setDbEpochSyncNumber)
+import Cardano.DbSync.Metrics (setCacheHitRate, setDbEpochSyncDuration, setDbEpochSyncNumber)
 import Control.Concurrent.Class.MonadSTM.Strict (readTVarIO, writeTVar)
 import Control.Monad.Extra (whenJust)
 import qualified Data.Map.Strict as Map
@@ -96,6 +96,17 @@ insertNewEpochLedgerEvents syncEnv applyRes currentEpochNo@(EpochNo curEpoch) =
           liftIO $ setDbEpochSyncDuration metricSetters (epochDurationSeconds (elsStartTime epochStats) currentTime)
           liftIO $ setDbEpochSyncNumber metricSetters (fromIntegral $ unEpochNo en - 1)
 
+          -- Calculate and set cache hit rates
+          let cacheStats = elsCaches epochStats
+              hitRate hits queries = if queries == 0 then 0.0 else fromIntegral hits / fromIntegral queries
+          liftIO $ setCacheHitRate metricSetters "stake" (hitRate (credsHits cacheStats) (credsQueries cacheStats))
+          liftIO $ setCacheHitRate metricSetters "pools" (hitRate (poolsHits cacheStats) (poolsQueries cacheStats))
+          liftIO $ setCacheHitRate metricSetters "datum" (hitRate (datumHits cacheStats) (datumQueries cacheStats))
+          liftIO $ setCacheHitRate metricSetters "multi_assets" (hitRate (multiAssetsHits cacheStats) (multiAssetsQueries cacheStats))
+          liftIO $ setCacheHitRate metricSetters "prev_block" (hitRate (prevBlockHits cacheStats) (prevBlockQueries cacheStats))
+          liftIO $ setCacheHitRate metricSetters "address" (hitRate (addressHits cacheStats) (addressQueries cacheStats))
+          liftIO $ setCacheHitRate metricSetters "tx_ids" (hitRate (txIdsHits cacheStats) (txIdsQueries cacheStats))
+
           -- Log comprehensive epoch statistics
           liftIO . logInfo tracer $
             mconcat

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

@@ -9,6 +9,9 @@ module Cardano.DbSync.Metrics (
   setDbSlotHeight,
   setDbEpochSyncDuration,
   setDbEpochSyncNumber,
+  setDbBlocksPerSecond,
+  setInsertDuration,
+  setCacheHitRate,
   makeMetrics,
   withMetricSetters,
   withMetricsServer,
@@ -41,6 +44,24 @@ data Metrics = Metrics
   -- ^ The duration of the last epoch sync in seconds.
   , mDbEpochSyncNumber :: !Gauge
   -- ^ The number of the last epoch that was synced.
+  , mDbBlocksPerSecond :: !Gauge
+  -- ^ The number of blocks being processes per second.
+  , mInsertDuration :: !Gauge
+  -- ^ The duration of the last insert operation in seconds.
+  , mCacheStakeHitRate :: !Gauge
+  -- ^ Cache hit rate for stake cache.
+  , mCachePoolsHitRate :: !Gauge
+  -- ^ Cache hit rate for pools cache.
+  , mCacheDatumHitRate :: !Gauge
+  -- ^ Cache hit rate for datum cache.
+  , mCacheMultiAssetsHitRate :: !Gauge
+  -- ^ Cache hit rate for multi_assets cache.
+  , mCachePrevBlockHitRate :: !Gauge
+  -- ^ Cache hit rate for prev_block cache.
+  , mCacheAddressHitRate :: !Gauge
+  -- ^ Cache hit rate for address cache.
+  , mCacheTxIdsHitRate :: !Gauge
+  -- ^ Cache hit rate for tx_ids cache.
   }
 
 -- This enables us to be much more flexibile with what we actually measure.
@@ -61,6 +82,20 @@ withMetricSetters prometheusPort action =
             Gauge.set duration $ mDbEpochSyncDuration metrics
         , metricsSetDbEpochSyncNumber = \epochNo ->
             Gauge.set (fromIntegral epochNo) $ mDbEpochSyncNumber metrics
+        , metricsSetDbBlocksPerSecond = \bps ->
+            Gauge.set bps $ mDbBlocksPerSecond metrics
+        , metricsSetInsertDuration = \duration ->
+            Gauge.set duration $ mInsertDuration metrics
+        , metricsSetCacheHitRate = \cacheName hitRate ->
+            case cacheName of
+              "stake" -> Gauge.set hitRate $ mCacheStakeHitRate metrics
+              "pools" -> Gauge.set hitRate $ mCachePoolsHitRate metrics
+              "datum" -> Gauge.set hitRate $ mCacheDatumHitRate metrics
+              "multi_assets" -> Gauge.set hitRate $ mCacheMultiAssetsHitRate metrics
+              "prev_block" -> Gauge.set hitRate $ mCachePrevBlockHitRate metrics
+              "address" -> Gauge.set hitRate $ mCacheAddressHitRate metrics
+              "tx_ids" -> Gauge.set hitRate $ mCacheTxIdsHitRate metrics
+              _ -> pure ()
         }
 
 withMetricsServer :: Int -> (Metrics -> IO a) -> IO a
@@ -83,6 +118,15 @@ makeMetrics =
     <*> registerGauge "cardano_db_sync_db_slot_height" mempty
     <*> registerGauge "cardano_db_sync_db_epoch_sync_duration_seconds" mempty
     <*> registerGauge "cardano_db_sync_db_epoch_sync_number" mempty
+    <*> registerGauge "cardano_db_sync_blocks_per_second" mempty
+    <*> registerGauge "cardano_db_sync_insert_duration_seconds" mempty
+    <*> registerGauge "cardano_db_sync_cache_stake_hit_rate" mempty
+    <*> registerGauge "cardano_db_sync_cache_pools_hit_rate" mempty
+    <*> registerGauge "cardano_db_sync_cache_datum_hit_rate" mempty
+    <*> registerGauge "cardano_db_sync_cache_multi_assets_hit_rate" mempty
+    <*> registerGauge "cardano_db_sync_cache_prev_block_hit_rate" mempty
+    <*> registerGauge "cardano_db_sync_cache_address_hit_rate" mempty
+    <*> registerGauge "cardano_db_sync_cache_tx_ids_hit_rate" mempty
 
 setNodeBlockHeight :: MetricSetters -> WithOrigin BlockNo -> IO ()
 setNodeBlockHeight setters woBlkNo =
@@ -102,3 +146,12 @@ setDbEpochSyncDuration = metricsSetDbEpochSyncDuration
 
 setDbEpochSyncNumber :: MetricSetters -> Word64 -> IO ()
 setDbEpochSyncNumber = metricsSetDbEpochSyncNumber
+
+setDbBlocksPerSecond :: MetricSetters -> Double -> IO ()
+setDbBlocksPerSecond = metricsSetDbBlocksPerSecond
+
+setInsertDuration :: MetricSetters -> Double -> IO ()
+setInsertDuration = metricsSetInsertDuration
+
+setCacheHitRate :: MetricSetters -> Text -> Double -> IO ()
+setCacheHitRate = metricsSetCacheHitRate

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

@@ -139,6 +139,9 @@ data MetricSetters = MetricSetters
   , metricsSetDbSlotHeight :: SlotNo -> IO ()
   , metricsSetDbEpochSyncDuration :: Double -> IO ()
   , metricsSetDbEpochSyncNumber :: Word64 -> IO ()
+  , metricsSetDbBlocksPerSecond :: Double -> IO ()
+  , metricsSetInsertDuration :: Double -> IO ()
+  , metricsSetCacheHitRate :: Text -> Double -> IO ()
   }
 
 data SyncState = SyncLagging | SyncFollowing

cardano-db/src/Cardano/Db/Schema/Core/EpochAndProtocol.hs

@@ -324,6 +324,7 @@ data EpochSyncTime = EpochSyncTime
   { epochSyncTimeNo :: !Word64 -- sqltype=word31type
   , epochSyncTimeSeconds :: !Word64 -- sqltype=word63type
   , epochSyncTimeState :: !SyncState -- sqltype=syncstatetype
+  , epochSyncTimeSyncedAt :: !(Maybe UTCTime)
   }
   deriving (Show, Eq, Generic)
 
@@ -338,6 +339,7 @@ epochSyncTimeEncoder =
     [ epochSyncTimeNo >$< E.param (E.nonNullable $ fromIntegral >$< E.int8)
     , epochSyncTimeSeconds >$< E.param (E.nonNullable $ fromIntegral >$< E.int8)
     , epochSyncTimeState >$< E.param (E.nonNullable syncStateEncoder)
+    , epochSyncTimeSyncedAt >$< E.param (E.nullable E.timestamptz)
     ]
 
 -----------------------------------------------------------------------------------------------------------------------------------