diff --git a/cli/cmd/mcp_tools_chaos.go b/cli/cmd/mcp_tools_chaos.go index 8afdec64..f2a54ffa 100644 --- a/cli/cmd/mcp_tools_chaos.go +++ b/cli/cmd/mcp_tools_chaos.go @@ -180,7 +180,7 @@ var mcpCaveatsChaos = []mcpCaveat{ "unknown or was unavailable when the backend started, the backend falls back to noop, which injects " + "nothing and marks every step Skipped, and says so only in its log.", Refs: []string{ - mcpAnchoredRef(mcpJava+"chaos/CompoundChaosOrchestrator.java:150-169", + mcpAnchoredRef(mcpJava+"chaos/CompoundChaosOrchestrator.java:155-174", "List availableProviders()", `" (available)" : " (unavailable)"`, "return names;"), mcpAnchoredRef(mcpJava+"chaos/ChaosCoordinator.java:34-75", `name = "kates.chaos.provider"`, "matches no chaos provider"), diff --git a/cli/cmd/mcp_tools_runs.go b/cli/cmd/mcp_tools_runs.go index d7662ba4..cc8ef6d0 100644 --- a/cli/cmd/mcp_tools_runs.go +++ b/cli/cmd/mcp_tools_runs.go @@ -2392,9 +2392,9 @@ var mcpCaveatsRuns = []mcpCaveat{ mcpJava + "disruption/DisruptionTemplateResource.java:63-65", mcpJava + "disruption/DisruptionScheduler.java:95-107", mcpJava + "disruption/DisruptionReportEntity.java:39-47", - mcpAnchoredRef(mcpJava+"disruption/DisruptionAnalysisResource.java:110-141", + mcpAnchoredRef(mcpJava+"disruption/DisruptionAnalysisResource.java:118-192", `@Path("/compound")`, `"results", outcome.results()`), - mcpAnchoredRef(mcpJava+"resilience/ResilienceResource.java:97-127", + mcpAnchoredRef(mcpJava+"resilience/ResilienceResource.java:107-137", "StreamingOutput executeWithKeepAlive(", "objectMapper.writeValue(os, payload);", `"Failed to execute resilience test"`), }, diff --git a/cli/cmd/mcp_tools_scenario.go b/cli/cmd/mcp_tools_scenario.go index 1376612d..1d2290a1 100644 --- a/cli/cmd/mcp_tools_scenario.go +++ b/cli/cmd/mcp_tools_scenario.go @@ -1286,7 +1286,7 @@ var mcpCaveatsScenario = []mcpCaveat{ "switch (task.getWorkloadType())", "case INTEGRITY -> runIntegrity(task, state);", "void runIntegrity(BenchmarkTask task, WorkerState state)", "state.chaosStartNanos.get()"), mcpJava + "engine/TrogdorBackend.java:162-163", - mcpAnchoredRef(mcpJava+"resilience/ResilienceOrchestrator.java:144-145", + mcpAnchoredRef(mcpJava+"resilience/ResilienceOrchestrator.java:158-159", "chaosCoordinator.triggerFault(", "testOrchestrator.markChaosStart(runId, injectedAt)"), }, }, diff --git a/docs/book/01-introduction.md b/docs/book/01-introduction.md index 8c9e1cda..ff14ef02 100644 --- a/docs/book/01-introduction.md +++ b/docs/book/01-introduction.md @@ -70,7 +70,7 @@ Every run produces a JSON report, which you can export as CSV, as JUnit XML for ### 5. Safe by Default -Kates refuses a disruption plan that would hit every broker, or more brokers than the plan's `maxAffectedBrokers` allows. Before each fault, it checks that every Kafka pod is Ready. By default, when a step fails, [rollback](appendix-a-glossary.md#gl-disruption-rollback) gives back a broker that a scale-down removed and deletes the NetworkPolicies Kates created for a network partition. This safety guard counts brokers, not replicas. It doesn't read the ISR or the KRaft quorum, and a [resilience run](appendix-a-glossary.md#gl-resilience-run) doesn't go through it. [Chaos Engineering in Practice](07-chaos-practice.md#safety-guardrails) says exactly what it checks. +Kates refuses a disruption plan that would hit every broker, or more brokers than the plan's `maxAffectedBrokers` allows. Before each fault, it checks that every Kafka pod is Ready. By default, when a step fails, [rollback](appendix-a-glossary.md#gl-disruption-rollback) gives back a broker that a scale-down removed and deletes the NetworkPolicies Kates created for a network partition. This safety guard counts brokers, not replicas. It doesn't read the ISR or the KRaft quorum, and a [resilience run](appendix-a-glossary.md#gl-resilience-run) goes through only part of it: the broker count and the one-at-a-time rule before its test starts, with no Ready check before the fault and no rollback (see [Limits of the Guard](07-chaos-practice.md#limits-of-the-guard)). [Chaos Engineering in Practice](07-chaos-practice.md#safety-guardrails) says exactly what it checks. ## Feature Overview diff --git a/docs/book/02-architecture.md b/docs/book/02-architecture.md index 96084cd3..83942e2c 100644 --- a/docs/book/02-architecture.md +++ b/docs/book/02-architecture.md @@ -227,7 +227,7 @@ Kates's [disruption types](appendix-a-glossary.md#gl-disruption-type) run from k ### Safety Guardrails -Before a disruption plan touches the cluster, Kates counts the brokers its steps would hit, and refuses the plan if that's every broker, or more than the plan's `maxAffectedBrokers`. Before each fault it checks that every Kafka pod is Running and Ready. When a step fails with `autoRollback` on, [rollback](appendix-a-glossary.md#gl-disruption-rollback) gives a scaled-down node pool its broker back, or deletes the [NetworkPolicies](appendix-a-glossary.md#gl-networkpolicy) Kates created for a network partition. The safety guard doesn't read the [ISR](appendix-a-glossary.md#gl-isr) or the [KRaft](appendix-a-glossary.md#gl-kraft) quorum, and a resilience run doesn't go through it: [Chaos Engineering in Practice](07-chaos-practice.md#safety-guardrails) says exactly what it checks. +Before a disruption plan touches the cluster, Kates counts the brokers its steps would hit, and refuses the plan if that's every broker, or more than the plan's `maxAffectedBrokers`. Before each fault it checks that every Kafka pod is Running and Ready. When a step fails with `autoRollback` on, [rollback](appendix-a-glossary.md#gl-disruption-rollback) gives a scaled-down node pool its broker back, or deletes the [NetworkPolicies](appendix-a-glossary.md#gl-networkpolicy) Kates created for a network partition. The safety guard doesn't read the [ISR](appendix-a-glossary.md#gl-isr) or the [KRaft](appendix-a-glossary.md#gl-kraft) quorum, and a resilience run goes through only part of it: the broker count and the one-at-a-time rule before its test starts, with no Ready check before the fault and no rollback (see [Limits of the Guard](07-chaos-practice.md#limits-of-the-guard)). [Chaos Engineering in Practice](07-chaos-practice.md#safety-guardrails) says exactly what it checks. ## CLI Architecture @@ -519,7 +519,7 @@ kates report show - Kates has two halves: the Go CLI on your machine and the Kates API, one Quarkus service in the cluster that serves REST and gRPC. - The `TestOrchestrator` owns a run's lifecycle: resolve defaults, create the topic, launch workers, poll status and collect heatmap rows; reports are built on request. - Latency measurement rests on HdrHistogram (1µs–60s, microsecond precision), compressed into 25 heatmap buckets for export. -- The safety guard refuses a disruption plan that would hit too many brokers, and rollback undoes some faults; a resilience run skips both. +- The safety guard refuses a disruption plan or resilience run that would hit too many brokers, and rollback undoes some faults; a resilience run skips rollback. - Results live in PostgreSQL, not Kafka; heatmaps, security grades and built reports live only in the Kates API's memory. [The Cluster Under Test](03-cluster.md) comes next: it builds the Kubernetes and Kafka environment that the Kates API runs in and tests. diff --git a/docs/book/05-test-types.md b/docs/book/05-test-types.md index 2d044e28..b75b74b5 100644 --- a/docs/book/05-test-types.md +++ b/docs/book/05-test-types.md @@ -485,7 +485,7 @@ scenarios: ### Integrity + Chaos -The real power of INTEGRITY tests emerges when combined with chaos engineering. An INTEGRITY run injects no [fault](appendix-a-glossary.md#gl-fault) by itself: `kates resilience run` pairs a test request with one fault, injected while the test runs. That pairing is a [resilience run](appendix-a-glossary.md#gl-resilience-run), one of the two ways Kates runs a [chaos experiment](appendix-a-glossary.md#gl-chaos-experiment), and it doesn't go through the [safety guard](appendix-a-glossary.md#gl-safety-guard) that checks a disruption plan (see [Chaos Engineering in Practice](07-chaos-practice.md)): +The real power of INTEGRITY tests emerges when combined with chaos engineering. An INTEGRITY run injects no [fault](appendix-a-glossary.md#gl-fault) by itself: `kates resilience run` pairs a test request with one fault, injected while the test runs. That pairing is a [resilience run](appendix-a-glossary.md#gl-resilience-run), one of the two ways Kates runs a [chaos experiment](appendix-a-glossary.md#gl-chaos-experiment), and it goes through only part of the [safety guard](appendix-a-glossary.md#gl-safety-guard) that checks a disruption plan: the broker count and the one-at-a-time rule before its test starts, with no Ready check before the fault and no rollback (see [Limits of the Guard](07-chaos-practice.md#limits-of-the-guard)): ```yaml # resilience-integrity.yaml diff --git a/docs/book/07-chaos-practice.md b/docs/book/07-chaos-practice.md index 89f74d6a..a2d07f8b 100644 --- a/docs/book/07-chaos-practice.md +++ b/docs/book/07-chaos-practice.md @@ -75,8 +75,8 @@ The table sets a plan and a resilience run side by side. Read the guard, rollbac | Faults | One per step | One | | Workload | None | The Kates test in `testRequest` | | What it measures | Pod recovery, Prometheus snapshots before and after each fault, and ISR and lag when asked | The test's throughput, latency and error rate just before the fault, and again over the whole run after the recovery wait | -| Safety guard | Checks the plan before the first fault, and the Kafka pods before each fault | None, apart from the [fault parameter limits](#fault-parameter-limits) | -| Another plan running | Refused | Not checked | +| Safety guard | Checks the plan before the first fault, and the Kafka pods before each fault | Counts the brokers the fault hits, and holds it to the [fault parameter limits](#fault-parameter-limits), before the test starts; doesn't check the Kafka pods before the fault | +| Another disruption running | Refused | Refused with `409` before the test starts; a run that finds one when `steadyStateSec` is up ends `ERROR` with no fault | | Rollback | After a step fails its recovery check or errors, with `autoRollback` | None | | Grade | An [SLA grade](appendix-a-glossary.md#gl-sla-grade) when the plan has an `sla` block | None | | Report | Saved: `kates disruption status ` and `kates chaos list` | Printed once; the [test run](appendix-a-glossary.md#gl-test-run) is saved like any other | @@ -85,8 +85,8 @@ The table sets a plan and a resilience run side by side. Read the guard, rollbac Both paths get the fault itself from the same chaos provider, which [Choosing a Chaos Provider](#choosing-a-chaos-provider) describes, so the same fault spec (a plan step's `faultSpec`, a resilience file's `chaosSpec`) does the same thing either way. The one difference is aiming: only a plan step looks up a partition's [leader](appendix-a-glossary.md#gl-partition-leader) for you, when the step names a `targetTopic`. -::: {.callout-caution title="A Resilience Run Has No Safety Guard"} -Nothing checks a resilience run's `chaosSpec` against `maxAffectedBrokers` or waits for a running plan to finish, and nothing rolls its fault back. Aim it with a selector that matches only the pods you mean to hit, such as the brokers. +::: {.callout-caution title="A Resilience Run Has No Rollback"} +Kates refuses a resilience run whose `chaosSpec` would leave no broker untouched, and one started while another disruption holds the cluster. A resilience run has no `maxAffectedBrokers`, though, nothing checks that the Kafka pods are Ready before its fault, and nothing rolls the fault back. Aim it with a selector that matches only the pods you mean to hit, such as the brokers. ::: ## Disruption Types @@ -578,7 +578,7 @@ Kates runs the plan anyway, with a warning in `validationWarnings`, in three cas - A step is a `SCALE_DOWN`, whose broker stays removed until rollback puts it back. - The `sla` block sets a threshold a plan can't evaluate (see [SLA Grading](#sla-grading)). -Only one plan runs against the cluster at a time. While one runs, Kates refuses a second, with `409 Conflict` over the API, because both would rely on the original replica counts that rollback restores from. +Only one disruption runs against the cluster at a time, whether a plan, a compound run or a resilience run's fault. While one runs, Kates refuses another, with `409 Conflict` over the API, because two would take down brokers together, and two plans would both rely on the original replica counts that rollback restores from. A resilience run is refused before its test starts, and checked again just before its fault: one that finds the cluster taken then ends `ERROR` with no fault injected. ### What Kates Checks Before Each Fault @@ -588,7 +588,7 @@ If one isn't, the step fails with `Cluster is not in a stable baseline state bef ### Fault Parameter Limits -Every number in a fault spec has a range, and Kates refuses a fault with one outside it before anything is injected. A plan is checked before its first fault, whether you post it or it comes from a playbook, a template or a schedule, and one with a fault outside its range ends `REJECTED`. A resilience run's `chaosSpec` is checked before its test starts, and every fault of a compound run (`POST /api/disruptions/compound`) before any goes in; both get a `400` naming each parameter. The ranges are generous on purpose: they stop a typo or a runaway value, not a fault you mean to run. The table gives each range with the defaults, and the Kates API setting that raises its ceiling: +Every number in a fault spec has a range, and Kates refuses a fault with one outside it before anything is injected. A plan is checked before its first fault, whether you post it or it comes from a playbook, a template or a schedule, and one with a fault outside its range ends `REJECTED`. A resilience run's `chaosSpec` is checked before its test starts, and every fault of a compound run (`POST /api/disruptions/compound`) before any goes in; both get a `400` naming each parameter. Both are then counted by the safety guard, each fault as a plan step, and get a `400` with the guard's reasons when it refuses them. The ranges are generous on purpose: they stop a typo or a runaway value, not a fault you mean to run. The table gives each range with the defaults, and the Kates API setting that raises its ceiling: | Parameter | Range with the defaults | Setting for the ceiling | |-----------|-------------------------|-------------------------| @@ -656,7 +656,7 @@ The guard counts brokers, reads pod readiness and, for a drain, where pods run, - A step whose selector looks in another namespace, such as the consumers `consumer-isolation` isolates in `kates`, counts no broker, whatever it hits, unless it's a `NODE_DRAIN`. - Steps that pick one pod at random from the same selector count as one broker between them, although each can hit a different broker. - A `NODE_DRAIN` step counts the brokers on its node as the pods run when you submit the plan. A pod that moves to another node before the step runs takes the drain with it. -- 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. +- A resilience run (`kates resilience run`) gets the broker count before its test starts, with no `maxAffectedBrokers`, and the one-at-a-time rule then and again just before its fault. It gets no check of the Kafka pods before the fault and no rollback. - `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`. @@ -817,7 +817,7 @@ The Kates API emits these events under the plan's name, not under the disruption ## Resilience Testing: Performance + Chaos Combined -The `kates resilience run` command combines a performance test with chaos injection, providing a **before/after impact analysis**. It is the second of the [Two Ways to Run a Fault](#two-ways-to-run-a-fault), and it doesn't go through the safety guard: see [Limits of the Guard](#limits-of-the-guard). The diagram shows its phases: +The `kates resilience run` command combines a performance test with chaos injection, providing a **before/after impact analysis**. It is the second of the [Two Ways to Run a Fault](#two-ways-to-run-a-fault), and the safety guard checks less of it than of a plan: see [Limits of the Guard](#limits-of-the-guard). The diagram shows its phases: ```mermaid %%| label: fig-practice-resilience-run diff --git a/docs/book/10-cli-reference.md b/docs/book/10-cli-reference.md index 756d053b..1bbc97de 100644 --- a/docs/book/10-cli-reference.md +++ b/docs/book/10-cli-reference.md @@ -1067,7 +1067,7 @@ kates chaos show ### Resilience -Combined performance and chaos testing: a resilience run starts one Kates test, injects one fault while it runs, and prints the change in throughput, latency and error rate from before the fault. Unlike a disruption plan, it has no safety guard, rollback or grade, as [Chaos Engineering in Practice](07-chaos-practice.md) explains. +Combined performance and chaos testing: a resilience run starts one Kates test, injects one fault while it runs, and prints the change in throughput, latency and error rate from before the fault. Unlike a disruption plan, it has no rollback or grade, and the safety guard checks less of it, as [Chaos Engineering in Practice](07-chaos-practice.md) explains. ```bash kates resilience run -f resilience-test.yaml diff --git a/docs/book/11-api-reference.md b/docs/book/11-api-reference.md index 189ebd1c..81cf6857 100644 --- a/docs/book/11-api-reference.md +++ b/docs/book/11-api-reference.md @@ -708,7 +708,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. +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 disruption may run against a cluster at a time: a `POST` while another plan, a compound run or a resilience run's fault 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. @@ -870,7 +870,7 @@ Run a playbook. It takes no body and goes through the launcher `POST /api/disrup ### Resilience Testing -A resilience run starts a test run, injects one fault while it runs, and compares a snapshot of the run taken before the fault with a summary of the whole run taken after the recovery wait; `kates resilience run` sends this request. A disruption plan starts no test run and measures the cluster itself; [Chaos Engineering in Practice](07-chaos-practice.md) explains both. A resilience run doesn't go through the safety guard: nothing counts the brokers its fault hits, and nothing rolls it back. Its fault is held to the same [Fault Parameter Limits](07-chaos-practice.md#fault-parameter-limits) as a plan's, though. +A resilience run starts a test run, injects one fault while it runs, and compares a snapshot of the run taken before the fault with a summary of the whole run taken after the recovery wait; `kates resilience run` sends this request. A disruption plan starts no test run and measures the cluster itself; [Chaos Engineering in Practice](07-chaos-practice.md) explains both. A resilience run goes through part of the safety guard: before the test starts, Kates counts the brokers its fault hits and refuses the run while another disruption holds the cluster. Nothing checks the Kafka pods before the fault, though, and nothing rolls it back. Its fault is held to the same [Fault Parameter Limits](07-chaos-practice.md#fault-parameter-limits) as a plan's. #### POST /api/resilience @@ -915,9 +915,9 @@ The call returns once the probes pass after the fault, or once `maxRecoveryWaitS } ``` -A `testRequest` that `POST /api/tests` would refuse for a field its type or benchmark backend cannot apply is refused here too, with the same `400` and `fieldErrors`, before the stream starts and before any fault is injected. The same goes for a `null` in a scenario's `phases`, and for a `testRequest` without a `type`, keyed `type`, unless its `scenario` has a `type` of its own. So is a `testRequest` whose `spec` has a value outside the limits `POST /api/tests` sets, such as a `numRecords` below 1 or a `topic` that isn't a legal Kafka topic name. As on `POST /api/tests`, a scenario's `baseSpec` and each phase's `spec` are held to the same limits, with `fieldErrors` keyed by their path in the scenario, such as `phases[0].spec.numRecords`. A `chaosSpec` with a parameter outside the fault parameter limits is refused the same way: `fieldErrors` names each parameter, and `message` prefixes it with `chaosSpec.`. +A `testRequest` that `POST /api/tests` would refuse for a field its type or benchmark backend cannot apply is refused here too, with the same `400` and `fieldErrors`, before the stream starts and before any fault is injected. The same goes for a `null` in a scenario's `phases`, and for a `testRequest` without a `type`, keyed `type`, unless its `scenario` has a `type` of its own. So is a `testRequest` whose `spec` has a value outside the limits `POST /api/tests` sets, such as a `numRecords` below 1 or a `topic` that isn't a legal Kafka topic name. As on `POST /api/tests`, a scenario's `baseSpec` and each phase's `spec` are held to the same limits, with `fieldErrors` keyed by their path in the scenario, such as `phases[0].spec.numRecords`. A `chaosSpec` with a parameter outside the fault parameter limits is refused the same way: `fieldErrors` names each parameter, and `message` prefixes it with `chaosSpec.`. Last, the safety guard counts the brokers `chaosSpec` hits, as for a plan of one step, and a fault it refuses gets a `400` whose `message` starts `No fault was injected:` and gives each reason, such as `Plan would affect ALL 3 brokers — cluster would lose availability`. While another disruption (a plan, a compound run or another resilience run's fault) holds the cluster, the call gets `409 Conflict` instead. -`status` is one of `COMPLETED`, `CHAOS_FAILED`, `INTERRUPTED`, or `ERROR`; with `ERROR`, `error` says why, for example that the benchmark did not start. Kates injects the fault only into a test run that is `RUNNING`, and `steadyStateSec` counts from the moment the run's tasks are submitted. A run that has failed or finished by then, or when `steadyStateSec` is up, gets neither the fault nor the probes: `error` says the benchmark ended before the fault, names the run, and quotes the errors its tasks reported. A run whose tasks are still being submitted after five minutes gets `ERROR` too, with an `error` saying it had not started. Impact deltas are percentage changes between the pre- and post-chaos summaries. Durations are in seconds: `chaosDuration` runs from the moment Kates creates the fault to the chaos outcome's `verdict`, so on Litmus it includes the experiment's start-up. `recoveryTime` runs from that `verdict` to the end of the first poll in which every probe passed. Kates polls the probes every 5 s, `maxRecoveryWaitSec` ÷ 5 times at most and once at least. When no poll finds them all passing, the report has `unrecoveredAfter` in place of `recoveryTime`: the time to the end of the last poll, the least the recovery took. Either way, `postRecoveryProbes` holds the last poll's results. +`status` is one of `COMPLETED`, `CHAOS_FAILED`, `INTERRUPTED`, or `ERROR`; with `ERROR`, `error` says why, for example that the benchmark did not start. Kates injects the fault only into a test run that is `RUNNING`, and `steadyStateSec` counts from the moment the run's tasks are submitted. A run that has failed or finished by then, or when `steadyStateSec` is up, gets neither the fault nor the probes: `error` says the benchmark ended before the fault, names the run, and quotes the errors its tasks reported. A run whose tasks are still being submitted after five minutes gets `ERROR` too, with an `error` saying it had not started. So does a run that finds another disruption holding the cluster when `steadyStateSec` is up: the check before the stream starts doesn't reserve the cluster, and the run takes it only just before its fault. Impact deltas are percentage changes between the pre- and post-chaos summaries. Durations are in seconds: `chaosDuration` runs from the moment Kates creates the fault to the chaos outcome's `verdict`, so on Litmus it includes the experiment's start-up. `recoveryTime` runs from that `verdict` to the end of the first poll in which every probe passed. Kates polls the probes every 5 s, `maxRecoveryWaitSec` ÷ 5 times at most and once at least. When no poll finds them all passing, the report has `unrecoveredAfter` in place of `recoveryTime`: the time to the end of the last poll, the least the recovery took. Either way, `postRecoveryProbes` holds the last poll's results. #### GET /api/resilience/scenarios @@ -969,6 +969,9 @@ The body is checked before the stream starts and before any fault is injected, a - Its `testRequest` is one `POST /api/resilience` would refuse, as above, and the answer has the same `fieldErrors`. - Its `targetLabel` isn't a label selector, and `fieldErrors` has the key `targetLabel`. - Its `chaosDurationSec` is outside the [Fault Parameter Limits](07-chaos-practice.md#fault-parameter-limits), and `fieldErrors` has the key `chaosDurationSec`. +- The safety guard refuses the scenario's fault, as on `POST /api/resilience`. + +While another disruption holds the cluster, the call gets `409 Conflict` instead. An unknown `id` gets `404 Not Found`. The `node-maintenance` scenario is listed, but every call to run it gets a `400`. A scenario can't name the node its `NODE_DRAIN` drains, and Kates sets no `TARGET_NODE`, so the Litmus `node-drain` experiment would drain the node of a random pod in any namespace. To drain a node, send a `NODE_DRAIN` to `POST /api/resilience` with the node in `chaosSpec.envOverrides.TARGET_NODE`, as [LitmusChaos Integration](07-chaos-practice.md#litmuschaos-integration) describes. diff --git a/docs/book/appendix-a-glossary.md b/docs/book/appendix-a-glossary.md index 52e82766..389ecdf5 100644 --- a/docs/book/appendix-a-glossary.md +++ b/docs/book/appendix-a-glossary.md @@ -586,7 +586,7 @@ The share of time a broker's request-handler threads are free: 0.3 means they ar ### Resilience Run {#gl-resilience-run} -A Kates test run with one fault injected after `steadyStateSec`, which reports the run's before-and-after impact and, for an INTEGRITY workload, lets the verifier measure RPO from the fault. Start one with `kates resilience run -f ` or `POST /api/resilience`, giving a `testRequest`, a `chaosSpec` and optional probes. Unlike a disruption plan, it sees what a client sees; it passes through no safety guard and has no rollback or SLA grade. See [Chaos Engineering in Practice](07-chaos-practice.md). +A Kates test run with one fault injected after `steadyStateSec`, which reports the run's before-and-after impact and, for an INTEGRITY workload, lets the verifier measure RPO from the fault. Start one with `kates resilience run -f ` or `POST /api/resilience`, giving a `testRequest`, a `chaosSpec` and optional probes. Unlike a disruption plan, it sees what a client sees; the safety guard only counts the brokers its fault hits and refuses it while another disruption holds the cluster, and it has no rollback or SLA grade. See [Chaos Engineering in Practice](07-chaos-practice.md). ### Resilience Test {#gl-resilience-test} @@ -608,7 +608,7 @@ Recovery time objective: how long recovery from a failure may take [@swanson2010 ### Safety Guard {#gl-safety-guard} -The check Kates runs on a disruption plan before any fault: it refuses the plan when no broker pod matches, when the steps would hit more brokers than `maxAffectedBrokers`, or when no broker would be left untouched. Before each step's fault it also requires every Kafka pod, controllers included, to be Running and Ready. It never counts controllers and doesn't check under-replicated partitions or the KRaft quorum, and a resilience run doesn't pass through it. See [Chaos Engineering in Practice](07-chaos-practice.md#safety-guardrails). +The check Kates runs on a disruption plan before any fault, and on a compound or resilience run before it starts: it refuses the run when no broker pod matches, when a plan's steps would hit more brokers than `maxAffectedBrokers`, or when no broker would be left untouched. Before each step's fault, a plan also requires every Kafka pod, controllers included, to be Running and Ready; a resilience run's fault gets no such check. It never counts controllers and doesn't check under-replicated partitions or the KRaft quorum. See [Chaos Engineering in Practice](07-chaos-practice.md#safety-guardrails). ### Saturation Point {#gl-saturation-point} diff --git a/docs/book/part-chaos-integrity.md b/docs/book/part-chaos-integrity.md index e6326901..d3a9ae27 100644 --- a/docs/book/part-chaos-integrity.md +++ b/docs/book/part-chaos-integrity.md @@ -34,7 +34,7 @@ This Part checks the loss target: `krafter` loses no acknowledged record when on A plan can't measure loss, so the verdicts come from two resilience files, `payments-integrity-broker.yaml` and `payments-integrity-zone.yaml`. Each runs an INTEGRITY test through one of the same two faults with `kates resilience run -f `, which prints that run's ID as `Test Run` and can return before the run ends. Wait until `kates test get ` shows the run `DONE`, and read its `Lost`, `RPO` and `Verdict` there. The target holds when `kates resilience run` reported `Status COMPLETED` and the INTEGRITY run shows `Lost` 0 with a measured RPO. When `Lost` is 0, an RPO of `not measured` means the fault never reached the run. -Run the zone file only once the broker file's INTEGRITY run is `DONE`, because the two runs share a topic and a consumer group. A resilience run's fault skips the safety guard, so first preview the zone's pods with `kates disruption playbook run az-failure --dry-run`. +Run the zone file only once the broker file's INTEGRITY run is `DONE`, because the two runs share a topic and a consumer group. The safety guard only checks that a resilience run's fault leaves a broker untouched, so first preview the zone's pods with `kates disruption playbook run az-failure --dry-run`. ## Practice diff --git a/docs/stretched-kafka/04-test-and-chaos-scenarios.md b/docs/stretched-kafka/04-test-and-chaos-scenarios.md index 3529a602..8a39f3a4 100644 --- a/docs/stretched-kafka/04-test-and-chaos-scenarios.md +++ b/docs/stretched-kafka/04-test-and-chaos-scenarios.md @@ -594,7 +594,7 @@ Correlated pod kills (T0) and the active-controller kill (T10) run as Kates faul Set `maxAffectedBrokers` to exactly the number of pods the selector should hit (four for AZ1 in A1, A2 and A7 at production size: one controller and three brokers), so a selector that matches more is rejected. `targetAll` outranks `targetBrokerId`, and on `main` fields a JSON spec leaves out get the builder defaults; the example still sets them explicitly. The `isrTrackingTopic` must exist: point it at the topic of an INTEGRITY run that writes during the plan. -A resilience run adds an INTEGRITY workload and is the only way to get `rpoMs`. It bypasses the safety guard, so dry-run the equivalent plan first: +A resilience run adds an INTEGRITY workload and is the only way to get `rpoMs`. It has no `maxAffectedBrokers`, and the safety guard refuses it only when it would leave no broker untouched, so dry-run the equivalent plan first: ```yaml # t0-integrity.yaml: replace __NAME__ and __SELECTOR__ @@ -707,7 +707,7 @@ Each test lists what to inject, the workload, the expected result per architectu 1. S1: one random broker (`strimzi.io/cluster=krafter,strimzi.io/broker-role=true`, `targetAll: false`, `targetBrokerId: -1`, `maxAffectedBrokers: 1`). 2. S2: one random controller (`strimzi.io/cluster=krafter,strimzi.io/controller-role=true`, same settings). 3. S3, S4, S5: every Kafka pod of one room, as three separate plans: `site=a` for AZ1 (A4 and A5 pods carry `zone=az1a` or `zone=az1b`), `zone=az2`, `zone=az3`. -4. A site-B transient: every Kafka pod with `site=b`, `maxAffectedBrokers` = the site-B pod count. In A6 that is every pod of KB, which the safety guard rejects as a plan: run it as a resilience run only, and check its selector with `kubectl get pods -l` instead of a dry run. +4. A site-B transient: every Kafka pod with `site=b`, `maxAffectedBrokers` = the site-B pod count. In A6 that is every pod of KB, which the safety guard rejects as a plan and as a resilience run: run it outside Kates, with `kubectl delete pod -l` as `make gameday` does, while a REST INTEGRITY run writes (no `rpoMs`), and check its selector with `kubectl get pods -l` first. The pods come back within seconds: this is not an outage test. @@ -739,7 +739,7 @@ for c in "az1 site=a" "az2 zone=az2" "az3 zone=az3" "siteb site=b"; do | .maxAffectedBrokers = $n | .isrTrackingTopic = $name | .steps[0].faultSpec.experimentName = $name | .steps[0].faultSpec.targetLabel = $sel' \ t0-plan.json > "$name.json" - # The resilience run skips the guard: check the targets with the plan's dry run + # The resilience run gets only the broker count: check the targets with the plan's dry run curl -sf -X POST "http://localhost:30083/api/disruptions?dryRun=true" \ -H "Authorization: Bearer $KEY_A" -H 'Content-Type: application/json' -d @"$name.json" | jq -e '.wouldSucceed' >/dev/null || { echo "rejected: $name"; break; } diff --git a/kates/src/main/java/com/bmscomp/kates/chaos/CompoundChaosOrchestrator.java b/kates/src/main/java/com/bmscomp/kates/chaos/CompoundChaosOrchestrator.java index 5d1264c8..0fc7464a 100644 --- a/kates/src/main/java/com/bmscomp/kates/chaos/CompoundChaosOrchestrator.java +++ b/kates/src/main/java/com/bmscomp/kates/chaos/CompoundChaosOrchestrator.java @@ -71,13 +71,18 @@ public CompoundOutcome executeConcurrent(List faults, int timeout List results = new ArrayList<>(); boolean allOk = true; - for (CompletableFuture f : futures) { + for (int i = 0; i < futures.size(); i++) { + // At least the wait a sequential run gives the fault: a shorter + // timeoutSec reported a longer fault as timed out while it still + // ran, and the disruption lease was given up with it still in. + FaultSpec spec = faults.get(i).spec(); + int waitSec = Math.max(timeoutSec, spec.delayBeforeSec() + spec.chaosDurationSec() + 120); try { - ProviderOutcome outcome = f.get(timeoutSec, TimeUnit.SECONDS); + ProviderOutcome outcome = futures.get(i).get(waitSec, TimeUnit.SECONDS); results.add(outcome); if (!outcome.succeeded()) allOk = false; } catch (TimeoutException e) { - results.add(new ProviderOutcome("unknown", "unknown", false, "Timed out after " + timeoutSec + "s")); + results.add(new ProviderOutcome("unknown", "unknown", false, "Timed out after " + waitSec + "s")); allOk = false; } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -171,10 +176,10 @@ public List availableProviders() { /** * Refuses the whole run, before any fault is triggered, when one fault has * no spec or a parameter outside the chaos limits. A compound run calls the - * providers itself, past the coordinator and the safety guard, so this is - * the only check its faults get. + * providers itself, past the coordinator; its resource asks this first, then + * the safety guard, and holds the disruption lease while the faults run. */ - private void checkLimits(List faults) { + public void checkLimits(List faults) { List refused = new ArrayList<>(); for (int i = 0; i < faults.size(); i++) { String path = "faults[" + i + "].faultSpec"; diff --git a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionAnalysisResource.java b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionAnalysisResource.java index 9be7a23d..b099cfc9 100644 --- a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionAnalysisResource.java +++ b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionAnalysisResource.java @@ -1,7 +1,9 @@ package com.bmscomp.kates.disruption; +import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.UUID; import jakarta.annotation.security.RolesAllowed; import jakarta.inject.Inject; import jakarta.ws.rs.Consumes; @@ -38,6 +40,12 @@ public class DisruptionAnalysisResource { @Inject com.bmscomp.kates.chaos.CompoundChaosOrchestrator compoundOrchestrator; + @Inject + DisruptionSafetyGuard safetyGuard; + + @Inject + DisruptionConcurrencyGuard concurrencyGuard; + @Inject DisruptionReportRepository repository; @@ -115,8 +123,12 @@ public Response historyByTopic( @APIResponse(responseCode = "200", description = "Each fault's outcome") @APIResponse( responseCode = "400", - description = "No fault, or a fault without a faultSpec or with a parameter outside the chaos limits;" - + " no fault is triggered") + description = "No fault, a fault without a faultSpec or with a parameter outside the chaos limits, or" + + " faults the safety guard refuses, such as faults that would hit every broker; no fault is" + + " triggered") + @APIResponse( + responseCode = "409", + description = "Another disruption is running against this cluster; no fault is triggered") public Response executeCompound(DisruptionDtos.CompoundChaosRequest request) { if (request.faults == null || request.faults.isEmpty()) { return Response.status(400) @@ -129,15 +141,54 @@ public Response executeCompound(DisruptionDtos.CompoundChaosRequest request) { f.faultSpec, f.providerName != null ? f.providerName : "kubernetes")) .toList(); - var outcome = request.sequential - ? compoundOrchestrator.executeSequential(faults, request.delayBetweenSec) - : compoundOrchestrator.executeConcurrent(faults, request.timeoutSec); + // The limits first, so a fault without a spec is refused by its place: + // the guard reads every spec. + compoundOrchestrator.checkLimits(faults); + + // Each fault is counted as a plan step, aimed as posted. A compound + // fault is never aimed at a topic's leader, so targetTopic is dropped: + // the guard would count the leader instead. + DisruptionPlan plan = new DisruptionPlan(); + plan.setName("compound"); + List steps = new ArrayList<>(); + for (int i = 0; i < faults.size(); i++) { + var spec = faults.get(i).spec().toBuilder().targetTopic("").build(); + steps.add(new DisruptionPlan.DisruptionStep("faults[" + i + "]", spec, 0, 0, false)); + } + plan.setSteps(steps); + DisruptionSafetyGuard.ValidationResult validation = safetyGuard.validatePlan(plan); + if (!validation.safe()) { + return Response.status(400) + .entity(ApiError.of( + 400, "Bad Request", "No fault was triggered: " + String.join("; ", validation.errors()))) + .build(); + } - return Response.ok(Map.of( - "allSucceeded", outcome.allSucceeded(), - "mode", request.sequential ? "sequential" : "concurrent", - "results", outcome.results())) - .build(); + // The lease a plan holds, so a compound run and a plan don't take + // down brokers together or undo each other's faults. + String target = concurrencyGuard.currentTarget(); + String token = "compound#" + UUID.randomUUID(); + if (!concurrencyGuard.tryAcquire(target, token)) { + return Response.status(409) + .entity(ApiError.of( + 409, + "Conflict", + "Another disruption is running against " + target + ", so no fault was triggered")) + .build(); + } + try { + var outcome = request.sequential + ? compoundOrchestrator.executeSequential(faults, request.delayBetweenSec) + : compoundOrchestrator.executeConcurrent(faults, request.timeoutSec); + + return Response.ok(Map.of( + "allSucceeded", outcome.allSucceeded(), + "mode", request.sequential ? "sequential" : "concurrent", + "results", outcome.results())) + .build(); + } finally { + concurrencyGuard.release(target, token); + } } @GET diff --git a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionConcurrencyGuard.java b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionConcurrencyGuard.java index c4295c06..c352e555 100644 --- a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionConcurrencyGuard.java +++ b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionConcurrencyGuard.java @@ -7,8 +7,8 @@ import org.eclipse.microprofile.config.inject.ConfigProperty; /** - * Serialises disruption plans per target cluster: at most one plan may be - * running against a given namespace/cluster at a time. + * Serialises plans, compound runs and resilience faults per target cluster: + * at most one of them may be running against a given namespace/cluster. * *

Why. Plans are executed asynchronously, so nothing stopped two of * them overlapping. Rollback state is stored on the target itself — the diff --git a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionLauncher.java b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionLauncher.java index 20e1445a..8c5be9e3 100644 --- a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionLauncher.java +++ b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionLauncher.java @@ -70,8 +70,8 @@ public LaunchResult launch(DisruptionPlan plan) { return new LaunchResult( Status.CONFLICT, null, - List.of("A disruption plan is already running against " + target - + ". Wait for it to finish before starting another.")); + List.of("Another disruption (a plan, a compound run or a resilience run's fault) is running" + + " against " + target + ". Wait for it to finish before starting another.")); } String id = UUID.randomUUID().toString().substring(0, 8); diff --git a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrchestrator.java b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrchestrator.java index 5aa7dcf6..0b130ed6 100644 --- a/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrchestrator.java +++ b/kates/src/main/java/com/bmscomp/kates/disruption/DisruptionOrchestrator.java @@ -74,9 +74,9 @@ public class DisruptionOrchestrator { int recoveryTimeoutSec; /** - * Executes a plan, refusing to start while another plan is running against - * the same cluster. Every entry point (API, playbook, scheduler) funnels - * through here, so the guard cannot be bypassed by adding a caller. + * Executes a plan, refusing to start while another disruption holds the + * cluster's lease. Plans from every entry point take it here; compound runs + * and resilience faults take the same lease themselves. */ public DisruptionReport execute(DisruptionPlan plan) { String target = concurrencyGuard.currentTarget(); @@ -90,7 +90,7 @@ public DisruptionReport execute(DisruptionPlan plan) { DisruptionReport rejected = new DisruptionReport(); rejected.setPlanName(plan.getName()); rejected.setStatus("REJECTED"); - rejected.setValidationWarnings(List.of("ERROR: another disruption plan is already running against " + rejected.setValidationWarnings(List.of("ERROR: another disruption is already running against " + target + " (" + holder + "). Concurrent plans race the rollback" + " state stored on the target, which can leave the cluster under-scaled.")); return rejected; diff --git a/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceOrchestrator.java b/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceOrchestrator.java index c8b6c8f1..70db5215 100644 --- a/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceOrchestrator.java +++ b/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceOrchestrator.java @@ -15,6 +15,7 @@ import org.jboss.logging.Logger; import com.bmscomp.kates.chaos.*; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; import com.bmscomp.kates.domain.TestResult; import com.bmscomp.kates.domain.TestRun; import com.bmscomp.kates.engine.TestOrchestrator; @@ -30,7 +31,7 @@ * 1. Start benchmark, and wait for its tasks to be submitted * 2. Wait for steady-state period * 3. Evaluate baseline probes - * 4. Inject fault via ChaosCoordinator + * 4. Take the disruption lease, held to the end, and inject fault via ChaosCoordinator * 5. Run continuous probes during chaos * 6. Wait for fault to complete * 7. Measure recovery time (RTO) via probe polling @@ -58,6 +59,9 @@ public class ResilienceOrchestrator { @Inject ProbeExecutor probeExecutor; + @Inject + DisruptionConcurrencyGuard concurrencyGuard; + /** * How long to wait for a run's tasks to be submitted. Creating its topic * can take about two minutes to fail (four attempts of up to 30 s), and a @@ -71,6 +75,10 @@ public class ResilienceOrchestrator { @ActivateRequestContext public ResilienceReport execute(ResilienceTestRequest request) { ResilienceReport report = new ResilienceReport(); + String target = concurrencyGuard.currentTarget(); + // Set when the fault is about to go in; release() frees the lease only + // while this token holds it. + String leaseToken = null; try { List probes = resolveProbes(request); @@ -122,7 +130,13 @@ public ResilienceReport execute(ResilienceTestRequest request) { LOG.infof("Baseline probes: %d/%d passed", passCount, baseline.size()); } - // 4. Inject fault + start continuous probes + // 4. Take the lease a plan takes, then inject fault + start + // continuous probes. It is held through the recovery wait, so a + // plan's fault or rollback doesn't land on this one's. + leaseToken = "resilience:" + run.getId() + "#" + UUID.randomUUID(); + if (!concurrencyGuard.tryAcquire(target, leaseToken)) { + return refuse(report, "Another disruption is running against " + target + ", so no fault was injected"); + } LOG.info("Resilience test: injecting fault '" + request.getChaosSpec().experimentName() + "'"); @@ -221,6 +235,10 @@ public ResilienceReport execute(ResilienceTestRequest request) { report.setError( e.getMessage() != null ? e.getMessage() : e.getClass().getName()); LOG.error("Resilience test failed", e); + } finally { + if (leaseToken != null) { + concurrencyGuard.release(target, leaseToken); + } } return report; diff --git a/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceResource.java b/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceResource.java index 60067b9c..cf48f7aa 100644 --- a/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceResource.java +++ b/kates/src/main/java/com/bmscomp/kates/resilience/ResilienceResource.java @@ -1,5 +1,6 @@ package com.bmscomp.kates.resilience; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.stream.Collectors; @@ -20,6 +21,9 @@ import com.bmscomp.kates.audit.Audited; import com.bmscomp.kates.chaos.FaultLimits; import com.bmscomp.kates.chaos.FaultSpec; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; +import com.bmscomp.kates.disruption.DisruptionPlan; +import com.bmscomp.kates.disruption.DisruptionSafetyGuard; import com.bmscomp.kates.domain.TestSpec; import com.bmscomp.kates.security.Scopes; @@ -53,6 +57,12 @@ public class ResilienceResource { @Inject com.bmscomp.kates.engine.SpecLimits specLimits; + @Inject + DisruptionSafetyGuard safetyGuard; + + @Inject + DisruptionConcurrencyGuard concurrencyGuard; + /** * A 400 naming each parameter of the chaos spec outside the chaos limits, * or empty when all are within them. Asked before the stream starts: the @@ -162,6 +172,38 @@ private static Optional unparsedTargetLabel(FaultSpec chaosSpec) { } } + /** + * A 409 while another disruption holds the cluster, or a 400 naming each + * reason the safety guard refuses the fault, counted as a one-step plan, + * or empty. Asked last, since it reads the cluster, and before the stream + * starts. The run takes the lease itself, just before the fault goes in. + */ + private Optional unsafeOrBusy(FaultSpec chaosSpec) { + String target = concurrencyGuard.currentTarget(); + if (concurrencyGuard.isBusy(target)) { + return Optional.of(Response.status(409) + .entity(ApiError.of( + 409, + "Conflict", + "Another disruption is running against " + target + ", so no fault was injected")) + .build()); + } + // A resilience fault is never aimed at a topic's leader, so + // targetTopic is dropped: the guard would count the leader instead. + DisruptionPlan plan = new DisruptionPlan(); + plan.setName("resilience"); + plan.setSteps(List.of(new DisruptionPlan.DisruptionStep( + "chaosSpec", chaosSpec.toBuilder().targetTopic("").build(), 0, 0, false))); + DisruptionSafetyGuard.ValidationResult validation = safetyGuard.validatePlan(plan); + if (validation.safe()) { + return Optional.empty(); + } + return Optional.of(Response.status(400) + .entity(ApiError.of( + 400, "Bad Request", "No fault was injected: " + String.join("; ", validation.errors()))) + .build()); + } + @Audited(action = "RUN", type = "resilience") @RolesAllowed(Scopes.CHAOS_RUN) @POST @@ -173,7 +215,11 @@ private static Optional unparsedTargetLabel(FaultSpec chaosSpec) { responseCode = "400", description = "Invalid request, including a testRequest with no type, a testRequest spec value outside" + " its limits, a testRequest spec field the test type or backend cannot apply, or a chaosSpec" - + " parameter outside the chaos limits; fieldErrors names each field") + + " parameter outside the chaos limits, where fieldErrors names each field; or a chaosSpec the" + + " safety guard refuses, such as one that would hit every broker") + @APIResponse( + responseCode = "409", + description = "Another disruption is running against this cluster; no test is started") public Response executeResilienceTest(ResilienceTestRequest request) { if (request.getTestRequest() == null) { return Response.status(400) @@ -194,6 +240,10 @@ public Response executeResilienceTest(ResilienceTestRequest request) { if (outsideLimits.isPresent()) { return outsideLimits.get(); } + Optional unsafeOrBusy = unsafeOrBusy(request.getChaosSpec()); + if (unsafeOrBusy.isPresent()) { + return unsafeOrBusy.get(); + } StreamingOutput stream = executeWithKeepAlive(request, null); return Response.ok(stream).build(); @@ -224,8 +274,12 @@ public Response listScenarios() { description = "No testRequest, or the node-maintenance scenario, which can't name the node to drain;" + " or, with fieldErrors naming each field, a testRequest spec value outside its limits, a" + " testRequest spec field the test type or backend cannot apply, a targetLabel that isn't a" - + " label selector, or a chaosDurationSec outside the chaos limits") + + " label selector, or a chaosDurationSec outside the chaos limits; or a fault the safety guard" + + " refuses, such as one that would hit every broker") @APIResponse(responseCode = "404", description = "Scenario not found") + @APIResponse( + responseCode = "409", + description = "Another disruption is running against this cluster; no test is started") public Response runScenario( @Parameter(description = "Scenario ID") @PathParam("id") String id, Map overrides) { @@ -282,6 +336,10 @@ public Response runScenario( if (refused.isPresent()) { return refused.get(); } + Optional unsafeOrBusy = unsafeOrBusy(request.getChaosSpec()); + if (unsafeOrBusy.isPresent()) { + return unsafeOrBusy.get(); + } StreamingOutput stream = executeWithKeepAlive(request, id); return Response.ok(stream).build(); diff --git a/kates/src/test/java/com/bmscomp/kates/chaos/CompoundChaosOrchestratorTest.java b/kates/src/test/java/com/bmscomp/kates/chaos/CompoundChaosOrchestratorTest.java index f36c71d2..3ccdd059 100644 --- a/kates/src/test/java/com/bmscomp/kates/chaos/CompoundChaosOrchestratorTest.java +++ b/kates/src/test/java/com/bmscomp/kates/chaos/CompoundChaosOrchestratorTest.java @@ -114,6 +114,28 @@ void aSequentialRunWaitsForEachFaultsDelayAndDurationAsWell() throws Exception { verify(kill).get(KILL.chaosDurationSec() + 120, TimeUnit.SECONDS); } + /** + * A concurrent run waited for each fault only timeoutSec, 120 s by + * default, whatever its delay and duration. A longer fault was reported + * timed out while it still ran, and the resource gave up the disruption + * lease with it still in. + */ + @Test + void aConcurrentFaultLongerThanTimeoutSecIsWaitedFor() { + ChaosProvider slow = mock(ChaosProvider.class); + when(slow.name()).thenReturn("slow"); + when(slow.isAvailable()).thenReturn(true); + when(slow.triggerFault(any())) + .thenReturn(CompletableFuture.supplyAsync( + () -> passed("kill").join(), CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS))); + when(orchestrator.providers.iterator()) + .thenAnswer(invocation -> List.of(slow).iterator()); + + var outcome = orchestrator.executeConcurrent(List.of(new CompoundFault(KILL, "slow")), 0); + + assertTrue(outcome.allSucceeded(), outcome.toString()); + } + private static CompletableFuture passed(String experimentName) { Instant now = Instant.now(); return spy(CompletableFuture.completedFuture( diff --git a/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionAnalysisResourceCompoundTest.java b/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionAnalysisResourceCompoundTest.java new file mode 100644 index 00000000..ca3038c3 --- /dev/null +++ b/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionAnalysisResourceCompoundTest.java @@ -0,0 +1,142 @@ +package com.bmscomp.kates.disruption; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.*; + +import java.util.ArrayList; +import java.util.List; +import jakarta.ws.rs.core.Response; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import com.bmscomp.kates.api.ApiError; +import com.bmscomp.kates.chaos.CompoundChaosOrchestrator; +import com.bmscomp.kates.chaos.CompoundChaosOrchestrator.CompoundOutcome; +import com.bmscomp.kates.chaos.DisruptionType; +import com.bmscomp.kates.chaos.FaultSpec; + +/** + * A compound run goes through the safety guard and takes the lease a plan + * takes. It called the providers past both: a POD_KILL with targetAll took + * down every broker at once, which the same fault as a plan step is refused + * for, and it ran on top of a plan's faults. + */ +class DisruptionAnalysisResourceCompoundTest { + + private final DisruptionAnalysisResource resource = new DisruptionAnalysisResource(); + + private final DisruptionConcurrencyGuard guard = new DisruptionConcurrencyGuard(); + + private static final FaultSpec EVERY_BROKER = FaultSpec.builder("kill") + .disruptionType(DisruptionType.POD_KILL) + .targetAll(true) + .build(); + + @BeforeEach + void wire() { + guard.kafkaNamespace = "kafka"; + guard.kafkaCluster = "krafter"; + resource.concurrencyGuard = guard; + resource.compoundOrchestrator = mock(CompoundChaosOrchestrator.class); + resource.safetyGuard = mock(DisruptionSafetyGuard.class); + when(resource.safetyGuard.validatePlan(any())).thenReturn(DisruptionSafetyGuard.ValidationResult.ok(List.of())); + when(resource.compoundOrchestrator.executeConcurrent(any(), anyInt())) + .thenReturn(new CompoundOutcome(true, List.of())); + } + + private static DisruptionDtos.CompoundChaosRequest request(FaultSpec... specs) { + DisruptionDtos.CompoundChaosRequest request = new DisruptionDtos.CompoundChaosRequest(); + request.faults = new ArrayList<>(); + for (FaultSpec spec : specs) { + DisruptionDtos.CompoundFaultEntry entry = new DisruptionDtos.CompoundFaultEntry(); + entry.faultSpec = spec; + request.faults.add(entry); + } + return request; + } + + private void assertNoFault() { + verify(resource.compoundOrchestrator, never()).executeConcurrent(any(), anyInt()); + verify(resource.compoundOrchestrator, never()).executeSequential(any(), anyInt()); + } + + @Test + void anUnsafeCompoundRunIsRefusedBeforeAnyFault() { + when(resource.safetyGuard.validatePlan(any())) + .thenReturn(DisruptionSafetyGuard.ValidationResult.rejected( + List.of(), List.of("Plan would affect ALL 3 brokers — cluster would lose availability"))); + + Response response = resource.executeCompound(request(EVERY_BROKER)); + + assertEquals(400, response.getStatus()); + assertEquals( + "No fault was triggered: Plan would affect ALL 3 brokers — cluster would lose availability", + ((ApiError) response.getEntity()).getMessage()); + assertNoFault(); + } + + @Test + void aCompoundRunWhileTheLeaseIsHeldIsAConflict() { + assertTrue(guard.tryAcquire(guard.currentTarget(), "leader-cascade#1")); + + Response response = resource.executeCompound(request(EVERY_BROKER)); + + assertEquals(409, response.getStatus()); + assertEquals( + "Another disruption is running against kafka/krafter, so no fault was triggered", + ((ApiError) response.getEntity()).getMessage()); + assertNoFault(); + assertEquals( + "leader-cascade#1", guard.activeToken(guard.currentTarget()).orElseThrow()); + } + + @Test + void theLeaseIsHeldWhileTheFaultsRunAndReleasedAfter() { + List busy = new ArrayList<>(); + when(resource.compoundOrchestrator.executeConcurrent(any(), anyInt())).thenAnswer(invocation -> { + busy.add(guard.isBusy(guard.currentTarget())); + return new CompoundOutcome(true, List.of()); + }); + + assertEquals(200, resource.executeCompound(request(EVERY_BROKER)).getStatus()); + assertEquals(List.of(true), busy); + assertFalse(guard.isBusy(guard.currentTarget()), "released after a run that answered"); + + doThrow(new IllegalStateException("the provider broke")) + .when(resource.compoundOrchestrator) + .executeConcurrent(any(), anyInt()); + + assertThrows(IllegalStateException.class, () -> resource.executeCompound(request(EVERY_BROKER))); + assertFalse(guard.isBusy(guard.currentTarget()), "released after a run that threw"); + } + + /** + * The guard replaces a step's targetBrokerId with the leader of its + * targetTopic, but a compound fault is never aimed by topic, so it would + * count the wrong broker. + */ + @Test + void eachFaultIsAStepCountedWithoutItsTargetTopic() { + FaultSpec byTopic = FaultSpec.builder("kill-1") + .disruptionType(DisruptionType.POD_KILL) + .targetBrokerId(1) + .targetTopic("orders") + .build(); + + resource.executeCompound(request(EVERY_BROKER, byTopic)); + + ArgumentCaptor plan = ArgumentCaptor.forClass(DisruptionPlan.class); + verify(resource.safetyGuard).validatePlan(plan.capture()); + List steps = plan.getValue().getSteps(); + assertEquals( + List.of("faults[0]", "faults[1]"), + steps.stream().map(DisruptionPlan.DisruptionStep::name).toList()); + assertEquals("", steps.get(1).faultSpec().targetTopic()); + assertEquals(byTopic.toBuilder().targetTopic("").build(), steps.get(1).faultSpec()); + assertEquals(EVERY_BROKER, steps.get(0).faultSpec()); + } +} diff --git a/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionResourceTest.java b/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionResourceTest.java index 2df2b3b2..3f757b91 100644 --- a/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionResourceTest.java +++ b/kates/src/test/java/com/bmscomp/kates/disruption/DisruptionResourceTest.java @@ -73,6 +73,28 @@ void aCompoundRunWithAFaultOutsideTheChaosLimitsIsRefused() { startsWith("No fault was triggered: faults[1].faultSpec.chaosDurationSec 0 is below 1")); } + /** + * A compound run within the limits goes to the safety guard, which counts + * the brokers its faults hit. In CI there is no cluster, so it counts + * none; on a machine whose kubeconfig reaches one, the targetAll kill + * hits every broker. Either way it is refused, and nothing is killed. + */ + @Test + void aCompoundRunWithNoBrokerToCountIsRefused() { + given().contentType("application/json") + .body("{\"faults\":[{\"faultSpec\":{\"experimentName\":\"kill\",\"disruptionType\":\"POD_KILL\"," + + "\"targetAll\":true}}]}") + .when() + .post("/api/disruptions/compound") + .then() + .statusCode(400) + .body( + "message", + anyOf( + startsWith("No fault was triggered: No broker pods found"), + startsWith("No fault was triggered: Plan would affect ALL "))); + } + @Test void executeDisruptionReturns400WithEmptyPlan() { given().contentType("application/json") diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorBenchmarkStatusTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorBenchmarkStatusTest.java index 5d6ba312..a96dd83a 100644 --- a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorBenchmarkStatusTest.java +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorBenchmarkStatusTest.java @@ -24,6 +24,7 @@ import com.bmscomp.kates.chaos.ProbeExecutor; import com.bmscomp.kates.chaos.ProbeResult; import com.bmscomp.kates.chaos.ProbeSpec; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; import com.bmscomp.kates.domain.CreateTestRequest; import com.bmscomp.kates.domain.ScenarioPhase; import com.bmscomp.kates.domain.TestResult; @@ -78,6 +79,7 @@ void wire() { orchestrator.probeExecutor = mock(ProbeExecutor.class); when(orchestrator.probeExecutor.evaluateAll(any(), any())).thenReturn(List.of(ProbeResult.pass("up", "", 1))); + orchestrator.concurrencyGuard = new DisruptionConcurrencyGuard(); orchestrator.submissionPollIntervalMs = 1; } diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorChaosStartTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorChaosStartTest.java index d01ed021..583627bd 100644 --- a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorChaosStartTest.java +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorChaosStartTest.java @@ -27,6 +27,7 @@ import com.bmscomp.kates.chaos.ProbeExecutor; import com.bmscomp.kates.chaos.ProbeResult; import com.bmscomp.kates.chaos.ProbeSpec; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; import com.bmscomp.kates.domain.TestResult; import com.bmscomp.kates.domain.TestRun; import com.bmscomp.kates.domain.TestSpec; @@ -59,6 +60,7 @@ void wire() { // A passing Edge probe keeps the run off the fixed 10s no-probe wait. orchestrator.probeExecutor = mock(ProbeExecutor.class); when(orchestrator.probeExecutor.evaluateAll(any(), any())).thenReturn(List.of(ProbeResult.pass("up", "", 1))); + orchestrator.concurrencyGuard = new DisruptionConcurrencyGuard(); } private static ChaosOutcome passed() { diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorLeaseTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorLeaseTest.java new file mode 100644 index 00000000..42ff5027 --- /dev/null +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorLeaseTest.java @@ -0,0 +1,127 @@ +package com.bmscomp.kates.resilience; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +import java.time.Instant; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.bmscomp.kates.chaos.ChaosCoordinator; +import com.bmscomp.kates.chaos.ChaosOutcome; +import com.bmscomp.kates.chaos.FaultSpec; +import com.bmscomp.kates.chaos.ProbeExecutor; +import com.bmscomp.kates.chaos.ProbeResult; +import com.bmscomp.kates.chaos.ProbeSpec; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; +import com.bmscomp.kates.domain.CreateTestRequest; +import com.bmscomp.kates.domain.TestResult.TaskStatus; +import com.bmscomp.kates.domain.TestRun; +import com.bmscomp.kates.domain.TestSpec; +import com.bmscomp.kates.domain.TestType; +import com.bmscomp.kates.engine.TestOrchestrator; +import com.bmscomp.kates.report.ReportGenerator; +import com.bmscomp.kates.report.TestReport; +import com.bmscomp.kates.util.Result; + +/** + * A resilience run's fault takes the lease a disruption plan takes. It went in + * whatever else was running: a plan killing broker 0 and a resilience run + * killing broker 1 took down two of three brokers together, and either one's + * cleanup undid the other's fault. + */ +class ResilienceOrchestratorLeaseTest { + + private final ResilienceOrchestrator orchestrator = new ResilienceOrchestrator(); + + private final DisruptionConcurrencyGuard guard = new DisruptionConcurrencyGuard(); + + private final String target = guard.currentTarget(); + + /** Whether the lease was held, at each probe evaluation and when the fault went in. */ + private final List seen = new CopyOnWriteArrayList<>(); + + @BeforeEach + void wire() { + TestRun running = new TestRun(TestType.LOAD, new TestSpec()).withStatus(TaskStatus.RUNNING); + orchestrator.testOrchestrator = mock(TestOrchestrator.class); + when(orchestrator.testOrchestrator.executeTest(any())).thenReturn(Result.success(running)); + when(orchestrator.testOrchestrator.refreshStatus(running.getId())).thenReturn(running); + + Instant now = Instant.now(); + orchestrator.chaosCoordinator = mock(ChaosCoordinator.class); + when(orchestrator.chaosCoordinator.triggerFault(any(), any())).thenAnswer(invocation -> { + seen.add("fault " + guard.isBusy(target)); + return CompletableFuture.completedFuture( + ChaosOutcome.success("engine", "pod-kill", now, now, System.nanoTime(), null, null, null)); + }); + + orchestrator.reportGenerator = mock(ReportGenerator.class); + when(orchestrator.reportGenerator.generate(any())).thenReturn(new TestReport()); + + // An Edge probe: evaluated for the baseline, then by the recovery wait. + orchestrator.probeExecutor = mock(ProbeExecutor.class); + when(orchestrator.probeExecutor.evaluateAll(any(), any())).thenAnswer(invocation -> { + seen.add("probes " + guard.isBusy(target)); + return List.of(ProbeResult.pass("up", "", 1)); + }); + + orchestrator.concurrencyGuard = guard; + } + + private ResilienceReport execute() { + ResilienceTestRequest request = new ResilienceTestRequest(); + request.setTestRequest(new CreateTestRequest()); + request.setSteadyStateSec(0); + request.setMaxRecoveryWaitSec(5); + request.setProbes(List.of(ProbeSpec.builder("up").build())); + request.setChaosSpec(FaultSpec.builder("pod-kill") + .targetNamespace("kafka") + .chaosDurationSec(1) + .build()); + return orchestrator.execute(request); + } + + @Test + void aFaultWhileAnotherDisruptionHoldsTheLeaseIsNotInjected() { + assertTrue(guard.tryAcquire(target, "leader-cascade#1")); + + ResilienceReport report = execute(); + + assertEquals("ERROR", report.getStatus()); + assertEquals( + "Another disruption is running against " + target + ", so no fault was injected", report.getError()); + verify(orchestrator.chaosCoordinator, never()).triggerFault(any(), any()); + assertEquals("leader-cascade#1", guard.activeToken(target).orElseThrow(), "the plan keeps its lease"); + } + + /** Taken after the baseline probes, and held from the fault to the last poll of the recovery wait. */ + @Test + void theLeaseIsHeldFromTheFaultThroughRecovery() { + ResilienceReport report = execute(); + + assertEquals("COMPLETED", report.getStatus()); + assertEquals(List.of("probes false", "fault true", "probes true"), seen); + } + + @Test + void theLeaseIsReleasedWhenTheRunEndsOrFails() { + assertEquals("COMPLETED", execute().getStatus()); + assertFalse(guard.isBusy(target), "released after a run that ended"); + + doThrow(new IllegalStateException("the provider broke")) + .when(orchestrator.chaosCoordinator) + .triggerFault(any(), any()); + + ResilienceReport failed = execute(); + + assertEquals("ERROR", failed.getStatus()); + assertEquals("the provider broke", failed.getError()); + assertFalse(guard.isBusy(target), "released after a run that failed"); + } +} diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorRecoveryTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorRecoveryTest.java index 7d30b6fe..039e6b79 100644 --- a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorRecoveryTest.java +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorRecoveryTest.java @@ -20,6 +20,7 @@ import com.bmscomp.kates.chaos.ProbeExecutor; import com.bmscomp.kates.chaos.ProbeResult; import com.bmscomp.kates.chaos.ProbeSpec; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; import com.bmscomp.kates.domain.CreateTestRequest; import com.bmscomp.kates.domain.TestResult.TaskStatus; import com.bmscomp.kates.domain.TestRun; @@ -66,6 +67,7 @@ void wire() { when(orchestrator.reportGenerator.generate(any())).thenReturn(new TestReport()); orchestrator.probeExecutor = mock(ProbeExecutor.class); + orchestrator.concurrencyGuard = new DisruptionConcurrencyGuard(); orchestrator.recoveryPollIntervalMs = 1; } diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorTestRunIdTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorTestRunIdTest.java index c357eaba..ee759272 100644 --- a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorTestRunIdTest.java +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceOrchestratorTestRunIdTest.java @@ -19,6 +19,7 @@ import com.bmscomp.kates.chaos.ProbeExecutor; import com.bmscomp.kates.chaos.ProbeResult; import com.bmscomp.kates.chaos.ProbeSpec; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; import com.bmscomp.kates.domain.CreateTestRequest; import com.bmscomp.kates.domain.TestResult.TaskStatus; import com.bmscomp.kates.domain.TestRun; @@ -67,6 +68,7 @@ void wire() { orchestrator.probeExecutor = mock(ProbeExecutor.class); when(orchestrator.probeExecutor.evaluateAll(any(), any())).thenReturn(List.of(ProbeResult.pass("up", "", 1))); + orchestrator.concurrencyGuard = new DisruptionConcurrencyGuard(); orchestrator.submissionPollIntervalMs = 1; } diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceStreamTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceStreamTest.java index c0a9ef8f..c485e36e 100644 --- a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceStreamTest.java +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceStreamTest.java @@ -21,6 +21,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import com.bmscomp.kates.disruption.DisruptionSafetyGuard; + /** * The answer to a resilience run that gets through every check: a space every * 10 s while the run goes on, then its report. The orchestrator is a mock, so @@ -35,6 +37,10 @@ class ResilienceResourceStreamTest { @InjectMock ResilienceOrchestrator orchestrator; + /** Lets the fault through: in CI the real guard finds no broker, and refuses every fault. */ + @InjectMock + DisruptionSafetyGuard safetyGuard; + /** Held so the logger, and the handler on it, outlive the test's own references. */ private final java.util.logging.Logger resourceLog = java.util.logging.Logger.getLogger(ResilienceResource.class.getName()); @@ -46,6 +52,11 @@ void captureLogs() { resourceLog.addHandler(captured); } + @BeforeEach + void aSafeFault() { + when(safetyGuard.validatePlan(any())).thenReturn(DisruptionSafetyGuard.ValidationResult.ok(List.of())); + } + @AfterEach void releaseLogs() { resourceLog.removeHandler(captured); diff --git a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceTest.java b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceTest.java index de4a1998..faa0e9ac 100644 --- a/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceTest.java +++ b/kates/src/test/java/com/bmscomp/kates/resilience/ResilienceResourceTest.java @@ -3,20 +3,26 @@ import static io.restassured.RestAssured.given; import static org.hamcrest.Matchers.*; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.util.List; import java.util.Map; import java.util.Set; +import jakarta.inject.Inject; import io.quarkus.test.InjectMock; import io.quarkus.test.junit.QuarkusTest; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import com.bmscomp.kates.chaos.DisruptionType; +import com.bmscomp.kates.disruption.DisruptionConcurrencyGuard; +import com.bmscomp.kates.disruption.DisruptionSafetyGuard; import com.bmscomp.kates.domain.TestType; @QuarkusTest @@ -26,6 +32,20 @@ class ResilienceResourceTest { @InjectMock ResilienceOrchestrator orchestrator; + @InjectMock + DisruptionSafetyGuard safetyGuard; + + @Inject + DisruptionConcurrencyGuard concurrencyGuard; + + private static final String EVERY_BROKER = "Plan would affect ALL 3 brokers — cluster would lose availability"; + + /** A fault the guard lets through: in CI the real one finds no broker, and refuses every fault. */ + @BeforeEach + void aSafeFault() { + when(safetyGuard.validatePlan(any())).thenReturn(DisruptionSafetyGuard.ValidationResult.ok(List.of())); + } + @Test void postResilienceReturnsBadRequestWithoutTestRequest() { given().contentType("application/json") @@ -446,6 +466,71 @@ void aScenarioTargetLabelThatIsNotASelectorIsRefusedByName() { verify(orchestrator, never()).execute(any()); } + /** + * A fault the safety guard refuses, here one that would hit every broker, + * is refused before the stream starts, naming why. It went in past the + * guard, which refuses the same fault as a plan step. + */ + @Test + void anUnsafeFaultIsRefusedBeforeTheStream() { + when(safetyGuard.validatePlan(any())) + .thenReturn(DisruptionSafetyGuard.ValidationResult.rejected(List.of(), List.of(EVERY_BROKER))); + + given().contentType("application/json") + .body("{\"testRequest\":{\"type\":\"LOAD\"},\"chaosSpec\":{\"experimentName\":\"kill\"," + + "\"disruptionType\":\"POD_KILL\",\"targetAll\":true}}") + .when() + .post("/api/resilience") + .then() + .statusCode(400) + .body("error", is("Bad Request")) + .body("message", is("No fault was injected: " + EVERY_BROKER)); + given().contentType("application/json") + .body("{\"testRequest\":{\"type\":\"LOAD\"}}") + .when() + .post("/api/resilience/scenarios/broker-crash") + .then() + .statusCode(400) + .body("message", is("No fault was injected: " + EVERY_BROKER)); + verify(orchestrator, never()).execute(any()); + } + + /** + * A run started while a disruption plan holds the cluster is refused + * before the stream starts. Its fault went in on top of the plan's: a + * broker each, two of three down together. + */ + @Test + void aRunWhileADisruptionHoldsTheLeaseIsAConflict() { + String target = concurrencyGuard.currentTarget(); + assertTrue(concurrencyGuard.tryAcquire(target, "leader-cascade#1")); + try { + given().contentType("application/json") + .body("{\"testRequest\":{\"type\":\"LOAD\"},\"chaosSpec\":{\"experimentName\":\"kill\"," + + "\"disruptionType\":\"POD_KILL\",\"chaosDurationSec\":30}}") + .when() + .post("/api/resilience") + .then() + .statusCode(409) + .body("error", is("Conflict")) + .body( + "message", + is("Another disruption is running against " + target + ", so no fault was injected")); + given().contentType("application/json") + .body("{\"testRequest\":{\"type\":\"LOAD\"}}") + .when() + .post("/api/resilience/scenarios/broker-crash") + .then() + .statusCode(409) + .body( + "message", + is("Another disruption is running against " + target + ", so no fault was injected")); + } finally { + concurrencyGuard.release(target, "leader-cascade#1"); + } + verify(orchestrator, never()).execute(any()); + } + /** * node-maintenance is listed but not run: a scenario can't name the node * to drain, and Litmus node-drain, given none, drains the node of a