From 5eee0b7e9b92b00b695ad684615c62e61880ccf5 Mon Sep 17 00:00:00 2001 From: Zakaria Date: Sun, 25 Oct 2026 18:10:33 -0400 Subject: [PATCH] feat(api): add Athena workgroups and a query editor to the cloud explorer Create and delete Athena workgroups (with a result folder, optionally enforced), and run SQL in one: pick a database from the Glue Data Catalog, preview a table, run with Ctrl+Enter, cancel a running query, read up to 1,000 rows and download the full result CSV from S3. Recent queries in the workgroup can be reopened. TheCloud runs the SQL for real over the files a Glue table points at. The Glue page links to the editor. --- README.md | 1 + packages/api/package.json | 1 + .../src/adapter-aws/AwsQueryAdapter.test.ts | 166 +++++++++++ .../api/src/adapter-aws/AwsQueryAdapter.ts | 268 ++++++++++++++++++ packages/api/src/aws.ts | 3 + packages/api/src/cloud-spi/querySchema.ts | 80 ++++++ packages/api/src/cloud-spi/serviceCatalog.ts | 7 + packages/api/src/cloud-spi/types.ts | 46 +++ packages/api/src/cloudProxy.test.ts | 20 +- packages/api/src/cloudProxy.ts | 2 + packages/api/src/routes/clouds.test.ts | 23 ++ packages/api/src/routes/clouds.ts | 39 +++ packages/api/src/service/CloudProxyService.ts | 34 +++ packages/frontend/src/api/api.ts | 42 +++ packages/frontend/src/api/cloudProxyClient.ts | 69 +++++ .../src/components/DataCatalogPanel.test.tsx | 4 +- .../src/components/DataCatalogPanel.tsx | 2 + .../src/components/DynamicResourceView.tsx | 4 + .../src/components/QueryPanel.test.tsx | 100 +++++++ .../frontend/src/components/QueryPanel.tsx | 185 ++++++++++++ .../frontend/src/components/serviceIcons.tsx | 4 + packages/frontend/src/index.css | 21 ++ packages/frontend/src/types/resource.ts | 33 +++ packages/frontend/src/types/schema.ts | 2 + pnpm-lock.yaml | 256 +++++++++-------- 25 files changed, 1291 insertions(+), 121 deletions(-) create mode 100644 packages/api/src/adapter-aws/AwsQueryAdapter.test.ts create mode 100644 packages/api/src/adapter-aws/AwsQueryAdapter.ts create mode 100644 packages/api/src/cloud-spi/querySchema.ts create mode 100644 packages/frontend/src/components/QueryPanel.test.tsx create mode 100644 packages/frontend/src/components/QueryPanel.tsx diff --git a/README.md b/README.md index a4ffddc..16d6e47 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,7 @@ cd packages/api && bun run scripts/service-matrix.ts | Analytics | MSK Connect / Managed Streaming for Kafka Connect | Yes (legacy page) | No | No | | Analytics | Kinesis / Kinesis Data Streams | Yes (list, create, inspect, delete) | No | No | | Analytics | Glue Data Catalog / Data Catalog | Yes (list, create, delete, inspect) | No | No | +| Analytics | Athena / Query | Yes (list, create, delete, inspect) | No | No | | Provisioning | CloudFormation / Infrastructure as Code | Yes (list, create, update, delete, inspect) | No | No | | Provisioning | AppConfig / Configuration | Yes (list, create, delete, inspect) | No | No | | Security | Identity | Yes (list, create, delete, inspect) | No | No | diff --git a/packages/api/package.json b/packages/api/package.json index dca7e7d..231ba9a 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -14,6 +14,7 @@ "@aws-sdk/client-api-gateway": "^3.1076.0", "@aws-sdk/client-apigatewayv2": "^3.1076.0", "@aws-sdk/client-appconfig": "^3.1076.0", + "@aws-sdk/client-athena": "^3.1146.0", "@aws-sdk/client-auto-scaling": "^3.1146.0", "@aws-sdk/client-cloudformation": "^3.1111.0", "@aws-sdk/client-cloudfront": "^3.1146.0", diff --git a/packages/api/src/adapter-aws/AwsQueryAdapter.test.ts b/packages/api/src/adapter-aws/AwsQueryAdapter.test.ts new file mode 100644 index 0000000..4ac7423 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsQueryAdapter.test.ts @@ -0,0 +1,166 @@ +import {describe, expect, test} from 'bun:test' +import { + type AthenaClient, + CreateWorkGroupCommand, + DeleteWorkGroupCommand, + GetQueryExecutionCommand, + GetQueryResultsCommand, + GetWorkGroupCommand, + ListDatabasesCommand, + ListQueryExecutionsCommand, + ListTableMetadataCommand, + ListWorkGroupsCommand, + StartQueryExecutionCommand, + StopQueryExecutionCommand, +} from '@aws-sdk/client-athena' +import {AwsQueryAdapter} from './AwsQueryAdapter' +import {NotFoundError, ValidationError} from '../cloud-spi/errors' + +type SendResult = Record + +function execution(id: string, overrides: Record = {}) { + return { + QueryExecutionId: id, + Query: 'SELECT name, amount FROM orders', + StatementType: 'DML', + WorkGroup: 'analysts', + QueryExecutionContext: {Database: 'shop', Catalog: 'AwsDataCatalog'}, + ResultConfiguration: {OutputLocation: `s3://results/${id}.csv`}, + Status: {State: 'SUCCEEDED', SubmissionDateTime: new Date('2026-10-25T21:00:00Z'), CompletionDateTime: new Date('2026-10-25T21:00:01Z')}, + Statistics: {TotalExecutionTimeInMillis: 42, DataScannedInBytes: 0}, + ...overrides, + } +} + +const header = {Data: [{VarCharValue: 'name'}, {VarCharValue: 'amount'}]} +const columnInfo = [{Name: 'name', Label: 'name', Type: 'varchar'}, {Name: 'amount', Label: 'amount', Type: 'double'}] +const row = (name: string, amount: string) => ({Data: [{VarCharValue: name}, {VarCharValue: amount}]}) + +function adapterWith(handler: (command: object) => SendResult | undefined = () => undefined) { + const sent: object[] = [] + const client = { + async send(command: object) { + sent.push(command) + const result = handler(command) + if (result) return result + if (command instanceof GetQueryExecutionCommand) return {QueryExecution: execution(command.input.QueryExecutionId ?? '')} + if (command instanceof GetWorkGroupCommand) { + return {WorkGroup: {Name: command.input.WorkGroup, State: 'ENABLED', Description: 'Team', Configuration: {ResultConfiguration: {OutputLocation: 's3://results/'}, EnforceWorkGroupConfiguration: true, EngineVersion: {EffectiveEngineVersion: 'Athena engine version 3'}}}} + } + if (command instanceof StartQueryExecutionCommand) return {QueryExecutionId: 'q-new'} + return {} + }, + } + return {sent, adapter: new AwsQueryAdapter(client as unknown as AthenaClient)} +} + +describe('AwsQueryAdapter', () => { + test('lists workgroups across pages and describes one with its result location', async () => { + const {adapter} = adapterWith((command) => command instanceof ListWorkGroupsCommand + ? command.input.NextToken ? {WorkGroups: [{Name: 'analysts', State: 'ENABLED', Description: 'Team'}]} : {WorkGroups: [{Name: 'primary', State: 'ENABLED'}], NextToken: 'n'} + : undefined) + + expect((await adapter.list()).map((wg) => [wg.name, wg.status])).toEqual([['primary', 'ENABLED'], ['analysts', 'ENABLED']]) + expect((await adapter.list({search: 'team'})).map((wg) => wg.name)).toEqual(['analysts']) + expect((await adapter.get('analysts'))?.metadata).toEqual({description: 'Team', outputLocation: 's3://results/', enforceConfiguration: true, engineVersion: 'Athena engine version 3'}) + }) + + test('creates a workgroup with a result folder, and keeps the primary workgroup', async () => { + const {adapter, sent} = adapterWith() + await adapter.create({values: {name: 'analysts', description: 'Team', outputLocation: 's3://results/athena', enforce: 'true'}}) + expect((sent.find((command) => command instanceof CreateWorkGroupCommand) as CreateWorkGroupCommand).input).toEqual({ + Name: 'analysts', Description: 'Team', Configuration: {ResultConfiguration: {OutputLocation: 's3://results/athena/'}, EnforceWorkGroupConfiguration: true}, + }) + + await expect(adapter.create({values: {name: 'primary'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {name: 'bad name'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.delete('primary')).rejects.toBeInstanceOf(ValidationError) + await adapter.delete('analysts') + expect((sent.find((command) => command instanceof DeleteWorkGroupCommand) as DeleteWorkGroupCommand).input).toEqual({WorkGroup: 'analysts', RecursiveDeleteOption: true}) + }) + + test('starts a query in the workgroup against a database', async () => { + const {adapter, sent} = adapterWith() + const started = await adapter.startQuery('analysts', {sql: ' SELECT 1 ', database: 'shop'}) + + expect((sent.find((command) => command instanceof StartQueryExecutionCommand) as StartQueryExecutionCommand).input).toEqual({ + QueryString: 'SELECT 1', WorkGroup: 'analysts', QueryExecutionContext: {Catalog: 'AwsDataCatalog', Database: 'shop'}, + }) + expect(started).toMatchObject({id: 'q-new', state: 'SUCCEEDED', runtimeMs: 42, outputLocation: 's3://results/q-new.csv'}) + await expect(adapter.startQuery('analysts', {sql: ' '})).rejects.toBeInstanceOf(ValidationError) + }) + + test('returns a succeeded query with its columns and rows, without the header row', async () => { + const {adapter} = adapterWith((command) => command instanceof GetQueryResultsCommand + ? {ResultSet: {ResultSetMetadata: {ColumnInfo: columnInfo}, Rows: [header, row('alpha', '10'), row('beta', '25')]}} + : undefined) + + const result = await adapter.getQueryResult('analysts', 'q1') + expect(result.columns).toEqual([{name: 'name', type: 'varchar'}, {name: 'amount', type: 'double'}]) + expect(result.rows).toEqual([['alpha', '10'], ['beta', '25']]) + expect(result.truncated).toBe(false) + }) + + test('stops reading after the row limit and says the result is truncated', async () => { + const page = Array.from({length: 1000}, (_, i) => row(`n${i}`, String(i))) + const {adapter, sent} = adapterWith((command) => command instanceof GetQueryResultsCommand + ? command.input.NextToken ? {ResultSet: {Rows: page}, NextToken: 'more'} : {ResultSet: {ResultSetMetadata: {ColumnInfo: columnInfo}, Rows: [header, ...page.slice(0, 999)]}, NextToken: 'p2'} + : undefined) + + const result = await adapter.getQueryResult('analysts', 'q1') + expect(result.rows).toHaveLength(1000) + expect(result.truncated).toBe(true) + expect(sent.filter((command) => command instanceof GetQueryResultsCommand)).toHaveLength(2) + }) + + test('does not read results of a query that has not succeeded', async () => { + const {adapter, sent} = adapterWith((command) => command instanceof GetQueryExecutionCommand + ? {QueryExecution: execution('q1', {Status: {State: 'FAILED', StateChangeReason: 'Table orders not found'}})} + : undefined) + + const result = await adapter.getQueryResult('analysts', 'q1') + expect(result.execution).toMatchObject({state: 'FAILED', stateReason: 'Table orders not found'}) + expect(result.rows).toEqual([]) + expect(sent.some((command) => command instanceof GetQueryResultsCommand)).toBe(false) + }) + + test('refuses a query from another workgroup', async () => { + const {adapter, sent} = adapterWith() + await expect(adapter.getQueryResult('primary', 'q1')).rejects.toBeInstanceOf(NotFoundError) + await expect(adapter.stopQuery('primary', 'q1')).rejects.toBeInstanceOf(NotFoundError) + expect(sent.some((command) => command instanceof StopQueryExecutionCommand)).toBe(false) + + await adapter.stopQuery('analysts', 'q1') + expect(sent.some((command) => command instanceof StopQueryExecutionCommand)).toBe(true) + }) + + test('lists the workgroup queries newest first', async () => { + const {adapter} = adapterWith((command) => { + if (command instanceof ListQueryExecutionsCommand) return {QueryExecutionIds: ['old', 'new', 'other']} + if (command instanceof GetQueryExecutionCommand) { + const id = command.input.QueryExecutionId + return {QueryExecution: execution(id ?? '', { + WorkGroup: id === 'other' ? 'primary' : 'analysts', + Status: {State: 'SUCCEEDED', SubmissionDateTime: new Date(id === 'new' ? '2026-10-25T22:00:00Z' : '2026-10-25T21:00:00Z')}, + })} + } + return undefined + }) + + expect((await adapter.listQueries('analysts')).map((query) => query.id)).toEqual(['new', 'old']) + }) + + test('lists databases and the tables of the chosen one', async () => { + const {adapter, sent} = adapterWith((command) => { + if (command instanceof ListDatabasesCommand) return {DatabaseList: [{Name: 'shop'}, {Name: 'logs'}]} + if (command instanceof ListTableMetadataCommand) { + return {TableMetadataList: [{Name: 'orders', Columns: [{Name: 'id', Type: 'int'}], PartitionKeys: [{Name: 'day', Type: 'string'}]}]} + } + return undefined + }) + + expect(await adapter.describeQueryCatalog('shop')).toEqual({databases: ['logs', 'shop'], database: 'shop', tables: [{name: 'orders', columns: [{name: 'id', type: 'int'}, {name: 'day', type: 'string'}]}]}) + expect((await adapter.describeQueryCatalog('missing')).database).toBe('logs') + expect((sent.filter((command) => command instanceof ListTableMetadataCommand) as ListTableMetadataCommand[]).map((command) => command.input.DatabaseName)).toEqual(['shop', 'logs']) + }) +}) diff --git a/packages/api/src/adapter-aws/AwsQueryAdapter.ts b/packages/api/src/adapter-aws/AwsQueryAdapter.ts new file mode 100644 index 0000000..d288d47 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsQueryAdapter.ts @@ -0,0 +1,268 @@ +import { + type AthenaClient, + CreateWorkGroupCommand, + DeleteWorkGroupCommand, + GetQueryExecutionCommand, + GetQueryResultsCommand, + type GetQueryResultsCommandOutput, + GetWorkGroupCommand, + ListDatabasesCommand, + ListQueryExecutionsCommand, + ListTableMetadataCommand, + type ListTableMetadataCommandOutput, + ListWorkGroupsCommand, + type QueryExecution, + StartQueryExecutionCommand, + StopQueryExecutionCommand, + type TableMetadata, + type WorkGroup, + type WorkGroupSummary, +} from '@aws-sdk/client-athena' +import {NotFoundError, RuntimeError, ValidationError} from '../cloud-spi/errors' +import { + awsQuerySchema, + PRIMARY_WORKGROUP, + QUERY_HISTORY_LIMIT, + QUERY_RESULT_ROW_LIMIT, + WORKGROUP_NAME_PATTERN, +} from '../cloud-spi/querySchema' +import type { + CloudResource, + CloudServiceAdapter, + CreateResourceInput, + QueryCatalog, + QueryExecutionSummary, + QueryInput, + QueryResult, + ResourceQuery, + ServiceSchema, +} from '../cloud-spi/types' + +/** Athena's own catalog name for the Glue Data Catalog. */ +const DATA_CATALOG = 'AwsDataCatalog' +/** Athena's cap on a query string. */ +const MAX_QUERY_LENGTH = 262_144 +const S3_URI = /^s3:\/\/[a-z0-9.-]{3,63}(\/.*)?$/ + +/** + * Talks to Amazon Athena, verified against Floci core (community, `latest`). + * + * Notes from probing the runtime: + * - Queries really run: TheCloud executes the SQL with DuckDB over the files a + * Glue table points at, and writes the result to S3 as CSV, as Athena does. + * - Statements that return no rows, such as SHOW TABLES, fail in the runtime, + * so the editor lists databases and tables through the catalog APIs instead. + * - Data scanned is always reported as 0 bytes, so the editor shows only the + * run time. + */ +export class AwsQueryAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'query' as const + + constructor(private readonly athena: AthenaClient) {} + + schema(): ServiceSchema { + return awsQuerySchema() + } + + async list(query: ResourceQuery = {}): Promise { + const found: WorkGroupSummary[] = [] + let next: string | undefined + do { + const res = await this.athena.send(new ListWorkGroupsCommand({MaxResults: 50, ...(next ? {NextToken: next} : {})})) + found.push(...(res.WorkGroups ?? [])) + next = res.NextToken + } while (next) + + const search = query.search?.trim().toLowerCase() + return found.map(toResource).filter((workGroup) => !search + || workGroup.name.toLowerCase().includes(search) + || String(workGroup.metadata.description ?? '').toLowerCase().includes(search)) + } + + async get(id: string): Promise { + try { + const workGroup = (await this.athena.send(new GetWorkGroupCommand({WorkGroup: id}))).WorkGroup + return workGroup ? toResource(workGroup) : null + } catch (error) { + if (errorName(error) === 'InvalidRequestException') return null + throw error + } + } + + async create(input: CreateResourceInput): Promise { + const values = input.values + const name = text(values.name) + if (!new RegExp(WORKGROUP_NAME_PATTERN).test(name)) throw new ValidationError('name must be up to 128 letters, numbers, dots, dashes or underscores') + if (name === PRIMARY_WORKGROUP) throw new ValidationError('Every account already has the primary workgroup') + const outputLocation = text(values.outputLocation) + if (outputLocation && !S3_URI.test(outputLocation)) throw new ValidationError('outputLocation must be an S3 URI such as s3://my-bucket/athena-results/') + const description = text(values.description) + + await this.athena.send(new CreateWorkGroupCommand({ + Name: name, + ...(description ? {Description: description} : {}), + Configuration: { + ...(outputLocation ? {ResultConfiguration: {OutputLocation: outputLocation.endsWith('/') ? outputLocation : `${outputLocation}/`}} : {}), + EnforceWorkGroupConfiguration: values.enforce === 'true', + }, + })) + const created = await this.get(name) + if (!created) throw new RuntimeError(`Athena did not return the new workgroup ${name}`) + return created + } + + async delete(id: string): Promise { + if (id === PRIMARY_WORKGROUP) throw new ValidationError('The primary workgroup cannot be deleted') + await this.athena.send(new DeleteWorkGroupCommand({WorkGroup: id, RecursiveDeleteOption: true})) + } + + async startQuery(workGroup: string, input: QueryInput): Promise { + const sql = typeof input.sql === 'string' ? input.sql.trim() : '' + if (!sql) throw new ValidationError('Enter a query to run') + if (sql.length > MAX_QUERY_LENGTH) throw new ValidationError(`A query can be at most ${MAX_QUERY_LENGTH} characters`) + const database = text(input.database) + + const started = await this.athena.send(new StartQueryExecutionCommand({ + QueryString: sql, + WorkGroup: workGroup, + ...(database ? {QueryExecutionContext: {Catalog: DATA_CATALOG, Database: database}} : {}), + })) + if (!started.QueryExecutionId) throw new RuntimeError('Athena did not return a query id') + return (await this.execution(workGroup, started.QueryExecutionId)) + } + + async getQueryResult(workGroup: string, queryId: string): Promise { + const execution = await this.execution(workGroup, queryId) + if (execution.state !== 'SUCCEEDED' || !execution.outputLocation) return {execution, columns: [], rows: [], truncated: false} + + const pages: GetQueryResultsCommandOutput[] = [] + let next: string | undefined + let rowCount = 0 + do { + const page = await this.athena.send(new GetQueryResultsCommand({QueryExecutionId: queryId, MaxResults: 1000, ...(next ? {NextToken: next} : {})})) + pages.push(page) + rowCount += page.ResultSet?.Rows?.length ?? 0 + next = page.NextToken + } while (next && rowCount <= QUERY_RESULT_ROW_LIMIT) + + const columnInfo = pages[0]?.ResultSet?.ResultSetMetadata?.ColumnInfo ?? [] + const raw = pages.flatMap((page) => page.ResultSet?.Rows ?? []).map((row) => (row.Data ?? []).map((cell) => cell.VarCharValue ?? null)) + const header = raw[0] ?? [] + const columns = columnInfo.length > 0 + ? columnInfo.map((column) => ({name: column.Label ?? column.Name ?? '', type: column.Type ?? null})) + : header.map((name) => ({name: name ?? '', type: null})) + // Athena repeats the column names as the first row of a SELECT's results. + const body = raw.length > 0 && header.length === columns.length && header.every((name, index) => name === columns[index].name) ? raw.slice(1) : raw + return {execution, columns, rows: body.slice(0, QUERY_RESULT_ROW_LIMIT), truncated: body.length > QUERY_RESULT_ROW_LIMIT || Boolean(next)} + } + + async stopQuery(workGroup: string, queryId: string): Promise { + await this.execution(workGroup, queryId) + await this.athena.send(new StopQueryExecutionCommand({QueryExecutionId: queryId})) + } + + async listQueries(workGroup: string): Promise { + const ids = (await this.athena.send(new ListQueryExecutionsCommand({WorkGroup: workGroup, MaxResults: QUERY_HISTORY_LIMIT}))).QueryExecutionIds ?? [] + const executions = await Promise.all(ids.slice(0, QUERY_HISTORY_LIMIT).map(async (id) => { + const res = await this.athena.send(new GetQueryExecutionCommand({QueryExecutionId: id})) + return res.QueryExecution ? toExecution(res.QueryExecution) : null + })) + return executions + .filter((execution): execution is QueryExecutionSummary => execution !== null && (execution.workGroup ?? PRIMARY_WORKGROUP) === workGroup) + .sort((a, b) => (b.submittedAt ?? '').localeCompare(a.submittedAt ?? '')) + } + + async describeQueryCatalog(database?: string): Promise { + const databases: string[] = [] + let next: string | undefined + do { + const res = await this.athena.send(new ListDatabasesCommand({CatalogName: DATA_CATALOG, MaxResults: 50, ...(next ? {NextToken: next} : {})})) + databases.push(...(res.DatabaseList ?? []).map((entry) => entry.Name ?? '').filter(Boolean)) + next = res.NextToken + } while (next) + databases.sort() + + const chosen = database && databases.includes(database) ? database : databases[0] ?? null + if (!chosen) return {databases, database: null, tables: []} + + const tables: TableMetadata[] = [] + let tableToken: string | undefined + do { + const res: ListTableMetadataCommandOutput = await this.athena.send(new ListTableMetadataCommand({CatalogName: DATA_CATALOG, DatabaseName: chosen, MaxResults: 50, ...(tableToken ? {NextToken: tableToken} : {})})) + tables.push(...(res.TableMetadataList ?? [])) + tableToken = res.NextToken + } while (tableToken) + return { + databases, + database: chosen, + tables: tables.map((table) => ({ + name: table.Name ?? '', + columns: [...(table.Columns ?? []), ...(table.PartitionKeys ?? [])].map((column) => ({name: column.Name ?? '', type: column.Type ?? ''})), + })).sort((a, b) => a.name.localeCompare(b.name)), + } + } + + /** A query, refused unless it ran in this workgroup, so one workgroup's page never shows another's. */ + private async execution(workGroup: string, queryId: string): Promise { + let execution: QueryExecution | undefined + try { + execution = (await this.athena.send(new GetQueryExecutionCommand({QueryExecutionId: queryId}))).QueryExecution + } catch (error) { + if (errorName(error) === 'InvalidRequestException') throw new NotFoundError(`Query ${queryId} not found`) + throw error + } + if (!execution || (execution.WorkGroup ?? PRIMARY_WORKGROUP) !== workGroup) throw new NotFoundError(`Query ${queryId} is not in the ${workGroup} workgroup`) + return toExecution(execution) + } +} + +function toResource(workGroup: WorkGroup | WorkGroupSummary): CloudResource { + const name = workGroup.Name ?? '' + const detail = 'Configuration' in workGroup ? workGroup : undefined + const configuration = detail?.Configuration + return { + id: name, + name, + cloud: 'aws', + service: 'query', + type: 'athenaWorkGroup', + region: null, + createdAt: workGroup.CreationTime ? workGroup.CreationTime.toISOString() : null, + status: workGroup.State ?? null, + metadata: { + description: workGroup.Description || null, + ...(detail ? { + outputLocation: configuration?.ResultConfiguration?.OutputLocation ?? null, + enforceConfiguration: configuration?.EnforceWorkGroupConfiguration === true, + engineVersion: configuration?.EngineVersion?.EffectiveEngineVersion ?? configuration?.EngineVersion?.SelectedEngineVersion ?? null, + } : {}), + }, + } +} + +function toExecution(execution: QueryExecution): QueryExecutionSummary { + const status = execution.Status + return { + id: execution.QueryExecutionId ?? '', + sql: execution.Query ?? '', + database: execution.QueryExecutionContext?.Database ?? null, + workGroup: execution.WorkGroup ?? null, + state: status?.State ?? 'QUEUED', + stateReason: status?.StateChangeReason ?? status?.AthenaError?.ErrorMessage ?? null, + statementType: execution.StatementType ?? null, + submittedAt: status?.SubmissionDateTime ? status.SubmissionDateTime.toISOString() : null, + completedAt: status?.CompletionDateTime ? status.CompletionDateTime.toISOString() : null, + runtimeMs: execution.Statistics?.TotalExecutionTimeInMillis ?? execution.Statistics?.EngineExecutionTimeInMillis ?? null, + dataScannedBytes: execution.Statistics?.DataScannedInBytes ?? null, + outputLocation: execution.ResultConfiguration?.OutputLocation ?? null, + } +} + +function text(value: unknown): string { + return typeof value === 'string' ? value.trim() : '' +} + +function errorName(error: unknown): string | undefined { + return typeof error === 'object' && error !== null ? (error as {name?: string}).name : undefined +} diff --git a/packages/api/src/aws.ts b/packages/api/src/aws.ts index 57818dc..6e9ba28 100644 --- a/packages/api/src/aws.ts +++ b/packages/api/src/aws.ts @@ -39,6 +39,7 @@ import { ACMClient } from "@aws-sdk/client-acm"; import { PipesClient } from "@aws-sdk/client-pipes"; import { EFSClient } from "@aws-sdk/client-efs"; import { GlueClient } from "@aws-sdk/client-glue"; +import { AthenaClient } from "@aws-sdk/client-athena"; const endpoint = process.env.FLOCI_ENDPOINT; const region = process.env.AWS_REGION || "us-east-1"; @@ -109,6 +110,7 @@ export type AwsClients = { pipes: PipesClient; efs: EFSClient; glue: GlueClient; + athena: AthenaClient; }; export type AwsClientName = keyof AwsClients; @@ -163,6 +165,7 @@ function buildClients(accountId: string): AwsClients { pipes: new PipesClient(base), efs: new EFSClient(base), glue: new GlueClient(base), + athena: new AthenaClient(base), }; } diff --git a/packages/api/src/cloud-spi/querySchema.ts b/packages/api/src/cloud-spi/querySchema.ts new file mode 100644 index 0000000..f9071d0 --- /dev/null +++ b/packages/api/src/cloud-spi/querySchema.ts @@ -0,0 +1,80 @@ +import type {CloudProvider, FieldSchema, ServiceSchema, TableColumnSchema} from './types' + +/** CreateWorkGroup's name rule. */ +export const WORKGROUP_NAME_PATTERN = '^[a-zA-Z0-9._-]{1,128}$' +/** The workgroup every account has, which Athena does not let you delete. */ +export const PRIMARY_WORKGROUP = 'primary' +/** Rows the editor shows for one query; the full result stays in the CSV file in S3. */ +export const QUERY_RESULT_ROW_LIMIT = 1000 +/** Recent queries the history shows. */ +export const QUERY_HISTORY_LIMIT = 25 + +const workGroupColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'status', label: 'State', format: 'badge'}, + {name: 'description', label: 'Description', path: 'metadata.description', emptyText: '—'}, + {name: 'createdAt', label: 'Created', format: 'datetime', emptyText: '—'}, +] + +const workGroupFilters: FieldSchema[] = [{name: 'search', label: 'Search', type: 'text', required: false}] + +export function awsQuerySchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'query', + displayName: 'Athena workgroups', + fields: [ + { + name: 'name', + label: 'Name', + type: 'text', + required: true, + validation: {pattern: WORKGROUP_NAME_PATTERN, message: 'Use up to 128 letters, numbers, dots, dashes and underscores'}, + }, + {name: 'description', label: 'Description', type: 'text', required: false, validation: {maxLength: 1024}}, + { + name: 'outputLocation', + label: 'Query result location', + type: 'text', + required: false, + span: true, + description: 'Where each query writes its results as a CSV file, such as s3://my-bucket/athena-results/. Leave empty to use TheCloud\'s default results bucket.', + validation: {pattern: '^s3://[a-z0-9.-]{3,63}(/.*)?$', message: 'Use an S3 URI such as s3://my-bucket/athena-results/'}, + }, + { + name: 'enforce', + label: 'Override client-side settings', + type: 'select', + required: false, + defaultValue: 'false', + description: 'When on, queries always use this workgroup\'s result location, even if a client asks for another.', + options: [ + {label: 'Off', value: 'false'}, + {label: 'On', value: 'true'}, + ], + }, + ], + actions: ['list', 'create', 'delete', 'inspect'], + capabilities: { + resourceActions: [ + {name: 'list', label: 'List workgroups', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create workgroup', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete workgroup', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect workgroup', enabled: true, status: 'available', runtimeRequired: true}, + ], + queryActions: [ + {name: 'run', label: 'Run query', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'results', label: 'Get query results', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'stop', label: 'Cancel query', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'history', label: 'Recent queries', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'catalog', label: 'Browse databases and tables', enabled: true, status: 'available', runtimeRequired: true}, + ], + }, + filters: workGroupFilters, + columns: workGroupColumns, + } +} + +export function querySchemaFor(cloud: CloudProvider): ServiceSchema | null { + return cloud === 'aws' ? awsQuerySchema() : null +} diff --git a/packages/api/src/cloud-spi/serviceCatalog.ts b/packages/api/src/cloud-spi/serviceCatalog.ts index fe35516..1e0eeda 100644 --- a/packages/api/src/cloud-spi/serviceCatalog.ts +++ b/packages/api/src/cloud-spi/serviceCatalog.ts @@ -229,6 +229,13 @@ export const SERVICE_CATALOG = { group: 'Analytics', order: 20, }, + query: { + displayName: 'Query', + displayNameByCloud: {aws: 'Athena'}, + iconKey: 'query', + group: 'Analytics', + order: 21, + }, identity: {displayName: 'Identity', iconKey: 'iam', group: 'Security', order: 5}, apigateway: {displayName: 'API Gateway', iconKey: 'apigateway', group: 'Integration', order: 10}, email: { diff --git a/packages/api/src/cloud-spi/types.ts b/packages/api/src/cloud-spi/types.ts index 8713395..d8188c7 100644 --- a/packages/api/src/cloud-spi/types.ts +++ b/packages/api/src/cloud-spi/types.ts @@ -188,6 +188,7 @@ export type UserPoolActionName = export type DistributionActionName = 'setEnabled' | 'createInvalidation' export type AutoScalingActionName = 'setCapacity' | 'terminateInstance' export type CertificateActionName = 'download' | 'export' | 'revoke' +export type QueryActionName = 'run' | 'results' | 'stop' | 'history' | 'catalog' export type DataCatalogActionName = 'createTable' | 'deleteTable' export type FileSystemActionName = 'createMountTarget' | 'deleteMountTarget' | 'createAccessPoint' | 'deleteAccessPoint' | 'attachFunction' | 'detachFunction' export type MonitoringActionName = 'listMetrics' | 'getMetricData' | 'alarmHistory' | 'setAlarmState' | 'setAlarmActions' @@ -253,6 +254,7 @@ export interface ServiceSchema { certificateActions?: CapabilitySchema[] fileSystemActions?: CapabilitySchema[] dataCatalogActions?: CapabilitySchema[] + queryActions?: CapabilitySchema[] monitoringActions?: CapabilitySchema[] /** API Gateway v2 (HTTP and WebSocket APIs): what the routes, stages and security panel can do. */ httpApiActions?: CapabilitySchema[] @@ -560,6 +562,43 @@ export interface AccessPointInput { permissions: string } +export interface QueryInput { + sql: string + /** The Glue database unqualified table names resolve in. */ + database?: string +} + +export interface QueryExecutionSummary { + id: string + sql: string + database: string | null + workGroup: string | null + /** QUEUED, RUNNING, SUCCEEDED, FAILED or CANCELLED. */ + state: string + stateReason: string | null + statementType: string | null + submittedAt: string | null + completedAt: string | null + runtimeMs: number | null + dataScannedBytes: number | null + /** The S3 object holding the full result as CSV. */ + outputLocation: string | null +} + +export interface QueryResult { + execution: QueryExecutionSummary + columns: Array<{name: string; type: string | null}> + rows: Array> + /** More rows exist than the console returns; the full result is in the output file. */ + truncated: boolean +} + +export interface QueryCatalog { + databases: string[] + database: string | null + tables: Array<{name: string; columns: Array<{name: string; type: string}>}> +} + export interface CatalogColumnInput { name: string type: string @@ -1255,6 +1294,13 @@ export interface CloudServiceAdapter { /** With `replace`, the group launches a new instance in its place; otherwise desired capacity drops by one. */ terminateAutoScalingInstance?(groupName: string, instanceId: string, replace: boolean): Promise getCertificatePem?(certificateId: string): Promise + startQuery?(workGroup: string, input: QueryInput): Promise + /** The query's state, and its first rows once it has succeeded. */ + getQueryResult?(workGroup: string, queryId: string): Promise + stopQuery?(workGroup: string, queryId: string): Promise + listQueries?(workGroup: string): Promise + /** The databases a query can use and, for the one named, its tables. */ + describeQueryCatalog?(database?: string): Promise createCatalogTable?(databaseName: string, input: CatalogTableInput): Promise deleteCatalogTable?(databaseName: string, tableName: string): Promise createMountTarget?(fileSystemId: string, subnetId: string): Promise diff --git a/packages/api/src/cloudProxy.test.ts b/packages/api/src/cloudProxy.test.ts index 616d0ac..9331415 100644 --- a/packages/api/src/cloudProxy.test.ts +++ b/packages/api/src/cloudProxy.test.ts @@ -2,7 +2,7 @@ import {describe, expect, test} from 'bun:test' import {createCloudAdapterRegistry} from './cloudProxy' import {checkChildCapabilities} from './cloud-spi/childCapabilities' import {isServiceType} from './cloud-spi/serviceCatalog' -import type {AutoScalingActionName, CloudProvider, CloudServiceAdapter, CloudServiceType, CertificateActionName, DatabaseActionName, DataCatalogActionName, DistributionActionName, FileSystemActionName, HttpApiActionName, KubernetesActionName, MonitoringActionName, ResourceActionName, UserPoolActionName} from './cloud-spi/types' +import type {AutoScalingActionName, CloudProvider, CloudServiceAdapter, CloudServiceType, CertificateActionName, DatabaseActionName, DataCatalogActionName, DistributionActionName, FileSystemActionName, HttpApiActionName, QueryActionName, KubernetesActionName, MonitoringActionName, ResourceActionName, UserPoolActionName} from './cloud-spi/types' /** * Guards the schema-to-implementation contract for every registered adapter. @@ -81,6 +81,14 @@ const METHOD_FOR_CERTIFICATE_ACTION: Record = { + run: 'startQuery', + results: 'getQueryResult', + stop: 'stopQuery', + history: 'listQueries', + catalog: 'describeQueryCatalog', +} + const METHOD_FOR_DATA_CATALOG_ACTION: Record = { createTable: 'createCatalogTable', deleteTable: 'deleteCatalogTable', @@ -231,6 +239,15 @@ describe('registered adapters honour their schema', () => { ).toBe('function') } + const queryClaims = (adapter.schema().capabilities?.queryActions ?? []).filter((capability) => capability.status === 'available') + for (const capability of queryClaims) { + const method = METHOD_FOR_QUERY_ACTION[capability.name] + expect( + typeof adapter[method], + `${label} advertises ${capability.name} as available but has no ${String(method)}()`, + ).toBe('function') + } + const dataCatalogClaims = (adapter.schema().capabilities?.dataCatalogActions ?? []).filter((capability) => capability.status === 'available') for (const capability of dataCatalogClaims) { const method = METHOD_FOR_DATA_CATALOG_ACTION[capability.name] @@ -274,6 +291,7 @@ describe('registered adapters honour their schema', () => { ...(adapter.schema().capabilities?.autoScalingActions ?? []), ...(adapter.schema().capabilities?.fileSystemActions ?? []), ...(adapter.schema().capabilities?.dataCatalogActions ?? []), + ...(adapter.schema().capabilities?.queryActions ?? []), ...(adapter.schema().capabilities?.certificateActions ?? []), ...(adapter.schema().capabilities?.monitoringActions ?? []), ...(adapter.schema().capabilities?.httpApiActions ?? []), diff --git a/packages/api/src/cloudProxy.ts b/packages/api/src/cloudProxy.ts index 602bf46..dd5b44d 100644 --- a/packages/api/src/cloudProxy.ts +++ b/packages/api/src/cloudProxy.ts @@ -55,6 +55,7 @@ import {AwsCertificatesAdapter} from './adapter-aws/AwsCertificatesAdapter' import {AwsPipesAdapter} from './adapter-aws/AwsPipesAdapter' import {AwsFileSystemsAdapter} from './adapter-aws/AwsFileSystemsAdapter' import {AwsDataCatalogAdapter} from './adapter-aws/AwsDataCatalogAdapter' +import {AwsQueryAdapter} from './adapter-aws/AwsQueryAdapter' import {AwsMonitoringAdapter} from './adapter-aws/AwsMonitoringAdapter' import {awsClientsForAccount, resolveAccountId} from './aws' import {createEc2Service} from './services/ec2' @@ -113,6 +114,7 @@ export function createCloudAdapterRegistry(accountId?: string | null): CloudAdap new AwsPipesAdapter(clients.pipes), new AwsFileSystemsAdapter(clients.efs, clients.lambda, clients.ec2), new AwsDataCatalogAdapter(clients.glue), + new AwsQueryAdapter(clients.athena), new AwsMonitoringAdapter(clients.cloudwatch), new AzureStorageAdapter(), new AzureServiceBusAdapter(), diff --git a/packages/api/src/routes/clouds.test.ts b/packages/api/src/routes/clouds.test.ts index 5b05d40..919ce3d 100644 --- a/packages/api/src/routes/clouds.test.ts +++ b/packages/api/src/routes/clouds.test.ts @@ -21,6 +21,7 @@ import {awsCdnSchema} from '../cloud-spi/cdnSchema' import {awsAutoScalingSchema} from '../cloud-spi/autoScalingSchema' import {awsFileSystemsSchema} from '../cloud-spi/fileSystemsSchema' import {awsDataCatalogSchema} from '../cloud-spi/dataCatalogSchema' +import {awsQuerySchema} from '../cloud-spi/querySchema' import {awsCertificatesSchema} from '../cloud-spi/certificatesSchema' import {awsMonitoringSchema} from '../cloud-spi/monitoringSchema' import {awsAppConfigSchema} from '../cloud-spi/appConfigSchema' @@ -877,6 +878,28 @@ describe('cloud schema routes', () => { expect(calls).toEqual([['capacity', 'web-asg', {desiredCapacity: 3}], ['terminate', 'web-asg', 'i-abc', true]]) }) + test('runs, reads, lists and cancels queries through the Athena adapter', async () => { + const calls: unknown[] = [] + const execution = {id: 'q1', sql: 'SELECT 1', database: 'shop', workGroup: 'analysts', state: 'SUCCEEDED', stateReason: null, statementType: 'DML', submittedAt: null, completedAt: null, runtimeMs: 5, dataScannedBytes: 0, outputLocation: 's3://r/q1.csv'} + const app = appWithRoutes([mockAdapter('aws', { + service: 'query', + schema: awsQuerySchema, + startQuery: async (id, input) => { calls.push(['start', id, input.sql, input.database]); return execution }, + getQueryResult: async (id, queryId) => { calls.push(['result', id, queryId]); return {execution, columns: [{name: '_col0', type: 'integer'}], rows: [['1']], truncated: false} }, + stopQuery: async (id, queryId) => { calls.push(['stop', id, queryId]) }, + listQueries: async (id) => { calls.push(['list', id]); return [execution] }, + describeQueryCatalog: async (database) => { calls.push(['catalog', database]); return {databases: ['shop'], database: 'shop', tables: []} }, + })]) + const base = '/api/clouds/aws/services/query/resources/analysts/queries' + + expect((await app.request(base, {method: 'POST', headers: {'content-type': 'application/json'}, body: JSON.stringify({sql: 'SELECT 1', database: 'shop'})})).status).toBe(201) + expect((await (await app.request(`${base}/q1`)).json()).rows).toEqual([['1']]) + expect((await app.request(base)).status).toBe(200) + expect((await app.request(`${base}/q1/stop`, {method: 'POST'})).status).toBe(200) + expect((await (await app.request('/api/clouds/aws/services/query/catalog?database=shop')).json()).databases).toEqual(['shop']) + expect(calls).toEqual([['start', 'analysts', 'SELECT 1', 'shop'], ['result', 'analysts', 'q1'], ['list', 'analysts'], ['stop', 'analysts', 'q1'], ['catalog', 'shop']]) + }) + test('creates and deletes catalog tables through the Glue adapter', async () => { const calls: unknown[] = [] const database = {id: 'shop', name: 'shop', cloud: 'aws' as const, service: 'datacatalog' as const, type: 'glueDatabase', region: null, createdAt: null, metadata: {}} diff --git a/packages/api/src/routes/clouds.ts b/packages/api/src/routes/clouds.ts index 4f48fbe..ecd484b 100644 --- a/packages/api/src/routes/clouds.ts +++ b/packages/api/src/routes/clouds.ts @@ -5,6 +5,7 @@ import type { AutoScalingCapacityInput, AccessPointInput, CatalogTableInput, + QueryInput, FileSystemAttachmentInput, CertificateRevocationReason, CloudProvider, @@ -557,6 +558,44 @@ export function createCloudRoutes(injectedService?: CloudProxyService) { }) }) + app.get('/:cloud/services/query/catalog', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).describeQueryCatalog(cloud, c.req.query('database') || undefined))) + }) + + app.get('/:cloud/services/query/resources/:id/queries', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).listQueries(cloud, c.req.param('id')))) + }) + + app.post('/:cloud/services/query/resources/:id/queries', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).startQuery(cloud, c.req.param('id'), await c.req.json()), 201)) + }) + + app.get('/:cloud/services/query/resources/:id/queries/:queryId', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => c.json(await svc(c).getQueryResult(cloud, c.req.param('id'), c.req.param('queryId')))) + }) + + app.post('/:cloud/services/query/resources/:id/queries/:queryId/stop', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => { + await svc(c).stopQuery(cloud, c.req.param('id'), c.req.param('queryId')) + return c.json({ok: true}) + }) + }) + app.post('/:cloud/services/datacatalog/resources/:id/tables', async (c) => { const cloud = c.req.param('cloud') as CloudProvider if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) diff --git a/packages/api/src/service/CloudProxyService.ts b/packages/api/src/service/CloudProxyService.ts index 45ed341..c3b3bd3 100644 --- a/packages/api/src/service/CloudProxyService.ts +++ b/packages/api/src/service/CloudProxyService.ts @@ -96,6 +96,10 @@ import type { AutoScalingCapacityInput, AccessPointInput, CatalogTableInput, + QueryCatalog, + QueryExecutionSummary, + QueryInput, + QueryResult, FileSystemAttachmentInput, CertificatePem, CertificateRevocationReason, @@ -967,6 +971,36 @@ export class CloudProxyService { return adapter.revokeCertificate(certificateId, reason) } + async startQuery(cloud: CloudProvider, workGroup: string, input: QueryInput): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.startQuery) throw new NotSupportedError(`Running queries is not supported for ${cloud}/query`) + return adapter.startQuery(workGroup, input) + } + + async getQueryResult(cloud: CloudProvider, workGroup: string, queryId: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.getQueryResult) throw new NotSupportedError(`Query results are not supported for ${cloud}/query`) + return adapter.getQueryResult(workGroup, queryId) + } + + async stopQuery(cloud: CloudProvider, workGroup: string, queryId: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.stopQuery) throw new NotSupportedError(`Cancelling queries is not supported for ${cloud}/query`) + return adapter.stopQuery(workGroup, queryId) + } + + async listQueries(cloud: CloudProvider, workGroup: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.listQueries) throw new NotSupportedError(`Query history is not supported for ${cloud}/query`) + return adapter.listQueries(workGroup) + } + + async describeQueryCatalog(cloud: CloudProvider, database?: string): Promise { + const adapter = this.requireAdapter(cloud, 'query') + if (!adapter.describeQueryCatalog) throw new NotSupportedError(`Browsing the catalog is not supported for ${cloud}/query`) + return adapter.describeQueryCatalog(database) + } + async createCatalogTable(cloud: CloudProvider, databaseName: string, input: CatalogTableInput): Promise { const adapter = this.requireAdapter(cloud, 'datacatalog') if (!adapter.createCatalogTable) throw new NotSupportedError(`Creating tables is not supported for ${cloud}/datacatalog`) diff --git a/packages/frontend/src/api/api.ts b/packages/frontend/src/api/api.ts index df5cc14..3ced107 100644 --- a/packages/frontend/src/api/api.ts +++ b/packages/frontend/src/api/api.ts @@ -155,6 +155,13 @@ export const apiEndpointKeys = { export: "clouds.services.certificates.export", revoke: "clouds.services.certificates.revoke", }, + query: { + catalog: "clouds.services.query.catalog", + queries: "clouds.services.query.queries", + start: "clouds.services.query.start", + result: "clouds.services.query.result", + stop: "clouds.services.query.stop", + }, datacatalog: { tables: "clouds.services.datacatalog.tables", table: "clouds.services.datacatalog.table", @@ -968,6 +975,41 @@ export const endpointRegistry: EndpointRegistry = new Map([ method: "POST", }, ], + [ + apiEndpointKeys.clouds.query.catalog, + { + path: "/clouds/:cloud/services/query/catalog", + method: "GET", + }, + ], + [ + apiEndpointKeys.clouds.query.queries, + { + path: "/clouds/:cloud/services/query/resources/:id/queries", + method: "GET", + }, + ], + [ + apiEndpointKeys.clouds.query.start, + { + path: "/clouds/:cloud/services/query/resources/:id/queries", + method: "POST", + }, + ], + [ + apiEndpointKeys.clouds.query.result, + { + path: "/clouds/:cloud/services/query/resources/:id/queries/:queryId", + method: "GET", + }, + ], + [ + apiEndpointKeys.clouds.query.stop, + { + path: "/clouds/:cloud/services/query/resources/:id/queries/:queryId/stop", + method: "POST", + }, + ], [ apiEndpointKeys.clouds.datacatalog.tables, { diff --git a/packages/frontend/src/api/cloudProxyClient.ts b/packages/frontend/src/api/cloudProxyClient.ts index 397824a..c3afda3 100644 --- a/packages/frontend/src/api/cloudProxyClient.ts +++ b/packages/frontend/src/api/cloudProxyClient.ts @@ -36,6 +36,10 @@ import type { AutoScalingCapacityInput, AccessPointInput, CatalogTableInput, + QueryCatalog, + QueryExecutionSummary, + QueryInput, + QueryResult, FileSystemAttachmentInput, CertificatePem, CertificateRevocationReason, @@ -1724,6 +1728,71 @@ export async function revokeCertificate( return res.data; } +export async function describeQueryCatalog( + cloud: CloudProvider, + database?: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.catalog, + requestOptions(cloud, "query", { signal, params: { database } }), + { cloud }, + ); + return res.data; +} + +export async function listQueries( + cloud: CloudProvider, + workGroup: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.queries, + requestOptions(cloud, "query", { signal }), + { cloud, id: workGroup }, + ); + return res.data; +} + +export async function startQuery( + cloud: CloudProvider, + workGroup: string, + input: QueryInput, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.start, + requestOptions(cloud, "query", { body: input }), + { cloud, id: workGroup }, + ); + return res.data; +} + +export async function getQueryResult( + cloud: CloudProvider, + workGroup: string, + queryId: string, + signal?: AbortSignal, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.query.result, + requestOptions(cloud, "query", { signal }), + { cloud, id: workGroup, queryId }, + ); + return res.data; +} + +export async function stopQuery( + cloud: CloudProvider, + workGroup: string, + queryId: string, +): Promise { + await apiClient.call( + apiEndpointKeys.clouds.query.stop, + requestOptions(cloud, "query"), + { cloud, id: workGroup, queryId }, + ); +} + export async function createCatalogTable( cloud: CloudProvider, databaseName: string, diff --git a/packages/frontend/src/components/DataCatalogPanel.test.tsx b/packages/frontend/src/components/DataCatalogPanel.test.tsx index 0d197be..7cfa915 100644 --- a/packages/frontend/src/components/DataCatalogPanel.test.tsx +++ b/packages/frontend/src/components/DataCatalogPanel.test.tsx @@ -1,6 +1,7 @@ import {QueryClient, QueryClientProvider} from '@tanstack/react-query' import {render, screen, within} from '@testing-library/react' import userEvent from '@testing-library/user-event' +import {MemoryRouter} from 'react-router-dom' import {beforeEach, expect, test, vi} from 'vitest' import {DataCatalogPanel} from './DataCatalogPanel' import type {CloudResource} from '@/types/resource' @@ -24,7 +25,7 @@ function database(tables: unknown[] = [{ function renderPanel(resource: CloudResource) { api.getCloudResource.mockResolvedValue(resource) const client = new QueryClient({defaultOptions: {queries: {retry: false}}}) - return render() + return render() } beforeEach(() => { @@ -40,6 +41,7 @@ test('lists tables with their format, location and columns', async () => { expect(within(row).getByText('s3://data/orders/')).toBeInTheDocument() expect(within(row).getByText('semicolon-separated, header row')).toBeInTheDocument() expect(within(row).getByText('amount')).toBeInTheDocument() + expect(screen.getByRole('link', {name: 'Query in Athena'})).toHaveAttribute('href', '/cloud-explorer/aws/query') }) test('creates a CSV table from the columns entered', async () => { diff --git a/packages/frontend/src/components/DataCatalogPanel.tsx b/packages/frontend/src/components/DataCatalogPanel.tsx index 5fbc3e0..119684e 100644 --- a/packages/frontend/src/components/DataCatalogPanel.tsx +++ b/packages/frontend/src/components/DataCatalogPanel.tsx @@ -1,6 +1,7 @@ import {Plus, RefreshCw, X} from 'lucide-react' import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' import {useEffect, useState} from 'react' +import {Link} from 'react-router-dom' import {createCatalogTable, deleteCatalogTable, getCloudResource} from '@/api/cloudProxyClient' import {cloudQueryKeys} from '@/api/queries/cloudQueries' import {useAccountId} from '@/lib/accountStore' @@ -90,6 +91,7 @@ export function DataCatalogPanel({resource, runtimeReachable}: Props) {

Glue database

{database.metadata.description ? String(database.metadata.description) : 'Tables here describe files in S3 so Athena can query them.'}
+ {tables.length > 0 && Query in Athena}
diff --git a/packages/frontend/src/components/DynamicResourceView.tsx b/packages/frontend/src/components/DynamicResourceView.tsx index 1c192c2..3e6655f 100644 --- a/packages/frontend/src/components/DynamicResourceView.tsx +++ b/packages/frontend/src/components/DynamicResourceView.tsx @@ -71,6 +71,7 @@ import { CloudFrontPanel } from "@/components/CloudFrontPanel"; import { AutoScalingPanel } from "@/components/AutoScalingPanel"; import { FileSystemPanel } from "@/components/FileSystemPanel"; import { DataCatalogPanel } from "@/components/DataCatalogPanel"; +import { QueryPanel } from "@/components/QueryPanel"; import { CertificatePanel } from "@/components/CertificatePanel"; import { PipePanel } from "@/components/PipePanel"; import { CloudWatchAlarmPanel } from "@/components/CloudWatchAlarmPanel"; @@ -1007,6 +1008,9 @@ export function DynamicResourceView({ {service === "datacatalog" && cloud === "aws" && ( )} + {service === "query" && cloud === "aws" && ( + + )} {service === "cdn" && cloud === "aws" && ( )} diff --git a/packages/frontend/src/components/QueryPanel.test.tsx b/packages/frontend/src/components/QueryPanel.test.tsx new file mode 100644 index 0000000..b650f3e --- /dev/null +++ b/packages/frontend/src/components/QueryPanel.test.tsx @@ -0,0 +1,100 @@ +import {QueryClient, QueryClientProvider} from '@tanstack/react-query' +import {render, screen, within} from '@testing-library/react' +import userEvent from '@testing-library/user-event' +import {MemoryRouter} from 'react-router-dom' +import {beforeEach, expect, test, vi} from 'vitest' +import {QueryPanel} from './QueryPanel' +import type {CloudResource, QueryExecutionSummary} from '@/types/resource' + +const api = vi.hoisted(() => ({ + describeQueryCatalog: vi.fn(), + getCloudResource: vi.fn(), + listQueries: vi.fn(), + startQuery: vi.fn(), + getQueryResult: vi.fn(), + stopQuery: vi.fn(), + storageObjectDownloadUrl: vi.fn((_cloud: string, bucket: string, key: string) => `/download/${bucket}/${key}`), +})) + +vi.mock('@/api/cloudProxyClient', () => api) +vi.mock('@/lib/accountStore', () => ({useAccountId: () => '000000000000'})) + +const workGroup: CloudResource = {id: 'analysts', name: 'analysts', cloud: 'aws', service: 'query', type: 'athenaWorkGroup', region: null, createdAt: null, status: 'ENABLED', metadata: {outputLocation: 's3://results/'}} + +function execution(overrides: Partial = {}): QueryExecutionSummary { + return { + id: 'q1', sql: 'SELECT * FROM orders LIMIT 10', database: 'shop', workGroup: 'analysts', state: 'SUCCEEDED', stateReason: null, statementType: 'DML', + submittedAt: '2026-10-25T21:00:00.000Z', completedAt: null, runtimeMs: 42, dataScannedBytes: 0, outputLocation: 's3://results/q1.csv', ...overrides, + } +} + +function renderPanel() { + const client = new QueryClient({defaultOptions: {queries: {retry: false}}}) + return render() +} + +beforeEach(() => { + vi.clearAllMocks() + api.getCloudResource.mockResolvedValue({...workGroup, metadata: {outputLocation: 's3://results/athena/'}}) + api.describeQueryCatalog.mockResolvedValue({databases: ['logs', 'shop'], database: 'logs', tables: [{name: 'orders', columns: [{name: 'id', type: 'int'}, {name: 'amount', type: 'double'}]}]}) + api.listQueries.mockResolvedValue([]) + api.startQuery.mockResolvedValue(execution({state: 'RUNNING'})) + api.getQueryResult.mockResolvedValue({execution: execution(), columns: [{name: 'id', type: 'integer'}, {name: 'amount', type: 'double'}], rows: [['1', '10.5'], ['2', null]], truncated: false}) +}) + +test('starts from a preview of the first table and lists its columns', async () => { + renderPanel() + + expect(await screen.findByDisplayValue('SELECT * FROM orders LIMIT 10')).toBeInTheDocument() + expect(screen.getByText('amount')).toBeInTheDocument() + expect(screen.getByLabelText('Database')).toHaveValue('logs') + expect(await screen.findByText('s3://results/athena/')).toBeInTheDocument() +}) + +test('runs the query in the workgroup and shows its rows', async () => { + const user = userEvent.setup() + renderPanel() + + await screen.findByDisplayValue('SELECT * FROM orders LIMIT 10') + await user.click(screen.getByRole('button', {name: 'Run'})) + + expect(api.startQuery).toHaveBeenCalledWith('aws', 'analysts', {sql: 'SELECT * FROM orders LIMIT 10', database: 'logs'}) + expect(await screen.findByText('10.5')).toBeInTheDocument() + expect(screen.getByText('NULL')).toBeInTheDocument() + expect(screen.getByText('2 rows')).toBeInTheDocument() + expect(screen.getByRole('link', {name: 'Download CSV'})).toHaveAttribute('href', '/download/results/q1.csv') +}) + +test('shows why a query failed', async () => { + const user = userEvent.setup() + api.getQueryResult.mockResolvedValue({execution: execution({state: 'FAILED', stateReason: 'Table nope does not exist'}), columns: [], rows: [], truncated: false}) + renderPanel() + + const sql = await screen.findByDisplayValue('SELECT * FROM orders LIMIT 10') + await user.clear(sql) + await user.type(sql, 'SELECT * FROM nope') + await user.keyboard('{Control>}{Enter}{/Control}') + + expect(api.startQuery).toHaveBeenCalledWith('aws', 'analysts', {sql: 'SELECT * FROM nope', database: 'logs'}) + expect(await screen.findByText('Table nope does not exist')).toBeInTheDocument() + expect(screen.queryByRole('link', {name: 'Download CSV'})).not.toBeInTheDocument() +}) + +test('opens a query from the history', async () => { + const user = userEvent.setup() + api.listQueries.mockResolvedValue([execution({id: 'q0', sql: 'SELECT count(*) FROM orders', database: 'shop'})]) + renderPanel() + + await user.click(await screen.findByRole('button', {name: 'SELECT count(*) FROM orders'})) + expect(screen.getByLabelText('SQL')).toHaveValue('SELECT count(*) FROM orders') + expect(api.getQueryResult).toHaveBeenCalledWith('aws', 'analysts', 'q0', expect.anything()) +}) + +test('points to the Glue Data Catalog when there is nothing to query', async () => { + api.describeQueryCatalog.mockResolvedValue({databases: [], database: null, tables: []}) + renderPanel() + + const link = await screen.findByRole('link', {name: 'Glue Data Catalog'}) + expect(link).toHaveAttribute('href', '/cloud-explorer/aws/datacatalog') + expect(within(screen.getByRole('form', {name: 'Query'})).getByRole('button', {name: 'Run'})).toBeDisabled() +}) diff --git a/packages/frontend/src/components/QueryPanel.tsx b/packages/frontend/src/components/QueryPanel.tsx new file mode 100644 index 0000000..84a4c81 --- /dev/null +++ b/packages/frontend/src/components/QueryPanel.tsx @@ -0,0 +1,185 @@ +import {Download, Play, RefreshCw, Square} from 'lucide-react' +import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' +import {useEffect, useMemo, useState} from 'react' +import {Link} from 'react-router-dom' +import {describeQueryCatalog, getCloudResource, getQueryResult, listQueries, startQuery, stopQuery, storageObjectDownloadUrl} from '@/api/cloudProxyClient' +import {useAccountId} from '@/lib/accountStore' +import {formatDateTime} from '@/lib/format' +import type {CloudResource, QueryExecutionSummary} from '@/types/resource' + +type Props = { + resource?: CloudResource + runtimeReachable: boolean +} + +/** Athena reports a running query's state; it is re-read this often until it finishes. */ +const RUNNING_POLL_MS = 1000 +const FINISHED = new Set(['SUCCEEDED', 'FAILED', 'CANCELLED']) + +export function QueryPanel({resource, runtimeReachable}: Props) { + const accountId = useAccountId() + const queryClient = useQueryClient() + const workGroup = resource?.id ?? '' + const [database, setDatabase] = useState('') + const [sql, setSql] = useState('') + const [activeQueryId, setActiveQueryId] = useState(null) + const [seededFor, setSeededFor] = useState(null) + const historyKey = useMemo(() => ['athena-history', accountId, workGroup], [accountId, workGroup]) + + const workGroupQuery = useQuery({ + queryKey: ['athena-workgroup', accountId, workGroup], + queryFn: ({signal}) => getCloudResource('aws', 'query', workGroup, signal), + enabled: Boolean(resource) && runtimeReachable, + }) + const catalogQuery = useQuery({ + queryKey: ['athena-catalog', accountId, database], + queryFn: ({signal}) => describeQueryCatalog('aws', database || undefined, signal), + enabled: Boolean(resource) && runtimeReachable, + }) + const historyQuery = useQuery({ + queryKey: historyKey, + queryFn: ({signal}) => listQueries('aws', workGroup, signal), + enabled: Boolean(resource) && runtimeReachable, + }) + const resultQuery = useQuery({ + queryKey: ['athena-result', accountId, workGroup, activeQueryId], + queryFn: ({signal}) => getQueryResult('aws', workGroup, activeQueryId ?? '', signal), + enabled: Boolean(activeQueryId) && runtimeReachable, + refetchInterval: (query) => query.state.data && !FINISHED.has(query.state.data.execution.state) ? RUNNING_POLL_MS : false, + }) + + const catalog = catalogQuery.data + const chosenDatabase = database || catalog?.database || '' + const firstTable = catalog?.tables[0]?.name + useEffect(() => { + setActiveQueryId(null) + setSql('') + }, [workGroup]) + // Start each workgroup from a query that runs as-is, the way the Athena console opens on a table. + useEffect(() => { + if (!firstTable || seededFor === workGroup) return + setSeededFor(workGroup) + setSql((current) => current || `SELECT * FROM ${quote(firstTable)} LIMIT 10`) + }, [firstTable, seededFor, workGroup]) + + const execution = resultQuery.data?.execution + const finished = execution ? FINISHED.has(execution.state) : true + const executionState = execution?.state + useEffect(() => { + if (executionState && FINISHED.has(executionState)) void queryClient.invalidateQueries({queryKey: historyKey}) + }, [executionState, historyKey, queryClient]) + + const run = useMutation({ + mutationFn: () => startQuery('aws', workGroup, {sql, ...(chosenDatabase ? {database: chosenDatabase} : {})}), + onSuccess: (started) => { + setActiveQueryId(started.id) + return queryClient.invalidateQueries({queryKey: historyKey}) + }, + }) + const cancel = useMutation({ + mutationFn: () => stopQuery('aws', workGroup, activeQueryId ?? ''), + onSuccess: () => resultQuery.refetch(), + }) + + if (!resource) { + return

Select a workgroup

Choose a workgroup to run SQL against the tables in the Glue Data Catalog.

+ } + + const outputLocation = (workGroupQuery.data ?? resource).metadata.outputLocation + const disabled = !runtimeReachable || run.isPending + const result = resultQuery.data + const error = run.error ?? cancel.error ?? resultQuery.error ?? catalogQuery.error + const history = historyQuery.data ?? [] + const download = execution?.outputLocation ? s3Location(execution.outputLocation) : null + const submit = () => { if (sql.trim() && !disabled) run.mutate() } + const openHistory = (entry: QueryExecutionSummary) => { + setActiveQueryId(entry.id) + setSql(entry.sql) + if (entry.database) setDatabase(entry.database) + } + + return
+
+

Athena query editor

Workgroup {resource.name}{outputLocation ? <> · results in {String(outputLocation)} : null}
+
+ +
+
+ +
+ + +
+
{ event.preventDefault(); submit() }}> +