diff --git a/cli/cmd/mcp_tools_chaos.go b/cli/cmd/mcp_tools_chaos.go index 8afdec64..26f1e00c 100644 --- a/cli/cmd/mcp_tools_chaos.go +++ b/cli/cmd/mcp_tools_chaos.go @@ -282,7 +282,7 @@ var mcpCaveatsChaos = []mcpCaveat{ "backend replica: a replica that starts while another runs a plan marks that plan's report too, until " + "the plan ends and stores its outcome.", Refs: []string{ - mcpJava + "disruption/DisruptionOrphanReconciler.java:53-54,62-69,85-92,152-203", + mcpJava + "disruption/DisruptionOrphanReconciler.java:54-55,63-70,86-93,171-222", mcpJava + "disruption/DisruptionReportRepository.java:35-60,93-102", mcpJava + "disruption/DisruptionPersistence.java:26-37", mcpJava + "disruption/DisruptionLauncher.java:95-102,110-123", diff --git a/cli/cmd/mcp_tools_runs.go b/cli/cmd/mcp_tools_runs.go index d7662ba4..dd4a694f 100644 --- a/cli/cmd/mcp_tools_runs.go +++ b/cli/cmd/mcp_tools_runs.go @@ -1928,7 +1928,7 @@ func mcpActivityTestsSince(ctx context.Context, call *mcpCall, since time.Time, // are injected for part of its run, or one whose backend process stopped // before it ended and has not started again to mark it INTERRUPTED // (DisruptionLauncher.java:95-123, DisruptionReportRepository.java:35-43, -// DisruptionOrphanReconciler.java:152-203). +// DisruptionOrphanReconciler.java:171-222). func mcpActivityDisruptionsSince(ctx context.Context, call *mcpCall, since time.Time, limit int) (mcpActivityDisruptions, error) { out := mcpActivityDisruptions{StoredAsRunning: []mcpActivityDisruptionRow{}, Since: []mcpActivityDisruptionRow{}} for p := 0; p < mcpActivityMaxPages; p++ { @@ -2388,7 +2388,7 @@ var mcpCaveatsRuns = []mcpCaveat{ mcpJava + "disruption/DisruptionLauncher.java:95-123", mcpJava + "disruption/DisruptionPersistence.java:16-37", mcpJava + "disruption/DisruptionReportRepository.java:20-43", - mcpJava + "disruption/DisruptionOrphanReconciler.java:152-203", + mcpJava + "disruption/DisruptionOrphanReconciler.java:171-222", mcpJava + "disruption/DisruptionTemplateResource.java:63-65", mcpJava + "disruption/DisruptionScheduler.java:95-107", mcpJava + "disruption/DisruptionReportEntity.java:39-47", diff --git a/cli/examples/resilience-scale-down.yaml b/cli/examples/resilience-scale-down.yaml index 99f1529b..54d9eb37 100644 --- a/cli/examples/resilience-scale-down.yaml +++ b/cli/examples/resilience-scale-down.yaml @@ -19,7 +19,7 @@ # so min.isr=2 is still satisfied. The producer, with acks=all, sees no errors. # # A resilience test does not bring the broker back. Kates orphan recovery -# does, if Kates restarts 15 minutes or more after the test. To restore it +# does, the next time the Kates API starts. To restore it # yourself: # # kubectl patch kafkanodepool brokers-sigma -n kafka --type merge -p '{"spec":{"replicas":1}}' diff --git a/docs/book/07-chaos-practice.md b/docs/book/07-chaos-practice.md index 89f74d6a..bc8306a5 100644 --- a/docs/book/07-chaos-practice.md +++ b/docs/book/07-chaos-practice.md @@ -628,7 +628,7 @@ So rollback matters for two faults. For a partition, it deletes the policies the Rollback is per step, and a rolled-back step can still count as passed. The plan goes on to its next step, and a step whose fault went in counts toward `passedSteps` even when it rolled back afterward. Read each step's `rolledBack` and `rollbackReason`, which `kates disruption run` prints under the step. `rolledBack` says that rollback ran, not that it put anything back: it's `true` for a fault with nothing to undo, and for a rollback that failed. -If the Kates API pod dies in the middle of a plan, no rollback runs. When the Kates API starts again, it deletes the `managed-by=kates` NetworkPolicies and restores the scaled-down node pools and StatefulSets it finds in the Kafka namespace. It also marks `INTERRUPTED` the report of a plan that `kates disruption run` or `kates disruption playbook run` started. It does this once, at startup, and touches only faults older than `kates.chaos.orphan-recovery.min-age-sec`, 900 seconds by default. So when the Kates API is back within 15 minutes of the fault, the fault stays until you undo it, or until the Kates API starts again after those 15 minutes. +If the Kates API pod dies in the middle of a plan, no rollback runs. When the Kates API starts again, it deletes the `managed-by=kates` NetworkPolicies in every namespace, and restores the scaled-down node pools and StatefulSets it finds in the Kafka namespace. It also marks `INTERRUPTED` the report of a plan that `kates disruption run` or `kates disruption playbook run` started. It does this once, at startup, and undoes every fault made before the Kates API started, however recent: the `kates` chart runs one Kates API pod, so no other pod can still be running the plan. `kates.chaos.orphan-recovery.min-age-sec`, 0 by default, moves that cutoff earlier by as many seconds. ### Guardrail Parameters @@ -659,7 +659,7 @@ The guard counts brokers, reads pod readiness and, for a drain, where pods run, - A resilience run (`kates resilience run`) doesn't go through the guard at all: no broker count, no one-plan rule, no check before the fault and no rollback. Only the [fault parameter limits](#fault-parameter-limits) apply to it. - `make gameday` doesn't go through it either: it deletes a broker pod with `kubectl`. - The one-plan rule lives in the Kates API process, so it holds only while one Kates API pod runs. The `kates` chart refuses a second replica and rolls with `Recreate`, so it never runs two. -- Orphan recovery covers the Kafka namespace only. A NetworkPolicy that a Kates API left in another namespace when it died stays until you run `kubectl delete networkpolicy -n -l managed-by=kates`. +- Orphan recovery deletes Kates NetworkPolicies in every namespace, but restores scale-downs in the Kafka namespace only. A StatefulSet that a Kates API left scaled down in another namespace when it died stays so until you scale it back to the count in its `kates.io/original-replicas` annotation. - The dry run's RBAC check (see [Previewing a Playbook](#previewing-a-playbook)) asks about the Kates API's own service account, and a missing permission is a warning, not a refusal. ## Kafka Intelligence diff --git a/docs/book/11-api-reference.md b/docs/book/11-api-reference.md index 189ebd1c..3f001b51 100644 --- a/docs/book/11-api-reference.md +++ b/docs/book/11-api-reference.md @@ -710,7 +710,7 @@ Then poll `GET /api/disruptions/{id}` until the status is terminal: Disruption IDs are 8-character UUID prefixes. `status` is `RUNNING` while the plan executes, then one of `COMPLETED`, `PARTIAL` (some steps failed), `FAILED`, or `INTERRUPTED` (the process running the plan died; see below). Plans that violate the safety guard are rejected up front and return `422 Unprocessable Entity` with status `REJECTED` (see [Error Responses](#error-responses)). Only one plan may run against a cluster at a time: a `POST` while another plan is in flight returns `409 Conflict`. Concurrent plans would race the rollback state stored on the target and could leave a node pool or StatefulSet under-scaled. [Safety Guardrails](07-chaos-practice.md#safety-guardrails) in Chaos Engineering in Practice lists every reason the safety guard refuses a plan. -> **Orphan recovery.** If the Kates API pod is killed mid-plan, the injected faults would otherwise persist with nothing left to undo them. On startup Kates removes abandoned `managed-by=kates` NetworkPolicies in the Kafka namespace (`kates.chaos.kafka.namespace`), restores the KafkaNodePools and StatefulSets there that still carry a scale-down snapshot, and marks stranded `RUNNING` reports as `INTERRUPTED`. It runs once, at startup, and touches only faults older than `kates.chaos.orphan-recovery.min-age-sec` (default 900), so a fault younger than that when Kates restarts stays until you remove it or Kates restarts again. +> **Orphan recovery.** If the Kates API pod is killed mid-plan, the injected faults would otherwise persist with nothing left to undo them. On startup Kates removes abandoned `managed-by=kates` NetworkPolicies in every namespace, restores the KafkaNodePools and StatefulSets in the Kafka namespace (`kates.chaos.kafka.namespace`) that still carry a scale-down snapshot, and marks stranded `RUNNING` reports as `INTERRUPTED`. It runs once, at startup, and touches every fault made before Kates started; `kates.chaos.orphan-recovery.min-age-sec` (default 0) moves that cutoff earlier by as many seconds. **Dry run** — `POST /api/disruptions?dryRun=true` with the same request body returns `200 OK`: diff --git a/docs/stretched-kafka/03-kubernetes-and-strimzi.md b/docs/stretched-kafka/03-kubernetes-and-strimzi.md index efc1a0aa..50bf936d 100644 --- a/docs/stretched-kafka/03-kubernetes-and-strimzi.md +++ b/docs/stretched-kafka/03-kubernetes-and-strimzi.md @@ -578,7 +578,7 @@ MirrorMaker 2's `replication-latency-ms` and `record-age-ms` update only while r ## Where to Run the Test Harness (B0-H) -- **Pin the harness to the survivors.** Pin Kates, its PostgreSQL, Prometheus and the Litmus operator to the site each test expects to survive. The Kates chart's defaults are unpinned, with one replica. A harness in the lost site dies with it. The faults it injected are cleaned up only by orphan recovery at its next start, and only once they are 15 min old. +- **Pin the harness to the survivors.** Pin Kates, its PostgreSQL, Prometheus and the Litmus operator to the site each test expects to survive. The Kates chart's defaults are unpinned, with one replica. A harness in the lost site dies with it. The faults it injected are cleaned up only by orphan recovery at its next start. - **Partition tests need both sites.** For S7 tests (T3), run one Kates per site, each with its own INTEGRITY workload. - **Set the real cluster name** in `KATES_CHAOS_KAFKA_CLUSTER`. - **One Kates per chaos provider.** Use the direct backend (`kubernetes`) for simultaneous kills, and Litmus (`litmus-crd`) for network faults. The provider is fixed for the life of the process. diff --git a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrphanReconciler.java b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrphanReconciler.java index f2e66f22..23b89f22 100644 --- a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrphanReconciler.java +++ b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrphanReconciler.java @@ -1,6 +1,7 @@ package com.bmscomp.kates.disruption; import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.List; import jakarta.enterprise.context.ApplicationScoped; @@ -29,12 +30,12 @@ * brokers simply stayed, with nothing left running that knew to clean them up. * Tests already had {@code recoverOrphans}; disruptions had no equivalent. * - *

Why age matters. Kates may run more than one replica, so a fault - * seen at startup is not necessarily abandoned — another pod could be running - * that plan right now. Only faults older than - * {@code kates.chaos.orphan-recovery.min-age-sec} (default 15m, comfortably past - * the per-step recovery timeout) are treated as orphans, so live chaos is never - * ripped out from under a peer. + *

Why the process start is the cutoff. The chart pins Kates to one + * replica and rolls it with {@code Recreate}, so no other pod can be running a + * plan: every Kates fault created before this process started is an orphan, + * however recent. {@code kates.chaos.orphan-recovery.min-age-sec} (default 0) + * moves the cutoff that many seconds earlier, for a deployment that runs a + * second Kates pod anyway. */ @ApplicationScoped public class DisruptionOrphanReconciler { @@ -53,7 +54,7 @@ public class DisruptionOrphanReconciler { @ConfigProperty(name = "kates.chaos.orphan-recovery.enabled", defaultValue = "true") boolean enabled; - @ConfigProperty(name = "kates.chaos.orphan-recovery.min-age-sec", defaultValue = "900") + @ConfigProperty(name = "kates.chaos.orphan-recovery.min-age-sec", defaultValue = "0") long minAgeSec; @ConfigProperty(name = "kates.chaos.kafka.namespace", defaultValue = "kafka") @@ -80,15 +81,15 @@ void onStart(@Observes StartupEvent event) { // boot-time categories are at WARN, so a blocked boot prints nothing // at all. // - // Recovery is cleanup of faults at least `min-age-sec` old. Nothing - // about it needs to happen before the first request is served. + // Recovery is cleanup of faults made before this process started. + // Nothing about it needs to happen before the first request is served. Thread.ofVirtual().name("disruption-orphan-recovery").start(() -> reconcile(startedAt)); } private void reconcile(Instant startedAt) { try { - int policies = reconcileNetworkPolicies(); - ScaleDownSnapshots.Restored scaled = reconcileScaleDowns(); + int policies = reconcileNetworkPolicies(startedAt); + ScaleDownSnapshots.Restored scaled = reconcileScaleDowns(startedAt); int reports = markInterruptedReports(startedAt); if (policies + scaled.total() + reports > 0) { LOG.warnf( @@ -102,38 +103,50 @@ private void reconcile(Instant startedAt) { } } - /** Deletes stale Kates-managed NetworkPolicies (network partitions). */ - private int reconcileNetworkPolicies() { - long cutoff = System.currentTimeMillis() - minAgeSec * 1000L; - int removed = 0; + /** + * Deletes the Kates-managed NetworkPolicies (network partitions) left from + * before this process started, in every namespace, as cleanup does: a + * playbook can isolate pods outside the Kafka namespace. + */ + int reconcileNetworkPolicies(Instant startedAt) { + long cutoff = cutoff(startedAt); + List policies; try { - List policies = kubeClient + policies = kubeClient .network() .networkPolicies() - .inNamespace(kafkaNamespace) + .inAnyNamespace() .withLabel("managed-by", "kates") .list() .getItems(); - for (NetworkPolicy policy : policies) { - if (!isOlderThan(policy.getMetadata().getCreationTimestamp(), cutoff)) { - continue; - } + } catch (Exception e) { + // Concise on purpose: running without a reachable Kubernetes API + // (dev, unit tests) is normal and must not dump a stack trace on + // every boot. + LOG.warnf("Could not reconcile orphaned NetworkPolicies: %s", e.getMessage()); + return 0; + } + int removed = 0; + for (NetworkPolicy policy : policies) { + if (!isOlderThan(policy.getMetadata().getCreationTimestamp(), cutoff)) { + continue; + } + String namespace = policy.getMetadata().getNamespace(); + String name = policy.getMetadata().getName(); + // One try per policy: a namespace that refuses the delete must not + // stop recovery in the others. + try { kubeClient .network() .networkPolicies() - .inNamespace(kafkaNamespace) - .withName(policy.getMetadata().getName()) + .inNamespace(namespace) + .withName(name) .delete(); - LOG.warnf( - "Orphan recovery: deleted abandoned NetworkPolicy %s", - policy.getMetadata().getName()); + LOG.warnf("Orphan recovery: deleted abandoned NetworkPolicy %s/%s", namespace, name); removed++; + } catch (Exception e) { + LOG.warnf("Could not delete orphaned NetworkPolicy %s/%s: %s", namespace, name, e.getMessage()); } - } catch (Exception e) { - // Concise on purpose: running without a reachable Kubernetes API - // (dev, unit tests) is normal and must not dump a stack trace on - // every boot. - LOG.warnf("Could not reconcile orphaned NetworkPolicies: %s", e.getMessage()); } return removed; } @@ -144,11 +157,17 @@ private int reconcileNetworkPolicies() { * durable record of the original size — it survives the pod that created * it, which is exactly what makes recovery possible here. */ - ScaleDownSnapshots.Restored reconcileScaleDowns() { - long cutoff = System.currentTimeMillis() - minAgeSec * 1000L; + ScaleDownSnapshots.Restored reconcileScaleDowns(Instant startedAt) { + long cutoff = cutoff(startedAt); return ScaleDownSnapshots.restore(kubeClient, kafkaNamespace, meta -> isScaleDownOlderThan(meta, cutoff)); } + private long cutoff(Instant startedAt) { + // creationTimestamp has whole seconds: a policy made in this process's + // first second can be stamped before startedAt itself. + return startedAt.truncatedTo(ChronoUnit.SECONDS).toEpochMilli() - minAgeSec * 1000L; + } + /** * A RUNNING report row whose process is gone can never complete — leaving it * RUNNING forever misreports an in-flight disruption that no longer exists. diff --git a/kates/src/main/resources/application.properties b/kates/src/main/resources/application.properties index 142a732f..964a959f 100644 --- a/kates/src/main/resources/application.properties +++ b/kates/src/main/resources/application.properties @@ -195,7 +195,7 @@ mp.messaging.incoming.test-events-in.health-enabled=false # Disruption orphan recovery (faults abandoned by a killed pod) kates.chaos.orphan-recovery.enabled=true -kates.chaos.orphan-recovery.min-age-sec=900 +kates.chaos.orphan-recovery.min-age-sec=0 %test.kates.chaos.orphan-recovery.enabled=false %dev.kates.chaos.orphan-recovery.enabled=false diff --git a/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionOrphanReconcilerTest.java b/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionOrphanReconcilerTest.java index c218d305..1afcba70 100644 --- a/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionOrphanReconcilerTest.java +++ b/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionOrphanReconcilerTest.java @@ -1,8 +1,15 @@ package com.bmscomp.kates.disruption; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.time.Instant; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicy; +import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyBuilder; import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient; import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer; @@ -13,8 +20,10 @@ import com.bmscomp.kates.chaos.VertxPerMockClient; /** - * Startup recovery of a SCALE_DOWN whose plan died with the Kates pod. It - * used to look only at StatefulSets, which Strimzi does not create. + * Startup recovery of the faults a plan left when it died with the Kates pod. + * The scale-down recovery used to look only at StatefulSets, which Strimzi does + * not create, and every recovery used to skip a fault less than 15 minutes old, + * which is every fault a normal restart leaves. */ @EnableKubernetesMockClient(crud = true, kubernetesClientBuilderCustomizer = VertxPerMockClient.class) class DisruptionOrphanReconcilerTest { @@ -24,25 +33,84 @@ class DisruptionOrphanReconcilerTest { @Test void restoresANodePoolAbandonedMidPlanAndLeavesARecentScaleDownAlone() { + Instant startedAt = Instant.now(); StrimziTestCluster cluster = new StrimziTestCluster(server, client) .pool("brokers-alpha", "broker", 3) .pool("brokers-sigma", "broker", 4) - .pool("brokers-gamma", "broker", 5); + .pool("brokers-gamma", "broker", 5) + .pool("brokers-delta", "broker", 6); cluster.scaledDown("brokers-sigma", 0); - // Could be a plan another Kates replica is running right now. - cluster.scaledDown("brokers-gamma", 0, System.currentTimeMillis()); - - DisruptionOrphanReconciler reconciler = new DisruptionOrphanReconciler(); - reconciler.kubeClient = client; - reconciler.kafkaNamespace = StrimziTestCluster.NAMESPACE; - reconciler.minAgeSec = 900; + // A plan this process started. + cluster.scaledDown("brokers-gamma", 0, startedAt.plusSeconds(1).toEpochMilli()); + // The pod was killed 2 minutes into the plan and came back 90 s later. + cluster.scaledDown("brokers-delta", 0, startedAt.minusSeconds(120).toEpochMilli()); - ScaleDownSnapshots.Restored restored = reconciler.reconcileScaleDowns(); + ScaleDownSnapshots.Restored restored = reconciler().reconcileScaleDowns(startedAt); - assertEquals(new ScaleDownSnapshots.Restored(0, 1), restored); + assertEquals(new ScaleDownSnapshots.Restored(0, 2), restored); assertEquals(1, cluster.replicas("brokers-sigma")); assertTrue(cluster.annotations("brokers-sigma").isEmpty(), "snapshot cleared"); + assertEquals(1, cluster.replicas("brokers-delta")); assertEquals(0, cluster.replicas("brokers-gamma")); assertEquals(1, cluster.replicas("brokers-alpha")); } + + @Test + void deletesKatesNetworkPoliciesFromBeforeTheStartInEveryNamespace() { + // The mock server stamps creationTimestamp itself, so the start moves + // around the policies instead. + policy("kafka", "kates-netpol-krafter-brokers-sigma-4", true); + policy("kates", "kates-netpol-consumer", true); + policy("kafka", "allow-brokers", false); + DisruptionOrphanReconciler reconciler = reconciler(); + + assertEquals(0, reconciler.reconcileNetworkPolicies(Instant.now().minusSeconds(60))); + assertNotNull(get("kafka", "kates-netpol-krafter-brokers-sigma-4"), "made after the start"); + + assertEquals(2, reconciler.reconcileNetworkPolicies(Instant.now().plusSeconds(60))); + assertNull(get("kafka", "kates-netpol-krafter-brokers-sigma-4")); + assertNull(get("kates", "kates-netpol-consumer")); + assertNotNull(get("kafka", "allow-brokers"), "not Kates-managed"); + } + + @Test + void aPolicyStampedInTheSecondTheProcessStartedIsKept() { + NetworkPolicy created = policy("kafka", "kates-netpol-krafter-brokers-sigma-4", true); + // Stamped 12:00:00Z by a plan this process launched at 12:00:00.700. + Instant startedAt = + Instant.parse(created.getMetadata().getCreationTimestamp()).plusMillis(700); + + assertEquals(0, reconciler().reconcileNetworkPolicies(startedAt)); + assertNotNull(get("kafka", "kates-netpol-krafter-brokers-sigma-4")); + } + + private DisruptionOrphanReconciler reconciler() { + DisruptionOrphanReconciler reconciler = new DisruptionOrphanReconciler(); + reconciler.kubeClient = client; + reconciler.kafkaNamespace = StrimziTestCluster.NAMESPACE; + reconciler.minAgeSec = 0; + return reconciler; + } + + private NetworkPolicy policy(String namespace, String name, boolean katesManaged) { + return client.network() + .networkPolicies() + .inNamespace(namespace) + .resource(new NetworkPolicyBuilder() + .withNewMetadata() + .withName(name) + .withNamespace(namespace) + .withLabels(katesManaged ? Map.of("managed-by", "kates") : Map.of()) + .endMetadata() + .build()) + .create(); + } + + private NetworkPolicy get(String namespace, String name) { + return client.network() + .networkPolicies() + .inNamespace(namespace) + .withName(name) + .get(); + } }