From 12b50f48688cdfae0c334bade4d349067bd65cf1 Mon Sep 17 00:00:00 2001 From: Zakaria Date: Sun, 25 Oct 2026 12:49:15 -0400 Subject: [PATCH] feat(api): add EventBridge Pipes to the cloud explorer Create pipes from SQS, Kinesis, DynamoDB streams or MSK to Lambda, SQS, SNS, EventBridge buses or Step Functions, with an optional filter, enrichment and input template; start and stop them, see the flow with links to each resource, and send a test message to an SQS source. Sources and targets the runtime's poller cannot deliver between are refused on create. --- README.md | 1 + packages/api/package.json | 1 + .../src/adapter-aws/AwsPipesAdapter.test.ts | 122 +++++++++ .../api/src/adapter-aws/AwsPipesAdapter.ts | 237 ++++++++++++++++++ packages/api/src/aws.ts | 3 + packages/api/src/cloud-spi/pipesSchema.ts | 151 +++++++++++ packages/api/src/cloud-spi/serviceCatalog.ts | 7 + packages/api/src/cloudProxy.ts | 2 + .../src/components/DynamicResourceView.tsx | 4 + .../src/components/PipePanel.test.tsx | 81 ++++++ .../frontend/src/components/PipePanel.tsx | 135 ++++++++++ .../frontend/src/components/serviceIcons.tsx | 2 + packages/frontend/src/index.css | 9 + pnpm-lock.yaml | 18 ++ 14 files changed, 773 insertions(+) create mode 100644 packages/api/src/adapter-aws/AwsPipesAdapter.test.ts create mode 100644 packages/api/src/adapter-aws/AwsPipesAdapter.ts create mode 100644 packages/api/src/cloud-spi/pipesSchema.ts create mode 100644 packages/frontend/src/components/PipePanel.test.tsx create mode 100644 packages/frontend/src/components/PipePanel.tsx diff --git a/README.md b/README.md index 25cf884..9568f25 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,7 @@ cd packages/api && bun run scripts/service-matrix.ts | Integration | API Gateway | Yes (list, create, delete, inspect) | No | No | | Integration | SNS / Notifications | Yes (list, create, inspect, delete) | No | No | | Integration | EventBridge / Events | Yes (list, create, delete, inspect) | No | No | +| Integration | EventBridge Pipes / Pipes | Yes (list, create, delete, inspect) | No | No | | Integration | SES Mailbox / Email | Yes (list, inspect) | No | No | | Integration | Cloud Scheduler | No | No | Yes (list, create, delete, inspect) | | Integration | Step Functions / Workflows | Yes (list, create, update, delete, inspect) | No | No | diff --git a/packages/api/package.json b/packages/api/package.json index a06db3e..e85a9a6 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -37,6 +37,7 @@ "@aws-sdk/client-kms": "^3.1096.0", "@aws-sdk/client-lambda": "^3.1076.0", "@aws-sdk/client-lightsail": "^3.1076.0", + "@aws-sdk/client-pipes": "^3.1146.0", "@aws-sdk/client-rds": "^3.1076.0", "@aws-sdk/client-rds-data": "^3.1076.0", "@aws-sdk/client-route-53": "^3.1146.0", diff --git a/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts b/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts new file mode 100644 index 0000000..ecf91c0 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsPipesAdapter.test.ts @@ -0,0 +1,122 @@ +import {describe, expect, test} from 'bun:test' +import { + CreatePipeCommand, + DeletePipeCommand, + DescribePipeCommand, + ListPipesCommand, + type PipesClient, + StartPipeCommand, + StopPipeCommand, +} from '@aws-sdk/client-pipes' +import {AwsPipesAdapter, describeArn} from './AwsPipesAdapter' +import {ValidationError} from '../cloud-spi/errors' + +type SendResult = Record + +function stubPipes(handler: (command: object) => SendResult | Promise) { + const sent: object[] = [] + const client = { + async send(command: object) { + sent.push(command) + return handler(command) + }, + } as unknown as PipesClient + return {client, sent, adapter: new AwsPipesAdapter(client)} +} + +const SRC = 'arn:aws:sqs:us-east-1:000000000000:pipe-src' +const DST = 'arn:aws:lambda:us-east-1:000000000000:function:worker' +const ROLE = 'arn:aws:iam::000000000000:role/pipe-role' +const described = { + Name: 'orders-pipe', Arn: 'arn:aws:pipes:us-east-1:000000000000:pipe/orders-pipe', + DesiredState: 'RUNNING', CurrentState: 'RUNNING', Source: SRC, Target: DST, RoleArn: ROLE, + SourceParameters: {FilterCriteria: {Filters: [{Pattern: '{"body":{"kind":["order"]}}'}]}, SqsQueueParameters: {BatchSize: 5}}, + TargetParameters: {InputTemplate: '{"id": <$.messageId>}'}, + CreationTime: new Date('2026-10-25T16:42:45Z'), +} + +describe('AwsPipesAdapter', () => { + test('lists pipes across pages with readable source and target names', async () => { + const {adapter} = stubPipes((command) => { + if (!(command instanceof ListPipesCommand)) return {} + return command.input.NextToken + ? {Pipes: [{Name: 'clicks', CurrentState: 'STOPPED', Source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', Target: 'arn:aws:events:us-east-1:000000000000:event-bus/default'}]} + : {Pipes: [{Name: 'orders-pipe', CurrentState: 'RUNNING', Source: SRC, Target: DST}], NextToken: 'n'} + }) + + const pipes = await adapter.list() + expect(pipes.map((pipe) => [pipe.name, pipe.status, pipe.metadata.sourceName, pipe.metadata.targetName])).toEqual([ + ['orders-pipe', 'RUNNING', 'pipe-src (SQS queue)', 'worker (Lambda function)'], + ['clicks', 'STOPPED', 'clicks (Kinesis stream)', 'default (EventBridge event bus)'], + ]) + expect((await adapter.list({search: 'kinesis'})).map((pipe) => pipe.name)).toEqual(['clicks']) + }) + + test('describes the filter, batch size, template and role', async () => { + const {adapter} = stubPipes(() => described) + expect((await adapter.get('orders-pipe'))?.metadata).toMatchObject({ + filterPatterns: ['{"body":{"kind":["order"]}}'], + batchSize: 5, + inputTemplate: '{"id": <$.messageId>}', + roleArn: ROLE, + }) + }) + + test('a missing pipe is null', async () => { + const {adapter} = stubPipes(() => { throw Object.assign(new Error('gone'), {name: 'NotFoundException'}) }) + expect(await adapter.get('nope')).toBeNull() + }) + + test('creates an SQS pipe with its filter, batch size and template', async () => { + const {adapter, sent} = stubPipes((command) => command instanceof DescribePipeCommand ? described : {}) + + await adapter.create({values: {name: 'orders-pipe', source: SRC, target: DST, roleArn: ROLE, batchSize: '5', filterPattern: '{ "body": { "kind": ["order"] } }', inputTemplate: '{"id": <$.messageId>}'}}) + expect((sent[0] as CreatePipeCommand).input).toEqual({ + Name: 'orders-pipe', + Source: SRC, + SourceParameters: {FilterCriteria: {Filters: [{Pattern: '{"body":{"kind":["order"]}}'}]}, SqsQueueParameters: {BatchSize: 5}}, + Target: DST, + TargetParameters: {InputTemplate: '{"id": <$.messageId>}'}, + RoleArn: ROLE, + DesiredState: 'RUNNING', + }) + }) + + test('gives stream and MSK sources the starting position they require', async () => { + const {adapter, sent} = stubPipes((command) => command instanceof DescribePipeCommand ? described : {}) + + await adapter.create({values: {name: 'clicks', source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', target: DST, roleArn: ROLE, startingPosition: 'TRIM_HORIZON', desiredState: 'STOPPED'}}) + expect((sent[0] as CreatePipeCommand).input).toMatchObject({SourceParameters: {KinesisStreamParameters: {StartingPosition: 'TRIM_HORIZON'}}, DesiredState: 'STOPPED'}) + + await adapter.create({values: {name: 'k', source: 'arn:aws:kafka:us-east-1:000000000000:cluster/events/abc', target: DST, roleArn: ROLE, topicName: 'orders'}}) + expect((sent[2] as CreatePipeCommand).input.SourceParameters).toEqual({ManagedStreamingKafkaParameters: {TopicName: 'orders', StartingPosition: 'LATEST'}}) + }) + + test('refuses pipes the runtime could not run, before calling it', async () => { + const {adapter, sent} = stubPipes(() => ({})) + const base = {name: 'p', source: SRC, target: DST, roleArn: ROLE} + await expect(adapter.create({values: {...base, target: 'arn:aws:s3:::bucket'}})).rejects.toThrow('target must be the ARN of a') + await expect(adapter.create({values: {...base, source: 'arn:aws:sns:us-east-1:000000000000:topic'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, enrichment: 'arn:aws:sqs:us-east-1:000000000000:q'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, filterPattern: '{kind: order}'}})).rejects.toThrow('filterPattern must be JSON') + await expect(adapter.create({values: {...base, batchSize: '20000'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, source: 'arn:aws:kafka:us-east-1:000000000000:cluster/c/1'}})).rejects.toThrow('topicName is required') + await expect(adapter.create({values: {...base, roleArn: 'pipe-role'}})).rejects.toBeInstanceOf(ValidationError) + expect(sent).toHaveLength(0) + }) + + test('starts, stops and deletes by name', async () => { + const {adapter, sent} = stubPipes(() => ({})) + await adapter.stop('orders-pipe') + await adapter.start('orders-pipe') + await adapter.delete('orders-pipe') + expect(sent.map((command) => command.constructor)).toEqual([StopPipeCommand, StartPipeCommand, DeletePipeCommand]) + expect(sent.every((command) => (command as StopPipeCommand).input.Name === 'orders-pipe')).toBe(true) + }) + + test('names every kind of source and target ARN', () => { + expect(describeArn('arn:aws:dynamodb:us-east-1:000000000000:table/orders/stream/2026-10-25T00:00:00.000')).toBe('orders (DynamoDB stream)') + expect(describeArn('arn:aws:states:us-east-1:000000000000:stateMachine:flow')).toBe('flow (Step Functions state machine)') + expect(describeArn('arn:aws:sns:us-east-1:000000000000:alerts')).toBe('alerts (SNS topic)') + }) +}) diff --git a/packages/api/src/adapter-aws/AwsPipesAdapter.ts b/packages/api/src/adapter-aws/AwsPipesAdapter.ts new file mode 100644 index 0000000..4dd23a4 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsPipesAdapter.ts @@ -0,0 +1,237 @@ +import { + CreatePipeCommand, + DeletePipeCommand, + DescribePipeCommand, + type DescribePipeCommandOutput, + ListPipesCommand, + type Pipe, + type PipeSourceParameters, + type PipesClient, + StartPipeCommand, + StopPipeCommand, +} from '@aws-sdk/client-pipes' +import {RuntimeError, ValidationError} from '../cloud-spi/errors' +import { + awsPipesSchema, + PIPE_ENRICHMENT_SERVICES, + PIPE_NAME_PATTERN, + PIPE_SOURCE_SERVICES, + PIPE_STARTING_POSITIONS, + PIPE_TARGET_SERVICES, +} from '../cloud-spi/pipesSchema' +import type {CloudResource, CloudServiceAdapter, CreateResourceInput, ResourceQuery, ServiceSchema} from '../cloud-spi/types' + +const MAX_BATCH_SIZE = 10_000 + +/** + * Talks to EventBridge Pipes, verified against Floci core (community, `latest`). + * + * Notes from probing the runtime: + * - Pipes really run: the runtime polls SQS, Kinesis, DynamoDB streams and MSK, + * applies the filter, enrichment and input template, and calls the target. + * - Its poller delivers only to Lambda, SQS, SNS, EventBridge buses and Step + * Functions, so create refuses other targets instead of making a pipe that + * would never deliver. + * - ListPipes returns no parameters, so the table shows only what it carries; + * the filter, template and role come from DescribePipe. + */ +export class AwsPipesAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'pipes' as const + + constructor(private readonly pipes: PipesClient) {} + + schema(): ServiceSchema { + return awsPipesSchema() + } + + async list(query: ResourceQuery = {}): Promise { + const found: Pipe[] = [] + let next: string | undefined + do { + const res = await this.pipes.send(new ListPipesCommand({Limit: 100, ...(next ? {NextToken: next} : {})})) + found.push(...(res.Pipes ?? [])) + next = res.NextToken + } while (next) + + const search = query.search?.trim().toLowerCase() + return found.map((pipe) => toResource(pipe)).filter((pipe) => !search + || pipe.name.toLowerCase().includes(search) + || String(pipe.metadata.source).toLowerCase().includes(search) + || String(pipe.metadata.target).toLowerCase().includes(search)) + } + + async get(id: string): Promise { + try { + return toResource(await this.pipes.send(new DescribePipeCommand({Name: id}))) + } catch (error) { + if (errorName(error) === 'NotFoundException') return null + throw error + } + } + + async create(input: CreateResourceInput): Promise { + const values = input.values + const name = text(values.name) + if (!new RegExp(PIPE_NAME_PATTERN).test(name)) throw new ValidationError('name must be 1 to 64 letters, numbers, dots, dashes or underscores') + + const source = text(values.source) + const sourceService = supported(source, PIPE_SOURCE_SERVICES, 'source') + const target = text(values.target) + supported(target, PIPE_TARGET_SERVICES, 'target') + const enrichment = text(values.enrichment) + if (enrichment) supported(enrichment, PIPE_ENRICHMENT_SERVICES, 'enrichment') + const roleArn = text(values.roleArn) + if (!/^arn:aws[a-z-]*:iam::\d{12}:role\/.+$/.test(roleArn)) throw new ValidationError('roleArn must be an IAM role ARN') + + const inputTemplate = text(values.inputTemplate) + const description = text(values.description) + const desiredState = values.desiredState === 'STOPPED' ? 'STOPPED' : 'RUNNING' + + await this.pipes.send(new CreatePipeCommand({ + Name: name, + Source: source, + SourceParameters: sourceParameters(sourceService, values), + Target: target, + ...(inputTemplate ? {TargetParameters: {InputTemplate: inputTemplate}} : {}), + ...(enrichment ? {Enrichment: enrichment} : {}), + RoleArn: roleArn, + DesiredState: desiredState, + ...(description ? {Description: description} : {}), + })) + + const created = await this.get(name) + if (!created) throw new RuntimeError(`EventBridge Pipes did not return the new pipe ${name}`) + return created + } + + async delete(id: string): Promise { + await this.pipes.send(new DeletePipeCommand({Name: id})) + } + + async start(id: string): Promise { + await this.pipes.send(new StartPipeCommand({Name: id})) + } + + async stop(id: string): Promise { + await this.pipes.send(new StopPipeCommand({Name: id})) + } +} + +/** Each source kind carries its own parameter block, as CreatePipe expects. */ +function sourceParameters(service: keyof typeof PIPE_SOURCE_SERVICES, values: Record): PipeSourceParameters { + const batchSize = optionalBatchSize(values.batchSize) + const filterPattern = text(values.filterPattern) + const parameters: PipeSourceParameters = filterPattern ? {FilterCriteria: {Filters: [{Pattern: jsonObject(filterPattern, 'filterPattern')}]}} : {} + const batch = batchSize ? {BatchSize: batchSize} : {} + + if (service === 'sqs') return {...parameters, SqsQueueParameters: batch} + + const startingPosition = text(values.startingPosition) || 'LATEST' + if (!(PIPE_STARTING_POSITIONS as readonly string[]).includes(startingPosition)) { + throw new ValidationError(`startingPosition must be one of ${PIPE_STARTING_POSITIONS.join(', ')}`) + } + const position = startingPosition as (typeof PIPE_STARTING_POSITIONS)[number] + if (service === 'kinesis') return {...parameters, KinesisStreamParameters: {StartingPosition: position, ...batch}} + if (service === 'dynamodb') return {...parameters, DynamoDBStreamParameters: {StartingPosition: position, ...batch}} + + const topicName = text(values.topicName) + if (!topicName) throw new ValidationError('topicName is required for an MSK source') + return {...parameters, ManagedStreamingKafkaParameters: {TopicName: topicName, StartingPosition: position, ...batch}} +} + +function toResource(pipe: Pipe | DescribePipeCommandOutput): CloudResource { + const name = pipe.Name ?? '' + const detail = 'SourceParameters' in pipe ? pipe : undefined + const sourceParameters = detail?.SourceParameters + const batchSize = sourceParameters?.SqsQueueParameters?.BatchSize + ?? sourceParameters?.KinesisStreamParameters?.BatchSize + ?? sourceParameters?.DynamoDBStreamParameters?.BatchSize + ?? sourceParameters?.ManagedStreamingKafkaParameters?.BatchSize + return { + id: name, + name, + cloud: 'aws', + service: 'pipes', + type: 'pipe', + region: pipe.Arn?.split(':')[3] || null, + createdAt: pipe.CreationTime ? pipe.CreationTime.toISOString() : null, + status: pipe.CurrentState ?? null, + metadata: { + arn: pipe.Arn, + desiredState: pipe.DesiredState ?? null, + currentState: pipe.CurrentState ?? null, + stateReason: pipe.StateReason ?? null, + source: pipe.Source ?? null, + sourceName: describeArn(pipe.Source), + target: pipe.Target ?? null, + targetName: describeArn(pipe.Target), + enrichment: pipe.Enrichment ?? null, + lastModified: pipe.LastModifiedTime ? pipe.LastModifiedTime.toISOString() : null, + ...(detail ? { + description: detail.Description ?? null, + roleArn: detail.RoleArn ?? null, + filterPatterns: (sourceParameters?.FilterCriteria?.Filters ?? []).map((filter) => filter.Pattern ?? '').filter(Boolean), + batchSize: batchSize ?? null, + startingPosition: sourceParameters?.KinesisStreamParameters?.StartingPosition + ?? sourceParameters?.DynamoDBStreamParameters?.StartingPosition + ?? sourceParameters?.ManagedStreamingKafkaParameters?.StartingPosition + ?? null, + topicName: sourceParameters?.ManagedStreamingKafkaParameters?.TopicName ?? null, + inputTemplate: detail.TargetParameters?.InputTemplate ?? null, + enrichmentInputTemplate: detail.EnrichmentParameters?.InputTemplate ?? null, + } : {}), + }, + } +} + +/** + * A readable name for an ARN: `queue (SQS queue)`. The resource part differs by + * service — `function:name`, `stream/name`, `table/name/stream/…`, `cluster/name/uuid`. + */ +export function describeArn(arn: string | undefined): string | null { + if (!arn) return null + const parts = arn.split(':') + const service = parts[2] ?? '' + const resource = parts.slice(5).join(':') + const name = service === 'lambda' || service === 'states' ? resource.split(':')[1] ?? resource + : service === 'dynamodb' || service === 'kafka' || service === 'kinesis' || service === 'events' ? resource.split('/')[1] ?? resource + : resource + const kind = (PIPE_SOURCE_SERVICES as Record)[service] ?? (PIPE_TARGET_SERVICES as Record)[service] + return kind ? `${name} (${kind})` : name +} + +function supported>(arn: string, services: T, field: string): keyof T { + const service = arn.split(':')[2] ?? '' + if (!arn.startsWith('arn:') || !(service in services)) { + throw new ValidationError(`${field} must be the ARN of a ${Object.values(services).join(', ')}`) + } + return service as keyof T +} + +function optionalBatchSize(value: unknown): number | undefined { + const raw = text(value) + if (!raw) return undefined + const size = Number(raw) + if (!Number.isInteger(size) || size < 1 || size > MAX_BATCH_SIZE) throw new ValidationError(`batchSize must be a whole number from 1 to ${MAX_BATCH_SIZE}`) + return size +} + +function jsonObject(value: string, field: string): string { + let parsed: unknown + try { + parsed = JSON.parse(value) + } catch { + throw new ValidationError(`${field} must be JSON, such as {"body": {"kind": ["order"]}}`) + } + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) throw new ValidationError(`${field} must be a JSON object`) + return JSON.stringify(parsed) +} + +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 187566a..b1468e9 100644 --- a/packages/api/src/aws.ts +++ b/packages/api/src/aws.ts @@ -35,6 +35,7 @@ import { CognitoIdentityProviderClient } from "@aws-sdk/client-cognito-identity- import { Route53Client } from "@aws-sdk/client-route-53"; import { CloudFrontClient } from "@aws-sdk/client-cloudfront"; import { ACMClient } from "@aws-sdk/client-acm"; +import { PipesClient } from "@aws-sdk/client-pipes"; const endpoint = process.env.FLOCI_ENDPOINT; const region = process.env.AWS_REGION || "us-east-1"; @@ -101,6 +102,7 @@ export type AwsClients = { route53: Route53Client; cloudfront: CloudFrontClient; acm: ACMClient; + pipes: PipesClient; }; export type AwsClientName = keyof AwsClients; @@ -151,6 +153,7 @@ function buildClients(accountId: string): AwsClients { route53: new Route53Client(base), cloudfront: new CloudFrontClient(base), acm: new ACMClient(base), + pipes: new PipesClient(base), }; } diff --git a/packages/api/src/cloud-spi/pipesSchema.ts b/packages/api/src/cloud-spi/pipesSchema.ts new file mode 100644 index 0000000..daa7e46 --- /dev/null +++ b/packages/api/src/cloud-spi/pipesSchema.ts @@ -0,0 +1,151 @@ +import type {CloudProvider, FieldSchema, ServiceSchema, TableColumnSchema} from './types' + +/** CreatePipe's name rule. */ +export const PIPE_NAME_PATTERN = '^[\\.\\-_A-Za-z0-9]{1,64}$' +export const PIPE_STARTING_POSITIONS = ['TRIM_HORIZON', 'LATEST'] as const + +/** + * The sources and targets the runtime's pipe poller actually moves events + * between, keyed by the ARN's service field. Anything else is refused on create + * rather than accepted into a pipe that would never deliver. + */ +export const PIPE_SOURCE_SERVICES = {sqs: 'SQS queue', kinesis: 'Kinesis stream', dynamodb: 'DynamoDB stream', kafka: 'MSK topic'} as const +export const PIPE_TARGET_SERVICES = {lambda: 'Lambda function', sqs: 'SQS queue', sns: 'SNS topic', events: 'EventBridge event bus', states: 'Step Functions state machine'} as const +export const PIPE_ENRICHMENT_SERVICES = {lambda: 'Lambda function', states: 'Step Functions state machine'} as const + +const ARN = '^arn:aws[a-z-]*:[a-z0-9-]+:[a-z0-9-]*:\\d{12}:.+$' + +const pipeColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'status', label: 'State', format: 'badge'}, + {name: 'source', label: 'Source', path: 'metadata.sourceName'}, + {name: 'target', label: 'Target', path: 'metadata.targetName'}, + {name: 'createdAt', label: 'Created', format: 'datetime'}, +] + +const pipeFilters: FieldSchema[] = [{name: 'search', label: 'Search', type: 'text', required: false}] + +export function awsPipesSchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'pipes', + displayName: 'EventBridge Pipes', + fields: [ + { + name: 'name', + label: 'Name', + type: 'text', + required: true, + validation: {pattern: PIPE_NAME_PATTERN, message: 'Use up to 64 letters, numbers, dots, dashes and underscores'}, + }, + {name: 'description', label: 'Description', type: 'text', required: false, validation: {maxLength: 512}}, + { + name: 'source', + label: 'Source ARN', + type: 'text', + required: true, + span: true, + description: 'Where events come from: an SQS queue, a Kinesis stream, a DynamoDB stream or an MSK cluster.', + validation: {pattern: ARN, message: 'Use the ARN of a queue or stream'}, + }, + { + name: 'startingPosition', + label: 'Starting position', + type: 'select', + required: false, + defaultValue: 'LATEST', + description: 'For streams and MSK: read new records only, or everything still in the stream.', + options: [ + {label: 'New records only (LATEST)', value: 'LATEST'}, + {label: 'Oldest record (TRIM_HORIZON)', value: 'TRIM_HORIZON'}, + ], + }, + { + name: 'topicName', + label: 'MSK topic', + type: 'text', + required: false, + description: 'Required when the source is an MSK cluster.', + }, + { + name: 'batchSize', + label: 'Batch size', + type: 'text', + required: false, + description: 'Records sent to the target at once. 1 to 10,000; leave empty for the default of 10.', + validation: {pattern: '^[1-9][0-9]{0,4}$', message: 'Use a whole number from 1 to 10000'}, + }, + { + name: 'filterPattern', + label: 'Filter pattern', + type: 'textarea', + required: false, + span: true, + description: 'Optional EventBridge event pattern; only matching records go on. For SQS, match the parsed message under body, for example {"body": {"kind": ["order"]}}.', + }, + { + name: 'enrichment', + label: 'Enrichment ARN', + type: 'text', + required: false, + span: true, + description: 'Optional Lambda function or Step Functions state machine that transforms each batch before the target gets it.', + validation: {pattern: ARN, message: 'Use the ARN of a Lambda function or state machine'}, + }, + { + name: 'target', + label: 'Target ARN', + type: 'text', + required: true, + span: true, + description: 'Where events go: a Lambda function, SQS queue, SNS topic, EventBridge event bus or Step Functions state machine.', + validation: {pattern: ARN, message: 'Use the ARN of the target'}, + }, + { + name: 'inputTemplate', + label: 'Target input template', + type: 'textarea', + required: false, + span: true, + description: 'Optional. Reshape what the target receives, with <$.path> placeholders, for example {"id": <$.messageId>}.', + }, + { + name: 'roleArn', + label: 'Execution role ARN', + type: 'text', + required: true, + span: true, + description: 'The IAM role the pipe uses to read the source and call the target, as AWS requires. Create one in IAM first.', + validation: {pattern: '^arn:aws[a-z-]*:iam::\\d{12}:role/.+$', message: 'Use an IAM role ARN such as arn:aws:iam::000000000000:role/pipe-role'}, + }, + { + name: 'desiredState', + label: 'State after creating', + type: 'select', + required: false, + defaultValue: 'RUNNING', + options: [ + {label: 'Running', value: 'RUNNING'}, + {label: 'Stopped', value: 'STOPPED'}, + ], + }, + ], + actions: ['list', 'create', 'delete', 'inspect'], + capabilities: { + resourceActions: [ + {name: 'list', label: 'List pipes', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'start', label: 'Start pipe', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'stop', label: 'Stop pipe', enabled: true, status: 'available', runtimeRequired: true}, + ], + }, + filters: pipeFilters, + columns: pipeColumns, + } +} + +export function pipesSchemaFor(cloud: CloudProvider): ServiceSchema | null { + return cloud === 'aws' ? awsPipesSchema() : null +} diff --git a/packages/api/src/cloud-spi/serviceCatalog.ts b/packages/api/src/cloud-spi/serviceCatalog.ts index f0ca116..14fea20 100644 --- a/packages/api/src/cloud-spi/serviceCatalog.ts +++ b/packages/api/src/cloud-spi/serviceCatalog.ts @@ -281,6 +281,13 @@ export const SERVICE_CATALOG = { group: 'Provisioning', order: 20, }, + pipes: { + displayName: 'Pipes', + displayNameByCloud: {aws: 'EventBridge Pipes'}, + iconKey: 'pipes', + group: 'Integration', + order: 25, + }, scheduler: { displayName: 'Cloud Scheduler', displayNameByCloud: {gcp: 'Cloud Scheduler'}, diff --git a/packages/api/src/cloudProxy.ts b/packages/api/src/cloudProxy.ts index c1f4d4d..9c616f0 100644 --- a/packages/api/src/cloudProxy.ts +++ b/packages/api/src/cloudProxy.ts @@ -51,6 +51,7 @@ import {AwsUserPoolsAdapter} from './adapter-aws/AwsUserPoolsAdapter' import {AwsDnsAdapter} from './adapter-aws/AwsDnsAdapter' import {AwsCdnAdapter} from './adapter-aws/AwsCdnAdapter' import {AwsCertificatesAdapter} from './adapter-aws/AwsCertificatesAdapter' +import {AwsPipesAdapter} from './adapter-aws/AwsPipesAdapter' import {AwsMonitoringAdapter} from './adapter-aws/AwsMonitoringAdapter' import {awsClientsForAccount, resolveAccountId} from './aws' import {createEc2Service} from './services/ec2' @@ -105,6 +106,7 @@ export function createCloudAdapterRegistry(accountId?: string | null): CloudAdap new AwsDnsAdapter(clients.route53), new AwsCdnAdapter(clients.cloudfront, clients.s3), new AwsCertificatesAdapter(clients.acm, resolveAccountId(accountId)), + new AwsPipesAdapter(clients.pipes), new AwsMonitoringAdapter(clients.cloudwatch), new AzureStorageAdapter(), new AzureServiceBusAdapter(), diff --git a/packages/frontend/src/components/DynamicResourceView.tsx b/packages/frontend/src/components/DynamicResourceView.tsx index 7b34487..d241159 100644 --- a/packages/frontend/src/components/DynamicResourceView.tsx +++ b/packages/frontend/src/components/DynamicResourceView.tsx @@ -69,6 +69,7 @@ import { CognitoUsersPanel } from "@/components/CognitoUsersPanel"; import { Route53RecordsPanel } from "@/components/Route53RecordsPanel"; import { CloudFrontPanel } from "@/components/CloudFrontPanel"; import { CertificatePanel } from "@/components/CertificatePanel"; +import { PipePanel } from "@/components/PipePanel"; import { CloudWatchAlarmPanel } from "@/components/CloudWatchAlarmPanel"; import { CloudWatchMetricsExplorer } from "@/components/CloudWatchMetricsExplorer"; import { WorkflowExecutionDebugger } from "@/components/WorkflowExecutionDebugger"; @@ -991,6 +992,9 @@ export function DynamicResourceView({ {service === "certificates" && cloud === "aws" && ( )} + {service === "pipes" && cloud === "aws" && ( + + )} {service === "cdn" && cloud === "aws" && ( )} diff --git a/packages/frontend/src/components/PipePanel.test.tsx b/packages/frontend/src/components/PipePanel.test.tsx new file mode 100644 index 0000000..db7e622 --- /dev/null +++ b/packages/frontend/src/components/PipePanel.test.tsx @@ -0,0 +1,81 @@ +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 {PipePanel} from './PipePanel' +import type {CloudResource} from '@/types/resource' + +const api = vi.hoisted(() => ({ + getCloudResource: vi.fn(), + startCloudResource: vi.fn(), + stopCloudResource: vi.fn(), + sendSqsMessage: vi.fn(), +})) + +vi.mock('@/api/cloudProxyClient', () => api) +vi.mock('@/lib/accountStore', () => ({useAccountId: () => '000000000000'})) + +function pipe(status = 'RUNNING', metadata: Record = {}): CloudResource { + return { + id: 'orders-pipe', name: 'orders-pipe', cloud: 'aws', service: 'pipes', type: 'pipe', region: 'us-east-1', createdAt: null, status, + metadata: { + desiredState: status, source: 'arn:aws:sqs:us-east-1:000000000000:pipe-src', sourceName: 'pipe-src (SQS queue)', + target: 'arn:aws:lambda:us-east-1:000000000000:function:worker', targetName: 'worker (Lambda function)', + filterPatterns: ['{"body":{"kind":["order"]}}'], batchSize: 5, inputTemplate: '{"id": <$.messageId>}', + roleArn: 'arn:aws:iam::000000000000:role/pipe-role', + ...metadata, + }, + } +} + +function renderPanel(resource: CloudResource) { + api.getCloudResource.mockResolvedValue(resource) + const client = new QueryClient({defaultOptions: {queries: {retry: false}}}) + return render() +} + +beforeEach(() => { + vi.clearAllMocks() + api.startCloudResource.mockResolvedValue(undefined) + api.stopCloudResource.mockResolvedValue(undefined) + api.sendSqsMessage.mockResolvedValue({messageId: 'm-1'}) +}) + +test('shows the flow from source through the filter to the target, linked to each service', async () => { + renderPanel(pipe()) + + const flow = await screen.findByRole('list', {name: 'Event flow'}) + expect(within(flow).getByRole('link', {name: 'pipe-src (SQS queue)'})).toHaveAttribute('href', '/cloud-explorer/aws/messaging') + expect(within(flow).getByRole('link', {name: 'worker (Lambda function)'})).toHaveAttribute('href', '/cloud-explorer/aws/serverless') + expect(within(flow).getByText('{"body":{"kind":["order"]}}')).toBeInTheDocument() + expect(within(flow).getByText('batches of 5')).toBeInTheDocument() + expect(within(flow).getByText('{"id": <$.messageId>}')).toBeInTheDocument() +}) + +test('stops a running pipe', async () => { + const user = userEvent.setup() + renderPanel(pipe()) + + await user.click(await screen.findByRole('button', {name: 'Stop'})) + expect(api.stopCloudResource).toHaveBeenCalledWith('aws', 'pipes', 'orders-pipe') + expect(await screen.findByText(/Pipe stopped/)).toBeInTheDocument() +}) + +test('sends a test message to an SQS source queue', async () => { + const user = userEvent.setup() + renderPanel(pipe()) + + const form = await screen.findByRole('form', {name: 'Send test message'}) + await user.click(within(form).getByRole('button', {name: 'Send to pipe-src'})) + expect(api.sendSqsMessage).toHaveBeenCalledWith('aws', 'pipe-src', {messageBody: '{"kind": "order", "id": 1}'}) + expect(await screen.findByText(/Sent message m-1/)).toBeInTheDocument() +}) + +test('offers no test message for a stream source, and cannot toggle a pipe mid-change', async () => { + renderPanel(pipe('STARTING', {source: 'arn:aws:kinesis:us-east-1:000000000000:stream/clicks', sourceName: 'clicks (Kinesis stream)', desiredState: 'RUNNING'})) + + expect(await screen.findByRole('link', {name: 'clicks (Kinesis stream)'})).toBeInTheDocument() + expect(screen.queryByRole('form', {name: 'Send test message'})).not.toBeInTheDocument() + expect(screen.getByRole('button', {name: 'Start'})).toBeDisabled() +}) diff --git a/packages/frontend/src/components/PipePanel.tsx b/packages/frontend/src/components/PipePanel.tsx new file mode 100644 index 0000000..817f71d --- /dev/null +++ b/packages/frontend/src/components/PipePanel.tsx @@ -0,0 +1,135 @@ +import {ArrowRight, RefreshCw} from 'lucide-react' +import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' +import {type ReactNode, useEffect, useState} from 'react' +import {Link} from 'react-router-dom' +import {getCloudResource, sendSqsMessage, startCloudResource, stopCloudResource} from '@/api/cloudProxyClient' +import {cloudQueryKeys} from '@/api/queries/cloudQueries' +import {useAccountId} from '@/lib/accountStore' +import type {CloudResource} from '@/types/resource' + +type Props = { + resource?: CloudResource + runtimeReachable: boolean +} + +/** Where each ARN's service lives in TheCloud, so a step links to the resource behind it. */ +const CONSOLE_SERVICE: Record = { + sqs: 'messaging', + lambda: 'serverless', + sns: 'notifications', + events: 'events', + states: 'workflows', + kinesis: 'kinesis', + dynamodb: 'nosql', +} + +const SAMPLE_MESSAGE = '{"kind": "order", "id": 1}' + +export function PipePanel({resource, runtimeReachable}: Props) { + const accountId = useAccountId() + const queryClient = useQueryClient() + const [message, setMessage] = useState(SAMPLE_MESSAGE) + const [notice, setNotice] = useState('') + const pipeKey = ['pipe', accountId, resource?.id] + + // The table row is a ListPipes entry; the filter, template and role come from DescribePipe. + const pipeQuery = useQuery({ + queryKey: pipeKey, + queryFn: ({signal}) => getCloudResource('aws', 'pipes', resource?.id ?? '', signal), + enabled: Boolean(resource) && runtimeReachable, + }) + + useEffect(() => { + setMessage(SAMPLE_MESSAGE) + setNotice('') + }, [resource?.id]) + + const refresh = () => Promise.all([ + queryClient.invalidateQueries({queryKey: pipeKey}), + queryClient.invalidateQueries({queryKey: cloudQueryKeys.resources('aws', 'pipes')}), + ]) + + const stateMutation = useMutation({ + mutationFn: (running: boolean) => running ? startCloudResource('aws', 'pipes', resource?.id ?? '') : stopCloudResource('aws', 'pipes', resource?.id ?? ''), + onSuccess: (_result, running) => { + setNotice(running ? 'Pipe started. It reads the source again.' : 'Pipe stopped. Records wait in the source until it starts again.') + return refresh() + }, + }) + const sendMutation = useMutation({ + mutationFn: (queue: string) => sendSqsMessage('aws', queue, {messageBody: message}), + onSuccess: (sent) => setNotice(`Sent message ${sent.messageId} to the source queue. The pipe picks it up within a few seconds.`), + }) + + if (!resource) { + return

Select a pipe

Choose a pipe to see where its events flow, start or stop it, or send it a test message.

+ } + + const pipe = pipeQuery.data ?? resource + const metadata = pipe.metadata + const state = pipe.status ?? 'UNKNOWN' + const running = state === 'RUNNING' + const filters = Array.isArray(metadata.filterPatterns) ? metadata.filterPatterns as string[] : [] + const source = text(metadata.source) + const sourceQueue = source.split(':')[2] === 'sqs' ? source.split(':')[5] : '' + const pending = stateMutation.isPending || sendMutation.isPending + const disabled = !runtimeReachable || pending + const error = pipeQuery.error ?? stateMutation.error ?? sendMutation.error + + return
+
+

Pipe

{state}{metadata.desiredState && metadata.desiredState !== state ? ` · going to ${String(metadata.desiredState)}` : ''}{metadata.description ? ` · ${String(metadata.description)}` : ''}
+
+ + +
+
+ {typeof metadata.stateReason === 'string' && metadata.stateReason &&

{metadata.stateReason}

} + +
    + + {filters.length > 0 &&
  1. Filter{filters.map((pattern) => {pattern})}
    2. } + {text(metadata.enrichment) && } + input template {text(metadata.inputTemplate)} : undefined}/> +
+ {text(metadata.roleArn) &&

Runs as {text(metadata.roleArn)}

} + + {sourceQueue && <> +

Try it

Send a message to the source queue{filters.length > 0 ? '. Only messages matching the filter reach the target' : ''}.
+
{ event.preventDefault(); setNotice(''); sendMutation.mutate(sourceQueue) }}> +