added observability support to processor app

apps/processor/src/index.ts

@@ -1,8 +1,12 @@
+import "./instrumentation.js";
+
 import { mobileRouter } from "./domains/mobile/router.js";
+import { httpTelemetryMiddleware } from "./telemetry/http.middleware.js";
 import { serve } from "@hono/node-server";
 import { Hono } from "hono";
 
 const app = new Hono();
+app.use("*", httpTelemetryMiddleware);
 
 app.get("/health", (c) => {
     return c.json({ ok: true });

apps/processor/src/instrumentation.ts

@@ -0,0 +1,50 @@
+import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
+import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
+import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
+import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-proto";
+import { createAddHookMessageChannel } from "import-in-the-middle";
+import { BatchLogRecordProcessor } from "@opentelemetry/sdk-logs";
+import { NodeSDK } from "@opentelemetry/sdk-node";
+import { settings } from "@pkg/settings";
+import { register } from "node:module";
+
+const { registerOptions } = createAddHookMessageChannel();
+register("import-in-the-middle/hook.mjs", import.meta.url, registerOptions);
+
+const normalizedEndpoint = settings.otelExporterOtlpHttpEndpoint.startsWith(
+    "http",
+)
+    ? settings.otelExporterOtlpHttpEndpoint
+    : `http://${settings.otelExporterOtlpHttpEndpoint}`;
+
+if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT) {
+    process.env.OTEL_EXPORTER_OTLP_ENDPOINT = normalizedEndpoint;
+}
+
+const sdk = new NodeSDK({
+    serviceName: `${settings.otelServiceName}-processor`,
+    traceExporter: new OTLPTraceExporter(),
+    metricReader: new PeriodicExportingMetricReader({
+        exporter: new OTLPMetricExporter(),
+        exportIntervalMillis: 10_000,
+    }),
+    logRecordProcessors: [new BatchLogRecordProcessor(new OTLPLogExporter())],
+    instrumentations: [
+        getNodeAutoInstrumentations({
+            "@opentelemetry/instrumentation-winston": {
+                // We add OpenTelemetryTransportV3 explicitly in @pkg/logger.
+                disableLogSending: true,
+            },
+        }),
+    ],
+});
+
+sdk.start();
+
+const shutdown = () => {
+    void sdk.shutdown();
+};
+
+process.on("SIGTERM", shutdown);
+process.on("SIGINT", shutdown);

apps/processor/src/telemetry/http.middleware.ts

@@ -0,0 +1,77 @@
+import { context, metrics, SpanStatusCode, trace } from "@opentelemetry/api";
+import type { MiddlewareHandler } from "hono";
+
+const tracer = trace.getTracer("processor.http");
+const meter = metrics.getMeter("processor.http");
+
+const requestCount = meter.createCounter("processor.http.server.requests", {
+    description: "Total number of processor HTTP requests",
+});
+
+const requestDuration = meter.createHistogram(
+    "processor.http.server.duration",
+    {
+        description: "Processor HTTP request duration",
+        unit: "ms",
+    },
+);
+
+const activeRequests = meter.createUpDownCounter(
+    "processor.http.server.active_requests",
+    {
+        description: "Number of in-flight processor HTTP requests",
+    },
+);
+
+export const httpTelemetryMiddleware: MiddlewareHandler = async (c, next) => {
+    const startedAt = performance.now();
+    const method = c.req.method;
+    const route = c.req.path;
+
+    const span = tracer.startSpan(`http ${method} ${route}`, {
+        attributes: {
+            "http.request.method": method,
+            "url.path": route,
+        },
+    });
+
+    activeRequests.add(1, {
+        "http.request.method": method,
+        "url.path": route,
+    });
+
+    try {
+        await context.with(trace.setSpan(context.active(), span), next);
+
+        span.setAttribute("http.response.status_code", c.res.status);
+        span.setStatus({
+            code:
+                c.res.status >= 500
+                    ? SpanStatusCode.ERROR
+                    : SpanStatusCode.OK,
+        });
+    } catch (error) {
+        span.recordException(error as Error);
+        span.setStatus({
+            code: SpanStatusCode.ERROR,
+            message: "Unhandled processor request error",
+        });
+        throw error;
+    } finally {
+        const durationMs = performance.now() - startedAt;
+        const attrs = {
+            "http.request.method": method,
+            "http.response.status_code": c.res.status,
+            "url.path": route,
+        };
+
+        requestCount.add(1, attrs);
+        requestDuration.record(durationMs, attrs);
+        activeRequests.add(-1, {
+            "http.request.method": method,
+            "url.path": route,
+        });
+
+        span.end();
+    }
+};