feat: emit CloudWatch metric for inflight requests in gateway (#1850)

- Add AWS CloudWatch SDK client dependency
- Create CloudWatch metrics service to track inflight requests per task
- Add middleware to increment/decrement inflight request counter
- Emit metric every 10 seconds with TaskId and Environment dimensions
- Only enable when AWS credentials are available (AWS_ACCESS_KEY and AWS_ACCESS_SECRET)
- Metric name: inflight_per_task under Latitude/Gateway namespace

Author: geclos

apps/gateway/package.json

@@ -15,23 +15,24 @@
     "datadog:sourcemaps": "datadog-ci sourcemaps upload dist --service=latitude-llm-gateway --minified-path-prefix=/app/apps/gateway/dist --release-version=${RELEASE_VERSION:-unknown}"
   },
   "dependencies": {
+    "@aws-sdk/client-cloudwatch": "^3.914.0",
     "@hono/node-server": "1.13.2",
     "@hono/swagger-ui": "0.4.1",
     "@hono/zod-openapi": "1.1.1",
     "@latitude-data/constants": "workspace:^",
     "@latitude-data/core": "workspace:^",
     "@latitude-data/env": "workspace:^",
-    "@latitude-data/telemetry": "workspace:*",
     "@latitude-data/sdk": "workspace:^",
+    "@latitude-data/telemetry": "workspace:*",
     "@t3-oss/env-core": "0.13.8",
+    "dd-trace": "catalog:",
     "drizzle-orm": "catalog:",
     "hono": "4.9.7",
     "ioredis": "5.6.0",
     "lodash-es": "4.17.21",
     "promptl-ai": "catalog:",
     "rate-limiter-flexible": "5.0.3",
-    "zod": "catalog:",
-    "dd-trace": "catalog:"
+    "zod": "catalog:"
   },
   "devDependencies": {
     "@datadog/datadog-ci": "catalog:",

apps/gateway/src/middlewares/inflightRequests.ts

@@ -0,0 +1,14 @@
+import { MiddlewareHandler } from 'hono'
+import { cloudWatchMetrics } from '../services/cloudwatchMetrics'
+
+export const inflightRequestsMiddleware = (): MiddlewareHandler => {
+  return async (_c, next) => {
+    cloudWatchMetrics.incrementInflightRequests()
+
+    try {
+      await next()
+    } finally {
+      cloudWatchMetrics.decrementInflightRequests()
+    }
+  }
+}

apps/gateway/src/routes/app.ts

@@ -3,13 +3,15 @@ import { logger } from 'hono/logger'
 import authMiddleware from '$/middlewares/auth'
 import { rateLimitMiddleware } from '$/middlewares/rateLimit'
 import errorHandlerMiddleware from '$/middlewares/errorHandler'
+import { inflightRequestsMiddleware } from '$/middlewares/inflightRequests'
 
 import createApp from '$/openApi/createApp'
 import configureOpenAPI from '$/openApi/configureOpenAPI'
 import { configureApiRoutes } from './api'
 import { configureWebhookRoutes } from './webhook'
 import { memoryUsageMiddleware } from '$/middlewares/memoryLogger'
 import { tracerMiddleware } from '$/middlewares/tracer'
+import { env } from '@latitude-data/env'
 
 const app = createApp()
 
@@ -30,6 +32,10 @@ configureWebhookRoutes(app)
 app.use(rateLimitMiddleware())
 app.use(authMiddleware())
 
+if (env.AWS_ACCESS_KEY && env.AWS_ACCESS_SECRET) {
+  app.use(inflightRequestsMiddleware())
+}
+
 configureApiRoutes(app)
 
 app.onError(errorHandlerMiddleware)

apps/gateway/src/server.ts

@@ -6,6 +6,7 @@ import cluster from 'cluster'
 import os from 'os'
 
 import { env } from '@latitude-data/env'
+import { cloudWatchMetrics } from './services/cloudwatchMetrics'
 
 const HOSTNAME = env.GATEWAY_BIND_ADDRESS
 const PORT = env.GATEWAY_BIND_PORT
@@ -60,11 +61,26 @@ if (cluster.isPrimary) {
       console.log(
         `Worker ${process.pid} listening on http://${HOSTNAME}:${PORT}`,
       )
+
+      if (env.AWS_ACCESS_KEY && env.AWS_ACCESS_SECRET) {
+        cloudWatchMetrics.startPeriodicEmission()
+        console.log('Started CloudWatch metrics emission for inflight requests')
+      }
     },
   )
 
