lake.ts 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331
  1. import { domain } from "./stage"
  2. const current = aws.getCallerIdentityOutput({})
  3. const partition = aws.getPartitionOutput({})
  4. const region = aws.getRegionOutput({})
  5. const tableBucketName = `opencode-${$app.stage}-lake`
  6. const glueCatalogName = "s3tablescatalog"
  7. const glueCatalogArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:catalog`
  8. const glueS3TablesCatalogArn = $interpolate`${glueCatalogArn}/${glueCatalogName}`
  9. const glueS3TablesChildCatalogArn = $interpolate`${glueS3TablesCatalogArn}/${tableBucketName}`
  10. const glueS3TablesDatabaseWildcardArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/${glueCatalogName}/${tableBucketName}/*`
  11. const glueS3TablesTableWildcardArn = $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/${tableBucketName}/*/*`
  12. const s3TablesBucketWildcardArn = $interpolate`arn:${partition.partition}:s3tables:${region.region}:${current.accountId}:bucket/*`
  13. export const tableBucket = new aws.s3tables.TableBucket("LakeTableBucket", {
  14. name: tableBucketName,
  15. forceDestroy: $app.stage !== "production",
  16. })
  17. const s3TablesCatalog = new aws.cloudcontrol.Resource(
  18. "LakeS3TablesCatalog",
  19. {
  20. typeName: "AWS::Glue::Catalog",
  21. desiredState: $jsonStringify({
  22. Name: glueCatalogName,
  23. Description: "Federated catalog for S3 Tables",
  24. FederatedCatalog: {
  25. Identifier: s3TablesBucketWildcardArn,
  26. ConnectionName: "aws:s3tables",
  27. },
  28. CreateDatabaseDefaultPermissions: [
  29. {
  30. Principal: {
  31. DataLakePrincipalIdentifier: "IAM_ALLOWED_PRINCIPALS",
  32. },
  33. Permissions: ["ALL"],
  34. },
  35. ],
  36. CreateTableDefaultPermissions: [
  37. {
  38. Principal: {
  39. DataLakePrincipalIdentifier: "IAM_ALLOWED_PRINCIPALS",
  40. },
  41. Permissions: ["ALL"],
  42. },
  43. ],
  44. AllowFullTableExternalDataAccess: "True",
  45. }),
  46. },
  47. { dependsOn: [tableBucket] },
  48. )
  49. const athenaResultsBucket = new aws.s3.Bucket("LakeAthenaResults", {
  50. bucket: `opencode-${$app.stage}-lake-athena-results`,
  51. forceDestroy: $app.stage !== "production",
  52. })
  53. const firehoseErrorBucket = new aws.s3.Bucket("LakeFirehoseErrors", {
  54. bucket: `opencode-${$app.stage}-lake-firehose-errors`,
  55. forceDestroy: $app.stage !== "production",
  56. })
  57. const athenaWorkgroup = new aws.athena.Workgroup("LakeAthenaWorkgroup", {
  58. name: `opencode-${$app.stage}-lake-workgroup`,
  59. forceDestroy: $app.stage !== "production",
  60. configuration: {
  61. enforceWorkgroupConfiguration: true,
  62. publishCloudwatchMetricsEnabled: true,
  63. // Athena bills $5/TB scanned; kill any query that would scan more than 2 TB
  64. // so a regression cannot silently burn money. Stats sync full passes scan
  65. // ~250 GB as of 2026-07.
  66. bytesScannedCutoffPerQuery: 2 * 1024 ** 4,
  67. resultConfiguration: {
  68. outputLocation: $interpolate`s3://${athenaResultsBucket.bucket}/`,
  69. },
  70. },
  71. })
  72. const firehoseRole = new aws.iam.Role("LakeFirehoseRole", {
  73. assumeRolePolicy: aws.iam.getPolicyDocumentOutput({
  74. statements: [
  75. {
  76. effect: "Allow",
  77. actions: ["sts:AssumeRole"],
  78. principals: [
  79. {
  80. type: "Service",
  81. identifiers: ["firehose.amazonaws.com"],
  82. },
  83. ],
  84. },
  85. ],
  86. }).json,
  87. })
  88. const firehosePolicy = new aws.iam.RolePolicy("LakeFirehosePolicy", {
  89. role: firehoseRole.id,
  90. policy: aws.iam.getPolicyDocumentOutput({
  91. statements: [
  92. {
  93. effect: "Allow",
  94. actions: [
  95. "s3tables:ListTableBuckets",
  96. "s3tables:GetTableBucket",
  97. "s3tables:GetNamespace",
  98. "s3tables:GetTable",
  99. "s3tables:GetTableData",
  100. "s3tables:GetTableMetadataLocation",
  101. "s3tables:ListNamespaces",
  102. "s3tables:ListTables",
  103. "s3tables:PutTableData",
  104. "s3tables:UpdateTableMetadataLocation",
  105. ],
  106. resources: ["*"],
  107. },
  108. {
  109. effect: "Allow",
  110. actions: [
  111. "glue:GetCatalog",
  112. "glue:GetCatalogs",
  113. "glue:GetDatabase",
  114. "glue:GetDatabases",
  115. "glue:GetTable",
  116. "glue:GetTables",
  117. "glue:UpdateTable",
  118. ],
  119. resources: [
  120. glueCatalogArn,
  121. glueS3TablesCatalogArn,
  122. $interpolate`${glueS3TablesCatalogArn}/*`,
  123. glueS3TablesDatabaseWildcardArn,
  124. glueS3TablesTableWildcardArn,
  125. $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/*`,
  126. $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/*/*`,
  127. $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/*`,
  128. ],
  129. },
  130. {
  131. effect: "Allow",
  132. actions: [
  133. "s3:AbortMultipartUpload",
  134. "s3:GetBucketLocation",
  135. "s3:GetObject",
  136. "s3:ListBucket",
  137. "s3:ListBucketMultipartUploads",
  138. "s3:PutObject",
  139. ],
  140. resources: [firehoseErrorBucket.arn, $interpolate`${firehoseErrorBucket.arn}/*`],
  141. },
  142. {
  143. effect: "Allow",
  144. actions: ["lakeformation:GetDataAccess"],
  145. resources: ["*"],
  146. },
  147. ],
  148. }).json,
  149. })
  150. const firehose = new aws.kinesis.FirehoseDeliveryStream(
  151. "LakeFirehose",
  152. {
  153. name: `opencode-${$app.stage}-lake-ingest`,
  154. destination: "iceberg",
  155. icebergConfiguration: {
  156. appendOnly: true,
  157. bufferingInterval: 60,
  158. bufferingSize: 1,
  159. catalogArn: glueS3TablesChildCatalogArn,
  160. processingConfiguration: {
  161. enabled: true,
  162. processors: [
  163. {
  164. type: "MetadataExtraction",
  165. parameters: [
  166. { parameterName: "JsonParsingEngine", parameterValue: "JQ-1.6" },
  167. {
  168. parameterName: "MetadataExtractionQuery",
  169. parameterValue:
  170. '{destinationDatabaseName:._lake_database,destinationTableName:._lake_table,operation:(._lake_operation // "insert")}',
  171. },
  172. ],
  173. },
  174. ],
  175. },
  176. roleArn: firehoseRole.arn,
  177. s3BackupMode: "FailedDataOnly",
  178. s3Configuration: {
  179. roleArn: firehoseRole.arn,
  180. bucketArn: firehoseErrorBucket.arn,
  181. errorOutputPrefix: "errors/!{firehose:error-output-type}/",
  182. },
  183. },
  184. },
  185. { dependsOn: [s3TablesCatalog, firehosePolicy] },
  186. )
  187. export const lakeVpc = new sst.aws.Vpc("LakeVpc")
  188. export const lakeCluster = new sst.aws.Cluster("LakeCluster", { vpc: lakeVpc })
  189. export const lakeRegion = region.region
  190. export const lakeCatalog = $interpolate`${glueCatalogName}/${tableBucket.name}`
  191. export const lakeAthenaWorkgroup = athenaWorkgroup
  192. const ingestSecret = new random.RandomPassword("LakeIngestSecret", { length: 32 })
  193. export const ingestSecretSsm = new aws.ssm.Parameter("LakeIngestSecretSsm", {
  194. name: $interpolate`/${$app.name}/${$app.stage}/lake/ingest/secret`,
  195. type: "SecureString",
  196. value: ingestSecret.result,
  197. })
  198. const ingestConfig = new sst.Linkable("LakeIngestConfig", {
  199. properties: {
  200. streamName: firehose.name,
  201. secret: ingestSecret.result,
  202. },
  203. })
  204. const ingestService = new sst.aws.Service("LakeIngestService", {
  205. cluster: lakeCluster,
  206. architecture: "arm64",
  207. cpu: "1 vCPU",
  208. memory: "4 GB",
  209. image: {
  210. context: ".",
  211. dockerfile: "packages/stats/server/Dockerfile",
  212. },
  213. link: [ingestConfig],
  214. permissions: [
  215. {
  216. actions: ["firehose:PutRecord", "firehose:PutRecordBatch"],
  217. resources: [firehose.arn],
  218. },
  219. ],
  220. scaling: {
  221. min: $app.stage === "production" ? 2 : 1,
  222. max: $app.stage === "production" ? 32 : 4,
  223. cpuUtilization: 60,
  224. memoryUtilization: 70,
  225. },
  226. loadBalancer: {
  227. domain: {
  228. name: `lake.${domain}`,
  229. dns: sst.cloudflare.dns(),
  230. },
  231. rules: [
  232. { listen: "80/http", redirect: "443/https" },
  233. { listen: "443/https", forward: "3000/http" },
  234. ],
  235. health: {
  236. "3000/http": {
  237. path: "/ready",
  238. successCodes: "200-299",
  239. },
  240. },
  241. },
  242. health: {
  243. command: [
  244. "CMD-SHELL",
  245. "bun --eval \"fetch('http://localhost:3000/health').then((r) => process.exit(r.ok ? 0 : 1)).catch(() => process.exit(1))\"",
  246. ],
  247. interval: "30 seconds",
  248. retries: 3,
  249. startPeriod: "30 seconds",
  250. timeout: "5 seconds",
  251. },
  252. dev: {
  253. command: "bun run start",
  254. directory: "packages/stats/server",
  255. url: "http://localhost:3000",
  256. },
  257. wait: $app.stage === "production",
  258. })
  259. export const lakeIngest = new sst.Linkable("LakeIngest", {
  260. properties: {
  261. url: ingestService.url,
  262. secret: ingestSecret.result,
  263. },
  264. })
  265. export const lakeQueryPermissions = [
  266. {
  267. actions: ["athena:StartQueryExecution", "athena:GetQueryExecution", "athena:GetQueryResults"],
  268. resources: [athenaWorkgroup.arn],
  269. },
  270. {
  271. actions: [
  272. "glue:GetCatalog",
  273. "glue:GetCatalogs",
  274. "glue:GetDatabase",
  275. "glue:GetDatabases",
  276. "glue:GetTable",
  277. "glue:GetTables",
  278. "glue:GetPartitions",
  279. ],
  280. resources: [
  281. glueCatalogArn,
  282. glueS3TablesCatalogArn,
  283. $interpolate`${glueS3TablesCatalogArn}/*`,
  284. glueS3TablesDatabaseWildcardArn,
  285. glueS3TablesTableWildcardArn,
  286. $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:database/*`,
  287. $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/*/*`,
  288. $interpolate`arn:${partition.partition}:glue:${region.region}:${current.accountId}:table/${glueCatalogName}/*`,
  289. ],
  290. },
  291. {
  292. actions: ["s3:GetBucketLocation", "s3:ListBucket"],
  293. resources: [athenaResultsBucket.arn],
  294. },
  295. {
  296. actions: ["s3:GetObject", "s3:PutObject", "s3:AbortMultipartUpload", "s3:ListBucketMultipartUploads"],
  297. resources: [$interpolate`${athenaResultsBucket.arn}/*`],
  298. },
  299. {
  300. actions: [
  301. "s3tables:GetTableBucket",
  302. "s3tables:GetNamespace",
  303. "s3tables:GetTable",
  304. "s3tables:GetTableData",
  305. "s3tables:GetTableMetadataLocation",
  306. "s3tables:ListNamespaces",
  307. "s3tables:ListTables",
  308. ],
  309. resources: ["*"],
  310. },
  311. {
  312. actions: ["lakeformation:GetDataAccess"],
  313. resources: ["*"],
  314. },
  315. ]