diff --git a/README.md b/README.md index b61c410..b93e969 100644 --- a/README.md +++ b/README.md @@ -203,7 +203,7 @@ Run AWS-compatible services locally without an AWS account, auth token, or paid
Real Docker where fidelity matters -Lambda, RDS, Neptune, ElastiCache, MSK, ECS, EC2, EKS, OpenSearch, CodeBuild, and Managed Service for Apache Flink use real Docker-backed execution instead of shallow mocks. +Lambda, RDS, Neptune, ElastiCache, MSK, MSK Connect, ECS, EC2, EKS, OpenSearch, CodeBuild, and Managed Service for Apache Flink use real Docker-backed execution instead of shallow mocks.
@@ -244,7 +244,7 @@ LocalStack's community edition [sunset in March 2026](https://blog.localstack.cl | License | MIT | Restricted | | API Gateway v2 / HTTP API | Yes | No | | Cognito | Yes | No | -| RDS, ElastiCache, MSK | Real Docker | No | +| RDS, ElastiCache, MSK, MSK Connect | Real Docker | No | | Neptune (graph DB + Gremlin WebSocket) | Real Docker | No | | DocumentDB (MongoDB-compatible) | Real Docker | No | | ECS, EC2, EKS | Real Docker | No | @@ -298,7 +298,7 @@ Floci supports local emulation for application services, data services, eventing | Containers and compute | ECS, EC2, Lightsail, EKS, MWAA, ECR, EFS, CodeBuild, CodeDeploy, CodePipeline, CodeGuru Reviewer, CodeArtifact, AWS Batch, Auto Scaling, Application Auto Scaling, Elastic Beanstalk, ELB v2, ELB Classic | | Data, analytics, and AI | Athena, Glue, Lake Formation, EMR, EMR Serverless, Redshift, Redshift Data API, Firehose, Managed Service for Apache Flink, OpenSearch, S3 Tables, S3 Vectors, Textract, Transcribe, Comprehend, Rekognition, Translate, Bedrock, Bedrock Runtime, Bedrock AgentCore, Bedrock AgentCore Control, SageMaker | | Databases and caching | RDS, RDS Data API, Neptune, DocumentDB, DMS, MemoryDB, ElastiCache, Timestream for InfluxDB | -| Messaging and transfer | SES, Kinesis, MSK, Amazon MQ, Transfer Family, DataSync, IoT Core, Amazon Connect, Amazon AppIntegrations | +| Messaging and transfer | SES, Kinesis, MSK, MSK Connect, Amazon MQ, Transfer Family, DataSync, IoT Core, Amazon Connect, Amazon AppIntegrations | | Security and governance | AWS Network Firewall, AWS RAM, Service Quotas, WAF v2, GuardDuty, Amazon Inspector, CloudTrail, CloudFront, Resource Groups Tagging API, Resource Explorer 2, CloudHSM v2, Organizations, AWS Account Management, IAM Access Analyzer, IAM Identity Center (SSO Admin, OIDC, Access Portal, SCIM), Identity Store, Amazon Macie, Amazon Detective, Security Hub, Amazon Verified Permissions, Control Catalog, Control Tower, Service Catalog, AWS Marketplace | | Cost and billing | AWS Budgets, Pricing, Cost Explorer, Cost and Usage Reports, BCM Pricing Calculator, BCM Data Exports | | Resilience, backup, and config | AWS FIS, AWS Backup, AWS Config, AppConfig, AppConfigData, CloudFormation, Cloud Control API | @@ -342,6 +342,7 @@ For operation-level compatibility, see the [Services Overview](https://floci.io/ | Neptune | Real Docker | Graph DB via TinkerPop Gremlin Server (default) or Neo4j for openCypher/Bolt (`NEPTUNE_DB_TYPE`); RDS-shaped control plane; SigV4 proxy on port 8182 | | DocumentDB | Real Docker, mock mode available | MongoDB-compatible cluster via real MongoDB containers; RDS-shaped control plane; MongoDB wire protocol on port 27017 | | MSK | Real Docker | Kafka-compatible broker via Redpanda | +| MSK Connect | Real Docker | Stock Kafka Connect workers with S3 custom plugin ZIPs | | Amazon MQ | Real Docker | RabbitMQ broker via rabbitmq:3-management; AMQP + management console | | Athena | In-process with DuckDB sidecar | Real SQL execution over S3 and Glue-backed views | | Glue | In-process | Data Catalog, Schema Registry, tables consumed by Athena | diff --git a/docker/ec2/ami-images/debian-13/Dockerfile b/docker/ec2/ami-images/debian-13/Dockerfile new file mode 100644 index 0000000..56e1d68 --- /dev/null +++ b/docker/ec2/ami-images/debian-13/Dockerfile @@ -0,0 +1,19 @@ +FROM docker.io/library/debian:trixie-slim + +RUN apt-get update \ + && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ + systemd \ + systemd-sysv \ + openssh-client \ + openssh-server \ + iproute2 \ + iptables \ + socat \ + curl \ + ca-certificates \ + && systemctl enable ssh \ + && rm -rf /var/lib/apt/lists/* + +ENV container=docker +STOPSIGNAL SIGRTMIN+3 +CMD ["/sbin/init"] diff --git a/docker/ec2/ami-images/debian-13/README.md b/docker/ec2/ami-images/debian-13/README.md new file mode 100644 index 0000000..c43758d --- /dev/null +++ b/docker/ec2/ami-images/debian-13/README.md @@ -0,0 +1,10 @@ +# Debian 13 EC2 guest + +Build the local guest image before launching `ami-debian13`: + +```text +docker build --platform linux/amd64 -t floci/ami-debian:13 docker/ec2/ami-images/debian-13 +``` + +The image runs systemd and starts SSH on every EC2 boot. Enable application services with +`systemctl enable --now ` so they restart after `StopInstances` and `StartInstances`. diff --git a/docker/ec2/ami-images/fedora-42/Dockerfile b/docker/ec2/ami-images/fedora-42/Dockerfile new file mode 100644 index 0000000..8939b2f --- /dev/null +++ b/docker/ec2/ami-images/fedora-42/Dockerfile @@ -0,0 +1,17 @@ +FROM registry.fedoraproject.org/fedora:42 + +RUN dnf install -y \ + systemd \ + openssh-clients \ + openssh-server \ + iproute \ + iptables \ + socat \ + curl \ + ca-certificates \ + && dnf clean all \ + && systemctl enable sshd + +ENV container=docker +STOPSIGNAL SIGRTMIN+3 +CMD ["/sbin/init"] diff --git a/docker/ec2/ami-images/fedora-42/README.md b/docker/ec2/ami-images/fedora-42/README.md new file mode 100644 index 0000000..a8dd252 --- /dev/null +++ b/docker/ec2/ami-images/fedora-42/README.md @@ -0,0 +1,10 @@ +# Fedora 42 systemd EC2 guest + +Build the local guest image before launching `ami-fedora42`: + +```text +docker build --platform linux/amd64 -t floci/ami-fedora:42 docker/ec2/ami-images/fedora-42 +``` + +The image runs systemd and starts SSH on every EC2 boot. Enable application services with +`systemctl enable --now ` so they restart after `StopInstances` and `StartInstances`. diff --git a/docker/ec2/ami-images/opensuse-leap-15.6/Dockerfile b/docker/ec2/ami-images/opensuse-leap-15.6/Dockerfile new file mode 100644 index 0000000..4fe663e --- /dev/null +++ b/docker/ec2/ami-images/opensuse-leap-15.6/Dockerfile @@ -0,0 +1,4 @@ +FROM registry.opensuse.org/opensuse/leap:15.6 + +RUN zypper --non-interactive install -y openssh iproute2 iptables socat curl ca-certificates \ + && zypper clean --all diff --git a/docker/ec2/ami-images/opensuse-leap-15.6/README.md b/docker/ec2/ami-images/opensuse-leap-15.6/README.md new file mode 100644 index 0000000..d8627eb --- /dev/null +++ b/docker/ec2/ami-images/opensuse-leap-15.6/README.md @@ -0,0 +1,9 @@ +# openSUSE Leap 15.6 EC2 guest + +Build the local guest image before launching `ami-opensuse-leap156`: + +```text +docker build --platform linux/amd64 -t floci/ami-opensuse:15.6 docker/ec2/ami-images/opensuse-leap-15.6 +``` + +The image includes OpenSSH and the EC2 metadata proxy dependencies. diff --git a/docker/ec2/ami-images/ubuntu-20.04/Dockerfile b/docker/ec2/ami-images/ubuntu-20.04/Dockerfile new file mode 100644 index 0000000..7efff9f --- /dev/null +++ b/docker/ec2/ami-images/ubuntu-20.04/Dockerfile @@ -0,0 +1,19 @@ +FROM public.ecr.aws/docker/library/ubuntu:20.04 + +RUN apt-get update \ + && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ + systemd \ + systemd-sysv \ + openssh-client \ + openssh-server \ + iproute2 \ + iptables \ + socat \ + curl \ + ca-certificates \ + && systemctl enable ssh \ + && rm -rf /var/lib/apt/lists/* + +ENV container=docker +STOPSIGNAL SIGRTMIN+3 +CMD ["/sbin/init"] diff --git a/docker/ec2/ami-images/ubuntu-20.04/README.md b/docker/ec2/ami-images/ubuntu-20.04/README.md new file mode 100644 index 0000000..4e96977 --- /dev/null +++ b/docker/ec2/ami-images/ubuntu-20.04/README.md @@ -0,0 +1,10 @@ +# Ubuntu 20.04 systemd EC2 guest + +Build the local guest image before launching `ami-0abcdef1234567892`: + +```text +docker build --platform linux/amd64 -t floci/ami-ubuntu:20.04 docker/ec2/ami-images/ubuntu-20.04 +``` + +The image runs systemd and starts SSH on every EC2 boot. Enable application services with +`systemctl enable --now ` so they restart after `StopInstances` and `StartInstances`. diff --git a/docs/services/ec2.md b/docs/services/ec2.md index a681504..6c68f8d 100644 --- a/docs/services/ec2.md +++ b/docs/services/ec2.md @@ -26,14 +26,16 @@ metadata. | AMI ID | Aliases | Docker image | |---|---|---| | `ami-0abcdef1234567890` | `ami-amazonlinux2` | `public.ecr.aws/amazonlinux/amazonlinux:2` | -| `ami-0abcdef1234567891` | `ami-amazonlinux2023` | `public.ecr.aws/amazonlinux/amazonlinux:2023` | -| `ami-0abcdef1234567892` | `ami-ubuntu2004` | `public.ecr.aws/docker/library/ubuntu:20.04` | +| `ami-0abcdef1234567891` | `ami-amazonlinux2023` | `floci/ami-amazonlinux:2023` | +| `ami-0abcdef1234567892` | `ami-ubuntu2004` | `floci/ami-ubuntu:20.04` | | `ami-ubuntu2204` | | `public.ecr.aws/docker/library/ubuntu:22.04` | | `ami-ubuntu2404-arm64` | `ami-ubuntu2404` | `public.ecr.aws/docker/library/ubuntu:24.04` | | `ami-ubuntu2404-amd64` | | `public.ecr.aws/docker/library/ubuntu:24.04` | | `ami-ubuntu2404-cloud-arm64` | `ami-ubuntu2404-cloud` | `floci/ami-ubuntu:24.04-arm64` | | `ami-debian12` | | `public.ecr.aws/docker/library/debian:12` | +| `ami-debian13` | | `floci/ami-debian:13` | | `ami-alpine` | | `public.ecr.aws/docker/library/alpine:latest` | +| `ami-fedora42` | | `floci/ami-fedora:42` | | `ami-0abcdef1234567893` | | `public.ecr.aws/amazonlinux/amazonlinux:2023` | Any unrecognized AMI ID (including real AWS AMI IDs like `ami-0abc12345678`) falls back to the catalog `defaultDockerImage` (`public.ecr.aws/amazonlinux/amazonlinux:2023` by default). diff --git a/docs/services/index.md b/docs/services/index.md index 2f5a48c..9695123 100644 --- a/docs/services/index.md +++ b/docs/services/index.md @@ -65,6 +65,7 @@ Operation counts are exact. For dispatch-table services (Query and JSON 1.1) eac | [RDS Data API](rds-data.md) | `/Execute`, `/BeginTransaction`, `/CommitTransaction`, `/RollbackTransaction` | REST JSON | 4 | | [Timestream for InfluxDB](timestream-influxdb.md) | `POST /` + `X-Amz-Target: AmazonTimestreamInfluxDB.*` + InfluxDB container | JSON 1.0 + InfluxDB HTTP | 24 | | [MSK](msk.md) | `/v1/clusters/...`, `/api/v2/clusters/...` + Redpanda broker | REST JSON + Kafka | 8 | +| [MSK Connect](kafka-connect.md) | `/v1/custom-plugins`, `/v1/connectors`, `/v1/worker-configurations` + Kafka Connect worker | REST JSON + Kafka Connect | 13 | | [Amazon MQ](amazonmq.md) | `/v1/brokers/...` + RabbitMQ broker | REST JSON + AMQP | 5 | | [MWAA](mwaa.md) | `/` REST paths for environments + CLI/web proxy | REST JSON | 10 | | [Athena](athena.md) | `POST /` + `X-Amz-Target: AmazonAthena.*` | JSON 1.1 | 4 | diff --git a/docs/services/kafka-connect.md b/docs/services/kafka-connect.md new file mode 100644 index 0000000..7da8129 --- /dev/null +++ b/docs/services/kafka-connect.md @@ -0,0 +1,45 @@ +# MSK Connect + +**Protocol:** REST JSON +**Signing name:** `kafkaconnect` + +Floci implements the MSK Connect control plane separately from MSK. Custom plugins are ZIP +archives stored in Floci S3. A connector starts a pinned stock Apache Kafka Connect worker, extracts +the referenced ZIP, and mounts only that extracted plugin directory into the worker. + +## Supported Actions + +| Action | Description | +|---|---| +| `CreateCustomPlugin` | Register a ZIP custom plugin from an S3 bucket ARN and key | +| `ListCustomPlugins` | List custom plugins | +| `DescribeCustomPlugin` | Describe a custom plugin | +| `DeleteCustomPlugin` | Delete a custom plugin | +| `CreateWorkerConfiguration` | Create a worker properties configuration | +| `ListWorkerConfigurations` | List worker configurations | +| `DescribeWorkerConfiguration` | Describe a worker configuration | +| `DeleteWorkerConfiguration` | Delete a worker configuration | +| `CreateConnector` | Start a connector using a custom plugin | +| `ListConnectors` | List connectors | +| `DescribeConnector` | Describe a connector | +| `UpdateConnector` | Update connector configuration | +| `DeleteConnector` | Stop and delete a connector | + +## Worker behavior + +Workers use the internal Redpanda address of a Floci-managed MSK cluster. The public bootstrap +address supplied to `CreateConnector` is resolved to that cluster but is never passed into the +container. This keeps the worker on Floci's Docker network while host clients continue using the +public MSK bootstrap endpoint. + +No connector plugin is bundled. To run an S3 sink, upload a compatible S3 sink connector ZIP to +Floci S3, create a custom plugin for it, and use that plugin in `CreateConnector`. The worker's +container logs are streamed to CloudWatch Logs under `/aws/msk-connect/{connectorName}`. + +## Configuration + +| Variable | Default | Description | +|---|---|---| +| `FLOCI_SERVICES_KAFKACONNECT_ENABLED` | `true` | Enable or disable MSK Connect | +| `FLOCI_SERVICES_KAFKACONNECT_MOCK` | `false` | Persist control-plane metadata without starting workers | +| `FLOCI_SERVICES_KAFKACONNECT_WORKER_IMAGE` | `apache/kafka:3.9.0` | Pinned stock Kafka Connect worker image | diff --git a/docs/services/kinesis.md b/docs/services/kinesis.md index e506602..c3ad8f8 100644 --- a/docs/services/kinesis.md +++ b/docs/services/kinesis.md @@ -91,6 +91,15 @@ TPS over 10 will trigger the LimitExceededException"), independent of the ten-ca limit above. Floci does not simulate real-time request throttling for this or any other action, so that limit is intentionally unenforced here. +## Current Limitations + +- `ListStreams`, `ListShards`, `ListTagsForStream`, and `ListStreamConsumers` do not yet + support AWS pagination tokens. `ListShards` honors `MaxResults`, but does not return a + continuation token. +- Resource policy actions are not implemented. +- KMS encryption settings are recorded in stream metadata, but record payloads are not + cryptographically encrypted. + ## Configuration | Variable | Default | Description | diff --git a/mkdocs.yml b/mkdocs.yml index 74062f3..9d67a16 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -142,6 +142,7 @@ nav: - RDS Data API: services/rds-data.md - Timestream for InfluxDB: services/timestream-influxdb.md - MSK (Kafka): services/msk.md + - MSK Connect: services/kafka-connect.md - Amazon MQ: services/amazonmq.md - Glue: services/glue.md - Lake Formation: services/lakeformation.md diff --git a/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java b/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java index 4d96d8f..b5e75f1 100644 --- a/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java +++ b/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java @@ -302,6 +302,7 @@ public interface EmulatorConfig { interface ServiceStorageOverrides { SsmStorageConfig ssm(); + KafkaConnectStorageConfig kafkaconnect(); SqsStorageConfig sqs(); S3StorageConfig s3(); DynamoDbStorageConfig dynamodb(); @@ -365,6 +366,13 @@ public interface EmulatorConfig { long flushIntervalMs(); } + interface KafkaConnectStorageConfig { + Optional mode(); + + @WithDefault("5000") + long flushIntervalMs(); + } + interface SqsStorageConfig { Optional mode(); } @@ -673,6 +681,7 @@ public interface EmulatorConfig { ApiGatewayServiceConfig apigateway(); IamServiceConfig iam(); MskServiceConfig msk(); + KafkaConnectServiceConfig kafkaconnect(); AmazonMqServiceConfig amazonmq(); KinesisAnalyticsServiceConfig kinesisAnalytics(); ElastiCacheServiceConfig elasticache(); @@ -1345,6 +1354,19 @@ public interface EmulatorConfig { int kafkaHostPortMax(); } + interface KafkaConnectServiceConfig { + @WithDefault("true") + boolean enabled(); + + /** Metadata-only mode is useful where Docker is intentionally unavailable. */ + @WithDefault("false") + boolean mock(); + + /** Stock Apache Kafka distribution, never a connector-plugin image. */ + @WithDefault("apache/kafka:3.9.0") + String workerImage(); + } + interface AmazonMqServiceConfig { @WithDefault("true") boolean enabled(); diff --git a/src/main/java/io/github/hectorvent/floci/core/common/ResolvedServiceCatalog.java b/src/main/java/io/github/hectorvent/floci/core/common/ResolvedServiceCatalog.java index 5098370..8e37171 100644 --- a/src/main/java/io/github/hectorvent/floci/core/common/ResolvedServiceCatalog.java +++ b/src/main/java/io/github/hectorvent/floci/core/common/ResolvedServiceCatalog.java @@ -146,6 +146,12 @@ public class ResolvedServiceCatalog { "msk", config.storage().mode(), 5000L, null, ServiceProtocol.REST_JSON, protocols(ServiceProtocol.REST_JSON), Set.of(), Set.of("kafka"), Set.of(), Set.of(io.github.hectorvent.floci.services.msk.MskController.class)), + descriptor("kafkaconnect", "kafkaconnect", config.services().kafkaconnect().enabled(), true, + "kafkaconnect", storageMode(config.storage().services().kafkaconnect().mode(), config.storage().mode()), + config.storage().services().kafkaconnect().flushIntervalMs(), null, ServiceProtocol.REST_JSON, + protocols(ServiceProtocol.REST_JSON), + Set.of(), Set.of("kafkaconnect"), Set.of(), + Set.of(io.github.hectorvent.floci.services.kafkaconnect.KafkaConnectController.class)), descriptor("mq", "amazonmq", config.services().amazonmq().enabled(), true, "amazonmq", config.storage().mode(), 5000L, null, ServiceProtocol.REST_JSON, protocols(ServiceProtocol.REST_JSON), diff --git a/src/main/java/io/github/hectorvent/floci/core/common/docker/ContainerLifecycleManager.java b/src/main/java/io/github/hectorvent/floci/core/common/docker/ContainerLifecycleManager.java index 2285080..4699bc7 100644 --- a/src/main/java/io/github/hectorvent/floci/core/common/docker/ContainerLifecycleManager.java +++ b/src/main/java/io/github/hectorvent/floci/core/common/docker/ContainerLifecycleManager.java @@ -110,6 +110,10 @@ public class ContainerLifecycleManager { } } + public String refreshImage(String image) { + return imageCacheService.refreshImage(image); + } + /** * Creates a container without starting it. Use {@link #startCreated} to * start it after any pre-start setup (e.g. copying files into the diff --git a/src/main/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoController.java b/src/main/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoController.java index b8b4d63..3d25c79 100644 --- a/src/main/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoController.java +++ b/src/main/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoController.java @@ -6,6 +6,7 @@ import io.github.hectorvent.floci.config.FlociCertificateAuthority; import io.github.hectorvent.floci.core.common.ContainerTeardown; import io.github.hectorvent.floci.core.common.ContainerTeardowns; import io.github.hectorvent.floci.core.common.ServiceRegistry; +import io.github.hectorvent.floci.services.ec2.Ec2Service; import io.github.hectorvent.floci.lifecycle.inithook.InitializationHook; import jakarta.inject.Inject; import jakarta.ws.rs.GET; @@ -44,15 +45,17 @@ public class EmulatorInfoController { private final Instance containerTeardowns; private final FlociCertificateAuthority certificateAuthority; private final EmulatorConfig config; + private final Ec2Service ec2Service; @Inject public EmulatorInfoController(ServiceRegistry serviceRegistry, InitLifecycleState initLifecycleState, StorageFactory storageFactory, Instance resettables, - Instance containerTeardowns, - FlociCertificateAuthority certificateAuthority, - EmulatorConfig config) { + Instance containerTeardowns, + FlociCertificateAuthority certificateAuthority, + EmulatorConfig config, + Ec2Service ec2Service) { this.serviceRegistry = serviceRegistry; this.initLifecycleState = initLifecycleState; this.storageFactory = storageFactory; @@ -60,6 +63,7 @@ public class EmulatorInfoController { this.containerTeardowns = containerTeardowns; this.certificateAuthority = certificateAuthority; this.config = config; + this.ec2Service = ec2Service; this.version = resolveVersion(); } @@ -104,7 +108,7 @@ public class EmulatorInfoController { @GET @Path("/diagnose") public Response diagnose() { - return Response.ok(Map.of()).build(); + return Response.ok(Map.of("ec2", Map.of("instances", ec2Service.diagnosticInstances()))).build(); } @GET diff --git a/src/main/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployService.java b/src/main/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployService.java index 267aaad..552f0bb 100644 --- a/src/main/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployService.java +++ b/src/main/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployService.java @@ -19,8 +19,11 @@ import io.github.hectorvent.floci.services.ecs.EcsService; import io.github.hectorvent.floci.services.ecs.model.CreateTaskSetRequest; import io.github.hectorvent.floci.services.ecs.model.EcsLoadBalancer; import io.github.hectorvent.floci.services.ecs.model.EcsServiceModel; +import io.github.hectorvent.floci.services.ecs.model.LaunchType; import io.github.hectorvent.floci.services.ecs.model.TaskSet; import io.github.hectorvent.floci.services.elbv2.ElbV2Service; +import io.github.hectorvent.floci.services.elbv2.model.Action; +import io.github.hectorvent.floci.services.elbv2.model.Rule; import io.github.hectorvent.floci.services.elbv2.model.TargetGroup; import io.github.hectorvent.floci.services.elbv2.model.TargetHealth; import io.github.hectorvent.floci.services.lambda.LambdaService; @@ -63,6 +66,8 @@ public class CodeDeployService { // Lambda/ECS validation hooks must call PutLifecycleEventHookExecutionStatus within one hour. private static final Duration DEFAULT_HOOK_CALLBACK_TIMEOUT = Duration.ofHours(1); private static final Duration STOP_POLL_INTERVAL = Duration.ofMillis(200); + private static final int ECS_READINESS_MAX_POLLS = 150; + private static final int ECS_TARGET_HEALTH_MAX_POLLS = 900; @FunctionalInterface interface Sleeper { @@ -1543,6 +1548,12 @@ public class CodeDeployService { createTaskSet.setLoadBalancers(List.of(loadBalancer)); } greenTaskSet = ecsService.createTaskSet(createTaskSet, region); + verifyGreenTaskSet(region, clusterName, serviceName, deploymentId, targetService, greenTaskSet, + greenTgArn); + if (greenTaskSet.getLaunchType() == LaunchType.FARGATE) { + greenTaskSet = waitForGreenTaskSetRunning(region, clusterName, serviceName, greenTaskSet, + stopFlag); + } appendTaskSetInfo(ecsTargetMap, greenTaskSet, greenTgArn, 0.0); finishLifecycleEvent(installEvent, "Succeeded"); } catch (Exception e) { @@ -1585,9 +1596,9 @@ public class CodeDeployService { if (stopFlag.get()) { finishEcsStopped(deployment, ecsTargetMap); return; } executeEcsAllowTraffic(region, deployment, configName, ecsTargetMap, listenerArns, blueTgArn, greenTgArn, stopFlag); + verifyListenersForwardToGreen(region, listenerArns, greenTgArn); } else { - Map allowEvent = addLifecycleEvent(ecsTargetMap, "AllowTraffic"); - finishLifecycleEvent(allowEvent, "Succeeded"); + throw new IllegalStateException("ECS deployment group must have production listeners and blue/green target groups"); } if (stopFlag.get()) { finishEcsStopped(deployment, ecsTargetMap); return; } @@ -1615,10 +1626,11 @@ public class CodeDeployService { try { ecsService.deleteTaskSet(clusterName, serviceName, blueTaskSet.getTaskSetArn(), true, region); + verifyBlueTaskSetDrained(region, clusterName, serviceName, blueTaskSet.getTaskSetArn()); finishLifecycleEvent(terminateEvent, "Succeeded"); } catch (Exception e) { - LOG.debugv("Could not delete blue task set: {0}", e.getMessage()); - finishLifecycleEvent(terminateEvent, "Succeeded"); + finishLifecycleEvent(terminateEvent, "Failed"); + throw e; } } @@ -1637,18 +1649,120 @@ public class CodeDeployService { } } - /** A real target group is switched only once every registered green task is routable. */ - private void waitForGreenTargets(String region, String greenTgArn, AtomicBoolean stopFlag) + void verifyGreenTaskSet(String region, String clusterName, String serviceName, String deploymentId, + EcsServiceModel targetService, TaskSet createdTaskSet, String greenTgArn) { + if (createdTaskSet == null || createdTaskSet.getTaskSetArn() == null) { + throw new IllegalStateException("ECS did not create a green task set"); + } + List taskSets = ecsService.describeTaskSets(clusterName, serviceName, + List.of(createdTaskSet.getTaskSetArn()), region); + if (taskSets.size() != 1) { + throw new IllegalStateException("ECS did not retain the green task set"); + } + TaskSet greenTaskSet = taskSets.getFirst(); + if (!deploymentId.equals(greenTaskSet.getExternalId()) + || !"CODE_DEPLOY".equals(greenTaskSet.getStartedBy()) + || !"ACTIVE".equals(greenTaskSet.getStatus())) { + throw new IllegalStateException("ECS green task set is not a CodeDeploy active task set"); + } + if (greenTaskSet.getLaunchType() != LaunchType.FARGATE && greenTaskSet.getLaunchType() != LaunchType.EC2) { + throw new IllegalStateException("ECS green task set does not use FARGATE or EC2"); + } + if (targetService.getLaunchType() != null && greenTaskSet.getLaunchType() != targetService.getLaunchType()) { + throw new IllegalStateException("ECS green task set launch type does not match the service"); + } + if (greenTaskSet.getLaunchType() == LaunchType.EC2 + && !hasRunningTasks(greenTaskSet)) { + throw new IllegalStateException("ECS green task set has no running tasks"); + } + if (greenTgArn == null || greenTaskSet.getLoadBalancers() == null || greenTaskSet.getLoadBalancers().stream() + .noneMatch(loadBalancer -> greenTgArn.equals(loadBalancer.getTargetGroupArn()))) { + throw new IllegalStateException("ECS green task set is not registered with the green target group"); + } + } + + /** Fargate task launch is asynchronous, so the initial CreateTaskSet result is not yet stable. */ + TaskSet waitForGreenTaskSetRunning(String region, String clusterName, String serviceName, + TaskSet createdTaskSet, AtomicBoolean stopFlag) throws InterruptedException { + for (int poll = 0; poll < ECS_READINESS_MAX_POLLS && !stopFlag.get(); poll++) { + List taskSets = ecsService.describeTaskSets(clusterName, serviceName, + List.of(createdTaskSet.getTaskSetArn()), region); + if (taskSets.size() == 1 && hasRunningTasks(taskSets.getFirst())) { + return taskSets.getFirst(); + } + sleeper.sleep(STOP_POLL_INTERVAL.toMillis()); + } + if (!stopFlag.get()) { + throw new IllegalStateException("ECS green task set has no running tasks"); + } + return createdTaskSet; + } + + private boolean hasRunningTasks(TaskSet taskSet) { + return taskSet.getComputedDesiredCount() > 0 + && taskSet.getRunningCount() >= taskSet.getComputedDesiredCount(); + } + + /** A production listener is switched only once green tasks are registered and routable. */ + void waitForGreenTargets(String region, String greenTgArn, AtomicBoolean stopFlag) throws InterruptedException { - while (!stopFlag.get()) { + boolean registeredTargets = false; + for (int poll = 0; poll < ECS_TARGET_HEALTH_MAX_POLLS && !stopFlag.get(); poll++) { List targets = elbV2Service.describeTargetHealth(region, greenTgArn, null); - // Mock ECS does not start containers or register target addresses. Preserve that control-plane - // behavior while Docker-backed deployments wait for the ELB health checker. - if (targets.isEmpty() || targets.stream().allMatch(target -> "healthy".equals(target.getState()))) { + registeredTargets = !targets.isEmpty(); + if (registeredTargets && targets.stream().allMatch(target -> "healthy".equals(target.getState()))) { return; } sleeper.sleep(STOP_POLL_INTERVAL.toMillis()); } + if (!stopFlag.get()) { + throw new IllegalStateException(registeredTargets + ? "ECS green target group has unhealthy targets" + : "ECS green target group has no registered targets"); + } + } + + void verifyListenersForwardToGreen(String region, List listenerArns, String greenTgArn) { + for (String listenerArn : listenerArns) { + List defaultRules = elbV2Service.describeRules(region, listenerArn, null).stream() + .filter(Rule::isDefault) + .toList(); + if (defaultRules.size() != 1 || !forwardsOnlyTo(defaultRules.getFirst().getActions(), greenTgArn)) { + throw new IllegalStateException("Production listener did not switch to the green target group"); + } + } + } + + private boolean forwardsOnlyTo(List actions, String greenTgArn) { + if (actions == null) { + return false; + } + List forwardActions = actions.stream() + .filter(action -> "forward".equals(action.getType())) + .toList(); + return !forwardActions.isEmpty() + && forwardActions.stream().allMatch(action -> forwardsOnlyTo(action, greenTgArn)); + } + + private boolean forwardsOnlyTo(Action action, String greenTgArn) { + if (action.getTargetGroups() == null || action.getTargetGroups().isEmpty()) { + return greenTgArn.equals(action.getTargetGroupArn()); + } + boolean greenReceivesTraffic = action.getTargetGroups().stream() + .anyMatch(target -> greenTgArn.equals(target.getTargetGroupArn()) && effectiveWeight(target) > 0); + boolean otherTargetReceivesTraffic = action.getTargetGroups().stream() + .anyMatch(target -> !greenTgArn.equals(target.getTargetGroupArn()) && effectiveWeight(target) > 0); + return greenReceivesTraffic && !otherTargetReceivesTraffic; + } + + private int effectiveWeight(Action.TargetGroupTuple target) { + return target.getWeight() != null ? target.getWeight() : 1; + } + + void verifyBlueTaskSetDrained(String region, String clusterName, String serviceName, String blueTaskSetArn) { + if (!ecsService.describeTaskSets(clusterName, serviceName, List.of(blueTaskSetArn), region).isEmpty()) { + throw new IllegalStateException("ECS blue task set was not drained"); + } } void executeEcsAllowTraffic(String region, Deployment deployment, String configName, diff --git a/src/main/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineService.java b/src/main/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineService.java index d77f13c..b632080 100644 --- a/src/main/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineService.java +++ b/src/main/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineService.java @@ -23,6 +23,7 @@ import io.github.hectorvent.floci.services.codepipeline.model.CodePipelinePipeli import io.github.hectorvent.floci.services.codepipeline.model.CodePipelinePipeline.TransitionState; import io.github.hectorvent.floci.services.codepipeline.model.CodePipelineStoredItem; import io.github.hectorvent.floci.services.ecs.EcsService; +import io.github.hectorvent.floci.services.ecs.model.ContainerDefinition; import io.github.hectorvent.floci.services.ecs.model.RegisterTaskDefinitionRequest; import io.github.hectorvent.floci.services.ecs.model.TaskDefinition; import io.github.hectorvent.floci.services.lambda.LambdaService; @@ -1325,10 +1326,10 @@ public class CodePipelineService { JsonNode config = action.path("configuration"); String applicationName = requiredCodeDeployToEcsConfiguration(config, "ApplicationName"); String deploymentGroupName = requiredCodeDeployToEcsConfiguration(config, "DeploymentGroupName"); - String taskDefinitionPath = requiredCodeDeployToEcsConfiguration(config, "TaskDefinitionTemplatePath"); - String appSpecPath = requiredCodeDeployToEcsConfiguration(config, "AppSpecTemplatePath"); - String taskDefinitionTemplate = readCodeDeployToEcsTemplate(execution, action, taskDefinitionPath); - String appSpecTemplate = readCodeDeployToEcsTemplate(execution, action, appSpecPath); + String taskDefinitionTemplate = readCodeDeployToEcsTemplate(execution, action, config, + "TaskDefinitionTemplateArtifact", "TaskDefinitionTemplatePath"); + String appSpecTemplate = readCodeDeployToEcsTemplate(execution, action, config, + "AppSpecTemplateArtifact", "AppSpecTemplatePath"); RegisterTaskDefinitionRequest taskDefinitionRequest; try { @@ -1341,6 +1342,7 @@ public class CodePipelineService { throw new AwsException("InvalidActionDeclarationException", "Task definition template must specify family", 400); } + substituteCodeDeployToEcsImage(execution, action, config, taskDefinitionRequest); TaskDefinition taskDefinition = ecsService.registerTaskDefinition(taskDefinitionRequest, execution.getRegion()); String appSpec = appSpecTemplate.replace("", taskDefinition.getTaskDefinitionArn()); Map revision = Map.of("revisionType", "AppSpecContent", @@ -1351,13 +1353,25 @@ public class CodePipelineService { waitForCodeDeployDeployment(execution, state, deploymentId); } - private String readCodeDeployToEcsTemplate(CodePipelineExecution execution, JsonNode action, String reference) { - int separator = reference.indexOf("::"); - if (separator <= 0 || separator == reference.length() - 2 || reference.indexOf("::", separator + 2) >= 0) { - throw new AwsException("InvalidActionDeclarationException", - "CodeDeployToECS template paths must use ArtifactName::TemplateFilePath", 400); + private String readCodeDeployToEcsTemplate(CodePipelineExecution execution, JsonNode action, JsonNode config, + String artifactField, String pathField) { + String templatePath = requiredCodeDeployToEcsConfiguration(config, pathField); + int separator = templatePath.indexOf("::"); + if (separator >= 0) { + if (separator == 0 || separator == templatePath.length() - 2 + || templatePath.indexOf("::", separator + 2) >= 0) { + throw new AwsException("InvalidActionDeclarationException", + "CodeDeployToECS template paths must use ArtifactName::TemplateFilePath", 400); + } + return readCodeDeployToEcsTemplate(execution, action, templatePath.substring(0, separator), + templatePath.substring(separator + 2)); } - String artifactName = reference.substring(0, separator); + String artifactName = requiredCodeDeployToEcsConfiguration(config, artifactField); + return readCodeDeployToEcsTemplate(execution, action, artifactName, templatePath); + } + + private String readCodeDeployToEcsTemplate(CodePipelineExecution execution, JsonNode action, String artifactName, + String templatePath) { boolean declaredInput = false; for (JsonNode input : action.path("inputArtifacts")) { if (artifactName.equals(input.path("name").asText())) { @@ -1373,7 +1387,7 @@ public class CodePipelineService { if (artifact == null) { throw new AwsException("InvalidJobStateException", "Input artifact is not available: " + artifactName, 400); } - return readTemplateFromArtifact(artifact, reference.substring(separator + 2)); + return readTemplateFromArtifact(artifact, templatePath); } private static String requiredCodeDeployToEcsConfiguration(JsonNode config, String field) { @@ -1385,6 +1399,40 @@ public class CodePipelineService { return value; } + private void substituteCodeDeployToEcsImage(CodePipelineExecution execution, JsonNode action, JsonNode config, + RegisterTaskDefinitionRequest taskDefinitionRequest) { + String imageArtifactName = requiredCodeDeployToEcsConfiguration(config, "Image1ArtifactName"); + String containerName = requiredCodeDeployToEcsConfiguration(config, "Image1ContainerName"); + String imageDetail = readCodeDeployToEcsTemplate(execution, action, imageArtifactName, "imageDetail.json"); + JsonNode imageDetailJson; + try { + imageDetailJson = mapper.readTree(imageDetail); + } catch (java.io.IOException exception) { + throw new AwsException("InvalidActionDeclarationException", + "imageDetail.json must contain a valid image detail JSON object", 400); + } + String imageUri = imageDetailJson.path("ImageURI").asText(null); + if (imageUri == null || imageUri.isBlank()) { + throw new AwsException("InvalidActionDeclarationException", "imageDetail.json must specify ImageURI", 400); + } + if (taskDefinitionRequest.getContainerDefinitions() == null) { + throw new AwsException("InvalidActionDeclarationException", + "Task definition template must specify containerDefinitions", 400); + } + for (ContainerDefinition container : taskDefinitionRequest.getContainerDefinitions()) { + if (containerName.equals(container.getName())) { + if (container.getImage() == null) { + throw new AwsException("InvalidActionDeclarationException", + "Image1ContainerName must identify a container with an image", 400); + } + container.setImage(container.getImage().replace("", imageUri)); + return; + } + } + throw new AwsException("InvalidActionDeclarationException", + "Image1ContainerName does not match a task definition container", 400); + } + private void waitForCodeDeployDeployment(CodePipelineExecution execution, ActionExecution state, String deploymentId) throws InterruptedException { Deployment deployment = codeDeployService.getDeployment(execution.getRegion(), deploymentId); diff --git a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManager.java b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManager.java index 151ca7a..c86a122 100644 --- a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManager.java +++ b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManager.java @@ -510,12 +510,11 @@ public class Ec2ContainerManager { if (publicKey != null && !publicKey.isBlank()) { injectSshKey(containerId, publicKey); } - // sshd runs on every instance regardless of whether a key pair was supplied, - // matching real AWS AMIs (which start it as part of normal boot, independent of - // key-pair association) - starting it only when a key was present meant - // run-instances without --key-name produced a "connection refused" instead of AWS's - // actual behavior: a running daemon with simply nothing to authenticate against. - startSshd(containerId, instanceId); + // Systemd-backed AMIs start their enabled SSH unit during boot. Starting a second + // daemon here races it for port 22; lightweight guests still need this bootstrap. + if (!image.systemd()) { + startSshd(containerId, instanceId); + } // Execute UserData String userData = instance.getUserData(); @@ -1793,6 +1792,7 @@ public class Ec2ContainerManager { // and sshd is never installed. " elif command -v yum >/dev/null 2>&1; then yum install -y openssh-server openssh-clients >/dev/null 2>&1;" + " elif command -v apt-get >/dev/null 2>&1; then DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server openssh-client >/dev/null 2>&1;" + + " elif command -v zypper >/dev/null 2>&1; then zypper --non-interactive install -y openssh >/dev/null 2>&1;" + " elif command -v apk >/dev/null 2>&1; then apk add --no-cache openssh >/dev/null 2>&1;" + " fi;" + "fi;" + diff --git a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2MetadataProxy.java b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2MetadataProxy.java index e141265..629ddb4 100644 --- a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2MetadataProxy.java +++ b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2MetadataProxy.java @@ -35,6 +35,8 @@ public final class Ec2MetadataProxy { // leaving the instance without a link-local IMDS endpoint. "elif command -v yum >/dev/null 2>&1; then", " yum install -y iproute iptables socat curl ca-certificates >/dev/null", + "elif command -v zypper >/dev/null 2>&1; then", + " zypper --non-interactive install -y iproute2 iptables socat curl ca-certificates >/dev/null", "elif command -v apk >/dev/null 2>&1; then", " apk add --no-cache iproute2 iptables socat curl ca-certificates >/dev/null", "else", diff --git a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2QueryHandler.java b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2QueryHandler.java index 7c8e068..738f81b 100644 --- a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2QueryHandler.java +++ b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2QueryHandler.java @@ -702,6 +702,15 @@ public class Ec2QueryHandler { int networkInterfaceDeviceIndex = parseIntParam(p, "NetworkInterface.1.DeviceIndex", 0); String clientToken = p.getFirst("ClientToken"); List sgIds = getList(p, "SecurityGroupId"); + if (sgIds.isEmpty()) { + sgIds = getList(p, "NetworkInterface.1.SecurityGroupId"); + } + if (sgIds.isEmpty()) { + sgIds = getList(p, "NetworkInterface.1.Groups"); + } + if (sgIds.isEmpty()) { + sgIds = getList(p, "NetworkInterface.1.GroupId"); + } // UserData is base64-encoded in the wire format String userDataEncoded = p.getFirst("UserData"); diff --git a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2Service.java b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2Service.java index 7036585..8d64283 100644 --- a/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2Service.java +++ b/src/main/java/io/github/hectorvent/floci/services/ec2/Ec2Service.java @@ -5005,6 +5005,21 @@ public class Ec2Service implements ContainerTeardown, ResourceProvider { return containerManager.isContainerRunning(instance.getDockerContainerId()); } + public List> diagnosticInstances() { + return instances.scan(key -> true).stream().map(instance -> { + boolean containerRunning = isInstanceContainerRunning(instance.getInstanceId()); + String containerStatus = instance.getDockerContainerId() == null + ? "unavailable" : containerRunning ? "running" : "stopped"; + Map diagnostic = new LinkedHashMap<>(); + diagnostic.put("instanceId", instance.getInstanceId()); + diagnostic.put("sshHost", "localhost"); + diagnostic.put("sshPort", instance.getSshHostPort()); + diagnostic.put("containerId", instance.getDockerContainerId()); + diagnostic.put("containerStatus", containerStatus); + return diagnostic; + }).toList(); + } + private static void requireKeyName(String keyName) { if (keyName == null || keyName.isBlank()) { throw new AwsException("MissingParameter", "The request must contain the parameter KeyName", 400); diff --git a/src/main/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManager.java b/src/main/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManager.java index d6c9230..bdee7c4 100644 --- a/src/main/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManager.java +++ b/src/main/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManager.java @@ -226,6 +226,9 @@ public class EcsContainerManager { metadataId)); imagesByContainer.put(def, ecrRegistryManager.rewriteImageUri(def.getImage())); } + for (String image : new HashSet<>(imagesByContainer.values())) { + lifecycleManager.refreshImage(image); + } PreparedNetwork protectedNetwork = prepareNetwork(task, taskDef, region, taskId); diff --git a/src/main/java/io/github/hectorvent/floci/services/elbv2/ElbV2Service.java b/src/main/java/io/github/hectorvent/floci/services/elbv2/ElbV2Service.java index 78b11fd..f1ec4d0 100644 --- a/src/main/java/io/github/hectorvent/floci/services/elbv2/ElbV2Service.java +++ b/src/main/java/io/github/hectorvent/floci/services/elbv2/ElbV2Service.java @@ -1099,6 +1099,7 @@ public class ElbV2Service implements ResourceProvider { if (defaultRule == null) { return; } + Listener listener = requireListener(region, listenerArn); Action action = new Action(); action.setType("forward"); if (greenWeightPct >= 100) { @@ -1113,6 +1114,9 @@ public class ElbV2Service implements ResourceProvider { action.setTargetGroups(List.of(blueTuple, greenTuple)); } defaultRule.setActions(List.of(action)); + listener.setDefaultActions(List.of(action)); + persistRegion(rules, region); + persistRegion(listeners, region); dataPlane.recompileRules(listenerArn, getListenerRules(region, listenerArn)); } diff --git a/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectController.java b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectController.java new file mode 100644 index 0000000..63a7f5a --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectController.java @@ -0,0 +1,73 @@ +package io.github.hectorvent.floci.services.kafkaconnect; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectConnector; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectCustomPlugin; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectWorkerConfiguration; +import jakarta.inject.Inject; +import jakarta.ws.rs.Consumes; +import jakarta.ws.rs.DELETE; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.POST; +import jakarta.ws.rs.PUT; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.PathParam; +import jakarta.ws.rs.Produces; +import jakarta.ws.rs.core.MediaType; +import jakarta.ws.rs.core.Response; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** AWS MSK Connect REST JSON control plane. */ +@Path("/") +@Produces(MediaType.APPLICATION_JSON) +@Consumes(MediaType.APPLICATION_JSON) +public class KafkaConnectController { + private final KafkaConnectService service; + private final ObjectMapper objectMapper; + + @Inject + public KafkaConnectController(KafkaConnectService service, ObjectMapper objectMapper) { this.service = service; this.objectMapper = objectMapper; } + + @POST @Path("/v1/custom-plugins") + public Response createCustomPlugin(JsonNode request) { + JsonNode location = request.path("location").path("s3Location"); + KafkaConnectCustomPlugin plugin = service.createPlugin(text(request, "name"), text(request, "contentType"), + text(location, "bucketArn"), text(location, "fileKey"), strings(request.get("tags"))); + return Response.ok(pluginView(plugin)).build(); + } + @GET @Path("/v1/custom-plugins") public Response listCustomPlugins() { return Response.ok(Map.of("customPlugins", service.plugins().stream().map(this::pluginView).toList())).build(); } + @GET @Path("/v1/custom-plugins/{arn}") public Response describeCustomPlugin(@PathParam("arn") String arn) { return Response.ok(pluginView(service.plugin(arn))).build(); } + @DELETE @Path("/v1/custom-plugins/{arn}") public Response deleteCustomPlugin(@PathParam("arn") String arn) { service.deletePlugin(arn); return Response.ok(Map.of("customPluginArn", arn)).build(); } + + @POST @Path("/v1/worker-configurations") public Response createWorkerConfiguration(JsonNode request) { return Response.ok(workerView(service.createWorkerConfiguration(text(request, "name"), text(request, "propertiesFileContent")))).build(); } + @GET @Path("/v1/worker-configurations") public Response listWorkerConfigurations() { return Response.ok(Map.of("workerConfigurations", service.workers().stream().map(this::workerView).toList())).build(); } + @GET @Path("/v1/worker-configurations/{arn}") public Response describeWorkerConfiguration(@PathParam("arn") String arn) { return Response.ok(workerView(service.worker(arn))).build(); } + @DELETE @Path("/v1/worker-configurations/{arn}") public Response deleteWorkerConfiguration(@PathParam("arn") String arn) { service.deleteWorker(arn); return Response.ok(Map.of("workerConfigurationArn", arn)).build(); } + + @POST @Path("/v1/connectors") + public Response createConnector(JsonNode request) { + JsonNode cluster = request.path("kafkaCluster").path("apacheKafkaCluster"); + JsonNode plugin = request.path("plugins"); + String pluginArn = plugin.isArray() && !plugin.isEmpty() ? text(plugin.get(0).path("customPlugin"), "customPluginArn") : null; + JsonNode worker = request.path("workerConfiguration"); + String bootstrapServers = cluster.path("bootstrapServers").isArray() + ? cluster.path("bootstrapServers").path(0).asText(null) : text(cluster, "bootstrapServers"); + KafkaConnectConnector connector = service.createConnector(text(request, "connectorName"), bootstrapServers, pluginArn, + text(worker, "workerConfigurationArn"), strings(request.get("connectorConfiguration")), strings(request.get("tags"))); + return Response.ok(connectorView(connector)).build(); + } + @GET @Path("/v1/connectors") public Response listConnectors() { return Response.ok(Map.of("connectors", service.connectors().stream().map(this::connectorView).toList())).build(); } + @GET @Path("/v1/connectors/{arn}") public Response describeConnector(@PathParam("arn") String arn) { return Response.ok(connectorView(service.connector(arn))).build(); } + @PUT @Path("/v1/connectors/{arn}") public Response updateConnector(@PathParam("arn") String arn, JsonNode request) { return Response.ok(connectorView(service.updateConnector(arn, strings(request.get("connectorConfiguration"))))).build(); } + @DELETE @Path("/v1/connectors/{arn}") public Response deleteConnector(@PathParam("arn") String arn) { service.deleteConnector(arn); return Response.ok(Map.of("connectorArn", arn)).build(); } + + private Map pluginView(KafkaConnectCustomPlugin plugin) { return Map.of("customPluginArn", plugin.getArn(), "name", plugin.getName(), "customPluginName", plugin.getName(), "contentType", plugin.getContentType(), "creationTime", plugin.getCreationTime(), "state", "ACTIVE", "contentLocation", Map.of("s3Location", Map.of("bucketArn", plugin.getBucketArn(), "fileKey", plugin.getFileKey()))); } + private Map workerView(KafkaConnectWorkerConfiguration worker) { return Map.of("workerConfigurationArn", worker.getArn(), "name", worker.getName(), "revision", worker.getRevision(), "creationTime", worker.getCreationTime(), "propertiesFileContent", worker.getPropertiesFileContent()); } + private Map connectorView(KafkaConnectConnector connector) { Map view = new LinkedHashMap<>(); view.put("connectorArn", connector.getArn()); view.put("connectorName", connector.getName()); view.put("connectorState", connector.getState()); view.put("currentVersion", String.valueOf(connector.getCurrentVersion())); view.put("creationTime", connector.getCreationTime()); view.put("connectorConfiguration", connector.getConnectorConfiguration()); return view; } + private String text(JsonNode node, String name) { JsonNode value = node.get(name); return value == null || value.isNull() ? null : value.asText(); } + private Map strings(JsonNode node) { Map values = new LinkedHashMap<>(); if (node != null && node.isObject()) node.fields().forEachRemaining(entry -> values.put(entry.getKey(), entry.getValue().asText())); return values; } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectService.java b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectService.java new file mode 100644 index 0000000..6d35096 --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectService.java @@ -0,0 +1,125 @@ +package io.github.hectorvent.floci.services.kafkaconnect; + +import com.fasterxml.jackson.core.type.TypeReference; +import io.github.hectorvent.floci.config.EmulatorConfig; +import io.github.hectorvent.floci.core.common.AwsArnUtils; +import io.github.hectorvent.floci.core.common.AwsException; +import io.github.hectorvent.floci.core.common.RegionResolver; +import io.github.hectorvent.floci.core.storage.StorageBackend; +import io.github.hectorvent.floci.core.storage.StorageFactory; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectConnector; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectCustomPlugin; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectWorkerConfiguration; +import io.github.hectorvent.floci.services.msk.MskService; +import io.github.hectorvent.floci.services.msk.model.MskCluster; +import io.github.hectorvent.floci.services.s3.S3Service; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import org.jboss.logging.Logger; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +@ApplicationScoped +public class KafkaConnectService { + private static final Logger LOG = Logger.getLogger(KafkaConnectService.class); + private final StorageBackend plugins; + private final StorageBackend connectors; + private final StorageBackend workers; + private final RegionResolver regionResolver; + private final S3Service s3Service; + private final MskService mskService; + private final KafkaConnectWorkerManager workerManager; + private final EmulatorConfig config; + + @Inject + public KafkaConnectService(StorageFactory storageFactory, RegionResolver regionResolver, S3Service s3Service, + MskService mskService, KafkaConnectWorkerManager workerManager, EmulatorConfig config) { + this.plugins = storageFactory.create("kafkaconnect", "custom-plugins.json", new TypeReference>() { }); + this.connectors = storageFactory.create("kafkaconnect", "connectors.json", new TypeReference>() { }); + this.workers = storageFactory.create("kafkaconnect", "worker-configurations.json", new TypeReference>() { }); + this.regionResolver = regionResolver; + this.s3Service = s3Service; + this.mskService = mskService; + this.workerManager = workerManager; + this.config = config; + } + + public KafkaConnectCustomPlugin createPlugin(String name, String contentType, String bucketArn, String fileKey, Map tags) { + require(name, "customPluginName"); require(bucketArn, "contentLocation.s3Location.bucketArn"); require(fileKey, "contentLocation.s3Location.fileKey"); + if (!"JAR".equals(contentType)) { throw badRequest("contentType must be JAR"); } + String bucket = bucketName(bucketArn); + s3Service.getObject(bucket, fileKey); + KafkaConnectCustomPlugin plugin = new KafkaConnectCustomPlugin(); + plugin.setArn(arn("custom-plugin", name)); plugin.setName(name); plugin.setContentType(contentType); + plugin.setBucketArn(bucketArn); plugin.setFileKey(fileKey); plugin.setCreationTime(Instant.now()); plugin.setTags(tags); + plugins.put(plugin.getArn(), plugin); + return plugin; + } + + public KafkaConnectWorkerConfiguration createWorkerConfiguration(String name, String properties) { + require(name, "name"); require(properties, "propertiesFileContent"); + KafkaConnectWorkerConfiguration worker = new KafkaConnectWorkerConfiguration(); + worker.setArn(arn("worker-configuration", name)); worker.setName(name); worker.setPropertiesFileContent(properties); + worker.setRevision(1); worker.setCreationTime(Instant.now()); workers.put(worker.getArn(), worker); + return worker; + } + + public KafkaConnectConnector createConnector(String name, String bootstrapServers, String pluginArn, String workerArn, + Map connectorConfiguration, Map tags) { + require(name, "connectorName"); require(bootstrapServers, "kafkaCluster.apacheKafkaCluster.bootstrapServers"); require(pluginArn, "plugins"); + KafkaConnectCustomPlugin plugin = plugin(pluginArn); + MskCluster cluster = mskService.listClusters().stream() + .filter(candidate -> bootstrapServers.equals(candidate.getBootstrapBrokers())) + .findFirst().orElseThrow(() -> badRequest("bootstrapServers must name an MSK cluster managed by Floci")); + if (cluster.getInternalBootstrapBrokers() == null && !config.services().kafkaconnect().mock()) { + throw badRequest("The MSK cluster does not have an internal Redpanda bootstrap endpoint"); + } + KafkaConnectConnector connector = new KafkaConnectConnector(); + applyS3Defaults(connectorConfiguration); + connector.setArn(arn("connector", name)); connector.setName(name); connector.setState("RUNNING"); + connector.setClusterArn(cluster.getClusterArn()); connector.setPluginArn(pluginArn); connector.setWorkerConfigurationArn(workerArn); + connector.setCurrentVersion(1); connector.setCreationTime(Instant.now()); connector.setConnectorConfiguration(connectorConfiguration); connector.setTags(tags); + if (!config.services().kafkaconnect().mock()) { + String properties = workerArn == null ? null : worker(workerArn).getPropertiesFileContent(); + byte[] archive = s3Service.getObject(bucketName(plugin.getBucketArn()), plugin.getFileKey()).getData(); + workerManager.start(connector, plugin, archive, cluster.getInternalBootstrapBrokers(), properties, + regionResolver.getRegion(), regionResolver.getAccountId()); + } + connectors.put(connector.getArn(), connector); + return connector; + } + + public KafkaConnectConnector updateConnector(String arn, Map configuration) { + KafkaConnectConnector connector = connector(arn); + Map updatedConfiguration = new LinkedHashMap<>(connector.getConnectorConfiguration()); + updatedConfiguration.putAll(configuration); + applyS3Defaults(updatedConfiguration); + connector.setConnectorConfiguration(updatedConfiguration); connector.setCurrentVersion(connector.getCurrentVersion() + 1); + connectors.put(arn, connector); if (!config.services().kafkaconnect().mock()) { workerManager.configure(connector); } return connector; + } + public void deleteConnector(String arn) { connector(arn); workerManager.stop(arn); connectors.delete(arn); } + public void deletePlugin(String arn) { plugin(arn); plugins.delete(arn); } + public void deleteWorker(String arn) { worker(arn); workers.delete(arn); } + public KafkaConnectCustomPlugin plugin(String arn) { return plugins.get(arn).orElseThrow(() -> notFound("Custom plugin", arn)); } + public KafkaConnectConnector connector(String arn) { return connectors.get(arn).orElseThrow(() -> notFound("Connector", arn)); } + public KafkaConnectWorkerConfiguration worker(String arn) { return workers.get(arn).orElseThrow(() -> notFound("Worker configuration", arn)); } + public List plugins() { return plugins.scan(key -> true); } + public List connectors() { return connectors.scan(key -> true); } + public List workers() { return workers.scan(key -> true); } + private void applyS3Defaults(Map configuration) { + configuration.putIfAbsent("store.url", config.baseUrl()); + configuration.putIfAbsent("aws.access.key.id", "test"); + configuration.putIfAbsent("aws.secret.access.key", "test"); + configuration.putIfAbsent("format.class", "io.confluent.connect.s3.format.json.JsonFormat"); + configuration.putIfAbsent("storage.class", "io.confluent.connect.s3.storage.S3Storage"); + } + private String arn(String type, String name) { return regionResolver.buildArn("kafkaconnect", regionResolver.getRegion(), type + "/" + name + "/" + UUID.randomUUID()); } + private String bucketName(String bucketArn) { try { String resource = AwsArnUtils.parse(bucketArn).resource(); if (!"s3".equals(AwsArnUtils.parse(bucketArn).service()) || resource.isBlank()) throw new IllegalArgumentException(); return resource; } catch (IllegalArgumentException e) { throw badRequest("contentLocation.s3Location.bucketArn must be an S3 bucket ARN"); } } + private void require(String value, String field) { if (value == null || value.isBlank()) throw badRequest(field + " is required"); } + private AwsException badRequest(String message) { return new AwsException("BadRequestException", message, 400); } + private AwsException notFound(String type, String arn) { return new AwsException("NotFoundException", type + " " + arn + " does not exist.", 404); } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectWorkerManager.java b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectWorkerManager.java new file mode 100644 index 0000000..5e77ec9 --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectWorkerManager.java @@ -0,0 +1,197 @@ +package io.github.hectorvent.floci.services.kafkaconnect; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.github.hectorvent.floci.config.EmulatorConfig; +import io.github.hectorvent.floci.core.common.docker.ContainerBuilder; +import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager; +import io.github.hectorvent.floci.core.common.docker.ContainerLogStreamer; +import io.github.hectorvent.floci.core.common.docker.ContainerSpec; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectConnector; +import io.github.hectorvent.floci.services.kafkaconnect.model.KafkaConnectCustomPlugin; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import org.jboss.logging.Logger; +import org.apache.commons.compress.archivers.tar.TarArchiveEntry; +import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URI; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.zip.ZipInputStream; + +/** Runs stock Kafka Connect workers. Connector implementations are supplied exclusively by S3 ZIPs. */ +@ApplicationScoped +public class KafkaConnectWorkerManager { + private static final Logger LOG = Logger.getLogger(KafkaConnectWorkerManager.class); + private static final int REST_PORT = 8083; + private final ContainerBuilder containerBuilder; + private final ContainerLifecycleManager lifecycleManager; + private final ContainerLogStreamer logStreamer; + private final EmulatorConfig config; + private final ObjectMapper objectMapper; + private final Map workers = new ConcurrentHashMap<>(); + + @Inject + public KafkaConnectWorkerManager(ContainerBuilder containerBuilder, ContainerLifecycleManager lifecycleManager, + ContainerLogStreamer logStreamer, EmulatorConfig config, ObjectMapper objectMapper) { + this.containerBuilder = containerBuilder; + this.lifecycleManager = lifecycleManager; + this.logStreamer = logStreamer; + this.config = config; + this.objectMapper = objectMapper; + } + + public void start(KafkaConnectConnector connector, KafkaConnectCustomPlugin plugin, byte[] archive, + String internalBootstrap, String workerProperties, String region, String account) { + try { + Path directory = Files.createTempDirectory("floci-kafka-connect-"); + Path plugins = Files.createDirectories(directory.resolve("plugins")); + extractZip(archive, plugins); + Path properties = directory.resolve("connect-distributed.properties"); + Files.writeString(properties, distributedProperties(internalBootstrap, workerProperties)); + String name = containerName(connector.getArn()); + lifecycleManager.removeIfExists(name); + ContainerSpec spec = containerBuilder.newContainer(config.services().kafkaconnect().workerImage()) + .withName(name).withDockerNetwork(config.services().dockerNetwork()).withLogRotation() + .withDynamicPort(REST_PORT) + .withCmd(List.of("/opt/kafka/bin/connect-distributed.sh", "/opt/floci/connect-distributed.properties")) + .build(); + String containerId = lifecycleManager.create(spec); + copyWorkDirectory(containerId, directory); + ContainerLifecycleManager.ContainerInfo info = lifecycleManager.startCreated(containerId, spec); + String endpoint = "http://" + info.getEndpoint(REST_PORT).host() + ":" + info.getEndpoint(REST_PORT).port(); + Closeable logs = logStreamer.attachForAccount(account, info.containerId(), "/aws/msk-connect/" + connector.getName(), + logStreamer.generateLogStreamName(info.containerId().substring(0, 8)), region, "kafkaconnect:" + connector.getName()); + workers.put(connector.getArn(), new Worker(info.containerId(), endpoint, directory, logs)); + try { + configure(connector); + } catch (RuntimeException e) { + stop(connector.getArn()); + throw e; + } + } catch (IOException e) { + throw new IllegalStateException("Unable to prepare Kafka Connect plugin archive", e); + } + } + + public void configure(KafkaConnectConnector connector) { + Worker worker = workers.get(connector.getArn()); + if (worker == null) { + return; + } + IOException lastFailure = null; + for (int attempt = 0; attempt < 30; attempt++) { + try { + HttpURLConnection connection = (HttpURLConnection) URI.create(worker.endpoint() + "/connectors/" + connector.getName() + "/config") + .toURL().openConnection(); + connection.setRequestMethod("PUT"); + connection.setDoOutput(true); + connection.setConnectTimeout(5000); + connection.setReadTimeout(10000); + connection.setRequestProperty("Content-Type", "application/json"); + objectMapper.writeValue(connection.getOutputStream(), connector.getConnectorConfiguration()); + int responseCode = connection.getResponseCode(); + if (responseCode == 404) { + throw new IOException("Kafka Connect worker has not registered its connector API yet"); + } + if (responseCode >= 300) { + throw new IllegalStateException("Kafka Connect rejected connector configuration: " + responseCode); + } + return; + } catch (IOException e) { + lastFailure = e; + try { + Thread.sleep(1000); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while waiting for Kafka Connect worker", interrupted); + } + } + } + throw new IllegalStateException("Kafka Connect worker is not ready for connector " + connector.getName(), lastFailure); + } + + public void stop(String connectorArn) { + Worker worker = workers.remove(connectorArn); + if (worker == null) { + lifecycleManager.removeIfExists(containerName(connectorArn)); + return; + } + lifecycleManager.stopAndRemove(worker.containerId(), worker.logs()); + try { + Files.walk(worker.directory()).sorted(java.util.Comparator.reverseOrder()).forEach(path -> { + try { + Files.deleteIfExists(path); + } catch (IOException ignored) { + // A failed cleanup only leaves a disposable plugin extraction directory. + } + }); + } catch (IOException e) { + LOG.debugv("Could not remove Kafka Connect work directory {0}: {1}", worker.directory(), e.getMessage()); + } + } + + private String distributedProperties(String bootstrap, String extra) { + return "bootstrap.servers=" + bootstrap + "\n" + + "group.id=floci-kafka-connect\nconfig.storage.topic=floci-connect-configs\n" + + "offset.storage.topic=floci-connect-offsets\nstatus.storage.topic=floci-connect-status\n" + + "config.storage.replication.factor=1\noffset.storage.replication.factor=1\n" + + "status.storage.replication.factor=1\nkey.converter=org.apache.kafka.connect.json.JsonConverter\n" + + "value.converter=org.apache.kafka.connect.json.JsonConverter\nplugin.path=/opt/floci/plugins\n" + + (extra == null ? "" : extra + "\n"); + } + + private String containerName(String connectorArn) { + return "floci-kafkaconnect-" + connectorArn.substring(connectorArn.lastIndexOf('/') + 1); + } + + private void extractZip(byte[] archive, Path output) throws IOException { + try (ZipInputStream zip = new ZipInputStream(new java.io.ByteArrayInputStream(archive))) { + java.util.zip.ZipEntry entry; + while ((entry = zip.getNextEntry()) != null) { + Path target = output.resolve(entry.getName()).normalize(); + if (!target.startsWith(output)) { + throw new IOException("Plugin ZIP contains an invalid path"); + } + if (entry.isDirectory()) { Files.createDirectories(target); } + else { Files.createDirectories(target.getParent()); Files.copy(zip, target); } + } + } + } + + private void copyWorkDirectory(String containerId, Path directory) throws IOException { + ByteArrayOutputStream archive = new ByteArrayOutputStream(); + try (TarArchiveOutputStream tar = new TarArchiveOutputStream(archive); + java.util.stream.Stream paths = Files.walk(directory)) { + tar.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX); + tar.setBigNumberMode(TarArchiveOutputStream.BIGNUMBER_POSIX); + for (Path path : paths.sorted(Comparator.naturalOrder()).toList()) { + String relative = directory.relativize(path).toString(); + if (relative.isEmpty()) { + continue; + } + TarArchiveEntry entry = new TarArchiveEntry(path.toFile(), "floci/" + relative); + tar.putArchiveEntry(entry); + if (Files.isRegularFile(path)) { + Files.copy(path, tar); + } + tar.closeArchiveEntry(); + } + } + lifecycleManager.getDockerClient().copyArchiveToContainerCmd(containerId) + .withRemotePath("/opt") + .withTarInputStream(new ByteArrayInputStream(archive.toByteArray())) + .exec(); + } + + private record Worker(String containerId, String endpoint, Path directory, Closeable logs) { } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectConnector.java b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectConnector.java new file mode 100644 index 0000000..6b293b2 --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectConnector.java @@ -0,0 +1,41 @@ +package io.github.hectorvent.floci.services.kafkaconnect.model; + +import io.quarkus.runtime.annotations.RegisterForReflection; +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.Map; + +@RegisterForReflection +public class KafkaConnectConnector { + private String arn; + private String name; + private String state; + private String clusterArn; + private String pluginArn; + private String workerConfigurationArn; + private long currentVersion; + private Instant creationTime; + private Map connectorConfiguration = new LinkedHashMap<>(); + private Map tags = new LinkedHashMap<>(); + + public String getArn() { return arn; } + public void setArn(String arn) { this.arn = arn; } + public String getName() { return name; } + public void setName(String name) { this.name = name; } + public String getState() { return state; } + public void setState(String state) { this.state = state; } + public String getClusterArn() { return clusterArn; } + public void setClusterArn(String clusterArn) { this.clusterArn = clusterArn; } + public String getPluginArn() { return pluginArn; } + public void setPluginArn(String pluginArn) { this.pluginArn = pluginArn; } + public String getWorkerConfigurationArn() { return workerConfigurationArn; } + public void setWorkerConfigurationArn(String workerConfigurationArn) { this.workerConfigurationArn = workerConfigurationArn; } + public long getCurrentVersion() { return currentVersion; } + public void setCurrentVersion(long currentVersion) { this.currentVersion = currentVersion; } + public Instant getCreationTime() { return creationTime; } + public void setCreationTime(Instant creationTime) { this.creationTime = creationTime; } + public Map getConnectorConfiguration() { return connectorConfiguration; } + public void setConnectorConfiguration(Map connectorConfiguration) { this.connectorConfiguration = connectorConfiguration; } + public Map getTags() { return tags; } + public void setTags(Map tags) { this.tags = tags; } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectCustomPlugin.java b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectCustomPlugin.java new file mode 100644 index 0000000..a787ffc --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectCustomPlugin.java @@ -0,0 +1,32 @@ +package io.github.hectorvent.floci.services.kafkaconnect.model; + +import io.quarkus.runtime.annotations.RegisterForReflection; +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.Map; + +@RegisterForReflection +public class KafkaConnectCustomPlugin { + private String arn; + private String name; + private String contentType; + private String bucketArn; + private String fileKey; + private Instant creationTime; + private Map tags = new LinkedHashMap<>(); + + public String getArn() { return arn; } + public void setArn(String arn) { this.arn = arn; } + public String getName() { return name; } + public void setName(String name) { this.name = name; } + public String getContentType() { return contentType; } + public void setContentType(String contentType) { this.contentType = contentType; } + public String getBucketArn() { return bucketArn; } + public void setBucketArn(String bucketArn) { this.bucketArn = bucketArn; } + public String getFileKey() { return fileKey; } + public void setFileKey(String fileKey) { this.fileKey = fileKey; } + public Instant getCreationTime() { return creationTime; } + public void setCreationTime(Instant creationTime) { this.creationTime = creationTime; } + public Map getTags() { return tags; } + public void setTags(Map tags) { this.tags = tags; } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectWorkerConfiguration.java b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectWorkerConfiguration.java new file mode 100644 index 0000000..f0b8dd4 --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/kafkaconnect/model/KafkaConnectWorkerConfiguration.java @@ -0,0 +1,24 @@ +package io.github.hectorvent.floci.services.kafkaconnect.model; + +import io.quarkus.runtime.annotations.RegisterForReflection; +import java.time.Instant; + +@RegisterForReflection +public class KafkaConnectWorkerConfiguration { + private String arn; + private String name; + private String propertiesFileContent; + private long revision; + private Instant creationTime; + + public String getArn() { return arn; } + public void setArn(String arn) { this.arn = arn; } + public String getName() { return name; } + public void setName(String name) { this.name = name; } + public String getPropertiesFileContent() { return propertiesFileContent; } + public void setPropertiesFileContent(String propertiesFileContent) { this.propertiesFileContent = propertiesFileContent; } + public long getRevision() { return revision; } + public void setRevision(long revision) { this.revision = revision; } + public Instant getCreationTime() { return creationTime; } + public void setCreationTime(Instant creationTime) { this.creationTime = creationTime; } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandler.java b/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandler.java index 656a55e..5e4ef55 100644 --- a/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandler.java +++ b/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandler.java @@ -110,7 +110,11 @@ public class KinesisJsonHandler { private Response handleCreateStream(JsonNode request, String region) { String streamName = request.path("StreamName").asText(); - int shardCount = request.path("ShardCount").asInt(1); + JsonNode shardCountNode = request.path("ShardCount"); + boolean hasShardCount = !shardCountNode.isMissingNode() && !shardCountNode.isNull(); + if (hasShardCount && (!shardCountNode.isIntegralNumber() || !shardCountNode.canConvertToInt())) { + throw new AwsException("InvalidArgumentException", "ShardCount must be an integer.", 400); + } String streamMode = null; JsonNode modeDetails = request.path("StreamModeDetails"); if (modeDetails.isObject()) { @@ -119,6 +123,18 @@ public class KinesisJsonHandler { streamMode = mode; } } + if (hasShardCount && streamMode != null) { + throw new AwsException("InvalidArgumentException", + "Specify either ShardCount or StreamModeDetails, not both.", 400); + } + if (!hasShardCount && streamMode == null) { + throw new AwsException("InvalidArgumentException", + "Either ShardCount or StreamModeDetails must be specified.", 400); + } + if (hasShardCount && shardCountNode.intValue() < 1) { + throw new AwsException("InvalidArgumentException", "ShardCount must be at least 1.", 400); + } + int shardCount = hasShardCount ? shardCountNode.intValue() : 1; // CreateStream's optional Tags member was being dropped. That is invisible to a // hand-written script but not to Terraform: aws_kinesis_stream sets tags at create // time and then reads them back with ListTagsForStream, so an empty read produces a @@ -633,6 +649,15 @@ public class KinesisJsonHandler { String shardId = request.path("ShardId").asText(); String type = request.path("ShardIteratorType").asText(); String seq = request.has("StartingSequenceNumber") ? request.path("StartingSequenceNumber").asText() : null; + if (!Set.of("AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "AT_TIMESTAMP", "LATEST", "TRIM_HORIZON") + .contains(type)) { + throw new AwsException("InvalidArgumentException", "Invalid ShardIteratorType: " + type, 400); + } + if (("AT_SEQUENCE_NUMBER".equals(type) || "AFTER_SEQUENCE_NUMBER".equals(type)) + && (seq == null || seq.isBlank())) { + throw new AwsException("InvalidArgumentException", + "ShardIteratorType " + type + " requires a StartingSequenceNumber", 400); + } // AWS sends Timestamp as epoch seconds (double with fractional ms). // Convert to long millis at the boundary; the emulator stores time in ms everywhere. // Use Math.round to avoid 1ms drift from FP multiplication (e.g. X.999...). @@ -661,6 +686,9 @@ public class KinesisJsonHandler { private Response handleGetRecords(JsonNode request, String region) { String iterator = request.path("ShardIterator").asText(); Integer limit = request.has("Limit") ? request.path("Limit").asInt() : null; + if (limit != null && (limit < 1 || limit > 10000)) { + throw new AwsException("InvalidArgumentException", "Limit must be between 1 and 10000.", 400); + } Map result = service.getRecords(iterator, limit, region); List records = (List) result.get("Records"); diff --git a/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisService.java b/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisService.java index 0ce6b79..e0bd105 100644 --- a/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisService.java +++ b/src/main/java/io/github/hectorvent/floci/services/kinesis/KinesisService.java @@ -906,7 +906,7 @@ public class KinesisService implements ResourceProvider { } } - int max = limit != null ? Math.min(limit, 1000) : 1000; + int max = limit != null ? Math.min(limit, 10000) : 10000; List result = new ArrayList<>(); int nextIndex = startIndex; for (int i = startIndex; i < allRecords.size() && result.size() < max; i++) { diff --git a/src/main/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheService.java b/src/main/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheService.java index 6a7bb0a..1166cbc 100644 --- a/src/main/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheService.java +++ b/src/main/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheService.java @@ -68,25 +68,43 @@ public class ImageCacheService { LOG.infov("Image already present locally, skipping pull: {0}", imageUri); return resolvedImage; } - LOG.infov("Pulling image: {0}", imageUri); - try { - runWithRetry(imageUri, MAX_PULL_ATTEMPTS, INITIAL_BACKOFF_MS, () -> { - PullImageCmd pullImage = dockerClient.pullImageCmd(imageUri) - .withAuthConfig(resolveAuth(imageUri)); - if (explicitPlatform) { - pullImage.withPlatform(requestedPlatform); - } - pullImage.exec(new PullImageResultCallback()) - .awaitCompletion(5, TimeUnit.MINUTES); - }); - resolvedImage = resolvedImageReference(imageUri, inspectLocalImage(imageUri)); - resolvedImages.put(imageKey, resolvedImage); - LOG.infov("Image pulled successfully: {0}", imageUri); - return resolvedImage; - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException("Interrupted while pulling image: " + imageUri, e); - } + return pullImage(imageUri, requestedPlatform, explicitPlatform, imageKey); + } + } + + public String refreshImage(String imageUri) { + return refreshImage(imageUri, null); + } + + public String refreshImage(String imageUri, String platform) { + boolean explicitPlatform = platform != null && !platform.isBlank(); + String requestedPlatform = explicitPlatform ? platform.trim() : daemonPlatform(); + ImageKey imageKey = new ImageKey(imageUri, requestedPlatform); + Object lock = locks.computeIfAbsent(imageUri, k -> new Object()); + synchronized (lock) { + return pullImage(imageUri, requestedPlatform, explicitPlatform, imageKey); + } + } + + private String pullImage(String imageUri, String requestedPlatform, boolean explicitPlatform, ImageKey imageKey) { + LOG.infov("Pulling image: {0}", imageUri); + try { + runWithRetry(imageUri, MAX_PULL_ATTEMPTS, INITIAL_BACKOFF_MS, () -> { + PullImageCmd pullImage = dockerClient.pullImageCmd(imageUri) + .withAuthConfig(resolveAuth(imageUri)); + if (explicitPlatform) { + pullImage.withPlatform(requestedPlatform); + } + pullImage.exec(new PullImageResultCallback()) + .awaitCompletion(5, TimeUnit.MINUTES); + }); + String resolvedImage = resolvedImageReference(imageUri, inspectLocalImage(imageUri)); + resolvedImages.put(imageKey, resolvedImage); + LOG.infov("Image pulled successfully: {0}", imageUri); + return resolvedImage; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while pulling image: " + imageUri, e); } } diff --git a/src/main/java/io/github/hectorvent/floci/services/msk/MskController.java b/src/main/java/io/github/hectorvent/floci/services/msk/MskController.java index 936e46d..cac12dd 100644 --- a/src/main/java/io/github/hectorvent/floci/services/msk/MskController.java +++ b/src/main/java/io/github/hectorvent/floci/services/msk/MskController.java @@ -91,8 +91,15 @@ public class MskController { @GET @Path("/v1/clusters/{clusterArn}/bootstrap-brokers") - public Response getBootstrapBrokers(@PathParam("clusterArn") String clusterArn) { + public Response getBootstrapBrokers(@PathParam("clusterArn") String clusterArn, + @HeaderParam("X-Floci-Internal-Client") String internalClient) { String bootstrapBrokers = mskService.getBootstrapBrokers(clusterArn); + if ("true".equals(internalClient)) { + MskCluster cluster = mskService.describeCluster(clusterArn); + if (cluster.getInternalBootstrapBrokers() != null) { + bootstrapBrokers = cluster.getInternalBootstrapBrokers(); + } + } return Response.ok(Map.of("bootstrapBrokerString", bootstrapBrokers)).build(); } diff --git a/src/main/java/io/github/hectorvent/floci/services/msk/RedpandaManager.java b/src/main/java/io/github/hectorvent/floci/services/msk/RedpandaManager.java index 605e8e0..d83fd19 100644 --- a/src/main/java/io/github/hectorvent/floci/services/msk/RedpandaManager.java +++ b/src/main/java/io/github/hectorvent/floci/services/msk/RedpandaManager.java @@ -7,7 +7,6 @@ import io.github.hectorvent.floci.core.common.docker.ContainerBuilder; import io.github.hectorvent.floci.core.common.docker.ContainerDetector; import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager; import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager.ContainerInfo; -import io.github.hectorvent.floci.core.common.docker.ContainerLifecycleManager.EndpointInfo; import io.github.hectorvent.floci.core.common.docker.ContainerLogStreamer; import io.github.hectorvent.floci.core.common.docker.ContainerSpec; import io.github.hectorvent.floci.core.common.docker.ContainerStorageHelper; @@ -32,6 +31,7 @@ public class RedpandaManager { private static final Logger LOG = Logger.getLogger(RedpandaManager.class); private static final int KAFKA_PORT = 9092; + private static final int INTERNAL_KAFKA_PORT = 9093; static final int ADMIN_PORT = 9644; // Redpanda's Admin API exposes readiness at /v1/status/ready. /ready always returns 404 @@ -124,28 +124,22 @@ public class RedpandaManager { // reachable address up front. The host port has to be known before the // container starts (the flag is a startup arg), so in native mode we // pre-allocate it ourselves instead of letting Docker assign one dynamically. - String kafkaAdvertiseAddr; - int kafkaHostPort = 0; - if (!containerDetector.isRunningInContainer()) { - kafkaHostPort = portAllocator.allocate( - config.services().msk().kafkaHostPortBase(), - config.services().msk().kafkaHostPortMax()); - kafkaAdvertiseAddr = "localhost:" + kafkaHostPort; - } else { - // Sibling containers on the docker network resolve the broker via its - // container name through Docker's embedded DNS. - kafkaAdvertiseAddr = containerName + ":" + KAFKA_PORT; - } + int kafkaHostPort = portAllocator.allocate( + config.services().msk().kafkaHostPortBase(), + config.services().msk().kafkaHostPortMax()); + String externalKafkaAdvertiseAddr = "external://localhost:" + kafkaHostPort; + String internalKafkaAdvertiseAddr = "internal://" + containerName + ":" + INTERNAL_KAFKA_PORT; // Build command List cmd = new ArrayList<>(List.of( "redpanda", "start", "--overprovisioned", "--smp", "1", "--memory", "512M", "--reserve-memory", "0M", - "--advertise-kafka-addr", kafkaAdvertiseAddr)); + "--kafka-addr", "internal://0.0.0.0:" + INTERNAL_KAFKA_PORT + ",external://0.0.0.0:" + KAFKA_PORT, + "--advertise-kafka-addr", internalKafkaAdvertiseAddr + "," + externalKafkaAdvertiseAddr)); - // Build container spec. Publish Kafka/admin ports to the host only in - // native mode; in Docker mode producers/consumers reach the broker via - // the docker network IP resolved from container inspect. + // MSK clients run outside Floci's Docker network as often as inside it. Publish and + // advertise the Kafka listener in both topologies so GetBootstrapBrokers is usable by + // real host clients, not merely sibling containers. ContainerBuilder.Builder specBuilder = containerBuilder.newContainer(image) .withName(containerName) .withDockerNetwork(config.services().dockerNetwork()) @@ -155,11 +149,7 @@ public class RedpandaManager { AwsArnUtils.accountOrDefault(cluster.getClusterArn(), regionResolver.getAccountId()), AwsArnUtils.regionOrDefault(cluster.getClusterArn(), regionResolver.getDefaultRegion()))); - if (!containerDetector.isRunningInContainer()) { - specBuilder.withPortBinding(KAFKA_PORT, kafkaHostPort).withDynamicPort(ADMIN_PORT); - } else { - specBuilder.withExposedPort(KAFKA_PORT).withExposedPort(ADMIN_PORT); - } + specBuilder.withPortBinding(KAFKA_PORT, kafkaHostPort).withDynamicPort(ADMIN_PORT); // Handle persistence mounting if (ContainerStorageHelper.isNamedVolumeMode(config)) { @@ -183,17 +173,14 @@ public class RedpandaManager { try { info = lifecycleManager.createAndStart(spec); } catch (RuntimeException e) { - if (!containerDetector.isRunningInContainer()) { - portAllocator.release(kafkaHostPort); - } + portAllocator.release(kafkaHostPort); throw e; } cluster.setContainerId(info.containerId()); // Resolve endpoints - EndpointInfo kafkaEndpoint = info.getEndpoint(KAFKA_PORT); - - cluster.setBootstrapBrokers(kafkaEndpoint.host() + ":" + kafkaEndpoint.port()); + cluster.setBootstrapBrokers("localhost:" + kafkaHostPort); + cluster.setInternalBootstrapBrokers(containerName + ":" + INTERNAL_KAFKA_PORT); LOG.infov("Redpanda container {0} started. Bootstrap: {1}", info.containerId(), cluster.getBootstrapBrokers()); // Attach log streaming (new feature) @@ -231,7 +218,13 @@ public class RedpandaManager { return false; } } else { - String containerIp = bootstrap.split(":")[0]; + String containerIp = lifecycleManager.getDockerClient().inspectContainerCmd(cluster.getContainerId()) + .exec().getNetworkSettings().getNetworks().values().stream() + .map(network -> network.getIpAddress()).filter(ip -> ip != null && !ip.isBlank()) + .findFirst().orElse(null); + if (containerIp == null) { + return false; + } adminUrl = "http://" + containerIp + ":" + ADMIN_PORT + ADMIN_READY_PATH; } diff --git a/src/main/java/io/github/hectorvent/floci/services/msk/model/MskCluster.java b/src/main/java/io/github/hectorvent/floci/services/msk/model/MskCluster.java index 79dc645..34e8fee 100644 --- a/src/main/java/io/github/hectorvent/floci/services/msk/model/MskCluster.java +++ b/src/main/java/io/github/hectorvent/floci/services/msk/model/MskCluster.java @@ -89,6 +89,10 @@ public class MskCluster { @JsonProperty("bootstrapBrokers") private String bootstrapBrokers; + // Broker address for Floci services sharing the Docker network. + @JsonProperty("internalBootstrapBrokers") + private String internalBootstrapBrokers; + // Docker container ID for mock=false @JsonProperty("containerId") private String containerId; @@ -178,6 +182,11 @@ public class MskCluster { public String getBootstrapBrokers() { return bootstrapBrokers; } public void setBootstrapBrokers(String bootstrapBrokers) { this.bootstrapBrokers = bootstrapBrokers; } + public String getInternalBootstrapBrokers() { return internalBootstrapBrokers; } + public void setInternalBootstrapBrokers(String internalBootstrapBrokers) { + this.internalBootstrapBrokers = internalBootstrapBrokers; + } + public String getContainerId() { return containerId; } public void setContainerId(String containerId) { this.containerId = containerId; } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 1f5a380..cb6b997 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -155,6 +155,8 @@ floci: services: ssm: flush-interval-ms: 5000 + kafkaconnect: + flush-interval-ms: 5000 dynamodb: flush-interval-ms: 5000 sns: @@ -357,6 +359,10 @@ floci: default-image: "redpandadata/redpanda:latest" kafka-host-port-base: 9300 kafka-host-port-max: 9399 + kafkaconnect: + enabled: true + mock: false + worker-image: apache/kafka:3.9.0 amazonmq: enabled: true mock: false diff --git a/src/main/resources/ec2/image-catalog.yaml b/src/main/resources/ec2/image-catalog.yaml index 66ef560..38a98df 100644 --- a/src/main/resources/ec2/image-catalog.yaml +++ b/src/main/resources/ec2/image-catalog.yaml @@ -35,13 +35,14 @@ images: - imageId: ami-0abcdef1234567892 aliases: - ami-ubuntu2004 - dockerImage: public.ecr.aws/docker/library/ubuntu:20.04 + dockerImage: floci/ami-ubuntu:20.04 name: ubuntu/images/hvm-ssd/ubuntu-focal-20.04-amd64-server-20230324 description: Canonical, Ubuntu, 20.04 LTS ownerId: "099720109477" imageOwnerAlias: canonical architecture: x86_64 creationDate: "2023-03-24T00:00:00.000Z" + guestRuntime: systemd - imageId: ami-ubuntu2204 dockerImage: public.ecr.aws/docker/library/ubuntu:22.04 @@ -95,6 +96,14 @@ images: architecture: x86_64 creationDate: "2023-04-11T00:00:00.000Z" + - imageId: ami-debian13 + dockerImage: floci/ami-debian:13 + name: debian-13-amd64 + description: Debian 13 AMI with SSH + architecture: x86_64 + creationDate: "2026-10-06T00:00:00.000Z" + guestRuntime: systemd + - imageId: ami-alpine dockerImage: public.ecr.aws/docker/library/alpine:latest name: alpine-linux-latest-x86_64 @@ -103,14 +112,15 @@ images: creationDate: "2023-04-01T00:00:00.000Z" - imageId: ami-fedora42 - dockerImage: registry.fedoraproject.org/fedora:42 + dockerImage: floci/ami-fedora:42 name: Fedora-Cloud-Base-42-x86_64 description: Fedora Linux 42 AMI architecture: x86_64 creationDate: "2025-04-15T00:00:00.000Z" + guestRuntime: systemd - imageId: ami-opensuse-leap156 - dockerImage: registry.opensuse.org/opensuse/leap:15.6 + dockerImage: floci/ami-opensuse:15.6 name: openSUSE-Leap-15.6-x86_64 description: openSUSE Leap 15.6 AMI architecture: x86_64 diff --git a/src/test/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoControllerTest.java b/src/test/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoControllerTest.java index 1023f69..ece447e 100644 --- a/src/test/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoControllerTest.java +++ b/src/test/java/io/github/hectorvent/floci/lifecycle/EmulatorInfoControllerTest.java @@ -1,6 +1,7 @@ package io.github.hectorvent.floci.lifecycle; import io.github.hectorvent.floci.config.EmulatorConfig; +import io.github.hectorvent.floci.services.ec2.Ec2Service; import io.github.hectorvent.floci.config.FlociCertificateAuthority; import io.github.hectorvent.floci.core.common.ContainerTeardown; import io.github.hectorvent.floci.core.common.Resettable; @@ -36,6 +37,7 @@ class EmulatorInfoControllerTest { @Mock private Instance containerTeardowns; @Mock private FlociCertificateAuthority certificateAuthority; @Mock private EmulatorConfig config; + @Mock private Ec2Service ec2Service; @Mock private ContainerTeardown sageMakerTeardown; @Mock private ContainerTeardown batchTeardown; @Mock private Resettable resettable; @@ -45,7 +47,7 @@ class EmulatorInfoControllerTest { @BeforeEach void setUp() { controller = new EmulatorInfoController(serviceRegistry, initLifecycleState, storageFactory, - resettables, containerTeardowns, certificateAuthority, config); + resettables, containerTeardowns, certificateAuthority, config, ec2Service); } @Test diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudwatch/logs/CloudWatchLogsMetricFilterResetTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudwatch/logs/CloudWatchLogsMetricFilterResetTest.java index f7698f9..8ac2afd 100644 --- a/src/test/java/io/github/hectorvent/floci/services/cloudwatch/logs/CloudWatchLogsMetricFilterResetTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/cloudwatch/logs/CloudWatchLogsMetricFilterResetTest.java @@ -115,6 +115,6 @@ class CloudWatchLogsMetricFilterResetTest { when(resettables.iterator()).thenAnswer(call -> List.of(resettable).iterator()); Instance teardowns = mock(Instance.class); when(teardowns.iterator()).thenAnswer(call -> List.of().iterator()); - return new EmulatorInfoController(null, null, f.factory, resettables, teardowns, null, null); + return new EmulatorInfoController(null, null, f.factory, resettables, teardowns, null, null, null); } } diff --git a/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployEcsIntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployEcsIntegrationTest.java index bdba45b..13c7929 100644 --- a/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployEcsIntegrationTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployEcsIntegrationTest.java @@ -281,7 +281,7 @@ class CodeDeployEcsIntegrationTest { @Test @Order(11) - void deploymentCompletesSuccessfully() throws InterruptedException { + void deploymentFailsWithoutRealGreenTargets() throws InterruptedException { Assumptions.assumeTrue(deploymentId != null, "deploymentId must be set"); String status = null; @@ -302,7 +302,7 @@ class CodeDeployEcsIntegrationTest { TimeUnit.MILLISECONDS.sleep(500); } - Assertions.assertEquals("Succeeded", status, "ECS deployment should succeed"); + Assertions.assertEquals("Failed", status, "ECS deployment without green targets must fail"); } @Test @@ -350,14 +350,14 @@ class CodeDeployEcsIntegrationTest { .then() .statusCode(200) .body("deploymentTargets[0].deploymentTargetType", equalTo("ECSTarget")) - .body("deploymentTargets[0].ecsTarget.status", equalTo("Succeeded")); + .body("deploymentTargets[0].ecsTarget.status", equalTo("Failed")); } @Test @Order(14) - void listenerNowPointsToGreen() { + void listenerRemainsOnBlueAfterFailedDeployment() { Assumptions.assumeTrue(listenerArn != null, "listenerArn must be set"); - Assumptions.assumeTrue(greenTgArn != null, "greenTgArn must be set"); + Assumptions.assumeTrue(blueTgArn != null, "blueTgArn must be set"); given() .contentType(ELBV2_CONTENT_TYPE) @@ -369,6 +369,6 @@ class CodeDeployEcsIntegrationTest { .then() .statusCode(200) .body("DescribeRulesResponse.DescribeRulesResult.Rules.member.Actions.member.TargetGroupArn", - equalTo(greenTgArn)); + equalTo(blueTgArn)); } } diff --git a/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployServiceTest.java b/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployServiceTest.java index d55a62a..2d30612 100644 --- a/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployServiceTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/codedeploy/CodeDeployServiceTest.java @@ -8,7 +8,15 @@ import io.github.hectorvent.floci.services.ec2.Ec2Service; import io.github.hectorvent.floci.services.ec2.model.Instance; import io.github.hectorvent.floci.services.ec2.model.Reservation; import io.github.hectorvent.floci.services.ecs.EcsService; +import io.github.hectorvent.floci.services.ecs.model.EcsLoadBalancer; +import io.github.hectorvent.floci.services.ecs.model.EcsServiceModel; +import io.github.hectorvent.floci.services.ecs.model.LaunchType; +import io.github.hectorvent.floci.services.ecs.model.TaskSet; import io.github.hectorvent.floci.services.elbv2.ElbV2Service; +import io.github.hectorvent.floci.services.elbv2.model.Action; +import io.github.hectorvent.floci.services.elbv2.model.Listener; +import io.github.hectorvent.floci.services.elbv2.model.Rule; +import io.github.hectorvent.floci.services.elbv2.model.TargetHealth; import io.github.hectorvent.floci.services.lambda.LambdaService; import io.github.hectorvent.floci.services.lambda.model.LambdaAlias; import io.github.hectorvent.floci.services.lambda.model.LambdaFunction; @@ -32,6 +40,7 @@ import java.util.concurrent.atomic.AtomicReference; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; @@ -494,6 +503,247 @@ class CodeDeployServiceTest { verify(elbV2Service).shiftListenerForward(REGION, "listener", "blue", "green", 100); } + @Test + void ecsGreenTaskSetWaitsForFargateTasksToStart() throws InterruptedException { + EcsService ecsService = mock(EcsService.class); + AtomicInteger sleeps = new AtomicInteger(); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService, + mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + ignored -> sleeps.incrementAndGet()); + TaskSet pending = greenTaskSet(); + pending.setRunningCount(0); + TaskSet running = greenTaskSet(); + when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)) + .thenReturn(List.of(pending), List.of(running)); + + TaskSet actual = ecsDeploymentService.waitForGreenTaskSetRunning(REGION, "cluster", "service", pending, + new java.util.concurrent.atomic.AtomicBoolean()); + + assertEquals(running, actual); + assertEquals(1, sleeps.get()); + } + + @Test + void ecsGreenTaskSetFailsAfterFargateStartupTimeout() { + EcsService ecsService = mock(EcsService.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService, + mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + ignored -> { }); + TaskSet pending = greenTaskSet(); + pending.setRunningCount(0); + when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(pending)); + + IllegalStateException exception = assertThrows(IllegalStateException.class, + () -> ecsDeploymentService.waitForGreenTaskSetRunning(REGION, "cluster", "service", pending, + new java.util.concurrent.atomic.AtomicBoolean())); + + assertEquals("ECS green task set has no running tasks", exception.getMessage()); + } + + @Test + void ecsGreenTaskSetMustMatchServiceLaunchTypeAndTargetGroup() { + EcsService ecsService = mock(EcsService.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService, + mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + Thread::sleep); + TaskSet green = greenTaskSet(); + green.setLaunchType(LaunchType.EC2); + when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(green)); + + assertThrows(IllegalStateException.class, () -> ecsDeploymentService.verifyGreenTaskSet(REGION, + "cluster", "service", "deployment", fargateService(), green, "green-target-group")); + } + + @Test + void ecsGreenTaskSetAcceptsTheExistingEc2Path() { + EcsService ecsService = mock(EcsService.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService, + mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + Thread::sleep); + TaskSet green = greenTaskSet(); + green.setLaunchType(LaunchType.EC2); + EcsServiceModel ec2Service = new EcsServiceModel(); + ec2Service.setLaunchType(LaunchType.EC2); + when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(green)); + + ecsDeploymentService.verifyGreenTaskSet(REGION, "cluster", "service", "deployment", ec2Service, + green, "green-target-group"); + } + + @Test + void ecsGreenTaskSetRejectsEc2WithoutRunningTasksImmediately() { + EcsService ecsService = mock(EcsService.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, ecsService, + mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + ignored -> { }); + TaskSet green = greenTaskSet(); + green.setLaunchType(LaunchType.EC2); + green.setRunningCount(0); + EcsServiceModel ec2Service = new EcsServiceModel(); + ec2Service.setLaunchType(LaunchType.EC2); + when(ecsService.describeTaskSets("cluster", "service", List.of("green"), REGION)).thenReturn(List.of(green)); + + IllegalStateException exception = assertThrows(IllegalStateException.class, + () -> ecsDeploymentService.verifyGreenTaskSet(REGION, "cluster", "service", "deployment", + ec2Service, green, "green-target-group")); + + assertEquals("ECS green task set has no running tasks", exception.getMessage()); + } + + @Test + void ecsDeploymentVerifiesTheAuthoritativeDefaultRuleInsteadOfStaleListenerActions() { + ElbV2Service elbV2Service = mock(ElbV2Service.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class), + elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + Thread::sleep); + Listener staleListener = new Listener(); + staleListener.setDefaultActions(List.of(forwardAction("blue-target-group"))); + when(elbV2Service.describeListeners(REGION, null, List.of("listener"))).thenReturn(List.of(staleListener)); + when(elbV2Service.describeRules(REGION, "listener", null)) + .thenReturn(List.of(defaultRule(forwardAction("green-target-group")))); + + ecsDeploymentService.verifyListenersForwardToGreen(REGION, List.of("listener"), "green-target-group"); + + verify(elbV2Service, never()).describeListeners(anyString(), any(), any()); + } + + @Test + void ecsDeploymentAcceptsWeightedDefaultRuleWithAllTrafficOnGreen() { + ElbV2Service elbV2Service = mock(ElbV2Service.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class), + elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + Thread::sleep); + Action weightedForward = new Action(); + weightedForward.setType("forward"); + weightedForward.setTargetGroups(List.of(targetGroup("blue-target-group", 0), + targetGroup("green-target-group", 100))); + when(elbV2Service.describeRules(REGION, "listener", null)) + .thenReturn(List.of(defaultRule(weightedForward))); + + ecsDeploymentService.verifyListenersForwardToGreen(REGION, List.of("listener"), "green-target-group"); + } + + @Test + void ecsDeploymentRejectsDefaultRuleWithResidualBlueTraffic() { + ElbV2Service elbV2Service = mock(ElbV2Service.class); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class), + elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + Thread::sleep); + Action weightedForward = new Action(); + weightedForward.setType("forward"); + weightedForward.setTargetGroups(List.of(targetGroup("blue-target-group", 1), + targetGroup("green-target-group", 99))); + when(elbV2Service.describeRules(REGION, "listener", null)) + .thenReturn(List.of(defaultRule(weightedForward))); + + IllegalStateException exception = assertThrows(IllegalStateException.class, + () -> ecsDeploymentService.verifyListenersForwardToGreen(REGION, + List.of("listener"), "green-target-group")); + + assertEquals("Production listener did not switch to the green target group", exception.getMessage()); + } + + @Test + void ecsDeploymentRejectsAnEmptyGreenTargetGroup() { + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class), + mock(ElbV2Service.class), ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + ignored -> { }); + + assertThrows(IllegalStateException.class, () -> ecsDeploymentService.waitForGreenTargets(REGION, + "green-target-group", new java.util.concurrent.atomic.AtomicBoolean())); + } + + @Test + void ecsDeploymentWaitsForHealthyGreenTargets() throws InterruptedException { + ElbV2Service elbV2Service = mock(ElbV2Service.class); + AtomicInteger sleeps = new AtomicInteger(); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class), + elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + ignored -> sleeps.incrementAndGet()); + TargetHealth target = new TargetHealth(); + target.setState("healthy"); + when(elbV2Service.describeTargetHealth(REGION, "green-target-group", null)) + .thenReturn(List.of(), List.of(target)); + + ecsDeploymentService.waitForGreenTargets(REGION, "green-target-group", + new java.util.concurrent.atomic.AtomicBoolean()); + + assertEquals(1, sleeps.get()); + } + + @Test + void ecsDeploymentWaitsForDefaultTargetGroupHealthThreshold() throws InterruptedException { + ElbV2Service elbV2Service = mock(ElbV2Service.class); + AtomicInteger polls = new AtomicInteger(); + CodeDeployService ecsDeploymentService = new CodeDeployService(lambdaService, mock(EcsService.class), + elbV2Service, ssmCommandService, mock(Ec2Service.class), new ObjectMapper(), + new RegionResolver(REGION, ACCOUNT_ID), null, Duration.ofHours(1), Clock.systemUTC(), null, + ignored -> { }); + TargetHealth initial = new TargetHealth(); + initial.setState("initial"); + TargetHealth healthy = new TargetHealth(); + healthy.setState("healthy"); + when(elbV2Service.describeTargetHealth(REGION, "green-target-group", null)) + .thenAnswer(ignored -> List.of(polls.incrementAndGet() > 150 ? healthy : initial)); + + ecsDeploymentService.waitForGreenTargets(REGION, "green-target-group", + new java.util.concurrent.atomic.AtomicBoolean()); + + assertEquals(151, polls.get()); + } + + private TaskSet greenTaskSet() { + EcsLoadBalancer loadBalancer = new EcsLoadBalancer(); + loadBalancer.setTargetGroupArn("green-target-group"); + TaskSet taskSet = new TaskSet(); + taskSet.setTaskSetArn("green"); + taskSet.setExternalId("deployment"); + taskSet.setStartedBy("CODE_DEPLOY"); + taskSet.setStatus("ACTIVE"); + taskSet.setLaunchType(LaunchType.FARGATE); + taskSet.setComputedDesiredCount(1); + taskSet.setRunningCount(1); + taskSet.setLoadBalancers(List.of(loadBalancer)); + return taskSet; + } + + private EcsServiceModel fargateService() { + EcsServiceModel ecsService = new EcsServiceModel(); + ecsService.setLaunchType(LaunchType.FARGATE); + return ecsService; + } + + private Rule defaultRule(Action action) { + Rule rule = new Rule(); + rule.setDefault(true); + rule.setActions(List.of(action)); + return rule; + } + + private Action forwardAction(String targetGroupArn) { + Action action = new Action(); + action.setType("forward"); + action.setTargetGroupArn(targetGroupArn); + return action; + } + + private Action.TargetGroupTuple targetGroup(String targetGroupArn, int weight) { + Action.TargetGroupTuple target = new Action.TargetGroupTuple(); + target.setTargetGroupArn(targetGroupArn); + target.setWeight(weight); + return target; + } + // ---- Helpers -------------------------------------------------------------------------- private String createServerDeployment(String appName, String groupName, int timeoutSeconds) { diff --git a/src/test/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineServiceTest.java b/src/test/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineServiceTest.java index b172fd3..e7ec822 100644 --- a/src/test/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineServiceTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/codepipeline/CodePipelineServiceTest.java @@ -394,14 +394,17 @@ class CodePipelineServiceTest { } @Test - void codeDeployToEcsRegistersTaskDefinitionAndSubstitutesAppSpec() throws Exception { + void codeDeployToEcsSubstitutesFargateTaskDefinitionImageFromImageDetail() throws Exception { S3Service s3Service = mock(S3Service.class); CodeDeployService codeDeployService = mock(CodeDeployService.class); EcsService ecsService = mock(EcsService.class); S3Object source = new S3Object(); source.setData(zipArtifact(Map.of( "taskdef.json", """ - {"family":"pipeline-task","containerDefinitions":[{"name":"app","image":"nginx"}]} + {"family":"pipeline-task","networkMode":"awsvpc","requiresCompatibilities":["FARGATE"],"containerDefinitions":[{"name":"app","image":""}]} + """, + "imageDetail.json", """ + {"ImageURI":"000000000000.dkr.ecr.us-east-1.amazonaws.com/pipeline-app:build-5"} """, "appspec.json", """ {"version":0.0,"Resources":[{"TargetService":{"Type":"AWS::ECS::Service","Properties":{"TaskDefinition":"","LoadBalancerInfo":{"ContainerName":"app","ContainerPort":80}}}}]} @@ -429,6 +432,9 @@ class CodePipelineServiceTest { ArgumentCaptor request = ArgumentCaptor.forClass(RegisterTaskDefinitionRequest.class); verify(ecsService).registerTaskDefinition(request.capture(), eq(REGION)); assertEquals("pipeline-task", request.getValue().getFamily()); + assertTrue(request.getValue().isFargate()); + assertEquals("000000000000.dkr.ecr.us-east-1.amazonaws.com/pipeline-app:build-5", + request.getValue().getContainerDefinitions().getFirst().getImage()); ArgumentCaptor> revision = ArgumentCaptor.forClass(Map.class); verify(codeDeployService).createDeployment(eq(REGION), eq("ecs-app"), eq("ecs-group"), eq(null), revision.capture(), anyString()); @@ -447,7 +453,8 @@ class CodePipelineServiceTest { EcsService ecsService = mock(EcsService.class); S3Object source = new S3Object(); source.setData(zipArtifact(Map.of( - "taskdef.json", "{\"family\":\"pipeline-task\",\"containerDefinitions\":[]}", + "taskdef.json", "{\"family\":\"pipeline-task\",\"containerDefinitions\":[{\"name\":\"app\",\"image\":\"\"}]}", + "imageDetail.json", "{\"ImageURI\":\"000000000000.dkr.ecr.us-east-1.amazonaws.com/pipeline-app:build-5\"}", "appspec.json", "{\"version\":0.0,\"Resources\":[{\"TargetService\":{\"Properties\":{\"TaskDefinition\":\"\"}}}]}"))); source.setETag("source-etag"); source.setLastModified(Instant.now()); @@ -964,8 +971,12 @@ class CodePipelineServiceTest { "configuration": { "ApplicationName": "ecs-app", "DeploymentGroupName": "ecs-group", - "TaskDefinitionTemplatePath": "SourceOutput::taskdef.json", - "AppSpecTemplatePath": "SourceOutput::appspec.json" + "TaskDefinitionTemplateArtifact": "SourceOutput", + "TaskDefinitionTemplatePath": "taskdef.json", + "AppSpecTemplateArtifact": "SourceOutput", + "AppSpecTemplatePath": "appspec.json", + "Image1ArtifactName": "SourceOutput", + "Image1ContainerName": "app" }, "inputArtifacts": [{"name": "SourceOutput"}] }] diff --git a/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManagerTest.java b/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManagerTest.java index db102d2..64726e1 100644 --- a/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManagerTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ContainerManagerTest.java @@ -854,6 +854,14 @@ class Ec2ContainerManagerTest { "dnf must be probed before yum"); } + @Test + void sshdInstallProbeHandlesOpenSuse() { + String script = Ec2ContainerManager.sshdInstallProbeCommand()[2]; + + assertTrue(script.contains("command -v zypper"), script); + assertTrue(script.contains("zypper --non-interactive install -y openssh"), script); + } + @Test void metadataProxyInstallCommandHandlesYumForAmazonLinux2() { // The IMDS proxy dependency install had the same gap, and unlike the sshd probe it @@ -866,6 +874,15 @@ class Ec2ContainerManagerTest { assertTrue(script.contains("yum install -y iproute iptables socat curl ca-certificates"), script); } + @Test + void metadataProxyInstallCommandHandlesOpenSuse() { + String script = Ec2ContainerManager.metadataProxyInstallCommand()[2]; + + assertTrue(script.contains("command -v zypper"), script); + assertTrue(script.contains("zypper --non-interactive install -y iproute2 iptables socat curl ca-certificates"), + script); + } + @Test void metadataProxyInstallCommandLetsDnfReplaceCurlMinimalWithCurl() { // public.ecr.aws/amazonlinux/amazonlinux:2023 -- the fallback image AmiImageResolver @@ -1354,9 +1371,10 @@ class Ec2ContainerManagerTest { private static final String[] SSHD_INSTALL_PROBE_CMD = {"sh", "-c", "if ! command -v sshd >/dev/null 2>&1 || ! command -v scp >/dev/null 2>&1; then" + " if command -v dnf >/dev/null 2>&1; then dnf install -y openssh-server openssh-clients >/dev/null 2>&1;" - + " elif command -v yum >/dev/null 2>&1; then yum install -y openssh-server openssh-clients >/dev/null 2>&1;" - + " elif command -v apt-get >/dev/null 2>&1; then DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server openssh-client >/dev/null 2>&1;" - + " elif command -v apk >/dev/null 2>&1; then apk add --no-cache openssh >/dev/null 2>&1;" + + " elif command -v yum >/dev/null 2>&1; then yum install -y openssh-server openssh-clients >/dev/null 2>&1;" + + " elif command -v apt-get >/dev/null 2>&1; then DEBIAN_FRONTEND=noninteractive apt-get install -y openssh-server openssh-client >/dev/null 2>&1;" + + " elif command -v zypper >/dev/null 2>&1; then zypper --non-interactive install -y openssh >/dev/null 2>&1;" + + " elif command -v apk >/dev/null 2>&1; then apk add --no-cache openssh >/dev/null 2>&1;" + " fi;" + "fi;" + "command -v sshd >/dev/null 2>&1 || exit 1;" diff --git a/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ImageCatalogTest.java b/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ImageCatalogTest.java index cfe06b6..6f7a7d5 100644 --- a/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ImageCatalogTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2ImageCatalogTest.java @@ -37,6 +37,7 @@ class Ec2ImageCatalogTest { "ami-ubuntu2404-amd64", "ami-ubuntu2404-cloud-arm64", "ami-debian12", + "ami-debian13", "ami-alpine", "ami-fedora42", "ami-opensuse-leap156"), imageIds); @@ -78,9 +79,14 @@ class Ec2ImageCatalogTest { } @Test - void resolverUsesFedoraAndOpenSuseImages() { - assertEquals("registry.fedoraproject.org/fedora:42", amiImageResolver.resolve("ami-fedora42")); - assertEquals("registry.opensuse.org/opensuse/leap:15.6", amiImageResolver.resolve("ami-opensuse-leap156")); + void resolverUsesFedoraAndLocalSshReadyImages() { + assertEquals("floci/ami-ubuntu:20.04", amiImageResolver.resolve("ami-0abcdef1234567892")); + assertEquals("floci/ami-debian:13", amiImageResolver.resolve("ami-debian13")); + assertEquals("floci/ami-fedora:42", amiImageResolver.resolve("ami-fedora42")); + assertEquals("floci/ami-opensuse:15.6", amiImageResolver.resolve("ami-opensuse-leap156")); + assertTrue(amiImageResolver.resolveImage("ami-0abcdef1234567892").systemd()); + assertTrue(amiImageResolver.resolveImage("ami-debian13").systemd()); + assertTrue(amiImageResolver.resolveImage("ami-fedora42").systemd()); } @Test diff --git a/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2IntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2IntegrationTest.java index 4693a3f..1e97b65 100644 --- a/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2IntegrationTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/ec2/Ec2IntegrationTest.java @@ -2876,6 +2876,26 @@ class Ec2IntegrationTest { .extract().path("RunInstancesResponse.instancesSet.item.instanceId"); } + @Test + @Order(80) + void runInstancesUsesSecurityGroupsFromNetworkInterface() { + given() + .formParam("Action", "RunInstances") + .formParam("ImageId", "ami-0abcdef1234567890") + .formParam("InstanceType", "t2.micro") + .formParam("MinCount", "1") + .formParam("MaxCount", "1") + .formParam("NetworkInterface.1.DeviceIndex", "0") + .formParam("NetworkInterface.1.SubnetId", subnetId) + .formParam("NetworkInterface.1.SecurityGroupId.1", securityGroupId) + .header("Authorization", AUTH_HEADER) + .when() + .post("/") + .then() + .statusCode(200) + .body("RunInstancesResponse.instancesSet.item.groupSet.item.groupId", equalTo(securityGroupId)); + } + @Test @Order(80) void runInstancesWithArm64ImageDescribesArm64Architecture() { diff --git a/src/test/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManagerEcrRewriteTest.java b/src/test/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManagerEcrRewriteTest.java index d362436..88debcb 100644 --- a/src/test/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManagerEcrRewriteTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/ecs/container/EcsContainerManagerEcrRewriteTest.java @@ -108,6 +108,8 @@ class EcsContainerManagerEcrRewriteTest { List images = imageCaptor.getAllValues(); assertEquals(REWRITTEN_IMAGE, images.get(0), "ECR-shaped image should be rewritten for the docker pull"); assertEquals("sidecar:latest", images.get(1), "non-ECR image should pass through unchanged"); + verify(lifecycleManager).refreshImage(REWRITTEN_IMAGE); + verify(lifecycleManager).refreshImage("sidecar:latest"); // RunTask/DescribeTasks must keep reporting the original task-def image, not the rewrite. List containers = task.getContainers(); diff --git a/src/test/java/io/github/hectorvent/floci/services/elasticache/ElastiCacheIntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/elasticache/ElastiCacheIntegrationTest.java index ad48ebe..02affc3 100644 --- a/src/test/java/io/github/hectorvent/floci/services/elasticache/ElastiCacheIntegrationTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/elasticache/ElastiCacheIntegrationTest.java @@ -300,6 +300,23 @@ class ElastiCacheIntegrationTest { @Test @Order(13) + void keyExpiresAtTheConfiguredTtl() throws Exception { + try (Socket socket = openSocket(firstProxyPort)) { + write(socket, respArray("AUTH", GROUP_AUTH_TOKEN)); + assertEquals("+OK\r\n", readLine(socket)); + + write(socket, respArray("SET", "expires-after-one-second", "value", "EX", "1")); + assertEquals("+OK\r\n", readLine(socket)); + + Thread.sleep(1_100); + + write(socket, respArray("GET", "expires-after-one-second")); + assertEquals("$-1\r\n", readLine(socket)); + } + } + + @Test + @Order(14) void deleteUserRemovesUserFromDescribeUsers() { given() .formParam("Action", "DeleteUser") @@ -322,7 +339,7 @@ class ElastiCacheIntegrationTest { } @Test - @Order(14) + @Order(15) void deleteReplicationGroupReleasesProxyPortForReuse() { given() .formParam("Action", "DeleteReplicationGroup") diff --git a/src/test/java/io/github/hectorvent/floci/services/elbv2/ElbV2ServiceTest.java b/src/test/java/io/github/hectorvent/floci/services/elbv2/ElbV2ServiceTest.java index fdc3d2f..00a2fed 100644 --- a/src/test/java/io/github/hectorvent/floci/services/elbv2/ElbV2ServiceTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/elbv2/ElbV2ServiceTest.java @@ -225,6 +225,35 @@ class ElbV2ServiceTest { .get(rule.getRuleArn()).containsKey("env")); } + @Test + void shiftListenerForwardPersistsTheGreenTargetGroup() { + SharedStorageFactory storageFactory = new SharedStorageFactory(); + ElbV2Service first = serviceWithStorage(storageFactory, mock(ElbV2DataPlane.class), mock(ElbV2HealthChecker.class)); + String lbArn = first.createLoadBalancer( + REGION, "traffic-lb", "internal", "application", "ipv4", + ALB_SUBNETS, List.of("sg-a"), Map.of()).getLoadBalancerArn(); + String blueTargetGroupArn = first.createTargetGroup( + REGION, "traffic-blue", "HTTP", "HTTP1", 8080, "vpc-a", "ip", + "HTTP", "traffic-port", true, "/", 30, 5, 5, 2, "200", "ipv4", Map.of()).getTargetGroupArn(); + String greenTargetGroupArn = first.createTargetGroup( + REGION, "traffic-green", "HTTP", "HTTP1", 8080, "vpc-a", "ip", + "HTTP", "traffic-port", true, "/", 30, 5, 5, 2, "200", "ipv4", Map.of()).getTargetGroupArn(); + String listenerArn = first.createListener( + REGION, lbArn, "HTTP", 9080, null, List.of(), + List.of(forwardAction(blueTargetGroupArn)), List.of(), Map.of()).getListenerArn(); + + first.shiftListenerForward(REGION, listenerArn, blueTargetGroupArn, greenTargetGroupArn, 100); + + ElbV2Service reloaded = serviceWithStorage(storageFactory, mock(ElbV2DataPlane.class), mock(ElbV2HealthChecker.class)); + Rule defaultRule = reloaded.describeRules(REGION, listenerArn, null).stream() + .filter(Rule::isDefault) + .findFirst() + .orElseThrow(); + assertEquals(greenTargetGroupArn, defaultRule.getActions().getFirst().getTargetGroupArn()); + assertEquals(greenTargetGroupArn, reloaded.describeListeners(REGION, lbArn, List.of(listenerArn)).getFirst() + .getDefaultActions().getFirst().getTargetGroupArn()); + } + @Test void initializeStorageDoesNotTouchCollaborators() { // #1913: ElbV2DataPlane.binding() calls back into this service, and during @PostConstruct the diff --git a/src/test/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectIntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectIntegrationTest.java new file mode 100644 index 0000000..b254813 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectIntegrationTest.java @@ -0,0 +1,60 @@ +package io.github.hectorvent.floci.services.kafkaconnect; + +import io.github.hectorvent.floci.services.msk.MskService; +import io.github.hectorvent.floci.services.msk.model.CreateClusterRequest; +import io.github.hectorvent.floci.services.msk.model.MskCluster; +import io.github.hectorvent.floci.services.s3.S3Service; +import jakarta.inject.Inject; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.util.Map; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; + +import static io.restassured.RestAssured.given; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.notNullValue; + +@io.quarkus.test.junit.QuarkusTest +class KafkaConnectIntegrationTest { + @Inject S3Service s3Service; + @Inject MskService mskService; + + @Test + void customPluginWorkerConfigurationAndConnectorPersistThroughAwsRoutes() throws Exception { + s3Service.createBucket("plugins", null); + s3Service.putObject("plugins", "sink.zip", zip(), "application/zip", Map.of()); + CreateClusterRequest clusterRequest = new CreateClusterRequest(); + clusterRequest.setClusterName("connect-cluster"); + MskCluster cluster = mskService.createCluster(clusterRequest); + + String pluginArn = given().contentType("application/json") + .body(""" + {"customPluginName":"s3-sink","contentType":"ZIP","contentLocation":{"s3Location":{"bucketArn":"arn:aws:s3:::plugins","fileKey":"sink.zip"}}} + """) + .when().post("/v1/custom-plugins").then().statusCode(200) + .body("customPluginArn", notNullValue()).extract().path("customPluginArn"); + String workerArn = given().contentType("application/json") + .body("{\"name\":\"worker\",\"propertiesFileContent\":\"offset.flush.interval.ms=1000\"}") + .when().post("/v1/worker-configurations").then().statusCode(200) + .body("revision", equalTo(1)).extract().path("workerConfigurationArn"); + + given().contentType("application/json") + .body(""" + {"connectorName":"sink","connectorConfiguration":{"connector.class":"example.Sink","topics":"events"},"kafkaCluster":{"apacheKafkaCluster":{"bootstrapServers":["%s"]}},"plugins":[{"customPlugin":{"customPluginArn":"%s"}}],"workerConfiguration":{"workerConfigurationArn":"%s"}} + """.formatted(cluster.getBootstrapBrokers(), pluginArn, workerArn)) + .when().post("/v1/connectors").then().statusCode(200).body("connectorState", equalTo("RUNNING")); + given().when().get("/v1/connectors").then().statusCode(200).body("connectors.size()", equalTo(1)); + } + + private byte[] zip() throws Exception { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (ZipOutputStream zip = new ZipOutputStream(output)) { + zip.putNextEntry(new ZipEntry("README.txt")); + zip.write("plugin".getBytes(java.nio.charset.StandardCharsets.UTF_8)); + zip.closeEntry(); + } + return output.toByteArray(); + } +} diff --git a/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisIntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisIntegrationTest.java index c09e467..b472cf3 100644 --- a/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisIntegrationTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisIntegrationTest.java @@ -584,7 +584,7 @@ class KinesisIntegrationTest { @Test @Order(31) - void createStreamWithOnDemandMode() { + void createStreamRejectsShardCountAndOnDemandModeTogether() { given() .header("X-Amz-Target", "Kinesis_20131202.CreateStream") .contentType(KINESIS_CONTENT_TYPE) @@ -594,19 +594,8 @@ class KinesisIntegrationTest { .when() .post("/") .then() - .statusCode(200); - - given() - .header("X-Amz-Target", "Kinesis_20131202.DescribeStream") - .contentType(KINESIS_CONTENT_TYPE) - .body(""" - {"StreamName": "on-demand-create-test"} - """) - .when() - .post("/") - .then() - .statusCode(200) - .body("StreamDescription.StreamModeDetails.StreamMode", equalTo("ON_DEMAND")); + .statusCode(400) + .body("__type", equalTo("InvalidArgumentException")); } @Test diff --git a/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandlerTest.java b/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandlerTest.java index fc754f7..a9db1ca 100644 --- a/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandlerTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/kinesis/KinesisJsonHandlerTest.java @@ -339,7 +339,6 @@ class KinesisJsonHandlerTest { void createStreamHonorsOnDemandStreamMode() { ObjectNode req = MAPPER.createObjectNode(); req.put("StreamName", "test-stream"); - req.put("ShardCount", 1); req.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND"); assertThat(handler.handle("CreateStream", req, REGION).getStatus(), is(200)); @@ -1249,7 +1248,6 @@ class KinesisJsonHandlerTest { void updateMaxRecordSizeRejectsOnDemandStreams() { ObjectNode create = MAPPER.createObjectNode(); create.put("StreamName", "on-demand-stream"); - create.put("ShardCount", 1); create.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND"); assertThat(handler.handle("CreateStream", create, REGION).getStatus(), is(200)); @@ -1363,7 +1361,6 @@ class KinesisJsonHandlerTest { void updateShardCountRejectsOnDemandStream() { ObjectNode create = MAPPER.createObjectNode(); create.put("StreamName", "on-demand-stream"); - create.put("ShardCount", 2); create.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND"); assertThat(handler.handle("CreateStream", create, REGION).getStatus(), is(200)); @@ -1584,6 +1581,76 @@ class KinesisJsonHandlerTest { assertEquals(Map.of("Foo", "Bar"), service.listTagsForStream("test-stream", REGION)); } + @Test + void createStreamRequiresExactlyOneCapacitySetting() { + ObjectNode missingCapacity = MAPPER.createObjectNode(); + missingCapacity.put("StreamName", "missing-capacity"); + AwsException missingException = assertThrows(AwsException.class, + () -> handler.handle("CreateStream", missingCapacity, REGION)); + assertEquals("InvalidArgumentException", missingException.getErrorCode()); + + ObjectNode bothCapacitySettings = MAPPER.createObjectNode(); + bothCapacitySettings.put("StreamName", "both-capacity-settings"); + bothCapacitySettings.put("ShardCount", 1); + bothCapacitySettings.putObject("StreamModeDetails").put("StreamMode", "ON_DEMAND"); + AwsException bothException = assertThrows(AwsException.class, + () -> handler.handle("CreateStream", bothCapacitySettings, REGION)); + assertEquals("InvalidArgumentException", bothException.getErrorCode()); + + ObjectNode zeroShards = MAPPER.createObjectNode(); + zeroShards.put("StreamName", "zero-shards"); + zeroShards.put("ShardCount", 0); + AwsException zeroException = assertThrows(AwsException.class, + () -> handler.handle("CreateStream", zeroShards, REGION)); + assertEquals("InvalidArgumentException", zeroException.getErrorCode()); + } + + @Test + void getShardIteratorValidatesIteratorTypeAndSequenceNumber() { + createStream("test-stream"); + + ObjectNode invalidType = MAPPER.createObjectNode(); + invalidType.put("StreamName", "test-stream"); + invalidType.put("ShardId", FIRST_SHARD_ID); + invalidType.put("ShardIteratorType", "INVALID"); + AwsException invalidTypeException = assertThrows(AwsException.class, + () -> handler.handle("GetShardIterator", invalidType, REGION)); + assertEquals("InvalidArgumentException", invalidTypeException.getErrorCode()); + + ObjectNode missingSequence = MAPPER.createObjectNode(); + missingSequence.put("StreamName", "test-stream"); + missingSequence.put("ShardId", FIRST_SHARD_ID); + missingSequence.put("ShardIteratorType", "AT_SEQUENCE_NUMBER"); + AwsException missingSequenceException = assertThrows(AwsException.class, + () -> handler.handle("GetShardIterator", missingSequence, REGION)); + assertEquals("InvalidArgumentException", missingSequenceException.getErrorCode()); + } + + @Test + void getRecordsAcceptsKinesisMaximumLimitAndRejectsOutOfRangeValues() { + createStream("test-stream"); + ObjectNode iteratorRequest = MAPPER.createObjectNode(); + iteratorRequest.put("StreamName", "test-stream"); + iteratorRequest.put("ShardId", FIRST_SHARD_ID); + iteratorRequest.put("ShardIteratorType", "TRIM_HORIZON"); + String iterator = responseEntity(handler.handle("GetShardIterator", iteratorRequest, REGION)) + .get("ShardIterator").asText(); + + ObjectNode maximumLimit = MAPPER.createObjectNode(); + maximumLimit.put("ShardIterator", iterator); + maximumLimit.put("Limit", 10000); + assertThat(handler.handle("GetRecords", maximumLimit, REGION).getStatus(), is(200)); + + for (int invalidLimit : List.of(0, 10001)) { + ObjectNode invalidLimitRequest = MAPPER.createObjectNode(); + invalidLimitRequest.put("ShardIterator", iterator); + invalidLimitRequest.put("Limit", invalidLimit); + AwsException exception = assertThrows(AwsException.class, + () -> handler.handle("GetRecords", invalidLimitRequest, REGION)); + assertEquals("InvalidArgumentException", exception.getErrorCode()); + } + } + /** Every non-string tag value shape — number, boolean, null, object, array — plus a map mixing a valid and an invalid value. */ private static List nonStringTagValues() { ObjectNode number = MAPPER.createObjectNode(); diff --git a/src/test/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheServiceTest.java b/src/test/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheServiceTest.java index 54ebdc3..a4dc5d0 100644 --- a/src/test/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheServiceTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/lambda/launcher/ImageCacheServiceTest.java @@ -284,6 +284,34 @@ class ImageCacheServiceTest { verify(pullImage).exec(any(PullImageResultCallback.class)); } + @Test + void refreshesMutableTagWhileCachedImageStillExists() throws Exception { + DockerClient dockerClient = mock(DockerClient.class); + InspectImageCmd inspectImage = mock(InspectImageCmd.class); + InspectImageCmd inspectCachedImage = mock(InspectImageCmd.class); + PullImageCmd pullImage = mock(PullImageCmd.class); + PullImageResultCallback callback = mock(PullImageResultCallback.class); + when(dockerClient.inspectImageCmd(IMAGE)).thenReturn(inspectImage); + when(dockerClient.inspectImageCmd("sha256:old")).thenReturn(inspectCachedImage); + when(dockerClient.inspectImageCmd("sha256:new")).thenReturn(inspectCachedImage); + when(inspectImage.exec()).thenReturn( + new InspectImageResponse().withId("sha256:old").withOs("linux").withArch("amd64"), + new InspectImageResponse().withId("sha256:new").withOs("linux").withArch("amd64")); + when(inspectCachedImage.exec()).thenReturn( + new InspectImageResponse().withId("sha256:old").withOs("linux").withArch("amd64"), + new InspectImageResponse().withId("sha256:new").withOs("linux").withArch("amd64")); + when(dockerClient.pullImageCmd(IMAGE)).thenReturn(pullImage); + when(pullImage.withAuthConfig(any())).thenReturn(pullImage); + when(pullImage.exec(any(PullImageResultCallback.class))).thenReturn(callback); + + ImageCacheService service = newService(dockerClient); + assertEquals("sha256:old", service.ensureImageExists(IMAGE)); + assertEquals("sha256:new", service.refreshImage(IMAGE)); + assertEquals("sha256:new", service.ensureImageExists(IMAGE)); + + verify(pullImage).exec(any(PullImageResultCallback.class)); + } + @Test void succeedsOnFirstAttempt() throws Exception { AtomicInteger calls = new AtomicInteger(); diff --git a/src/test/java/io/github/hectorvent/floci/services/msk/RedpandaManagerTest.java b/src/test/java/io/github/hectorvent/floci/services/msk/RedpandaManagerTest.java index 543febc..a41d428 100644 --- a/src/test/java/io/github/hectorvent/floci/services/msk/RedpandaManagerTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/msk/RedpandaManagerTest.java @@ -43,7 +43,6 @@ import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -177,14 +176,15 @@ class RedpandaManagerTest { int flagIndex = spec.cmd().indexOf("--advertise-kafka-addr"); assertTrue(flagIndex >= 0, "cmd should contain --advertise-kafka-addr"); - assertEquals("localhost:54321", spec.cmd().get(flagIndex + 1)); + assertEquals("internal://floci-msk-abc123:9093,external://localhost:54321", spec.cmd().get(flagIndex + 1)); assertEquals(Integer.valueOf(54321), spec.portBindings().get(KAFKA_PORT)); } @Test - void containerModeAdvertisesContainerNameAddress() { + void containerModeAdvertisesInternalAndExternalAddresses() { when(containerDetector.isRunningInContainer()).thenReturn(true); + when(portAllocator.allocate(9300, 9399)).thenReturn(54321); ContainerInfo info = new ContainerInfo("container-456", Map.of(KAFKA_PORT, new EndpointInfo("172.18.0.5", KAFKA_PORT))); @@ -199,15 +199,15 @@ class RedpandaManagerTest { int flagIndex = spec.cmd().indexOf("--advertise-kafka-addr"); assertTrue(flagIndex >= 0, "cmd should contain --advertise-kafka-addr"); - assertEquals("floci-msk-abc123:9092", spec.cmd().get(flagIndex + 1)); + assertEquals("internal://floci-msk-abc123:9093,external://localhost:54321", spec.cmd().get(flagIndex + 1)); - assertFalse(spec.portBindings().containsKey(KAFKA_PORT), "container mode should not publish ports to host"); + assertEquals(Integer.valueOf(54321), spec.portBindings().get(KAFKA_PORT)); assertTrue(spec.exposedPorts().contains(KAFKA_PORT)); assertTrue(spec.exposedPorts().contains(RedpandaManager.ADMIN_PORT)); assertFalse(Files.exists(tempDir.resolve("msk").resolve("test-cluster")), "container mode must not create host directories from inside the emulator container"); - verifyNoInteractions(portAllocator); + verify(portAllocator).allocate(9300, 9399); } @Test diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml index 3c37c56..e34e5f7 100644 --- a/src/test/resources/application.yml +++ b/src/test/resources/application.yml @@ -162,6 +162,10 @@ floci: enabled: true mock: true default-image: "redpandadata/redpanda:latest" + kafkaconnect: + enabled: true + mock: true + worker-image: apache/kafka:3.9.0 amazonmq: enabled: true mock: true diff --git a/tools/docs/services.yaml b/tools/docs/services.yaml index 57142a4..1361e60 100644 --- a/tools/docs/services.yaml +++ b/tools/docs/services.yaml @@ -10,6 +10,10 @@ # the new-service sentinel only fires on genuinely unregistered handlers. services: + - service: kafkaconnect + doc: docs/services/kafka-connect.md + sources: + - { path: src/main/java/io/github/hectorvent/floci/services/kafkaconnect/KafkaConnectController.java, mode: rest } - service: account doc: docs/services/account.md sources: