GITGUD

neiam neiam /mutant-operator

An Experimental Operator

public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages

⭐

Log in to mark this repository.

Mutation resources and an HTTP API to manage them

c8d9978 · gmorell · 2026-11-09 05:41

12 files +1245 -125
Message 
{commit_body(@commit)}

Files changed

modified Cargo.lock
+81 −0
@@ -305,6 +305,56 @@ version = "1.4.0"
305 305 source = "registry+https://github.com/rust-lang/crates.io-index"
306 306 checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26"
307 307
308 +[[package]]
309 +name = "axum"
310 +version = "0.8.9"
311 +source = "registry+https://github.com/rust-lang/crates.io-index"
312 +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
313 +dependencies = [
314 + "axum-core",
315 + "bytes",
316 + "form_urlencoded",
317 + "futures-util",
318 + "http",
319 + "http-body",
320 + "http-body-util",
321 + "hyper",
322 + "hyper-util",
323 + "itoa",
324 + "matchit",
325 + "memchr",
326 + "mime",
327 + "percent-encoding",
328 + "pin-project-lite",
329 + "serde_core",
330 + "serde_json",
331 + "serde_path_to_error",
332 + "serde_urlencoded",
333 + "sync_wrapper",
334 + "tokio",
335 + "tower",
336 + "tower-layer",
337 + "tower-service",
338 +]
339 +
340 +[[package]]
341 +name = "axum-core"
342 +version = "0.5.6"
343 +source = "registry+https://github.com/rust-lang/crates.io-index"
344 +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
345 +dependencies = [
346 + "bytes",
347 + "futures-core",
348 + "http",
349 + "http-body",
350 + "http-body-util",
351 + "mime",
352 + "pin-project-lite",
353 + "sync_wrapper",
354 + "tower-layer",
355 + "tower-service",
356 +]
357 +
308 358 [[package]]
309 359 name = "backon"
310 360 version = "1.5.0"
@@ -1221,6 +1271,7 @@ dependencies = [
1221 1271 "http",
1222 1272 "http-body",
1223 1273 "httparse",
1274 + "httpdate",
1224 1275 "itoa",
1225 1276 "pin-project-lite",
1226 1277 "smallvec",
@@ -1786,6 +1837,12 @@ dependencies = [
1786 1837 "windows-sys 0.61.2",
1787 1838 ]
1788 1839
1840 +[[package]]
1841 +name = "matchit"
1842 +version = "0.8.4"
1843 +source = "registry+https://github.com/rust-lang/crates.io-index"
1844 +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
1845 +
1789 1846 [[package]]
1790 1847 name = "md-5"
1791 1848 version = "0.11.0"
@@ -1834,6 +1891,7 @@ version = "0.1.0"
1834 1891 dependencies = [
1835 1892 "anyhow",
1836 1893 "async-trait",
1894 + "axum",
1837 1895 "derive",
1838 1896 "futures",
1839 1897 "k8s-openapi",
@@ -2804,6 +2862,29 @@ dependencies = [
2804 2862 "zmij",
2805 2863 ]
2806 2864
2865 +[[package]]
2866 +name = "serde_path_to_error"
2867 +version = "0.1.20"
2868 +source = "registry+https://github.com/rust-lang/crates.io-index"
2869 +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
2870 +dependencies = [
2871 + "itoa",
2872 + "serde",
2873 + "serde_core",
2874 +]
2875 +
2876 +[[package]]
2877 +name = "serde_urlencoded"
2878 +version = "0.7.1"
2879 +source = "registry+https://github.com/rust-lang/crates.io-index"
2880 +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
2881 +dependencies = [
2882 + "form_urlencoded",
2883 + "itoa",
2884 + "ryu",
2885 + "serde",
2886 +]
2887 +
2807 2888 [[package]]
2808 2889 name = "serde_yaml"
2809 2890 version = "0.9.34+deprecated"
modified Cargo.toml
+3 −2
@@ -9,7 +9,7 @@ k8s-openapi = { version = "0.24.0", features = ["latest"] }
9 9 kube = { version = "0.99.0", features = ["derive", "runtime"] }
10 10 schemars = "0.8.22"
11 11 serde = { version = "1.0.219", features = ["derive"] }
12 −tokio = { version = "1.44.2", features = ["rt", "rt-multi-thread", "macros"] }
12 +tokio = { version = "1.44.2", features = ["rt", "rt-multi-thread", "macros", "signal", "net"] }
13 13 async-trait = "0.1.88"
14 14 serde_json = "1.0.140"
15 15 anyhow = "1.0.98"
@@ -18,6 +18,7 @@ tracing = "0.1.41"
18 18 tracing-subscriber = "0.3.19"
19 19 sea-orm = { version = "2.0.4", features = ["sqlx-postgres", "runtime-tokio-native-tls"] }
20 20 futures = "0.3.31"
21 +axum = { version = "0.8", default-features = false, features = ["http1", "json", "tokio", "tower-log", "query"] }
21 22
22 23
23 24 [package.metadata.sqlx]
@@ -25,4 +26,4 @@ offline = true
25 26
26 27 [[bin]]
27 28 name = "mutant-deployment"
28 −path = "src/bin/main-deployment.rs"
29 +path = "src/bin/main-deployment.rs"
modified README.md
+101 −1
@@ -43,4 +43,104 @@ spec:
43 43 env:
44 44 NGINX_EXPERIMENTAL: "true"
45 45 instances: 3
46 −```
46 +```
47 +Removing a mutation from the list deletes its Deployment. Deleting the
48 +MutantDeployment deletes all of them.
49 +
50 +## Mutation resources
51 +
52 +Mutations can also be standalone `Mutation` resources, created at runtime
53 +without touching the MutantDeployment. Use them when the MutantDeployment is
54 +owned by Argo CD or another GitOps tool:
55 +
56 +```yaml
57 +apiVersion: neiam.org/v1
58 +kind: Mutation
59 +metadata:
60 + name: nginx-mutant-beta
61 +spec:
62 + mutantDeployment: nginx-mutant # same namespace
63 + label: beta
64 + env:
65 + FEATURE_X: "on"
66 + instances: 2
67 +```
68 +
69 +```console
70 +$ kubectl get mutations
71 +NAME TARGET LABEL INSTANCES READY STATE AGE
72 +nginx-mutant-beta nginx-mutant beta 2 2 Ready 1m
73 +```
74 +
75 +- A MutantDeployment runs its inline `spec.mutations` plus every Mutation
76 + that targets it. Labels are unique per MutantDeployment. An inline
77 + mutation wins a clash, then the oldest Mutation. A Mutation that loses is
78 + not run and shows `State: Conflict`.
79 +- `status` reports the generated Deployment, its ready replicas, and
80 + `Progressing`, `Ready`, `Invalid` or `Conflict`.
81 +- The operator makes each Mutation a dependent (`ownerReferences`) of its
82 + MutantDeployment. Argo CD shows it under the MutantDeployment in the
83 + resource tree, and deleting the MutantDeployment deletes its Mutations.
84 +
85 +### Argo CD
86 +
87 +Mutations created through the API are not in git and carry no Argo CD
88 +tracking label, so Argo CD neither prunes nor reverts them. Inline
89 +mutations remain in git and are read-only through the API. For Argo CD to
90 +show Mutation health, add to `argocd-cm`:
91 +
92 +```yaml
93 +resource.customizations.health.neiam.org_Mutation: |
94 + hs = { status = "Progressing", message = "Waiting for the operator" }
95 + if obj.status ~= nil and obj.status.state ~= nil then
96 + hs.message = obj.status.message or obj.status.state
97 + if obj.status.state == "Ready" then hs.status = "Healthy"
98 + elseif obj.status.state == "Progressing" then hs.status = "Progressing"
99 + else hs.status = "Degraded" end
100 + end
101 + return hs
102 +```
103 +
104 +## HTTP API
105 +
106 +The operator serves an HTTP API to list, create and destroy mutations.
107 +Creating one makes a Mutation resource named `<mutantdeployment>-<label>`.
108 +Destroying one deletes it.
109 +
110 +The API is off until `MUTANT_API_TOKEN` is set (`operator.yml` reads it from
111 +the `mutantops-api` secret). Every `/api` request then needs
112 +`Authorization: Bearer <token>`. `MUTANT_API_INSECURE=true` serves it without
113 +auth. `MUTANT_API_ADDR` sets the listen address (default `0.0.0.0:8080`).
114 +
115 +```sh
116 +kubectl -n mutants create secret generic mutantops-api --from-literal=token="$(openssl rand -hex 32)"
117 +kubectl -n mutants rollout restart deploy/mutantops-operator
118 +TOKEN=$(kubectl -n mutants get secret mutantops-api -o jsonpath='{.data.token}' | base64 -d)
119 +API=http://mutantops-api.mutants.svc # or: kubectl -n mutants port-forward svc/mutantops-api 8080:80
120 +MD=$API/api/v1/namespaces/default/mutantdeployments/nginx-mutant
121 +
122 +# MutantDeployments in a namespace, with their running mutation labels
123 +curl -H "Authorization: Bearer $TOKEN" $API/api/v1/namespaces/default/mutantdeployments
124 +# list / get mutations: inline and resource, with live status
125 +curl -H "Authorization: Bearer $TOKEN" $MD/mutations
126 +curl -H "Authorization: Bearer $TOKEN" $MD/mutations/canary
127 +# create (201; 409 if the label is taken, 400 if invalid, 404 if no such MutantDeployment)
128 +curl -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
129 + -d '{"label":"beta","env":{"FEATURE_X":"on"},"instances":2}' $MD/mutations
130 +# destroy (204; 404 if absent; 409 if inline, which is changed in git instead)
131 +curl -X DELETE -H "Authorization: Bearer $TOKEN" $MD/mutations/beta
132 +```
133 +
134 +A listed mutation looks like:
135 +
136 +```json
137 +{
138 + "label": "beta",
139 + "env": { "FEATURE_X": "on" },
140 + "instances": 2,
141 + "source": { "kind": "resource", "name": "nginx-mutant-beta" },
142 + "status": { "state": "Ready", "message": null, "deployment": "nginx-beta", "replicas": 2, "readyReplicas": 2 }
143 +}
144 +```
145 +
146 +`source.kind` is `inline` for mutations declared in the MutantDeployment.
modified crd.yml
+94 −0
@@ -41,3 +41,97 @@ spec:
41 41 kind: MutantDeployment
42 42 shortNames:
43 43 - md
44 +---
45 +# A single mutation of a MutantDeployment, created at runtime (e.g. through
46 +# the operator's HTTP API) so the MutantDeployment itself can stay owned by
47 +# GitOps. Schema matches `Mutation` in src/crd.rs.
48 +apiVersion: apiextensions.k8s.io/v1
49 +kind: CustomResourceDefinition
50 +metadata:
51 + name: mutations.neiam.org
52 +spec:
53 + group: neiam.org
54 + scope: Namespaced
55 + names:
56 + plural: mutations
57 + singular: mutation
58 + kind: Mutation
59 + shortNames:
60 + - mut
61 + versions:
62 + - name: v1
63 + served: true
64 + storage: true
65 + subresources:
66 + status: {}
67 + additionalPrinterColumns:
68 + - name: Target
69 + type: string
70 + jsonPath: .spec.mutantDeployment
71 + - name: Label
72 + type: string
73 + jsonPath: .spec.label
74 + - name: Instances
75 + type: integer
76 + jsonPath: .spec.instances
77 + - name: Ready
78 + type: integer
79 + jsonPath: .status.readyReplicas
80 + - name: State
81 + type: string
82 + jsonPath: .status.state
83 + - name: Age
84 + type: date
85 + jsonPath: .metadata.creationTimestamp
86 + schema:
87 + openAPIV3Schema:
88 + type: object
89 + required:
90 + - spec
91 + properties:
92 + spec:
93 + type: object
94 + required:
95 + - mutantDeployment
96 + - label
97 + - instances
98 + properties:
99 + mutantDeployment:
100 + description: Name of the MutantDeployment, in the same namespace, to mutate.
101 + type: string
102 + label:
103 + description: Suffix of the generated Deployment's name; unique per MutantDeployment.
104 + type: string
105 + pattern: '^[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?$'
106 + env:
107 + type: object
108 + default: {}
109 + additionalProperties:
110 + type: string
111 + instances:
112 + type: integer
113 + format: int32
114 + minimum: 0
115 + status:
116 + type: object
117 + required:
118 + - state
119 + properties:
120 + state:
121 + type: string
122 + enum: [Progressing, Ready, Invalid, Conflict]
123 + message:
124 + type: string
125 + nullable: true
126 + deployment:
127 + description: The generated Deployment.
128 + type: string
129 + nullable: true
130 + replicas:
131 + type: integer
132 + format: int32
133 + nullable: true
134 + readyReplicas:
135 + type: integer
136 + format: int32
137 + nullable: true
modified operator.yml
+27 −0
@@ -29,6 +29,31 @@ spec:
29 29 fieldPath: metadata.name
30 30 - name: OPERATOR_NAME
31 31 value: "mutantops-operator"
32 + # The HTTP API stays off until this is set. Create the secret with:
33 + # kubectl -n mutants create secret generic mutantops-api \
34 + # --from-literal=token="$(openssl rand -hex 32)"
35 + - name: MUTANT_API_TOKEN
36 + valueFrom:
37 + secretKeyRef:
38 + name: mutantops-api
39 + key: token
40 + optional: true
41 + ports:
42 + - name: http
43 + containerPort: 8080
44 +---
45 +apiVersion: v1
46 +kind: Service
47 +metadata:
48 + name: mutantops-api
49 + namespace: mutants
50 +spec:
51 + selector:
52 + app: mutantops-operator
53 + ports:
54 + - name: http
55 + port: 80
56 + targetPort: http
32 57
33 58 ---
34 59 apiVersion: v1
@@ -80,6 +105,8 @@ rules:
80 105 - neiam.org
81 106 resources:
82 107 - mutantdeployments
108 + - mutations
109 + - mutations/status
83 110 verbs:
84 111 - create
85 112 - delete
added src/api.rs
+307 −0
@@ -0,0 +1,307 @@
1 +//! HTTP API for listing, creating and deleting the mutations of a live
2 +//! MutantDeployment.
3 +//!
4 +//! Created mutations are standalone `Mutation` resources, never edits to the
5 +//! MutantDeployment, which can stay owned by Argo CD (or any GitOps tool)
6 +//! without the API and the sync fighting over it. Mutations declared inline
7 +//! in the MutantDeployment are listed too, but are read-only here: they
8 +//! change in git.
9 +//!
10 +//! ```text
11 +//! GET /healthz
12 +//! GET /api/v1/namespaces/{ns}/mutantdeployments
13 +//! GET /api/v1/namespaces/{ns}/mutantdeployments/{name}/mutations
14 +//! POST /api/v1/namespaces/{ns}/mutantdeployments/{name}/mutations
15 +//! GET /api/v1/namespaces/{ns}/mutantdeployments/{name}/mutations/{label}
16 +//! DELETE /api/v1/namespaces/{ns}/mutantdeployments/{name}/mutations/{label}
17 +//! ```
18 +
19 +use crate::crd::{
20 + applied_status, resolve, MutantDeployment, Mutation, MutationDef, MutationSpec, MutationStatus,
21 + Source, OWNER_LABEL, TARGET_LABEL,
22 +};
23 +use axum::{
24 + extract::{Path, Request, State},
25 + http::{header, StatusCode},
26 + middleware::{self, Next},
27 + response::{IntoResponse, Response},
28 + routing::get,
29 + Json, Router,
30 +};
31 +use k8s_openapi::api::apps::v1::Deployment;
32 +use kube::{
33 + api::{Api, DeleteParams, ListParams, PostParams},
34 + Client, Resource, ResourceExt,
35 +};
36 +use serde::Serialize;
37 +use serde_json::json;
38 +use std::{collections::BTreeMap, sync::Arc};
39 +use tracing::{info, warn};
40 +
41 +#[derive(Clone)]
42 +pub struct AppState {
43 + client: Client,
44 + /// Bearer token required on every `/api` request; `None` disables auth.
45 + token: Option<Arc<str>>,
46 +}
47 +
48 +impl AppState {
49 + pub fn new(client: Client, token: Option<String>) -> Self {
50 + Self { client, token: token.map(Into::into) }
51 + }
52 +}
53 +
54 +pub fn router(state: AppState) -> Router {
55 + let api = Router::new()
56 + .route("/namespaces/{ns}/mutantdeployments", get(list_mutants))
57 + .route(
58 + "/namespaces/{ns}/mutantdeployments/{name}/mutations",
59 + get(list_mutations).post(create_mutation),
60 + )
61 + .route(
62 + "/namespaces/{ns}/mutantdeployments/{name}/mutations/{label}",
63 + get(get_mutation).delete(delete_mutation),
64 + )
65 + .route_layer(middleware::from_fn_with_state(state.clone(), require_token));
66 + Router::new()
67 + .route("/healthz", get(|| async { "ok" }))
68 + .nest("/api/v1", api)
69 + .with_state(state)
70 +}
71 +
72 +async fn require_token(State(state): State<AppState>, req: Request, next: Next) -> Response {
73 + let presented = req
74 + .headers()
75 + .get(header::AUTHORIZATION)
76 + .and_then(|v| v.to_str().ok())
77 + .and_then(|v| v.strip_prefix("Bearer "));
78 + if authorized(state.token.as_deref(), presented) {
79 + next.run(req).await
80 + } else {
81 + ApiError::Unauthorized.into_response()
82 + }
83 +}
84 +
85 +fn authorized(expected: Option<&str>, presented: Option<&str>) -> bool {
86 + match (expected, presented) {
87 + (None, _) => true,
88 + (Some(want), Some(got)) => constant_time_eq(want.as_bytes(), got.as_bytes()),
89 + (Some(_), None) => false,
90 + }
91 +}
92 +
93 +fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
94 + a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
95 +}
96 +
97 +#[derive(Debug)]
98 +enum ApiError {
99 + Unauthorized,
100 + BadRequest(String),
101 + NotFound(String),
102 + Conflict(String),
103 + Kube(kube::Error),
104 +}
105 +
106 +impl From<kube::Error> for ApiError {
107 + fn from(err: kube::Error) -> Self {
108 + match err {
109 + kube::Error::Api(ref e) if e.code == 404 => ApiError::NotFound(e.message.clone()),
110 + kube::Error::Api(ref e) if e.code == 409 => ApiError::Conflict(e.message.clone()),
111 + kube::Error::Api(ref e) if e.code == 422 => ApiError::BadRequest(e.message.clone()),
112 + other => ApiError::Kube(other),
113 + }
114 + }
115 +}
116 +
117 +impl IntoResponse for ApiError {
118 + fn into_response(self) -> Response {
119 + let (status, msg) = match self {
120 + ApiError::Unauthorized => (StatusCode::UNAUTHORIZED, "missing or invalid bearer token".into()),
121 + ApiError::BadRequest(m) => (StatusCode::BAD_REQUEST, m),
122 + ApiError::NotFound(m) => (StatusCode::NOT_FOUND, m),
123 + ApiError::Conflict(m) => (StatusCode::CONFLICT, m),
124 + ApiError::Kube(e) => {
125 + warn!("kubernetes request failed: {e}");
126 + (StatusCode::BAD_GATEWAY, format!("kubernetes request failed: {e}"))
127 + }
128 + };
129 + (status, Json(json!({ "error": msg }))).into_response()
130 + }
131 +}
132 +
133 +type ApiResult<T> = Result<T, ApiError>;
134 +
135 +/// A mutation as the API reports it: its definition, where it was declared,
136 +/// and its live state.
137 +#[derive(Serialize, Debug)]
138 +struct MutationView {
139 + #[serde(flatten)]
140 + def: MutationDef,
141 + source: Source,
142 + status: MutationStatus,
143 +}
144 +
145 +#[derive(Serialize)]
146 +struct MutantSummary {
147 + name: String,
148 + deployment: String,
149 + mutations: Vec<String>,
150 +}
151 +
152 +/// Every mutation of `name`: the ones running, then Mutation resources that
153 +/// are not (invalid, or a duplicate label), with why.
154 +async fn views(client: &Client, ns: &str, name: &str) -> ApiResult<Vec<MutationView>> {
155 + let mutant = Api::<MutantDeployment>::namespaced(client.clone(), ns).get(name).await?;
156 + let resources = Api::<Mutation>::namespaced(client.clone(), ns)
157 + .list(&ListParams::default())
158 + .await?
159 + .items;
160 + // Status is read live from the Deployments rather than from Mutation
161 + // status, so inline mutations, which have no status, get one too.
162 + let deployments: BTreeMap<String, Deployment> = Api::<Deployment>::namespaced(client.clone(), ns)
163 + .list(&ListParams::default().labels(&format!("{OWNER_LABEL}={name}")))
164 + .await?
165 + .into_iter()
166 + .map(|d| (d.name_any(), d))
167 + .collect();
168 +
169 + let resolved = resolve(&mutant, &resources);
170 + let mut out: Vec<_> = resolved
171 + .active
172 + .into_iter()
173 + .map(|(def, source)| {
174 + let dname = mutant.deployment_name(&def.label);
175 + let status = applied_status(dname.clone(), def.instances, deployments.get(&dname));
176 + MutationView { def, source, status }
177 + })
178 + .collect();
179 + for (rname, state, message) in resolved.rejected {
180 + if let Some(m) = resources.iter().find(|m| m.name_any() == rname) {
181 + out.push(MutationView {
182 + def: m.spec.def.clone(),
183 + source: Source::Resource { name: rname },
184 + status: MutationStatus { state, message: Some(message), ..Default::default() },
185 + });
186 + }
187 + }
188 + Ok(out)
189 +}
190 +
191 +async fn list_mutants(State(st): State<AppState>, Path(ns): Path<String>) -> ApiResult<Json<Vec<MutantSummary>>> {
192 + let mutants = Api::<MutantDeployment>::namespaced(st.client.clone(), &ns)
193 + .list(&ListParams::default())
194 + .await?;
195 + let resources = Api::<Mutation>::namespaced(st.client, &ns)
196 + .list(&ListParams::default())
197 + .await?
198 + .items;
199 + Ok(Json(
200 + mutants
201 + .into_iter()
202 + .map(|md| MutantSummary {
203 + mutations: resolve(&md, &resources).active.into_iter().map(|(d, _)| d.label).collect(),
204 + deployment: md.base_name(),
205 + name: md.name_any(),
206 + })
207 + .collect(),
208 + ))
209 +}
210 +
211 +async fn list_mutations(
212 + State(st): State<AppState>,
213 + Path((ns, name)): Path<(String, String)>,
214 +) -> ApiResult<Json<Vec<MutationView>>> {
215 + Ok(Json(views(&st.client, &ns, &name).await?))
216 +}
217 +
218 +async fn get_mutation(
219 + State(st): State<AppState>,
220 + Path((ns, name, label)): Path<(String, String, String)>,
221 +) -> ApiResult<Json<MutationView>> {
222 + // The running one: listed first, ahead of any rejected duplicate.
223 + views(&st.client, &ns, &name)
224 + .await?
225 + .into_iter()
226 + .find(|v| v.def.label == label)
227 + .map(Json)
228 + .ok_or_else(|| ApiError::NotFound(format!("mutation {label:?} not found")))
229 +}
230 +
231 +/// Creates a Mutation resource named `{name}-{label}`.
232 +async fn create_mutation(
233 + State(st): State<AppState>,
234 + Path((ns, name)): Path<(String, String)>,
235 + Json(def): Json<MutationDef>,
236 +) -> ApiResult<(StatusCode, Json<Mutation>)> {
237 + def.validate().map_err(ApiError::BadRequest)?;
238 + let mutant = Api::<MutantDeployment>::namespaced(st.client.clone(), &ns).get(&name).await?;
239 + // Refused here rather than created only to sit in Conflict.
240 + if views(&st.client, &ns, &name).await?.iter().any(|v| v.def.label == def.label) {
241 + return Err(ApiError::Conflict(format!("mutation {:?} already exists", def.label)));
242 + }
243 +
244 + let label = def.label.clone();
245 + let mut mutation = Mutation::new(
246 + &format!("{name}-{label}"),
247 + MutationSpec { mutant_deployment: name.clone(), def },
248 + );
249 + mutation.metadata.labels = Some(BTreeMap::from([(TARGET_LABEL.to_string(), name.clone())]));
250 + // Set up front so the Mutation is a dependent of its MutantDeployment
251 + // from the start; the controller also adds it to hand-made ones.
252 + mutation.metadata.owner_references = mutant.owner_ref(&()).map(|o| vec![o]);
253 +
254 + let created = Api::<Mutation>::namespaced(st.client, &ns)
255 + .create(&PostParams::default(), &mutation)
256 + .await?;
257 + info!(%ns, %name, %label, "mutation created");
258 + Ok((StatusCode::CREATED, Json(created)))
259 +}
260 +
261 +/// Deletes every Mutation resource of `name` with this label: the running
262 +/// one and any duplicate waiting behind it, which would otherwise take its
263 +/// place. Inline mutations are refused.
264 +async fn delete_mutation(
265 + State(st): State<AppState>,
266 + Path((ns, name, label)): Path<(String, String, String)>,
267 +) -> ApiResult<StatusCode> {
268 + let matching: Vec<_> = views(&st.client, &ns, &name)
269 + .await?
270 + .into_iter()
271 + .filter(|v| v.def.label == label)
272 + .collect();
273 + if matching.is_empty() {
274 + return Err(ApiError::NotFound(format!("mutation {label:?} not found")));
275 + }
276 + if matching.iter().any(|v| v.source == Source::Inline) {
277 + return Err(ApiError::Conflict(format!(
278 + "mutation {label:?} is declared inline in MutantDeployment {name}; remove it from its spec.mutations (in git) instead"
279 + )));
280 + }
281 + let api = Api::<Mutation>::namespaced(st.client, &ns);
282 + for v in matching {
283 + if let Source::Resource { name: rname } = v.source {
284 + match api.delete(&rname, &DeleteParams::default()).await {
285 + Ok(_) | Err(kube::Error::Api(kube::core::ErrorResponse { code: 404, .. })) => {}
286 + Err(e) => return Err(e.into()),
287 + }
288 + }
289 + }
290 + info!(%ns, %name, %label, "mutation deleted");
291 + Ok(StatusCode::NO_CONTENT)
292 +}
293 +
294 +#[cfg(test)]
295 +mod tests {
296 + use super::*;
297 +
298 + #[test]
299 + fn auth() {
300 + assert!(authorized(None, None));
301 + assert!(authorized(None, Some("anything")));
302 + assert!(authorized(Some("s3cret"), Some("s3cret")));
303 + assert!(!authorized(Some("s3cret"), Some("s3cre")));
304 + assert!(!authorized(Some("s3cret"), Some("s3creT")));
305 + assert!(!authorized(Some("s3cret"), None));
306 + }
307 +}
modified src/bin/main-deployment.rs
+38 −124
@@ -1,135 +1,49 @@
1 −use futures::StreamExt;
2 −use k8s_openapi::api::apps::v1::Deployment;
3 −use kube::{
4 − api::{Api, PostParams},
5 − runtime::controller::{self, Action, Controller},
6 − Client, CustomResource, ResourceExt,
7 −};
8 −use schemars::JsonSchema;
9 −use serde::{Deserialize, Serialize};
10 −use std::{collections::HashMap, fmt, sync::Arc};
11 −use kube::runtime::watcher;
12 −use tokio::time::Duration;
13 −use tracing::info;
1 +use kube::Client;
2 +use mutantops::{api, controller};
3 +use std::net::SocketAddr;
4 +use tracing::{info, warn};
14 5
15 −// Define a custom error type
16 −#[derive(Debug)]
17 −pub enum MutantopsError {
18 − ParseError(String),
19 − IoError(std::io::Error),
20 − KubeError(kube::Error),
21 −}
22 −
23 −impl std::error::Error for MutantopsError {}
24 −
25 −// Implement the `fmt::Display` trait for `MyError`
26 −impl fmt::Display for MutantopsError {
27 − fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
28 − match self {
29 − MutantopsError::ParseError(msg) => write!(f, "Parse error: {}", msg),
30 − MutantopsError::IoError(err) => write!(f, "IO error: {}", err),
31 − MutantopsError::KubeError(err) => write!(f, "Kube error: {}", err),
32 − }
33 − }
34 −}
35 −
36 −// Implement the `From` trait for `std::io::Error`
37 −impl From<std::io::Error> for MutantopsError {
38 − fn from(err: std::io::Error) -> Self {
39 − MutantopsError::IoError(err)
40 − }
41 −}
42 −
43 −// Implement the `From` trait for `kube::Error`
44 −impl From<kube::Error> for MutantopsError {
45 − fn from(err: kube::Error) -> Self {
46 − MutantopsError::KubeError(err)
47 − }
48 −}
49 −
50 −#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
51 −#[kube(group = "neiam.org", version = "v1", kind = "MutantDeployment", namespaced)]
52 −pub struct MutantDeploymentSpec {
53 − #[schemars(skip)]
54 − pub deployment: Deployment,
55 − pub mutations: Vec<Mutation>,
56 −}
57 −
58 −#[derive(Deserialize, Serialize, Clone, Debug, JsonSchema)]
59 −pub struct Mutation {
60 − pub label: String,
61 − pub env: HashMap<String, String>,
62 − pub instances: i32,
63 −}
64 −
65 −// #[instrument(skip(_ctx))]
66 −async fn reconcile(mutant: Arc<MutantDeployment>, _ctx: Arc<()>) -> Result<Action, MutantopsError> {
67 − let client = Client::try_default().await?;
68 − let deployments: Api<Deployment> = Api::namespaced(client.clone(), &mutant.namespace().unwrap());
69 −
70 − let base_deploy = &mutant.spec.deployment;
71 −
72 − for mutation in &mutant.spec.mutations {
73 − let mut new_deploy = base_deploy.clone();
74 −
75 − // Modify name
76 − let name = format!("{}-{}", base_deploy.metadata.name.clone().unwrap_or("mutant".into()), mutation.label);
77 − new_deploy.metadata.name = Some(name.clone());
78 − new_deploy.metadata.labels.get_or_insert_with(Default::default).insert("mutation".to_string(), mutation.label.clone());
79 −
80 − // Modify replicas
81 − if let Some(spec) = &mut new_deploy.spec {
82 − spec.replicas = Some(mutation.instances);
83 −
84 − if let Some(template) = &mut spec.template.spec {
85 − for container in &mut template.containers {
86 − let envs = container.env.get_or_insert(vec![]);
87 − for (key, val) in &mutation.env {
88 − envs.push(k8s_openapi::api::core::v1::EnvVar {
89 − name: key.clone(),
90 − value: Some(val.clone()),
91 − value_from: None,
92 − });
93 − }
94 − }
95 − }
96 − }
97 −
98 − // Apply the new deployment
99 − match deployments.get(&name).await {
100 − Ok(_) => {
101 − deployments.replace(&name, &PostParams::default(), &new_deploy).await?;
102 − }
103 − Err(_) => {
104 − deployments.create(&PostParams::default(), &new_deploy).await?;
105 − }
106 − }
107 − }
108 −
109 − Ok(Action::await_change())
110 −}
111 −
112 −fn error_policy(_object: Arc<MutantDeployment>, error: &MutantopsError, _ctx: Arc<()>) -> controller::Action {
113 − println!("Error reconciling: {:?}", error);
114 − controller::Action::requeue(Duration::from_secs(10))
115 −}
6 +/// Address the HTTP API listens on.
7 +const DEFAULT_API_ADDR: &str = "0.0.0.0:8080";
116 8
117 9 #[tokio::main]
118 10 async fn main() -> anyhow::Result<()> {
119 11 tracing_subscriber::fmt::init();
120 12
121 13 let client = Client::try_default().await?;
122 − let crds = Api::<MutantDeployment>::all(client.clone());
123 14
124 − Controller::new(crds, watcher::Config::default())
125 − .run(reconcile, error_policy, Arc::new(()))
126 − .for_each(|res| async move {
127 − match res {
128 − Ok((mutant, _)) => info!("Reconciled {:?}", mutant.name),
129 − Err(err) => info!("Reconcile failed: {:?}", err),
130 − }
131 − })
132 − .await;
15 + // The API can create Deployments with arbitrary env in any namespace the
16 + // operator can write to, so it needs a token unless explicitly opted out.
17 + let token = std::env::var("MUTANT_API_TOKEN").ok().filter(|t| !t.is_empty());
18 + let insecure = std::env::var("MUTANT_API_INSECURE").is_ok_and(|v| v == "true");
19 + let api_enabled = token.is_some() || insecure;
20 + if token.is_none() && insecure {
21 + warn!("MUTANT_API_INSECURE=true: HTTP API is serving without authentication");
22 + }
23 +
24 + let addr: SocketAddr = std::env::var("MUTANT_API_ADDR")
25 + .unwrap_or_else(|_| DEFAULT_API_ADDR.into())
26 + .parse()?;
133 27
28 + let server = async {
29 + if !api_enabled {
30 + warn!("HTTP API disabled: set MUTANT_API_TOKEN (or MUTANT_API_INSECURE=true) to enable it");
31 + std::future::pending::<()>().await;
32 + }
33 + let listener = tokio::net::TcpListener::bind(addr).await?;
34 + info!(%addr, "HTTP API listening");
35 + axum::serve(listener, api::router(api::AppState::new(client.clone(), token)))
36 + .with_graceful_shutdown(async {
37 + let _ = tokio::signal::ctrl_c().await;
38 + })
39 + .await?;
40 + anyhow::Ok(())
41 + };
42 +
43 + // The controller stops on SIGINT/SIGTERM; when it does, so does the API.
44 + tokio::select! {
45 + () = controller::run(client.clone()) => {}
46 + res = server => res?,
47 + }
134 48 Ok(())
135 49 }
added src/controller.rs
+166 −0
@@ -0,0 +1,166 @@
1 +//! Reconciles each MutantDeployment, together with the Mutation resources
2 +//! that target it, into one Deployment per mutation; prunes Deployments
3 +//! whose mutation is gone; and reports each Mutation's state in its status.
4 +
5 +use crate::crd::{
6 + applied_status, build_deployment, resolve, MutantDeployment, Mutation, MutationStatus, Source,
7 + MUTATION_LABEL, OWNER_LABEL,
8 +};
9 +use futures::StreamExt;
10 +use k8s_openapi::api::apps::v1::Deployment;
11 +use kube::{
12 + api::{Api, DeleteParams, ListParams, Patch, PatchParams},
13 + runtime::{
14 + controller::{Action, Controller},
15 + reflector::ObjectRef,
16 + watcher,
17 + },
18 + Client, Resource, ResourceExt,
19 +};
20 +use serde_json::json;
21 +use std::{collections::HashSet, fmt, sync::Arc, time::Duration};
22 +use tracing::{info, warn};
23 +
24 +const FIELD_MANAGER: &str = "mutantops-operator";
25 +/// Periodic resync. Events drive reconciliation; this only catches what no
26 +/// event points at, such as a Mutation retargeted away from a
27 +/// MutantDeployment, whose old Deployment is then pruned.
28 +const RESYNC: Duration = Duration::from_secs(300);
29 +
30 +#[derive(Debug)]
31 +pub enum MutantopsError {
32 + Invalid(String),
33 + Kube(kube::Error),
34 +}
35 +
36 +impl std::error::Error for MutantopsError {}
37 +
38 +impl fmt::Display for MutantopsError {
39 + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
40 + match self {
41 + MutantopsError::Invalid(msg) => write!(f, "invalid resource: {msg}"),
42 + MutantopsError::Kube(err) => write!(f, "kube error: {err}"),
43 + }
44 + }
45 +}
46 +
47 +impl From<kube::Error> for MutantopsError {
48 + fn from(err: kube::Error) -> Self {
49 + MutantopsError::Kube(err)
50 + }
51 +}
52 +
53 +struct Ctx {
54 + client: Client,
55 +}
56 +
57 +async fn reconcile(mutant: Arc<MutantDeployment>, ctx: Arc<Ctx>) -> Result<Action, MutantopsError> {
58 + let ns = mutant
59 + .namespace()
60 + .ok_or_else(|| MutantopsError::Invalid("MutantDeployment has no namespace".into()))?;
61 + let controller_ref = mutant
62 + .controller_owner_ref(&())
63 + .ok_or_else(|| MutantopsError::Invalid("MutantDeployment has no uid".into()))?;
64 + let deployments: Api<Deployment> = Api::namespaced(ctx.client.clone(), &ns);
65 + let mutations: Api<Mutation> = Api::namespaced(ctx.client.clone(), &ns);
66 + let apply = PatchParams::apply(FIELD_MANAGER).force();
67 +
68 + // Inline entries can only be edited in the MutantDeployment itself, and
69 + // have no status to report into, so a bad one is just logged.
70 + for def in &mutant.spec.mutations {
71 + if let Err(msg) = def.validate() {
72 + warn!(mutant = %mutant.name_any(), %ns, "skipping inline mutation: {msg}");
73 + }
74 + }
75 +
76 + let resources = mutations.list(&ListParams::default()).await?.items;
77 + // Each Mutation becomes a dependent of its MutantDeployment: Argo CD then
78 + // shows it in the MutantDeployment's resource tree, and deleting the
79 + // MutantDeployment garbage-collects it. Done before the status writes
80 + // below, which would invalidate the resourceVersion this is guarded by.
81 + let owner = mutant.owner_ref(&()).expect("uid checked above");
82 + for m in resources.iter().filter(|m| m.spec.mutant_deployment == mutant.name_any()) {
83 + if !m.owner_references().iter().any(|o| o.uid == owner.uid) {
84 + let mut refs = m.owner_references().to_vec();
85 + refs.push(owner.clone());
86 + let patch = json!({ "metadata": { "ownerReferences": refs, "resourceVersion": m.resource_version() } });
87 + mutations.patch(&m.name_any(), &PatchParams::default(), &Patch::Merge(&patch)).await?;
88 + }
89 + }
90 +
91 + let resolved = resolve(&mutant, &resources);
92 + let find = |name: &str| resources.iter().find(|m| m.name_any() == name);
93 +
94 + let mut wanted = HashSet::new();
95 + for (def, source) in &resolved.active {
96 + let name = mutant.deployment_name(&def.label);
97 + let deploy = build_deployment(&mutant, def, controller_ref.clone());
98 + let applied = deployments.patch(&name, &apply, &Patch::Apply(&deploy)).await?;
99 + if let Source::Resource { name: mname } = source {
100 + if let Some(m) = find(mname) {
101 + set_status(&mutations, m, applied_status(name.clone(), def.instances, Some(&applied))).await?;
102 + }
103 + }
104 + wanted.insert(name);
105 + }
106 + for (mname, state, message) in &resolved.rejected {
107 + if let Some(m) = find(mname) {
108 + let status = MutationStatus { state: *state, message: Some(message.clone()), ..Default::default() };
109 + set_status(&mutations, m, status).await?;
110 + }
111 + }
112 +
113 + // Prune: Deployments this MutantDeployment owns whose mutation is gone.
114 + // Checking the owner uid as well as the label means a Deployment that
115 + // merely carries the label is never deleted.
116 + let selector = format!("{OWNER_LABEL}={}", mutant.name_any());
117 + for existing in deployments.list(&ListParams::default().labels(&selector)).await? {
118 + let name = existing.name_any();
119 + let owned = existing.owner_references().iter().any(|o| o.uid == controller_ref.uid);
120 + if owned && !wanted.contains(&name) {
121 + info!(%name, %ns, mutation = ?existing.labels().get(MUTATION_LABEL), "pruning removed mutation");
122 + deployments.delete(&name, &DeleteParams::background()).await?;
123 + }
124 + }
125 +
126 + Ok(Action::requeue(RESYNC))
127 +}
128 +
129 +/// Writes `status` unless it is already current. Skipping no-op writes
130 +/// matters: every write is a watch event that reconciles again.
131 +async fn set_status(api: &Api<Mutation>, m: &Mutation, status: MutationStatus) -> Result<(), kube::Error> {
132 + if m.status.as_ref() == Some(&status) {
133 + return Ok(());
134 + }
135 + let patch = json!({ "status": status });
136 + api.patch_status(&m.name_any(), &PatchParams::default(), &Patch::Merge(&patch)).await?;
137 + Ok(())
138 +}
139 +
140 +fn error_policy(mutant: Arc<MutantDeployment>, error: &MutantopsError, _ctx: Arc<Ctx>) -> Action {
141 + warn!(mutant = %mutant.name_any(), "reconcile failed: {error}");
142 + Action::requeue(Duration::from_secs(10))
143 +}
144 +
145 +/// Runs the controller until a shutdown signal.
146 +pub async fn run(client: Client) {
147 + let mutants = Api::<MutantDeployment>::all(client.clone());
148 + Controller::new(mutants, watcher::Config::default())
149 + // Watch only the Deployments we generate: a hand edit or delete is
150 + // put back, and replica readiness flows into Mutation status.
151 + .owns(Api::<Deployment>::all(client.clone()), watcher::Config::default().labels(OWNER_LABEL))
152 + // A Mutation created, changed or deleted reconciles its target.
153 + .watches(Api::<Mutation>::all(client.clone()), watcher::Config::default(), |m| {
154 + let ns = m.namespace()?;
155 + Some(ObjectRef::new(&m.spec.mutant_deployment).within(&ns))
156 + })
157 + .shutdown_on_signal()
158 + .run(reconcile, error_policy, Arc::new(Ctx { client }))
159 + .for_each(|res| async move {
160 + match res {
161 + Ok((obj, _)) => info!(name = %obj.name, ns = ?obj.namespace, "reconciled"),
162 + Err(err) => warn!("reconcile failed: {err}"),
163 + }
164 + })
165 + .await;
166 +}
added src/crd.rs
+427 −0
@@ -0,0 +1,427 @@
1 +//! The `MutantDeployment` and `Mutation` resources, and the pure logic that
2 +//! turns them into concrete `Deployment`s.
3 +//!
4 +//! A mutation can be declared two ways:
5 +//! - inline, in `MutantDeployment.spec.mutations` (lives in git with the
6 +//! MutantDeployment, so GitOps tools own it);
7 +//! - as a standalone `Mutation` resource naming its MutantDeployment in
8 +//! `spec.mutantDeployment` (created at runtime, e.g. through the HTTP API,
9 +//! without touching the GitOps-managed MutantDeployment).
10 +
11 +use k8s_openapi::api::apps::v1::Deployment;
12 +use k8s_openapi::api::core::v1::EnvVar;
13 +use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference};
14 +use kube::{CustomResource, ResourceExt};
15 +use schemars::JsonSchema;
16 +use serde::{Deserialize, Serialize};
17 +use std::collections::{BTreeMap, HashMap, HashSet};
18 +
19 +/// Set on every generated Deployment: the mutation it was built from.
20 +/// Kept unprefixed for compatibility with Deployments created before it
21 +/// was namespaced.
22 +pub const MUTATION_LABEL: &str = "mutation";
23 +/// Set on every generated Deployment: the MutantDeployment that owns it.
24 +/// Used to find Deployments to prune when a mutation is removed.
25 +pub const OWNER_LABEL: &str = "mutantops.neiam.org/owner";
26 +/// Set on Mutation resources the API creates: their MutantDeployment, so
27 +/// they can be selected with `kubectl get mutations -l ...`.
28 +pub const TARGET_LABEL: &str = "mutantops.neiam.org/mutant-deployment";
29 +
30 +#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
31 +#[kube(group = "neiam.org", version = "v1", kind = "MutantDeployment", namespaced)]
32 +pub struct MutantDeploymentSpec {
33 + #[schemars(skip)]
34 + pub deployment: Deployment,
35 + #[serde(default)]
36 + pub mutations: Vec<MutationDef>,
37 +}
38 +
39 +#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, PartialEq, JsonSchema)]
40 +#[kube(
41 + group = "neiam.org",
42 + version = "v1",
43 + kind = "Mutation",
44 + namespaced,
45 + status = "MutationStatus"
46 +)]
47 +#[serde(rename_all = "camelCase")]
48 +pub struct MutationSpec {
49 + /// Name of the MutantDeployment, in the same namespace, to mutate.
50 + pub mutant_deployment: String,
51 + #[serde(flatten)]
52 + pub def: MutationDef,
53 +}
54 +
55 +/// One mutation, however it was declared.
56 +#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, JsonSchema)]
57 +pub struct MutationDef {
58 + pub label: String,
59 + #[serde(default)]
60 + pub env: HashMap<String, String>,
61 + pub instances: i32,
62 +}
63 +
64 +#[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq, JsonSchema)]
65 +#[serde(rename_all = "camelCase")]
66 +pub struct MutationStatus {
67 + // `None` serializes as null on purpose: status is written as a merge
68 + // patch, where null clears a field left over from an earlier state.
69 + pub state: MutationState,
70 + #[serde(default)]
71 + pub message: Option<String>,
72 + /// The generated Deployment.
73 + #[serde(default)]
74 + pub deployment: Option<String>,
75 + #[serde(default)]
76 + pub replicas: Option<i32>,
77 + #[serde(default)]
78 + pub ready_replicas: Option<i32>,
79 +}
80 +
81 +#[derive(Deserialize, Serialize, Clone, Copy, Debug, Default, PartialEq, Eq, JsonSchema)]
82 +pub enum MutationState {
83 + /// Deployment applied; not all replicas ready yet.
84 + #[default]
85 + Progressing,
86 + /// Every requested replica is ready.
87 + Ready,
88 + /// Not applied: the spec is invalid.
89 + Invalid,
90 + /// Not applied: another mutation of the same MutantDeployment already
91 + /// uses this label.
92 + Conflict,
93 +}
94 +
95 +impl MutationDef {
96 + /// Checks the mutation can become a valid Deployment, so a bad entry is
97 + /// refused at the API instead of failing reconciliation later.
98 + pub fn validate(&self) -> Result<(), String> {
99 + validate_label(&self.label)?;
100 + if self.instances < 0 {
101 + return Err(format!("instances must be >= 0, got {}", self.instances));
102 + }
103 + for key in self.env.keys() {
104 + validate_env_name(key)?;
105 + }
106 + Ok(())
107 + }
108 +}
109 +
110 +/// The label becomes both a name suffix and a label value, so it must be an
111 +/// RFC 1123 label: lowercase alphanumerics and '-', 1-63 chars, alphanumeric
112 +/// at both ends.
113 +pub fn validate_label(label: &str) -> Result<(), String> {
114 + let ok = !label.is_empty()
115 + && label.len() <= 63
116 + && label
117 + .bytes()
118 + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
119 + && !label.starts_with('-')
120 + && !label.ends_with('-');
121 + if ok {
122 + Ok(())
123 + } else {
124 + Err(format!(
125 + "invalid label {label:?}: must be 1-63 lowercase alphanumerics or '-', \
126 + starting and ending alphanumeric"
127 + ))
128 + }
129 +}
130 +
131 +/// Kubernetes' rule for env var names: `[-._a-zA-Z][-._a-zA-Z0-9]*`.
132 +fn validate_env_name(name: &str) -> Result<(), String> {
133 + let mut bytes = name.bytes();
134 + let first_ok = matches!(bytes.next(), Some(b) if b.is_ascii_alphabetic() || b"-._".contains(&b));
135 + if first_ok && bytes.all(|b| b.is_ascii_alphanumeric() || b"-._".contains(&b)) {
136 + Ok(())
137 + } else {
138 + Err(format!("invalid env var name {name:?}"))
139 + }
140 +}
141 +
142 +impl MutantDeployment {
143 + /// Name of the base Deployment the mutations are derived from.
144 + pub fn base_name(&self) -> String {
145 + self.spec
146 + .deployment
147 + .metadata
148 + .name
149 + .clone()
150 + .unwrap_or_else(|| "mutant".into())
151 + }
152 +
153 + /// Name of the Deployment generated for `label`.
154 + pub fn deployment_name(&self, label: &str) -> String {
155 + format!("{}-{}", self.base_name(), label)
156 + }
157 +}
158 +
159 +/// Where an active mutation was declared.
160 +#[derive(Serialize, Clone, Debug, PartialEq, Eq)]
161 +#[serde(tag = "kind", rename_all = "camelCase")]
162 +pub enum Source {
163 + /// In the MutantDeployment's `spec.mutations`.
164 + Inline,
165 + /// A `Mutation` resource of this name.
166 + Resource { name: String },
167 +}
168 +
169 +/// The outcome of combining a MutantDeployment's inline mutations with the
170 +/// Mutation resources that target it.
171 +#[derive(Debug, Default)]
172 +pub struct Resolved {
173 + /// Mutations to run, one per label.
174 + pub active: Vec<(MutationDef, Source)>,
175 + /// Mutation resources not run, by name, with why.
176 + pub rejected: Vec<(String, MutationState, String)>,
177 +}
178 +
179 +/// Decides which mutations run. Inline mutations come first, since they
180 +/// live in git with the MutantDeployment; Mutation resources follow, oldest
181 +/// first, so an existing mutation is never displaced by a newer one with the
182 +/// same label. Invalid inline entries are dropped (the caller logs them).
183 +pub fn resolve(mutant: &MutantDeployment, resources: &[Mutation]) -> Resolved {
184 + let mut out = Resolved::default();
185 + let mut labels = HashSet::new();
186 +
187 + for def in &mutant.spec.mutations {
188 + if def.validate().is_ok() && labels.insert(def.label.clone()) {
189 + out.active.push((def.clone(), Source::Inline));
190 + }
191 + }
192 +
193 + let mut targeting: Vec<_> = resources
194 + .iter()
195 + .filter(|m| m.spec.mutant_deployment == mutant.name_any() && m.metadata.deletion_timestamp.is_none())
196 + .collect();
197 + targeting.sort_by(|a, b| {
198 + (a.creation_timestamp(), a.name_any()).cmp(&(b.creation_timestamp(), b.name_any()))
199 + });
200 + for m in targeting {
201 + let def = &m.spec.def;
202 + if let Err(msg) = def.validate() {
203 + out.rejected.push((m.name_any(), MutationState::Invalid, msg));
204 + } else if !labels.insert(def.label.clone()) {
205 + out.rejected.push((
206 + m.name_any(),
207 + MutationState::Conflict,
208 + format!("label {:?} is already used by another mutation of {}", def.label, mutant.name_any()),
209 + ));
210 + } else {
211 + out.active.push((def.clone(), Source::Resource { name: m.name_any() }));
212 + }
213 + }
214 + out
215 +}
216 +
217 +/// Status of an applied mutation, from its Deployment's state.
218 +pub fn applied_status(name: String, desired: i32, deploy: Option<&Deployment>) -> MutationStatus {
219 + let status = deploy.and_then(|d| d.status.as_ref());
220 + let ready = status.and_then(|s| s.ready_replicas).unwrap_or(0);
221 + let observed_current = deploy.is_some_and(|d| {
222 + d.status.as_ref().and_then(|s| s.observed_generation) >= d.metadata.generation
223 + });
224 + MutationStatus {
225 + state: if observed_current && ready >= desired {
226 + MutationState::Ready
227 + } else {
228 + MutationState::Progressing
229 + },
230 + message: None,
231 + deployment: Some(name),
232 + replicas: Some(desired),
233 + ready_replicas: Some(ready),
234 + }
235 +}
236 +
237 +/// Builds the Deployment for one mutation, ready for server-side apply.
238 +pub fn build_deployment(
239 + mutant: &MutantDeployment,
240 + mutation: &MutationDef,
241 + owner: OwnerReference,
242 +) -> Deployment {
243 + let base = &mutant.spec.deployment;
244 + let mut labels: BTreeMap<String, String> = base.metadata.labels.clone().unwrap_or_default();
245 + labels.insert(MUTATION_LABEL.into(), mutation.label.clone());
246 + labels.insert(OWNER_LABEL.into(), mutant.name_any());
247 +
248 + let mut deploy = Deployment {
249 + // Only the user-authored parts of the base metadata are carried over;
250 + // server-populated fields would make the apply fail.
251 + metadata: ObjectMeta {
252 + name: Some(mutant.deployment_name(&mutation.label)),
253 + namespace: mutant.namespace(),
254 + labels: Some(labels),
255 + annotations: base.metadata.annotations.clone(),
256 + owner_references: Some(vec![owner]),
257 + ..Default::default()
258 + },
259 + spec: base.spec.clone(),
260 + status: None,
261 + };
262 +
263 + if let Some(spec) = &mut deploy.spec {
264 + spec.replicas = Some(mutation.instances);
265 + // Pod labels (not the selector, which is immutable on existing
266 + // Deployments) so a Service or query can single out one mutation.
267 + spec.template
268 + .metadata
269 + .get_or_insert_with(Default::default)
270 + .labels
271 + .get_or_insert_with(Default::default)
272 + .insert(MUTATION_LABEL.into(), mutation.label.clone());
273 +
274 + if let Some(pod) = &mut spec.template.spec {
275 + let mut env: Vec<_> = mutation.env.iter().collect();
276 + env.sort(); // stable order, so re-applies don't churn the pod template
277 + for container in &mut pod.containers {
278 + let vars = container.env.get_or_insert_with(Vec::new);
279 + for (key, val) in &env {
280 + // Override a variable the base already sets rather than
281 + // adding a duplicate entry.
282 + vars.retain(|v| &v.name != *key);
283 + vars.push(EnvVar {
284 + name: (*key).clone(),
285 + value: Some((*val).clone()),
286 + value_from: None,
287 + });
288 + }
289 + }
290 + }
291 + }
292 + deploy
293 +}
294 +
295 +#[cfg(test)]
296 +mod tests {
297 + use super::*;
298 + use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
299 +
300 + fn mutant(inline: Vec<MutationDef>) -> MutantDeployment {
301 + let deployment: Deployment = serde_json::from_value(serde_json::json!({
302 + "apiVersion": "apps/v1",
303 + "kind": "Deployment",
304 + "metadata": { "name": "nginx", "labels": { "app": "nginx" }, "resourceVersion": "9" },
305 + "spec": {
306 + "replicas": 3,
307 + "selector": { "matchLabels": { "app": "nginx" } },
308 + "template": {
309 + "metadata": { "labels": { "app": "nginx" } },
310 + "spec": { "containers": [{
311 + "name": "nginx", "image": "nginx",
312 + "env": [{ "name": "MODE", "value": "base" }]
313 + }] }
314 + }
315 + }
316 + }))
317 + .unwrap();
318 + let mut md = MutantDeployment::new("nginx-mutant", MutantDeploymentSpec { deployment, mutations: inline });
319 + md.metadata.namespace = Some("default".into());
320 + md
321 + }
322 +
323 + fn def(label: &str) -> MutationDef {
324 + MutationDef {
325 + label: label.into(),
326 + env: HashMap::from([("MODE".into(), "canary".into()), ("EXTRA".into(), "1".into())]),
327 + instances: 1,
328 + }
329 + }
330 +
331 + fn resource(name: &str, target: &str, label: &str, created_secs: i64) -> Mutation {
332 + let mut m = Mutation::new(name, MutationSpec { mutant_deployment: target.into(), def: def(label) });
333 + m.metadata.creation_timestamp = Some(Time(k8s_openapi::chrono::DateTime::from_timestamp(created_secs, 0).unwrap()));
334 + m
335 + }
336 +
337 + #[test]
338 + fn builds_deployment() {
339 + let md = mutant(vec![]);
340 + let d = build_deployment(&md, &def("canary"), OwnerReference::default());
341 + assert_eq!(d.metadata.name.as_deref(), Some("nginx-canary"));
342 + assert_eq!(d.metadata.namespace.as_deref(), Some("default"));
343 + assert_eq!(d.metadata.resource_version, None);
344 + let labels = d.metadata.labels.unwrap();
345 + assert_eq!(labels[MUTATION_LABEL], "canary");
346 + assert_eq!(labels[OWNER_LABEL], "nginx-mutant");
347 + assert_eq!(labels["app"], "nginx");
348 +
349 + let spec = d.spec.unwrap();
350 + assert_eq!(spec.replicas, Some(1));
351 + assert_eq!(spec.template.metadata.unwrap().labels.unwrap()[MUTATION_LABEL], "canary");
352 + let env = spec.template.spec.unwrap().containers[0].env.clone().unwrap();
353 + let names: Vec<_> = env.iter().map(|e| (e.name.as_str(), e.value.as_deref().unwrap())).collect();
354 + assert_eq!(names, [("EXTRA", "1"), ("MODE", "canary")]);
355 + }
356 +
357 + #[test]
358 + fn validates() {
359 + assert!(def("canary").validate().is_ok());
360 + for bad in ["", "Canary", "-a", "a-", "a_b", &"a".repeat(64)] {
361 + assert!(validate_label(bad).is_err(), "{bad:?}");
362 + }
363 + let mut m = def("ok");
364 + m.instances = -1;
365 + assert!(m.validate().is_err());
366 + let mut m = def("ok");
367 + m.env.insert("1BAD".into(), "x".into());
368 + assert!(m.validate().is_err());
369 + }
370 +
371 + #[test]
372 + fn resolves_inline_first_then_oldest_resource() {
373 + let md = mutant(vec![def("canary")]);
374 + let mut bad = resource("bad", "nginx-mutant", "Nope", 1);
375 + bad.spec.def.label = "Nope".into();
376 + let resources = [
377 + resource("newer-beta", "nginx-mutant", "beta", 20),
378 + resource("older-beta", "nginx-mutant", "beta", 10),
379 + resource("dup-canary", "nginx-mutant", "canary", 5),
380 + resource("elsewhere", "other-mutant", "gamma", 1),
381 + bad,
382 + ];
383 + let r = resolve(&md, &resources);
384 +
385 + let active: Vec<_> = r.active.iter().map(|(d, s)| (d.label.as_str(), s.clone())).collect();
386 + assert_eq!(
387 + active,
388 + [
389 + ("canary", Source::Inline),
390 + ("beta", Source::Resource { name: "older-beta".into() }),
391 + ]
392 + );
393 + let rejected: Vec<_> = r.rejected.iter().map(|(n, s, _)| (n.as_str(), *s)).collect();
394 + assert_eq!(
395 + rejected,
396 + [
397 + ("bad", MutationState::Invalid),
398 + ("dup-canary", MutationState::Conflict),
399 + ("newer-beta", MutationState::Conflict),
400 + ]
401 + );
402 + }
403 +
404 + #[test]
405 + fn status_from_deployment() {
406 + let mut d: Deployment = serde_json::from_value(serde_json::json!({
407 + "metadata": { "generation": 2 },
408 + "status": { "observedGeneration": 2, "readyReplicas": 2 }
409 + }))
410 + .unwrap();
411 + assert_eq!(applied_status("x".into(), 2, Some(&d)).state, MutationState::Ready);
412 + assert_eq!(applied_status("x".into(), 3, Some(&d)).state, MutationState::Progressing);
413 + d.metadata.generation = Some(3); // spec changed, not yet observed
414 + assert_eq!(applied_status("x".into(), 2, Some(&d)).state, MutationState::Progressing);
415 + assert_eq!(applied_status("x".into(), 0, None).state, MutationState::Progressing);
416 + }
417 +
418 + #[test]
419 + fn mutation_wire_format() {
420 + let m: MutationSpec = serde_json::from_value(serde_json::json!({
421 + "mutantDeployment": "nginx-mutant", "label": "beta", "env": { "A": "1" }, "instances": 2
422 + }))
423 + .unwrap();
424 + assert_eq!(m.mutant_deployment, "nginx-mutant");
425 + assert_eq!(m.def.label, "beta");
426 + }
427 +}
added src/lib.rs
+3 −0
@@ -0,0 +1,3 @@
1 +pub mod api;
2 +pub mod controller;
3 +pub mod crd;

Parents: e4ce82a