mirror of
https://github.com/Heretek-AI/opencode.git
synced 2026-10-12 02:04:25 +00:00
chore(stats): remove s3 lake ingestion (#52572)
This commit is contained in:
Adam
@@ -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/[email protected]", "", { "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/[email protected]", "", { "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/[email protected]", "", { "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/[email protected]", "", { "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/[email protected]", "", { "dependencies": { "@smithy/types": "^4.14.2", "tslib": "^2.6.2" } }, "sha512-kuBfgQVdcz5Bmapc4A13YbpVw/pXkesfhetcFYwbntqas8sF41OHyd4o28+/TG2ZQdHBsv90Lsu5y6oitvYCdg=="],
|
||||
|
||||
"@aws-sdk/client-firehose/@smithy/util-utf8": ["@smithy/[email protected]", "", { "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/[email protected]", "", { "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/[email protected]", "", { "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=="],
|
||||
|
||||
+8
-207
@@ -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,
|
||||
},
|
||||
})
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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:",
|
||||
|
||||
@@ -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<string, unknown>
|
||||
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<Ingest, Ingest.Service>()("@opencode/stats/Ingest") {
|
||||
static readonly layer: Layer.Layer<Ingest> = 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<number, IngestError> = 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,
|
||||
}),
|
||||
),
|
||||
}
|
||||
}
|
||||
-11
@@ -1,11 +0,0 @@
|
||||
import "sst/resource"
|
||||
|
||||
declare module "sst/resource" {
|
||||
export interface Resource {
|
||||
LakeIngestConfig: {
|
||||
secret: string
|
||||
streamName: string
|
||||
type: "sst.sst.Linkable"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<string, string | undefined>) {
|
||||
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<string, unknown>) {
|
||||
return HttpServerResponse.json(body, { status }).pipe(Effect.orDie)
|
||||
}
|
||||
@@ -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 })
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,
|
||||
}
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user