From dc31828006702f1a88c0113b69afccad389832e9 Mon Sep 17 00:00:00 2001 From: Adam <2363879+adamdotdevin@users.noreply.github.com> Date: Thu, 1 Oct 2026 15:58:14 -0500 Subject: [PATCH] chore(stats): remove s3 lake ingestion (#52572) --- bun.lock | 5 - infra/lake.ts | 215 +----------------------- packages/stats/server/Dockerfile | 4 +- packages/stats/server/package.json | 7 +- packages/stats/server/src/ingest.ts | 166 ------------------ packages/stats/server/src/resource.d.ts | 11 -- packages/stats/server/src/router.ts | 73 -------- packages/stats/server/src/server.ts | 28 --- packages/stats/server/src/shutdown.ts | 17 -- sst.config.ts | 7 - 10 files changed, 12 insertions(+), 521 deletions(-) delete mode 100644 packages/stats/server/src/ingest.ts delete mode 100644 packages/stats/server/src/resource.d.ts delete mode 100644 packages/stats/server/src/router.ts delete mode 100644 packages/stats/server/src/server.ts delete mode 100644 packages/stats/server/src/shutdown.ts diff --git a/bun.lock b/bun.lock index 89315fb65a..5ca5f3ac46 100644 --- a/bun.lock +++ b/bun.lock @@ -898,7 +898,6 @@ "name": "@opencode-ai/stats-server", "version": "1.18.34", "dependencies": { - "@aws-sdk/client-firehose": "3.933.0", "@effect/platform-node": "catalog:", "@opencode-ai/stats-core": "workspace:*", "effect": "catalog:", @@ -1269,8 +1268,6 @@ "@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=="], - "@aws-sdk/client-lambda": ["@aws-sdk/client-lambda@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-JoaE3QNvPqOSHJtcAFfza7BRoGUNerIKlSVhsKCBsa1i54Vto1YWUS7itkUELK2rpvS8kNZMPliPxDdi/oV5dw=="], "@aws-sdk/client-s3": ["@aws-sdk/client-s3@3.933.0", "", { "dependencies": { "@aws-crypto/sha1-browser": "5.2.0", "@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-bucket-endpoint": "3.930.0", "@aws-sdk/middleware-expect-continue": "3.930.0", "@aws-sdk/middleware-flexible-checksums": "3.932.0", "@aws-sdk/middleware-host-header": "3.930.0", "@aws-sdk/middleware-location-constraint": "3.930.0", "@aws-sdk/middleware-logger": "3.930.0", "@aws-sdk/middleware-recursion-detection": "3.933.0", "@aws-sdk/middleware-sdk-s3": "3.932.0", "@aws-sdk/middleware-ssec": "3.930.0", "@aws-sdk/middleware-user-agent": "3.932.0", "@aws-sdk/region-config-resolver": "3.930.0", "@aws-sdk/signature-v4-multi-region": "3.932.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/eventstream-serde-browser": "^4.2.5", "@smithy/eventstream-serde-config-resolver": "^4.3.5", "@smithy/eventstream-serde-node": "^4.2.5", "@smithy/fetch-http-handler": "^5.3.6", "@smithy/hash-blob-browser": "^4.2.6", "@smithy/hash-node": "^4.2.5", "@smithy/hash-stream-node": "^4.2.5", "@smithy/invalid-dependency": "^4.2.5", "@smithy/md5-js": "^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-stream": "^4.5.6", "@smithy/util-utf8": "^4.2.0", "@smithy/util-waiter": "^4.2.5", "tslib": "^2.6.2" } }, "sha512-KxwZvdxdCeWK6o8mpnb+kk7Kgb8V+8AjTwSXUWH1UAD85B0tjdo1cSfE5zoR5fWGol4Ml5RLez12a6LPhsoTqA=="], @@ -5765,8 +5762,6 @@ "@aws-sdk/client-cognito-identity/@aws-sdk/types": ["@aws-sdk/types@3.973.9", "", { "dependencies": { "@smithy/types": "^4.14.2", "tslib": "^2.6.2" } }, "sha512-kuBfgQVdcz5Bmapc4A13YbpVw/pXkesfhetcFYwbntqas8sF41OHyd4o28+/TG2ZQdHBsv90Lsu5y6oitvYCdg=="], - "@aws-sdk/client-firehose/@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-lambda/@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-lambda/@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/lake.ts b/infra/lake.ts index a6125a1ee1..ccad966006 100644 --- a/infra/lake.ts +++ b/infra/lake.ts @@ -1,8 +1,8 @@ -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. +// Retired in favor of the R2 lake. Nothing writes to it anymore, but the table, +// catalog and Athena workgroup stay while the anomalyco/anomaly inference and +// analytics dashboards still query its history. Pulumi reads forceDestroy and +// retainOnDelete from state at delete time, so keep those settings until the +// lake is removed. const current = aws.getCallerIdentityOutput({}) const partition = aws.getPartitionOutput({}) @@ -10,11 +10,6 @@ const region = aws.getRegionOutput({}) const tableBucketName = `opencode-${$app.stage}-lake` const glueCatalogName = "s3tablescatalog" -const glueCatalogArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:catalog` -const glueS3TablesCatalogArn = $interpolate`${glueCatalogArn}/${glueCatalogName}` -const glueS3TablesChildCatalogArn = $interpolate`${glueS3TablesCatalogArn}/${tableBucketName}` -const glueS3TablesDatabaseWildcardArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/${glueCatalogName}/${tableBucketName}/*` -const glueS3TablesTableWildcardArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/${tableBucketName}/*/*` const s3TablesBucketWildcardArn = $interpolate`arn:${partition.partition}:s3tables:${region.region}:${current.accountId}:bucket/*` export const tableBucket = new aws.s3tables.TableBucket("LakeTableBucket", { @@ -22,7 +17,7 @@ export const tableBucket = new aws.s3tables.TableBucket("LakeTableBucket", { forceDestroy: true, }) -const s3TablesCatalog = new aws.cloudcontrol.Resource( +new aws.cloudcontrol.Resource( "LakeS3TablesCatalog", { typeName: "AWS::Glue::Catalog", @@ -64,7 +59,8 @@ const athenaResultsBucket = new aws.s3.Bucket( { retainOnDelete: false }, ) -const firehoseErrorBucket = new aws.s3.Bucket( +// Keep the archived delivery failures until their records have been reconciled. +new aws.s3.Bucket( "LakeFirehoseErrors", { bucket: `opencode-${$app.stage}-lake-firehose-errors`, @@ -89,200 +85,5 @@ new aws.athena.Workgroup("LakeAthenaWorkgroup", { }, }) -const firehoseRole = new aws.iam.Role("LakeFirehoseRole", { - assumeRolePolicy: aws.iam.getPolicyDocumentOutput({ - statements: [ - { - effect: "Allow", - actions: ["sts:AssumeRole"], - principals: [ - { - type: "Service", - identifiers: ["firehose.amazonaws.com"], - }, - ], - }, - ], - }).json, -}) - -const firehosePolicy = new aws.iam.RolePolicy("LakeFirehosePolicy", { - role: firehoseRole.id, - policy: aws.iam.getPolicyDocumentOutput({ - statements: [ - { - effect: "Allow", - actions: [ - "s3tables:ListTableBuckets", - "s3tables:GetTableBucket", - "s3tables:GetNamespace", - "s3tables:GetTable", - "s3tables:GetTableData", - "s3tables:GetTableMetadataLocation", - "s3tables:ListNamespaces", - "s3tables:ListTables", - "s3tables:PutTableData", - "s3tables:UpdateTableMetadataLocation", - ], - resources: ["*"], - }, - { - effect: "Allow", - actions: [ - "glue:GetCatalog", - "glue:GetCatalogs", - "glue:GetDatabase", - "glue:GetDatabases", - "glue:GetTable", - "glue:GetTables", - "glue:UpdateTable", - ], - 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}/*`, - ], - }, - { - effect: "Allow", - actions: [ - "s3:AbortMultipartUpload", - "s3:GetBucketLocation", - "s3:GetObject", - "s3:ListBucket", - "s3:ListBucketMultipartUploads", - "s3:PutObject", - ], - resources: [firehoseErrorBucket.arn, $interpolate`${firehoseErrorBucket.arn}/*`], - }, - { - effect: "Allow", - actions: ["lakeformation:GetDataAccess"], - resources: ["*"], - }, - ], - }).json, -}) - -const firehose = new aws.kinesis.FirehoseDeliveryStream( - "LakeFirehose", - { - name: `opencode-${$app.stage}-lake-ingest`, - destination: "iceberg", - icebergConfiguration: { - appendOnly: true, - bufferingInterval: 60, - bufferingSize: 1, - catalogArn: glueS3TablesChildCatalogArn, - processingConfiguration: { - enabled: true, - processors: [ - { - type: "MetadataExtraction", - parameters: [ - { parameterName: "JsonParsingEngine", parameterValue: "JQ-1.6" }, - { - parameterName: "MetadataExtractionQuery", - parameterValue: - '{destinationDatabaseName:._lake_database,destinationTableName:._lake_table,operation:(._lake_operation // "insert")}', - }, - ], - }, - ], - }, - roleArn: firehoseRole.arn, - s3BackupMode: "FailedDataOnly", - s3Configuration: { - roleArn: firehoseRole.arn, - bucketArn: firehoseErrorBucket.arn, - errorOutputPrefix: "errors/!{firehose:error-output-type}/", - }, - }, - }, - { dependsOn: [s3TablesCatalog, firehosePolicy] }, -) - export const lakeVpc = new sst.aws.Vpc("LakeVpc") export const lakeCluster = new sst.aws.Cluster("LakeCluster", { vpc: lakeVpc }) - -const ingestSecret = new random.RandomPassword("LakeIngestSecret", { length: 32 }) -export const ingestSecretSsm = new aws.ssm.Parameter("LakeIngestSecretSsm", { - name: $interpolate`/${$app.name}/${$app.stage}/lake/ingest/secret`, - type: "SecureString", - value: ingestSecret.result, -}) - -const ingestConfig = new sst.Linkable("LakeIngestConfig", { - properties: { - streamName: firehose.name, - secret: ingestSecret.result, - }, -}) - -const ingestService = new sst.aws.Service("LakeIngestService", { - cluster: lakeCluster, - architecture: "arm64", - cpu: "1 vCPU", - memory: "4 GB", - image: { - context: ".", - dockerfile: "packages/stats/server/Dockerfile", - }, - link: [ingestConfig], - permissions: [ - { - actions: ["firehose:PutRecord", "firehose:PutRecordBatch"], - resources: [firehose.arn], - }, - ], - scaling: { - min: $app.stage === "production" ? 2 : 1, - max: $app.stage === "production" ? 32 : 4, - cpuUtilization: 60, - memoryUtilization: 70, - }, - loadBalancer: { - domain: { - name: `lake.${domain}`, - dns: sst.cloudflare.dns(), - }, - rules: [ - { listen: "80/http", redirect: "443/https" }, - { listen: "443/https", forward: "3000/http" }, - ], - health: { - "3000/http": { - path: "/ready", - successCodes: "200-299", - }, - }, - }, - health: { - command: [ - "CMD-SHELL", - "bun --eval \"fetch('http://localhost:3000/health').then((r) => process.exit(r.ok ? 0 : 1)).catch(() => process.exit(1))\"", - ], - interval: "30 seconds", - retries: 3, - startPeriod: "30 seconds", - timeout: "5 seconds", - }, - dev: { - command: "bun run start", - directory: "packages/stats/server", - url: "http://localhost:3000", - }, - wait: $app.stage === "production", -}) - -export const lakeIngest = new sst.Linkable("LakeIngest", { - properties: { - url: ingestService.url, - secret: ingestSecret.result, - }, -}) diff --git a/packages/stats/server/Dockerfile b/packages/stats/server/Dockerfile index 01a9004b7b..c42476e8d5 100644 --- a/packages/stats/server/Dockerfile +++ b/packages/stats/server/Dockerfile @@ -29,6 +29,4 @@ COPY --from=pruner /app/out/full/ ./ WORKDIR /app/packages/stats/server -EXPOSE 3000 - -CMD ["bun", "src/server.ts"] +CMD ["bun", "src/stat-sync.ts"] diff --git a/packages/stats/server/package.json b/packages/stats/server/package.json index 9bae345f81..5c30056b26 100644 --- a/packages/stats/server/package.json +++ b/packages/stats/server/package.json @@ -5,16 +5,15 @@ "private": true, "type": "module", "license": "MIT", - "main": "./src/server.ts", + "main": "./src/stat-sync.ts", "exports": { - ".": "./src/server.ts" + ".": "./src/stat-sync.ts" }, "scripts": { - "start": "bun src/server.ts", + "start": "bun src/stat-sync.ts", "typecheck": "tsgo --noEmit" }, "dependencies": { - "@aws-sdk/client-firehose": "3.933.0", "@effect/platform-node": "catalog:", "@opencode-ai/stats-core": "workspace:*", "effect": "catalog:", diff --git a/packages/stats/server/src/ingest.ts b/packages/stats/server/src/ingest.ts deleted file mode 100644 index eda662d989..0000000000 --- a/packages/stats/server/src/ingest.ts +++ /dev/null @@ -1,166 +0,0 @@ -import { Buffer } from "node:buffer" -import { FirehoseClient, PutRecordBatchCommand } from "@aws-sdk/client-firehose" -import { Effect, Layer } from "effect" -import * as Context from "effect/Context" -import { Resource } from "sst/resource" - -const MAX_FIREHOSE_BATCH_SIZE = 500 -const MAX_FIREHOSE_ATTEMPTS = 3 -const LAKE_TYPE = /^([A-Za-z0-9_]+)\.([A-Za-z0-9_]+)$/ - -type IngestEvent = Record -type LakeRoute = { database: string; table: string } -type FirehoseRecord = { Data: Uint8Array } - -export class IngestError extends Error { - readonly _tag = "IngestError" - readonly failed: number - - constructor(input: { message: string; failed: number; cause?: unknown }) { - super(input.message, { cause: input.cause }) - this.name = "IngestError" - this.failed = input.failed - } -} - -export declare namespace Ingest { - export interface Service { - readonly write: (events: unknown[]) => Effect.Effect<{ records: number }, IngestError> - } -} - -export class Ingest extends Context.Service()("@opencode/stats/Ingest") { - static readonly layer: Layer.Layer = Layer.effect( - Ingest, - Effect.sync(() => { - const client = new FirehoseClient({}) - - const write = Effect.fn("Ingest.write")(function* (events: unknown[]) { - if (events.length === 0) return { records: 0 } - const counts = countRoutedEvents(events) - if (counts.unsupported > 0) { - yield* Effect.logWarning( - `lake ingest rejected ${JSON.stringify({ records: counts.records, unsupported: counts.unsupported })}`, - ) - return yield* Effect.fail( - new IngestError({ - message: "Unsupported lake event type", - failed: counts.unsupported, - }), - ) - } - if (counts.records === 0) return { records: 0 } - - let batch: FirehoseRecord[] = [] - let batches = 0 - let failed = 0 - - for (const event of events) { - if (!isRecord(event)) continue - const route = routeEvent(event) - if (!route) continue - batch.push(toFirehoseRecord(event, route)) - if (batch.length < MAX_FIREHOSE_BATCH_SIZE) continue - failed += yield* putRecords(client, Resource.LakeIngestConfig.streamName, batch) - batches++ - batch = [] - } - - if (batch.length > 0) { - failed += yield* putRecords(client, Resource.LakeIngestConfig.streamName, batch) - batches++ - } - - if (failed > 0) { - yield* Effect.logWarning(`lake ingest incomplete ${JSON.stringify({ records: counts.records, failed })}`) - return yield* Effect.fail(new IngestError({ message: "Failed to ingest all lake records", failed })) - } - - yield* Effect.logInfo(`lake ingest complete ${JSON.stringify({ records: counts.records, batches })}`) - return { records: counts.records } - }) - - return Ingest.of({ write }) - }), - ) -} - -const putRecords: ( - client: FirehoseClient, - streamName: string, - records: FirehoseRecord[], - attempt?: number, -) => Effect.Effect = Effect.fn("Ingest.putRecords")(function* ( - client, - streamName, - records, - attempt = 1, -) { - const result = yield* Effect.tryPromise({ - try: () => client.send(new PutRecordBatchCommand({ DeliveryStreamName: streamName, Records: records })), - catch: (cause) => - new IngestError({ message: "Failed to write lake records to Firehose", failed: records.length, cause }), - }).pipe( - Effect.tapError(() => - Effect.logWarning(`firehose batch write failed ${JSON.stringify({ records: records.length, attempt })}`), - ), - ) - const failed = - result.RequestResponses?.flatMap((item, index) => { - const record = records[index] - if (!item.ErrorCode || !record) return [] - return [record] - }) ?? [] - - if (failed.length === 0) return 0 - if (attempt >= MAX_FIREHOSE_ATTEMPTS) { - yield* Effect.logWarning( - `firehose batch failed ${JSON.stringify({ records: failed.length, attempts: MAX_FIREHOSE_ATTEMPTS })}`, - ) - return failed.length - } - - yield* Effect.logWarning( - `firehose batch retrying ${JSON.stringify({ records: failed.length, attempt: attempt + 1 })}`, - ) - yield* Effect.sleep(`${250 * 2 ** (attempt - 1)} millis`) - return yield* putRecords(client, streamName, failed, attempt + 1) -}) - -function countRoutedEvents(events: unknown[]) { - let records = 0 - let unsupported = 0 - for (const event of events) { - if (!isRecord(event)) continue - if (routeEvent(event)) records++ - else unsupported++ - } - return { records, unsupported } -} - -function isRecord(item: unknown): item is IngestEvent { - return Boolean(item) && typeof item === "object" && !Array.isArray(item) -} - -function routeEvent(event: IngestEvent): LakeRoute | undefined { - if (typeof event._datalake_key !== "string") return - const match = event._datalake_key.match(LAKE_TYPE) - if (!match?.[1] || !match[2]) return - return { - database: match[1], - table: match[2], - } -} - -function toFirehoseRecord(event: IngestEvent, route: LakeRoute): FirehoseRecord { - return { - Data: Buffer.from( - JSON.stringify({ - ...Object.fromEntries(Object.entries(event).filter(([key]) => key !== "_datalake_key")), - _lake_database: route.database, - _lake_table: route.table, - _lake_operation: "insert" as const, - }), - ), - } -} diff --git a/packages/stats/server/src/resource.d.ts b/packages/stats/server/src/resource.d.ts deleted file mode 100644 index 514b4d2afe..0000000000 --- a/packages/stats/server/src/resource.d.ts +++ /dev/null @@ -1,11 +0,0 @@ -import "sst/resource" - -declare module "sst/resource" { - export interface Resource { - LakeIngestConfig: { - secret: string - streamName: string - type: "sst.sst.Linkable" - } - } -} diff --git a/packages/stats/server/src/router.ts b/packages/stats/server/src/router.ts deleted file mode 100644 index e951118079..0000000000 --- a/packages/stats/server/src/router.ts +++ /dev/null @@ -1,73 +0,0 @@ -import { Buffer } from "node:buffer" -import { timingSafeEqual } from "node:crypto" -import { Effect, Schema } from "effect" -import * as Semaphore from "effect/Semaphore" -import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" -import { Resource } from "sst/resource" -import { Ingest } from "./ingest" -import { isShuttingDown } from "./shutdown" - -const MAX_CONCURRENT_INGEST_REQUESTS = 8 - -const IngestPayload = Schema.Struct({ - events: Schema.optional(Schema.Unknown), -}) - -export const Routes = HttpRouter.use((router) => - Effect.gen(function* () { - const ingestService = yield* Ingest - const ingestRequests = yield* Semaphore.make(MAX_CONCURRENT_INGEST_REQUESTS) - - yield* Effect.all( - [ - router.add("GET", "/health", () => json(200, { ok: true })), - router.add("GET", "/ready", () => json(isShuttingDown() ? 503 : 200, { ok: !isShuttingDown() })), - router.add("POST", "/", ingestRequests.withPermit(ingest(ingestService))), - ], - { discard: true }, - ) - }), -) - -const ingest = (ingestService: Ingest.Service) => - Effect.gen(function* () { - const request = yield* HttpServerRequest.HttpServerRequest - if (!isAuthorized(request.headers)) return yield* json(401, { ok: false, error: "Unauthorized" }) - - const payload = yield* HttpServerRequest.schemaBodyJson(IngestPayload).pipe( - Effect.match({ - onFailure: () => undefined, - onSuccess: (value) => value, - }), - ) - if (!payload) return yield* json(400, { ok: false, error: "Invalid JSON body" }) - - const events = Array.isArray(payload.events) ? payload.events : [] - if (events.length === 0) return yield* json(202, { ok: true, records: 0 }) - - return yield* ingestService.write(events).pipe( - Effect.flatMap((result) => json(202, { ok: true, records: result.records })), - Effect.catchTag("IngestError", (error) => - json(502, { ok: false, records: countRecords(events), failed: error.failed }), - ), - ) - }) - -function isAuthorized(headers: Record) { - const actual = Buffer.from(headers.authorization ?? headers.Authorization ?? "") - const expected = Buffer.from(`Bearer ${Resource.LakeIngestConfig.secret}`) - if (actual.length !== expected.length) return false - return timingSafeEqual(actual, expected) -} - -function countRecords(items: unknown[]) { - let records = 0 - for (const item of items) { - if (Boolean(item) && typeof item === "object" && !Array.isArray(item)) records++ - } - return records -} - -function json(status: number, body: Record) { - return HttpServerResponse.json(body, { status }).pipe(Effect.orDie) -} diff --git a/packages/stats/server/src/server.ts b/packages/stats/server/src/server.ts deleted file mode 100644 index 95104bc45e..0000000000 --- a/packages/stats/server/src/server.ts +++ /dev/null @@ -1,28 +0,0 @@ -import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer" -import * as NodeRuntime from "@effect/platform-node/NodeRuntime" -import { Config, Layer } from "effect" -import { HttpRouter } from "effect/unstable/http" -import { createServer } from "node:http" -import { Ingest } from "./ingest" -import { Routes } from "./router" -import { registerShutdownSignalHandlers } from "./shutdown" - -registerShutdownSignalHandlers() - -const ServerLive = NodeHttpServer.layerConfig( - () => createServer(), - Config.all({ - port: Config.number("PORT").pipe(Config.withDefault(3000)), - host: Config.string("HOST").pipe(Config.withDefault("0.0.0.0")), - }), -) - -const runtimeLayer = Ingest.layer -const programLayer = Routes.pipe(Layer.provide(runtimeLayer)) -const main = Layer.launch( - HttpRouter.serve(programLayer, { - disableLogger: true, - }).pipe(Layer.provideMerge(ServerLive)), -) - -NodeRuntime.runMain(main, { disableErrorReporting: true }) diff --git a/packages/stats/server/src/shutdown.ts b/packages/stats/server/src/shutdown.ts deleted file mode 100644 index 43938b3268..0000000000 --- a/packages/stats/server/src/shutdown.ts +++ /dev/null @@ -1,17 +0,0 @@ -let shuttingDown = false -let signalHandlersRegistered = false - -export function isShuttingDown() { - return shuttingDown -} - -export function registerShutdownSignalHandlers() { - if (signalHandlersRegistered) return - signalHandlersRegistered = true - process.once("SIGTERM", markShuttingDown) - process.once("SIGINT", markShuttingDown) -} - -function markShuttingDown() { - shuttingDown = true -} diff --git a/sst.config.ts b/sst.config.ts index b245a51324..41321005bf 100644 --- a/sst.config.ts +++ b/sst.config.ts @@ -30,7 +30,6 @@ export default $config({ async run() { const stage = await import("./infra/stage.js") await import("./infra/app.js") - const lake = stage.deployAws ? await import("./infra/lake.js") : undefined const stats = stage.deployAws ? await import("./infra/stats.js") : undefined const { stat } = await import("./infra/console.js") await import("./infra/enterprise.js") @@ -41,12 +40,6 @@ export default $config({ return { StatWorkerUrl: stat.url, ...(stats ? { StatsUrl: stats.app.url } : {}), - ...(lake - ? { - LakeUrl: lake.lakeIngest.properties.url, - LakeSecretSsm: lake.ingestSecretSsm.name, - } - : {}), AwsStage: stage.awsStage, } },