feat: clickhouse container collecting api metrics

Taskfile.yml

@@ -14,15 +14,15 @@ tasks:
       - docker compose -f ./deployment/docker-compose.yaml up -d
 
 
-  seed:
+  migrate:
     cmds:
       - task: migrate-db
       - task: migrate-clickhouse
 
   migrate-clickhouse:
     env:
       GOOSE_DRIVER: clickhouse
-      GOOSE_DBSTRING: "tcp://127.0.0.1:9000"
+      GOOSE_DBSTRING: "tcp://default:password@127.0.0.1:9000"
       GOOSE_MIGRATION_DIR: ./apps/agent/pkg/clickhouse/schema
     cmds:
       - goose up

apps/agent/pkg/clickhouse/schema/001_create_requests_table.sql

@@ -21,7 +21,13 @@ CREATE TABLE default.raw_api_requests_v1(
     response_headers Array(String),
     response_body String,
     -- internal err.Error() string, empty if no error
-    error String
+    error String,
+
+    -- milliseconds
+    service_latency Int64,
+
+    user_agent String,
+    ip_address String
 
 )
 ENGINE = MergeTree()

apps/agent/pkg/clickhouse/schema/002_create_key_verifications_table.sql

@@ -12,7 +12,7 @@ CREATE TABLE default.raw_key_verifications_v1(
 
     -- Right now this is a 3 character airport code, but when we move to aws,
     -- this will be the region code such as `us-east-1`
-    region String,
+    region LowCardinality(String),
 
     -- Examples:
     -- - "VALID"
@@ -24,6 +24,8 @@ CREATE TABLE default.raw_key_verifications_v1(
     -- Empty string if the key has no identity
     identity_id String,
 
+
+
 )
 ENGINE = MergeTree()
 ORDER BY (workspace_id, key_space_id, key_id, time)

apps/api/package.json

@@ -36,6 +36,7 @@
     "@unkey/logs": "workspace:^",
     "@unkey/metrics": "workspace:^",
     "@unkey/rbac": "workspace:^",
+    "@unkey/clickhouse-zod": "workspace:^",
     "@unkey/schema": "workspace:^",
     "@unkey/worker-logging": "workspace:^",
     "hono": "^4.5.8",

apps/api/src/pkg/analytics.ts

@@ -1,4 +1,5 @@
 import { NoopTinybird, Tinybird } from "@chronark/zod-bird";
+import * as ch from "@unkey/clickhouse-zod";
 import { newId } from "@unkey/id";
 import { auditLogSchemaV1, unkeyAuditLogEvents } from "@unkey/schema/src/auditlog";
 import { ratelimitSchemaV1 } from "@unkey/schema/src/ratelimit-tinybird";
@@ -17,13 +18,17 @@ const dateToUnixMilli = z.string().transform((t) => new Date(t.split(" ").at(0)
 export class Analytics {
   public readonly readClient: Tinybird | NoopTinybird;
   public readonly writeClient: Tinybird | NoopTinybird;
+  private clickhouse: ch.Clickhouse;
 
   constructor(opts: {
     tinybirdToken?: string;
     tinybirdProxy?: {
       url: string;
       token: string;
     };
+    clickhouse?: {
+      url: string;
+    };
   }) {
     this.readClient = opts.tinybirdToken
       ? new Tinybird({ token: opts.tinybirdToken })
@@ -32,6 +37,8 @@ export class Analytics {
     this.writeClient = opts.tinybirdProxy
       ? new Tinybird({ token: opts.tinybirdProxy.token, baseUrl: opts.tinybirdProxy.url })
       : this.readClient;
+
+    this.clickhouse = opts.clickhouse ? new ch.Client({ url: opts.clickhouse.url }) : new ch.Noop();
   }
 
   public get ingestSdkTelemetry() {
@@ -93,6 +100,53 @@ export class Analytics {
     });
   }
 
+  public get insertKeyVerification() {
+    return this.clickhouse.insert({
+      table: "default.raw_key_verifications_v1",
+      schema: z.object({
+        request_id: z.string(),
+        time: z.number().int(),
+        workspace_id: z.string(),
+        key_space_id: z.string(),
+        key_id: z.string(),
+        region: z.string(),
+        outcome: z.enum([
+          "VALID",
+          "RATE_LIMITED",
+          "EXPIRED",
+          "DISABLED",
+          "FORBIDDEN",
+          "USAGE_EXCEEDED",
+          "INSUFFICIENT_PERMISSIONS",
+        ]),
+        identity_id: z.string().optional().default(""),
+      }),
+    });
+  }
+
+  public get insertApiRequest() {
+    return this.clickhouse.insert({
+      table: "default.raw_api_requests_v1",
+      schema: z.object({
+        request_id: z.string(),
+        time: z.number().int(),
+        workspace_id: z.string(),
+        host: z.string(),
+        method: z.string(),
+        path: z.string(),
+        request_headers: z.array(z.string()),
+        request_body: z.string(),
+        response_status: z.number().int(),
+        response_headers: z.array(z.string()),
+        response_body: z.string(),
+        error: z.string().optional().default(""),
+        service_latency: z.number().int(),
+        user_agent: z.string(),
+        ip_address: z.string(),
+      }),
+    });
+  }
+
   public get ingestKeyVerification() {
     return this.writeClient.buildIngestEndpoint({
       datasource: "key_verifications__v2",

apps/api/src/pkg/env.ts

@@ -28,6 +28,9 @@ export const zEnv = z.object({
     }),
   AGENT_URL: z.string().url(),
   AGENT_TOKEN: z.string(),
+
+  CLICKHOUSE_URL: z.string().optional(),
+
   SYNC_RATELIMIT_ON_NO_DATA: z
     .string()
     .optional()

apps/api/src/pkg/hono/env.ts

@@ -29,6 +29,8 @@ export type HonoEnv = {
     isolateId: string;
     isolateCreatedAt: number;
     requestId: string;
+    requestStartedAt: number;
+    workspaceId?: string;
     metricsContext: {
       keyId?: string;
       [key: string]: unknown;

apps/api/src/pkg/keys/service.ts

@@ -151,6 +151,7 @@ export class KeyService {
         });
         return res;
       }
+      c.set("workspaceId", res.val.key?.forWorkspaceId ?? res.val.key?.workspaceId);
 
       this.metrics.emit({
         metric: "metric.key.verification",

apps/api/src/pkg/middleware/init.ts

@@ -1,5 +1,6 @@
 import { Analytics } from "@/pkg/analytics";
 import { createConnection } from "@/pkg/db";
+
 import { KeyService } from "@/pkg/keys/service";
 import { AgentRatelimiter } from "@/pkg/ratelimit";
 import { DurableUsageLimiter, NoopUsageLimiter } from "@/pkg/usagelimit";
@@ -45,6 +46,8 @@ export function init(): MiddlewareHandler<HonoEnv> {
     const requestId = newId("request");
     c.set("requestId", requestId);
 
+    c.set("requestStartedAt", Date.now());
+
     c.res.headers.set("Unkey-Request-Id", requestId);
 
     const logger = new ConsoleLogger({
@@ -104,6 +107,11 @@ export function init(): MiddlewareHandler<HonoEnv> {
     const analytics = new Analytics({
       tinybirdProxy,
       tinybirdToken: c.env.TINYBIRD_TOKEN,
+      clickhouse: c.env.CLICKHOUSE_URL
+        ? {
+            url: c.env.CLICKHOUSE_URL,
+          }
+        : undefined,
     });
     const rateLimiter = new AgentRatelimiter({
       agent: { url: c.env.AGENT_URL, token: c.env.AGENT_TOKEN },

apps/api/src/pkg/middleware/metrics.ts

@@ -8,6 +8,8 @@ export function metrics(): MiddlewareHandler<HonoEnv> {
   return async (c, next) => {
     const { metrics, analytics, logger } = c.get("services");
 
+    let requestBody = await c.req.raw.clone().text();
+    requestBody = requestBody.replaceAll(/"key":\s*"[a-zA-Z0-9_]+"/g, '"key": "<REDACTED>"');
     const start = performance.now();
     const m = {
       isolateId: c.get("isolateId"),
@@ -79,6 +81,36 @@ export function metrics(): MiddlewareHandler<HonoEnv> {
       c.res.headers.append("Unkey-Version", c.env.VERSION);
       metrics.emit(m);
       c.executionCtx.waitUntil(metrics.flush());
+
+      const responseHeaders: Array<string> = [];
+      c.res.headers.forEach((v, k) => {
+        responseHeaders.push(`${k}: ${v}`);
+      });
+
+      c.executionCtx.waitUntil(
+        analytics.insertApiRequest({
+          request_id: c.get("requestId"),
+          time: c.get("requestStartedAt"),
+          workspace_id: c.get("workspaceId") ?? "",
+          host: new URL(c.req.url).host,
+          method: c.req.method,
+          path: c.req.path,
+          request_headers: Object.entries(c.req.header()).map(([k, v]) => {
+            if (k.toLowerCase() === "authorization") {
+              return `${k}: <REDACTED>`;
+            }
+            return `${k}: ${v}`;
+          }),
+          request_body: requestBody,
+          response_status: c.res.status,
+          response_headers: responseHeaders,
+          response_body: await c.res.clone().text(),
+          error: m.error ?? "",
+          service_latency: Date.now() - c.get("requestStartedAt"),
+          ip_address: c.req.header("True-Client-IP") ?? c.req.header("CF-Connecting-IP") ?? "",
+          user_agent: c.req.header("User-Agent") ?? "",
+        }),
+      );
     }
   };
 }

apps/api/src/routes/v1_keys_verifyKey.ratelimit_accuracy.test.ts

@@ -31,7 +31,7 @@ const testCases: {
   },
   {
     limit: 20,
-    duration: 5000,
+    duration: 30000,
     rps: 50,
     seconds: 60,
   },

apps/api/src/routes/v1_keys_verifyKey.ts

@@ -349,6 +349,22 @@ export const registerV1KeysVerifyKey = (app: App) =>
         : undefined,
     };
     c.executionCtx.waitUntil(
+      // new clickhouse
+      analytics.insertKeyVerification({
+        request_id: c.get("requestId"),
+        time: Date.now(),
+        workspace_id: val.key.workspaceId,
+        key_space_id: val.key.keyAuthId,
+        key_id: val.key.id,
+        // @ts-expect-error
+        region: c.req.raw.cf.colo ?? "",
+        outcome: val.code ?? "VALID",
+        identity_id: val.identity?.id,
+      }),
+    );
+
+    c.executionCtx.waitUntil(
+      // old tinybird
       analytics.ingestKeyVerification({
         workspaceId: val.key.workspaceId,
         apiId: val.api.id,

apps/dashboard/app/(app)/logs/page.tsx

@@ -0,0 +1,28 @@
+import { PageHeader } from "@/components/dashboard/page-header";
+import { getTenantId } from "@/lib/auth";
+import { getLogs } from "@/lib/clickhouse";
+import { db } from "@/lib/db";
+
+export const revalidate = 0;
+
+export default async function Page() {
+  const tenantId = getTenantId();
+
+  const workspace = await db.query.workspaces.findFirst({
+    where: (table, { and, eq, isNull }) =>
+      and(eq(table.tenantId, tenantId), isNull(table.deletedAt)),
+  });
+  if (!workspace) {
+    return <div>Workspace with tenantId: {tenantId} not found</div>;
+  }
+
+  const logs = await getLogs({ workspaceId: workspace.id, limit: 10 });
+
+  return (
+    <div>
+      <PageHeader title="Logs" />
+
+      <pre>{JSON.stringify(logs, null, 2)}</pre>
+    </div>
+  );
+}

apps/dashboard/lib/clickhouse/index.ts

@@ -0,0 +1,52 @@
+import { type Clickhouse, Client, Noop } from "@unkey/clickhouse-zod";
+import { z } from "zod";
+import { env } from "../env";
+
+// dummy example of how to query stuff from clickhouse
+export async function getLogs(args: { workspaceId: string; limit: number }) {
+  const { CLICKHOUSE_URL } = env();
+
+  const ch: Clickhouse = CLICKHOUSE_URL ? new Client({ url: CLICKHOUSE_URL }) : new Noop();
+  const query = ch.query({
+    query: `
+    SELECT
+      request_id,
+      time,
+      workspace_id,
+      host,
+      method,
+      path,
+      request_headers,
+      request_body,
+      response_status,
+      response_headers,
+      response_body,
+      error,
+      service_latency
+    FROM default.raw_api_requests_v1
+    WHERE workspace_id = {workspaceId: String}
+    ORDER BY time DESC
+    LIMIT {limit: Int}`,
+    params: z.object({
+      workspaceId: z.string(),
+      limit: z.number().int(),
+    }),
+    schema: z.object({
+      request_id: z.string(),
+      time: z.number().int(),
+      workspace_id: z.string(),
+      host: z.string(),
+      method: z.string(),
+      path: z.string(),
+      request_headers: z.array(z.string()),
+      request_body: z.string(),
+      response_status: z.number().int(),
+      response_headers: z.array(z.string()),
+      response_body: z.string(),
+      error: z.string(),
+      service_latency: z.number().int(),
+    }),
+  });
+
+  return query(args);
+}

apps/dashboard/lib/env.ts

@@ -37,6 +37,8 @@ export const env = () =>
       // - `ratelimit.*.create_namespace`
       // - `ratelimit.*.limit`
       UNKEY_ROOT_KEY: z.string().optional(),
+
+      CLICKHOUSE_URL: z.string().optional(),
     })
     .parse(process.env);
 

apps/dashboard/package.json

@@ -42,6 +42,7 @@
     "@trpc/react-query": "^10.45.2",
     "@trpc/server": "^10.45.2",
     "@unkey/billing": "workspace:^",
+    "@unkey/clickhouse-zod": "workspace:^",
     "@unkey/db": "workspace:^",
     "@unkey/encryption": "workspace:^",
     "@unkey/error": "workspace:^",