Made metrics instrumentation modular

packages/api/src/index.ts

@@ -12,3 +12,5 @@ export * from "./graphql/modules/MerkleWitnessResolver";
 export * from "./graphql/modules/LinkedMerkleWitnessResolver";
 export * from "./graphql/VanillaGraphqlModules";
 export * from "./metrics/OpenTelemetryServer";
+export * from "./metrics/OpenTelemetryTracer";
+export * from "./metrics/ModularizedInstrumentation";

packages/api/src/metrics/ModularizedInstrumentation.ts

@@ -0,0 +1,97 @@
+import { injectable, injectAll } from "tsyringe";
+import { PollInstrumentation, PushInstrumentation } from "@proto-kit/sequencer";
+import { InstrumentationBase } from "@opentelemetry/instrumentation";
+import { mapSequential, splitArray } from "@proto-kit/common";
+
+const INSTRUMENTATION_PREFIX = "protokit";
+
+@injectable()
+export class ModularizedInstrumentation extends InstrumentationBase<{}> {
+  // private readonly pushInstrumentations: PushInstrumentation[];
+
+  // private readonly pollInstrumentations: PollInstrumentation[];
+
+  public constructor(
+    @injectAll("Instrumentation")
+    private readonly instrumentations: (
+      | PushInstrumentation
+      | PollInstrumentation
+    )[]
+  ) {
+    super("protokit", "canary", {});
+  }
+
+  public async start() {
+    const { pushInstrumentations } = this.splitInstrumentations();
+    await mapSequential(pushInstrumentations, async (i) => await i.start());
+  }
+
+  public splitInstrumentations() {
+    // PushInstrumentation is abstract, therefore we can filter like this, while Poll
+    // isn't. But all remainders are of that type (bcs of the annotation) so it's fine
+    const split = splitArray(this.instrumentations, (i) =>
+      i instanceof PushInstrumentation ? "push" : "poll"
+    );
+    // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+    const pushInstrumentations = (split.push as PushInstrumentation[]) ?? [];
+    // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+    const pollInstrumentations = (split.poll as PollInstrumentation[]) ?? [];
+
+    return {
+      pushInstrumentations,
+      pollInstrumentations,
+    };
+  }
+
+  public initialize() {
+    const { pushInstrumentations, pollInstrumentations } =
+      this.splitInstrumentations();
+
+    pollInstrumentations.forEach((instrumentation) => {
+      const observableCounter = this.meter.createObservableCounter(
+        `${INSTRUMENTATION_PREFIX}_${instrumentation.name}`,
+        {
+          description: instrumentation.description,
+        }
+      );
+
+      this.meter.addBatchObservableCallback(
+        async (observableResult) => {
+          const metric = await instrumentation.poll();
+
+          observableResult.observe(observableCounter, metric);
+        },
+        [observableCounter]
+      );
+    });
+
+    pushInstrumentations.forEach((instrumentation) => {
+      const gauges = instrumentation.names.map((name) => {
+        const gauge = this.meter.createGauge(
+          `${INSTRUMENTATION_PREFIX}_${name}`,
+          {
+            description: instrumentation.description,
+          }
+        );
+        return [name, gauge] as const;
+      });
+
+      instrumentation.setPushFn((name, v) => {
+        const gauge = gauges.find(([candidate]) => candidate === name);
+        gauge![1].record(v);
+      });
+    });
+  }
+
+  // Called when a new `MeterProvider` is set
+  // the Meter (result of @opentelemetry/api's getMeter) is
+  // available as this.meter within this method
+  // eslint-disable-next-line no-underscore-dangle
+  override _updateMetricInstruments() {
+    if (this.instrumentations !== undefined) {
+      this.initialize();
+    }
+  }
+
+  init() {}
+}

packages/api/src/metrics/OpenTelemetryServer.ts

@@ -16,8 +16,8 @@ import { diag, DiagConsoleLogger, DiagLogLevel } from "@opentelemetry/api";
 import { inject } from "tsyringe";
 import { dependencyFactory, DependencyRecord, log } from "@proto-kit/common";
 
