crabka-operator 0.3.0

Kubernetes operator for Crabka clusters
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

use std::net::SocketAddr;

use clap::Args;
use serde::{Deserialize, Serialize};

/// Operator runtime configuration.
///
/// All fields can be set via CLI (`--watch-namespaces`, `--health-addr`, …)
/// or via env (`WATCH_NAMESPACES`, `HEALTH_ADDR`, …). CLI wins on conflict.
#[derive(Debug, Clone, Args, Serialize, Deserialize)]
pub struct OperatorConfig {
    /// Comma-separated namespaces to watch. Empty = cluster-scoped.
    #[arg(long, env = "WATCH_NAMESPACES", value_delimiter = ',', num_args = 0..)]
    pub watch_namespaces: Vec<String>,

    /// Namespace the operator runs in (used for the leader-election Lease).
    #[arg(long, env = "OPERATOR_NAMESPACE", default_value = "crabka-operator")]
    pub operator_namespace: String,

    /// Lease name for leader election.
    #[arg(long, env = "LEASE_NAME", default_value = "crabka-operator-leader")]
    pub lease_name: String,

    /// Identity advertised in the Lease (typically the pod name).
    #[arg(long, env = "POD_NAME", default_value = "crabka-operator-local")]
    pub pod_name: String,

    /// Address for `/healthz`, `/readyz`, `/metrics`.
    #[arg(long, env = "HEALTH_ADDR", default_value = "0.0.0.0:8080")]
    pub health_addr: SocketAddr,

    /// Tracing filter (e.g. `info,kube=warn`).
    #[arg(
        long,
        env = "RUST_LOG",
        default_value = "info,kube_client::client::builder=warn"
    )]
    pub log_filter: String,

    /// Default broker image used when `Kafka.spec.image` is unset.
    #[arg(long, env = "DEFAULT_BROKER_IMAGE")]
    pub default_broker_image: Option<String>,

    /// Default gateway image used when `KafkaGrpcGateway.spec.image` is unset.
    #[arg(long, env = "DEFAULT_GATEWAY_IMAGE")]
    pub default_gateway_image: Option<String>,
    /// Default schema-registry image used when `SchemaRegistry.spec.image` is unset.
    #[arg(long, env = "DEFAULT_SCHEMA_REGISTRY_IMAGE")]
    pub default_schema_registry_image: Option<String>,
}

impl OperatorConfig {
    /// Iterator over watched namespaces, or `None` for cluster-scoped.
    #[must_use]
    pub fn watched(&self) -> Option<&[String]> {
        if self.watch_namespaces.is_empty() {
            None
        } else {
            Some(&self.watch_namespaces)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use assert2::assert;
    use clap::Parser;

    #[derive(Parser)]
    struct Wrap {
        #[command(flatten)]
        cfg: OperatorConfig,
    }

    #[test]
    fn cli_defaults_compute_cluster_scope() {
        let parsed = Wrap::parse_from(["bin"]);
        assert!(parsed.cfg.watched().is_none());
        assert!(parsed.cfg.operator_namespace == "crabka-operator");
    }

    #[test]
    fn comma_separated_namespaces_parse() {
        let parsed = Wrap::parse_from(["bin", "--watch-namespaces=a,b,c"]);
        assert!(parsed.cfg.watch_namespaces == vec!["a", "b", "c"]);
        assert!(parsed.cfg.watched().is_some());
    }
}