Added trace_id to prometheus metrics and logs

package.json

@@ -41,13 +41,13 @@
         "prepublishOnly": "npm run build"
     },
     "dependencies": {
-        "@substreams/core": "^0.1.18",
-        "@substreams/manifest": "^0.0.8",
-        "@substreams/node": "^0.2.0",
+        "@substreams/core": "^0.1.19",
+        "@substreams/manifest": "^0.0.9",
+        "@substreams/node": "^0.2.1",
         "commander": "latest",
         "dotenv": "latest",
-        "node-fetch": "^3.3.2",
-        "prom-client": "^14.2.0",
+        "node-fetch": "latest",
+        "prom-client": "latest",
         "tslog": "latest"
     },
     "devDependencies": {

src/prometheus.ts

@@ -1,4 +1,4 @@
-import type { BlockScopedData, Clock } from "@substreams/core/proto";
+import type { BlockScopedData, SessionInit, Clock } from "@substreams/core/proto";
 import type { BlockEmitter } from "@substreams/node";
 import client, { Counter, Gauge, Summary, Histogram, type CounterConfiguration, type GaugeConfiguration, type SummaryConfiguration, type HistogramConfiguration } from "prom-client";
 
@@ -56,6 +56,7 @@ const substreams_sink_unknown_message = registerCounter("substreams_sink_unknown
 const substreams_sink_progress_message = registerCounter("substreams_sink_progress_message", "The number of progress message received", ["module"]);
 
 // Gauges
+const trace_id = registerGauge("trace_id", "Substreams session trace id", ["trace_id"]);
 const head_block_number = registerGauge("head_block_number", "Last processed block number");
 const head_block_time_drift = registerGauge("head_block_time_drift", "Head block time drift in seconds");
 const head_block_timestamp = registerGauge("head_block_timestamp", "Head block timestamp");
@@ -79,8 +80,14 @@ function updateBlockDataMetrics(block: BlockScopedData) {
     substreams_sink_backprocessing_completion?.set(1);
 }
 
+export function updateSessionMetrics(session: SessionInit) {
+    logger.info("trace_id", session.traceId);
+    trace_id?.labels({ trace_id: session.traceId }).set(1);
+}
+
 export function onPrometheusMetrics(emitter: BlockEmitter) {
     emitter.on("undo", () => substreams_sink_undo_message?.inc(1));
+    emitter.on("session", session => updateSessionMetrics(session));
     emitter.on("block", block => {
         updateBlockDataMetrics(block);
         if (block.clock) updateClockMetrics(block.clock);
@@ -101,4 +108,4 @@ export function handleManifest(emitter: BlockEmitter, moduleHash: string, option
         stopBlockNum: String(emitter.request.stopBlockNum),
         productionMode: String(emitter.request.productionMode)
     }).set(1);
-}
+}