diff --git a/.github/workflows/ci-operator.yml b/.github/workflows/ci-operator.yml index 980c7c1d..f607984a 100644 --- a/.github/workflows/ci-operator.yml +++ b/.github/workflows/ci-operator.yml @@ -6,7 +6,7 @@ name: CI · Operator on: push: - branches: [main] + branches: [main, operator] pull_request: branches: [main] @@ -44,6 +44,7 @@ jobs: - 'charts/kates/**' - 'charts/kafka-cluster/**' - 'config/cluster.yaml' + - 'scripts/deploy-kafka-generic.sh' - 'versions.env' - '.github/workflows/ci-operator.yml' @@ -84,11 +85,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 +109,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" @@ -123,15 +124,29 @@ jobs: fi e2e: - name: CRDs on Kind + name: E2E on Kind needs: changes if: needs.changes.outputs.e2e == 'true' runs-on: ubuntu-latest - timeout-minutes: 15 + timeout-minutes: 45 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - uses: ./.github/actions/load-versions + - uses: azure/setup-helm@9bc31f4ebc9c6b171d7bfbaa5d006ae7abdb4310 # v5.0.1 + with: + version: ${{ env.HELM_VERSION }} + + # deploy-kafka-generic.sh detects the cluster with it. + - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: cli/go.mod + cache-dependency-path: cli/go.sum + + - name: Build kates CLI + working-directory: cli + run: go build -o ../build/kates . + # node_image is deliberately not passed, as in integration.yml: # config/cluster.yaml names the image. - name: Create Kind cluster @@ -146,7 +161,25 @@ 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 + + - name: Build the operator image + run: | + docker build -f operator/Dockerfile -t kates-operator:ci . + kind load docker-image kates-operator:ci --name ci-test + + - name: Deploy Kafka, Kates and the operator + run: operator/e2e-stack.sh docker.io/library/kates-operator:ci + + - name: Run the e2e checks + run: operator/e2e.sh payments kafka krafter + + - name: Show the stack + if: failure() + run: | + kubectl get pods -A + kubectl -n kates logs deploy/kates -c kates --tail=50 || true + kubectl -n kates-operator logs deploy/kates-operator --tail=50 || true gate: name: Gate · Operator diff --git a/.github/workflows/publish-operator.yml b/.github/workflows/publish-operator.yml new file mode 100644 index 00000000..7507bbfe --- /dev/null +++ b/.github/workflows/publish-operator.yml @@ -0,0 +1,133 @@ +name: Publish · Operator + +# The operator's image and chart, on an operator-v tag. It lives apart +# from release.yml on purpose: the operator ships on its own versions, and a +# v* Kates release never sees it. + +on: + push: + tags: + - "operator-v*" + +permissions: + contents: read + +concurrency: + group: publish-operator-${{ github.ref }} + cancel-in-progress: false + +env: + IMAGE: ghcr.io/bmscomp/kates-operator + CHART_REGISTRY: oci://ghcr.io/bmscomp/charts + # github.ref, so a signature names this tag, as publish-charts.yml's do. + IDENTITY: https://github.com/${{ github.repository }}/.github/workflows/publish-operator.yml@${{ github.ref }} + ISSUER: https://token.actions.githubusercontent.com + +jobs: + check: + name: Tag matches the chart + runs-on: ubuntu-latest + timeout-minutes: 5 + outputs: + version: ${{ steps.version.outputs.version }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - id: version + run: | + set -euo pipefail + version=${GITHUB_REF_NAME#operator-v} + chart=$(yq .version operator/chart/Chart.yaml) + app=$(yq .appVersion operator/chart/Chart.yaml) + if [ "${version}" != "${chart}" ] || [ "${version}" != "${app}" ]; then + echo "::error::tag ${version}, chart ${chart}, appVersion ${app}: all three must match" + exit 1 + fi + echo "version=${version}" >> "${GITHUB_OUTPUT}" + + build: + name: Build (${{ matrix.arch }}) + needs: check + runs-on: ${{ matrix.runner }} + timeout-minutes: 45 + permissions: + contents: read + packages: write + strategy: + matrix: + include: + - {arch: amd64, runner: ubuntu-latest} + - {arch: arm64, runner: ubuntu-24.04-arm} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: docker/setup-buildx-action@f87e5991a6d7451dcb8d9637bfbc97413f497069 # v4.4.1 + - uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - uses: docker/build-push-action@c3c9e263c25d99ce0380d002d59b67737d91b0dc # v7.4.0 + with: + context: . + file: operator/Dockerfile + platforms: linux/${{ matrix.arch }} + push: true + tags: ${{ env.IMAGE }}:sha-${{ github.sha }}-${{ matrix.arch }} + provenance: mode=max + sbom: true + + publish: + name: Merge, sign, push the chart + needs: [check, build] + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + packages: write + id-token: write # cosign keyless signing + env: + VERSION: ${{ needs.check.outputs.version }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: ./.github/actions/load-versions + - uses: azure/setup-helm@9bc31f4ebc9c6b171d7bfbaa5d006ae7abdb4310 # v5.0.1 + with: + version: ${{ env.HELM_VERSION }} + - uses: sigstore/cosign-installer@6f9f17788090df1f26f669e9d70d6ae9567deba6 # v4.1.2 + - uses: docker/setup-buildx-action@f87e5991a6d7451dcb8d9637bfbc97413f497069 # v4.4.1 + - uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - name: Helm logs in + env: + TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: helm registry login ghcr.io -u "${GITHUB_ACTOR}" --password-stdin <<<"${TOKEN}" + + - name: Merge and sign the image + run: | + set -euo pipefail + docker buildx imagetools create -t "${IMAGE}:${VERSION}" \ + "${IMAGE}:sha-${GITHUB_SHA}-amd64" "${IMAGE}:sha-${GITHUB_SHA}-arm64" + digest=$(docker buildx imagetools inspect "${IMAGE}:${VERSION}" --format '{{json .Manifest.Digest}}' | tr -d '"') + cosign sign --yes "${IMAGE}@${digest}" + cosign verify "${IMAGE}@${digest}" --certificate-identity "${IDENTITY}" \ + --certificate-oidc-issuer "${ISSUER}" > /dev/null + + - name: Push and sign the chart + run: | + set -euo pipefail + # helm package is not reproducible: a re-run would replace a digest + # someone may already pin, so a published version is never pushed again. + if helm show chart "${CHART_REGISTRY}/kates-operator" --version "${VERSION}" > /dev/null 2>&1; then + echo "kates-operator ${VERSION} is already published" + exit 0 + fi + helm package operator/chart --destination .build/ + out=$(helm push ".build/kates-operator-${VERSION}.tgz" "${CHART_REGISTRY}" 2>&1) + echo "${out}" + digest=$(sed -n 's/^Digest: //p' <<<"${out}") + repo=ghcr.io/bmscomp/charts/kates-operator + cosign sign --yes "${repo}@${digest}" + cosign verify "${repo}@${digest}" --certificate-identity "${IDENTITY}" \ + --certificate-oidc-issuer "${ISSUER}" > /dev/null 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/.helmignore b/operator/chart/.helmignore new file mode 100644 index 00000000..d69cba37 --- /dev/null +++ b/operator/chart/.helmignore @@ -0,0 +1,2 @@ +.git +*.md.orig diff --git a/operator/chart/Chart.yaml b/operator/chart/Chart.yaml new file mode 100644 index 00000000..6a31d86d --- /dev/null +++ b/operator/chart/Chart.yaml @@ -0,0 +1,17 @@ +apiVersion: v2 +name: kates-operator +description: Kates runs and compliance checks as Kubernetes resources, driving an existing Kates API +type: application +version: 0.1.1 +appVersion: "0.1.1" +kubeVersion: ">=1.29.0-0" +home: https://github.com/bmscomp/kates +sources: + - https://github.com/bmscomp/kates +maintainers: + - name: bmscomp + url: https://github.com/bmscomp +keywords: + - kafka + - kates + - operator diff --git a/operator/chart/README.md b/operator/chart/README.md new file mode 100644 index 00000000..dc11b626 --- /dev/null +++ b/operator/chart/README.md @@ -0,0 +1,216 @@ +# kates-operator + +Kates runs and compliance checks as Kubernetes resources. `KatesTest` is one run and `KafkaCompliance` a standing check. A team can commit `loadtest.yaml`, watch it with `kubectl get katestests`, and gate a pipeline on it. + +The operator drives a Kates API that is already running, through its REST API. It doesn't install Kates, and it needs Kates 1.38.0 or newer: older versions can't honour an idempotency key, so every `KatesTest` waits at `Synced=False`, reason `ApiTooOld`, and starts no run. + +## Install + +```bash +helm install kates-operator oci://ghcr.io/bmscomp/charts/kates-operator --version 0.1.1 \ + -n kates-operator --create-namespace +``` + +It needs Kubernetes 1.29 or newer and pulls `ghcr.io/bmscomp/kates-operator`, for amd64 and arm64. On a cluster that pulls from a mirror, set `global.imageRegistry` and `global.imagePullSecrets`. The image and the chart are signed with cosign: + +```bash +cosign verify ghcr.io/bmscomp/kates-operator:0.1.1 \ + --certificate-identity-regexp 'https://github.com/bmscomp/kates/.github/workflows/publish-operator.yml@refs/tags/operator-v.*' \ + --certificate-oidc-issuer https://token.actions.githubusercontent.com +``` + +To run your own build, from the repository root: + +```bash +docker build -f operator/Dockerfile -t /kates-operator:dev . +docker push /kates-operator:dev +helm install kates-operator operator/chart -n kates-operator --create-namespace \ + --set image.repository=/kates-operator --set image.tag=dev +``` + +| Mode | Values | Rendered | +| --- | --- | --- | +| Its own namespace | (default) | A ClusterRole, a RoleBinding in the release namespace | +| Some namespaces | `watchNamespaces: [a, b]` | The same ClusterRole, a RoleBinding in each | +| Every namespace | `watchAnyNamespace: true` | The same ClusterRole, a ClusterRoleBinding | +| No cluster-scoped rights | `rbac.clusterRoles.create: false` | A Role in the release namespace, and nothing cluster-scoped | + +For the last mode a cluster admin applies the CRDs and the aggregated roles once, and the namespace admin installs with `--skip-crds`: + +```bash +helm pull oci://ghcr.io/bmscomp/charts/kates-operator --version 0.1.1 --untar +kubectl apply --server-side -f kates-operator/crds/ +helm template kates-operator ./kates-operator -n team --show-only templates/aggregate-roles.yaml | kubectl apply -f - +helm install kates-operator ./kates-operator -n team --skip-crds \ + --set rbac.clusterRoles.create=false --set rbac.aggregateRoles.create=false +``` + +CRDs are applied by a pre-install and pre-upgrade hook that never downgrades: with several releases on a cluster, the newest wins. On a cluster that already has the CRDs, pass `--skip-crds`. A conflict on `customresourcedefinitions.apiextensions.k8s.io` during install means the flag was missed, and `--force-conflicts` is a downgrade, not the fix. + +Don't let two releases watch the same namespace: each would reconcile the same resources. + +## Connect a namespace + +Each watched namespace holds a Secret named `kates-connection`, with the keys `url`, `apiKey` and optional `caBundle`: + +```bash +kubectl create secret generic kates-connection -n payments \ + --from-literal=url=http://kates.kates.svc:8080 --from-literal=apiKey= +``` + +The key needs `read` and `test:run`. Use a named key from the Kates keys file, so the audit log names the team. `connection.fallback: true` lends the Secret `connection.existingSecret` to every watched namespace that has no `kates-connection`. With `watchAnyNamespace`, that lends the key to everyone who can create a `KatesTest`. + +To rotate a key, add the new hash to the Kates keys file, swap the Secret, then remove the old hash. The operator reads the key on every pass, so it needs no restart. + +## KatesTest + +```yaml +apiVersion: kates.io/v1alpha1 +kind: KatesTest +metadata: + name: nightly-load +spec: + type: LOAD + spec: + numRecords: 100000 + acks: all + sla: + maxP99LatencyMs: 100 +``` + +`spec.spec` takes the wire names of `POST /api/tests`, and the CRD refuses a mistyped key. `spec.sla` is graded by the Kates API, the same way the report and the JUnit export grade it. Once the run has started, the whole `spec` is immutable, because the resource is the evidence of what the run was judged against. + +A CI gate is two commands, race-free on either outcome: + +```bash +kubectl wait --for=jsonpath='{.status.finishedAt}' katestest/nightly-load --timeout=30m +kubectl wait --for=condition=Ready katestest/nightly-load --timeout=0 +``` + +The first returns on every ending, and the second exits non-zero on a failed run. `kubectl describe` shows the failing key. The value-less `jsonpath` form needs kubectl 1.29 or newer. + +Deleting a `KatesTest` whose run is live cancels the run first. The run's record stays in Kates. A deleted and re-created `KatesTest` is a new run. + +### On a schedule + +There is no schedule kind. A `CronJob` creates one `KatesTest` per firing, and waits for it so `Forbid` holds across firings: + +```yaml +apiVersion: v1 +kind: ServiceAccount +metadata: + name: nightly-load +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: nightly-load +rules: + - apiGroups: [kates.io] + resources: [katestests] + verbs: [create, get, list, watch] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: nightly-load +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: nightly-load +subjects: + - kind: ServiceAccount + name: nightly-load +--- +apiVersion: batch/v1 +kind: CronJob +metadata: + name: nightly-load +spec: + schedule: "0 2 * * *" + timeZone: Etc/UTC + concurrencyPolicy: Forbid + startingDeadlineSeconds: 600 + jobTemplate: + spec: + backoffLimit: 0 + template: + spec: + serviceAccountName: nightly-load + restartPolicy: Never + containers: + - name: run + image: ghcr.io/bmscomp/kates-tester:1.38.0 + command: [/bin/sh, -ec] + args: + - | + run=$(kubectl create -o name -f - <` | Forces a pass, recorded in `status.lastHandledReconcileAt` | + +Before `helm uninstall`, before dropping a namespace from `watchNamespaces`, and before deleting a namespace, delete its live `KatesTest`s. A finished `KatesTest` holds no finalizer and blocks nothing. + +A live one left behind needs a hand. With `watchNamespaces` or the default mode, the RoleBinding that gives the operator its rights lives in the namespace, so `kubectl delete namespace` removes those rights along with everything else. The operator can then no longer clear its finalizer, and the namespace stays Terminating until you do: + +```bash +kubectl patch katestest -n --type=merge -p '{"metadata":{"finalizers":null}}' +``` + +Kates ends the run at its own timeout. With `watchAnyNamespace` the ClusterRoleBinding survives: if the Secret goes first, the finalizer comes off with a `CancelSkipped` Event. + +The NetworkPolicy allows egress on 8080, 443 and 6443. A Kates URL on any other port needs `networkPolicy.egressRules`, or the blocked call shows as `ApiUnavailable`. + +Metrics are on `:8080/metrics` under `kates_operator_`, and `monitoring.enabled` renders a PodMonitor. `/readyz` fails on an unreachable API server or a stale served CRD, and it reports the Lease and any refused Kafka namespace in its body without failing on them. + +One pod holds the Lease `kates-operator` in the release namespace. `replicas: 2` only shortens the gap after a crash, to between 5 and 15 seconds. 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/chart/templates/NOTES.txt b/operator/chart/templates/NOTES.txt new file mode 100644 index 00000000..7b440fd9 --- /dev/null +++ b/operator/chart/templates/NOTES.txt @@ -0,0 +1,10 @@ +kates-operator reconciles KatesTest and KafkaCompliance in: {{ include "kates-operator.namespaces" . }} + +Each of those namespaces needs a Secret named kates-connection (keys url, apiKey, optional caBundle): + + kubectl create secret generic kates-connection -n \ + --from-literal=url=http://kates.kates.svc:8080 --from-literal=apiKey= + +Then: + + kubectl get katestests,kafkacompliances -A diff --git a/operator/chart/templates/_helpers.tpl b/operator/chart/templates/_helpers.tpl new file mode 100644 index 00000000..eaa942e2 --- /dev/null +++ b/operator/chart/templates/_helpers.tpl @@ -0,0 +1,71 @@ +{{- define "kates-operator.fullname" -}} +{{- printf "kates-operator-%s" .Release.Namespace | trunc 63 | trimSuffix "-" -}} +{{- end -}} + +{{- define "kates-operator.selectorLabels" -}} +app.kubernetes.io/name: kates-operator +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end -}} + +{{- define "kates-operator.labels" -}} +{{ include "kates-operator.selectorLabels" . }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +helm.sh/chart: {{ printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" }} +{{- end -}} + +{{- define "kates-operator.image" -}} +{{- $image := printf "%s:%s" .Values.image.repository (.Values.image.tag | default .Chart.AppVersion) -}} +{{- with .Values.image.digest -}}{{- $image = printf "%s@%s" $.Values.image.repository . -}}{{- end -}} +{{- with .Values.global.imageRegistry -}}{{ printf "%s/%s" (trimSuffix "/" .) $image }}{{- else -}}{{ $image }}{{- end -}} +{{- end -}} + +{{- define "kates-operator.imagePullSecrets" -}} +{{- $secrets := concat (.Values.global.imagePullSecrets | default list) (.Values.imagePullSecrets | default list) -}} +{{- with $secrets }} +imagePullSecrets: + {{- range . }} + - name: {{ if kindIs "string" . }}{{ . }}{{ else }}{{ .name }}{{ end }} + {{- end }} +{{- end }} +{{- end -}} + +{{/* The namespaces this release reconciles, or * for all of them. */}} +{{- define "kates-operator.namespaces" -}} +{{- if .Values.watchAnyNamespace -}}* +{{- else if .Values.watchNamespaces -}}{{ join "," .Values.watchNamespaces }} +{{- else -}}{{ .Release.Namespace }} +{{- end -}} +{{- end -}} + +{{/* What the operator may do where it reconciles. */}} +{{- define "kates-operator.rules" -}} +- apiGroups: ["kates.io"] + resources: ["katestests", "kafkacompliances"] + verbs: ["get", "list", "watch", "patch", "update", "delete"] +- apiGroups: ["kates.io"] + resources: ["katestests/status", "kafkacompliances/status", "katestests/finalizers", "kafkacompliances/finalizers"] + verbs: ["get", "patch", "update"] +- apiGroups: ["events.k8s.io"] + resources: ["events"] + verbs: ["create", "patch"] +- apiGroups: [""] + resources: ["secrets"] + verbs: ["get"] + resourceNames: ["kates-connection"] +{{- end -}} + +{{- define "kates-operator.podSecurityContext" -}} +runAsNonRoot: true +runAsUser: 65532 +runAsGroup: 65532 +seccompProfile: + type: RuntimeDefault +{{- end -}} + +{{- define "kates-operator.securityContext" -}} +allowPrivilegeEscalation: false +readOnlyRootFilesystem: true +capabilities: + drop: + - ALL +{{- end -}} diff --git a/operator/chart/templates/aggregate-roles.yaml b/operator/chart/templates/aggregate-roles.yaml new file mode 100644 index 00000000..e736d4e7 --- /dev/null +++ b/operator/chart/templates/aggregate-roles.yaml @@ -0,0 +1,35 @@ +{{- if .Values.rbac.aggregateRoles.create }} +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: kates-operator-view-{{ .Release.Namespace }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + rbac.authorization.k8s.io/aggregate-to-view: "true" +rules: + - apiGroups: ["kates.io"] + resources: ["katestests", "kafkacompliances"] + verbs: ["get", "list", "watch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: kates-operator-edit-{{ .Release.Namespace }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + rbac.authorization.k8s.io/aggregate-to-edit: "true" +rules: + - apiGroups: ["kates.io"] + resources: ["katestests", "kafkacompliances"] + verbs: ["*"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: kates-operator-admin-{{ .Release.Namespace }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + rbac.authorization.k8s.io/aggregate-to-admin: "true" +rules: + {{- include "kates-operator.rules" . | nindent 2 }} +{{- end }} diff --git a/operator/chart/templates/crd-upgrade.yaml b/operator/chart/templates/crd-upgrade.yaml new file mode 100644 index 00000000..2858da86 --- /dev/null +++ b/operator/chart/templates/crd-upgrade.yaml @@ -0,0 +1,83 @@ +{{- if and .Values.crdUpgrade.enabled .Values.rbac.clusterRoles.create }} +{{- $name := printf "kates-operator-crd-upgrade-%s" .Release.Namespace | trunc 63 | trimSuffix "-" }} +apiVersion: v1 +kind: ServiceAccount +metadata: + name: kates-operator-crd-upgrade + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + annotations: + helm.sh/hook: pre-install,pre-upgrade + helm.sh/hook-weight: "1" + helm.sh/hook-delete-policy: before-hook-creation,hook-succeeded +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: {{ $name }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + annotations: + helm.sh/hook: pre-install,pre-upgrade + helm.sh/hook-weight: "1" + helm.sh/hook-delete-policy: before-hook-creation,hook-succeeded +rules: + - apiGroups: ["apiextensions.k8s.io"] + resources: ["customresourcedefinitions"] + resourceNames: ["katestests.kates.io", "kafkacompliances.kates.io"] + verbs: ["get", "create", "patch"] + - apiGroups: ["apiextensions.k8s.io"] + resources: ["customresourcedefinitions"] + verbs: ["list", "watch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: {{ $name }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + annotations: + helm.sh/hook: pre-install,pre-upgrade + helm.sh/hook-weight: "1" + helm.sh/hook-delete-policy: before-hook-creation,hook-succeeded +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: {{ $name }} +subjects: + - kind: ServiceAccount + name: kates-operator-crd-upgrade + namespace: {{ .Release.Namespace }} +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: kates-operator-crd-upgrade + labels: + {{- include "kates-operator.labels" . | nindent 4 }} + annotations: + helm.sh/hook: pre-install,pre-upgrade + helm.sh/hook-weight: "5" + helm.sh/hook-delete-policy: before-hook-creation,hook-succeeded +spec: + backoffLimit: 3 + template: + metadata: + labels: + {{- toYaml (merge (dict "app.kubernetes.io/component" "crd-upgrade" "app.kubernetes.io/managed-by" .Release.Service) .Values.podLabels) | nindent 8 }} + spec: + restartPolicy: Never + serviceAccountName: kates-operator-crd-upgrade + {{- include "kates-operator.imagePullSecrets" . | nindent 6 }} + securityContext: + {{- include "kates-operator.podSecurityContext" . | nindent 8 }} + containers: + - name: crd-upgrade + image: {{ include "kates-operator.image" . | quote }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + args: ["crds", "apply"] + securityContext: + {{- include "kates-operator.securityContext" . | nindent 12 }} + resources: + {{- toYaml .Values.crdUpgrade.resources | nindent 12 }} +{{- end }} diff --git a/operator/chart/templates/deployment.yaml b/operator/chart/templates/deployment.yaml new file mode 100644 index 00000000..98c67921 --- /dev/null +++ b/operator/chart/templates/deployment.yaml @@ -0,0 +1,97 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: kates-operator + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +spec: + replicas: {{ .Values.replicas }} + # A rolling update would run two reconcilers until the old pod lets go of the Lease. + strategy: + type: Recreate + selector: + matchLabels: + {{- include "kates-operator.selectorLabels" . | nindent 6 }} + template: + metadata: + labels: + {{- toYaml (merge (include "kates-operator.labels" . | fromYaml) .Values.podLabels) | nindent 8 }} + {{- with .Values.podAnnotations }} + annotations: + {{- toYaml . | nindent 8 }} + {{- end }} + spec: + serviceAccountName: kates-operator + {{- include "kates-operator.imagePullSecrets" . | nindent 6 }} + securityContext: + {{- include "kates-operator.podSecurityContext" . | nindent 8 }} + {{- with .Values.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: operator + image: {{ include "kates-operator.image" . | quote }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + securityContext: + {{- include "kates-operator.securityContext" . | nindent 12 }} + ports: + - name: http + containerPort: 8080 + protocol: TCP + env: + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: KATES_OPERATOR_NAMESPACES + value: {{ include "kates-operator.namespaces" . | quote }} + - name: KATES_OPERATOR_KAFKA_NAMESPACES + value: {{ if .Values.watchAnyNamespace }}"*"{{ else }}{{ join "," .Values.kafka.namespaces | quote }}{{ end }} + - name: LOG_FORMAT + value: {{ .Values.logFormat | quote }} + {{- with .Values.tracing.endpoint }} + - name: OTEL_EXPORTER_OTLP_ENDPOINT + value: {{ . | quote }} + {{- end }} + livenessProbe: + httpGet: + path: /healthz + port: http + periodSeconds: 30 + readinessProbe: + httpGet: + path: /readyz + port: http + periodSeconds: 10 + resources: + {{- toYaml .Values.resources | nindent 12 }} + {{- if .Values.connection.fallback }} + volumeMounts: + - name: connection + mountPath: /etc/kates-operator/connection + readOnly: true + {{- end }} + {{- if .Values.connection.fallback }} + # Projected with no subPath, so the kubelet refreshes a rotated key; optional, so a + # missing Secret leaves the resources at NoConnection and the pod running. + volumes: + - name: connection + projected: + sources: + - secret: + name: {{ .Values.connection.existingSecret }} + optional: true + {{- end }} diff --git a/operator/chart/templates/kafka-reader.yaml b/operator/chart/templates/kafka-reader.yaml new file mode 100644 index 00000000..de5cc3f9 --- /dev/null +++ b/operator/chart/templates/kafka-reader.yaml @@ -0,0 +1,34 @@ +{{- if and .Values.rbac.kafkaNamespaceRole (not .Values.watchAnyNamespace) }} +{{- $root := . }} +{{- $name := printf "kates-operator-kafka-reader-%s" .Release.Namespace | trunc 63 | trimSuffix "-" }} +{{- range .Values.kafka.namespaces }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: {{ $name }} + namespace: {{ . }} + labels: + {{- include "kates-operator.labels" $root | nindent 4 }} +rules: + - apiGroups: ["kafka.strimzi.io"] + resources: ["kafkas"] + verbs: ["get", "list", "watch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: {{ $name }} + namespace: {{ . }} + labels: + {{- include "kates-operator.labels" $root | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: {{ $name }} +subjects: + - kind: ServiceAccount + name: kates-operator + namespace: {{ $root.Release.Namespace }} +{{- end }} +{{- end }} diff --git a/operator/chart/templates/networkpolicy.yaml b/operator/chart/templates/networkpolicy.yaml new file mode 100644 index 00000000..677d7c5b --- /dev/null +++ b/operator/chart/templates/networkpolicy.yaml @@ -0,0 +1,43 @@ +{{- if .Values.networkPolicy.enabled }} +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: kates-operator + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +spec: + podSelector: + matchLabels: + {{- include "kates-operator.selectorLabels" . | nindent 6 }} + policyTypes: + - Ingress + - Egress + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: {{ .Values.networkPolicy.monitoringNamespace }} + ports: + - port: http + protocol: TCP + egress: + - ports: + - port: 53 + protocol: UDP + - port: 53 + protocol: TCP + # Ports only, with no to: the template cannot know which namespace a kates-connection names. + - ports: + - port: 8080 + protocol: TCP + # 6443 is the API server's own port after the kubernetes Service's DNAT, and NetworkPolicy + # leaves undefined whether a policy runs before or after that rewrite. + - ports: + - port: 443 + protocol: TCP + - port: 6443 + protocol: TCP + {{- with .Values.networkPolicy.egressRules }} + {{- toYaml . | nindent 4 }} + {{- end }} +{{- end }} diff --git a/operator/chart/templates/podmonitor.yaml b/operator/chart/templates/podmonitor.yaml new file mode 100644 index 00000000..48d69101 --- /dev/null +++ b/operator/chart/templates/podmonitor.yaml @@ -0,0 +1,15 @@ +{{- if .Values.monitoring.enabled }} +apiVersion: monitoring.coreos.com/v1 +kind: PodMonitor +metadata: + name: kates-operator + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +spec: + selector: + matchLabels: + {{- include "kates-operator.selectorLabels" . | nindent 6 }} + podMetricsEndpoints: + - port: http + path: /metrics +{{- end }} diff --git a/operator/chart/templates/rbac.yaml b/operator/chart/templates/rbac.yaml new file mode 100644 index 00000000..b50386af --- /dev/null +++ b/operator/chart/templates/rbac.yaml @@ -0,0 +1,110 @@ +{{- $name := include "kates-operator.fullname" . -}} +{{- $clusterScoped := .Values.rbac.clusterRoles.create -}} +{{- if not $clusterScoped }} +{{- if or .Values.watchNamespaces .Values.watchAnyNamespace }} +{{- fail "rbac.clusterRoles.create=false installs in the release namespace alone: leave watchNamespaces and watchAnyNamespace unset" }} +{{- end }} +{{- if .Values.kafka.namespaces }} +{{- fail "rbac.clusterRoles.create=false takes no kafka.namespaces: a Kafka namespace's Role is that namespace's admin's to apply, and compliance then runs on its interval" }} +{{- end }} +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: kates-operator + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +rules: + {{- include "kates-operator.rules" . | nindent 2 }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: kates-operator + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: kates-operator +subjects: + - kind: ServiceAccount + name: kates-operator + namespace: {{ .Release.Namespace }} +{{- else }} +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: kates-operator-namespaced-{{ .Release.Namespace }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +rules: + {{- include "kates-operator.rules" . | nindent 2 }} + {{- if .Values.watchAnyNamespace }} + - apiGroups: ["kafka.strimzi.io"] + resources: ["kafkas"] + verbs: ["get", "list", "watch"] + {{- end }} +{{- if .Values.watchAnyNamespace }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: {{ $name }} + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: kates-operator-namespaced-{{ .Release.Namespace }} +subjects: + - kind: ServiceAccount + name: kates-operator + namespace: {{ .Release.Namespace }} +{{- else }} +{{- $root := . }} +{{- range ternary .Values.watchNamespaces (list .Release.Namespace) (gt (len .Values.watchNamespaces) 0) }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: {{ $name }} + namespace: {{ . }} + labels: + {{- include "kates-operator.labels" $root | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: kates-operator-namespaced-{{ $root.Release.Namespace }} +subjects: + - kind: ServiceAccount + name: kates-operator + namespace: {{ $root.Release.Namespace }} +{{- end }} +{{- end }} +{{- end }} +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: kates-operator-leader-election + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +rules: + - apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["get", "list", "watch", "create", "update", "patch"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: kates-operator-leader-election + labels: + {{- include "kates-operator.labels" . | nindent 4 }} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: kates-operator-leader-election +subjects: + - kind: ServiceAccount + name: kates-operator + namespace: {{ .Release.Namespace }} diff --git a/operator/chart/templates/serviceaccount.yaml b/operator/chart/templates/serviceaccount.yaml new file mode 100644 index 00000000..d9db1d4a --- /dev/null +++ b/operator/chart/templates/serviceaccount.yaml @@ -0,0 +1,6 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: kates-operator + labels: + {{- include "kates-operator.labels" . | nindent 4 }} diff --git a/operator/chart/values.schema.json b/operator/chart/values.schema.json new file mode 100644 index 00000000..38d75560 --- /dev/null +++ b/operator/chart/values.schema.json @@ -0,0 +1,105 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "kates-operator values", + "type": "object", + "additionalProperties": false, + "properties": { + "global": { + "type": "object", + "properties": { + "imageRegistry": { "type": "string" }, + "imagePullSecrets": { "$ref": "#/$defs/pullSecrets" } + } + }, + "image": { + "type": "object", + "additionalProperties": false, + "properties": { + "repository": { "type": "string", "minLength": 1 }, + "tag": { "type": "string", "not": { "const": "latest" } }, + "digest": { "type": "string", "pattern": "^(sha256:[0-9a-f]{64})?$" }, + "pullPolicy": { "enum": ["Always", "IfNotPresent", "Never"] } + } + }, + "imagePullSecrets": { "$ref": "#/$defs/pullSecrets" }, + "watchNamespaces": { "$ref": "#/$defs/names" }, + "watchAnyNamespace": { "type": "boolean" }, + "kafka": { + "type": "object", + "additionalProperties": false, + "properties": { "namespaces": { "$ref": "#/$defs/names" } } + }, + "rbac": { + "type": "object", + "additionalProperties": false, + "properties": { + "clusterRoles": { + "type": "object", + "additionalProperties": false, + "properties": { "create": { "type": "boolean" } } + }, + "aggregateRoles": { + "type": "object", + "additionalProperties": false, + "properties": { "create": { "type": "boolean" } } + }, + "kafkaNamespaceRole": { "type": "boolean" } + } + }, + "connection": { + "type": "object", + "additionalProperties": false, + "properties": { + "fallback": { "type": "boolean" }, + "existingSecret": { "type": "string", "minLength": 1 } + } + }, + "crdUpgrade": { + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { "type": "boolean" }, + "resources": { "type": "object" } + } + }, + "replicas": { "type": "integer", "minimum": 1 }, + "resources": { "type": "object" }, + "logFormat": { "enum": ["text", "json"] }, + "tracing": { + "type": "object", + "additionalProperties": false, + "properties": { "endpoint": { "type": "string" } } + }, + "networkPolicy": { + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { "type": "boolean" }, + "monitoringNamespace": { "type": "string" }, + "egressRules": { "type": "array", "items": { "type": "object" } } + } + }, + "monitoring": { + "type": "object", + "additionalProperties": false, + "properties": { "enabled": { "type": "boolean" } } + }, + "podLabels": { "type": "object", "additionalProperties": { "type": "string" } }, + "podAnnotations": { "type": "object" }, + "nodeSelector": { "type": "object" }, + "tolerations": { "type": "array" }, + "affinity": { "type": "object" } + }, + "$defs": { + "names": { "type": "array", "items": { "type": "string", "minLength": 1 } }, + "pullSecrets": { + "type": "array", + "items": { + "anyOf": [ + { "type": "string", "minLength": 1 }, + { "type": "object", "required": ["name"], "properties": { "name": { "type": "string", "minLength": 1 } } } + ] + } + } + } +} diff --git a/operator/chart/values.yaml b/operator/chart/values.yaml new file mode 100644 index 00000000..d7f3eb2e --- /dev/null +++ b/operator/chart/values.yaml @@ -0,0 +1,99 @@ +# The Kates operator. It drives a Kates API that is already running and never +# installs one. A resource reads its API and key from the Secret +# kates-connection in its own namespace (keys url, apiKey, optional caBundle). + +global: + # -- Registry prepended to the image, with a slash, for clusters that pull + # from a mirror. + imageRegistry: "" + # -- Pull secrets for that registry, as names or {name: …}. + imagePullSecrets: [] + +image: + repository: ghcr.io/bmscomp/kates-operator + # -- Empty: the chart's appVersion. + tag: "" + # -- sha256:… pins by content and wins over the tag. + digest: "" + pullPolicy: IfNotPresent + +# -- Pull secrets of this chart's pods. +imagePullSecrets: [] + +# -- Namespaces whose resources this release reconciles. Empty: its own. +watchNamespaces: [] +# -- Reconcile every namespace, with a ClusterRoleBinding. +watchAnyNamespace: false + +kafka: + # -- Namespaces whose Strimzi Kafkas trigger a compliance check when they + # change. Empty: checks run on their interval only. + namespaces: [] + +rbac: + clusterRoles: + # -- false renders nothing cluster-scoped, for a namespace admin: apply the + # CRDs and aggregate-roles.yaml once as a cluster admin, and install with + # --skip-crds. Not with watchNamespaces or watchAnyNamespace. + create: true + aggregateRoles: + # -- Fold the kinds into the built-in view, edit and admin roles. + create: true + # -- Render the Role that lets the operator read Kafkas in kafka.namespaces. + # false: someone with rights there applies it from helm template --show-only. + kafkaNamespaceRole: true + +connection: + # -- Lend every watched namespace the Secret below when it has no + # kates-connection of its own. With watchAnyNamespace that lends this key to + # every user who can create a KatesTest. + fallback: false + # -- The fallback Secret (keys url, apiKey, optional caBundle). This chart + # never renders it, so no key sits in Helm values. + existingSecret: kates-operator-connection + +crdUpgrade: + # -- Apply the CRDs from a pre-install and pre-upgrade hook, never over a + # newer version. Off with rbac.clusterRoles.create=false. + enabled: true + resources: + requests: + cpu: 10m + memory: 32Mi + limits: + memory: 64Mi + +# -- 2 only shortens the gap after a crash: one pod holds the Lease. +replicas: 1 + +resources: + requests: + cpu: 25m + memory: 48Mi + limits: + memory: 128Mi + +# -- text or json. +logFormat: json + +tracing: + # -- OTLP endpoint; empty sends no traces. + endpoint: "" + +networkPolicy: + enabled: true + # -- Namespace allowed to scrape metrics on 8080. + monitoringNamespace: monitoring + # -- Extra egress rules, for a Kates URL on a port other than 8080 or 443. + egressRules: [] + +monitoring: + # -- Render a PodMonitor. Needs the Prometheus Operator's CRDs. + enabled: false + +# -- Labels added to the pods; the chart's own win on a clash. +podLabels: {} +podAnnotations: {} +nodeSelector: {} +tolerations: [] +affinity: {} 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..0d41972f --- /dev/null +++ b/operator/crates/kates-client/src/lib.rs @@ -0,0 +1,760 @@ +//! 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::{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); +const MAX_BODY: usize = 8 * 1024 * 1024; + +/// 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)) + .bearer_auth(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 neither the URL nor the headers here, so no key can reach it. + detail: chain(&error.without_url()), + })?; + 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 = capped(response).await.unwrap_or_default(); + let body = String::from_utf8_lossy(&body); + Err(classify(status, &headers, &body, &host)) + } +} + +// reqwest's own text is "error sending request"; the cause is at the end of its sources. +fn chain(error: &dyn std::error::Error) -> String { + let mut text = error.to_string(); + let mut source = error.source(); + while let Some(cause) = source { + text = format!("{text}: {cause}"); + source = cause.source(); + } + text +} + +async fn decode(response: Response) -> Result { + let status = response.status().as_u16(); + let unavailable = |detail| Error::Unavailable { + status: Some(status), + api_message: None, + detail: format!("an answer that is not the API's: {detail}"), + }; + let body = capped(response).await.map_err(unavailable)?; + serde_json::from_slice(&body).map_err(|error| unavailable(error.to_string())) +} + +// A kates-connection can name any host, and one huge answer would kill the pod every namespace +// shares. +async fn capped(mut response: Response) -> Result, String> { + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|error| error.without_url().to_string())? + { + if body.len() + chunk.len() > MAX_BODY { + return Err(format!("over {MAX_BODY} bytes")); + } + body.extend_from_slice(&chunk); + } + Ok(body) +} + +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 an_answer_over_the_cap_is_an_outage() { + let error = + answering(ResponseTemplate::new(200).set_body_bytes(vec![b' '; MAX_BODY + 1])).await; + assert_eq!(error.name(), "Unavailable"); + assert!(error.to_string().contains("over"), "{error}"); + } + + #[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)]); + } + + #[tokio::test] + async fn a_network_error_says_why() { + let client = Client::new("http://127.0.0.1:1", None).unwrap(); + let error = client.get_test(&key(), "r").await.unwrap_err(); + let Error::Unavailable { status, detail, .. } = error else { + panic!("{error:?}"); + }; + assert_eq!(status, None); + assert!(detail.to_lowercase().contains("refused"), "{detail}"); + } + + #[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