-import { SequencerInstrumentation } from "./SequencerInstrumentation";
 import { OpenTelemetryTracer } from "./OpenTelemetryTracer";
+import { ModularizedInstrumentation } from "./ModularizedInstrumentation";
 
 export type OpenTelemetryServerConfig = {
   metrics?: {
@@ -54,9 +54,8 @@ export class OpenTelemetryServer extends SequencerModule<OpenTelemetryServerConf
       config: { metrics, tracing },
     } = this;
 
-    // TODO Modularize Instrumentations
     const seqMetrics = this.sequencer.dependencyContainer.resolve(
-      SequencerInstrumentation
+      ModularizedInstrumentation
     );
 
     const metricReader =
@@ -91,6 +90,8 @@ export class OpenTelemetryServer extends SequencerModule<OpenTelemetryServerConf
 
     sdk.start();
 
+    await seqMetrics.start();
+
     // TODO Write logger to directly integrate with our logging library
     diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ERROR);
 

packages/sequencer/src/index.ts

@@ -136,3 +136,6 @@ export * from "./appChain/AppChain";
 export * from "./appChain/AppChainModule";
 export * from "./appChain/AreProofsEnabledFactory";
 export * from "./appChain/SharedDependencyFactory";
+export * from "./metrics/Instrumentation";
+export * from "./metrics/BlockProductionInstrumentation";
+export * from "./metrics/MempoolInstrumentation";

packages/sequencer/src/mempool/private/PrivateMempool.ts

@@ -1,4 +1,4 @@
-import { EventEmitter, log, noop } from "@proto-kit/common";
+import { dependencyFactory, EventEmitter, log, noop } from "@proto-kit/common";
 import { inject } from "tsyringe";
 
 import type { Mempool, MempoolEvents } from "../Mempool";
@@ -14,13 +14,15 @@ import { trace } from "../../logging/trace";
 import { IncomingMessagesService } from "../../settlement/messages/IncomingMessagesService";
 import { MempoolSorting } from "../sorting/MempoolSorting";
 import { DefaultMempoolSorting } from "../sorting/DefaultMempoolSorting";
+import { MempoolInstrumentation } from "../../metrics/MempoolInstrumentation";
 
 type PrivateMempoolConfig = {
   type?: "hybrid" | "private" | "based";
   targetBlockSize?: number;
 };
 
 @sequencerModule()
+@dependencyFactory()
 export class PrivateMempool
   extends SequencerModule<PrivateMempoolConfig>
   implements Mempool
@@ -43,6 +45,14 @@ export class PrivateMempool
     this.mempoolSorting = mempoolSorting ?? new DefaultMempoolSorting();
   }
 
+  public static dependencies() {
+    return {
+      MempoolInstrumentation: {
+        useClass: MempoolInstrumentation,
+      },
+    };
+  }
+
   private type() {
     return this.config.type ?? "hybrid";
   }

packages/sequencer/src/metrics/BlockProductionInstrumentation.ts

@@ -0,0 +1,37 @@
+import { inject, injectable } from "tsyringe";
+
+import type { BlockTriggerBase } from "../protocol/production/trigger/BlockTrigger";
+
+import { instrumentation, PushInstrumentation } from "./Instrumentation";
+
+@instrumentation()
+@injectable()
+export class BlockProductionInstrumentation extends PushInstrumentation {
+  names = ["block_height", "block_result", "batch_height"];
+
+  description = "L2 block height";
+
+  public constructor(
+    @inject("BlockTrigger")
+    private readonly trigger: BlockTriggerBase
+  ) {
+    super();
+  }
+
+  public async start() {
+    this.trigger.events.on("block-produced", (block) => {
+      this.pushValue("block_height", parseInt(block.height.toString(), 10));
+    });
+
+    this.trigger.events.on("block-metadata-produced", (block) => {
+      this.pushValue(
+        "block_result",
+        parseInt(block.block.height.toString(), 10)
+      );
+    });
+
+    this.trigger.events.on("batch-produced", (batch) => {
+      this.pushValue("batch_height", parseInt(batch.height.toString(), 10));
+    });
+  }
+}

