Merge pull request #6473 from IntersectMBO/russoul/integrate-timeseries-io

bench: Integrate cardano-timeseries-io

Author: mgmeier

bench/cardano-timeseries-io/src/Cardano/Timeseries/Component.hs

@@ -4,7 +4,6 @@
 {-# HLINT ignore "Use newtype instead of data" #-}
 {-# LANGUAGE AllowAmbiguousTypes #-}
 {-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
@@ -81,16 +80,16 @@ create tr mbCfg = do
     runPruner handle = threadLabelMe "timeseries-pruner-thread" >> do
       forever $ do
         cfg <- readTVarIO handle.config
-        case cfg.pruningPeriodSec of
+        case cfg.pruningPeriodMillis of
           Nothing ->
             -- If the current configuration doesn't specify a pruning period, we block
             -- the thread until a reconfiguration happens.
             takeMVar handle.reconfigured
-          Just period -> do
+          Just periodMs -> do
             prune handle
             -- Wait for the given period or wake up on a reconfiguration.
             race_
-              (threadDelay (fromIntegral period * 1_000_000))
+              (threadDelay (fromIntegral periodMs * 1000))
               (takeMVar handle.reconfigured)
 
 -- | Reconfigure the store. The new parameters are applied immediately.

bench/cardano-timeseries-io/src/Cardano/Timeseries/Component/Types.hs

@@ -13,8 +13,8 @@ type QueryId = Word64
 data TimeseriesConfig = TimeseriesConfig {
     -- | How long the store entries are retained for (ms).
     retentionMillis :: Word64
-    -- | How often the pruner thread shall prune the store (sec), if enabled.
-  , pruningPeriodSec :: Maybe Word64
+    -- | How often the pruner thread shall prune the store (ms), if enabled.
+  , pruningPeriodMillis :: Maybe Word64
     -- | Parameters of timeseries query interpretation.
   , interpCfg :: Interp.Config
 } deriving (Show, Generic, ToJSON)

cardano-tracer/CHANGELOG.md

@@ -1,6 +1,8 @@
 # ChangeLog
 
 ## NEXT
+
+## 0.4.0 (April 2026)
 * RTView: Remove monitoring based on the `NodePeers` datapoint, which has been removed
   since Node 10.6.2; fixes the RTView-enabled build.
 * Introduce secure communication via HTTPS for EKG and Prometheus metric servers, enabled
@@ -16,6 +18,10 @@
        , "certificateKeyFile": "/path/to/key.pem"
        , "certificateChain": ["/path/to/intermediate1.pem", "/path/to/intermediate2.pem"]
        }
+* Integrate a timeseries store as a REST API:
+   - Tweak ekg-forward to relay node-emanated metrics to the timeseries store
+   - The store can be queried, pruned and configured dynamically via the API
+   - The store can be pruned periodically by an automatic process (configurable)
 
 ## 0.3.6 (November 2025)
 * Implement Prometheus HTTP service discovery (SD) under the URL `/targets`

cardano-tracer/bench/cardano-tracer-bench.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 import           Cardano.Logging hiding (LocalSocket)
@@ -61,7 +62,7 @@ main = do
   rtViewPageOpened <- newTVarIO False
 #endif
 
-  tracer <- mkTracerTracer $ SeverityF $ Just Warning
+  tracer <- mkTraceBundle $ SeverityF $ Just Warning
 
   let tracerEnv :: TracerConfig -> HandleRegistry -> TracerEnv
       tracerEnv config handleRegistry = TracerEnv
@@ -73,11 +74,12 @@ main = do
         , teCurrentDPLock         = currentDPLock
         , teDPRequestors          = dpRequestors
         , teProtocolsBrake        = protocolsBrake
-        , teTracer                = tracer
+        , teTracer                = tracer.assorted
         , teReforwardTraceObjects = \_-> pure ()
         , teRegistry              = handleRegistry
         , teStateDir              = Nothing
         , teMetricsHelp           = []
+        , teTimeseriesHandle      = Nothing
         }
 
       tracerEnvRTView :: TracerEnvRTView
@@ -140,22 +142,23 @@ main = do
 
   mkConfig :: FilePath -> LogFormat -> TracerConfig
   mkConfig root format = TracerConfig
-    { networkMagic   = 764824073
-    , network        = AcceptAt (Net.LocalPipe "")
-    , loRequestNum   = Nothing
-    , ekgRequestFreq = Nothing
-    , hasEKG         = Nothing
-    , hasPrometheus  = Nothing
-    , hasRTView      = Nothing
-    , tlsCertificate = Nothing
-    , logging        = NE.fromList [LoggingParams root FileMode format]
-    , rotation       = Nothing
-    , verbosity      = Nothing
-    , metricsNoSuffix = Nothing
-    , metricsHelp    = Nothing
-    , hasForwarding  = Nothing
-    , resourceFreq   = Nothing
-    , ekgRequestFull = Nothing
+    { networkMagic     = 764824073
+    , network          = AcceptAt (Net.LocalPipe "")
+    , loRequestNum     = Nothing
+    , ekgRequestFreq   = Nothing
+    , hasEKG           = Nothing
+    , hasPrometheus    = Nothing
+    , hasRTView        = Nothing
+    , hasTimeseries    = Nothing
+    , tlsCertificate   = Nothing
+    , logging          = NE.fromList [LoggingParams root FileMode format]
+    , rotation         = Nothing
+    , verbosity        = Nothing
+    , metricsNoSuffix  = Nothing
+    , metricsHelp      = Nothing
+    , hasForwarding    = Nothing
+    , resourceFreq     = Nothing
+    , ekgRequestFull   = Nothing
     , prometheusLabels = Nothing
     }
 

cardano-tracer/cardano-tracer.cabal

@@ -1,7 +1,7 @@
 cabal-version: 3.0
 
 name:                   cardano-tracer
-version:                0.3.7
+version:                0.4.0
 synopsis:               A service for logging and monitoring over Cardano nodes
 description:            A service for logging and monitoring over Cardano nodes.
 category:               Cardano,
@@ -123,6 +123,7 @@ library
                         Cardano.Tracer.Handlers.Metrics.Monitoring
                         Cardano.Tracer.Handlers.Metrics.Prometheus
                         Cardano.Tracer.Handlers.Metrics.Servers
+                        Cardano.Tracer.Handlers.Metrics.TimeseriesServer
                         Cardano.Tracer.Handlers.Metrics.Utils
 
                         Cardano.Tracer.Handlers.Notifications.Check
@@ -147,6 +148,8 @@ library
 
   other-modules:        Cardano.Tracer.Handlers.Logs.Journal.NoSystemd
                         Cardano.Tracer.Handlers.Notifications.Timer
+                        Cardano.Tracer.Time
+
                         Paths_cardano_tracer
 
   autogen-modules:      Paths_cardano_tracer
@@ -155,6 +158,7 @@ library
     build-depends:
                         cardano-git-rev ^>=0.2.2
                       , cassava
+                      , string-qq
                       , threepenny-gui
                       , utf8-string
                       , vector
@@ -201,6 +205,7 @@ library
                       , warp ^>= 3.4
                       , warp-tls
                       , yaml
+                      , cardano-timeseries-io
 
   if os(windows)
     build-depends:    Win32

cardano-tracer/src/Cardano/Tracer/Acceptors/Client.hs

@@ -41,7 +41,7 @@ import           Data.Word (Word32)
 import qualified Network.Mux as Mux
 import qualified Network.Socket as Socket
 import qualified System.Metrics.Configuration as EKGF
-import           System.Metrics.Network.Acceptor (acceptEKGMetricsInit)
+import           System.Metrics.Network.Acceptor (acceptMetricsInit)
 
 import qualified Trace.Forward.Configuration.DataPoint as DPF
 import qualified Trace.Forward.Configuration.TraceObject as TF
@@ -192,9 +192,10 @@ runEKGAcceptorInit
                      respoinderCtx
                      LBS.ByteString IO () Void
 runEKGAcceptorInit tracerEnv ekgConfig errorHandler =
-  acceptEKGMetricsInit
+  acceptMetricsInit
     ekgConfig
     (prepareMetricsStores tracerEnv . micConnectionId)
+    (store tracerEnv . connIdToNodeId . micConnectionId)
     (errorHandler . micConnectionId)
 
 runTraceObjectsAcceptorInit

cardano-tracer/src/Cardano/Tracer/Acceptors/Server.hs

@@ -5,8 +5,6 @@ module Cardano.Tracer.Acceptors.Server
   ( runAcceptorsServer
   ) where
 
-import "contra-tracer" Control.Tracer (nullTracer)
-
 import           Cardano.Logging (TraceObject)
 import qualified Cardano.Logging.Types as Net
 import           Cardano.Tracer.Acceptors.Utils
@@ -25,25 +23,25 @@ import           Ouroboros.Network.Mux (MiniProtocol (..), MiniProtocolLimits (.
                    miniProtocolNum, miniProtocolRun)
 import           Ouroboros.Network.Protocol.Handshake (Handshake, HandshakeArguments (..))
 import qualified Ouroboros.Network.Protocol.Handshake as Handshake
+import qualified Ouroboros.Network.Server.Simple as Server
 import           Ouroboros.Network.Snocket (LocalAddress, LocalSocket, Snocket,
                    localAddressFromPath, localSnocket, makeLocalBearer, makeSocketBearer,
                    socketSnocket)
-import           Ouroboros.Network.Socket (ConnectionId (..),
-                   SomeResponderApplication (..))
-import qualified Ouroboros.Network.Server.Simple as Server
+import           Ouroboros.Network.Socket (ConnectionId (..), SomeResponderApplication (..))
 
 import           Codec.CBOR.Term (Term)
 import           Control.Concurrent.Async (wait)
+import           "contra-tracer" Control.Tracer (nullTracer)
 import qualified Data.ByteString.Lazy as LBS
+import           Data.Functor (void)
 import           Data.List.NonEmpty (NonEmpty ((:|)))
 import qualified Data.Text as Text
-import           Data.Functor (void)
 import           Data.Void (Void)
 import           Data.Word (Word32)
 import qualified Network.Mux as Mux
 import qualified Network.Socket as Socket
 import qualified System.Metrics.Configuration as EKGF
-import           System.Metrics.Network.Acceptor (acceptEKGMetricsResp)
+import           System.Metrics.Network.Acceptor (acceptMetricsResp)
 
 import qualified Trace.Forward.Configuration.DataPoint as DPF
 import qualified Trace.Forward.Configuration.TraceObject as TF
@@ -184,9 +182,10 @@ runEKGAcceptor
   -> (ConnectionId addr -> IO ())
   -> RunMiniProtocol 'Mux.ResponderMode initiatorCtx (ResponderContext addr) LBS.ByteString IO Void ()
 runEKGAcceptor tracerEnv ekgConfig errorHandler =
-  acceptEKGMetricsResp
+  acceptMetricsResp
     ekgConfig
     (prepareMetricsStores tracerEnv . rcConnectionId)
+    (store tracerEnv . connIdToNodeId . rcConnectionId)
     (errorHandler . rcConnectionId)
 
 runTraceObjectsAcceptor

cardano-tracer/src/Cardano/Tracer/Acceptors/Utils.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TupleSections #-}
+{-# LANGUAGE ViewPatterns #-}
 
 {-# OPTIONS_GHC -Wno-redundant-constraints #-}
 
@@ -9,31 +10,41 @@ module Cardano.Tracer.Acceptors.Utils
   , prepareMetricsStores
   , removeDisconnectedNode
   , notifyAboutNodeDisconnected
+  , store
   ) where
 
 #if RTVIEW
 import           Cardano.Logging (SeverityS (..))
+#endif
+import qualified Cardano.Timeseries.Component as Timeseries
+import           Cardano.Timeseries.Domain.Types (MetricIdentifier)
+import           Cardano.Tracer.Environment
+#if RTVIEW
 import           Cardano.Tracer.Handlers.Notifications.Types
 import           Cardano.Tracer.Handlers.Notifications.Utils
 #endif
-import           Cardano.Tracer.Environment
+import           Cardano.Tracer.Time (getTimeMs)
 import           Cardano.Tracer.Types
 import           Cardano.Tracer.Utils
 import           Ouroboros.Network.Socket (ConnectionId (..))
 
 import           Control.Concurrent.STM (atomically)
 import           Control.Concurrent.STM.TVar (TVar, modifyTVar', newTVarIO)
 import qualified Data.Bimap as BM
+import           Data.Foldable
 import qualified Data.Map.Strict as M
+import           Data.Maybe (mapMaybe)
 import qualified Data.Set as S
-import           Data.Time.Clock.POSIX (getPOSIXTime)
 #if RTVIEW
 import           Data.Time.Clock.System (getSystemTime, systemToUTCTime)
 #endif
 import qualified System.Metrics as EKG
-import           System.Metrics.Store.Acceptor (MetricsLocalStore, emptyMetricsLocalStore)
+import           System.Metrics.ReqResp
+import           System.Metrics.Store.Acceptor (MetricsLocalStore, emptyMetricsLocalStore,
+                   storeMetrics)
 
 import           Trace.Forward.Utils.DataPoint (DataPointRequestor, initDataPointRequestor)
+import qualified Data.Text.Read as Text
 
 prepareDataPointRequestor
   :: Show addr
@@ -54,26 +65,17 @@ prepareMetricsStores
   -> IO (EKG.Store, TVar MetricsLocalStore)
 prepareMetricsStores TracerEnv{teConnectedNodes, teAcceptedMetrics} connId = do
   addConnectedNode teConnectedNodes connId
-  store <- EKG.newStore
+  st <- EKG.newStore
 
-  EKG.registerCounter "ekg.server_timestamp_ms" getTimeMs store
-  storesForNewNode <- (store ,) <$> newTVarIO emptyMetricsLocalStore
+  EKG.registerCounter "ekg.server_timestamp_ms" getTimeMs st
+  storesForNewNode <- (st ,) <$> newTVarIO emptyMetricsLocalStore
 
   atomically do
     modifyTVar' teAcceptedMetrics do
       M.insert (connIdToNodeId connId) storesForNewNode
 
   return storesForNewNode
 
-  where
-    -- forkServer definition of `getTimeMs'. The ekg frontend relies
-    -- on the "ekg.server_timestamp_ms" metric being in every
-    -- store. While forkServer adds that that automatically we must
-    -- manually add it.
-    -- url
-    --  + https://github.com/tvh/ekg-wai/blob/master/System/Remote/Monitoring/Wai.hs#L237-L238
-    getTimeMs = (round . (* 1000)) `fmap` getPOSIXTime
-
 addConnectedNode
   :: Show addr
   => ConnectedNodes
@@ -115,3 +117,23 @@ notifyAboutNodeDisconnected TracerEnvRTView{teEventsQueues} connId = do
 #else
 notifyAboutNodeDisconnected _ _ = pure ()
 #endif
+
+store :: TracerEnv -> NodeId -> (EKG.Store, TVar MetricsLocalStore) -> Response -> IO ()
+store tracerEnv (NodeId nodeId) (ekgStore, localStore) resp@(ResponseMetrics ms) = do
+  storeMetrics resp ekgStore localStore
+  for_ (teTimeseriesHandle tracerEnv) $ \h -> do
+    ts <- getTimeMs
+    Timeseries.insert h "node_id" nodeId (fromIntegral ts) (mapMaybe parseMetric ms)
+
+  where
+    numeralOnly :: MetricValue -> Maybe Double
+    numeralOnly (GaugeValue x) = Just (fromIntegral x)
+    numeralOnly (CounterValue x) = Just (fromIntegral x)
+    -- If the label is readable as double, accept it
+    numeralOnly (LabelValue (Text.double -> Right (x, ""))) = Just x
+    numeralOnly _ = Nothing
+
+    parseMetric :: (MetricName, MetricValue) -> Maybe (MetricIdentifier, Double)
+    parseMetric (k, numeralOnly -> Just v) = Just (k, v)
+    parseMetric _ = Nothing
+

cardano-tracer/src/Cardano/Tracer/Configuration.hs

@@ -164,6 +164,7 @@ data TracerConfig = TracerConfig
   , hasEKG           :: !(Maybe Endpoint)             -- ^ Endpoint for EKG web-page.
   , hasPrometheus    :: !(Maybe Endpoint)             -- ^ Endpoint for Prometheus web-page.
   , hasRTView        :: !(Maybe Endpoint)             -- ^ Endpoint for RTView web-page.
+  , hasTimeseries    :: !(Maybe Endpoint)
   , tlsCertificate   :: !(Maybe Certificate)
     -- | Socket for tracer's to reforward on. Second member of the triplet is the list of prefixes to reforward.
     -- Third member of the triplet is the forwarder config.

cardano-tracer/src/Cardano/Tracer/Environment.hs

@@ -6,6 +6,7 @@ module Cardano.Tracer.Environment
   ) where
 
 import           Cardano.Logging.Types
+import           Cardano.Timeseries.Component (TimeseriesHandle)
 import           Cardano.Tracer.Configuration
 #if RTVIEW
 import           Cardano.Tracer.Handlers.Notifications.Types
@@ -36,6 +37,7 @@ data TracerEnv = TracerEnv
   , teRegistry              :: !HandleRegistry
   , teStateDir              :: !(Maybe FilePath)
   , teMetricsHelp           :: ![(Text, Builder)]
+  , teTimeseriesHandle      :: !(Maybe TimeseriesHandle)
   }
 
 #if RTVIEW