AWS MSK Fleet Health Gate

An AWS MSK cluster that enters state FAILED does not page anyone by itself - AWS's own console shows it, but nothing watches for it on a schedule unless a CloudWatch alarm was deliberately wired up for that specific state transition, on that specific cluster. This blueprint lists every MSK cluster in a region in one call, flags any that are FAILED, and alerts Slack the moment that happens or the MSK API itself cannot be reached - a fleet-wide check rather than a single cluster's.

How it works

1. periodic_fleet_check (io.kestra.plugin.core.trigger.Schedule) runs every 30 minutes, shipped disabled so you can validate a manual run first.
2. list_msk_clusters (io.kestra.plugin.aws.msk.ListClusters, allowFailure: true) returns every cluster matching cluster_name_filter (or every cluster in the region if left empty), each carrying clusterArn, clusterName, state, kafkaVersion, numberOfBrokerNodes, and creationTime.
3. evaluate_cluster_health (io.kestra.plugin.core.debug.Return) classifies UNREACHABLE (the call did not return), AT_RISK (the count of FAILED clusters exceeds max_failed_clusters), or HEALTHY.
4. route_by_health (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY logs a clean fleet, AT_RISK pages Slack naming every failed cluster, UNREACHABLE pages Slack that the API itself could not be reached, and the defaults: block catches any unexpected classification value.
5. log_audit_complete always runs last, printing the fleet total and classification regardless of which branch fired.
6. The errors block alerts Slack separately if the flow itself fails outside these handled branches.

What you get

• A fleet-wide, named list of every FAILED MSK cluster in a region, computed from AWS's own ListClusters API, not inferred from a per-cluster alarm someone had to remember to configure.
• A three-way classification (HEALTHY/AT_RISK/UNREACHABLE) so a broken audit (bad credentials, a missing IAM permission) is never mistaken for a healthy fleet.
• A final status log every run, so the execution history doubles as a fleet-health trend.

Who it's for

• Platform teams running multiple MSK clusters across environments or teams who want one scheduled check instead of a CloudWatch alarm wired per cluster.
• SREs who want a cluster-state safety net independent of whatever alerting the Kafka consumers themselves have.
• Anyone evaluating this plugin's MSK tasks who wants a worked example of ListClusters used as a standing fleet audit rather than a one-off lookup.

Why orchestrate this with Kestra

ListClusters is a single API call with no memory of the last check and no notification path of its own. Kestra supplies the schedule, the three-way risk classification as a first-class branch, and an execution history that shows exactly when a cluster first entered FAILED and for how long it stayed that way.

Prerequisites

• An AWS account with at least one MSK cluster, and IAM permissions for kafka:ListClusters.
• A Slack incoming webhook for alerts.

Local testing: AWS MSK is a fully managed service with no official self-hosted or LocalStack-community emulator for actual cluster provisioning and state transitions (LocalStack's MSK support, where it exists, is a Pro-tier feature). Validate this flow against a real AWS account and a real (even single-broker, smallest-instance-type) MSK cluster rather than attempting a fully local run - this has not been exercised in this environment.

Secrets

• AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_DEFAULT_REGION: AWS credentials used by ListClusters, matching this plugin's own documented connection properties (see build-aws-ecr-image.yaml for the same pattern on a different AWS service).
• SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_at_risk, alert_unreachable, and the errors block.
• In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_AWS_ACCESS_KEY_ID=$(echo -n '<your-access-key>' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

• cluster_name_filter (STRING, default empty): optional substring filter on cluster name; empty audits every cluster in the region.
• max_failed_clusters (INT, default 0): breach threshold; 0 means any FAILED cluster at all breaches.

Outputs

• outputs.list_msk_clusters.clusters / .total: the raw cluster listing and count for every run (absent when the API call failed).
• outputs.evaluate_cluster_health.value: HEALTHY, AT_RISK, or UNREACHABLE for every run.

Quick start

1. Point this flow at an AWS account with at least one MSK cluster (any state is fine for the first HEALTHY-path check).
2. Add the AWS credential secrets and SLACK_WEBHOOK_URL.
3. Run the flow manually and confirm log_healthy fires naming the total cluster count.
4. To exercise AT_RISK safely, do this against a non-production, disposable test cluster only - deliberately misconfiguring a cluster's VPC/subnet/security-group association before creation is one documented way to land it in FAILED - then re-run and confirm alert_at_risk names it.
5. Enable periodic_fleet_check once you trust the check.

How to extend

• Chain DescribeCluster (also in this plugin) after a Slack alert to pull kafkaVersion and numberOfBrokerNodes for the specific failed cluster into a richer incident message.
• Loop over several cluster_name_filter values with io.kestra.plugin.core.flow.Loop to report per-team or per-environment fleet health separately instead of one combined count.
• Add a second classification tier that also flags clusters stuck in a transitional state (CREATING, UPDATING, REBOOTING_BROKER) for longer than an expected duration, using each entry's creationTime as a reference point.
• If AWS later adds a documented single-call cluster repair action to this plugin, extend this blueprint with the same double-gated auto_remediate/dry_run pattern used by this repo's other sentinels - deliberately not attempted here; see Pitfalls.

Pitfalls

• This blueprint does not attempt automated remediation. Verified task list for this plugin's MSK package: CreateCluster, DeleteCluster, DescribeCluster, GetBootstrapBrokers, ListClusters, plus a Trigger. None of them offer a safe, targeted "repair" or "reboot" action for a cluster already in FAILED - DeleteCluster is destructive and CreateCluster makes a new one, neither fixes the existing failed cluster. This is audit-and-alert only, by design.
• ListClusters results are capped at 100 entries. Verified from the plugin's own documentation and source: "Results are capped at 100 entries; for larger accounts use the AWS Console or CLI for full enumeration." An account with more than 100 MSK clusters matching the filter will be under-audited by this blueprint as written.
• state is one of eight documented values, not just ACTIVE/FAILED. Verified from the plugin's own source: ACTIVE, CREATING, DELETING, FAILED, HEALING, MAINTENANCE, REBOOTING_BROKER, UPDATING. This blueprint only treats FAILED as a breach; the other five non-ACTIVE states are transitional and intentionally not alerted on, to avoid paging on a routine maintenance window or broker reboot.
• allowFailure: true is what lets UNREACHABLE be classified instead of failing the execution outright. A base Task property, not specific to this plugin; without it, a missing IAM permission or bad credential would abort the flow before evaluate_cluster_health ever runs.
• Switch case keys are quoted strings matching the classifier's exact output. "HEALTHY", "AT_RISK", and "UNREACHABLE" must match evaluate_cluster_health.value verbatim; the defaults: block exists specifically to catch any value that does not match one of those three cases.

Links

• plugin-aws MSK documentation
• plugin-aws MSK source
• AWS MSK cluster states reference
• Kestra secrets (OSS vs Enterprise)