From 8bb2ccf82944eaec6c40b0bcdd03886bbfe86ef2 Mon Sep 17 00:00:00 2001 From: Adam <2363879+adamdotdevin@users.noreply.github.com> Date: Thu, 1 Oct 2026 10:12:55 -0500 Subject: [PATCH] chore(stats): retire the S3 data lake (#52514) --- bun.lock | 5 - infra/console.ts | 6 +- infra/lake.ts | 89 +++-------- infra/stats.ts | 20 +-- .../console/function/src/log-processor.ts | 134 ++-------------- packages/stats/core/package.json | 2 - packages/stats/core/src/athena.ts | 151 ------------------ packages/stats/core/src/index.ts | 1 - packages/stats/core/src/resource.d.ts | 9 -- 9 files changed, 37 insertions(+), 380 deletions(-) delete mode 100644 packages/stats/core/src/athena.ts diff --git a/bun.lock b/bun.lock index cb30678c4fe5..89315fb65a62 100644 --- a/bun.lock +++ b/bun.lock @@ -880,7 +880,6 @@ "name": "@opencode-ai/stats-core", "version": "1.18.34", "dependencies": { - "@aws-sdk/client-athena": "3.933.0", "@planetscale/database": "1.19.0", "drizzle-orm": "catalog:", "effect": "catalog:", @@ -1268,8 +1267,6 @@ "@aws-crypto/util": ["@aws-crypto/util@5.2.0", "", { "dependencies": { "@aws-sdk/types": "^3.222.0", "@smithy/util-utf8": "^2.0.0", "tslib": "^2.6.2" } }, "sha512-4RkU9EsI6ZpBve5fseQlGNUWKMa1RLPQ1dnjnQoe07ldfIzcsGb5hC5W0Dm7u423KWzawlrpbjXBrXCEv9zazQ=="], - "@aws-sdk/client-athena": ["@aws-sdk/client-athena@3.933.0", "", { "dependencies": { "@aws-crypto/sha256-browser": "5.2.0", "@aws-crypto/sha256-js": "5.2.0", "@aws-sdk/core": "3.932.0", "@aws-sdk/credential-provider-node": "3.933.0", "@aws-sdk/middleware-host-header": "3.930.0", "@aws-sdk/middleware-logger": "3.930.0", "@aws-sdk/middleware-recursion-detection": "3.933.0", "@aws-sdk/middleware-user-agent": "3.932.0", "@aws-sdk/region-config-resolver": "3.930.0", "@aws-sdk/types": "3.930.0", "@aws-sdk/util-endpoints": "3.930.0", "@aws-sdk/util-user-agent-browser": "3.930.0", "@aws-sdk/util-user-agent-node": "3.932.0", "@smithy/config-resolver": "^4.4.3", "@smithy/core": "^3.18.2", "@smithy/fetch-http-handler": "^5.3.6", "@smithy/hash-node": "^4.2.5", "@smithy/invalid-dependency": "^4.2.5", "@smithy/middleware-content-length": "^4.2.5", "@smithy/middleware-endpoint": "^4.3.9", "@smithy/middleware-retry": "^4.4.9", "@smithy/middleware-serde": "^4.2.5", "@smithy/middleware-stack": "^4.2.5", "@smithy/node-config-provider": "^4.3.5", "@smithy/node-http-handler": "^4.4.5", "@smithy/protocol-http": "^5.3.5", "@smithy/smithy-client": "^4.9.5", "@smithy/types": "^4.9.0", "@smithy/url-parser": "^4.2.5", "@smithy/util-base64": "^4.3.0", "@smithy/util-body-length-browser": "^4.2.0", "@smithy/util-body-length-node": "^4.2.1", "@smithy/util-defaults-mode-browser": "^4.3.8", "@smithy/util-defaults-mode-node": "^4.2.11", "@smithy/util-endpoints": "^3.2.5", "@smithy/util-middleware": "^4.2.5", "@smithy/util-retry": "^4.2.5", "@smithy/util-utf8": "^4.2.0", "tslib": "^2.6.2" } }, "sha512-9eMUCu1Ay3C9ojo+dJcynSdpbxuwDVtZUt/Xhce+c2+mgDsmvRzjww+wfLpZwRNWxBWmeauQQAZk52tCwQgXsQ=="], - "@aws-sdk/client-cognito-identity": ["@aws-sdk/client-cognito-identity@3.1057.0", "", { "dependencies": { "@aws-crypto/sha256-browser": "5.2.0", "@aws-crypto/sha256-js": "5.2.0", "@aws-sdk/core": "^3.974.15", "@aws-sdk/credential-provider-node": "^3.972.47", "@aws-sdk/types": "^3.973.9", "@smithy/core": "^3.24.5", "@smithy/fetch-http-handler": "^5.4.5", "@smithy/node-http-handler": "^4.7.5", "@smithy/types": "^4.14.2", "tslib": "^2.6.2" } }, "sha512-5MliYkp2u0+2arTp5fZIaxl+xmm90LEKv/VeSxhfNQW4t0fvWJrNO429/jchWQenNoDRrOGE59VfbuZUfwFujg=="], "@aws-sdk/client-firehose": ["@aws-sdk/client-firehose@3.933.0", "", { "dependencies": { "@aws-crypto/sha256-browser": "5.2.0", "@aws-crypto/sha256-js": "5.2.0", "@aws-sdk/core": "3.932.0", "@aws-sdk/credential-provider-node": "3.933.0", "@aws-sdk/middleware-host-header": "3.930.0", "@aws-sdk/middleware-logger": "3.930.0", "@aws-sdk/middleware-recursion-detection": "3.933.0", "@aws-sdk/middleware-user-agent": "3.932.0", "@aws-sdk/region-config-resolver": "3.930.0", "@aws-sdk/types": "3.930.0", "@aws-sdk/util-endpoints": "3.930.0", "@aws-sdk/util-user-agent-browser": "3.930.0", "@aws-sdk/util-user-agent-node": "3.932.0", "@smithy/config-resolver": "^4.4.3", "@smithy/core": "^3.18.2", "@smithy/fetch-http-handler": "^5.3.6", "@smithy/hash-node": "^4.2.5", "@smithy/invalid-dependency": "^4.2.5", "@smithy/middleware-content-length": "^4.2.5", "@smithy/middleware-endpoint": "^4.3.9", "@smithy/middleware-retry": "^4.4.9", "@smithy/middleware-serde": "^4.2.5", "@smithy/middleware-stack": "^4.2.5", "@smithy/node-config-provider": "^4.3.5", "@smithy/node-http-handler": "^4.4.5", "@smithy/protocol-http": "^5.3.5", "@smithy/smithy-client": "^4.9.5", "@smithy/types": "^4.9.0", "@smithy/url-parser": "^4.2.5", "@smithy/util-base64": "^4.3.0", "@smithy/util-body-length-browser": "^4.2.0", "@smithy/util-body-length-node": "^4.2.1", "@smithy/util-defaults-mode-browser": "^4.3.8", "@smithy/util-defaults-mode-node": "^4.2.11", "@smithy/util-endpoints": "^3.2.5", "@smithy/util-middleware": "^4.2.5", "@smithy/util-retry": "^4.2.5", "@smithy/util-utf8": "^4.2.0", "tslib": "^2.6.2" } }, "sha512-tDrtgczN2lQsflLDPYu/wdOoyCZLVYtgzmWnYzSEOBWd/cp2AbuQ7D+FemSwUTzyoMTuhhIevyEJKzqsF+QYxA=="], @@ -5762,8 +5759,6 @@ "@aws-crypto/util/@smithy/util-utf8": ["@smithy/util-utf8@2.3.0", "", { "dependencies": { "@smithy/util-buffer-from": "^2.2.0", "tslib": "^2.6.2" } }, "sha512-R8Rdn8Hy72KKcebgLiv8jQcQkXoLMOGGv5uI1/k0l+snqkOzQ1R0ChUBCxWMlBsFMekWjq0wRudIweFs7sKT5A=="], - "@aws-sdk/client-athena/@smithy/util-utf8": ["@smithy/util-utf8@4.2.2", "", { "dependencies": { "@smithy/util-buffer-from": "^4.2.2", "tslib": "^2.6.2" } }, "sha512-75MeYpjdWRe8M5E3AW0O4Cx3UadweS+cwdXjwYGBW5h/gxxnbeZ877sLPX/ZJA9GVTlL/qG0dXP29JWFCD1Ayw=="], - "@aws-sdk/client-cognito-identity/@aws-sdk/core": ["@aws-sdk/core@3.974.15", "", { "dependencies": { "@aws-sdk/types": "^3.973.9", "@aws-sdk/xml-builder": "^3.972.26", "@aws/lambda-invoke-store": "^0.2.2", "@smithy/core": "^3.24.5", "@smithy/signature-v4": "^5.4.5", "@smithy/types": "^4.14.2", "bowser": "^2.11.0", "tslib": "^2.6.2" } }, "sha512-UpA0rTGW/tHGITcCqHisbuuEPraYg9GG+mWmXjY5+RxZBMLGe6aL9oe0ix50LztwAcPIkGZLH0yWdMIkCM10hw=="], "@aws-sdk/client-cognito-identity/@aws-sdk/credential-provider-node": ["@aws-sdk/credential-provider-node@3.972.47", "", { "dependencies": { "@aws-sdk/credential-provider-env": "^3.972.41", "@aws-sdk/credential-provider-http": "^3.972.43", "@aws-sdk/credential-provider-ini": "^3.972.46", "@aws-sdk/credential-provider-process": "^3.972.41", "@aws-sdk/credential-provider-sso": "^3.972.45", "@aws-sdk/credential-provider-web-identity": "^3.972.45", "@aws-sdk/types": "^3.973.9", "@smithy/core": "^3.24.5", "@smithy/credential-provider-imds": "^4.3.6", "@smithy/types": "^4.14.2", "tslib": "^2.6.2" } }, "sha512-HrId+C0DWA5qDIyLG64/kjUB2RNtPypxmABnIctK+TA1P1kHlOYoE/Wf5T5tKOMKgb08P7k/zNyhvfJ3lh5Oag=="], diff --git a/infra/console.ts b/infra/console.ts index 630bb56fec7d..77d6fd0fc114 100644 --- a/infra/console.ts +++ b/infra/console.ts @@ -1,9 +1,7 @@ -import { deployAws, domain } from "./stage" +import { domain } from "./stage" import { EMAILOCTOPUS_API_KEY } from "./app" import { SECRET } from "./secret" -const lake = deployAws ? await import("./lake") : undefined - //////////////// // DATABASE //////////////// @@ -253,7 +251,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, ...(lake?.lakeIngest ? [lake.lakeIngest] : [])], + link: [SECRET.HoneycombApiKey], }) new sst.cloudflare.x.SolidStart("Console", { diff --git a/infra/lake.ts b/infra/lake.ts index d4e46168e366..a6125a1ee1c7 100644 --- a/infra/lake.ts +++ b/infra/lake.ts @@ -1,5 +1,9 @@ import { domain } from "./stage" +// Retired in favor of the R2 lake and pending removal. Pulumi reads forceDestroy +// and retainOnDelete from state at delete time, so these settings must deploy to +// every stage before the resources are removed from code. + const current = aws.getCallerIdentityOutput({}) const partition = aws.getPartitionOutput({}) const region = aws.getRegionOutput({}) @@ -15,7 +19,7 @@ const s3TablesBucketWildcardArn = $interpolate`arn:${partition.partition}:s3tabl export const tableBucket = new aws.s3tables.TableBucket("LakeTableBucket", { name: tableBucketName, - forceDestroy: $app.stage !== "production", + forceDestroy: true, }) const s3TablesCatalog = new aws.cloudcontrol.Resource( @@ -51,19 +55,27 @@ const s3TablesCatalog = new aws.cloudcontrol.Resource( { dependsOn: [tableBucket] }, ) -const athenaResultsBucket = new aws.s3.Bucket("LakeAthenaResults", { - bucket: `opencode-${$app.stage}-lake-athena-results`, - forceDestroy: $app.stage !== "production", -}) +const athenaResultsBucket = new aws.s3.Bucket( + "LakeAthenaResults", + { + bucket: `opencode-${$app.stage}-lake-athena-results`, + forceDestroy: true, + }, + { retainOnDelete: false }, +) -const firehoseErrorBucket = new aws.s3.Bucket("LakeFirehoseErrors", { - bucket: `opencode-${$app.stage}-lake-firehose-errors`, - forceDestroy: $app.stage !== "production", -}) +const firehoseErrorBucket = new aws.s3.Bucket( + "LakeFirehoseErrors", + { + bucket: `opencode-${$app.stage}-lake-firehose-errors`, + forceDestroy: true, + }, + { retainOnDelete: false }, +) -const athenaWorkgroup = new aws.athena.Workgroup("LakeAthenaWorkgroup", { +new aws.athena.Workgroup("LakeAthenaWorkgroup", { name: `opencode-${$app.stage}-lake-workgroup`, - forceDestroy: $app.stage !== "production", + forceDestroy: true, configuration: { enforceWorkgroupConfiguration: true, publishCloudwatchMetricsEnabled: true, @@ -197,9 +209,6 @@ const firehose = new aws.kinesis.FirehoseDeliveryStream( export const lakeVpc = new sst.aws.Vpc("LakeVpc") export const lakeCluster = new sst.aws.Cluster("LakeCluster", { vpc: lakeVpc }) -export const lakeRegion = region.region -export const lakeCatalog = $interpolate`${glueCatalogName}/${tableBucket.name}` -export const lakeAthenaWorkgroup = athenaWorkgroup const ingestSecret = new random.RandomPassword("LakeIngestSecret", { length: 32 }) export const ingestSecretSsm = new aws.ssm.Parameter("LakeIngestSecretSsm", { @@ -277,55 +286,3 @@ export const lakeIngest = new sst.Linkable("LakeIngest", { secret: ingestSecret.result, }, }) - -export const lakeQueryPermissions = [ - { - actions: ["athena:StartQueryExecution", "athena:GetQueryExecution", "athena:GetQueryResults"], - resources: [athenaWorkgroup.arn], - }, - { - actions: [ - "glue:GetCatalog", - "glue:GetCatalogs", - "glue:GetDatabase", - "glue:GetDatabases", - "glue:GetTable", - "glue:GetTables", - "glue:GetPartitions", - ], - resources: [ - glueCatalogArn, - glueS3TablesCatalogArn, - $interpolate`${glueS3TablesCatalogArn}/*`, - glueS3TablesDatabaseWildcardArn, - glueS3TablesTableWildcardArn, - $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/*`, - $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/*/*`, - $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/*`, - ], - }, - { - actions: ["s3:GetBucketLocation", "s3:ListBucket"], - resources: [athenaResultsBucket.arn], - }, - { - actions: ["s3:GetObject", "s3:PutObject", "s3:AbortMultipartUpload", "s3:ListBucketMultipartUploads"], - resources: [$interpolate`${athenaResultsBucket.arn}/*`], - }, - { - actions: [ - "s3tables:GetTableBucket", - "s3tables:GetNamespace", - "s3tables:GetTable", - "s3tables:GetTableData", - "s3tables:GetTableMetadataLocation", - "s3tables:ListNamespaces", - "s3tables:ListTables", - ], - resources: ["*"], - }, - { - actions: ["lakeformation:GetDataAccess"], - resources: ["*"], - }, -] diff --git a/infra/stats.ts b/infra/stats.ts index 29c9537daf88..a867aa695a16 100644 --- a/infra/stats.ts +++ b/infra/stats.ts @@ -1,4 +1,4 @@ -import { lakeAthenaWorkgroup, lakeCatalog, lakeCluster, lakeQueryPermissions, lakeRegion, tableBucket } from "./lake" +import { lakeCluster, tableBucket } from "./lake" import { EMAILOCTOPUS_API_KEY } from "./app" import { domain } from "./stage" @@ -11,7 +11,7 @@ const inferenceNamespace = new aws.s3tables.Namespace("LakeInferenceNamespace", tableBucketArn: tableBucket.arn, }) -const inferenceEventTable = new aws.s3tables.Table( +new aws.s3tables.Table( "LakeInferenceEventTable", { name: "event", @@ -89,17 +89,6 @@ const inferenceEventTable = new aws.s3tables.Table( { deleteBeforeReplace: $app.stage !== "production", ignoreChanges: ["metadata"] }, ) -export const inferenceEvent = new sst.Linkable("InferenceEvent", { - properties: { - region: lakeRegion, - catalog: lakeCatalog, - database: inferenceNamespace.namespace, - table: inferenceEventTable.name, - tableBucket: tableBucket.name, - workgroup: lakeAthenaWorkgroup.name, - }, -}) - //////////////// // DATABASE //////////////// @@ -203,10 +192,7 @@ export const statSync = new sst.aws.Service("StatsSyncService", { dockerfile: "packages/stats/server/Dockerfile", }, command: ["bun", "src/stat-sync.ts"], - // Keep the legacy Athena link and IAM permissions during the first R2-backed - // release so reverting the application code remains a one-deploy rollback. - link: [database, inferenceEvent, r2Sql, r2SqlAuthToken, statsSyncConfig], - permissions: lakeQueryPermissions, + link: [database, r2Sql, r2SqlAuthToken, statsSyncConfig], scaling: { min: 1, max: 1, diff --git a/packages/console/function/src/log-processor.ts b/packages/console/function/src/log-processor.ts index 75daba2b5046..71f57c37b389 100644 --- a/packages/console/function/src/log-processor.ts +++ b/packages/console/function/src/log-processor.ts @@ -37,6 +37,7 @@ export default { "ip.prefix": ipPrefix(ip), } const time = new Date(event.eventTimestamp ?? Date.now()).toISOString() + // This also merges each _metric log into data, so keep it while the Honeycomb export below is disabled. const events = [ ...event.logs.flatMap((log) => log.message.flatMap((message: string) => { @@ -53,107 +54,20 @@ export default { ] console.log(JSON.stringify(data, null, 2)) - const lakeIngest = getLakeIngest() - const [lake] = await Promise.all([ - // fetch("https://api.honeycomb.io/1/batch/zen", { - // method: "POST", - // headers: { - // "Content-Type": "application/json", - // "X-Honeycomb-Team": Resource.HONEYCOMB_API_KEY.value, - // }, - // body: JSON.stringify(events), - // }), - ...(lakeIngest - ? [ - fetch(lakeIngest.url, { - method: "POST", - headers: { - "Content-Type": "application/json", - Authorization: `Bearer ${lakeIngest.secret}`, - }, - body: JSON.stringify({ events: events.map((event) => toLakeEvent(event.time, event.data)) }), - }), - ] - : []), - ]) + // const honeycomb = await fetch("https://api.honeycomb.io/1/batch/zen", { + // method: "POST", + // headers: { + // "Content-Type": "application/json", + // "X-Honeycomb-Team": Resource.HONEYCOMB_API_KEY.value, + // }, + // body: JSON.stringify(events), + // }) // console.log(honeycomb.status) // console.log(await honeycomb.text()) - if (lake) { - console.log(lake.status) - console.log(await lake.text()) - } } }, } -function getLakeIngest(): { url: string; secret: string } | undefined { - try { - return Resource.LakeIngest - } catch { - return undefined - } -} - -function toLakeEvent(time: string, data: Record) { - return { - _datalake_key: "inference.event", - event_timestamp: time, - event_date: time.slice(0, 10), - event_type: string(data, "event_type"), - dataset: "zen", - cf_continent: string(data, "cf.continent"), - cf_country: string(data, "cf.country"), - cf_city: string(data, "cf.city"), - cf_region: string(data, "cf.region"), - cf_latitude: number(data, "cf.latitude"), - cf_longitude: number(data, "cf.longitude"), - cf_timezone: string(data, "cf.timezone"), - duration: number(data, "duration"), - request_length: integer(data, "request_length"), - status: integer(data, "status"), - ip: string(data, "ip"), - ip_prefix: string(data, "ip.prefix"), - is_stream: boolean(data, "is_stream"), - session: string(data, "session"), - request: string(data, "request"), - client: string(data, "client"), - user_agent: string(data, "user_agent"), - model: string(data, "model"), - model_tier: string(data, "model.tier"), - model_variant: string(data, "model.variant"), - source: string(data, "source"), - provider: string(data, "provider"), - provider_model: string(data, "provider.model"), - llm_error_code: integer(data, "llm.error.code"), - llm_error_message: string(data, "llm.error.message"), - error_response: string(data, "error.response"), - error_type: string(data, "error.type"), - error_message: string(data, "error.message"), - error_cause: string(data, "error.cause"), - error_cause2: string(data, "error.cause2"), - api_key: string(data, "api_key"), - workspace: string(data, "workspace"), - user_id: string(data, "user_id"), - is_subscription: boolean(data, "isSubscription"), // removed - subscription: string(data, "subscription"), - response_length: integer(data, "response_length"), - time_to_first_byte: integer(data, "time_to_first_byte"), - timestamp_first_byte: integer(data, "timestamp.first_byte"), - timestamp_last_byte: integer(data, "timestamp.last_byte"), - tokens_input: integer(data, "tokens.input"), - tokens_output: integer(data, "tokens.output"), - tokens_reasoning: integer(data, "tokens.reasoning"), - tokens_cache_read: integer(data, "tokens.cache_read"), - tokens_cache_write_5m: integer(data, "tokens.cache_write_5m"), - tokens_cache_write_1h: integer(data, "tokens.cache_write_1h"), - cost_input_microcents: integer(data, "cost.input.microcents"), - cost_output_microcents: integer(data, "cost.output.microcents"), - cost_cache_read_microcents: integer(data, "cost.cache_read.microcents"), - cost_cache_write_microcents: integer(data, "cost.cache_write.microcents"), - cost_total_microcents: integer(data, "cost.total.microcents"), - } -} - // Returns a stable lookup key for an IP address. // IPv4: full address as /32 (e.g. "203.0.113.45/32"). // IPv6: the /64 network prefix (e.g. "2001:db8:abcd:1234::/64"). ISPs commonly @@ -179,33 +93,3 @@ function ipPrefix(ip: string | undefined) { .join(":") return `${prefix}::/64` } - -function string(data: Record, 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: Record, 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: Record, key: string) { - const value = number(data, key) - if (value === undefined) return undefined - return Math.round(value) -} - -function number(data: Record, key: string) { - const value = data[key] - if (typeof value === "number") return Number.isFinite(value) ? value : undefined - if (typeof value === "string") { - const parsed = Number(value) - return Number.isFinite(parsed) ? parsed : undefined - } - return undefined -} diff --git a/packages/stats/core/package.json b/packages/stats/core/package.json index d938182cb4d0..aefcf4d8cfcb 100644 --- a/packages/stats/core/package.json +++ b/packages/stats/core/package.json @@ -7,7 +7,6 @@ "license": "MIT", "exports": { ".": "./src/index.ts", - "./athena": "./src/athena.ts", "./config": "./src/config.ts", "./database": "./src/database.ts", "./database/*": "./src/database/*.ts", @@ -27,7 +26,6 @@ "typecheck": "tsgo --noEmit" }, "dependencies": { - "@aws-sdk/client-athena": "3.933.0", "@planetscale/database": "1.19.0", "drizzle-orm": "catalog:", "effect": "catalog:", diff --git a/packages/stats/core/src/athena.ts b/packages/stats/core/src/athena.ts deleted file mode 100644 index 23aba70b3ce6..000000000000 --- a/packages/stats/core/src/athena.ts +++ /dev/null @@ -1,151 +0,0 @@ -import { - AthenaClient as AwsAthenaClient, - GetQueryExecutionCommand, - GetQueryResultsCommand, - StartQueryExecutionCommand, - type Row, -} from "@aws-sdk/client-athena" -import { Effect, Layer } from "effect" -import * as Context from "effect/Context" -import { Resource } from "sst/resource" - -const ATHENA_MAX_POLL_ATTEMPTS = 900 -const ATHENA_PAGE_SIZE = 1000 - -export type AthenaData = Record - -export class AthenaQueryError extends Error { - readonly _tag = "AthenaQueryError" - readonly queryExecutionId?: string - - constructor(input: { message: string; queryExecutionId?: string; cause?: unknown }) { - super(input.message, { cause: input.cause }) - this.name = "AthenaQueryError" - this.queryExecutionId = input.queryExecutionId - } -} - -export class AthenaQueryTimeoutError extends Error { - readonly _tag = "AthenaQueryTimeoutError" - readonly queryExecutionId: string - - constructor(input: { message: string; queryExecutionId: string }) { - super(input.message) - this.name = "AthenaQueryTimeoutError" - this.queryExecutionId = input.queryExecutionId - } -} - -export declare namespace Athena { - export interface Service { - readonly query: (query: string) => Effect.Effect - } -} - -export class Athena extends Context.Service()("@opencode/stats/Athena") { - static readonly layer: Layer.Layer = Layer.effect( - Athena, - Effect.sync(() => { - const client = new AwsAthenaClient({ region: Resource.InferenceEvent.region }) - - const query = Effect.fn("Athena.query")(function* (query: string) { - const started = yield* Effect.tryPromise({ - try: () => - client.send( - new StartQueryExecutionCommand({ - QueryString: query, - WorkGroup: Resource.InferenceEvent.workgroup, - QueryExecutionContext: { - Catalog: Resource.InferenceEvent.catalog, - Database: Resource.InferenceEvent.database, - }, - }), - ), - catch: (cause) => new AthenaQueryError({ message: "Failed to start Athena stats query", cause }), - }) - const queryExecutionId = started.QueryExecutionId - if (!queryExecutionId) - return yield* Effect.fail(new AthenaQueryError({ message: "Athena did not return a query execution id" })) - - yield* poll(client, queryExecutionId) - return yield* results(client, queryExecutionId) - }) - - return Athena.of({ query }) - }), - ) -} - -const poll: ( - client: AwsAthenaClient, - queryExecutionId: string, - attempt?: number, -) => Effect.Effect = Effect.fn("Athena.poll")(function* ( - client: AwsAthenaClient, - queryExecutionId: string, - attempt = 0, -) { - if (attempt > 0) yield* Effect.sleep("2 seconds") - - const result = yield* Effect.tryPromise({ - try: () => client.send(new GetQueryExecutionCommand({ QueryExecutionId: queryExecutionId })), - catch: (cause) => new AthenaQueryError({ message: "Failed to poll Athena stats query", queryExecutionId, cause }), - }) - const status = result.QueryExecution?.Status - - if (status?.State === "SUCCEEDED") return - if (status?.State === "FAILED" || status?.State === "CANCELLED") - return yield* Effect.fail( - new AthenaQueryError({ - message: `Athena stats query ${status.State.toLowerCase()}: ${status.StateChangeReason ?? "unknown reason"}`, - queryExecutionId, - }), - ) - - if (attempt >= ATHENA_MAX_POLL_ATTEMPTS - 1) - return yield* Effect.fail( - new AthenaQueryTimeoutError({ - message: `Athena stats query ${queryExecutionId} did not complete`, - queryExecutionId, - }), - ) - - return yield* poll(client, queryExecutionId, attempt + 1) -}) - -const results: (client: AwsAthenaClient, queryExecutionId: string) => Effect.Effect = - Effect.fn("Athena.results")(function* (client: AwsAthenaClient, queryExecutionId: string) { - // Accumulate pages iteratively; recursive spreads copied every previously - // fetched row per page and blew up memory on large result sets. - const rows: AthenaData[] = [] - let nextToken: string | undefined - while (true) { - const result = yield* Effect.tryPromise({ - try: () => - client.send( - new GetQueryResultsCommand({ - QueryExecutionId: queryExecutionId, - NextToken: nextToken, - MaxResults: ATHENA_PAGE_SIZE, - }), - ), - catch: (cause) => - new AthenaQueryError({ message: "Failed to read Athena stats results", queryExecutionId, cause }), - }) - const columns = result.ResultSet?.ResultSetMetadata?.ColumnInfo?.map((item) => item.Name ?? "") ?? [] - // The first page starts with the header row. - for (const row of (result.ResultSet?.Rows ?? []).slice(nextToken ? 0 : 1)) rows.push(rowData(columns, row)) - if (!result.NextToken) return rows - nextToken = result.NextToken - } - }) - -function rowData(columns: string[], row: Row): AthenaData { - return Object.fromEntries( - columns.flatMap((column, index) => { - const value = row.Data?.[index]?.VarCharValue - if (!column || value === undefined) return [] - return [[column, value]] - }), - ) -} diff --git a/packages/stats/core/src/index.ts b/packages/stats/core/src/index.ts index 834625f60165..9b7fcd0f6981 100644 --- a/packages/stats/core/src/index.ts +++ b/packages/stats/core/src/index.ts @@ -1,4 +1,3 @@ -export * as Athena from "./athena" export * as AppConfig from "./config" export * as Database from "./database" export * as GeoStat from "./domain/geo" diff --git a/packages/stats/core/src/resource.d.ts b/packages/stats/core/src/resource.d.ts index b8017777971e..db2167b03354 100644 --- a/packages/stats/core/src/resource.d.ts +++ b/packages/stats/core/src/resource.d.ts @@ -2,15 +2,6 @@ import "sst/resource" declare module "sst/resource" { export interface Resource { - InferenceEvent: { - catalog: string - database: string - region: string - table: string - tableBucket: string - type: "sst.sst.Linkable" - workgroup: string - } R2Sql: { accountId: string bucket: string