packages/sequencer/src/metrics/Instrumentation.ts

@@ -0,0 +1,41 @@
+import { implement, Startable } from "@proto-kit/common";
+
+export function instrumentation() {
+  return implement<PollInstrumentation | PushInstrumentation>(
+    "Instrumentation"
+  );
+}
+
+export interface InstrumentationModule {
+  name: string;
+
+  description?: string;
+}
+
+export interface PollInstrumentation extends InstrumentationModule {
+  poll(): Promise<number>;
+}
+
+export abstract class PushInstrumentation implements Startable {
+  abstract names: string[];
+
+  abstract description?: string;
+
+  private pushFn?: (name: string, value: number) => void;
+
+  public setPushFn(f: (name: string, value: number) => void) {
+    this.pushFn = f;
+  }
+
+  protected pushValue(name: string, value: number) {
+    if (this.pushFn === undefined) {
+      throw new Error("Pushfn not initialized");
+    }
+    if (!this.names.includes(name)) {
+      throw new Error(`Name ${name} not specified in declared name list`);
+    }
+    this.pushFn(name, value);
+  }
+
+  abstract start(): Promise<void>;
+}

packages/sequencer/src/metrics/MempoolInstrumentation.ts

@@ -0,0 +1,22 @@
+import { inject, injectable } from "tsyringe";
+
+import type { PrivateMempool } from "../mempool/private/PrivateMempool";
+
+import { instrumentation, PollInstrumentation } from "./Instrumentation";
+
+@instrumentation()
+@injectable()
+export class MempoolInstrumentation implements PollInstrumentation {
+  name = "mempool_size";
+
+  description = "The size of the mempool";
+
+  public constructor(
+    @inject("Mempool")
+    private readonly mempool: PrivateMempool
+  ) {}
+
+  async poll() {
+    return await this.mempool.length();
+  }
+}

packages/sequencer/src/protocol/production/trigger/TimedBlockTrigger.ts

@@ -1,5 +1,5 @@
 import { inject, injectable } from "tsyringe";
-import { log } from "@proto-kit/common";
+import { dependencyFactory, log } from "@proto-kit/common";
 
 import { closeable, Closeable } from "../../../sequencer/builder/Closeable";
 import { BatchProducerModule } from "../BatchProducerModule";
@@ -13,6 +13,7 @@ import {
 } from "../../../settlement/BridgingModule";
 import { ensureNotBusy } from "../../../helpers/BusyGuard";
 import { SequencerStartupModule } from "../../../sequencer/SequencerStartupModule";
+import { BlockProductionInstrumentation } from "../../../metrics/BlockProductionInstrumentation";
 
 import { BlockTriggerBase } from "./BlockTrigger";
 
@@ -26,6 +27,7 @@ export interface TimedBlockTriggerConfig {
 
 @injectable()
 @closeable()
+@dependencyFactory()
 export class TimedBlockTrigger
   extends BlockTriggerBase<TimedBlockTriggerConfig>
   implements Closeable
@@ -59,6 +61,14 @@ export class TimedBlockTrigger
     );
   }
 
+  public static dependencies() {
+    return {
+      BlockProductionInstrumentation: {
+        useClass: BlockProductionInstrumentation,
+      },
+    };
+  }
+
   public async start(): Promise<void> {
     log.info("Starting timed block trigger");
     const { settlementInterval, blockInterval } = this.config;

packages/stack/src/scripts/graphql/server.ts

@@ -174,6 +174,10 @@ export async function startServer() {
         },
         metrics: {
           enabled: true,
+          prometheus: {
+            port: 9464,
+            host: "0.0.0.0",
+          },
         },
       },
 
@@ -239,7 +243,7 @@ export async function startServer() {
   let nonce = Number(as?.nonce.toString() ?? "0");
 
   setInterval(async () => {
-    const random = Math.floor(Math.random() * 5);
+    const random = 0; //Math.floor(Math.random() * 5);
     await mapSequential(range(0, random), async () => {
       const tx = await appChain.transaction(
         priv.toPublicKey(),