wip: stats

Author: adamdotdevin

bun.lock

@@ -2,7 +2,6 @@ import { domain } from "./stage"
 import { EMAILOCTOPUS_API_KEY } from "./app"
 import { SECRET } from "./secret"
 import { lakeIngest } from "./lake"
-import { inferenceEventLake } from "./stats"
 
 ////////////////
 // DATABASE
@@ -242,7 +241,7 @@ const SALESFORCE_INSTANCE_URL = new sst.Secret("SALESFORCE_INSTANCE_URL")
 
 const logProcessor = new sst.cloudflare.Worker("LogProcessor", {
   handler: "packages/console/function/src/log-processor.ts",
-  link: [SECRET.HoneycombApiKey, lakeIngest, inferenceEventLake],
+  link: [SECRET.HoneycombApiKey, lakeIngest],
 })
 
 new sst.cloudflare.x.SolidStart("Console", {

infra/console.ts

@@ -57,7 +57,7 @@ const s3TablesCatalog = new aws.cloudcontrol.Resource(
 const athenaResultsBucket = new aws.s3.Bucket(
   "LakeAthenaResults",
   {
-    bucket: `opencode-${$app.stage}-datalake-athena-results`,
+    bucket: `opencode-${$app.stage}-lake-athena-results`,
     forceDestroy: $app.stage !== "production",
   },
 )
@@ -73,7 +73,7 @@ const firehoseErrorBucket = new aws.s3.Bucket(
 const athenaWorkgroup = new aws.athena.Workgroup(
   "LakeAthenaWorkgroup",
   {
-    name: `opencode-${$app.stage}-datalake`,
+    name: `opencode-${$app.stage}-lake-workgroup`,
     forceDestroy: $app.stage !== "production",
     configuration: {
       enforceWorkgroupConfiguration: true,
@@ -248,7 +248,7 @@ const ingestService = new sst.aws.Service("LakeIngestService", {
   },
   loadBalancer: {
     domain: {
-      name: `lake-ingest.${domain}`,
+      name: `lake.${domain}`,
       dns: sst.cloudflare.dns(),
     },
     rules: [

infra/lake.ts

@@ -72,7 +72,7 @@ const inferenceEventTable = new aws.s3tables.Table("LakeInferenceEventTable", {
   },
 })
 
-export const inferenceEventLake = new sst.Linkable("InferenceEventLake", {
+export const inferenceEvent = new sst.Linkable("InferenceEvent", {
   properties: {
     region: lakeRegion,
     catalog: lakeCatalog,
@@ -144,13 +144,10 @@ new sst.x.DevCommand("StatsStudio", {
 // APP
 ////////////////
 
-export const app = new sst.aws.SolidStart("Stats", {
+export const app = new sst.cloudflare.x.SolidStart("Stats", {
   path: "packages/stats/app",
   buildCommand: "bun run build",
-  domain: {
-    name: domain,
-    dns: sst.cloudflare.dns(),
-  },
+  domain,
   link: [database],
   environment: {
     PUBLIC_URL: `https://${domain}`,
@@ -177,7 +174,7 @@ export const statSync = new sst.aws.Service("StatsSyncService", {
     dockerfile: "packages/stats/server/Dockerfile",
   },
   command: ["bun", "src/stat-sync.ts"],
-  link: [database, inferenceEventLake, statsSyncConfig],
+  link: [database, inferenceEvent, statsSyncConfig],
   permissions: lakeQueryPermissions,
   scaling: {
     min: 1,

infra/stats.ts

@@ -1,9 +1,6 @@
 import { Resource } from "@opencode-ai/console-resource"
 import type { TraceItem } from "@cloudflare/workers-types"
 
-type MetricData = Record<string, unknown>
-type MetricEvent = { time: string; data: MetricData }
-
 export default {
   async tail(events: TraceItem[]) {
     for (const event of events) {
@@ -24,7 +21,7 @@ export default {
       )
         continue
 
-      let data: MetricData = {
+      let data: Record<string, unknown> = {
         "cf.continent": event.event.request.cf?.continent,
         "cf.country": event.event.request.cf?.country,
         "cf.city": event.event.request.cf?.city,
@@ -38,18 +35,20 @@ export default {
         ip: event.event.request.headers["x-real-ip"],
       }
       const time = new Date(event.eventTimestamp ?? Date.now()).toISOString()
-      const events: MetricEvent[] = []
-      for (const log of event.logs) {
-        for (const message of log.message) {
-          if (!message.startsWith("_metric:")) continue
-          const json = JSON.parse(message.slice(8)) as MetricData
-          data = { ...data, ...json }
-          if ("llm.error.code" in json) {
-            events.push({ time, data: { ...data, event_type: "llm.error" } })
-          }
-        }
-      }
-      events.push({ time, data: { ...data, event_type: "completions" } })
+      const events = [
+        ...event.logs.flatMap((log) =>
+          log.message.flatMap((message: string) => {
+            if (!message.startsWith("_metric:")) return []
+            const json = JSON.parse(message.slice(8)) as Record<string, unknown>
+            data = { ...data, ...json }
+            if ("llm.error.code" in json) {
+              return [{ time, data: { ...data, event_type: "llm.error" } }]
+            }
+            return []
+          }),
+        ),
+        { time, data: { ...data, event_type: "completions" } },
+      ]
       console.log(JSON.stringify(data, null, 2))
 
       const [honeycomb, lake] = await Promise.all([
@@ -78,7 +77,7 @@ export default {
   },
 }
 
-function toLakeEvent(time: string, data: MetricData) {
+function toLakeEvent(time: string, data: Record<string, unknown>) {
   const tokensInput = integer(data, "tokens.input")
   const tokensOutput = integer(data, "tokens.output")
   const tokensReasoning = integer(data, "tokens.reasoning")
@@ -90,16 +89,14 @@ function toLakeEvent(time: string, data: MetricData) {
   const source = string(data, "source")
 
   return {
-    _lake_database: Resource.InferenceEventLake.database,
-    _lake_table: Resource.InferenceEventLake.table,
-    _lake_operation: "insert",
+    type: "inference.event",
     event_timestamp: time,
     event_date: time.slice(0, 10),
     event_type: string(data, "event_type"),
     dataset: "zen",
     client: string(data, "client"),
     source,
-    tier: tier(source),
+    tier: source,
     provider: string(data, "provider"),
     provider_model: string(data, "provider.model"),
     model: string(data, "model"),
@@ -145,40 +142,32 @@ function toLakeEvent(time: string, data: MetricData) {
   }
 }
 
-function tier(source: string | undefined) {
-  if (source === "anonymous" || source === "free") return "Free"
-  if (source === "lite") return "Go"
-  if (source === "subscription" || source === "balance") return "Zen"
-  if (source === "byok") return "BYOK"
-  return undefined
-}
-
 function outputTps(tokens: number | undefined, firstByte: number | undefined, lastByte: number | undefined) {
   if (!tokens || !firstByte || !lastByte || lastByte <= firstByte) return undefined
   return Number(((tokens / (lastByte - firstByte)) * 1000).toFixed(6))
 }
 
-function string(data: MetricData, key: string) {
+function string(data: Record<string, unknown>, key: string) {
   const value = data[key]
   if (typeof value === "string") return value
   if (typeof value === "number" || typeof value === "boolean") return String(value)
   return undefined
 }
 
-function boolean(data: MetricData, key: string) {
+function boolean(data: Record<string, unknown>, key: string) {
   const value = data[key]
   if (typeof value === "boolean") return value
   if (typeof value === "string") return value === "true" ? true : value === "false" ? false : undefined
   return undefined
 }
 
-function integer(data: MetricData, key: string) {
+function integer(data: Record<string, unknown>, key: string) {
   const value = number(data, key)
   if (value === undefined) return undefined
   return Math.round(value)
 }
 
-function number(data: MetricData, key: string) {
+function number(data: Record<string, unknown>, key: string) {
   const value = data[key]
   if (typeof value === "number") return Number.isFinite(value) ? value : undefined
   if (typeof value === "string") {

packages/console/function/src/log-processor.ts

@@ -1,5 +1,5 @@
 export default {
   server: {
-    preset: "aws-lambda",
+    preset: "cloudflare-module",
   },
 }

packages/console/function/src/resource.d.ts

@@ -24,6 +24,7 @@
     "vite": "catalog:"
   },
   "devDependencies": {
+    "@cloudflare/workers-types": "catalog:",
     "@types/bun": "catalog:",
     "@types/d3-scale": "4.0.9",
     "@typescript/native-preview": "catalog:",

packages/stats/app/app.config.ts

@@ -3,7 +3,16 @@ import { nitro } from "nitro/vite"
 import { defineConfig, type PluginOption } from "vite"
 
 export default defineConfig({
-  plugins: [solidStart() as PluginOption, nitro({ preset: "aws-lambda" })],
+  plugins: [
+    solidStart() as PluginOption,
+    nitro({
+      compatibilityDate: "2024-09-19",
+      preset: "cloudflare-module",
+      cloudflare: {
+        nodeCompat: true,
+      },
+    }),
+  ],
   server: {
     allowedHosts: true,
   },

packages/stats/app/package.json

@@ -1,4 +1,4 @@
-import { Resource } from "sst"
+import { Resource } from "sst/resource"
 import { defineConfig } from "drizzle-kit"
 
 export default defineConfig({