chore: generate
This commit is contained in:
opencode-agent[bot]
+99
-117
@@ -13,13 +13,10 @@ const glueS3TablesDatabaseWildcardArn = $interpolate`arn:${partition.partition}:
|
|||||||
const glueS3TablesTableWildcardArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${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/*`
|
const s3TablesBucketWildcardArn = $interpolate`arn:${partition.partition}:s3tables:${region.region}:${current.accountId}:bucket/*`
|
||||||
|
|
||||||
export const tableBucket = new aws.s3tables.TableBucket(
|
export const tableBucket = new aws.s3tables.TableBucket("LakeTableBucket", {
|
||||||
"LakeTableBucket",
|
name: tableBucketName,
|
||||||
{
|
forceDestroy: $app.stage !== "production",
|
||||||
name: tableBucketName,
|
})
|
||||||
forceDestroy: $app.stage !== "production",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
const s3TablesCatalog = new aws.cloudcontrol.Resource(
|
const s3TablesCatalog = new aws.cloudcontrol.Resource(
|
||||||
"LakeS3TablesCatalog",
|
"LakeS3TablesCatalog",
|
||||||
@@ -54,122 +51,107 @@ const s3TablesCatalog = new aws.cloudcontrol.Resource(
|
|||||||
{ dependsOn: [tableBucket] },
|
{ dependsOn: [tableBucket] },
|
||||||
)
|
)
|
||||||
|
|
||||||
const athenaResultsBucket = new aws.s3.Bucket(
|
const athenaResultsBucket = new aws.s3.Bucket("LakeAthenaResults", {
|
||||||
"LakeAthenaResults",
|
bucket: `opencode-${$app.stage}-lake-athena-results`,
|
||||||
{
|
forceDestroy: $app.stage !== "production",
|
||||||
bucket: `opencode-${$app.stage}-lake-athena-results`,
|
})
|
||||||
forceDestroy: $app.stage !== "production",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
const firehoseErrorBucket = new aws.s3.Bucket(
|
const firehoseErrorBucket = new aws.s3.Bucket("LakeFirehoseErrors", {
|
||||||
"LakeFirehoseErrors",
|
bucket: `opencode-${$app.stage}-lake-firehose-errors`,
|
||||||
{
|
forceDestroy: $app.stage !== "production",
|
||||||
bucket: `opencode-${$app.stage}-lake-firehose-errors`,
|
})
|
||||||
forceDestroy: $app.stage !== "production",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
const athenaWorkgroup = new aws.athena.Workgroup(
|
const athenaWorkgroup = new aws.athena.Workgroup("LakeAthenaWorkgroup", {
|
||||||
"LakeAthenaWorkgroup",
|
name: `opencode-${$app.stage}-lake-workgroup`,
|
||||||
{
|
forceDestroy: $app.stage !== "production",
|
||||||
name: `opencode-${$app.stage}-lake-workgroup`,
|
configuration: {
|
||||||
forceDestroy: $app.stage !== "production",
|
enforceWorkgroupConfiguration: true,
|
||||||
configuration: {
|
publishCloudwatchMetricsEnabled: true,
|
||||||
enforceWorkgroupConfiguration: true,
|
resultConfiguration: {
|
||||||
publishCloudwatchMetricsEnabled: true,
|
outputLocation: $interpolate`s3://${athenaResultsBucket.bucket}/`,
|
||||||
resultConfiguration: {
|
|
||||||
outputLocation: $interpolate`s3://${athenaResultsBucket.bucket}/`,
|
|
||||||
},
|
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
)
|
})
|
||||||
|
|
||||||
const firehoseRole = new aws.iam.Role(
|
const firehoseRole = new aws.iam.Role("LakeFirehoseRole", {
|
||||||
"LakeFirehoseRole",
|
assumeRolePolicy: aws.iam.getPolicyDocumentOutput({
|
||||||
{
|
statements: [
|
||||||
assumeRolePolicy: aws.iam.getPolicyDocumentOutput({
|
{
|
||||||
statements: [
|
effect: "Allow",
|
||||||
{
|
actions: ["sts:AssumeRole"],
|
||||||
effect: "Allow",
|
principals: [
|
||||||
actions: ["sts:AssumeRole"],
|
{
|
||||||
principals: [
|
type: "Service",
|
||||||
{
|
identifiers: ["firehose.amazonaws.com"],
|
||||||
type: "Service",
|
},
|
||||||
identifiers: ["firehose.amazonaws.com"],
|
],
|
||||||
},
|
},
|
||||||
],
|
],
|
||||||
},
|
}).json,
|
||||||
],
|
})
|
||||||
}).json,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
const firehosePolicy = new aws.iam.RolePolicy(
|
const firehosePolicy = new aws.iam.RolePolicy("LakeFirehosePolicy", {
|
||||||
"LakeFirehosePolicy",
|
role: firehoseRole.id,
|
||||||
{
|
policy: aws.iam.getPolicyDocumentOutput({
|
||||||
role: firehoseRole.id,
|
statements: [
|
||||||
policy: aws.iam.getPolicyDocumentOutput({
|
{
|
||||||
statements: [
|
effect: "Allow",
|
||||||
{
|
actions: [
|
||||||
effect: "Allow",
|
"s3tables:ListTableBuckets",
|
||||||
actions: [
|
"s3tables:GetTableBucket",
|
||||||
"s3tables:ListTableBuckets",
|
"s3tables:GetNamespace",
|
||||||
"s3tables:GetTableBucket",
|
"s3tables:GetTable",
|
||||||
"s3tables:GetNamespace",
|
"s3tables:GetTableData",
|
||||||
"s3tables:GetTable",
|
"s3tables:GetTableMetadataLocation",
|
||||||
"s3tables:GetTableData",
|
"s3tables:ListNamespaces",
|
||||||
"s3tables:GetTableMetadataLocation",
|
"s3tables:ListTables",
|
||||||
"s3tables:ListNamespaces",
|
"s3tables:PutTableData",
|
||||||
"s3tables:ListTables",
|
"s3tables:UpdateTableMetadataLocation",
|
||||||
"s3tables:PutTableData",
|
],
|
||||||
"s3tables:UpdateTableMetadataLocation",
|
resources: ["*"],
|
||||||
],
|
},
|
||||||
resources: ["*"],
|
{
|
||||||
},
|
effect: "Allow",
|
||||||
{
|
actions: [
|
||||||
effect: "Allow",
|
"glue:GetCatalog",
|
||||||
actions: [
|
"glue:GetCatalogs",
|
||||||
"glue:GetCatalog",
|
"glue:GetDatabase",
|
||||||
"glue:GetCatalogs",
|
"glue:GetDatabases",
|
||||||
"glue:GetDatabase",
|
"glue:GetTable",
|
||||||
"glue:GetDatabases",
|
"glue:GetTables",
|
||||||
"glue:GetTable",
|
"glue:UpdateTable",
|
||||||
"glue:GetTables",
|
],
|
||||||
"glue:UpdateTable",
|
resources: [
|
||||||
],
|
glueCatalogArn,
|
||||||
resources: [
|
glueS3TablesCatalogArn,
|
||||||
glueCatalogArn,
|
$interpolate`${glueS3TablesCatalogArn}/*`,
|
||||||
glueS3TablesCatalogArn,
|
glueS3TablesDatabaseWildcardArn,
|
||||||
$interpolate`${glueS3TablesCatalogArn}/*`,
|
glueS3TablesTableWildcardArn,
|
||||||
glueS3TablesDatabaseWildcardArn,
|
$interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/*`,
|
||||||
glueS3TablesTableWildcardArn,
|
$interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/*/*`,
|
||||||
$interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/*`,
|
$interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/*`,
|
||||||
$interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/*/*`,
|
],
|
||||||
$interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/*`,
|
},
|
||||||
],
|
{
|
||||||
},
|
effect: "Allow",
|
||||||
{
|
actions: [
|
||||||
effect: "Allow",
|
"s3:AbortMultipartUpload",
|
||||||
actions: [
|
"s3:GetBucketLocation",
|
||||||
"s3:AbortMultipartUpload",
|
"s3:GetObject",
|
||||||
"s3:GetBucketLocation",
|
"s3:ListBucket",
|
||||||
"s3:GetObject",
|
"s3:ListBucketMultipartUploads",
|
||||||
"s3:ListBucket",
|
"s3:PutObject",
|
||||||
"s3:ListBucketMultipartUploads",
|
],
|
||||||
"s3:PutObject",
|
resources: [firehoseErrorBucket.arn, $interpolate`${firehoseErrorBucket.arn}/*`],
|
||||||
],
|
},
|
||||||
resources: [firehoseErrorBucket.arn, $interpolate`${firehoseErrorBucket.arn}/*`],
|
{
|
||||||
},
|
effect: "Allow",
|
||||||
{
|
actions: ["lakeformation:GetDataAccess"],
|
||||||
effect: "Allow",
|
resources: ["*"],
|
||||||
actions: ["lakeformation:GetDataAccess"],
|
},
|
||||||
resources: ["*"],
|
],
|
||||||
},
|
}).json,
|
||||||
],
|
})
|
||||||
}).json,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
const firehose = new aws.kinesis.FirehoseDeliveryStream(
|
const firehose = new aws.kinesis.FirehoseDeliveryStream(
|
||||||
"LakeFirehose",
|
"LakeFirehose",
|
||||||
|
|||||||
@@ -2,9 +2,7 @@
|
|||||||
"version": "6",
|
"version": "6",
|
||||||
"dialect": "mysql",
|
"dialect": "mysql",
|
||||||
"id": "72655266-65da-408e-bfd8-9f3a4ad817a5",
|
"id": "72655266-65da-408e-bfd8-9f3a4ad817a5",
|
||||||
"prevIds": [
|
"prevIds": ["00000000-0000-0000-0000-000000000000"],
|
||||||
"00000000-0000-0000-0000-000000000000"
|
|
||||||
],
|
|
||||||
"ddl": [
|
"ddl": [
|
||||||
{
|
{
|
||||||
"name": "stat",
|
"name": "stat",
|
||||||
@@ -515,9 +513,7 @@
|
|||||||
"table": "stat"
|
"table": "stat"
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"columns": [
|
"columns": ["id"],
|
||||||
"id"
|
|
||||||
],
|
|
||||||
"name": "PRIMARY",
|
"name": "PRIMARY",
|
||||||
"table": "stat",
|
"table": "stat",
|
||||||
"entityType": "pks"
|
"entityType": "pks"
|
||||||
@@ -624,4 +620,4 @@
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
"renames": []
|
"renames": []
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,9 +2,7 @@
|
|||||||
"version": "6",
|
"version": "6",
|
||||||
"dialect": "mysql",
|
"dialect": "mysql",
|
||||||
"id": "e246639a-0da0-4fbd-b7bb-f1781d407780",
|
"id": "e246639a-0da0-4fbd-b7bb-f1781d407780",
|
||||||
"prevIds": [
|
"prevIds": ["72655266-65da-408e-bfd8-9f3a4ad817a5"],
|
||||||
"72655266-65da-408e-bfd8-9f3a4ad817a5"
|
|
||||||
],
|
|
||||||
"ddl": [
|
"ddl": [
|
||||||
{
|
{
|
||||||
"name": "geo_stat",
|
"name": "geo_stat",
|
||||||
@@ -1601,25 +1599,19 @@
|
|||||||
"table": "provider_stat"
|
"table": "provider_stat"
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"columns": [
|
"columns": ["id"],
|
||||||
"id"
|
|
||||||
],
|
|
||||||
"name": "PRIMARY",
|
"name": "PRIMARY",
|
||||||
"table": "geo_stat",
|
"table": "geo_stat",
|
||||||
"entityType": "pks"
|
"entityType": "pks"
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"columns": [
|
"columns": ["id"],
|
||||||
"id"
|
|
||||||
],
|
|
||||||
"name": "PRIMARY",
|
"name": "PRIMARY",
|
||||||
"table": "model_stat",
|
"table": "model_stat",
|
||||||
"entityType": "pks"
|
"entityType": "pks"
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"columns": [
|
"columns": ["id"],
|
||||||
"id"
|
|
||||||
],
|
|
||||||
"name": "PRIMARY",
|
"name": "PRIMARY",
|
||||||
"table": "provider_stat",
|
"table": "provider_stat",
|
||||||
"entityType": "pks"
|
"entityType": "pks"
|
||||||
@@ -2038,4 +2030,4 @@
|
|||||||
}
|
}
|
||||||
],
|
],
|
||||||
"renames": []
|
"renames": []
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,11 +10,7 @@ export type StatDimension = "model" | "provider" | "geo"
|
|||||||
export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: StatDimension) {
|
export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: StatDimension) {
|
||||||
const periodStartValue = sqlString(periodStart.toISOString())
|
const periodStartValue = sqlString(periodStart.toISOString())
|
||||||
const periodEndValue = sqlString(periodEnd.toISOString())
|
const periodEndValue = sqlString(periodEnd.toISOString())
|
||||||
const sourceTable = [
|
const sourceTable = [Resource.InferenceEvent.catalog, Resource.InferenceEvent.database, Resource.InferenceEvent.table]
|
||||||
Resource.InferenceEvent.catalog,
|
|
||||||
Resource.InferenceEvent.database,
|
|
||||||
Resource.InferenceEvent.table,
|
|
||||||
]
|
|
||||||
.map(sqlIdentifier)
|
.map(sqlIdentifier)
|
||||||
.join(".")
|
.join(".")
|
||||||
const dimensionSql = (() => {
|
const dimensionSql = (() => {
|
||||||
|
|||||||
@@ -40,15 +40,13 @@ export class Ingest extends Context.Service<Ingest, Ingest.Service>()("@opencode
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
const failed = (
|
const failed = (yield* Effect.all(
|
||||||
yield* Effect.all(
|
chunks(
|
||||||
chunks(
|
records.map((event) => ({ Data: Buffer.from(JSON.stringify(event)) })),
|
||||||
records.map((event) => ({ Data: Buffer.from(JSON.stringify(event)) })),
|
MAX_FIREHOSE_BATCH_SIZE,
|
||||||
MAX_FIREHOSE_BATCH_SIZE,
|
).map((batch) => putRecords(client, Resource.LakeIngestConfig.streamName, batch)),
|
||||||
).map((batch) => putRecords(client, Resource.LakeIngestConfig.streamName, batch)),
|
{ concurrency: 8 },
|
||||||
{ concurrency: 8 },
|
)).reduce((sum, item) => sum + item, 0)
|
||||||
)
|
|
||||||
).reduce((sum, item) => sum + item, 0)
|
|
||||||
|
|
||||||
if (failed > 0) {
|
if (failed > 0) {
|
||||||
return yield* new IngestError({ message: "Failed to ingest all lake records", failed })
|
return yield* new IngestError({ message: "Failed to ingest all lake records", failed })
|
||||||
@@ -75,7 +73,8 @@ const putRecords: (
|
|||||||
) {
|
) {
|
||||||
const result = yield* Effect.tryPromise({
|
const result = yield* Effect.tryPromise({
|
||||||
try: () => client.send(new PutRecordBatchCommand({ DeliveryStreamName: streamName, Records: records })),
|
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 }),
|
catch: (cause) =>
|
||||||
|
new IngestError({ message: "Failed to write lake records to Firehose", failed: records.length, cause }),
|
||||||
})
|
})
|
||||||
const failed =
|
const failed =
|
||||||
result.RequestResponses?.flatMap((item, index) => {
|
result.RequestResponses?.flatMap((item, index) => {
|
||||||
|
|||||||
@@ -25,26 +25,27 @@ export const Routes = HttpRouter.use((router) =>
|
|||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
const ingest = (ingestService: Ingest.Service) => Effect.gen(function* () {
|
const ingest = (ingestService: Ingest.Service) =>
|
||||||
const request = yield* HttpServerRequest.HttpServerRequest
|
Effect.gen(function* () {
|
||||||
if (!isAuthorized(request.headers)) return yield* json(401, { ok: false, error: "Unauthorized" })
|
const request = yield* HttpServerRequest.HttpServerRequest
|
||||||
|
if (!isAuthorized(request.headers)) return yield* json(401, { ok: false, error: "Unauthorized" })
|
||||||
|
|
||||||
const payload = yield* HttpServerRequest.schemaBodyJson(IngestPayload).pipe(
|
const payload = yield* HttpServerRequest.schemaBodyJson(IngestPayload).pipe(
|
||||||
Effect.match({
|
Effect.match({
|
||||||
onFailure: () => undefined,
|
onFailure: () => undefined,
|
||||||
onSuccess: (value) => value,
|
onSuccess: (value) => value,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
if (!payload) return yield* json(400, { ok: false, error: "Invalid JSON body" })
|
if (!payload) return yield* json(400, { ok: false, error: "Invalid JSON body" })
|
||||||
|
|
||||||
const events = Array.isArray(payload.events) ? payload.events.filter(isRecord) : []
|
const events = Array.isArray(payload.events) ? payload.events.filter(isRecord) : []
|
||||||
if (events.length === 0) return yield* json(202, { ok: true, records: 0 })
|
if (events.length === 0) return yield* json(202, { ok: true, records: 0 })
|
||||||
|
|
||||||
return yield* ingestService.write(events).pipe(
|
return yield* ingestService.write(events).pipe(
|
||||||
Effect.flatMap((result) => json(202, { ok: true, records: result.records })),
|
Effect.flatMap((result) => json(202, { ok: true, records: result.records })),
|
||||||
Effect.catchTag("IngestError", (error) => json(502, { ok: false, records: events.length, failed: error.failed })),
|
Effect.catchTag("IngestError", (error) => json(502, { ok: false, records: events.length, failed: error.failed })),
|
||||||
)
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
function isAuthorized(headers: Record<string, string | undefined>) {
|
function isAuthorized(headers: Record<string, string | undefined>) {
|
||||||
const actual = Buffer.from(headers.authorization ?? headers.Authorization ?? "")
|
const actual = Buffer.from(headers.authorization ?? headers.Authorization ?? "")
|
||||||
|
|||||||
Reference in New Issue
Block a user