-  process.on('SIGTERM', () => gracefulShutdown(server))
-  process.on('SIGINT', () => gracefulShutdown(server))
+  process.on('SIGTERM', () => {
+    if (env.AWS_ACCESS_KEY && env.AWS_ACCESS_SECRET) {
+      cloudWatchMetrics.stopPeriodicEmission()
+    }
+    gracefulShutdown(server)
+  })
+  process.on('SIGINT', () => {
+    if (env.AWS_ACCESS_KEY && env.AWS_ACCESS_SECRET) {
+      cloudWatchMetrics.stopPeriodicEmission()
+    }
+    gracefulShutdown(server)
+  })
 
   process.on('uncaughtException', function (err) {
     captureException(err)

apps/gateway/src/services/cloudwatchMetrics.ts

@@ -0,0 +1,86 @@
+import {
+  CloudWatchClient,
+  PutMetricDataCommand,
+} from '@aws-sdk/client-cloudwatch'
+import { env } from '@latitude-data/env'
+
+class CloudWatchMetricsService {
+  private client: CloudWatchClient | null = null
+  private inflightRequests = 0
+  private namespace = 'Latitude/Gateway'
+  private metricName = 'inflight_per_task'
+  private intervalId: ReturnType<typeof setInterval> | null = null
+
+  constructor() {
+    if (this.isConfigured()) {
+      this.client = new CloudWatchClient({
+        region: env.AWS_REGION || 'us-east-1',
+        credentials: {
+          accessKeyId: env.AWS_ACCESS_KEY!,
+          secretAccessKey: env.AWS_ACCESS_SECRET!,
+        },
+      })
+    }
+  }
+
+  private isConfigured(): boolean {
+    return Boolean(env.AWS_ACCESS_KEY && env.AWS_ACCESS_SECRET)
+  }
+
+  incrementInflightRequests(): void {
+    this.inflightRequests++
+  }
+
+  decrementInflightRequests(): void {
+    this.inflightRequests = Math.max(0, this.inflightRequests - 1)
+  }
+
+  getInflightRequests(): number {
+    return this.inflightRequests
+  }
+
+  async emitMetric(): Promise<void> {
+    if (!this.isConfigured() || !this.client) {
+      return
+    }
+
+    try {
+      const command = new PutMetricDataCommand({
+        Namespace: this.namespace,
+        MetricData: [
+          {
+            MetricName: this.metricName,
+            Value: this.inflightRequests,
+            Unit: 'Count',
+            Timestamp: new Date(),
+          },
+        ],
+      })
+
+      await this.client.send(command)
+    } catch (error) {
+      console.error('Failed to emit CloudWatch metric:', error)
+    }
+  }
+
+  startPeriodicEmission(intervalMs = 10000): void {
+    if (!this.isConfigured() || this.intervalId) {
+      return
+    }
+
+    this.intervalId = setInterval(() => {
+      this.emitMetric()
+    }, intervalMs)
+
+    this.intervalId.unref()
+  }
+
+  stopPeriodicEmission(): void {
+    if (this.intervalId) {
+      clearInterval(this.intervalId)
+      this.intervalId = null
+    }
+  }
+}
+
+export const cloudWatchMetrics = new CloudWatchMetricsService()