mirror of
https://github.com/anomalyco/opencode.git
synced 2026-10-21 18:29:58 +00:00
chore(stats): retire the S3 data lake (#52514)
This commit is contained in:
Adam
GitHub
parent
0112a92c41
commit
8bb2ccf829
9 changed files with 37 additions and 380 deletions
5
bun.lock
5
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=="],
|
||||
|
|
|
|||
|
|
@ -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", {
|
||||
|
|
|
|||
|
|
@ -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: ["*"],
|
||||
},
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>) {
|
||||
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<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: 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: Record<string, unknown>, key: string) {
|
||||
const value = number(data, key)
|
||||
if (value === undefined) return undefined
|
||||
return Math.round(value)
|
||||
}
|
||||
|
||||
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") {
|
||||
const parsed = Number(value)
|
||||
return Number.isFinite(parsed) ? parsed : undefined
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:",
|
||||
|
|
|
|||
|
|
@ -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<string, string>
|
||||
|
||||
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<AthenaData[], AthenaQueryError | AthenaQueryTimeoutError>
|
||||
}
|
||||
}
|
||||
|
||||
export class Athena extends Context.Service<Athena, Athena.Service>()("@opencode/stats/Athena") {
|
||||
static readonly layer: Layer.Layer<Athena> = 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<void, AthenaQueryError | AthenaQueryTimeoutError> = 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<AthenaData[], AthenaQueryError> =
|
||||
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]]
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
9
packages/stats/core/src/resource.d.ts
vendored
9
packages/stats/core/src/resource.d.ts
vendored
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue