feat: add prom metrics for webhook lister http requests

Author: saihaj

prometheus/prometheus.yml

@@ -7,8 +7,8 @@ scrape_configs:
     honor_timestamps: true
     scrape_interval: 3s
     scrape_timeout: 1s
-    metrics_path: /v1/metrics
+    metrics_path: /metrics
     scheme: http
     static_configs:
       - targets:
-          - host.docker.internal:4040
+          - host.docker.internal:10254

substream-listener/src/index.mts

@@ -2,7 +2,11 @@ import { createRouter, Response } from "fets";
 import { App } from "uWebSockets.js";
 import { isAddress } from "viem";
 import { sendWebhook } from "./send-webhook.mjs";
-import { registry } from "./prometheus.mjs";
+import {
+  invalidHttpRequests,
+  registry,
+  successfulHttpRequests,
+} from "./prometheus.mjs";
 
 // Creating a new router
 const router = createRouter({
@@ -59,36 +63,50 @@ const router = createRouter({
       },
     },
     async handler(req) {
+      const json = await req.json().catch(() => null);
+
+      if (!json) {
+        invalidHttpRequests.inc();
+        return Response.json(
+          { message: "Invalid JSON payload" },
+          { status: 400 },
+        );
+      }
+
       // Extracting the appId and startBlock from the request
-      const { appId, startBlock, contractAddress, substreamsToken } =
-        await req.json();
+      const { appId, startBlock, contractAddress, substreamsToken } = json;
 
       if (!appId) {
+        invalidHttpRequests.inc();
         return Response.json({ message: "appId is required" }, { status: 400 });
       }
 
       if (startBlock == null) {
+        invalidHttpRequests.inc();
         return Response.json(
           { message: "startBlock is required" },
           { status: 400 },
         );
       }
 
       if (!contractAddress) {
+        invalidHttpRequests.inc();
         return Response.json(
           { message: "contractAddress is required" },
           { status: 400 },
         );
       }
 
       if (!substreamsToken) {
+        invalidHttpRequests.inc();
         return Response.json(
           { message: "substreamsToken is required" },
           { status: 400 },
         );
       }
 
       if (!isAddress(contractAddress)) {
+        invalidHttpRequests.inc();
         return Response.json(
           { message: "contractAddress is invalid" },
           { status: 400 },
@@ -105,6 +123,8 @@ const router = createRouter({
         }),
       ]);
 
+      successfulHttpRequests.inc();
+
       // If the status code is not specified, it defaults to 200
       return Response.json({
         message: "Webhook registered",

substream-listener/src/prometheus.mts

@@ -20,6 +20,17 @@ function calculateHeadBlockTimeDrift(clock: Clock) {
 }
 
 // Counters
+
+export const invalidHttpRequests = new promClient.Counter({
+  name: "substreams_sink_invalid_http_requests",
+  help: "The number of invalid HTTP requests received",
+});
+
+export const successfulHttpRequests = new promClient.Counter({
+  name: "substreams_sink_successful_http_requests",
+  help: "The number of successful HTTP requests received",
+});
+
 export const substreams_sink_progress_message = new promClient.Counter({
   name: "substreams_sink_progress_message",
   help: "The number of progress message received",