feature: add telemetry to the HTTP layer (#9116)

---------

Co-authored-by: adrien2p <adrien.deperetti@gmail.com>

packages/framework/framework/src/http/middlewares/authenticate-middleware.ts

@@ -20,7 +20,10 @@ const API_KEY_AUTH = "api-key"
 // This is the only hard-coded actor type, as API keys have special handling for now. We could also generalize API keys to carry the actor type with them.
 const ADMIN_ACTOR_TYPE = "user"
 
-type AuthType = typeof SESSION_AUTH | typeof BEARER_AUTH | typeof API_KEY_AUTH
+export type AuthType =
+  | typeof SESSION_AUTH
+  | typeof BEARER_AUTH
+  | typeof API_KEY_AUTH
 
 type MedusaSession = {
   auth_context: AuthContext
@@ -31,7 +34,7 @@ export const authenticate = (
   authType: AuthType | AuthType[],
   options: { allowUnauthenticated?: boolean; allowUnregistered?: boolean } = {}
 ): RequestHandler => {
-  const handler = async (
+  const authenticateMiddleware = async (
     req: MedusaRequest,
     res: MedusaResponse,
     next: NextFunction
@@ -105,7 +108,7 @@ export const authenticate = (
     res.status(401).json({ message: "Unauthorized" })
   }
 
-  return handler as unknown as RequestHandler
+  return authenticateMiddleware as unknown as RequestHandler
 }
 
 const getApiKeyInfo = async (req: MedusaRequest): Promise<ApiKeyDTO | null> => {

packages/framework/framework/src/http/router.ts

@@ -1,6 +1,13 @@
 import { parseCorsOrigins, promiseAll, wrapHandler } from "@medusajs/utils"
 import cors from "cors"
-import { type Express, json, Router, text, urlencoded } from "express"
+import {
+  type Express,
+  json,
+  RequestHandler,
+  Router,
+  text,
+  urlencoded,
+} from "express"
 import { readdir } from "fs/promises"
 import { extname, join, parse, sep } from "path"
 import {
@@ -8,15 +15,17 @@ import {
   HTTP_METHODS,
   MedusaRequest,
   MedusaResponse,
+  MiddlewareFunction,
   MiddlewareRoute,
   MiddlewaresConfig,
   MiddlewareVerb,
   ParserConfigArgs,
   RouteConfig,
   RouteDescriptor,
+  RouteHandler,
   RouteVerb,
 } from "./types"
-import { authenticate, errorHandler } from "./middlewares"
+import { authenticate, AuthType, errorHandler } from "./middlewares"
 import { configManager } from "../config"
 import { logger } from "../logger"
 
@@ -212,6 +221,24 @@ class ApiRoutesLoader {
    */
   readonly #sourceDir: string
 
+  /**
+   * Wrap the original route handler implementation for
+   * instrumentation.
+   */
+  static traceRoute?: (
+    handler: RouteHandler,
+    route: { route: string; method: string }
+  ) => RouteHandler
+
+  /**
+   * Wrap the original middleware handler implementation for
+   * instrumentation.
+   */
+  static traceMiddleware?: (
+    handler: RequestHandler | MiddlewareFunction,
+    route: { route: string; method?: string }
+  ) => RequestHandler
+
   constructor({
     app,
     activityId,
@@ -560,6 +587,27 @@ class ApiRoutesLoader {
     return
   }
 
+  /**
+   * Applies the route middleware on a route. Encapsulates the logic
+   * needed to pass the middleware via the trace calls
+   */
+  applyAuthMiddleware(
+    route: string,
+    actorType: string | string[],
+    authType: AuthType | AuthType[],
+    options?: { allowUnauthenticated?: boolean; allowUnregistered?: boolean }
+  ) {
+    let authenticateMiddleware = authenticate(actorType, authType, options)
+    if (ApiRoutesLoader.traceMiddleware) {
+      authenticateMiddleware = ApiRoutesLoader.traceMiddleware(
+        authenticateMiddleware,
+        { route: route }
+      )
+    }
+
+    this.#router.use(route, authenticateMiddleware)
+  }
+
   /**
    * Apply the route specific middlewares to the router,
    * this includes the cors, authentication and
@@ -629,20 +677,22 @@ class ApiRoutesLoader {
 
       // We only apply the auth middleware to store routes to populate the auth context. For actual authentication, users can just reapply the middleware.
       if (!config.optedOutOfAuth && config.routeType === "store") {
-        this.#router.use(
+        this.applyAuthMiddleware(
           descriptor.route,
-          authenticate("customer", ["bearer", "session"], {
+          "customer",
+          ["bearer", "session"],
+          {
             allowUnauthenticated: true,
-          })
+          }
         )
       }
 
       if (!config.optedOutOfAuth && config.routeType === "admin") {
-        // We probably don't want to allow access to all endpoints using an api key, but it will do until we revamp our routing.
-        this.#router.use(
-          descriptor.route,
-          authenticate("user", ["bearer", "session", "api-key"])
-        )
+        this.applyAuthMiddleware(descriptor.route, "user", [
+          "bearer",
+          "session",
+          "api-key",
+        ])
       }
 
       for (const route of routes) {
@@ -708,13 +758,26 @@ class ApiRoutesLoader {
           }`,
         })
 
+        let handler: RequestHandler | RouteHandler = route.handler
+
+        /**
+         * Give handler to the trace route handler for instrumentation
+         * from outside-in.
+         */
+        if (ApiRoutesLoader.traceRoute) {
+          handler = ApiRoutesLoader.traceRoute(handler, {
+            method: route.method!,
+            route: descriptor.route,
+          })
+        }
+
         /**
          * If the user hasn't opted out of error handling then
          * we wrap the handler in a try/catch block.
          */
-        const handler = shouldWrapHandler
-          ? wrapHandler(route.handler as Parameters<typeof wrapHandler>[0])
-          : route.handler
+        if (shouldWrapHandler) {
+          handler = wrapHandler(handler as Parameters<typeof wrapHandler>[0])
+        }
 
         this.#router[route.method!.toLowerCase()](descriptor.route, handler)
       }
@@ -756,7 +819,17 @@ class ApiRoutesLoader {
           message: `Registering middleware [${method}] - ${route.matcher}`,
         })
 
-        this.#router[method.toLowerCase()](route.matcher, ...route.middlewares)
+        let middlewares = route.middlewares
+        if (ApiRoutesLoader.traceMiddleware) {
+          middlewares = middlewares.map((middleware) =>
+            ApiRoutesLoader.traceMiddleware!(middleware, {
+              route: String(route.matcher),
+              method,
+            })
+          )
+        }
+
+        this.#router[method.toLowerCase()](route.matcher, ...middlewares)
       }
     }
   }
@@ -840,6 +913,27 @@ export class RoutesLoader {
    */
   readonly #sourceDir: string | string[]
 
+  static instrument: {
+    /**
+     * Instrument middleware function calls by wrapping the original
+     * middleware handler inside a custom implementation
+     */
+    middleware: (callback: (typeof ApiRoutesLoader)["traceMiddleware"]) => void
+
+    /**
+     * Instrument route handler function calls by wrapping the original
+     * middleware handler inside a custom implementation
+     */
+    route: (callback: (typeof ApiRoutesLoader)["traceRoute"]) => void
+  } = {
+    middleware(callback) {
+      ApiRoutesLoader.traceMiddleware = callback
+    },
+    route(callback) {
+      ApiRoutesLoader.traceRoute = callback
+    },
+  }
+
   constructor({
     app,
     activityId,

packages/framework/framework/src/http/types.ts

@@ -36,7 +36,7 @@ export type AsyncRouteHandler = (
   res: MedusaResponse
 ) => Promise<void>
 
-type RouteHandler = SyncRouteHandler | AsyncRouteHandler
+export type RouteHandler = SyncRouteHandler | AsyncRouteHandler
 
 export type RouteImplementation = {
   method?: RouteVerb

packages/medusa/package.json

@@ -22,6 +22,11 @@
   "license": "MIT",
   "devDependencies": {
     "@medusajs/types": "^1.11.16",
+    "@opentelemetry/instrumentation": "^0.53.0",
+    "@opentelemetry/instrumentation-pg": "^0.44.0",
+    "@opentelemetry/resources": "^1.26.0",
+    "@opentelemetry/sdk-node": "^0.53.0",
+    "@opentelemetry/sdk-trace-node": "^1.26.0",
     "@swc/jest": "^0.2.36",
     "@types/express": "^4.17.17",
     "@types/ioredis": "^4.28.10",

packages/medusa/src/api/admin/products/utils/maybe-apply-price-lists-filter.ts

@@ -7,7 +7,11 @@ import { NextFunction } from "express"
 import { MedusaRequest } from "../../../../types/routing"
 
 export function maybeApplyPriceListsFilter() {
-  return async (req: MedusaRequest, _, next: NextFunction) => {
+  return async function applyPriceListsFilter(
+    req: MedusaRequest,
+    _,
+    next: NextFunction
+  ) {
     const filterableFields: HttpTypes.AdminProductListParams =
       req.filterableFields
 

packages/medusa/src/api/utils/maybe-apply-link-filter.ts

@@ -12,7 +12,7 @@ export function maybeApplyLinkFilter({
   filterableField,
   filterByField = "id",
 }) {
-  return async (req: MedusaRequest, _, next: NextFunction) => {
+  return async function linkFilter(req: MedusaRequest, _, next: NextFunction) {
     const filterableFields = req.filterableFields
 
     if (!filterableFields?.[filterableField]) {
@@ -76,27 +76,26 @@ export function maybeApplyLinkFilter({
   }
 */
 function transformFilterableFields(filterableFields: Record<string, unknown>) {
-  const result = {};
+  const result = {}
   for (const key of Object.keys(filterableFields)) {
-    const value = filterableFields[key];
-    const keys = key.split(".");
-    let current = result;
+    const value = filterableFields[key]
+    const keys = key.split(".")
+    let current = result
 
     // Iterate over the keys, creating nested objects as needed
     for (let i = 0; i < keys.length; i++) {
-      const part = keys[i];
-      current[part] ??= {};
+      const part = keys[i]
+      current[part] ??= {}
 
       if (i === keys.length - 1) {
         // If its the last key, assign the value
-        current[part] = value;
-        break;
+        current[part] = value
+        break
       }
 
-      current = current[part];
+      current = current[part]
     }
   }
 
-  return result;
+  return result
 }
-

packages/medusa/src/api/utils/validate-body.ts

@@ -14,7 +14,11 @@ export function validateAndTransformBody(
   res: MedusaResponse,
   next: NextFunction
 ) => Promise<void> {
-  return async (req: MedusaRequest, _: MedusaResponse, next: NextFunction) => {
+  return async function validateBody(
+    req: MedusaRequest,
+    _: MedusaResponse,
+    next: NextFunction
+  ) {
     try {
       let schema: z.ZodObject<any, any> | z.ZodEffects<any, any>
       if (typeof zodSchema === "function") {

packages/medusa/src/api/utils/validate-query.ts

@@ -65,7 +65,11 @@ export function validateAndTransformQuery<TEntity extends BaseEntity>(
   res: MedusaResponse,
   next: NextFunction
 ) => Promise<void> {
-  return async (req: MedusaRequest, _: MedusaResponse, next: NextFunction) => {
+  return async function validateQuery(
+    req: MedusaRequest,
+    _: MedusaResponse,
+    next: NextFunction
+  ) {
     try {
       const allowed = (req.allowed ?? queryConfig.allowed ?? []) as string[]
       delete req.allowed

packages/medusa/src/commands/start.ts

@@ -1,21 +1,59 @@
+import path from "path"
 import express from "express"
 import { track } from "medusa-telemetry"
 import { scheduleJob } from "node-schedule"
 
-import { gqlSchemaToTypes, logger } from "@medusajs/framework"
 import { GracefulShutdownServer } from "@medusajs/utils"
+import http, { IncomingMessage, ServerResponse } from "http"
+import { gqlSchemaToTypes, logger } from "@medusajs/framework"
+
 import loaders from "../loaders"
-import path from "path"
 
 const EVERY_SIXTH_HOUR = "0 */6 * * *"
 const CRON_SCHEDULE = EVERY_SIXTH_HOUR
 
-export default async function ({ port, directory, types }) {
-  async function start() {
+/**
+ * Imports the "instrumentation.js" file from the root of the
+ * directory and invokes the register function. The existence
+ * of this file is optional, hence we ignore "ENOENT"
+ * errors.
+ */
+async function registerInstrumentation(directory: string) {
+  try {
+    const instrumentation = await import(
+      path.join(directory, "instrumentation.js")
+    )
+    if (typeof instrumentation.register === "function") {
+      logger.info("OTEL registered")
+      instrumentation.register()
+    }
+  } catch (error) {
+    if (!["ENOENT", "MODULE_NOT_FOUND"].includes(error.code)) {
+      throw error
+    }
+  }
+}
+
+async function start({ port, directory, types }) {
+  async function internalStart() {
     track("CLI_START")
+    await registerInstrumentation(directory)
 
     const app = express()
 
+    const http_ = http.createServer(async (req, res) => {
+      await start.traceRequestHandler(
+        async () => {
+          return new Promise((resolve) => {
+            res.on("finish", resolve)
+            app(req, res)
+          })
+        },
+        req,
+        res
+      )
+    })
+
     try {
       const { shutdown, gqlSchema } = await loaders({
         directory,
@@ -33,7 +71,7 @@ export default async function ({ port, directory, types }) {
 
       const serverActivity = logger.activity(`Creating server`)
       const server = GracefulShutdownServer.create(
-        app.listen(port).on("listening", () => {
+        http_.listen(port).on("listening", () => {
           logger.success(serverActivity, `Server is ready on port: ${port}`)
           track("CLI_START_COMPLETED")
         })
@@ -68,5 +106,19 @@ export default async function ({ port, directory, types }) {
     }
   }
 
-  await start()
+  await internalStart()
 }
+
+/**
+ * Wrap request handler inside custom implementation to enabled
+ * instrumentation.
+ */
+start.traceRequestHandler = async (
+  requestHandler: () => Promise<void>,
+  _: IncomingMessage,
+  __: ServerResponse
+) => {
+  return await requestHandler()
+}
+
+export default start

packages/medusa/src/index.js

@@ -5,3 +5,4 @@ export * from "./types/middlewares"
 export * from "./types/routing"
 export * from "./types/subscribers"
 export * from "./utils"
+export * from "./instrumentation"

packages/medusa/src/instrumentation/index.ts

@@ -0,0 +1,151 @@
+import { snakeCase } from "lodash"
+import { NodeSDK } from "@opentelemetry/sdk-node"
+import { Resource } from "@opentelemetry/resources"
+import { SpanStatusCode } from "@opentelemetry/api"
+import { RoutesLoader, Tracer } from "@medusajs/framework"
+import {
+  type SpanExporter,
+  SimpleSpanProcessor,
+} from "@opentelemetry/sdk-trace-node"
+import { PgInstrumentation } from "@opentelemetry/instrumentation-pg"
+import type { Instrumentation } from "@opentelemetry/instrumentation"
+
+import start from "../commands/start"
+
+const EXCLUDED_RESOURCES = [".vite", "virtual:"]
+
+function shouldExcludeResource(resource: string) {
+  return EXCLUDED_RESOURCES.some((excludedResource) =>
+    resource.includes(excludedResource)
+  )
+}
+
+/**
+ * Instrumenting the first touch point of the Http layer to report traces to
+ * OpenTelemetry
+ */
+export function instrumentHttpLayer() {
+  const HTTPTracer = new Tracer("@medusajs/http", "2.0.0")
+
+  start.traceRequestHandler = async (requestHandler, req, res) => {
+    if (shouldExcludeResource(req.url!)) {
+      return await requestHandler()
+    }
+
+    const traceName = `${req.method} ${req.url}`
+    await HTTPTracer.trace(traceName, async (span) => {
+      span.setAttributes({
+        "http.url": req.url,
+        "http.method": req.method,
+        ...req.headers,
+      })
+
+      try {
+        await requestHandler()
+      } finally {
+        span.setAttributes({ "http.statusCode": res.statusCode })
+        span.end()
+      }
+    })
+  }
+
+  /**
+   * Instrumenting the route handler to report traces to
+   * OpenTelemetry
+   */
+  RoutesLoader.instrument.route((handler) => {
+    const traceName = `route: ${
+      handler.name ? snakeCase(handler.name) : `anonymous`
+    }`
+
+    return async (req, res) => {
+      if (shouldExcludeResource(req.originalUrl)) {
+        return await handler(req, res)
+      }
+
+      await HTTPTracer.trace(traceName, async (span) => {
+        try {
+          await handler(req, res)
+        } catch (error) {
+          span.setStatus({
+            code: SpanStatusCode.ERROR,
+            message: error.message || "Failed",
+          })
+        } finally {
+          span.end()
+        }
+      })
+    }
+  })
+
+  /**
+   * Instrumenting the middleware handler to report traces to
+   * OpenTelemetry
+   */
+  RoutesLoader.instrument.middleware((handler) => {
+    const traceName = `middleware: ${
+      handler.name ? snakeCase(handler.name) : `anonymous`
+    }`
+
+    return async (req, res, next) => {
+      if (shouldExcludeResource(req.originalUrl)) {
+        return handler(req, res, next)
+      }
+
+      await HTTPTracer.trace(traceName, async (span) => {
+        return new Promise<void>((resolve, reject) => {
+          const _next = (error?: any) => {
+            if (error) {
+              span.setStatus({
+                code: SpanStatusCode.ERROR,
+                message: error.message || "Failed",
+              })
+              span.end()
+              reject(error)
+            } else {
+              span.end()
+              resolve()
+            }
+          }
+
+          handler(req, res, _next)
+        })
+      })
+        .catch(next)
+        .then(next)
+    }
+  })
+}
+
+/**
+ * A helper function to configure the OpenTelemetry SDK with some defaults.
+ * For better/more control, please configure the SDK manually.
+ *
+ * You will have to install the following packages within your app for
+ * telemetry to work
+ *
+ * - @opentelemetry/sdk-node
+ * - @opentelemetry/resources
+ * - @opentelemetry/sdk-trace-node
+ * - @opentelemetry/instrumentation-pg
+ * - @opentelemetry/instrumentation
+ */
+export function registerOtel(options: {
+  serviceName: string
+  exporter: SpanExporter
+  instrumentations?: Instrumentation[]
+}) {
+  const sdk = new NodeSDK({
+    serviceName: options.serviceName,
+    resource: new Resource({
+      "service.name": options.serviceName,
+    }),
+    spanProcessor: new SimpleSpanProcessor(options.exporter),
+    instrumentations: [
+      new PgInstrumentation(),
+      ...(options.instrumentations || []),
+    ],
+  })
+  sdk.start()
+  return sdk
+}