Enable otel

Author: reednaa

witness/bun.lock

@@ -25,6 +25,11 @@
     "typescript": "^5"
   },
   "dependencies": {
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/exporter-metrics-otlp-http": "^0.213.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.213.0",
+    "@opentelemetry/sdk-node": "^0.213.0",
+    "@opentelemetry/semantic-conventions": "^1.40.0",
     "drizzle-orm": "^0.45.1",
     "elysia": "^1.4.27",
     "graphile-worker": "^0.16.6",

witness/package.json

@@ -1,10 +1,12 @@
+import { initTelemetry, shutdownTelemetry } from "../telemetry/index.ts";
 import { makeWorkerUtils } from "graphile-worker";
 import { loadApiConfig } from "../config/env.ts";
 import { createDb, closeDb } from "../db/client.ts";
 import { createApiServer } from "../api/server.ts";
 import { createHealthServer } from "../api/health.server.ts";
 
 async function main() {
+  initTelemetry();
   const config = loadApiConfig();
   const db = createDb(config.databaseUrl);
   const workerUtils = await makeWorkerUtils({ connectionString: config.databaseUrl });
@@ -36,6 +38,7 @@ async function main() {
     }
 
     await closeDb();
+    await shutdownTelemetry();
     process.exit(0);
   }
 

witness/src/entrypoints/api.ts

@@ -1,9 +1,11 @@
+import { initTelemetry, shutdownTelemetry } from "../telemetry/index.ts";
 import { loadWorkerConfig } from "../config/env.ts";
 import { createEnvSigner } from "../core/signing.ts";
 import { createDb, closeDb } from "../db/client.ts";
 import { setupWorker } from "../worker/setup.ts";
 
 async function main() {
+  initTelemetry();
   const config = loadWorkerConfig();
   const signer = createEnvSigner(config.witnessPrivateKey);
   const db = createDb(config.databaseUrl);
@@ -28,6 +30,7 @@ async function main() {
     }
 
     await closeDb();
+    await shutdownTelemetry();
     process.exit(0);
   }
 

witness/src/entrypoints/worker.ts

@@ -1,3 +1,4 @@
+import { initTelemetry, shutdownTelemetry } from "./telemetry/index.ts";
 import { loadConfig } from "./config/env.ts";
 import { createEnvSigner } from "./core/signing.ts";
 import { createDb, closeDb } from "./db/client.ts";
@@ -8,6 +9,7 @@ import { createApiServer } from "./api/server.ts";
 import path from "path";
 
 async function main() {
+  initTelemetry();
   const config = loadConfig();
 
   const signer = createEnvSigner(config.witnessPrivateKey);
@@ -54,6 +56,7 @@ async function main() {
 
     clearTimeout(timeout);
     await closeDb();
+    await shutdownTelemetry();
     process.exit(0);
   }
 

witness/src/index.ts

@@ -0,0 +1,55 @@
+import { trace, metrics, SpanStatusCode } from "@opentelemetry/api";
+import type { RpcClient } from "./client.ts";
+
+const tracer = trace.getTracer("vow-witness");
+const meter = metrics.getMeter("vow-witness");
+
+const durationHistogram = meter.createHistogram("vow.rpc.duration", {
+  description: "RPC call duration in milliseconds",
+  unit: "ms",
+});
+
+const errorsCounter = meter.createCounter("vow.rpc.errors", {
+  description: "RPC call error count",
+});
+
+type RpcMethod = "getBlock" | "getLogs" | "getBlockNumber";
+
+function wrapMethod<T extends (...args: any[]) => Promise<any>>(
+  method: T,
+  name: RpcMethod,
+  attrs: { "rpc.url": string; "chain.id": number }
+): T {
+  return (async (...args: any[]) => {
+    const spanAttrs = { "rpc.method": name, ...attrs };
+    const start = Date.now();
+
+    return tracer.startActiveSpan(`rpc.${name}`, { attributes: spanAttrs }, async (span) => {
+      try {
+        const result = await method(...args);
+        durationHistogram.record(Date.now() - start, spanAttrs);
+        return result;
+      } catch (err: any) {
+        durationHistogram.record(Date.now() - start, spanAttrs);
+        errorsCounter.add(1, { ...spanAttrs, "error.type": err?.constructor?.name ?? "Error" });
+        span.recordException(err);
+        span.setStatus({ code: SpanStatusCode.ERROR });
+        throw err;
+      } finally {
+        span.end();
+      }
+    });
+  }) as T;
+}
+
+export function instrumentRpcClient(
+  client: RpcClient,
+  attrs: { url: string; chainId: number }
+): RpcClient {
+  const commonAttrs = { "rpc.url": attrs.url, "chain.id": attrs.chainId };
+  return {
+    getBlock: wrapMethod(client.getBlock.bind(client), "getBlock", commonAttrs),
+    getLogs: wrapMethod(client.getLogs.bind(client), "getLogs", commonAttrs),
+    getBlockNumber: wrapMethod(client.getBlockNumber.bind(client), "getBlockNumber", commonAttrs),
+  };
+}

witness/src/rpc/instrumented-client.ts

@@ -0,0 +1,30 @@
+import { NodeSDK } from "@opentelemetry/sdk-node";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
+import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
+import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
+
+let sdk: NodeSDK | undefined;
+
+export function initTelemetry(): void {
+  if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT) return;
+
+  sdk = new NodeSDK({
+    traceExporter: new OTLPTraceExporter(),
+    metricReader: new PeriodicExportingMetricReader({
+      exporter: new OTLPMetricExporter(),
+      exportIntervalMillis: 15_000,
+    }),
+  });
+
+  sdk.start();
+  console.log("OpenTelemetry SDK started");
+}
+
+export async function shutdownTelemetry(): Promise<void> {
+  if (!sdk) return;
+  try {
+    await sdk.shutdown();
+  } catch (e) {
+    console.error("OpenTelemetry shutdown error:", e);
+  }
+}

witness/src/telemetry/index.ts

@@ -1,7 +1,9 @@
 import { type Task } from "graphile-worker";
 import { eq, sql } from "drizzle-orm";
 import { type Hex } from "viem";
+import { trace, SpanStatusCode } from "@opentelemetry/api";
 import { createRpcClient, type RpcClient } from "../rpc/client.ts";
+import { instrumentRpcClient } from "../rpc/instrumented-client.ts";
 import { fetchBlockConsistent, type ConsistentBlockResult } from "../rpc/consistency.ts";
 import { buildMerkleTree, ZERO_HASH } from "../core/merkle.ts";
 import { chains, rpcs, indexedBlocks, indexedEvents } from "../db/schema.ts";
@@ -30,6 +32,12 @@ export function createIndexBlockTask(
     const { chainId, blockNumber } = payload as IndexBlockPayload;
     const blockNumberBigInt = BigInt(blockNumber);
 
+    const tracer = trace.getTracer("vow-witness");
+    const span = tracer.startSpan("index-block", {
+      attributes: { "chain.id": chainId, "block.number": blockNumber },
+    });
+
+    try {
     // Fetch RPC URLs for this chain
     const rpcRows = await db
       .select({ url: rpcs.url })
@@ -43,7 +51,7 @@ export function createIndexBlockTask(
     }
 
     const rpcClients = rpcRows.map((row) =>
-      createRpcClient(row.url)
+      instrumentRpcClient(createRpcClient(row.url), { url: row.url, chainId })
     );
 
     // Fetch and validate block data from all RPCs
@@ -138,5 +146,12 @@ export function createIndexBlockTask(
         })
         .where(eq(chains.chainId, chainId));
     });
+    } catch (err: any) {
+      span.recordException(err);
+      span.setStatus({ code: SpanStatusCode.ERROR });
+      throw err;
+    } finally {
+      span.end();
+    }
   };
 }