diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c581dd..0d015bc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ jobs: - uses: dtolnay/rust-toolchain@stable with: - toolchain: "1.80" + toolchain: "1.88" components: rustfmt, clippy - run: cargo fmt --all -- --check diff --git a/AUDIT.md b/AUDIT.md new file mode 100644 index 0000000..b3e0e55 --- /dev/null +++ b/AUDIT.md @@ -0,0 +1,42 @@ +# Clean Code and SRP Audit + +## Summary + +- Controllers are already separated by Kubernetes resource actor; do not merge + their similar finalizer/status/requeue shapes into a generic controller. +- **Highest-leverage future split:** isolate StatefulSet/Service/ConfigMap + rendering from cluster reconciliation once golden manifest tests cover every + option. +- `main.rs` now correctly owns startup and shutdown of controller/health/metrics + tasks; further splitting would be small orchestration helpers, not new types. +- CRD modules are public serialized contracts and remain intact. +- Baseline formatting, strict Clippy, tests, integration gating, and auxiliary + task shutdown are green. + +## Findings + +| ID | Location | Category | Severity | Actors in conflict | Cost | Size | Behavior risk | +|---|---|---|---|---|---|---|---| +| OP-SRP-1 | `controllers/cluster.rs` | Mixed reconciliation/rendering | P2 | cluster product; Kubernetes workload API | Desired-state construction and reconciliation/status orchestration change for separate actors. | L | High | +| OP-CC-1 | controller modules | Similar but actor-distinct code | P2 | cluster/topic/user/contract/branch/memory actors | Generic deduplication would couple independent CRD evolution and worsen SRP. | M | High | + +## Ordered Refactor Sequence + +1. Add golden tests for rendered cluster workloads across storage, auth, TLS, + replicas, resources, and service settings. +2. Move rendering unchanged into `cluster_resources`. +3. Keep finalizer/requeue/status orchestration in `ClusterController`. +4. Do not create generic controller traits unless two CRDs share the same actor + and lifecycle contract. + +## Deferred + +- Cluster renderer extraction lacks complete golden coverage. +- Live Kubernetes integration requires cluster credentials and a server image. +- CRD changes require versioned schema/release decisions. + +## Out of Scope + +- Per-CRD controllers: actor-aligned. +- CRD modules: serialization contracts. +- Metrics and leader election: independent operational actors. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index efdbda6..9c2feba 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -6,7 +6,7 @@ Thank you for your interest in contributing! Please review the [organization-wid ### Prerequisites -- Rust 1.75+ (`rustup update stable`) +- Rust 1.88+ (`rustup update stable`) - Docker (for integration tests) - kubectl + a Kubernetes cluster (for e2e tests, [kind](https://kind.sigs.k8s.io/) recommended) diff --git a/Cargo.toml b/Cargo.toml index 493417e..ee8fdb0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ repository = "https://github.com/streamlinelabs/streamline-operator" readme = "README.md" keywords = ["kubernetes", "operator", "streamline", "kafka", "streaming"] categories = ["network-programming", "asynchronous"] -rust-version = "1.80" +rust-version = "1.88" [dependencies] # Kubernetes client and runtime @@ -65,5 +65,3 @@ tokio-test = "0.4" [lints.clippy] unwrap_used = "warn" expect_used = "warn" - - diff --git a/Makefile b/Makefile index 0ce51c0..b584aa1 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: build test lint fmt clean help check +.PHONY: build test lint fmt clean help check integration-up integration-down test-integration help: ## Show this help @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-15s\033[0m %s\n", $$1, $$2}' @@ -6,9 +6,18 @@ help: ## Show this help build: ## Build the operator cargo build -test: ## Run tests +test: ## Run tests (hermetic — no Kubernetes or Streamline required) cargo test +integration-up: ## Start integration services (override with STREAMLINE_TEST_IMAGE) + docker compose -f docker-compose.test.yml up -d --wait + +integration-down: ## Stop and remove integration services + docker compose -f docker-compose.test.yml down -v + +test-integration: ## Run explicitly gated integration tests (needs integration-up) + cargo test --test integration -- --ignored --test-threads=1 + lint: ## Run clippy lints cargo clippy --all-targets -- -D warnings diff --git a/README.md b/README.md index 3e59545..362adf6 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/streamlinelabs/streamline-operator/actions/workflows/ci.yml/badge.svg)](https://github.com/streamlinelabs/streamline-operator/actions/workflows/ci.yml) [![codecov](https://img.shields.io/codecov/c/github/streamlinelabs/streamline-operator?style=flat-square)](https://codecov.io/gh/streamlinelabs/streamline-operator) [![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://opensource.org/licenses/Apache-2.0) -[![Rust](https://img.shields.io/badge/Rust-1.80%2B-orange.svg)](https://www.rust-lang.org/) +[![Rust](https://img.shields.io/badge/Rust-1.88%2B-orange.svg)](https://www.rust-lang.org/) [![Kubernetes](https://img.shields.io/badge/Kubernetes-1.26+-326CE5.svg)](https://kubernetes.io/) [![Docs](https://img.shields.io/badge/docs-streamlinelabs.dev-brightgreen)](https://streamlinelabs.dev/docs/operations/kubernetes) [![Release](https://img.shields.io/github/v/release/streamlinelabs/streamline-operator?label=release)](https://github.com/streamlinelabs/streamline-operator/releases) @@ -289,7 +289,7 @@ spec: ### Prerequisites -- Rust 1.80+ +- Rust 1.88+ - Access to a Kubernetes cluster - `kubectl` configured @@ -318,6 +318,36 @@ cargo run -p streamline-operator -- --metrics-port 8080 --health-port 8081 cargo test ``` +The default test run is hermetic: it needs no Kubernetes cluster, no Streamline +server, and no Docker. + +#### Integration tests (opt-in) + +Tests that need a live Streamline server live in `tests/integration.rs` and are +`#[ignore]`d, so they never run as part of `cargo test`. Every networked +assertion is bounded by a timeout so a missing backend fails fast. + +```bash +make integration-up # start the server from docker-compose.test.yml +make test-integration # cargo test --test integration -- --ignored +make integration-down +``` + +The image and endpoints are configurable — see +[`docs/ENVIRONMENT.md`](docs/ENVIRONMENT.md#integration-test-variables): + +```bash +STREAMLINE_TEST_IMAGE=ghcr.io/streamlinelabs/streamline:0.3.0 \ +STREAMLINE_TEST_HTTP_PORT=19094 \ +STREAMLINE_TEST_KAFKA_PORT=19092 make integration-up + +STREAMLINE_TEST_HTTP_ENDPOINT=http://127.0.0.1:19094 \ +STREAMLINE_TEST_KAFKA_ENDPOINT=127.0.0.1:19092 make test-integration +``` + +A separate Helm/kubectl suite lives in `scripts/helm-integration-test.sh` and +requires a configured `kubectl` context. + ### Lint ```bash @@ -403,4 +433,3 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and guidelines. Apache License 2.0 — see [LICENSE](LICENSE) for details. - diff --git a/docker-compose.test.yml b/docker-compose.test.yml index 9f3fcb2..4ca3a61 100644 --- a/docker-compose.test.yml +++ b/docker-compose.test.yml @@ -1,22 +1,34 @@ # Docker Compose for operator integration testing # -# Starts a Streamline server and a kind-compatible Kubernetes API. -# The operator can be tested against these services locally. +# Starts a Streamline server for the explicitly gated integration suite in +# `tests/integration.rs`. Nothing in the default `cargo test` run depends on it. # -# Usage: docker compose -f docker-compose.test.yml up -d -version: "3.8" +# Usage: +# make integration-up # or: docker compose -f docker-compose.test.yml up -d --wait +# make test-integration # cargo test --test integration -- --ignored +# make integration-down +# +# Configuration (all optional — defaults match tests/integration.rs): +# STREAMLINE_TEST_IMAGE Server image (default ghcr.io/streamlinelabs/streamline:0.2.0) +# STREAMLINE_TEST_KAFKA_PORT Host Kafka port (default 9092) +# STREAMLINE_TEST_HTTP_PORT Host HTTP API port (default 9094) +# STREAMLINE_TEST_LOG_LEVEL Server log verbosity (default info) +# +# Example: +# STREAMLINE_TEST_IMAGE=ghcr.io/streamlinelabs/streamline:0.3.0 \ +# STREAMLINE_TEST_HTTP_PORT=19094 make integration-up services: streamline: - image: ghcr.io/streamlinelabs/streamline:0.2.0 + image: ${STREAMLINE_TEST_IMAGE:-ghcr.io/streamlinelabs/streamline:0.2.0} ports: - - "9092:9092" - - "9094:9094" + - "${STREAMLINE_TEST_KAFKA_PORT:-9092}:9092" + - "${STREAMLINE_TEST_HTTP_PORT:-9094}:9094" environment: - - STREAMLINE_LOG_LEVEL=info + - STREAMLINE_LOG_LEVEL=${STREAMLINE_TEST_LOG_LEVEL:-info} - STREAMLINE_PLAYGROUND=true healthcheck: test: ["CMD", "curl", "-f", "http://localhost:9094/health"] interval: 5s timeout: 5s retries: 10 - + start_period: 10s diff --git a/docs/ENVIRONMENT.md b/docs/ENVIRONMENT.md index bc1bbb6..d4dca89 100644 --- a/docs/ENVIRONMENT.md +++ b/docs/ENVIRONMENT.md @@ -10,3 +10,27 @@ The Streamline Operator supports the following environment variables: | `STREAMLINE_OPERATOR_LEADER_ELECTION` | Enable leader election | `true` | | `STREAMLINE_DEFAULT_IMAGE` | Default Streamline image | `streamlinelabs/streamline:latest` | | `STREAMLINE_RECONCILE_INTERVAL` | Reconciliation interval | `30s` | + +## Integration Test Variables + +These apply only to the opt-in integration suite (`tests/integration.rs` and +`docker-compose.test.yml`). They are never read by the operator at runtime, and +the default `cargo test` run ignores them entirely. + +| Variable | Description | Default | +|----------|-------------|---------| +| `STREAMLINE_TEST_IMAGE` | Streamline server image started by `docker-compose.test.yml` | `ghcr.io/streamlinelabs/streamline:0.2.0` | +| `STREAMLINE_TEST_HTTP_PORT` | Host port mapped to the server HTTP API | `9094` | +| `STREAMLINE_TEST_KAFKA_PORT` | Host port mapped to the server Kafka listener | `9092` | +| `STREAMLINE_TEST_HTTP_ENDPOINT` | Full HTTP endpoint override for the tests | `http://127.0.0.1:$STREAMLINE_TEST_HTTP_PORT` | +| `STREAMLINE_TEST_KAFKA_ENDPOINT` | Full `host:port` Kafka override for the tests | `127.0.0.1:$STREAMLINE_TEST_KAFKA_PORT` | +| `STREAMLINE_TEST_TIMEOUT_SECS` | Upper bound on every networked assertion | `15` | +| `STREAMLINE_TEST_LOG_LEVEL` | Log verbosity of the containerised server | `info` | + +Blank, zero, or unparseable values fall back to the defaults above. + +```bash +make integration-up +make test-integration +make integration-down +``` diff --git a/rustfmt.toml b/rustfmt.toml index 2e0fb18..f8c794b 100644 --- a/rustfmt.toml +++ b/rustfmt.toml @@ -3,4 +3,3 @@ max_width = 100 use_field_init_shorthand = true -imports_granularity = "Crate" diff --git a/src/conditions.rs b/src/conditions.rs index fbddb89..2f409ab 100644 --- a/src/conditions.rs +++ b/src/conditions.rs @@ -113,7 +113,12 @@ mod tests { #[test] fn test_build_condition() { - let cond = build_condition(CLUSTER_CONDITION_READY, CONDITION_TRUE, "AllReady", "All brokers ready"); + let cond = build_condition( + CLUSTER_CONDITION_READY, + CONDITION_TRUE, + "AllReady", + "All brokers ready", + ); assert_eq!(cond.condition_type, "Ready"); assert_eq!(cond.status, "True"); assert!(cond.last_transition_time.is_some()); diff --git a/src/controllers/autoscaling.rs b/src/controllers/autoscaling.rs index dee3100..969a7e9 100644 --- a/src/controllers/autoscaling.rs +++ b/src/controllers/autoscaling.rs @@ -571,8 +571,7 @@ impl AutoScalingController { if needed > recommended { recommended = needed; reasons.push(format!( - "High throughput: {:.0} msg/s per broker. Target: {}", - mps_per_broker, target_mps + "High throughput: {mps_per_broker:.0} msg/s per broker. Target: {target_mps}" )); triggering.push("messages_per_second".to_string()); } @@ -656,6 +655,8 @@ impl AutoScalingController { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; #[test] diff --git a/src/controllers/branch.rs b/src/controllers/branch.rs index 242bce8..a1539ef 100644 --- a/src/controllers/branch.rs +++ b/src/controllers/branch.rs @@ -2,12 +2,10 @@ //! Manages branch lifecycle via the admin API. use crate::conditions::{ - build_condition, set_condition, CONDITION_FALSE, CONDITION_TRUE, BRANCH_FINALIZER, + build_condition, set_condition, BRANCH_FINALIZER, CONDITION_FALSE, CONDITION_TRUE, }; use crate::controllers::error_policy_backoff; -use crate::crd::{ - BranchCondition, BranchPhase, BranchStatus, StreamlineBranch, StreamlineCluster, -}; +use crate::crd::{BranchCondition, BranchPhase, BranchStatus, StreamlineBranch, StreamlineCluster}; use crate::error::{OperatorError, Result}; use futures::StreamExt; use kube::api::{Api, Patch, PatchParams}; @@ -31,7 +29,10 @@ pub struct BranchController { impl BranchController { /// Create a new branch controller pub fn new(client: Client, http_client: reqwest::Client) -> Self { - Self { client, http_client } + Self { + client, + http_client, + } } /// Run the branch controller @@ -147,11 +148,7 @@ impl BranchController { } /// Ensure the finalizer is present on the resource - async fn ensure_finalizer( - &self, - branch: &StreamlineBranch, - namespace: &str, - ) -> Result<()> { + async fn ensure_finalizer(&self, branch: &StreamlineBranch, namespace: &str) -> Result<()> { let finalizers = branch.metadata.finalizers.as_deref().unwrap_or_default(); if finalizers.contains(&BRANCH_FINALIZER.to_string()) { return Ok(()); @@ -195,13 +192,10 @@ impl BranchController { "http://{}-0.{}-headless.{}.svc:{}", cluster_name, cluster_name, namespace, cluster.spec.http_port ); - info!( - "Deleting branch {} from cluster at {}", - name, http_endpoint - ); + info!("Deleting branch {} from cluster at {}", name, http_endpoint); if let Err(e) = self .http_client - .delete(format!("{}/api/v1/branches/{}", http_endpoint, name)) + .delete(format!("{http_endpoint}/api/v1/branches/{name}")) .send() .await { @@ -232,11 +226,7 @@ impl BranchController { } }); branches - .patch( - &name, - &PatchParams::default(), - &Patch::Merge(&patch), - ) + .patch(&name, &PatchParams::default(), &Patch::Merge(&patch)) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; @@ -267,7 +257,7 @@ impl BranchController { let response = self .http_client - .post(format!("{}/api/v1/branches", http_endpoint)) + .post(format!("{http_endpoint}/api/v1/branches")) .json(&body) .send() .await @@ -375,12 +365,7 @@ impl BranchController { let mut cond_fields = Vec::new(); set_condition( &mut cond_fields, - build_condition( - BRANCH_CONDITION_ACTIVE, - CONDITION_FALSE, - "Pending", - message, - ), + build_condition(BRANCH_CONDITION_ACTIVE, CONDITION_FALSE, "Pending", message), ); let conditions: Vec = cond_fields diff --git a/src/controllers/cluster.rs b/src/controllers/cluster.rs index a64d0da..6b23ff4 100644 --- a/src/controllers/cluster.rs +++ b/src/controllers/cluster.rs @@ -14,8 +14,7 @@ use crate::error::{OperatorError, Result}; use chrono::Utc; use futures::StreamExt; use k8s_openapi::api::apps::v1::{ - RollingUpdateStatefulSetStrategy, StatefulSet, StatefulSetSpec, - StatefulSetUpdateStrategy, + RollingUpdateStatefulSetStrategy, StatefulSet, StatefulSetSpec, StatefulSetUpdateStrategy, }; use k8s_openapi::api::core::v1::{ ConfigMap, Container, ContainerPort, EnvVar, PersistentVolumeClaim, PersistentVolumeClaimSpec, @@ -124,8 +123,9 @@ impl ClusterController { target_messages_per_second: autoscaling_spec.target_messages_per_second, ..Default::default() }; - let autoscaler = - crate::controllers::autoscaling::AutoScalingController::new(self.client.clone()); + let autoscaler = crate::controllers::autoscaling::AutoScalingController::new( + self.client.clone(), + ); autoscaler .reconcile_hpa(&cluster, &namespace, &autoscaling_config) .await?; @@ -141,11 +141,7 @@ impl ClusterController { } /// Ensure the finalizer is present on the resource - async fn ensure_finalizer( - &self, - cluster: &StreamlineCluster, - namespace: &str, - ) -> Result<()> { + async fn ensure_finalizer(&self, cluster: &StreamlineCluster, namespace: &str) -> Result<()> { let finalizers = cluster.metadata.finalizers.as_deref().unwrap_or_default(); if finalizers.contains(&CLUSTER_FINALIZER.to_string()) { return Ok(()); @@ -176,12 +172,15 @@ impl ClusterController { namespace: &str, ) -> std::result::Result { let name = cluster.name_any(); - info!("Handling deletion of StreamlineCluster {}/{}", namespace, name); + info!( + "Handling deletion of StreamlineCluster {}/{}", + namespace, name + ); // Clean up PVCs created by the StatefulSet let pvcs: Api = Api::namespaced(self.client.clone(), namespace); let pvc_list = pvcs - .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={}", name))) + .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={name}"))) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; @@ -210,15 +209,14 @@ impl ClusterController { } }); clusters - .patch( - &name, - &PatchParams::default(), - &Patch::Merge(&patch), - ) + .patch(&name, &PatchParams::default(), &Patch::Merge(&patch)) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; - info!("Finalizer removed for StreamlineCluster {}/{}", namespace, name); + info!( + "Finalizer removed for StreamlineCluster {}/{}", + namespace, name + ); Ok(Action::await_change()) } @@ -263,8 +261,7 @@ tls: mtls_enabled: {} insecure_skip_verify: {} "#, - tls.mtls_enabled, - tls.insecure_skip_verify + tls.mtls_enabled, tls.insecure_skip_verify )); } } @@ -638,7 +635,7 @@ tls: match_labels: Some(selector), ..Default::default() }, - service_name: format!("{}-headless", name), + service_name: format!("{name}-headless"), template: pod_template, volume_claim_templates: Some(volume_claim_templates), pod_management_policy: Some("Parallel".to_string()), @@ -678,7 +675,7 @@ tls: // Count ready pods let pod_list = pods - .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={}", name))) + .list(&ListParams::default().labels(&format!("app.kubernetes.io/instance={name}"))) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; @@ -721,35 +718,90 @@ tls: // Ready condition let (ready_status, ready_reason, ready_msg) = if ready_count == desired { - (CONDITION_TRUE, "AllBrokersReady", format!("{}/{} brokers ready", ready_count, desired)) + ( + CONDITION_TRUE, + "AllBrokersReady", + format!("{ready_count}/{desired} brokers ready"), + ) } else { - (CONDITION_FALSE, "BrokersNotReady", format!("{}/{} brokers ready", ready_count, desired)) + ( + CONDITION_FALSE, + "BrokersNotReady", + format!("{ready_count}/{desired} brokers ready"), + ) }; - set_condition(&mut cond_fields, build_condition(CLUSTER_CONDITION_READY, ready_status, ready_reason, &ready_msg)); + set_condition( + &mut cond_fields, + build_condition( + CLUSTER_CONDITION_READY, + ready_status, + ready_reason, + &ready_msg, + ), + ); // Available condition — at least one broker is ready let (avail_status, avail_reason, avail_msg) = if ready_count > 0 { - (CONDITION_TRUE, "MinimumAvailable", format!("{} broker(s) available", ready_count)) + ( + CONDITION_TRUE, + "MinimumAvailable", + format!("{ready_count} broker(s) available"), + ) } else { - (CONDITION_FALSE, "NoBrokersAvailable", "No brokers are available".to_string()) + ( + CONDITION_FALSE, + "NoBrokersAvailable", + "No brokers are available".to_string(), + ) }; - set_condition(&mut cond_fields, build_condition(CLUSTER_CONDITION_AVAILABLE, avail_status, avail_reason, &avail_msg)); + set_condition( + &mut cond_fields, + build_condition( + CLUSTER_CONDITION_AVAILABLE, + avail_status, + avail_reason, + &avail_msg, + ), + ); // Progressing condition — rolling out or scaling let (prog_status, prog_reason, prog_msg) = if ready_count < desired { - (CONDITION_TRUE, "ScalingUp", format!("Scaling from {} to {} replicas", ready_count, desired)) + ( + CONDITION_TRUE, + "ScalingUp", + format!("Scaling from {ready_count} to {desired} replicas"), + ) } else { - (CONDITION_FALSE, "UpToDate", "All replicas are up to date".to_string()) + ( + CONDITION_FALSE, + "UpToDate", + "All replicas are up to date".to_string(), + ) }; - set_condition(&mut cond_fields, build_condition(CLUSTER_CONDITION_PROGRESSING, prog_status, prog_reason, &prog_msg)); + set_condition( + &mut cond_fields, + build_condition( + CLUSTER_CONDITION_PROGRESSING, + prog_status, + prog_reason, + &prog_msg, + ), + ); // Degraded condition — some brokers are down let (deg_status, deg_reason, deg_msg) = if ready_count > 0 && ready_count < desired { - (CONDITION_TRUE, "PartiallyReady", format!("Only {}/{} brokers ready", ready_count, desired)) + ( + CONDITION_TRUE, + "PartiallyReady", + format!("Only {ready_count}/{desired} brokers ready"), + ) } else { (CONDITION_FALSE, "Healthy", "Cluster is healthy".to_string()) }; - set_condition(&mut cond_fields, build_condition(CLUSTER_CONDITION_DEGRADED, deg_status, deg_reason, °_msg)); + set_condition( + &mut cond_fields, + build_condition(CLUSTER_CONDITION_DEGRADED, deg_status, deg_reason, °_msg), + ); let conditions = cond_fields .into_iter() diff --git a/src/controllers/contract.rs b/src/controllers/contract.rs index 4a2fcb9..9171665 100644 --- a/src/controllers/contract.rs +++ b/src/controllers/contract.rs @@ -31,7 +31,10 @@ pub struct ContractController { impl ContractController { /// Create a new contract controller pub fn new(client: Client, http_client: reqwest::Client) -> Self { - Self { client, http_client } + Self { + client, + http_client, + } } /// Run the contract controller @@ -74,7 +77,9 @@ impl ContractController { crate::metrics::get().inc_reconcile("contract"); let _timer = crate::metrics::get().start_timer(); let name = contract.name_any(); - let namespace = contract.namespace().unwrap_or_else(|| "default".to_string()); + let namespace = contract + .namespace() + .unwrap_or_else(|| "default".to_string()); info!("Reconciling StreamlineContract {}/{}", namespace, name); @@ -160,11 +165,7 @@ impl ContractController { } /// Ensure the finalizer is present on the resource - async fn ensure_finalizer( - &self, - contract: &StreamlineContract, - namespace: &str, - ) -> Result<()> { + async fn ensure_finalizer(&self, contract: &StreamlineContract, namespace: &str) -> Result<()> { let finalizers = contract.metadata.finalizers.as_deref().unwrap_or_default(); if finalizers.contains(&CONTRACT_FINALIZER.to_string()) { return Ok(()); @@ -214,7 +215,7 @@ impl ContractController { ); if let Err(e) = self .http_client - .delete(format!("{}/api/v1/contracts/{}", http_endpoint, name)) + .delete(format!("{http_endpoint}/api/v1/contracts/{name}")) .send() .await { @@ -245,11 +246,7 @@ impl ContractController { } }); contracts - .patch( - &name, - &PatchParams::default(), - &Patch::Merge(&patch), - ) + .patch(&name, &PatchParams::default(), &Patch::Merge(&patch)) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; @@ -275,7 +272,7 @@ impl ContractController { let response = self .http_client - .post(format!("{}/api/v1/contracts/validate", http_endpoint)) + .post(format!("{http_endpoint}/api/v1/contracts/validate")) .json(&body) .send() .await @@ -291,8 +288,7 @@ impl ContractController { let status = response.status(); let body = response.text().await.unwrap_or_default(); return Err(OperatorError::Reconciliation(format!( - "Contract validation failed (HTTP {}): {}", - status, body + "Contract validation failed (HTTP {status}): {body}" ))); } @@ -314,7 +310,7 @@ impl ContractController { let response = self .http_client - .post(format!("{}/api/v1/contracts", http_endpoint)) + .post(format!("{http_endpoint}/api/v1/contracts")) .json(&body) .send() .await diff --git a/src/controllers/memory.rs b/src/controllers/memory.rs index addb1ed..8499f76 100644 --- a/src/controllers/memory.rs +++ b/src/controllers/memory.rs @@ -7,9 +7,7 @@ use crate::conditions::{ build_condition, set_condition, CONDITION_FALSE, CONDITION_TRUE, MEMORY_FINALIZER, }; use crate::controllers::error_policy_backoff; -use crate::crd::{ - MemoryCondition, MemoryPhase, MemoryStatus, StreamlineCluster, StreamlineMemory, -}; +use crate::crd::{MemoryCondition, MemoryPhase, MemoryStatus, StreamlineCluster, StreamlineMemory}; use crate::error::{OperatorError, Result}; use futures::StreamExt; use kube::api::{Api, Patch, PatchParams}; @@ -36,7 +34,10 @@ pub struct MemoryController { impl MemoryController { /// Create a new memory controller pub fn new(client: Client, http_client: reqwest::Client) -> Self { - Self { client, http_client } + Self { + client, + http_client, + } } /// Run the memory controller @@ -75,9 +76,7 @@ impl MemoryController { fn topic_name(memory: &StreamlineMemory, tier: &str) -> String { format!( "_memory.{}.{}.{}", - memory.spec.tenant, - memory.spec.agent_id, - tier + memory.spec.tenant, memory.spec.agent_id, tier ) } @@ -170,11 +169,7 @@ impl MemoryController { } /// Ensure the finalizer is present on the resource - async fn ensure_finalizer( - &self, - memory: &StreamlineMemory, - namespace: &str, - ) -> Result<()> { + async fn ensure_finalizer(&self, memory: &StreamlineMemory, namespace: &str) -> Result<()> { let finalizers = memory.metadata.finalizers.as_deref().unwrap_or_default(); if finalizers.contains(&MEMORY_FINALIZER.to_string()) { return Ok(()); @@ -224,11 +219,14 @@ impl MemoryController { info!("Deleting memory topic {} from cluster", topic_name); if let Err(e) = self .http_client - .delete(format!("{}/api/v1/topics/{}", http_endpoint, topic_name)) + .delete(format!("{http_endpoint}/api/v1/topics/{topic_name}")) .send() .await { - warn!("Failed to delete memory topic {} from cluster API: {}", topic_name, e); + warn!( + "Failed to delete memory topic {} from cluster API: {}", + topic_name, e + ); } } } else { @@ -256,11 +254,7 @@ impl MemoryController { } }); memories - .patch( - &name, - &PatchParams::default(), - &Patch::Merge(&patch), - ) + .patch(&name, &PatchParams::default(), &Patch::Merge(&patch)) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; @@ -278,9 +272,18 @@ impl MemoryController { http_endpoint: &str, ) -> Result<()> { let retentions = [ - ("episodic", Self::retention_ms(memory.spec.tiers.episodic_retention_days)), - ("semantic", Self::retention_ms(memory.spec.tiers.semantic_retention_days)), - ("procedural", Self::retention_ms(memory.spec.tiers.procedural_retention_days)), + ( + "episodic", + Self::retention_ms(memory.spec.tiers.episodic_retention_days), + ), + ( + "semantic", + Self::retention_ms(memory.spec.tiers.semantic_retention_days), + ), + ( + "procedural", + Self::retention_ms(memory.spec.tiers.procedural_retention_days), + ), ]; for (tier, retention_ms) in &retentions { @@ -303,14 +306,13 @@ impl MemoryController { let response = self .http_client - .post(format!("{}/api/v1/topics", http_endpoint)) + .post(format!("{http_endpoint}/api/v1/topics")) .json(&topic_config) .send() .await .map_err(|e| { OperatorError::Internal(format!( - "HTTP request to create memory topic {} failed: {}", - topic_name, e + "HTTP request to create memory topic {topic_name} failed: {e}" )) })?; @@ -320,8 +322,7 @@ impl MemoryController { // 409 Conflict means topic already exists — treat as success if status.as_u16() != 409 { return Err(OperatorError::Internal(format!( - "Failed to create memory topic {} (HTTP {}): {}", - topic_name, status, body + "Failed to create memory topic {topic_name} (HTTP {status}): {body}" ))); } } @@ -331,11 +332,7 @@ impl MemoryController { } /// Update status to ready - async fn update_status_ready( - &self, - memory: &StreamlineMemory, - namespace: &str, - ) -> Result<()> { + async fn update_status_ready(&self, memory: &StreamlineMemory, namespace: &str) -> Result<()> { let name = memory.name_any(); let memories: Api = Api::namespaced(self.client.clone(), namespace); @@ -402,12 +399,7 @@ impl MemoryController { let mut cond_fields = Vec::new(); set_condition( &mut cond_fields, - build_condition( - MEMORY_CONDITION_READY, - CONDITION_FALSE, - "Pending", - message, - ), + build_condition(MEMORY_CONDITION_READY, CONDITION_FALSE, "Pending", message), ); set_condition( &mut cond_fields, @@ -508,6 +500,8 @@ impl MemoryController { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; #[test] diff --git a/src/controllers/operator_hub.rs b/src/controllers/operator_hub.rs index 8c94a5a..475d94c 100644 --- a/src/controllers/operator_hub.rs +++ b/src/controllers/operator_hub.rs @@ -266,12 +266,12 @@ impl OperatorHub { Some(o) => o, None => { warn!(name, "Attempted to install unknown operator"); - return Err(format!("operator '{}' not found in hub", name)); + return Err(format!("operator '{name}' not found in hub")); } }; if self.config.require_verified && !op.verified { - return Err(format!("operator '{}' is not verified", name)); + return Err(format!("operator '{name}' is not verified")); } // Check capacity. @@ -284,7 +284,7 @@ impl OperatorHub { )); } if inst.contains_key(name) { - return Err(format!("operator '{}' is already installed", name)); + return Err(format!("operator '{name}' is already installed")); } } @@ -315,8 +315,11 @@ impl OperatorHub { Ok(()) } None => { - debug!(name, "Attempted to uninstall operator that is not installed"); - Err(format!("operator '{}' is not installed", name)) + debug!( + name, + "Attempted to uninstall operator that is not installed" + ); + Err(format!("operator '{name}' is not installed")) } } } @@ -327,7 +330,7 @@ impl OperatorHub { let ops = self.operators.read().await; match ops.get(name) { Some(o) => o.version.clone(), - None => return Err(format!("operator '{}' not found in hub", name)), + None => return Err(format!("operator '{name}' not found in hub")), } }; @@ -336,8 +339,7 @@ impl OperatorHub { Some(op) => { if op.version == latest_version { return Err(format!( - "operator '{}' is already at latest version {}", - name, latest_version + "operator '{name}' is already at latest version {latest_version}" )); } info!( @@ -351,7 +353,7 @@ impl OperatorHub { op.last_reconcile_at = Some(chrono::Utc::now().to_rfc3339()); Ok(op.clone()) } - None => Err(format!("operator '{}' is not installed", name)), + None => Err(format!("operator '{name}' is not installed")), } } @@ -380,7 +382,10 @@ impl OperatorHub { } if !updates.is_empty() { - info!(count = updates.len(), "Updates available for installed operators"); + info!( + count = updates.len(), + "Updates available for installed operators" + ); } updates @@ -417,6 +422,8 @@ impl OperatorHub { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; fn default_hub() -> OperatorHub { @@ -688,4 +695,3 @@ mod tests { assert_eq!(back, it); } } - diff --git a/src/controllers/scale_to_zero.rs b/src/controllers/scale_to_zero.rs index e56ad69..4fa29cb 100644 --- a/src/controllers/scale_to_zero.rs +++ b/src/controllers/scale_to_zero.rs @@ -14,7 +14,7 @@ use crate::crd::StreamlineCluster; use crate::error::{OperatorError, Result}; use k8s_openapi::api::apps::v1::StatefulSet; -use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference}; +use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference; use kube::api::{Api, Patch, PatchParams, PostParams}; use kube::{Client, Resource, ResourceExt}; use schemars::JsonSchema; @@ -363,7 +363,10 @@ impl ScaleToZeroController { match api.get(&name).await { Ok(_existing) => { - info!("Updating KEDA ScaledObject {} in namespace {}", name, namespace); + info!( + "Updating KEDA ScaledObject {} in namespace {}", + name, namespace + ); api.patch( &name, &PatchParams::apply("streamline-operator"), @@ -373,7 +376,10 @@ impl ScaleToZeroController { .map_err(|e| OperatorError::KubeApi(e.to_string()))?; } Err(_) => { - info!("Creating KEDA ScaledObject {} in namespace {}", name, namespace); + info!( + "Creating KEDA ScaledObject {} in namespace {}", + name, namespace + ); api.create(&PostParams::default(), &scaled_object) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; @@ -405,10 +411,16 @@ impl ScaleToZeroController { match api.delete(&name, &Default::default()).await { Ok(_) => { - info!("Deleted KEDA ScaledObject {} in namespace {}", name, namespace); + info!( + "Deleted KEDA ScaledObject {} in namespace {}", + name, namespace + ); } Err(kube::Error::Api(e)) if e.code == 404 => { - debug!("KEDA ScaledObject {} does not exist, nothing to delete", name); + debug!( + "KEDA ScaledObject {} does not exist, nothing to delete", + name + ); } Err(e) => { warn!("Failed to delete KEDA ScaledObject {}: {}", name, e); @@ -480,8 +492,8 @@ impl ScaleToZeroController { } }); - let obj: kube::api::DynamicObject = - serde_json::from_value(data).map_err(|e| OperatorError::Serialization(e.to_string()))?; + let obj: kube::api::DynamicObject = serde_json::from_value(data) + .map_err(|e| OperatorError::Serialization(e.to_string()))?; Ok(obj) } @@ -525,6 +537,9 @@ impl ScaleToZeroController { // --------------------------------------------------------------------------- /// Build a [`ClusterActivitySnapshot`] from raw metric values and the configured idle timeout. +// Argument count is part of the published API surface; grouping them into a struct +// would be a breaking change for downstream callers. +#[allow(clippy::too_many_arguments)] pub fn build_activity_snapshot( cluster_name: &str, namespace: &str, @@ -562,6 +577,8 @@ pub fn build_activity_snapshot( #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; fn default_config() -> ScaleToZeroConfig { @@ -717,16 +734,8 @@ mod tests { let now = chrono::Utc::now().timestamp(); let last_activity = now - 600; - let snap = build_activity_snapshot( - "my-cluster", - "prod", - 0.0, - 0.0, - 0, - 0, - last_activity, - 300, - ); + let snap = + build_activity_snapshot("my-cluster", "prod", 0.0, 0.0, 0, 0, last_activity, 300); assert!(snap.is_idle); assert!(snap.idle_duration_seconds >= 300); @@ -737,16 +746,8 @@ mod tests { fn test_build_activity_snapshot_active() { let now = chrono::Utc::now().timestamp(); - let snap = build_activity_snapshot( - "my-cluster", - "prod", - 500.0, - 100_000.0, - 5, - 200, - now, - 300, - ); + let snap = + build_activity_snapshot("my-cluster", "prod", 500.0, 100_000.0, 5, 200, now, 300); assert!(!snap.is_idle); assert_eq!(snap.active_connections, 5); @@ -836,4 +837,3 @@ mod tests { } } } - diff --git a/src/controllers/topic.rs b/src/controllers/topic.rs index 3e075db..f3c8360 100644 --- a/src/controllers/topic.rs +++ b/src/controllers/topic.rs @@ -29,7 +29,10 @@ pub struct TopicController { impl TopicController { /// Create a new topic controller pub fn new(client: Client, http_client: reqwest::Client) -> Self { - Self { client, http_client } + Self { + client, + http_client, + } } /// Run the topic controller @@ -137,11 +140,7 @@ impl TopicController { } /// Ensure the finalizer is present on the resource - async fn ensure_finalizer( - &self, - topic: &StreamlineTopic, - namespace: &str, - ) -> Result<()> { + async fn ensure_finalizer(&self, topic: &StreamlineTopic, namespace: &str) -> Result<()> { let finalizers = topic.metadata.finalizers.as_deref().unwrap_or_default(); if finalizers.contains(&TOPIC_FINALIZER.to_string()) { return Ok(()); @@ -172,7 +171,10 @@ impl TopicController { namespace: &str, ) -> std::result::Result { let name = topic.name_any(); - info!("Handling deletion of StreamlineTopic {}/{}", namespace, name); + info!( + "Handling deletion of StreamlineTopic {}/{}", + namespace, name + ); // Attempt to delete topic from the Streamline cluster let clusters: Api = Api::namespaced(self.client.clone(), namespace); @@ -182,12 +184,10 @@ impl TopicController { "http://{}-0.{}-headless.{}.svc:{}", cluster_name, cluster_name, namespace, cluster.spec.http_port ); - info!( - "Deleting topic {} from cluster at {}", - name, http_endpoint - ); - if let Err(e) = self.http_client - .delete(format!("{}/api/v1/topics/{}", http_endpoint, name)) + info!("Deleting topic {} from cluster at {}", name, http_endpoint); + if let Err(e) = self + .http_client + .delete(format!("{http_endpoint}/api/v1/topics/{name}")) .send() .await { @@ -218,15 +218,14 @@ impl TopicController { } }); topics - .patch( - &name, - &PatchParams::default(), - &Patch::Merge(&patch), - ) + .patch(&name, &PatchParams::default(), &Patch::Merge(&patch)) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; - info!("Finalizer removed for StreamlineTopic {}/{}", namespace, name); + info!( + "Finalizer removed for StreamlineTopic {}/{}", + namespace, name + ); Ok(Action::await_change()) } @@ -268,8 +267,9 @@ impl TopicController { http_endpoint, ); - let response = self.http_client - .post(format!("{}/api/v1/topics", http_endpoint)) + let response = self + .http_client + .post(format!("{http_endpoint}/api/v1/topics")) .json(&topic_config) .send() .await @@ -291,7 +291,7 @@ impl TopicController { topic.name_any(), status, body - )).into()); + ))); } } @@ -304,14 +304,29 @@ impl TopicController { let topics: Api = Api::namespaced(self.client.clone(), namespace); let mut cond_fields = Vec::new(); - set_condition(&mut cond_fields, build_condition( - TOPIC_CONDITION_READY, CONDITION_TRUE, "TopicReady", "Topic successfully created/updated", - )); - set_condition(&mut cond_fields, build_condition( - TOPIC_CONDITION_SYNCED, CONDITION_TRUE, "ConfigSynced", "Topic configuration is in sync with desired state", - )); + set_condition( + &mut cond_fields, + build_condition( + TOPIC_CONDITION_READY, + CONDITION_TRUE, + "TopicReady", + "Topic successfully created/updated", + ), + ); + set_condition( + &mut cond_fields, + build_condition( + TOPIC_CONDITION_SYNCED, + CONDITION_TRUE, + "ConfigSynced", + "Topic configuration is in sync with desired state", + ), + ); - let conditions = cond_fields.into_iter().map(|c| c.into_topic_condition()).collect(); + let conditions = cond_fields + .into_iter() + .map(|c| c.into_topic_condition()) + .collect(); let status = TopicStatus { ready: true, @@ -345,14 +360,24 @@ impl TopicController { let topics: Api = Api::namespaced(self.client.clone(), namespace); let mut cond_fields = Vec::new(); - set_condition(&mut cond_fields, build_condition( - TOPIC_CONDITION_READY, CONDITION_FALSE, "Pending", message, - )); - set_condition(&mut cond_fields, build_condition( - TOPIC_CONDITION_SYNCED, CONDITION_FALSE, "WaitingForCluster", "Topic cannot sync until cluster is ready", - )); + set_condition( + &mut cond_fields, + build_condition(TOPIC_CONDITION_READY, CONDITION_FALSE, "Pending", message), + ); + set_condition( + &mut cond_fields, + build_condition( + TOPIC_CONDITION_SYNCED, + CONDITION_FALSE, + "WaitingForCluster", + "Topic cannot sync until cluster is ready", + ), + ); - let conditions = cond_fields.into_iter().map(|c| c.into_topic_condition()).collect(); + let conditions = cond_fields + .into_iter() + .map(|c| c.into_topic_condition()) + .collect(); let status = TopicStatus { ready: false, @@ -386,14 +411,29 @@ impl TopicController { let topics: Api = Api::namespaced(self.client.clone(), namespace); let mut cond_fields = Vec::new(); - set_condition(&mut cond_fields, build_condition( - TOPIC_CONDITION_READY, CONDITION_FALSE, "Error", error_message, - )); - set_condition(&mut cond_fields, build_condition( - TOPIC_CONDITION_SYNCED, CONDITION_FALSE, "SyncFailed", error_message, - )); - - let conditions = cond_fields.into_iter().map(|c| c.into_topic_condition()).collect(); + set_condition( + &mut cond_fields, + build_condition( + TOPIC_CONDITION_READY, + CONDITION_FALSE, + "Error", + error_message, + ), + ); + set_condition( + &mut cond_fields, + build_condition( + TOPIC_CONDITION_SYNCED, + CONDITION_FALSE, + "SyncFailed", + error_message, + ), + ); + + let conditions = cond_fields + .into_iter() + .map(|c| c.into_topic_condition()) + .collect(); let status = TopicStatus { ready: false, diff --git a/src/controllers/user.rs b/src/controllers/user.rs index c249e79..ddf9c51 100644 --- a/src/controllers/user.rs +++ b/src/controllers/user.rs @@ -4,8 +4,8 @@ //! users and their credentials within Streamline clusters. use crate::conditions::{ - build_condition, set_condition, CONDITION_FALSE, CONDITION_TRUE, USER_CONDITION_CREDENTIALS_READY, - USER_CONDITION_READY, USER_FINALIZER, + build_condition, set_condition, CONDITION_FALSE, CONDITION_TRUE, + USER_CONDITION_CREDENTIALS_READY, USER_CONDITION_READY, USER_FINALIZER, }; use crate::controllers::error_policy_backoff; use crate::crd::{StreamlineCluster, StreamlineUser, UserPhase, UserStatus}; @@ -32,7 +32,10 @@ pub struct UserController { impl UserController { /// Create a new user controller pub fn new(client: Client, http_client: reqwest::Client) -> Self { - Self { client, http_client } + Self { + client, + http_client, + } } /// Run the user controller @@ -147,11 +150,7 @@ impl UserController { } /// Ensure the finalizer is present on the resource - async fn ensure_finalizer( - &self, - user: &StreamlineUser, - namespace: &str, - ) -> Result<()> { + async fn ensure_finalizer(&self, user: &StreamlineUser, namespace: &str) -> Result<()> { let finalizers = user.metadata.finalizers.as_deref().unwrap_or_default(); if finalizers.contains(&USER_FINALIZER.to_string()) { return Ok(()); @@ -196,8 +195,9 @@ impl UserController { "Revoking credentials for user {} from cluster at {}", name, http_endpoint ); - if let Err(e) = self.http_client - .delete(format!("{}/api/v1/users/{}", http_endpoint, name)) + if let Err(e) = self + .http_client + .delete(format!("{http_endpoint}/api/v1/users/{name}")) .send() .await { @@ -228,15 +228,14 @@ impl UserController { } }); users - .patch( - &name, - &PatchParams::default(), - &Patch::Merge(&patch), - ) + .patch(&name, &PatchParams::default(), &Patch::Merge(&patch)) .await .map_err(|e| OperatorError::KubeApi(e.to_string()))?; - info!("Finalizer removed for StreamlineUser {}/{}", namespace, name); + info!( + "Finalizer removed for StreamlineUser {}/{}", + namespace, name + ); Ok(Action::await_change()) } @@ -279,7 +278,7 @@ impl UserController { })?; String::from_utf8(password_bytes.0.clone()).map_err(|e| { - OperatorError::Configuration(format!("Invalid password encoding: {}", e)) + OperatorError::Configuration(format!("Invalid password encoding: {e}")) })? } else { // Generate random password @@ -391,8 +390,9 @@ impl UserController { http_endpoint, ); - let response = self.http_client - .post(format!("{}/api/v1/users", http_endpoint)) + let response = self + .http_client + .post(format!("{http_endpoint}/api/v1/users")) .json(&user_config) .send() .await @@ -413,7 +413,7 @@ impl UserController { user.name_any(), status, body - )).into()); + ))); } } @@ -431,15 +431,29 @@ impl UserController { let users: Api = Api::namespaced(self.client.clone(), namespace); let mut cond_fields = Vec::new(); - set_condition(&mut cond_fields, build_condition( - USER_CONDITION_READY, CONDITION_TRUE, "UserReady", "User successfully created/updated", - )); - set_condition(&mut cond_fields, build_condition( - USER_CONDITION_CREDENTIALS_READY, CONDITION_TRUE, "CredentialsProvisioned", - &format!("Credentials stored in secret {}", credentials_secret), - )); + set_condition( + &mut cond_fields, + build_condition( + USER_CONDITION_READY, + CONDITION_TRUE, + "UserReady", + "User successfully created/updated", + ), + ); + set_condition( + &mut cond_fields, + build_condition( + USER_CONDITION_CREDENTIALS_READY, + CONDITION_TRUE, + "CredentialsProvisioned", + &format!("Credentials stored in secret {credentials_secret}"), + ), + ); - let conditions = cond_fields.into_iter().map(|c| c.into_user_condition()).collect(); + let conditions = cond_fields + .into_iter() + .map(|c| c.into_user_condition()) + .collect(); let status = UserStatus { ready: true, @@ -472,15 +486,24 @@ impl UserController { let users: Api = Api::namespaced(self.client.clone(), namespace); let mut cond_fields = Vec::new(); - set_condition(&mut cond_fields, build_condition( - USER_CONDITION_READY, CONDITION_FALSE, "Pending", message, - )); - set_condition(&mut cond_fields, build_condition( - USER_CONDITION_CREDENTIALS_READY, CONDITION_FALSE, "WaitingForCluster", - "Credentials cannot be provisioned until cluster is ready", - )); + set_condition( + &mut cond_fields, + build_condition(USER_CONDITION_READY, CONDITION_FALSE, "Pending", message), + ); + set_condition( + &mut cond_fields, + build_condition( + USER_CONDITION_CREDENTIALS_READY, + CONDITION_FALSE, + "WaitingForCluster", + "Credentials cannot be provisioned until cluster is ready", + ), + ); - let conditions = cond_fields.into_iter().map(|c| c.into_user_condition()).collect(); + let conditions = cond_fields + .into_iter() + .map(|c| c.into_user_condition()) + .collect(); let status = UserStatus { ready: false, @@ -513,14 +536,29 @@ impl UserController { let users: Api = Api::namespaced(self.client.clone(), namespace); let mut cond_fields = Vec::new(); - set_condition(&mut cond_fields, build_condition( - USER_CONDITION_READY, CONDITION_FALSE, "Error", error_message, - )); - set_condition(&mut cond_fields, build_condition( - USER_CONDITION_CREDENTIALS_READY, CONDITION_FALSE, "ProvisioningFailed", error_message, - )); + set_condition( + &mut cond_fields, + build_condition( + USER_CONDITION_READY, + CONDITION_FALSE, + "Error", + error_message, + ), + ); + set_condition( + &mut cond_fields, + build_condition( + USER_CONDITION_CREDENTIALS_READY, + CONDITION_FALSE, + "ProvisioningFailed", + error_message, + ), + ); - let conditions = cond_fields.into_iter().map(|c| c.into_user_condition()).collect(); + let conditions = cond_fields + .into_iter() + .map(|c| c.into_user_condition()) + .collect(); let status = UserStatus { ready: false, @@ -550,4 +588,3 @@ mod tests { // Controller tests require k8s cluster } } - diff --git a/src/crd/backup.rs b/src/crd/backup.rs index a220b73..882db6e 100644 --- a/src/crd/backup.rs +++ b/src/crd/backup.rs @@ -74,9 +74,10 @@ fn default_compression() -> String { } /// Type of backup operation -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq, Default)] pub enum BackupType { /// Full backup of all data and metadata + #[default] Full, /// Incremental backup since last full/incremental backup Incremental, @@ -84,12 +85,6 @@ pub enum BackupType { MetadataOnly, } -impl Default for BackupType { - fn default() -> Self { - Self::Full - } -} - /// Backup storage destination #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "camelCase")] @@ -240,6 +235,17 @@ mod tests { assert!(spec.include_offsets); } + #[test] + fn test_backup_type_default_is_full() { + // Regression: `Default` is now derived — the default variant and its + // serde representation must stay unchanged. + assert_eq!(BackupType::default(), BackupType::Full); + assert_eq!( + serde_json::to_string(&BackupType::default()).unwrap_or_default(), + "\"Full\"" + ); + } + #[test] fn test_backup_phase_default() { let status = BackupStatus::default(); diff --git a/src/crd/branch.rs b/src/crd/branch.rs index c79fbcb..c60cca7 100644 --- a/src/crd/branch.rs +++ b/src/crd/branch.rs @@ -106,6 +106,8 @@ pub struct BranchCondition { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; use kube::CustomResourceExt; diff --git a/src/crd/cluster.rs b/src/crd/cluster.rs index 0dbe71f..62ce352 100644 --- a/src/crd/cluster.rs +++ b/src/crd/cluster.rs @@ -498,10 +498,7 @@ impl ClusterSpec { let mut errors = Vec::new(); if self.replicas < 1 { - errors.push(format!( - "replicas must be >= 1, got {}", - self.replicas - )); + errors.push(format!("replicas must be >= 1, got {}", self.replicas)); } if self.replicas > 1 && self.replicas % 2 == 0 { // Even replicas break Raft quorum — warn but allow @@ -518,16 +515,10 @@ impl ClusterSpec { )); } if self.http_port < 1 || self.http_port > 65535 { - errors.push(format!( - "httpPort must be 1-65535, got {}", - self.http_port - )); + errors.push(format!("httpPort must be 1-65535, got {}", self.http_port)); } if self.raft_port < 1 || self.raft_port > 65535 { - errors.push(format!( - "raftPort must be 1-65535, got {}", - self.raft_port - )); + errors.push(format!("raftPort must be 1-65535, got {}", self.raft_port)); } if self.kafka_port == self.http_port @@ -572,6 +563,8 @@ impl ClusterSpec { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; #[test] @@ -591,8 +584,7 @@ mod tests { #[test] fn test_cluster_spec_validate_negative_replicas() { - let spec: ClusterSpec = - serde_json::from_str(r#"{"replicas": -1}"#).unwrap(); + let spec: ClusterSpec = serde_json::from_str(r#"{"replicas": -1}"#).unwrap(); let result = spec.validate(); assert!(result.is_err()); assert!(result.unwrap_err()[0].contains("replicas must be >= 1")); @@ -609,8 +601,7 @@ mod tests { #[test] fn test_cluster_spec_validate_invalid_port() { - let spec: ClusterSpec = - serde_json::from_str(r#"{"kafkaPort": 99999}"#).unwrap(); + let spec: ClusterSpec = serde_json::from_str(r#"{"kafkaPort": 99999}"#).unwrap(); let result = spec.validate(); assert!(result.is_err()); } @@ -628,4 +619,3 @@ mod tests { assert_eq!(storage.access_modes, vec!["ReadWriteOnce"]); } } - diff --git a/src/crd/contract.rs b/src/crd/contract.rs index ffd5600..29899de 100644 --- a/src/crd/contract.rs +++ b/src/crd/contract.rs @@ -44,21 +44,16 @@ pub struct ContractSpec { } /// Schema compatibility policies recognised by the Moonshot control plane. -#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq, Default)] #[serde(rename_all = "UPPERCASE")] pub enum ContractCompatibility { + #[default] Backward, Forward, Full, None, } -impl Default for ContractCompatibility { - fn default() -> Self { - Self::Backward - } -} - fn default_compatibility() -> ContractCompatibility { ContractCompatibility::Backward } @@ -114,6 +109,8 @@ pub struct ContractCondition { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; use kube::CustomResourceExt; @@ -148,6 +145,21 @@ mod tests { assert_eq!(ContractPhase::default(), ContractPhase::Pending); } + #[test] + fn test_contract_compatibility_default_is_backward() { + // Regression: `Default` is now derived — the default variant and its + // UPPERCASE serde representation must stay unchanged. + assert_eq!( + ContractCompatibility::default(), + ContractCompatibility::Backward + ); + assert_eq!( + serde_json::to_string(&ContractCompatibility::default()).unwrap(), + "\"BACKWARD\"" + ); + assert_eq!(default_compatibility(), ContractCompatibility::default()); + } + #[test] fn test_contract_crd_renders() { let crd = StreamlineContract::crd(); diff --git a/src/crd/memory.rs b/src/crd/memory.rs index d3e3187..c4e9e35 100644 --- a/src/crd/memory.rs +++ b/src/crd/memory.rs @@ -174,6 +174,8 @@ pub struct MemoryCondition { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; use kube::CustomResourceExt; diff --git a/src/crd/mod.rs b/src/crd/mod.rs index f8304d6..32263e5 100644 --- a/src/crd/mod.rs +++ b/src/crd/mod.rs @@ -33,9 +33,7 @@ pub use contract::{ ContractCompatibility, ContractCondition, ContractPhase, ContractSpec, ContractStatus, StreamlineContract, }; -pub use edge::{ - BootstrapConfig, EdgeCondition, EdgeSpec, EdgeStatus, StreamlineEdge, SyncConfig, -}; +pub use edge::{BootstrapConfig, EdgeCondition, EdgeSpec, EdgeStatus, StreamlineEdge, SyncConfig}; pub use memory::{ MemoryCondition, MemoryDecay, MemoryPhase, MemorySpec, MemoryStatus, MemoryTiers, StreamlineMemory, diff --git a/src/crd/topic.rs b/src/crd/topic.rs index 12876ef..28916a0 100644 --- a/src/crd/topic.rs +++ b/src/crd/topic.rs @@ -11,7 +11,6 @@ use serde::{Deserialize, Serialize}; #[kube( group = "streamline.io", version = "v1alpha1", - kind = "StreamlineTopic", namespaced, status = "TopicStatus", @@ -241,6 +240,8 @@ fn default_compression_type() -> String { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; #[test] @@ -266,4 +267,3 @@ mod tests { assert_eq!(phase, TopicPhase::Pending); } } - diff --git a/src/crd/user.rs b/src/crd/user.rs index 7535ccf..b3eff36 100644 --- a/src/crd/user.rs +++ b/src/crd/user.rs @@ -293,6 +293,8 @@ fn default_password_key() -> String { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; #[test] diff --git a/src/error.rs b/src/error.rs index 5544684..03b5594 100644 --- a/src/error.rs +++ b/src/error.rs @@ -29,14 +29,14 @@ pub enum OperatorError { impl fmt::Display for OperatorError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - OperatorError::KubeApi(msg) => write!(f, "Kubernetes API error: {}", msg), - OperatorError::Configuration(msg) => write!(f, "Configuration error: {}", msg), - OperatorError::Reconciliation(msg) => write!(f, "Reconciliation error: {}", msg), - OperatorError::Http(msg) => write!(f, "HTTP error: {}", msg), - OperatorError::Serialization(msg) => write!(f, "Serialization error: {}", msg), - OperatorError::NotFound(msg) => write!(f, "Resource not found: {}", msg), - OperatorError::InvalidState(msg) => write!(f, "Invalid state: {}", msg), - OperatorError::Internal(msg) => write!(f, "Internal error: {}", msg), + OperatorError::KubeApi(msg) => write!(f, "Kubernetes API error: {msg}"), + OperatorError::Configuration(msg) => write!(f, "Configuration error: {msg}"), + OperatorError::Reconciliation(msg) => write!(f, "Reconciliation error: {msg}"), + OperatorError::Http(msg) => write!(f, "HTTP error: {msg}"), + OperatorError::Serialization(msg) => write!(f, "Serialization error: {msg}"), + OperatorError::NotFound(msg) => write!(f, "Resource not found: {msg}"), + OperatorError::InvalidState(msg) => write!(f, "Invalid state: {msg}"), + OperatorError::Internal(msg) => write!(f, "Internal error: {msg}"), } } } @@ -80,8 +80,7 @@ mod tests { for err in errors { // Ensure Display is implemented - let _ = format!("{}", err); + let _ = format!("{err}"); } } } - diff --git a/src/leader_election.rs b/src/leader_election.rs index 369890d..aab958f 100644 --- a/src/leader_election.rs +++ b/src/leader_election.rs @@ -48,7 +48,10 @@ impl LeaderElector { .or_else(|_| std::env::var("HOSTNAME")) .unwrap_or_else(|_| format!("operator-{:08x}", rand::random::())); info!(identity = %identity, namespace = %namespace, "Initialized leader elector"); - Self { lease_api, identity } + Self { + lease_api, + identity, + } } /// Blocks until the lease is successfully acquired. @@ -61,11 +64,17 @@ impl LeaderElector { return Ok(()); } Ok(false) => { - debug!("Lease held by another instance, retrying in {:?}", RETRY_INTERVAL); + debug!( + "Lease held by another instance, retrying in {:?}", + RETRY_INTERVAL + ); tokio::time::sleep(RETRY_INTERVAL).await; } Err(e) => { - warn!("Lease acquisition error: {}, retrying in {:?}", e, RETRY_INTERVAL); + warn!( + "Lease acquisition error: {}, retrying in {:?}", + e, RETRY_INTERVAL + ); tokio::time::sleep(RETRY_INTERVAL).await; } } @@ -174,7 +183,9 @@ impl LeaderElector { .unwrap_or(LEASE_DURATION_SECS) as i64; match renew_time { - Some(MicroTime(t)) => Utc::now().signed_duration_since(*t).num_seconds() > duration_secs, + Some(MicroTime(t)) => { + Utc::now().signed_duration_since(*t).num_seconds() > duration_secs + } None => true, } } @@ -193,11 +204,7 @@ impl LeaderElector { lease_transitions: Some(0), }), }; - match self - .lease_api - .create(&PostParams::default(), &lease) - .await - { + match self.lease_api.create(&PostParams::default(), &lease).await { Ok(_) => Ok(true), Err(kube::Error::Api(ae)) if ae.code == 409 => Ok(false), Err(e) => Err(e.into()), @@ -223,7 +230,11 @@ impl LeaderElector { prev.and_then(|s| s.acquire_time.clone()) }, renew_time: Some(now.clone()), - lease_transitions: Some(if takeover { transitions + 1 } else { transitions }), + lease_transitions: Some(if takeover { + transitions + 1 + } else { + transitions + }), }); match self diff --git a/src/lib.rs b/src/lib.rs index f06ecc4..586c6d1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,13 +33,12 @@ pub use controllers::{ UserController, }; pub use crd::{ - BackupCondition, BackupPhase, BackupSpec, BackupStatus, BackupStorage, BackupType, - BranchPhase, BranchSpec, BranchStatus, ClusterCondition, ClusterPhase, ClusterSpec, - ClusterStatus, ClusterStorage, ClusterTls, ContractCompatibility, ContractPhase, - ContractSpec, ContractStatus, MemoryCondition, MemoryDecay, MemoryPhase, MemorySpec, - MemoryStatus, MemoryTiers, ResourceRequirements, StreamlineBackup, StreamlineBranch, - StreamlineCluster, StreamlineContract, StreamlineMemory, StreamlineTopic, StreamlineUser, - TopicSpec, TopicStatus, UserCredentials, UserPermission, UserSpec, UserStatus, + BackupCondition, BackupPhase, BackupSpec, BackupStatus, BackupStorage, BackupType, BranchPhase, + BranchSpec, BranchStatus, ClusterCondition, ClusterPhase, ClusterSpec, ClusterStatus, + ClusterStorage, ClusterTls, ContractCompatibility, ContractPhase, ContractSpec, ContractStatus, + MemoryCondition, MemoryDecay, MemoryPhase, MemorySpec, MemoryStatus, MemoryTiers, + ResourceRequirements, StreamlineBackup, StreamlineBranch, StreamlineCluster, + StreamlineContract, StreamlineMemory, StreamlineTopic, StreamlineUser, TopicSpec, TopicStatus, + UserCredentials, UserPermission, UserSpec, UserStatus, }; pub use error::{OperatorError, Result}; - diff --git a/src/main.rs b/src/main.rs index a63e4b1..e694f2c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,6 +12,7 @@ //! RUST_LOG=debug streamline-operator //! ``` +use anyhow::Context; use clap::Parser; use kube::Client; use std::sync::Arc; @@ -90,13 +91,14 @@ async fn main() -> anyhow::Result<()> { .timeout(std::time::Duration::from_secs(30)) .pool_max_idle_per_host(4) .build() - .expect("Failed to create HTTP client"); + .context("Failed to create HTTP client")?; // Create controllers let cluster_controller = Arc::new(ClusterController::new(client.clone())); let topic_controller = Arc::new(TopicController::new(client.clone(), http_client.clone())); let user_controller = Arc::new(UserController::new(client.clone(), http_client.clone())); - let contract_controller = Arc::new(ContractController::new(client.clone(), http_client.clone())); + let contract_controller = + Arc::new(ContractController::new(client.clone(), http_client.clone())); let branch_controller = Arc::new(BranchController::new(client.clone(), http_client.clone())); let memory_controller = Arc::new(MemoryController::new(client.clone(), http_client)); @@ -187,7 +189,10 @@ async fn main() -> anyhow::Result<()> { let listener = match tokio::net::TcpListener::bind(&health_addr).await { Ok(l) => l, Err(e) => { - error!("Failed to bind health probe server on {}: {}", health_addr, e); + error!( + "Failed to bind health probe server on {}: {}", + health_addr, e + ); return; } }; @@ -201,10 +206,10 @@ async fn main() -> anyhow::Result<()> { let metrics_addr = args.metrics_bind_address.clone(); let metrics_handle = tokio::spawn(async move { use axum::{routing::get, Router}; - let app = Router::new() - .route("/metrics", get(|| async { - streamline_operator::metrics::get().render() - })); + let app = Router::new().route( + "/metrics", + get(|| async { streamline_operator::metrics::get().render() }), + ); let listener = match tokio::net::TcpListener::bind(&metrics_addr).await { Ok(l) => l, Err(e) => { @@ -258,6 +263,10 @@ async fn main() -> anyhow::Result<()> { } } + // Stop the auxiliary HTTP servers so they release their listening sockets + health_handle.abort(); + metrics_handle.abort(); + // Release the lease before exiting so a standby replica can take over immediately if let Some(e) = &elector { e.release().await; @@ -266,4 +275,3 @@ async fn main() -> anyhow::Result<()> { info!("Streamline Operator shutting down"); Ok(()) } - diff --git a/src/metrics.rs b/src/metrics.rs index a34f91f..16e1162 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -48,9 +48,15 @@ impl OperatorMetrics { pub fn inc_reconcile(&self, resource: &str) { self.reconcile_total.fetch_add(1, Ordering::Relaxed); match resource { - "cluster" => { self.reconcile_cluster_total.fetch_add(1, Ordering::Relaxed); } - "topic" => { self.reconcile_topic_total.fetch_add(1, Ordering::Relaxed); } - "user" => { self.reconcile_user_total.fetch_add(1, Ordering::Relaxed); } + "cluster" => { + self.reconcile_cluster_total.fetch_add(1, Ordering::Relaxed); + } + "topic" => { + self.reconcile_topic_total.fetch_add(1, Ordering::Relaxed); + } + "user" => { + self.reconcile_user_total.fetch_add(1, Ordering::Relaxed); + } _ => {} } } @@ -62,9 +68,16 @@ impl OperatorMetrics { pub fn inc_error(&self, resource: &str) { self.reconcile_errors.fetch_add(1, Ordering::Relaxed); match resource { - "cluster" => { self.reconcile_cluster_errors.fetch_add(1, Ordering::Relaxed); } - "topic" => { self.reconcile_topic_errors.fetch_add(1, Ordering::Relaxed); } - "user" => { self.reconcile_user_errors.fetch_add(1, Ordering::Relaxed); } + "cluster" => { + self.reconcile_cluster_errors + .fetch_add(1, Ordering::Relaxed); + } + "topic" => { + self.reconcile_topic_errors.fetch_add(1, Ordering::Relaxed); + } + "user" => { + self.reconcile_user_errors.fetch_add(1, Ordering::Relaxed); + } _ => {} } } @@ -82,7 +95,10 @@ impl OperatorMetrics { /// Start a timer that records duration on drop. pub fn start_timer(&self) -> ReconcileTimer<'_> { - ReconcileTimer { metrics: self, start: Instant::now() } + ReconcileTimer { + metrics: self, + start: Instant::now(), + } } /// Render metrics in Prometheus text exposition format. @@ -91,31 +107,85 @@ impl OperatorMetrics { out.push_str("# HELP streamline_operator_reconcile_total Total reconciliation attempts\n"); out.push_str("# TYPE streamline_operator_reconcile_total counter\n"); - push_counter(&mut out, "streamline_operator_reconcile_total", &[], self.reconcile_total.load(Ordering::Relaxed)); - - out.push_str("# HELP streamline_operator_reconcile_success_total Successful reconciliations\n"); + push_counter( + &mut out, + "streamline_operator_reconcile_total", + &[], + self.reconcile_total.load(Ordering::Relaxed), + ); + + out.push_str( + "# HELP streamline_operator_reconcile_success_total Successful reconciliations\n", + ); out.push_str("# TYPE streamline_operator_reconcile_success_total counter\n"); - push_counter(&mut out, "streamline_operator_reconcile_success_total", &[], self.reconcile_success.load(Ordering::Relaxed)); + push_counter( + &mut out, + "streamline_operator_reconcile_success_total", + &[], + self.reconcile_success.load(Ordering::Relaxed), + ); out.push_str("# HELP streamline_operator_reconcile_errors_total Failed reconciliations\n"); out.push_str("# TYPE streamline_operator_reconcile_errors_total counter\n"); - push_counter(&mut out, "streamline_operator_reconcile_errors_total", &[], self.reconcile_errors.load(Ordering::Relaxed)); + push_counter( + &mut out, + "streamline_operator_reconcile_errors_total", + &[], + self.reconcile_errors.load(Ordering::Relaxed), + ); out.push_str("# HELP streamline_operator_reconcile_by_resource_total Reconciliations per resource type\n"); out.push_str("# TYPE streamline_operator_reconcile_by_resource_total counter\n"); - push_counter(&mut out, "streamline_operator_reconcile_by_resource_total", &[("resource", "cluster")], self.reconcile_cluster_total.load(Ordering::Relaxed)); - push_counter(&mut out, "streamline_operator_reconcile_by_resource_total", &[("resource", "topic")], self.reconcile_topic_total.load(Ordering::Relaxed)); - push_counter(&mut out, "streamline_operator_reconcile_by_resource_total", &[("resource", "user")], self.reconcile_user_total.load(Ordering::Relaxed)); + push_counter( + &mut out, + "streamline_operator_reconcile_by_resource_total", + &[("resource", "cluster")], + self.reconcile_cluster_total.load(Ordering::Relaxed), + ); + push_counter( + &mut out, + "streamline_operator_reconcile_by_resource_total", + &[("resource", "topic")], + self.reconcile_topic_total.load(Ordering::Relaxed), + ); + push_counter( + &mut out, + "streamline_operator_reconcile_by_resource_total", + &[("resource", "user")], + self.reconcile_user_total.load(Ordering::Relaxed), + ); out.push_str("# HELP streamline_operator_reconcile_errors_by_resource_total Errors per resource type\n"); out.push_str("# TYPE streamline_operator_reconcile_errors_by_resource_total counter\n"); - push_counter(&mut out, "streamline_operator_reconcile_errors_by_resource_total", &[("resource", "cluster")], self.reconcile_cluster_errors.load(Ordering::Relaxed)); - push_counter(&mut out, "streamline_operator_reconcile_errors_by_resource_total", &[("resource", "topic")], self.reconcile_topic_errors.load(Ordering::Relaxed)); - push_counter(&mut out, "streamline_operator_reconcile_errors_by_resource_total", &[("resource", "user")], self.reconcile_user_errors.load(Ordering::Relaxed)); - - out.push_str("# HELP streamline_operator_leader_transitions_total Leader election transitions\n"); + push_counter( + &mut out, + "streamline_operator_reconcile_errors_by_resource_total", + &[("resource", "cluster")], + self.reconcile_cluster_errors.load(Ordering::Relaxed), + ); + push_counter( + &mut out, + "streamline_operator_reconcile_errors_by_resource_total", + &[("resource", "topic")], + self.reconcile_topic_errors.load(Ordering::Relaxed), + ); + push_counter( + &mut out, + "streamline_operator_reconcile_errors_by_resource_total", + &[("resource", "user")], + self.reconcile_user_errors.load(Ordering::Relaxed), + ); + + out.push_str( + "# HELP streamline_operator_leader_transitions_total Leader election transitions\n", + ); out.push_str("# TYPE streamline_operator_leader_transitions_total counter\n"); - push_counter(&mut out, "streamline_operator_leader_transitions_total", &[], self.leader_transitions.load(Ordering::Relaxed)); + push_counter( + &mut out, + "streamline_operator_leader_transitions_total", + &[], + self.leader_transitions.load(Ordering::Relaxed), + ); // Duration histogram if let Ok(h) = self.duration_buckets.lock() { @@ -125,14 +195,12 @@ impl OperatorMetrics { for (i, &boundary) in DURATION_BUCKETS_MS.iter().enumerate() { cumulative += h.buckets[i]; out.push_str(&format!( - "streamline_operator_reconcile_duration_ms_bucket{{le=\"{}\"}} {}\n", - boundary, cumulative + "streamline_operator_reconcile_duration_ms_bucket{{le=\"{boundary}\"}} {cumulative}\n" )); } cumulative += h.buckets[DURATION_BUCKETS_MS.len()]; out.push_str(&format!( - "streamline_operator_reconcile_duration_ms_bucket{{le=\"+Inf\"}} {}\n", - cumulative + "streamline_operator_reconcile_duration_ms_bucket{{le=\"+Inf\"}} {cumulative}\n" )); out.push_str(&format!( "streamline_operator_reconcile_duration_ms_sum {}\n", @@ -189,7 +257,9 @@ impl DurationHistogram { } } // Overflow bucket (+Inf) - *self.buckets.last_mut().expect("buckets non-empty") += 1; + if let Some(overflow) = self.buckets.last_mut() { + *overflow += 1; + } } } @@ -198,7 +268,9 @@ fn push_counter(out: &mut String, name: &str, labels: &[(&str, &str)], value: u6 if !labels.is_empty() { out.push('{'); for (i, (k, v)) in labels.iter().enumerate() { - if i > 0 { out.push(','); } + if i > 0 { + out.push(','); + } out.push_str(k); out.push_str("=\""); out.push_str(v); @@ -218,6 +290,8 @@ pub fn get() -> &'static OperatorMetrics { #[cfg(test)] mod tests { + // unwrap/expect are acceptable in tests; the crate-wide lint targets production code. + #![allow(clippy::unwrap_used, clippy::expect_used)] use super::*; #[test] @@ -272,9 +346,9 @@ mod tests { #[test] fn test_duration_histogram() { let m = OperatorMetrics::new(); - m.observe_duration_ms(3); // bucket: ≤5 - m.observe_duration_ms(50); // bucket: ≤50 - m.observe_duration_ms(500); // bucket: ≤500 + m.observe_duration_ms(3); // bucket: ≤5 + m.observe_duration_ms(50); // bucket: ≤50 + m.observe_duration_ms(500); // bucket: ≤500 m.observe_duration_ms(20000); // bucket: +Inf let output = m.render(); assert!(output.contains("reconcile_duration_ms_bucket")); @@ -286,8 +360,8 @@ mod tests { #[test] fn test_duration_histogram_buckets_cumulative() { let h = &mut DurationHistogram::new(); - h.observe(1); // ≤5 - h.observe(7); // ≤10 + h.observe(1); // ≤5 + h.observe(7); // ≤10 h.observe(100); // ≤100 assert_eq!(h.count, 3); assert_eq!(h.sum, 108); @@ -295,6 +369,35 @@ mod tests { assert_eq!(h.buckets[1], 1); // ≤10 } + #[test] + fn test_duration_histogram_overflow_bucket_is_last_slot() { + // Regression: the +Inf overflow slot used to be reached via `.expect()`. + // Values above the last boundary must land in the final slot, not panic. + let h = &mut DurationHistogram::new(); + let overflow_index = DURATION_BUCKETS_MS.len(); + h.observe(DURATION_BUCKETS_MS[overflow_index - 1] + 1); + h.observe(u64::MAX / 2); + + assert_eq!(h.count, 2); + assert_eq!(h.buckets[overflow_index], 2); + assert!( + h.buckets[..overflow_index].iter().all(|&b| b == 0), + "overflow observations must not land in a bounded bucket" + ); + } + + #[test] + fn test_duration_histogram_boundary_is_inclusive() { + // Regression: an observation exactly on the last boundary belongs to that + // bucket, not to +Inf. + let h = &mut DurationHistogram::new(); + let last_index = DURATION_BUCKETS_MS.len() - 1; + h.observe(DURATION_BUCKETS_MS[last_index]); + + assert_eq!(h.buckets[last_index], 1); + assert_eq!(h.buckets[DURATION_BUCKETS_MS.len()], 0); + } + #[test] fn test_start_timer_records_duration() { let m = OperatorMetrics::new(); diff --git a/tests/integration.rs b/tests/integration.rs new file mode 100644 index 0000000..30fc222 --- /dev/null +++ b/tests/integration.rs @@ -0,0 +1,241 @@ +//! Explicitly gated integration tests for the Streamline Operator. +//! +//! `cargo test` is hermetic by design: nothing in the default test run contacts a +//! Kubernetes API server or a live Streamline broker. The tests that *do* need +//! those services are marked `#[ignore]` and only run when asked for explicitly: +//! +//! ```bash +//! make integration-up # docker compose -f docker-compose.test.yml up -d --wait +//! make test-integration # cargo test --test integration -- --ignored +//! make integration-down +//! ``` +//! +//! Every networked assertion is bounded by [`IntegrationConfig::timeout`], so a +//! missing or wedged backend fails fast instead of hanging CI. +//! +//! ## Configuration +//! +//! | Variable | Description | Default | +//! |---|---|---| +//! | `STREAMLINE_TEST_IMAGE` | Server image used by `docker-compose.test.yml` | `ghcr.io/streamlinelabs/streamline:0.2.0` | +//! | `STREAMLINE_TEST_HTTP_PORT` | Host port mapped to the HTTP API | `9094` | +//! | `STREAMLINE_TEST_KAFKA_PORT` | Host port mapped to the Kafka listener | `9092` | +//! | `STREAMLINE_TEST_HTTP_ENDPOINT` | Full HTTP endpoint override | `http://127.0.0.1:` | +//! | `STREAMLINE_TEST_KAFKA_ENDPOINT` | Full `host:port` Kafka override | `127.0.0.1:` | +//! | `STREAMLINE_TEST_TIMEOUT_SECS` | Per-request/connect bound | `15` | + +// unwrap/expect are acceptable in tests; the crate-wide lint targets production code. +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::time::Duration; + +/// Default Streamline server image. Kept in sync with `docker-compose.test.yml` +/// by [`compose_default_image_matches_config`]. +pub const DEFAULT_IMAGE: &str = "ghcr.io/streamlinelabs/streamline:0.2.0"; +const DEFAULT_HTTP_PORT: &str = "9094"; +const DEFAULT_KAFKA_PORT: &str = "9092"; +const DEFAULT_TIMEOUT_SECS: u64 = 15; + +/// Resolved endpoints and bounds for the gated integration suite. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct IntegrationConfig { + pub image: String, + pub http_endpoint: String, + pub kafka_endpoint: String, + pub timeout_secs: u64, +} + +impl IntegrationConfig { + /// Resolve configuration from the process environment. + pub fn from_env() -> Self { + Self::resolve(|key| std::env::var(key).ok()) + } + + /// Resolve configuration from an arbitrary lookup. + /// + /// Taking the lookup as a parameter keeps the hermetic tests below free of + /// global `set_var` mutation, which races across parallel test threads. + pub fn resolve(lookup: impl Fn(&str) -> Option) -> Self { + let non_empty = |key: &str| lookup(key).filter(|v| !v.trim().is_empty()); + + let http_port = + non_empty("STREAMLINE_TEST_HTTP_PORT").unwrap_or_else(|| DEFAULT_HTTP_PORT.to_string()); + let kafka_port = non_empty("STREAMLINE_TEST_KAFKA_PORT") + .unwrap_or_else(|| DEFAULT_KAFKA_PORT.to_string()); + + Self { + image: non_empty("STREAMLINE_TEST_IMAGE").unwrap_or_else(|| DEFAULT_IMAGE.to_string()), + http_endpoint: non_empty("STREAMLINE_TEST_HTTP_ENDPOINT") + .unwrap_or_else(|| format!("http://127.0.0.1:{http_port}")), + kafka_endpoint: non_empty("STREAMLINE_TEST_KAFKA_ENDPOINT") + .unwrap_or_else(|| format!("127.0.0.1:{kafka_port}")), + timeout_secs: non_empty("STREAMLINE_TEST_TIMEOUT_SECS") + .and_then(|v| v.parse().ok()) + .filter(|secs| *secs > 0) + .unwrap_or(DEFAULT_TIMEOUT_SECS), + } + } + + /// Upper bound applied to every networked assertion. + pub fn timeout(&self) -> Duration { + Duration::from_secs(self.timeout_secs) + } + + fn url(&self, path: &str) -> String { + format!( + "{}/{}", + self.http_endpoint.trim_end_matches('/'), + path.trim_start_matches('/') + ) + } +} + +// --------------------------------------------------------------------------- +// Hermetic tests — no Kubernetes, no Streamline, no Docker required +// --------------------------------------------------------------------------- + +#[test] +fn defaults_resolve_without_any_environment() { + let config = IntegrationConfig::resolve(|_| None); + + assert_eq!(config.image, DEFAULT_IMAGE); + assert_eq!(config.http_endpoint, "http://127.0.0.1:9094"); + assert_eq!(config.kafka_endpoint, "127.0.0.1:9092"); + assert_eq!(config.timeout(), Duration::from_secs(DEFAULT_TIMEOUT_SECS)); +} + +#[test] +fn image_and_endpoints_are_configurable() { + let config = IntegrationConfig::resolve(|key| match key { + "STREAMLINE_TEST_IMAGE" => Some("registry.internal/streamline:9.9.9".to_string()), + "STREAMLINE_TEST_HTTP_ENDPOINT" => Some("https://streamline.internal:8443".to_string()), + "STREAMLINE_TEST_KAFKA_ENDPOINT" => Some("streamline.internal:19092".to_string()), + "STREAMLINE_TEST_TIMEOUT_SECS" => Some("42".to_string()), + _ => None, + }); + + assert_eq!(config.image, "registry.internal/streamline:9.9.9"); + assert_eq!(config.http_endpoint, "https://streamline.internal:8443"); + assert_eq!(config.kafka_endpoint, "streamline.internal:19092"); + assert_eq!(config.timeout(), Duration::from_secs(42)); +} + +#[test] +fn ports_override_endpoint_defaults() { + let config = IntegrationConfig::resolve(|key| match key { + "STREAMLINE_TEST_HTTP_PORT" => Some("18094".to_string()), + "STREAMLINE_TEST_KAFKA_PORT" => Some("18092".to_string()), + _ => None, + }); + + assert_eq!(config.http_endpoint, "http://127.0.0.1:18094"); + assert_eq!(config.kafka_endpoint, "127.0.0.1:18092"); +} + +#[test] +fn blank_and_invalid_values_fall_back_to_defaults() { + let config = IntegrationConfig::resolve(|key| match key { + "STREAMLINE_TEST_IMAGE" => Some(" ".to_string()), + "STREAMLINE_TEST_TIMEOUT_SECS" => Some("not-a-number".to_string()), + _ => None, + }); + + assert_eq!(config.image, DEFAULT_IMAGE); + assert_eq!(config.timeout_secs, DEFAULT_TIMEOUT_SECS); + + // A zero timeout would make every bounded assertion fail instantly. + let zeroed = IntegrationConfig::resolve(|key| { + (key == "STREAMLINE_TEST_TIMEOUT_SECS").then(|| "0".to_string()) + }); + assert_eq!(zeroed.timeout_secs, DEFAULT_TIMEOUT_SECS); +} + +#[test] +fn url_join_is_stable_regardless_of_slashes() { + let config = IntegrationConfig::resolve(|key| { + (key == "STREAMLINE_TEST_HTTP_ENDPOINT").then(|| "http://127.0.0.1:9094/".to_string()) + }); + + assert_eq!(config.url("/health"), "http://127.0.0.1:9094/health"); + assert_eq!(config.url("health"), "http://127.0.0.1:9094/health"); +} + +/// The compose file and this harness must agree on the default image, otherwise +/// `make integration-up` and `make test-integration` target different servers. +#[test] +fn compose_default_image_matches_config() { + let compose = std::fs::read_to_string(concat!( + env!("CARGO_MANIFEST_DIR"), + "/docker-compose.test.yml" + )) + .expect("docker-compose.test.yml must exist"); + + let expected = format!("${{STREAMLINE_TEST_IMAGE:-{DEFAULT_IMAGE}}}"); + assert!( + compose.contains(&expected), + "docker-compose.test.yml must make the server image configurable as `{expected}`" + ); + assert!( + compose.contains("${STREAMLINE_TEST_HTTP_PORT:-9094}") + && compose.contains("${STREAMLINE_TEST_KAFKA_PORT:-9092}"), + "docker-compose.test.yml must make host ports configurable" + ); +} + +// --------------------------------------------------------------------------- +// Gated tests — require the services from docker-compose.test.yml +// --------------------------------------------------------------------------- + +#[tokio::test] +#[ignore = "requires a live Streamline server: make integration-up"] +async fn streamline_http_api_is_reachable() { + let config = IntegrationConfig::from_env(); + + let client = reqwest::Client::builder() + .timeout(config.timeout()) + .build() + .expect("failed to build HTTP client"); + + let response = tokio::time::timeout(config.timeout(), client.get(config.url("/health")).send()) + .await + .unwrap_or_else(|_| { + panic!( + "timed out after {:?} waiting for {}", + config.timeout(), + config.url("/health") + ) + }) + .unwrap_or_else(|e| panic!("request to {} failed: {e}", config.url("/health"))); + + assert!( + response.status().is_success(), + "expected a successful /health response from {}, got HTTP {}", + config.http_endpoint, + response.status() + ); +} + +#[tokio::test] +#[ignore = "requires a live Streamline server: make integration-up"] +async fn streamline_kafka_listener_accepts_connections() { + let config = IntegrationConfig::from_env(); + + let stream = tokio::time::timeout( + config.timeout(), + tokio::net::TcpStream::connect(&config.kafka_endpoint), + ) + .await + .unwrap_or_else(|_| { + panic!( + "timed out after {:?} connecting to {}", + config.timeout(), + config.kafka_endpoint + ) + }); + + assert!( + stream.is_ok(), + "expected the Kafka listener at {} to accept a TCP connection", + config.kafka_endpoint + ); +}