diff --git a/README.md b/README.md index 25cf884..aa09c3c 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,7 @@ cd packages/api && bun run scripts/service-matrix.ts | Group | Service | AWS | Azure | GCP | |---|---|---|---|---| | Compute | Compute | Yes (list, inspect, create, delete) | Yes (list, inspect, create, delete) | No | +| Compute | EC2 Auto Scaling / Auto Scaling | Yes (list, create, delete, inspect) | No | No | | Compute | EKS / AKS / GKE | Yes (list, inspect) | Yes (list, inspect) | Yes (list, create, inspect, delete) | | Compute | Serverless | Yes (list, create, update, inspect, delete) | Runtime gap | Yes (list, create, inspect, delete) | | Compute | Containers / Cloud Run | No | No | Yes (list, create, delete, inspect) | @@ -81,6 +82,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..dbec764 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-auto-scaling": "^3.1146.0", "@aws-sdk/client-cloudformation": "^3.1111.0", "@aws-sdk/client-cloudfront": "^3.1146.0", "@aws-sdk/client-cloudwatch": "^3.1076.0", @@ -37,6 +38,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/AwsAutoScalingAdapter.test.ts b/packages/api/src/adapter-aws/AwsAutoScalingAdapter.test.ts new file mode 100644 index 0000000..ce9854e --- /dev/null +++ b/packages/api/src/adapter-aws/AwsAutoScalingAdapter.test.ts @@ -0,0 +1,160 @@ +import {describe, expect, test} from 'bun:test' +import { + type AutoScalingClient, + CreateAutoScalingGroupCommand, + DeleteAutoScalingGroupCommand, + DescribeAutoScalingGroupsCommand, + DescribeScalingActivitiesCommand, + TerminateInstanceInAutoScalingGroupCommand, + UpdateAutoScalingGroupCommand, +} from '@aws-sdk/client-auto-scaling' +import {DescribeSubnetsCommand, type EC2Client} from '@aws-sdk/client-ec2' +import {AwsAutoScalingAdapter} from './AwsAutoScalingAdapter' +import {NotFoundError, ValidationError} from '../cloud-spi/errors' + +type SendResult = Record + +function stub(handler: (command: object) => SendResult | Promise) { + const sent: object[] = [] + const client = { + async send(command: object) { + sent.push(command) + return handler(command) + }, + } + return {client, sent} +} + +const group = { + AutoScalingGroupName: 'web-asg', + AutoScalingGroupARN: 'arn:aws:autoscaling:us-east-1:000000000000:autoScalingGroup:web-asg', + LaunchTemplate: {LaunchTemplateName: 'web-lt', Version: '$Latest'}, + MinSize: 1, MaxSize: 3, DesiredCapacity: 2, + VPCZoneIdentifier: 'subnet-a,subnet-b', + HealthCheckType: 'EC2', + Instances: [ + {InstanceId: 'i-b8736c8b4a8c129a3', AvailabilityZone: 'us-east-1a', LifecycleState: 'InService', HealthStatus: 'Healthy', InstanceType: 't3.micro'}, + {InstanceId: 'i-a3155c876867976b1', AvailabilityZone: 'us-east-1b', LifecycleState: 'Pending', HealthStatus: 'Healthy', InstanceType: 't3.micro'}, + ], + CreatedTime: new Date('2026-10-25T16:49:54Z'), +} +const activity = {ActivityId: 'a1', Description: 'Launching a new EC2 instance: i-b8736c8b4a8c129a3', Cause: 'An instance was started in response to a desired capacity change.', StatusCode: 'Successful', StartTime: new Date('2026-10-25T16:50:09Z')} + +function adapterWith(handler: (command: object) => SendResult | undefined) { + const autoscaling = stub((command) => handler(command) ?? ( + command instanceof DescribeAutoScalingGroupsCommand ? {AutoScalingGroups: [group]} + : command instanceof DescribeScalingActivitiesCommand ? {Activities: [activity]} + : {})) + const ec2 = stub((command) => command instanceof DescribeSubnetsCommand ? {Subnets: [{SubnetId: 'subnet-default-b'}, {SubnetId: 'subnet-default-a'}]} : {}) + return { + sent: autoscaling.sent, + ec2Sent: ec2.sent, + adapter: new AwsAutoScalingAdapter(autoscaling.client as unknown as AutoScalingClient, ec2.client as unknown as EC2Client), + } +} + +describe('AwsAutoScalingAdapter', () => { + test('lists groups across pages with their capacity and template', async () => { + const {adapter} = adapterWith((command) => { + if (!(command instanceof DescribeAutoScalingGroupsCommand)) return undefined + return command.input.NextToken + ? {AutoScalingGroups: [{...group, AutoScalingGroupName: 'jobs', LaunchTemplate: {LaunchTemplateName: 'jobs-lt'}, Instances: []}]} + : {AutoScalingGroups: [group], NextToken: 'n'} + }) + + const groups = await adapter.list() + expect(groups.map((g) => [g.name, g.metadata.capacity, g.metadata.desiredCapacity, g.metadata.launchTemplateName])).toEqual([ + ['web-asg', '1 in service', 2, 'web-lt'], + ['jobs', '0 in service', 2, 'jobs-lt'], + ]) + expect((await adapter.list({search: 'jobs-lt'})).map((g) => g.name)).toEqual(['jobs']) + }) + + test('describes the instances and recent scaling activity', async () => { + const {adapter} = adapterWith(() => undefined) + const detail = await adapter.get('web-asg') + expect(detail?.metadata.instances).toEqual([ + {id: 'i-b8736c8b4a8c129a3', availabilityZone: 'us-east-1a', lifecycleState: 'InService', health: 'Healthy', instanceType: 't3.micro', protectedFromScaleIn: false}, + {id: 'i-a3155c876867976b1', availabilityZone: 'us-east-1b', lifecycleState: 'Pending', health: 'Healthy', instanceType: 't3.micro', protectedFromScaleIn: false}, + ]) + expect(detail?.metadata.activities).toEqual([{id: 'a1', description: activity.Description, cause: activity.Cause, status: 'Successful', statusMessage: null, startTime: '2026-10-25T16:50:09.000Z'}]) + }) + + test('keeps the group readable when its activities cannot be read', async () => { + const {adapter} = adapterWith((command) => { + if (command instanceof DescribeScalingActivitiesCommand) throw new Error('throttled') + return undefined + }) + expect((await adapter.get('web-asg'))?.metadata.activitiesUnavailable).toBe(true) + }) + + test('a missing group is null', async () => { + const {adapter} = adapterWith((command) => command instanceof DescribeAutoScalingGroupsCommand ? {AutoScalingGroups: []} : undefined) + expect(await adapter.get('nope')).toBeNull() + }) + + test('creates a group in the default subnets when none are given', async () => { + const {adapter, sent} = adapterWith(() => undefined) + + await adapter.create({values: {name: 'web-asg', launchTemplate: 'web-lt', minSize: '1', desiredCapacity: '2', maxSize: '3'}}) + expect((sent[0] as CreateAutoScalingGroupCommand).input).toEqual({ + AutoScalingGroupName: 'web-asg', + LaunchTemplate: {LaunchTemplateName: 'web-lt', Version: '$Latest'}, + MinSize: 1, MaxSize: 3, DesiredCapacity: 2, + VPCZoneIdentifier: 'subnet-default-a,subnet-default-b', + HealthCheckType: 'EC2', + }) + }) + + test('creates from a template ID with given subnets, target groups and ELB health checks', async () => { + const {adapter, sent, ec2Sent} = adapterWith(() => undefined) + const targetGroup = 'arn:aws:elasticloadbalancing:us-east-1:000000000000:targetgroup/web/abc' + + await adapter.create({values: {name: 'web-asg', launchTemplate: 'lt-ba034c5aa8346f602', launchTemplateVersion: '2', minSize: '0', maxSize: '2', subnets: 'subnet-a, subnet-b', targetGroupArns: targetGroup, healthCheckType: 'ELB'}}) + expect((sent[0] as CreateAutoScalingGroupCommand).input).toMatchObject({ + LaunchTemplate: {LaunchTemplateId: 'lt-ba034c5aa8346f602', Version: '2'}, + DesiredCapacity: 0, + VPCZoneIdentifier: 'subnet-a,subnet-b', + TargetGroupARNs: [targetGroup], + HealthCheckType: 'ELB', + HealthCheckGracePeriod: 300, + }) + expect(ec2Sent).toHaveLength(0) + }) + + test('refuses capacities AWS would reject, before calling it', async () => { + const {adapter, sent} = adapterWith(() => undefined) + const base = {name: 'g', launchTemplate: 'web-lt'} + await expect(adapter.create({values: {...base, minSize: '3', maxSize: '1'}})).rejects.toThrow('minSize cannot be larger than maxSize') + await expect(adapter.create({values: {...base, minSize: '1', desiredCapacity: '5', maxSize: '3'}})).rejects.toThrow('desiredCapacity must be between') + await expect(adapter.create({values: {...base, minSize: '0', maxSize: '500'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, minSize: '', maxSize: '1'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, minSize: '1', maxSize: '1', launchTemplateVersion: 'latest'}})).rejects.toBeInstanceOf(ValidationError) + await expect(adapter.create({values: {...base, name: 'a:b', minSize: '1', maxSize: '1'}})).rejects.toBeInstanceOf(ValidationError) + expect(sent).toHaveLength(0) + }) + + test('changes capacity, keeping the values not given', async () => { + const {adapter, sent} = adapterWith(() => undefined) + + await adapter.setAutoScalingCapacity('web-asg', {desiredCapacity: 3}) + expect((sent.find((command) => command instanceof UpdateAutoScalingGroupCommand) as UpdateAutoScalingGroupCommand).input).toEqual({AutoScalingGroupName: 'web-asg', MinSize: 1, DesiredCapacity: 3, MaxSize: 3}) + await expect(adapter.setAutoScalingCapacity('web-asg', {desiredCapacity: 4})).rejects.toThrow('desiredCapacity must be between') + }) + + test('terminates an instance of the group, replacing it or shrinking the group', async () => { + const {adapter, sent} = adapterWith(() => undefined) + + await adapter.terminateAutoScalingInstance('web-asg', 'i-b8736c8b4a8c129a3', true) + await adapter.terminateAutoScalingInstance('web-asg', 'i-b8736c8b4a8c129a3', false) + const terminations = sent.filter((command): command is TerminateInstanceInAutoScalingGroupCommand => command instanceof TerminateInstanceInAutoScalingGroupCommand) + expect(terminations.map((command) => command.input.ShouldDecrementDesiredCapacity)).toEqual([false, true]) + await expect(adapter.terminateAutoScalingInstance('web-asg', 'i-0000000000000000', true)).rejects.toBeInstanceOf(NotFoundError) + }) + + test('deletes a group together with its instances', async () => { + const {adapter, sent} = adapterWith(() => undefined) + await adapter.delete('web-asg') + expect((sent[0] as DeleteAutoScalingGroupCommand).input).toEqual({AutoScalingGroupName: 'web-asg', ForceDelete: true}) + }) +}) diff --git a/packages/api/src/adapter-aws/AwsAutoScalingAdapter.ts b/packages/api/src/adapter-aws/AwsAutoScalingAdapter.ts new file mode 100644 index 0000000..20c7344 --- /dev/null +++ b/packages/api/src/adapter-aws/AwsAutoScalingAdapter.ts @@ -0,0 +1,227 @@ +import { + type Activity, + type AutoScalingClient, + type AutoScalingGroup, + CreateAutoScalingGroupCommand, + DeleteAutoScalingGroupCommand, + DescribeAutoScalingGroupsCommand, + DescribeScalingActivitiesCommand, + TerminateInstanceInAutoScalingGroupCommand, + UpdateAutoScalingGroupCommand, +} from '@aws-sdk/client-auto-scaling' +import {DescribeSubnetsCommand, type EC2Client} from '@aws-sdk/client-ec2' +import {AUTO_SCALING_GROUP_NAME_PATTERN, AUTO_SCALING_MAX_INSTANCES, awsAutoScalingSchema} from '../cloud-spi/autoScalingSchema' +import {NotFoundError, RuntimeError, ValidationError} from '../cloud-spi/errors' +import type {AutoScalingCapacityInput, CloudResource, CloudServiceAdapter, CreateResourceInput, ResourceQuery, ServiceSchema} from '../cloud-spi/types' + +/** The most recent activities a group's detail shows. */ +const ACTIVITY_LIMIT = 20 + +/** + * Talks to EC2 Auto Scaling, verified against Floci core (community, `latest`). + * + * Notes from probing the runtime: + * - Groups really scale: a reconciler runs every 10 seconds and launches or + * terminates EC2 instances (containers) until the group matches its desired + * capacity, registering them with the group's target groups, and records each + * launch and termination as a scaling activity. + * - Scaling policies are stored but never run: the runtime's alarms drive only + * Application Auto Scaling (ECS services), so this adapter does not offer them. + * + * Groups are keyed by name, which is unique per region. + */ +export class AwsAutoScalingAdapter implements CloudServiceAdapter { + readonly cloud = 'aws' as const + readonly service = 'autoscaling' as const + + constructor(private readonly autoscaling: AutoScalingClient, private readonly ec2: EC2Client) {} + + schema(): ServiceSchema { + return awsAutoScalingSchema() + } + + async list(query: ResourceQuery = {}): Promise { + const groups: AutoScalingGroup[] = [] + let next: string | undefined + do { + const res = await this.autoscaling.send(new DescribeAutoScalingGroupsCommand({MaxRecords: 100, ...(next ? {NextToken: next} : {})})) + groups.push(...(res.AutoScalingGroups ?? [])) + next = res.NextToken + } while (next) + + const search = query.search?.trim().toLowerCase() + return groups.map((group) => toResource(group)).filter((group) => !search + || group.name.toLowerCase().includes(search) + || String(group.metadata.launchTemplateName ?? '').toLowerCase().includes(search)) + } + + async get(id: string): Promise { + const group = await this.describe(id) + if (!group) return null + return toResource(group, await this.activities(id)) + } + + async create(input: CreateResourceInput): Promise { + const values = input.values + const name = text(values.name) + if (!new RegExp(AUTO_SCALING_GROUP_NAME_PATTERN).test(name)) throw new ValidationError('name must be 1 to 255 characters, without a colon') + + const template = text(values.launchTemplate) + if (!template) throw new ValidationError('launchTemplate is required') + const version = text(values.launchTemplateVersion) || '$Latest' + if (!/^(\d+|\$Latest|\$Default)$/.test(version)) throw new ValidationError('launchTemplateVersion must be a version number, $Latest or $Default') + + const minSize = count(values.minSize, 'minSize') + const maxSize = count(values.maxSize, 'maxSize') + const desiredCapacity = text(values.desiredCapacity) ? count(values.desiredCapacity, 'desiredCapacity') : minSize + checkOrder({minSize, desiredCapacity, maxSize}) + + const subnets = text(values.subnets).split(',').map((subnet) => subnet.trim()).filter(Boolean) + if (subnets.some((subnet) => !/^subnet-[\w-]+$/.test(subnet))) throw new ValidationError('subnets must be subnet IDs separated by commas') + const targetGroupArns = text(values.targetGroupArns).split(/\s+/).filter(Boolean) + if (targetGroupArns.some((arn) => !/^arn:aws[a-z-]*:elasticloadbalancing:.+:targetgroup\/.+$/.test(arn))) { + throw new ValidationError('targetGroupArns must be target group ARNs, one per line') + } + const healthCheckType = values.healthCheckType === 'ELB' ? 'ELB' : 'EC2' + + await this.autoscaling.send(new CreateAutoScalingGroupCommand({ + AutoScalingGroupName: name, + LaunchTemplate: {...(template.startsWith('lt-') ? {LaunchTemplateId: template} : {LaunchTemplateName: template}), Version: version}, + MinSize: minSize, + MaxSize: maxSize, + DesiredCapacity: desiredCapacity, + VPCZoneIdentifier: (subnets.length > 0 ? subnets : await this.defaultSubnets()).join(','), + ...(targetGroupArns.length > 0 ? {TargetGroupARNs: targetGroupArns} : {}), + HealthCheckType: healthCheckType, + // AWS's default grace period when a load balancer decides health. + ...(healthCheckType === 'ELB' ? {HealthCheckGracePeriod: 300} : {}), + })) + + const created = await this.get(name) + if (!created) throw new RuntimeError(`EC2 Auto Scaling did not return the new group ${name}`) + return created + } + + /** ForceDelete terminates the group's instances too, which is what deleting a group in the console means. */ + async delete(id: string): Promise { + await this.autoscaling.send(new DeleteAutoScalingGroupCommand({AutoScalingGroupName: id, ForceDelete: true})) + } + + async setAutoScalingCapacity(id: string, input: AutoScalingCapacityInput): Promise { + const group = await this.describe(id) + if (!group) throw new NotFoundError(`Auto Scaling group ${id} was not found`) + + const next = { + minSize: input.minSize === undefined ? group.MinSize ?? 0 : count(input.minSize, 'minSize'), + desiredCapacity: input.desiredCapacity === undefined ? group.DesiredCapacity ?? 0 : count(input.desiredCapacity, 'desiredCapacity'), + maxSize: input.maxSize === undefined ? group.MaxSize ?? 0 : count(input.maxSize, 'maxSize'), + } + checkOrder(next) + + await this.autoscaling.send(new UpdateAutoScalingGroupCommand({ + AutoScalingGroupName: id, + MinSize: next.minSize, + DesiredCapacity: next.desiredCapacity, + MaxSize: next.maxSize, + })) + const updated = await this.get(id) + if (!updated) throw new NotFoundError(`Auto Scaling group ${id} was not found`) + return updated + } + + async terminateAutoScalingInstance(id: string, instanceId: string, replace: boolean): Promise { + if (!/^i-[0-9a-f]+$/.test(instanceId)) throw new ValidationError('instanceId must be an EC2 instance ID such as i-0abc1234') + const group = await this.describe(id) + if (!group?.Instances?.some((instance) => instance.InstanceId === instanceId)) { + throw new NotFoundError(`Instance ${instanceId} is not in Auto Scaling group ${id}`) + } + await this.autoscaling.send(new TerminateInstanceInAutoScalingGroupCommand({InstanceId: instanceId, ShouldDecrementDesiredCapacity: !replace})) + } + + private async describe(name: string): Promise { + const res = await this.autoscaling.send(new DescribeAutoScalingGroupsCommand({AutoScalingGroupNames: [name]})) + return res.AutoScalingGroups?.find((group) => group.AutoScalingGroupName === name) ?? null + } + + /** Activities are enrichment: a failure leaves the group readable rather than failing the request. */ + private async activities(name: string): Promise { + try { + const res = await this.autoscaling.send(new DescribeScalingActivitiesCommand({AutoScalingGroupName: name, MaxRecords: ACTIVITY_LIMIT})) + return res.Activities ?? [] + } catch { + return 'unavailable' + } + } + + /** AWS needs subnets or zones; like its console, fall back to the default VPC's subnets. */ + private async defaultSubnets(): Promise { + const res = await this.ec2.send(new DescribeSubnetsCommand({Filters: [{Name: 'default-for-az', Values: ['true']}]})) + const subnets = (res.Subnets ?? []).map((subnet) => subnet.SubnetId).filter((id): id is string => Boolean(id)).sort() + if (subnets.length === 0) throw new ValidationError('There is no default VPC to launch into; enter subnet IDs') + return subnets + } +} + +function toResource(group: AutoScalingGroup, activities?: Activity[] | 'unavailable'): CloudResource { + const name = group.AutoScalingGroupName ?? '' + const instances = group.Instances ?? [] + const template = group.LaunchTemplate ?? group.MixedInstancesPolicy?.LaunchTemplate?.LaunchTemplateSpecification + return { + id: name, + name, + cloud: 'aws', + service: 'autoscaling', + type: 'autoScalingGroup', + region: group.AutoScalingGroupARN?.split(':')[3] || null, + createdAt: group.CreatedTime ? group.CreatedTime.toISOString() : null, + status: group.Status ?? null, + metadata: { + arn: group.AutoScalingGroupARN, + minSize: group.MinSize ?? 0, + maxSize: group.MaxSize ?? 0, + desiredCapacity: group.DesiredCapacity ?? 0, + capacity: `${instances.filter((instance) => instance.LifecycleState === 'InService').length} in service`, + launchTemplateName: template?.LaunchTemplateName ?? template?.LaunchTemplateId ?? group.LaunchConfigurationName ?? null, + launchTemplateVersion: template?.Version ?? null, + subnets: (group.VPCZoneIdentifier ?? '').split(',').filter(Boolean), + targetGroupArns: group.TargetGroupARNs ?? [], + healthCheckType: group.HealthCheckType ?? null, + suspendedProcesses: (group.SuspendedProcesses ?? []).map((process) => process.ProcessName).filter(Boolean), + instances: instances.map((instance) => ({ + id: instance.InstanceId ?? '', + availabilityZone: instance.AvailabilityZone ?? null, + lifecycleState: instance.LifecycleState ?? null, + health: instance.HealthStatus ?? null, + instanceType: instance.InstanceType ?? null, + protectedFromScaleIn: instance.ProtectedFromScaleIn ?? false, + })), + ...(activities === 'unavailable' ? {activitiesUnavailable: true} : activities ? { + activities: activities.map((activity) => ({ + id: activity.ActivityId ?? '', + description: activity.Description ?? '', + cause: activity.Cause ?? '', + status: activity.StatusCode ?? null, + statusMessage: activity.StatusMessage ?? null, + startTime: activity.StartTime ? activity.StartTime.toISOString() : null, + })), + } : {}), + }, + } +} + +function checkOrder({minSize, desiredCapacity, maxSize}: Required): void { + if (minSize > maxSize) throw new ValidationError('minSize cannot be larger than maxSize') + if (desiredCapacity < minSize || desiredCapacity > maxSize) throw new ValidationError('desiredCapacity must be between minSize and maxSize') +} + +function count(value: unknown, field: string): number { + const raw = typeof value === 'number' ? value : Number(text(value)) + if (!Number.isInteger(raw) || raw < 0 || raw > AUTO_SCALING_MAX_INSTANCES || (typeof value === 'string' && !text(value))) { + throw new ValidationError(`${field} must be a whole number from 0 to ${AUTO_SCALING_MAX_INSTANCES}`) + } + return raw +} + +function text(value: unknown): string { + return typeof value === 'string' ? value.trim() : '' +} 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..ced2124 100644 --- a/packages/api/src/aws.ts +++ b/packages/api/src/aws.ts @@ -34,7 +34,9 @@ import { LightsailClient } from "@aws-sdk/client-lightsail"; import { CognitoIdentityProviderClient } from "@aws-sdk/client-cognito-identity-provider"; import { Route53Client } from "@aws-sdk/client-route-53"; import { CloudFrontClient } from "@aws-sdk/client-cloudfront"; +import { AutoScalingClient } from "@aws-sdk/client-auto-scaling"; 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"; @@ -100,7 +102,9 @@ export type AwsClients = { cognito: CognitoIdentityProviderClient; route53: Route53Client; cloudfront: CloudFrontClient; + autoscaling: AutoScalingClient; acm: ACMClient; + pipes: PipesClient; }; export type AwsClientName = keyof AwsClients; @@ -150,7 +154,9 @@ function buildClients(accountId: string): AwsClients { cognito: new CognitoIdentityProviderClient(base), route53: new Route53Client(base), cloudfront: new CloudFrontClient(base), + autoscaling: new AutoScalingClient(base), acm: new ACMClient(base), + pipes: new PipesClient(base), }; } diff --git a/packages/api/src/cloud-spi/autoScalingSchema.ts b/packages/api/src/cloud-spi/autoScalingSchema.ts new file mode 100644 index 0000000..b2d82c1 --- /dev/null +++ b/packages/api/src/cloud-spi/autoScalingSchema.ts @@ -0,0 +1,103 @@ +import type {CloudProvider, FieldSchema, ServiceSchema, TableColumnSchema} from './types' + +/** CreateAutoScalingGroup's name rule: 1 to 255 characters, no colon. */ +export const AUTO_SCALING_GROUP_NAME_PATTERN = '^[^:]{1,255}$' +/** A capacity the console accepts; AWS's own ceiling is far above what a laptop can run. */ +export const AUTO_SCALING_MAX_INSTANCES = 20 + +const groupColumns: TableColumnSchema[] = [ + {name: 'name', label: 'Name'}, + {name: 'capacity', label: 'Instances', path: 'metadata.capacity'}, + {name: 'desired', label: 'Desired', path: 'metadata.desiredCapacity'}, + {name: 'min', label: 'Min', path: 'metadata.minSize'}, + {name: 'max', label: 'Max', path: 'metadata.maxSize'}, + {name: 'launchTemplate', label: 'Launch template', path: 'metadata.launchTemplateName', emptyText: '—'}, + {name: 'createdAt', label: 'Created', format: 'datetime'}, +] + +const groupFilters: FieldSchema[] = [{name: 'search', label: 'Search', type: 'text', required: false}] + +const capacity = {pattern: '^\\d{1,2}$', message: `Use a whole number from 0 to ${AUTO_SCALING_MAX_INSTANCES}`} + +export function awsAutoScalingSchema(): ServiceSchema { + return { + cloud: 'aws', + service: 'autoscaling', + displayName: 'EC2 Auto Scaling groups', + fields: [ + { + name: 'name', + label: 'Name', + type: 'text', + required: true, + validation: {pattern: AUTO_SCALING_GROUP_NAME_PATTERN, message: 'Use up to 255 characters, without a colon'}, + }, + { + name: 'launchTemplate', + label: 'Launch template', + type: 'text', + required: true, + description: 'The name or ID (lt-…) of an EC2 launch template. New instances use its image and instance type.', + }, + { + name: 'launchTemplateVersion', + label: 'Template version', + type: 'text', + required: false, + defaultValue: '$Latest', + description: 'A version number, $Latest or $Default.', + validation: {pattern: '^(\\d+|\\$Latest|\\$Default)$', message: 'Use a version number, $Latest or $Default'}, + }, + {name: 'minSize', label: 'Minimum', type: 'text', required: true, defaultValue: '1', validation: capacity}, + {name: 'desiredCapacity', label: 'Desired', type: 'text', required: false, defaultValue: '1', validation: capacity, description: 'How many instances to run now. Between the minimum and the maximum.'}, + {name: 'maxSize', label: 'Maximum', type: 'text', required: true, defaultValue: '2', validation: capacity}, + { + name: 'subnets', + label: 'Subnets', + type: 'text', + required: false, + span: true, + description: 'Subnet IDs to launch into, separated by commas. Leave empty for the default subnets.', + validation: {pattern: '^\\s*subnet-[\\w-]+(\\s*,\\s*subnet-[\\w-]+)*\\s*$', message: 'Use subnet IDs separated by commas'}, + }, + { + name: 'targetGroupArns', + label: 'Target groups', + type: 'textarea', + required: false, + span: true, + description: 'Optional load balancer target group ARNs, one per line. Instances are registered with them as they launch.', + }, + { + name: 'healthCheckType', + label: 'Health check', + type: 'select', + required: false, + defaultValue: 'EC2', + options: [ + {label: 'EC2 instance status', value: 'EC2'}, + {label: 'Load balancer health checks', value: 'ELB'}, + ], + }, + ], + actions: ['list', 'create', 'delete', 'inspect'], + capabilities: { + resourceActions: [ + {name: 'list', label: 'List Auto Scaling groups', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'create', label: 'Create Auto Scaling group', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'delete', label: 'Delete group and its instances', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'inspect', label: 'Inspect Auto Scaling group', enabled: true, status: 'available', runtimeRequired: true}, + ], + autoScalingActions: [ + {name: 'setCapacity', label: 'Change minimum, desired and maximum', enabled: true, status: 'available', runtimeRequired: true}, + {name: 'terminateInstance', label: 'Terminate an instance', enabled: true, status: 'available', runtimeRequired: true}, + ], + }, + filters: groupFilters, + columns: groupColumns, + } +} + +export function autoScalingSchemaFor(cloud: CloudProvider): ServiceSchema | null { + return cloud === 'aws' ? awsAutoScalingSchema() : null +} 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..21d5b9a 100644 --- a/packages/api/src/cloud-spi/serviceCatalog.ts +++ b/packages/api/src/cloud-spi/serviceCatalog.ts @@ -74,6 +74,13 @@ export const SERVICE_CATALOG = { order: 20, }, serverless: {displayName: 'Serverless', iconKey: 'serverless', group: 'Compute', order: 30}, + autoscaling: { + displayName: 'Auto Scaling', + displayNameByCloud: {aws: 'EC2 Auto Scaling'}, + iconKey: 'autoscaling', + group: 'Compute', + order: 15, + }, containers: { displayName: 'Containers', displayNameByCloud: {gcp: 'Cloud Run'}, @@ -281,6 +288,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/cloud-spi/types.ts b/packages/api/src/cloud-spi/types.ts index 9fdc786..72208f6 100644 --- a/packages/api/src/cloud-spi/types.ts +++ b/packages/api/src/cloud-spi/types.ts @@ -186,6 +186,7 @@ export type UserPoolActionName = | 'describeClient' | 'updateClientOAuth' export type DistributionActionName = 'setEnabled' | 'createInvalidation' +export type AutoScalingActionName = 'setCapacity' | 'terminateInstance' export type CertificateActionName = 'download' | 'export' | 'revoke' export type MonitoringActionName = 'listMetrics' | 'getMetricData' | 'alarmHistory' | 'setAlarmState' | 'setAlarmActions' export type HttpApiActionName = @@ -246,6 +247,7 @@ export interface ServiceSchema { kubernetesActions?: CapabilitySchema[] userPoolActions?: CapabilitySchema[] distributionActions?: CapabilitySchema[] + autoScalingActions?: CapabilitySchema[] certificateActions?: CapabilitySchema[] monitoringActions?: CapabilitySchema[] /** API Gateway v2 (HTTP and WebSocket APIs): what the routes, stages and security panel can do. */ @@ -537,6 +539,13 @@ export type CertificateRevocationReason = | 'CESSATION_OF_OPERATION' | 'PRIVILEGE_WITHDRAWN' +/** Any of the three may be left out to keep its current value. */ +export interface AutoScalingCapacityInput { + minSize?: number + desiredCapacity?: number + maxSize?: number +} + export interface DistributionInvalidation { id: string status: string @@ -1204,6 +1213,9 @@ export interface CloudServiceAdapter { setAlarmState?(alarmName: string, state: AlarmStateValue, reason: string): Promise setAlarmActionsEnabled?(alarmName: string, enabled: boolean): Promise createDistributionInvalidation?(distributionId: string, paths: string[]): Promise + setAutoScalingCapacity?(groupName: string, input: AutoScalingCapacityInput): Promise + /** 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 exportCertificate?(certificateId: string, passphrase: string): Promise revokeCertificate?(certificateId: string, reason: CertificateRevocationReason): Promise diff --git a/packages/api/src/cloudProxy.test.ts b/packages/api/src/cloudProxy.test.ts index 2cacbcf..adc7043 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 {CloudProvider, CloudServiceAdapter, CloudServiceType, CertificateActionName, DatabaseActionName, DistributionActionName, HttpApiActionName, KubernetesActionName, MonitoringActionName, ResourceActionName, UserPoolActionName} from './cloud-spi/types' +import type {AutoScalingActionName, CloudProvider, CloudServiceAdapter, CloudServiceType, CertificateActionName, DatabaseActionName, DistributionActionName, HttpApiActionName, KubernetesActionName, MonitoringActionName, ResourceActionName, UserPoolActionName} from './cloud-spi/types' /** * Guards the schema-to-implementation contract for every registered adapter. @@ -81,6 +81,11 @@ const METHOD_FOR_CERTIFICATE_ACTION: Record = { + setCapacity: 'setAutoScalingCapacity', + terminateInstance: 'terminateAutoScalingInstance', +} + const METHOD_FOR_USER_POOL_ACTION: Record = { updateUser: 'updateUserPoolUser', setUserGroups: 'setUserPoolUserGroups', @@ -203,6 +208,15 @@ describe('registered adapters honour their schema', () => { ).toBe('function') } + const autoScalingClaims = (adapter.schema().capabilities?.autoScalingActions ?? []).filter((capability) => capability.status === 'available') + for (const capability of autoScalingClaims) { + const method = METHOD_FOR_AUTO_SCALING_ACTION[capability.name] + expect( + typeof adapter[method], + `${label} advertises ${capability.name} as available but has no ${String(method)}()`, + ).toBe('function') + } + const distributionClaims = (adapter.schema().capabilities?.distributionActions ?? []).filter((capability) => capability.status === 'available') for (const capability of distributionClaims) { const method = METHOD_FOR_DISTRIBUTION_ACTION[capability.name] @@ -225,6 +239,7 @@ describe('registered adapters honour their schema', () => { ...(adapter.schema().capabilities?.kubernetesActions ?? []), ...(adapter.schema().capabilities?.userPoolActions ?? []), ...(adapter.schema().capabilities?.distributionActions ?? []), + ...(adapter.schema().capabilities?.autoScalingActions ?? []), ...(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 c1f4d4d..96e85af 100644 --- a/packages/api/src/cloudProxy.ts +++ b/packages/api/src/cloudProxy.ts @@ -50,7 +50,9 @@ import {AwsLightsailAdapter} from './adapter-aws/AwsLightsailAdapter' import {AwsUserPoolsAdapter} from './adapter-aws/AwsUserPoolsAdapter' import {AwsDnsAdapter} from './adapter-aws/AwsDnsAdapter' import {AwsCdnAdapter} from './adapter-aws/AwsCdnAdapter' +import {AwsAutoScalingAdapter} from './adapter-aws/AwsAutoScalingAdapter' 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' @@ -104,7 +106,9 @@ export function createCloudAdapterRegistry(accountId?: string | null): CloudAdap new AwsUserPoolsAdapter(clients.cognito), new AwsDnsAdapter(clients.route53), new AwsCdnAdapter(clients.cloudfront, clients.s3), + new AwsAutoScalingAdapter(clients.autoscaling, clients.ec2), new AwsCertificatesAdapter(clients.acm, resolveAccountId(accountId)), + new AwsPipesAdapter(clients.pipes), 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 caa4d5f..636500b 100644 --- a/packages/api/src/routes/clouds.test.ts +++ b/packages/api/src/routes/clouds.test.ts @@ -18,6 +18,7 @@ import {awsStorageSchema, azureStorageSchema, gcpStorageSchema} from '../cloud-s import {awsSesEmailSchema} from '../cloud-spi/emailSchema' import {awsUserPoolsSchema} from '../cloud-spi/userPoolsSchema' import {awsCdnSchema} from '../cloud-spi/cdnSchema' +import {awsAutoScalingSchema} from '../cloud-spi/autoScalingSchema' import {awsCertificatesSchema} from '../cloud-spi/certificatesSchema' import {awsMonitoringSchema} from '../cloud-spi/monitoringSchema' import {awsAppConfigSchema} from '../cloud-spi/appConfigSchema' @@ -852,6 +853,28 @@ describe('cloud schema routes', () => { expect(calls).toEqual([['pem', 'c1'], ['export', 'c1', 'secret'], ['revoke', 'c1', 'UNSPECIFIED']]) }) + test('changes capacity and terminates an instance through the Auto Scaling adapter', async () => { + const calls: unknown[] = [] + const app = appWithRoutes([mockAdapter('aws', { + service: 'autoscaling', + schema: awsAutoScalingSchema, + setAutoScalingCapacity: async (id, input) => { + calls.push(['capacity', id, input]) + return {id, name: id, cloud: 'aws', service: 'autoscaling', type: 'autoScalingGroup', region: null, createdAt: null, metadata: {desiredCapacity: 3}} + }, + terminateAutoScalingInstance: async (id, instanceId, replace) => { + calls.push(['terminate', id, instanceId, replace]) + }, + })]) + const json = (method: string, body: unknown) => ({method, headers: {'content-type': 'application/json'}, body: JSON.stringify(body)}) + const base = '/api/clouds/aws/services/autoscaling/resources/web-asg' + + const updated = await app.request(`${base}/capacity`, json('PUT', {desiredCapacity: 3})) + expect((await updated.json()).metadata.desiredCapacity).toBe(3) + expect((await app.request(`${base}/instances/i-abc/terminate`, json('POST', {replace: true}))).status).toBe(200) + expect(calls).toEqual([['capacity', 'web-asg', {desiredCapacity: 3}], ['terminate', 'web-asg', 'i-abc', true]]) + }) + test('clears the SES mailbox through the email adapter', async () => { let cleared = false const app = appWithRoutes([mockAdapter('aws', { diff --git a/packages/api/src/routes/clouds.ts b/packages/api/src/routes/clouds.ts index a47d753..6eb6a31 100644 --- a/packages/api/src/routes/clouds.ts +++ b/packages/api/src/routes/clouds.ts @@ -2,6 +2,7 @@ import {Hono} from 'hono' import type {Context} from 'hono' import type { AlarmStateValue, + AutoScalingCapacityInput, CertificateRevocationReason, CloudProvider, CloudServiceType, @@ -553,6 +554,24 @@ export function createCloudRoutes(injectedService?: CloudProxyService) { }) }) + app.put('/:cloud/services/autoscaling/resources/:id/capacity', 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).setAutoScalingCapacity(cloud, c.req.param('id'), await c.req.json()))) + }) + + app.post('/:cloud/services/autoscaling/resources/:id/instances/:instanceId/terminate', async (c) => { + const cloud = c.req.param('cloud') as CloudProvider + if (!isCloudProvider(cloud)) return c.json({error: 'Unknown cloud'}, 404) + + return withRuntime(c, async () => { + const body = await c.req.json<{replace?: boolean}>() + await svc(c).terminateAutoScalingInstance(cloud, c.req.param('id'), c.req.param('instanceId'), body.replace === true) + return c.json({ok: true}) + }) + }) + app.get('/:cloud/services/k8s/resources/:id/nodegroups', 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 9127cba..21ee5a6 100644 --- a/packages/api/src/service/CloudProxyService.ts +++ b/packages/api/src/service/CloudProxyService.ts @@ -93,6 +93,7 @@ import type { UserPoolSignInTestResult, UserPoolClientSettings, DistributionInvalidation, + AutoScalingCapacityInput, CertificatePem, CertificateRevocationReason, ExportedCertificate, @@ -963,6 +964,18 @@ export class CloudProxyService { return adapter.revokeCertificate(certificateId, reason) } + async setAutoScalingCapacity(cloud: CloudProvider, groupName: string, input: AutoScalingCapacityInput): Promise { + const adapter = this.requireAdapter(cloud, 'autoscaling') + if (!adapter.setAutoScalingCapacity) throw new NotSupportedError(`Changing capacity is not supported for ${cloud}/autoscaling`) + return adapter.setAutoScalingCapacity(groupName, input) + } + + async terminateAutoScalingInstance(cloud: CloudProvider, groupName: string, instanceId: string, replace: boolean): Promise { + const adapter = this.requireAdapter(cloud, 'autoscaling') + if (!adapter.terminateAutoScalingInstance) throw new NotSupportedError(`Terminating instances is not supported for ${cloud}/autoscaling`) + return adapter.terminateAutoScalingInstance(groupName, instanceId, replace) + } + async describeUserPoolClient(cloud: CloudProvider, poolId: string, clientId: string): Promise { const adapter = this.requireAdapter(cloud, 'userpools') if (!adapter.describeUserPoolClient) throw new NotSupportedError(`App client settings are not supported for ${cloud}/userpools`) diff --git a/packages/frontend/src/api/api.ts b/packages/frontend/src/api/api.ts index ee0dafc..65678d0 100644 --- a/packages/frontend/src/api/api.ts +++ b/packages/frontend/src/api/api.ts @@ -155,6 +155,10 @@ export const apiEndpointKeys = { export: "clouds.services.certificates.export", revoke: "clouds.services.certificates.revoke", }, + autoscaling: { + capacity: "clouds.services.autoscaling.capacity", + terminate: "clouds.services.autoscaling.instances.terminate", + }, cdn: { enabled: "clouds.services.cdn.enabled", invalidations: "clouds.services.cdn.invalidations", @@ -952,6 +956,20 @@ export const endpointRegistry: EndpointRegistry = new Map([ method: "POST", }, ], + [ + apiEndpointKeys.clouds.autoscaling.capacity, + { + path: "/clouds/:cloud/services/autoscaling/resources/:id/capacity", + method: "PUT", + }, + ], + [ + apiEndpointKeys.clouds.autoscaling.terminate, + { + path: "/clouds/:cloud/services/autoscaling/resources/:id/instances/:instanceId/terminate", + method: "POST", + }, + ], [ apiEndpointKeys.clouds.cdn.enabled, { diff --git a/packages/frontend/src/api/cloudProxyClient.ts b/packages/frontend/src/api/cloudProxyClient.ts index 5dad178..178a278 100644 --- a/packages/frontend/src/api/cloudProxyClient.ts +++ b/packages/frontend/src/api/cloudProxyClient.ts @@ -33,6 +33,7 @@ import type { UserPoolSignInTestResult, UserPoolClientSettings, DistributionInvalidation, + AutoScalingCapacityInput, CertificatePem, CertificateRevocationReason, ExportedCertificate, @@ -1720,6 +1721,32 @@ export async function revokeCertificate( return res.data; } +export async function setAutoScalingCapacity( + cloud: CloudProvider, + groupName: string, + input: AutoScalingCapacityInput, +): Promise { + const res = await apiClient.call( + apiEndpointKeys.clouds.autoscaling.capacity, + requestOptions(cloud, "autoscaling", { body: input }), + { cloud, id: groupName }, + ); + return res.data; +} + +export async function terminateAutoScalingInstance( + cloud: CloudProvider, + groupName: string, + instanceId: string, + replace: boolean, +): Promise { + await apiClient.call( + apiEndpointKeys.clouds.autoscaling.terminate, + requestOptions(cloud, "autoscaling", { body: { replace } }), + { cloud, id: groupName, instanceId }, + ); +} + export async function setDistributionEnabled( cloud: CloudProvider, distributionId: string, diff --git a/packages/frontend/src/components/AutoScalingPanel.test.tsx b/packages/frontend/src/components/AutoScalingPanel.test.tsx new file mode 100644 index 0000000..4dbbd0d --- /dev/null +++ b/packages/frontend/src/components/AutoScalingPanel.test.tsx @@ -0,0 +1,85 @@ +import {QueryClient, QueryClientProvider} from '@tanstack/react-query' +import {render, screen, within} from '@testing-library/react' +import userEvent from '@testing-library/user-event' +import {beforeEach, expect, test, vi} from 'vitest' +import {AutoScalingPanel} from './AutoScalingPanel' +import type {CloudResource} from '@/types/resource' + +const api = vi.hoisted(() => ({ + getCloudResource: vi.fn(), + setAutoScalingCapacity: vi.fn(), + terminateAutoScalingInstance: vi.fn(), +})) + +vi.mock('@/api/cloudProxyClient', () => api) +vi.mock('@/lib/accountStore', () => ({useAccountId: () => '000000000000'})) + +function group(metadata: Record = {}): CloudResource { + return { + id: 'web-asg', name: 'web-asg', cloud: 'aws', service: 'autoscaling', type: 'autoScalingGroup', region: 'us-east-1', createdAt: null, status: null, + metadata: { + minSize: 1, desiredCapacity: 2, maxSize: 3, capacity: '2 in service', + instances: [ + {id: 'i-b8736c8b4a8c129a3', availabilityZone: 'us-east-1a', lifecycleState: 'InService', health: 'Healthy', instanceType: 't3.micro'}, + {id: 'i-a3155c876867976b1', availabilityZone: 'us-east-1b', lifecycleState: 'InService', health: 'Healthy', instanceType: 't3.micro'}, + ], + activities: [{id: 'a1', description: 'Launching a new EC2 instance: i-b8736c8b4a8c129a3', cause: 'An instance was started in response to a desired capacity change.', status: 'Successful', statusMessage: null, startTime: null}], + ...metadata, + }, + } +} + +function renderPanel(resource: CloudResource) { + api.getCloudResource.mockResolvedValue(resource) + const client = new QueryClient({defaultOptions: {queries: {retry: false}}}) + return render() +} + +beforeEach(() => { + vi.clearAllMocks() + api.setAutoScalingCapacity.mockImplementation(async (_cloud, _id, input) => group({desiredCapacity: input.desiredCapacity})) + api.terminateAutoScalingInstance.mockResolvedValue(undefined) +}) + +test('shows the instances and what launched them', async () => { + renderPanel(group()) + + expect(await screen.findByText('i-a3155c876867976b1')).toBeInTheDocument() + expect(screen.getByText('Launching a new EC2 instance: i-b8736c8b4a8c129a3')).toBeInTheDocument() + expect(screen.getByText(/never runs them/)).toBeInTheDocument() +}) + +test('changes the desired capacity', async () => { + const user = userEvent.setup() + renderPanel(group()) + + const form = await screen.findByRole('form', {name: 'Group capacity'}) + const update = within(form).getByRole('button', {name: 'Update capacity'}) + expect(update).toBeDisabled() + const desired = within(form).getByLabelText('Desired') + await user.clear(desired) + await user.type(desired, '3') + await user.click(update) + + expect(api.setAutoScalingCapacity).toHaveBeenCalledWith('aws', 'web-asg', {minSize: 1, desiredCapacity: 3, maxSize: 3}) + expect(await screen.findByText(/Capacity set to 3/)).toBeInTheDocument() +}) + +test('replaces an instance, or removes one while above the minimum', async () => { + const user = userEvent.setup() + vi.spyOn(window, 'confirm').mockReturnValue(true) + renderPanel(group()) + + await user.click(await screen.findByRole('button', {name: 'Replace i-b8736c8b4a8c129a3'})) + expect(api.terminateAutoScalingInstance).toHaveBeenCalledWith('aws', 'web-asg', 'i-b8736c8b4a8c129a3', true) + expect(await screen.findByText(/launches a replacement/)).toBeInTheDocument() + + await user.click(screen.getByRole('button', {name: 'Remove i-a3155c876867976b1'})) + expect(api.terminateAutoScalingInstance).toHaveBeenLastCalledWith('aws', 'web-asg', 'i-a3155c876867976b1', false) +}) + +test('cannot remove an instance from a group at its minimum', async () => { + renderPanel(group({minSize: 2})) + expect(await screen.findByRole('button', {name: 'Remove i-b8736c8b4a8c129a3'})).toBeDisabled() + expect(screen.getByRole('button', {name: 'Replace i-b8736c8b4a8c129a3'})).toBeEnabled() +}) diff --git a/packages/frontend/src/components/AutoScalingPanel.tsx b/packages/frontend/src/components/AutoScalingPanel.tsx new file mode 100644 index 0000000..5097eaa --- /dev/null +++ b/packages/frontend/src/components/AutoScalingPanel.tsx @@ -0,0 +1,146 @@ +import {RefreshCw} from 'lucide-react' +import {useMutation, useQuery, useQueryClient} from '@tanstack/react-query' +import {useEffect, useState} from 'react' +import {getCloudResource, setAutoScalingCapacity, terminateAutoScalingInstance} from '@/api/cloudProxyClient' +import {cloudQueryKeys} from '@/api/queries/cloudQueries' +import {useAccountId} from '@/lib/accountStore' +import {formatDateTime} from '@/lib/format' +import type {CloudResource} from '@/types/resource' + +type Props = { + resource?: CloudResource + runtimeReachable: boolean +} + +type Instance = {id: string; availabilityZone: string | null; lifecycleState: string | null; health: string | null; instanceType: string | null} +type Activity = {id: string; description: string; cause: string; status: string | null; statusMessage: string | null; startTime: string | null} +type Capacity = {minSize: string; desiredCapacity: string; maxSize: string} + +/** The runtime reconciles every 10 seconds, so a group that is still moving is re-read this often. */ +const SETTLING_POLL_MS = 4000 + +export function AutoScalingPanel({resource, runtimeReachable}: Props) { + const accountId = useAccountId() + const queryClient = useQueryClient() + const [capacity, setCapacity] = useState({minSize: '', desiredCapacity: '', maxSize: ''}) + const [notice, setNotice] = useState('') + const groupKey = ['autoscaling-group', accountId, resource?.id] + + const groupQuery = useQuery({ + queryKey: groupKey, + queryFn: ({signal}) => getCloudResource('aws', 'autoscaling', resource?.id ?? '', signal), + enabled: Boolean(resource) && runtimeReachable, + refetchInterval: (query) => query.state.data && !settled(query.state.data) ? SETTLING_POLL_MS : false, + }) + + const group = groupQuery.data ?? resource + useEffect(() => { + setNotice('') + }, [resource?.id]) + // The form follows the group's saved capacity whenever it changes, here or elsewhere. + const savedMin = String(group?.metadata.minSize ?? '') + const savedDesired = String(group?.metadata.desiredCapacity ?? '') + const savedMax = String(group?.metadata.maxSize ?? '') + useEffect(() => { + setCapacity({minSize: savedMin, desiredCapacity: savedDesired, maxSize: savedMax}) + }, [group?.id, savedMin, savedDesired, savedMax]) + + const refresh = () => Promise.all([ + queryClient.invalidateQueries({queryKey: groupKey}), + queryClient.invalidateQueries({queryKey: cloudQueryKeys.resources('aws', 'autoscaling')}), + ]) + + const capacityMutation = useMutation({ + mutationFn: () => setAutoScalingCapacity('aws', resource?.id ?? '', { + minSize: Number(capacity.minSize), + desiredCapacity: Number(capacity.desiredCapacity), + maxSize: Number(capacity.maxSize), + }), + onSuccess: (updated) => { + setNotice(`Capacity set to ${String(updated.metadata.desiredCapacity)}. TheCloud launches or terminates instances within about 10 seconds.`) + return refresh() + }, + }) + const terminateMutation = useMutation({ + mutationFn: ({instanceId, replace}: {instanceId: string; replace: boolean}) => terminateAutoScalingInstance('aws', resource?.id ?? '', instanceId, replace), + onSuccess: (_result, {instanceId, replace}) => { + setNotice(replace ? `Terminating ${instanceId}. The group launches a replacement to stay at its desired capacity.` : `Terminating ${instanceId} and lowering the desired capacity by one.`) + return refresh() + }, + }) + + if (!resource || !group) { + return

Select an Auto Scaling group

Choose a group to change its capacity, see its instances and follow its scaling activity.

+ } + + const metadata = group.metadata + const instances = Array.isArray(metadata.instances) ? metadata.instances as Instance[] : [] + const activities = Array.isArray(metadata.activities) ? metadata.activities as Activity[] : [] + const desired = Number(metadata.desiredCapacity ?? 0) + const min = Number(metadata.minSize ?? 0) + const pending = capacityMutation.isPending || terminateMutation.isPending + const disabled = !runtimeReachable || pending + const error = groupQuery.error ?? capacityMutation.error ?? terminateMutation.error + const changed = capacity.minSize !== String(metadata.minSize) || capacity.desiredCapacity !== String(metadata.desiredCapacity) || capacity.maxSize !== String(metadata.maxSize) + const field = (name: keyof Capacity, label: string) => + + return
+
+

Auto Scaling group

{String(metadata.capacity ?? '')} · desired {desired}{settled(group) ? '' : ' · scaling'}{group.status ? ` · ${group.status}` : ''}
+
+ +
+
+ +
{ event.preventDefault(); setNotice(''); capacityMutation.mutate() }}> + {field('minSize', 'Minimum')} + {field('desiredCapacity', 'Desired')} + {field('maxSize', 'Maximum')} +
+
+

Scaling policies are not offered: TheCloud stores EC2 scaling policies but never runs them, so set the capacity here. Alarm-driven scaling works for ECS services through Application Auto Scaling.

+ +

Instances

Each is a real EC2 instance in TheCloud, launched from the group's template.
+ {instances.length === 0 ?

No instances. {desired > 0 ? 'The group is launching them.' : 'Raise the desired capacity to launch some.'}

+ :
{instances.map((instance) => ( + + + + + + + + + ))}
InstanceStateHealthZoneType
{instance.id}{instance.lifecycleState ?? '—'}{instance.health ?? '—'}{instance.availabilityZone ?? '—'}{instance.instanceType ?? '—'} + + +
} + + {activities.length > 0 && <> +

Activity

The most recent launches and terminations.
+
{activities.map((activity) => ( + + + + + + ))}
StartedStatusActivity
{formatDateTime(activity.startTime) ?? '—'}{activity.status ?? '—'}{activity.description}
{activity.statusMessage || activity.cause}
+ } + {metadata.activitiesUnavailable === true &&

The scaling activity could not be read just now.

} + + {notice &&

{notice}

} + {error instanceof Error &&

{error.message}

} +
+} + +/** Settled once every desired instance is in service and none is still starting or stopping. */ +function settled(group: CloudResource): boolean { + const instances = Array.isArray(group.metadata.instances) ? group.metadata.instances as Instance[] : [] + const inService = instances.filter((instance) => instance.lifecycleState === 'InService').length + return inService === Number(group.metadata.desiredCapacity ?? 0) && instances.every((instance) => instance.lifecycleState === 'InService') +} diff --git a/packages/frontend/src/components/DynamicResourceView.tsx b/packages/frontend/src/components/DynamicResourceView.tsx index 7b34487..8f2560f 100644 --- a/packages/frontend/src/components/DynamicResourceView.tsx +++ b/packages/frontend/src/components/DynamicResourceView.tsx @@ -68,7 +68,9 @@ import { LoadBalancerListenerPanel } from "@/components/LoadBalancerListenerPane import { CognitoUsersPanel } from "@/components/CognitoUsersPanel"; import { Route53RecordsPanel } from "@/components/Route53RecordsPanel"; import { CloudFrontPanel } from "@/components/CloudFrontPanel"; +import { AutoScalingPanel } from "@/components/AutoScalingPanel"; 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 +993,12 @@ export function DynamicResourceView({ {service === "certificates" && cloud === "aws" && ( )} + {service === "pipes" && cloud === "aws" && ( + + )} + {service === "autoscaling" && 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) }}> +