From 29811d8d3388cf1aadcb57689bbc923da4d8efe9 Mon Sep 17 00:00:00 2001 From: Said Boudjelda Date: Thu, 5 Nov 2026 20:31:11 +0100 Subject: [PATCH 1/6] feat(operator): the reconcilers, kates-client, crds apply and leader election kates-client classifies every answer of the Kates API; KatesTest and KafkaCompliance reconcile on it, with their decisions in plain functions over the status. The binary takes the Lease, holds off on a stale CRD, and runs one controller per watched namespace. CRDs move to operator/chart/crds: the operator stays self-contained and nothing under charts/ learns about it. --- .github/workflows/ci-operator.yml | 14 +- operator/Cargo.lock | 878 +++++++++++++++++- operator/Cargo.toml | 2 + operator/Dockerfile | 18 + .../chart/crds/kafkacompliances.kates.io.yaml | 192 ++++ .../{ => chart}/crds/katestests.kates.io.yaml | 0 operator/crates/kates-client/Cargo.toml | 20 + operator/crates/kates-client/src/lib.rs | 709 ++++++++++++++ .../crates/kates-crds/src/kafkacompliance.rs | 100 ++ operator/crates/kates-crds/src/lib.rs | 44 +- .../kates-crds/tests/kafkacompliance.rs | 42 + operator/crates/kates-operator/Cargo.toml | 41 + .../crates/kates-operator/src/connection.rs | 181 ++++ operator/crates/kates-operator/src/context.rs | 80 ++ operator/crates/kates-operator/src/crds.rs | 122 +++ operator/crates/kates-operator/src/lib.rs | 10 + operator/crates/kates-operator/src/main.rs | 16 + operator/crates/kates-operator/src/metrics.rs | 281 ++++++ .../src/reconcile/compliance.rs | 618 ++++++++++++ .../kates-operator/src/reconcile/mod.rs | 56 ++ .../kates-operator/src/reconcile/test.rs | 732 +++++++++++++++ operator/crates/kates-operator/src/run.rs | 577 ++++++++++++ operator/crates/kates-operator/src/status.rs | 115 +++ .../crates/kates-operator/src/telemetry.rs | 81 ++ 24 files changed, 4915 insertions(+), 14 deletions(-) create mode 100644 operator/Dockerfile create mode 100644 operator/chart/crds/kafkacompliances.kates.io.yaml rename operator/{ => chart}/crds/katestests.kates.io.yaml (100%) create mode 100644 operator/crates/kates-client/Cargo.toml create mode 100644 operator/crates/kates-client/src/lib.rs create mode 100644 operator/crates/kates-crds/src/kafkacompliance.rs create mode 100644 operator/crates/kates-crds/tests/kafkacompliance.rs create mode 100644 operator/crates/kates-operator/Cargo.toml create mode 100644 operator/crates/kates-operator/src/connection.rs create mode 100644 operator/crates/kates-operator/src/context.rs create mode 100644 operator/crates/kates-operator/src/crds.rs create mode 100644 operator/crates/kates-operator/src/lib.rs create mode 100644 operator/crates/kates-operator/src/main.rs create mode 100644 operator/crates/kates-operator/src/metrics.rs create mode 100644 operator/crates/kates-operator/src/reconcile/compliance.rs create mode 100644 operator/crates/kates-operator/src/reconcile/mod.rs create mode 100644 operator/crates/kates-operator/src/reconcile/test.rs create mode 100644 operator/crates/kates-operator/src/run.rs create mode 100644 operator/crates/kates-operator/src/status.rs create mode 100644 operator/crates/kates-operator/src/telemetry.rs diff --git a/.github/workflows/ci-operator.yml b/.github/workflows/ci-operator.yml index 980c7c1d..74eea6f8 100644 --- a/.github/workflows/ci-operator.yml +++ b/.github/workflows/ci-operator.yml @@ -84,11 +84,11 @@ jobs: - name: Generated CRDs are fresh run: | - cargo run --locked --bin crdgen -- --out crds - if [ -n "$(git status --porcelain crds)" ]; then - git status --porcelain crds - git diff crds - echo "::error::operator/crds is stale: run cargo run --bin crdgen -- --out crds" + cargo run --locked --bin crdgen -- --out chart/crds + if [ -n "$(git status --porcelain chart/crds)" ]; then + git status --porcelain chart/crds + git diff chart/crds + echo "::error::operator/chart/crds is stale: run cargo run --bin crdgen -- --out chart/crds" exit 1 fi @@ -108,7 +108,7 @@ jobs: set -euo pipefail schemas="${RUNNER_TEMP}/schemas" mkdir -p "${schemas}" - yq -o=json '.spec.versions[0].schema.openAPIV3Schema' operator/crds/katestests.kates.io.yaml \ + yq -o=json '.spec.versions[0].schema.openAPIV3Schema' operator/chart/crds/katestests.kates.io.yaml \ | jq '.properties.spec |= walk(if type == "object" and (.properties | type) == "object" then . + {additionalProperties: false} else . end)' \ > "${schemas}/katestest_v1alpha1.json" @@ -146,7 +146,7 @@ jobs: # The apiserver is the only CEL compiler, cost and oldSelf-correlatability # checker there is. - name: The apiserver accepts the CRDs - run: kubectl apply --dry-run=server -f operator/crds + run: kubectl apply --dry-run=server -f operator/chart/crds gate: name: Gate · Operator diff --git a/operator/Cargo.lock b/operator/Cargo.lock index e767c582..f49477f6 100644 --- a/operator/Cargo.lock +++ b/operator/Cargo.lock @@ -73,6 +73,12 @@ dependencies = [ "uuid", ] +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + [[package]] name = "ar_archive_writer" version = "0.5.3" @@ -88,6 +94,16 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236" +[[package]] +name = "assert-json-diff" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e4f2b81832e72834d7518d8487a0396a28cc408186a2e8854c0f98011faf12" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "async-broadcast" version = "0.7.2" @@ -122,6 +138,17 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -157,6 +184,52 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "backon" version = "1.6.0" @@ -265,6 +338,23 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.45" @@ -318,6 +408,15 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48" +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + [[package]] name = "darling" version = "0.24.1" @@ -352,6 +451,24 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "deadpool" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" +dependencies = [ + "deadpool-runtime", + "lazy_static", + "num_cpus", + "tokio", +] + +[[package]] +name = "deadpool-runtime" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b" + [[package]] name = "defmt" version = "1.2.0" @@ -415,6 +532,12 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "dtoa" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c3cf4824e2d5f025c7b531afcb2325364084a16806f6d47fbc1f5fbd9960590" + [[package]] name = "dunce" version = "1.0.5" @@ -536,6 +659,12 @@ version = "0.1.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.2.0" @@ -565,6 +694,7 @@ checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" dependencies = [ "futures-channel", "futures-core", + "futures-executor", "futures-io", "futures-sink", "futures-task", @@ -587,6 +717,17 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.34" @@ -640,8 +781,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", ] [[package]] @@ -663,8 +806,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", + "js-sys", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", + "wasm-bindgen", ] [[package]] @@ -689,6 +835,25 @@ dependencies = [ "smallvec", ] +[[package]] +name = "h2" +version = "0.4.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d29020232d6aa3fb1daca64c1127cf662cf97f254ae16c18c05b8ab635fc118" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hashbrown" version = "0.16.1" @@ -700,6 +865,18 @@ dependencies = [ "foldhash", ] +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "hermit-abi" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17592d60ebacc7d5e169f4663c5f84f9161cc90328abcfe8456f41e4dfcb284" + [[package]] name = "hostname" version = "0.4.2" @@ -750,6 +927,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hyper" version = "1.12.0" @@ -758,10 +941,13 @@ checksum = "2c3e324da4c95177d6291d4c8730197c0d1822f8a9766814a4a44fa5ab797c9c" dependencies = [ "atomic-waker", "bytes", + "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -805,6 +991,7 @@ version = "0.1.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" dependencies = [ + "base64 0.23.1", "bytes", "futures-channel", "futures-util", @@ -812,7 +999,9 @@ dependencies = [ "http-body", "httparse", "hyper", + "ipnet", "libc", + "percent-encoding", "pin-project-lite", "socket2", "tokio", @@ -954,12 +1143,31 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "indexmap" +version = "2.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" +dependencies = [ + "equivalent", + "hashbrown 0.17.1", +] + [[package]] name = "ipnet" version = "2.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -975,10 +1183,12 @@ dependencies = [ "defmt", "jiff-core", "jiff-static", + "jiff-tzdb-platform", "log", "portable-atomic", "portable-atomic-util", "serde_core", + "windows-link", ] [[package]] @@ -1003,6 +1213,21 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "jiff-tzdb" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa8377070c6bae868759445e5a77f66d84f0b72f3a054bfb00e6d038b8282da7" + +[[package]] +name = "jiff-tzdb-platform" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8" +dependencies = [ + "jiff-tzdb", +] + [[package]] name = "jni" version = "0.22.4" @@ -1121,6 +1346,19 @@ dependencies = [ "serde_json", ] +[[package]] +name = "kates-client" +version = "0.1.0" +dependencies = [ + "reqwest", + "secrecy", + "serde", + "serde_json", + "thiserror 2.0.21", + "tokio", + "wiremock", +] + [[package]] name = "kates-crds" version = "0.1.0" @@ -1133,6 +1371,36 @@ dependencies = [ "serde_json", ] +[[package]] +name = "kates-operator" +version = "0.1.0" +dependencies = [ + "axum", + "futures", + "http", + "jiff", + "json-patch", + "k8s-openapi", + "kates-client", + "kates-crds", + "kube", + "kube-lease-manager", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", + "prometheus-client", + "rustls", + "secrecy", + "serde", + "serde_json", + "thiserror 2.0.21", + "tokio", + "tower-test", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", +] + [[package]] name = "kube" version = "5.0.0" @@ -1196,7 +1464,7 @@ dependencies = [ "tokio", "tokio-util", "tower", - "tower-http", + "tower-http 0.7.1", "tracing", ] @@ -1234,6 +1502,20 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "kube-lease-manager" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d0763a568229d7e0e6adb359bbc898e809898ccda81946a085c7238fc0e6c99" +dependencies = [ + "k8s-openapi", + "kube", + "rand 0.10.3", + "thiserror 2.0.21", + "tokio", + "tracing", +] + [[package]] name = "kube-runtime" version = "5.0.0" @@ -1246,7 +1528,7 @@ dependencies = [ "backon", "educe", "futures", - "hashbrown", + "hashbrown 0.16.1", "hostname", "json-patch", "k8s-openapi", @@ -1294,6 +1576,27 @@ version = "0.4.35" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d15f86d12da44cd8d4ec3c73984537eb7f4cadcb9776a575a0a8a05ca11005c" +[[package]] +name = "lru-slab" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4050469837a6ff301cd14c1f8f24f88549e6d548f24f64e2148eb0f72cebc51f" + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.8.3" @@ -1354,6 +1657,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -1363,6 +1675,16 @@ dependencies = [ "autocfg", ] +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi", + "libc", +] + [[package]] name = "object" version = "0.39.1" @@ -1384,6 +1706,65 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e036b411754132fd6b74e94b74e9a13c5cf606fa93d1885c35860d3060820b1" +[[package]] +name = "opentelemetry" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c1f1a6ac9c875a4941e4c7c085b61e8f4daa326de03b4701768c3491a6ad9ee" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.21", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0756af24841e367128ef31c831999937b8b5c5197d47914590d8f4e2e88c2e1a" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "thiserror 2.0.21", + "tokio", + "tonic", + "tonic-types", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d576f5dddc3299341bcd85a04be17b2564844c0c9563214f02d85f2524d1afae" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.33.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "372a912db2974d1990207006fd0e3aa0ef3fd4ad3be43c65cfaadf2206866061" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.5", + "thiserror 2.0.21", +] + [[package]] name = "ordered-float" version = "2.10.1" @@ -1554,14 +1935,78 @@ dependencies = [ ] [[package]] -name = "proc-macro2" -version = "1.0.107" +name = "ppv-lite86" +version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" dependencies = [ "unicode-ident", ] +[[package]] +name = "prometheus-client" +version = "0.25.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d150ad7bf80e60f45769d18a0a8c01028ab8f56355f33ed7bfce3a31f82e696" +dependencies = [ + "dtoa", + "itoa", + "parking_lot", + "prometheus-client-derive-encode", +] + +[[package]] +name = "prometheus-client-derive-encode" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01e34894696ff94f64a20c2c373a6440903e9c2789a303d68ec6e6f953f890e4" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "prost-types" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a" +dependencies = [ + "prost", +] + [[package]] name = "psm" version = "0.1.32" @@ -1572,6 +2017,63 @@ dependencies = [ "cc", ] +[[package]] +name = "quinn" +version = "0.11.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.21", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e750cca55fe4f0439a15d0bb529da9651e79993e8e72c61a899a36d462befbe" +dependencies = [ + "aws-lc-rs", + "bytes", + "getrandom 0.4.3", + "lru-slab", + "rand 0.10.3", + "rand_pcg", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.21", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af66907df18639dcf4db56ca65490cabc4b27a97dbadd96f2926cca73298f016" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.61.2", +] + [[package]] name = "quote" version = "1.0.47" @@ -1593,6 +2095,61 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1651,6 +2208,46 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reqwest" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029" +dependencies = [ + "base64 0.23.1", + "bytes", + "encoding_rs", + "futures-core", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "mime", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "rustls-platform-verifier", + "serde", + "serde_json", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http 0.6.11", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "ring" version = "0.17.14" @@ -1665,6 +2262,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustc_version" version = "0.4.1" @@ -1707,6 +2310,7 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ + "web-time", "zeroize", ] @@ -1923,6 +2527,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -2035,6 +2648,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -2087,6 +2703,15 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.5" @@ -2097,20 +2722,39 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinyvec" +version = "1.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd3ca314f692efd6c868f8408f53fe444634a845f96c028b97d35f6a1f79f0ee" + [[package]] name = "tokio" version = "1.54.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b52fd3255258d26fac7264ab5a76b53c13367136dc011d3d6ad4f8ed070396a3" dependencies = [ + "bytes", "libc", "mio", "pin-project-lite", "signal-hook-registry", "socket2", + "tokio-macros", "windows-sys 0.61.2", ] +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "tokio-rustls" version = "0.26.6" @@ -2121,6 +2765,28 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tokio-test" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12bc5db7778a8e08ee255090861e0e7aac55e784300077e7ed293c7be36168b2" +dependencies = [ + "futures-core", + "tokio", + "tokio-stream", +] + [[package]] name = "tokio-util" version = "0.7.20" @@ -2136,6 +2802,57 @@ dependencies = [ "tokio", ] +[[package]] +name = "tonic" +version = "0.14.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2cdfc673aabd30fe8fee92e65837b644c37c5d47f0f97ae6d5e0f1237d15e1b" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tokio-stream", + "tokio-util", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-prost" +version = "0.14.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eba464054e2226f182a34489b8b079bf4c6e3e7ab5fd7fc30e024f24aa92a484" +dependencies = [ + "bytes", + "prost", + "tonic", +] + +[[package]] +name = "tonic-types" +version = "0.14.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccbb6c31deb8e99dcdf87df19c5289b7838a600da74a8224baa671e52fe7c5ad" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -2144,7 +2861,9 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", + "indexmap", "pin-project-lite", + "slab", "sync_wrapper", "tokio", "tokio-util", @@ -2153,6 +2872,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags 2.13.2", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-http" version = "0.7.1" @@ -2185,6 +2922,20 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" +[[package]] +name = "tower-test" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4546773ffeab9e4ea02b8872faa49bb616a80a7da66afc2f32688943f97efa7" +dependencies = [ + "futures-util", + "pin-project", + "tokio", + "tokio-test", + "tower-layer", + "tower-service", +] + [[package]] name = "tracing" version = "0.1.44" @@ -2215,6 +2966,63 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-opentelemetry" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0a904802a1b902f43638b677ff2a650847e3b4404101b6c586d648e8c1e3e8fe" +dependencies = [ + "js-sys", + "opentelemetry", + "tracing", + "tracing-core", + "tracing-subscriber", + "web-time", +] + +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "serde", + "serde_json", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", + "tracing-serde", ] [[package]] @@ -2290,6 +3098,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "version_check" version = "0.9.5" @@ -2342,6 +3156,17 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.80" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e14e0327cb5368ad60ccb22d04a9827a693c95a926d8d125e9fa39e0baec501" +dependencies = [ + "js-sys", + "tokio", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.130" @@ -2374,6 +3199,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "web-sys" +version = "0.3.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7d179596ff5ec0f6de4db64d54011bde6289cc49603d3a6e4ce3f329207ab88" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-root-certs" version = "1.0.11" @@ -2533,6 +3378,29 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "wiremock" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031" +dependencies = [ + "assert-json-diff", + "base64 0.22.1", + "deadpool", + "futures", + "http", + "http-body-util", + "hyper", + "hyper-util", + "log", + "once_cell", + "regex", + "serde", + "serde_json", + "tokio", + "url", +] + [[package]] name = "wit-bindgen" version = "0.57.1" diff --git a/operator/Cargo.toml b/operator/Cargo.toml index 5f7b8e6b..b0bfaa97 100644 --- a/operator/Cargo.toml +++ b/operator/Cargo.toml @@ -15,6 +15,8 @@ schemars = "1" serde = { version = "1", features = ["derive"] } serde-saphyr = "2" serde_json = "1" +thiserror = "2" +tracing = "0.1" [workspace.lints.clippy] pedantic = { level = "warn", priority = -1 } diff --git a/operator/Dockerfile b/operator/Dockerfile new file mode 100644 index 00000000..b479835d --- /dev/null +++ b/operator/Dockerfile @@ -0,0 +1,18 @@ +# Built from the repository root (docker build -f operator/Dockerfile .): +# kates-crds reads schemas/ with include_str!. +FROM rust:1.99.1-bookworm@sha256:7085845219c18e7c5b1f34ed9ce89b84f822f2e3d54a02774cedc996ac0f304f AS builder +RUN apt-get update && apt-get install -y --no-install-recommends musl-tools && rm -rf /var/lib/apt/lists/* + +WORKDIR /build +COPY schemas/ schemas/ +COPY operator/ operator/ + +WORKDIR /build/operator +RUN target="$(uname -m)-unknown-linux-musl" \ + && rustup target add "$target" \ + && cargo build --release --locked --target "$target" -p kates-operator --features telemetry \ + && cp "target/$target/release/kates-operator" /kates-operator + +FROM gcr.io/distroless/static-debian12:nonroot@sha256:afa5c872c891853ca7fcf1f12c3edb23f7eeef36189728842dd51042ff57f7ab +COPY --from=builder /kates-operator /kates-operator +ENTRYPOINT ["/kates-operator"] diff --git a/operator/chart/crds/kafkacompliances.kates.io.yaml b/operator/chart/crds/kafkacompliances.kates.io.yaml new file mode 100644 index 00000000..73d945d0 --- /dev/null +++ b/operator/chart/crds/kafkacompliances.kates.io.yaml @@ -0,0 +1,192 @@ +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: kafkacompliances.kates.io +spec: + group: kates.io + names: + kind: KafkaCompliance + plural: kafkacompliances + singular: kafkacompliance + scope: Namespaced + versions: + - additionalPrinterColumns: + - jsonPath: .status.verdict + name: Verdict + type: string + - jsonPath: .status.checkedAt + name: Checked + type: date + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha1 + schema: + openAPIV3Schema: + description: "A standing compliance check of one Kafka. crd-version: 1" + properties: + spec: + description: A standing compliance check of one Kafka. + properties: + interval: + description: How often to check, such as 30m or 1h. + pattern: ^([0-9]+(s|m|h))+$ + type: string + reference: + description: A `KafkaReference` `.spec`, inline; its target names the Kafka. + properties: + target: + properties: + name: + type: string + namespace: + type: string + required: + - name + - namespace + type: object + x-kubernetes-preserve-unknown-fields: true + required: + - target + type: object + x-kubernetes-preserve-unknown-fields: true + suspend: + default: false + description: Stops new checks without touching what the last one found. + type: boolean + required: + - interval + - reference + type: object + status: + nullable: true + properties: + checkId: + description: "`/api/conformity/checks/{id}`; absent when the API kept no report." + nullable: true + type: string + checkedAt: + description: >- + Time is a wrapper around time.Time which supports correct marshaling to YAML + and JSON. Wrappers are provided for many of the factory methods that the time + package offers. + format: date-time + nullable: true + type: string + conditions: + items: + description: >- + Condition contains details for one aspect of the current state of this API + Resource. + properties: + lastTransitionTime: + description: >- + lastTransitionTime is the last time the condition transitioned from one status + to another. This should be when the underlying condition changed. If that is not + known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: >- + message is a human readable message indicating details about the transition. + This may be an empty string. + type: string + observedGeneration: + description: >- + observedGeneration represents the .metadata.generation that the condition was + set based upon. For instance, if .metadata.generation is currently 12, but the + .status.conditions[x].observedGeneration is 9, the condition is out of date with + respect to the current state of the instance. + format: int64 + type: integer + reason: + description: >- + reason contains a programmatic identifier indicating the reason for the + condition's last transition. Producers of specific condition types may define expected + values and meanings for this field, and whether the values are considered a + guaranteed API. The value should be a CamelCase string. This field may not be empty. + type: string + status: + description: status of the condition, one of True, False, Unknown. + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + type: array + x-kubernetes-list-map-keys: + - type + x-kubernetes-list-type: map + counts: + nullable: true + properties: + fail: + format: int64 + type: integer + pass: + format: int64 + type: integer + unknown: + format: int64 + type: integer + required: + - fail + - pass + - unknown + type: object + failed: + description: At most 20; the report has the rest. + items: + properties: + detail: + nullable: true + type: string + rule: + type: string + required: + - rule + type: object + type: array + lastHandledReconcileAt: + description: The `kates.io/reconcile-at` value last acted on. + nullable: true + type: string + nextCheckAt: + description: >- + Time is a wrapper around time.Time which supports correct marshaling to YAML + and JSON. Wrappers are provided for many of the factory methods that the time + package offers. + format: date-time + nullable: true + type: string + observedGeneration: + format: int64 + nullable: true + type: integer + observedKafkaGeneration: + description: The Kafka's `status.observedGeneration` at the last answered check. + format: int64 + nullable: true + type: integer + verdict: + description: PASS, FAIL or UNKNOWN. + nullable: true + type: string + type: object + required: + - spec + title: KafkaCompliance + type: object + x-kubernetes-validations: + - message: "name must be a DNS label (at most 63 lower-case letters, digits and -): it names the stored compliance check" + rule: self.metadata.name.matches('^[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?$') + served: true + storage: true + subresources: + status: {} diff --git a/operator/crds/katestests.kates.io.yaml b/operator/chart/crds/katestests.kates.io.yaml similarity index 100% rename from operator/crds/katestests.kates.io.yaml rename to operator/chart/crds/katestests.kates.io.yaml diff --git a/operator/crates/kates-client/Cargo.toml b/operator/crates/kates-client/Cargo.toml new file mode 100644 index 00000000..57c553e1 --- /dev/null +++ b/operator/crates/kates-client/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "kates-client" +version.workspace = true +edition.workspace = true +license.workspace = true +publish.workspace = true + +[dependencies] +reqwest = { version = "0.13", default-features = false, features = ["rustls", "json", "http2", "charset"] } +secrecy = "0.10" +serde.workspace = true +serde_json.workspace = true +thiserror.workspace = true + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +wiremock = "0.6" + +[lints] +workspace = true diff --git a/operator/crates/kates-client/src/lib.rs b/operator/crates/kates-client/src/lib.rs new file mode 100644 index 00000000..beaac498 --- /dev/null +++ b/operator/crates/kates-client/src/lib.rs @@ -0,0 +1,709 @@ +//! The Kates REST API, as the operator uses it. Every answer is classified into [`Error`], so the +//! reconcilers decide on variants and never on status codes. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use reqwest::header::{AUTHORIZATION, HeaderMap, LOCATION, RETRY_AFTER}; +use reqwest::{Certificate, Method, Response, StatusCode, redirect}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use serde_json::Value; + +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +// The CLI's figure, above the 35 s snapshot bound of a first report, so the API's own 503s arrive. +const REQUEST_TIMEOUT: Duration = Duration::from_mins(1); + +/// Told of every call: method, route template (never the URL), status (0 for a network error). +pub type Observer = Arc; + +#[derive(Debug, Clone, PartialEq, thiserror::Error)] +pub enum Error { + /// A 4xx with an `ApiError` body: the request is wrong as written. + #[error("{message}")] + Rejected { + status: u16, + message: String, + field_errors: BTreeMap, + }, + #[error("{message}")] + Conflict { message: String }, + /// 429: the engine is full. + #[error("{message}")] + RateLimited { + retry_after: Option, + message: String, + }, + #[error("the API refuses this key")] + InvalidKey, + #[error("this key lacks the scope for the call")] + MissingScope, + /// The run does not exist: only the API's own 404 says so, an ingress's never does. + #[error("the run does not exist")] + NotFound, + /// Everything else: a network error, a proxy's answer, a 5xx. `api_message` is an `ApiError` + /// body's message. + #[error("{detail}")] + Unavailable { + status: Option, + api_message: Option, + detail: String, + }, +} + +impl Error { + /// The variant's name, the bounded value of the metric's `error` label. + #[must_use] + pub fn name(&self) -> &'static str { + match self { + Self::Rejected { .. } => "Rejected", + Self::Conflict { .. } => "Conflict", + Self::RateLimited { .. } => "RateLimited", + Self::InvalidKey => "InvalidKey", + Self::MissingScope => "MissingScope", + Self::NotFound => "NotFound", + Self::Unavailable { .. } => "Unavailable", + } + } +} + +/// A `url` or `caBundle` no client can be built from. +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +pub struct InvalidConnection(String); + +#[derive(Debug, Deserialize)] +struct ApiError { + status: Option, + error: Option, + message: Option, + #[serde(default, rename = "fieldErrors")] + field_errors: BTreeMap, +} + +/// `TestResult.TaskStatus`; a status this build does not know reads as still running. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RunStatus { + Pending, + Running, + Done, + Failed, +} + +impl RunStatus { + fn parse(status: &str) -> Self { + match status { + "PENDING" => Self::Pending, + "DONE" => Self::Done, + "FAILED" => Self::Failed, + _ => Self::Running, + } + } + + #[must_use] + pub fn is_terminal(self) -> bool { + matches!(self, Self::Done | Self::Failed) + } +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct WireRun { + id: String, + status: String, + client_key: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Run { + pub id: String, + pub status: RunStatus, + pub client_key: Option, +} + +impl From for Run { + fn from(run: WireRun) -> Self { + Self { + status: RunStatus::parse(&run.status), + id: run.id, + client_key: run.client_key, + } + } +} + +#[derive(Debug, Clone, Deserialize)] +pub struct Violation { + pub metric: String, + pub threshold: f64, + pub actual: f64, + pub reason: Option, +} + +impl std::fmt::Display for Violation { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match &self.reason { + Some(reason) => write!(f, "{}: {reason}", self.metric), + None => write!( + f, + "{}: {} against {}", + self.metric, self.actual, self.threshold + ), + } + } +} + +#[derive(Debug, Clone, Deserialize)] +pub struct SlaVerdict { + pub passed: bool, + #[serde(default)] + pub violations: Vec, +} + +/// What the operator reads of `GET /api/tests/{id}/report`. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Report { + #[serde(default)] + pub summary: Value, + pub overall_sla_verdict: Option, +} + +#[derive(Debug, Clone, Default, Deserialize)] +pub struct ComplianceSummary { + pub pass: i64, + pub fail: i64, + pub warn: i64, + pub waived: i64, + pub unknown: i64, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct ComplianceResult { + pub rule: String, + pub status: String, + pub detail: Option, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct ComplianceReport { + pub verdict: String, + #[serde(default)] + pub summary: ComplianceSummary, + #[serde(default)] + pub results: Vec, +} + +/// An answered check. `check_id` is absent when the API ran it but could not store the report. +#[derive(Debug, Clone)] +pub struct ComplianceAnswer { + pub report: ComplianceReport, + pub check_id: Option, +} + +/// One Kates API. The key travels per call, so a rotated key needs no new client. +#[derive(Clone)] +pub struct Client { + http: reqwest::Client, + base: String, + observer: Option, +} + +impl Client { + /// # Errors + /// When `url` does not parse, `ca_bundle` holds no certificate, or no client can be built. + pub fn new(url: &str, ca_bundle: Option<&[u8]>) -> Result { + let parsed = reqwest::Url::parse(url) + .map_err(|error| InvalidConnection(format!("url {url:?}: {error}")))?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(InvalidConnection(format!("url {url:?}: not http or https"))); + } + let mut builder = reqwest::Client::builder() + .connect_timeout(CONNECT_TIMEOUT) + .timeout(REQUEST_TIMEOUT) + .redirect(redirect::Policy::none()); + if let Some(pem) = ca_bundle { + let certificates = Certificate::from_pem_bundle(pem) + .map_err(|error| InvalidConnection(format!("caBundle: {error}")))?; + if certificates.is_empty() { + return Err(InvalidConnection("caBundle: no certificate".to_owned())); + } + builder = builder.tls_certs_merge(certificates); + } + let http = builder + .build() + .map_err(|error| InvalidConnection(error.to_string()))?; + Ok(Self { + http, + base: url.trim_end_matches('/').to_owned(), + observer: None, + }) + } + + #[must_use] + pub fn observed_by(mut self, observer: Observer) -> Self { + self.observer = Some(observer); + self + } + + /// `GET /api/health`'s `capabilities`; an API older than the field lists none. + /// + /// # Errors + /// See [`Error`]. + pub async fn capabilities(&self, key: &SecretString) -> Result, Error> { + #[derive(Deserialize)] + struct Health { + #[serde(default)] + capabilities: Vec, + } + let health: Health = self + .json(Method::GET, "/api/health", "/api/health", key, None, None) + .await?; + Ok(health.capabilities) + } + + /// Starts a run, or finds the one `idempotency_key` already started. + /// + /// # Errors + /// See [`Error`]. + pub async fn create_test( + &self, + key: &SecretString, + idempotency_key: &str, + request: &Value, + ) -> Result { + let run: WireRun = self + .json( + Method::POST, + "/api/tests", + "/api/tests", + key, + Some(idempotency_key), + Some(request), + ) + .await?; + Ok(run.into()) + } + + /// # Errors + /// See [`Error`]. + pub async fn get_test(&self, key: &SecretString, id: &str) -> Result { + let path = format!("/api/tests/{id}"); + let run: WireRun = self + .json(Method::GET, "/api/tests/{id}", &path, key, None, None) + .await?; + Ok(run.into()) + } + + /// The run `client_key` created. An API older than the filter lists everything, so an item + /// counts only if it carries the key itself. + /// + /// # Errors + /// See [`Error`]. + pub async fn find_by_client_key( + &self, + key: &SecretString, + client_key: &str, + ) -> Result, Error> { + #[derive(Deserialize)] + struct Page { + #[serde(default)] + items: Vec, + } + let path = format!("/api/tests?clientKey={client_key}"); + let page: Page = self + .json(Method::GET, "/api/tests", &path, key, None, None) + .await?; + Ok(page + .items + .into_iter() + .find(|run| run.client_key.as_deref() == Some(client_key)) + .map(Run::from)) + } + + /// # Errors + /// `Conflict` when the run is not running. See [`Error`]. + pub async fn cancel_test(&self, key: &SecretString, id: &str) -> Result<(), Error> { + let path = format!("/api/tests/{id}/cancel"); + let _: Value = self + .json( + Method::POST, + "/api/tests/{id}/cancel", + &path, + key, + None, + None, + ) + .await?; + Ok(()) + } + + /// # Errors + /// See [`Error`]. + pub async fn report(&self, key: &SecretString, id: &str) -> Result { + let path = format!("/api/tests/{id}/report"); + self.json( + Method::GET, + "/api/tests/{id}/report", + &path, + key, + None, + None, + ) + .await + } + + /// Runs a compliance check. A stored report's id comes from the `Location` header. + /// + /// # Errors + /// See [`Error`]. + pub async fn check_compliance( + &self, + key: &SecretString, + reference: &Value, + ) -> Result { + const ROUTE: &str = "/api/conformity/check"; + let response = self + .call(Method::POST, ROUTE, ROUTE, key, None, Some(reference)) + .await?; + let check_id = response + .headers() + .get(LOCATION) + .and_then(|location| location.to_str().ok()) + .and_then(|location| location.rsplit('/').next()) + .filter(|id| !id.is_empty()) + .map(str::to_owned); + let report = decode(response).await?; + Ok(ComplianceAnswer { report, check_id }) + } + + async fn json( + &self, + method: Method, + route: &'static str, + path: &str, + key: &SecretString, + idempotency_key: Option<&str>, + body: Option<&Value>, + ) -> Result { + decode( + self.call(method, route, path, key, idempotency_key, body) + .await?, + ) + .await + } + + async fn call( + &self, + method: Method, + route: &'static str, + path: &str, + key: &SecretString, + idempotency_key: Option<&str>, + body: Option<&Value>, + ) -> Result { + let name = method_name(&method); + let mut request = self + .http + .request(method, format!("{}{path}", self.base)) + .header(AUTHORIZATION, format!("Bearer {}", key.expose_secret())); + if let Some(idempotency_key) = idempotency_key { + request = request.header("Idempotency-Key", idempotency_key); + } + if let Some(body) = body { + request = request.json(body); + } + let response = request.send().await; + let status = response.as_ref().map_or(0, |r| r.status().as_u16()); + if let Some(observer) = &self.observer { + observer(name, route, status); + } + let response = response.map_err(|error| Error::Unavailable { + status: None, + api_message: None, + // reqwest's text names the URL and never the headers, so no key can reach it. + detail: error.without_url().to_string(), + })?; + if response.status().is_success() { + return Ok(response); + } + let status = response.status(); + let headers = response.headers().clone(); + let host = response.url().host_str().unwrap_or_default().to_owned(); + let body = response.text().await.unwrap_or_default(); + Err(classify(status, &headers, &body, &host)) + } +} + +async fn decode(response: Response) -> Result { + let status = response.status().as_u16(); + response.json().await.map_err(|error| Error::Unavailable { + status: Some(status), + api_message: None, + detail: format!("an answer that is not the API's: {}", error.without_url()), + }) +} + +fn method_name(method: &Method) -> &'static str { + if *method == Method::GET { + "GET" + } else { + "POST" + } +} + +// Section 4.8 of plans/operator.md; the first match wins. An ingress's HTML 404 on a live run is +// an outage, never a RunGone, so only an ApiError body is believed. +fn classify(status: StatusCode, headers: &HeaderMap, body: &str, host: &str) -> Error { + let api_error = serde_json::from_str::(body) + .ok() + .filter(|error| error.status.is_some() || error.error.is_some()); + let message = || { + api_error + .as_ref() + .and_then(|error| error.message.clone()) + .unwrap_or_default() + }; + if status == StatusCode::TOO_MANY_REQUESTS { + return Error::RateLimited { + retry_after: headers + .get(RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse().ok()) + .map(Duration::from_secs), + message: message(), + }; + } + let unavailable = |api_message| Error::Unavailable { + status: Some(status.as_u16()), + api_message, + detail: format!("{host} answered {status}"), + }; + let Some(error) = &api_error else { + return unavailable(None); + }; + match status.as_u16() { + 409 => Error::Conflict { message: message() }, + 401 => Error::InvalidKey, + 403 if error.error.as_deref() == Some("Invalid API key") => Error::InvalidKey, + 403 => Error::MissingScope, + 404 if error.status == Some(404) && message().starts_with("Test run not found") => { + Error::NotFound + } + 400..=499 => Error::Rejected { + status: status.as_u16(), + message: message(), + field_errors: error.field_errors.clone(), + }, + _ => unavailable(error.message.clone()), + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + use wiremock::matchers::{header, method, path, query_param}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + fn key() -> SecretString { + SecretString::from("the-key") + } + + fn client(server: &MockServer) -> Client { + Client::new(&server.uri(), None).unwrap() + } + + fn api_error(status: u16, error: &str, message: &str) -> ResponseTemplate { + ResponseTemplate::new(status) + .set_body_json(json!({"status": status, "error": error, "message": message})) + } + + async fn answering(response: ResponseTemplate) -> Error { + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(response) + .mount(&server) + .await; + client(&server).get_test(&key(), "r").await.unwrap_err() + } + + #[tokio::test] + async fn create_sends_the_key_and_the_idempotency_key() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/prefix/api/tests")) + .and(header("authorization", "Bearer the-key")) + .and(header("idempotency-key", "uid-1")) + .respond_with(ResponseTemplate::new(202).set_body_json( + json!({"id": "r1", "status": "PENDING", "clientKey": "uid-1", "results": []}), + )) + .expect(1) + .mount(&server) + .await; + + let url = format!("{}/prefix/", server.uri()); + let run = Client::new(&url, None) + .unwrap() + .create_test(&key(), "uid-1", &json!({"type": "LOAD"})) + .await + .unwrap(); + assert_eq!(run.id, "r1"); + assert_eq!(run.status, RunStatus::Pending); + assert_eq!(run.client_key.as_deref(), Some("uid-1")); + } + + #[tokio::test] + async fn an_older_api_that_lists_everything_finds_nothing() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/tests")) + .and(query_param("clientKey", "uid-1")) + .respond_with(ResponseTemplate::new(200).set_body_json( + json!({"items": [{"id": "other", "status": "RUNNING"}], "total": 1}), + )) + .mount(&server) + .await; + let found = client(&server).find_by_client_key(&key(), "uid-1").await; + assert_eq!(found, Ok(None)); + } + + #[tokio::test] + async fn the_engine_full_answer_carries_retry_after() { + let error = answering( + api_error(429, "Too Many Requests", "engine full").insert_header("Retry-After", "30"), + ) + .await; + assert_eq!( + error, + Error::RateLimited { + retry_after: Some(Duration::from_secs(30)), + message: "engine full".to_owned() + } + ); + } + + #[tokio::test] + async fn only_the_apis_own_404_means_the_run_is_gone() { + let gone = answering(api_error(404, "Not Found", "Test run not found: r")).await; + assert_eq!(gone, Error::NotFound); + + let other = answering(api_error(404, "Not Found", "HTTP 404 Not Found")).await; + assert_eq!(other.name(), "Rejected"); + + let ingress = answering(ResponseTemplate::new(404).set_body_string("")).await; + assert_eq!(ingress.name(), "Unavailable"); + } + + #[tokio::test] + async fn refused_keys_are_told_apart() { + let invalid = answering(api_error(403, "Invalid API key", "m")).await; + assert_eq!(invalid, Error::InvalidKey); + let unauthorized = answering(api_error(401, "Unauthorized", "m")).await; + assert_eq!(unauthorized, Error::InvalidKey); + let scope = answering(api_error(403, "Forbidden", "m")).await; + assert_eq!(scope, Error::MissingScope); + } + + #[tokio::test] + async fn field_errors_and_conflicts_keep_the_apis_message() { + let rejected = answering( + ResponseTemplate::new(400).set_body_json(json!({"status": 400, "error": + "Validation Failed", "message": "bad", "fieldErrors": {"spec.acks": "no"}})), + ) + .await; + let Error::Rejected { field_errors, .. } = rejected else { + panic!("{rejected:?}"); + }; + assert_eq!(field_errors["spec.acks"], "no"); + + let conflict = answering(api_error(409, "Conflict", "Test is not running")).await; + assert_eq!( + conflict, + Error::Conflict { + message: "Test is not running".to_owned() + } + ); + } + + #[tokio::test] + async fn a_503_keeps_the_apis_message_and_a_proxy_has_none() { + let api = answering(api_error(503, "Service Unavailable", "kubernetes down")).await; + let Error::Unavailable { api_message, .. } = api else { + panic!("{api:?}"); + }; + assert_eq!(api_message.as_deref(), Some("kubernetes down")); + + let proxy = answering(ResponseTemplate::new(503).set_body_string("upstream")).await; + let Error::Unavailable { api_message, .. } = proxy else { + panic!("{proxy:?}"); + }; + assert_eq!(api_message, None); + } + + #[tokio::test] + async fn a_redirect_is_an_outage_and_not_followed() { + let error = answering(ResponseTemplate::new(302).insert_header("Location", "/x")).await; + assert_eq!(error.name(), "Unavailable"); + } + + #[tokio::test] + async fn the_check_id_comes_from_location_and_a_lost_report_has_none() { + let body = json!({"verdict": "FAIL", "summary": {"pass": 1, "fail": 1, "warn": 0, + "waived": 0, "unknown": 0}, "results": [{"rule": "kafka.replicas.min", + "status": "FAIL", "detail": "2 < 3"}]}); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/api/conformity/check")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(&body) + .insert_header("Location", "/api/conformity/checks/c1"), + ) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/api/conformity/check")) + .respond_with(ResponseTemplate::new(200).set_body_json(&body)) + .mount(&server) + .await; + + let client = client(&server); + let stored = client.check_compliance(&key(), &json!({})).await.unwrap(); + assert_eq!(stored.check_id.as_deref(), Some("c1")); + assert_eq!(stored.report.summary.fail, 1); + let lost = client.check_compliance(&key(), &json!({})).await.unwrap(); + assert_eq!(lost.check_id, None); + } + + #[tokio::test] + async fn the_observer_gets_the_route_template_and_never_the_url() { + let seen = Arc::new(std::sync::Mutex::new(Vec::new())); + let sink = seen.clone(); + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!({"id": "secret-id", "status": "DONE"})), + ) + .mount(&server) + .await; + client(&server) + .observed_by(Arc::new(move |m, r, s| { + sink.lock().unwrap().push((m, r, s)); + })) + .get_test(&key(), "secret-id") + .await + .unwrap(); + assert_eq!(*seen.lock().unwrap(), [("GET", "/api/tests/{id}", 200)]); + } + + #[test] + fn a_bad_url_or_bundle_is_an_invalid_connection() { + assert!(Client::new("not a url", None).is_err()); + assert!(Client::new("ftp://kates", None).is_err()); + assert!(Client::new("https://kates", Some(b"no certificate")).is_err()); + } +} diff --git a/operator/crates/kates-crds/src/kafkacompliance.rs b/operator/crates/kates-crds/src/kafkacompliance.rs new file mode 100644 index 00000000..07693c84 --- /dev/null +++ b/operator/crates/kates-crds/src/kafkacompliance.rs @@ -0,0 +1,100 @@ +use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time}; +use kube::{CustomResource, KubeSchema}; +use schemars::{JsonSchema, Schema, SchemaGenerator, json_schema}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// A standing compliance check of one Kafka. +#[derive(CustomResource, KubeSchema, Serialize, Deserialize, Clone, Debug)] +#[kube( + group = "kates.io", + version = "v1alpha1", + kind = "KafkaCompliance", + namespaced, + status = "KafkaComplianceStatus", + doc = "A standing compliance check of one Kafka. crd-version: 1", + cel, + validation = Rule::new("self.metadata.name.matches('^[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?$')") + .message("name must be a DNS label (at most 63 lower-case letters, digits and -): it names the stored compliance check"), + printcolumn = r#"{"name":"Verdict","type":"string","jsonPath":".status.verdict"}"#, + printcolumn = r#"{"name":"Checked","type":"date","jsonPath":".status.checkedAt"}"#, + printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"# +)] +#[serde(rename_all = "camelCase")] +pub struct KafkaComplianceSpec { + /// A `KafkaReference` `.spec`, inline; its target names the Kafka. + #[schemars(schema_with = "reference")] + pub reference: Value, + /// How often to check, such as 30m or 1h. + #[schemars(regex(pattern = r"^([0-9]+(s|m|h))+$"))] + pub interval: String, + /// Stops new checks without touching what the last one found. + #[serde(default)] + pub suspend: bool, +} + +// Pruning resumes at every declared property, so target keeps the flag too: without it the +// apiserver would drop clusterId, platform or a mistyped key before the strict reader sees them. +fn reference(_: &mut SchemaGenerator) -> Schema { + json_schema!({ + "type": "object", + "x-kubernetes-preserve-unknown-fields": true, + "required": ["target"], + "properties": { + "target": { + "type": "object", + "x-kubernetes-preserve-unknown-fields": true, + "required": ["namespace", "name"], + "properties": { + "namespace": {"type": "string"}, + "name": {"type": "string"}, + }, + }, + }, + }) +} + +#[derive(Serialize, Deserialize, JsonSchema, Clone, Debug, Default, PartialEq, Eq)] +pub struct ComplianceCounts { + pub pass: i64, + pub fail: i64, + pub unknown: i64, +} + +#[derive(Serialize, Deserialize, JsonSchema, Clone, Debug, PartialEq, Eq)] +pub struct FailedRule { + pub rule: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +#[derive(KubeSchema, Serialize, Deserialize, Clone, Debug, Default)] +#[serde(rename_all = "camelCase")] +pub struct KafkaComplianceStatus { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub observed_generation: Option, + /// PASS, FAIL or UNKNOWN. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub verdict: Option, + /// `/api/conformity/checks/{id}`; absent when the API kept no report. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub check_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub checked_at: Option