add metrics for events processed and tweak k8s service roles

Author: saihaj

deployments/index.ts

@@ -175,12 +175,11 @@ const role = new k8s.rbac.v1.Role(
         resources: ["jobs"],
         verbs: ["create", "update", "get", "list", "watch", "delete"],
       },
-      // Uncomment this if you need to manage Pods/Services/Deployments
-      // {
-      //   apiGroups: [""], // "" indicates the core API group
-      //   resources: ["services"],
-      //   verbs: ["create", "update", "get", "list", "watch", "delete"],
-      // },
+      {
+        apiGroups: [""], // "" indicates the core API group
+        resources: ["services"],
+        verbs: ["create", "update", "get", "list", "watch", "delete"],
+      },
     ],
   },
   { provider },

substream-listener/src/index.mts

@@ -129,6 +129,7 @@ emitter.on("cursor", async (cursor) => {
 // Stream Blocks
 emitter.on("anyMessage", async (message) => {
   const transfers = (message.transfers || []) as any[];
+  const messageCount = transfers.length;
 
   const events = transfers.map((transfer) => {
     return svix.message.create(APP_ID, {
@@ -142,7 +143,15 @@ emitter.on("anyMessage", async (message) => {
 
   try {
     await Promise.all(events);
+    prometheus.substream_listener_processed_messages.inc(
+      { app_id: APP_ID, status: "success" },
+      messageCount,
+    );
   } catch (e) {
+    prometheus.substream_listener_processed_messages.inc(
+      { app_id: APP_ID, status: "error" },
+      messageCount,
+    );
     logger.error({ error: e }, "Error sending events to Svix");
   }
 });

substream-listener/src/prometheus.mts

@@ -99,6 +99,12 @@ const sessionGauge = new promClient.Gauge({
   ],
 });
 
+export const substream_listener_processed_messages = new promClient.Gauge({
+  name: "substream_listener_processed_messages",
+  help: "The number of messages processed by the emitter",
+  labelNames: ["app_id", "status"],
+});
+
 // ------------------------------------------------------------------
 
 export function onPrometheusMetrics(