diff --git a/Cargo.lock b/Cargo.lock index 6f1e15a37..dcbd173f2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3121,6 +3121,18 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "stackable-agent" +version = "0.1.0" +source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.122.0#b0b8ca31a79da03b3fa20507b709390d2647efe4" +dependencies = [ + "k8s-openapi", + "schemars", + "serde", + "stackable-versioned", + "strum", +] + [[package]] name = "stackable-certs" version = "0.5.0" @@ -3145,6 +3157,21 @@ dependencies = [ "zeroize", ] +[[package]] +name = "stackable-kafka-agent" +version = "0.0.0-dev" +dependencies = [ + "anyhow", + "axum", + "built", + "clap", + "snafu 0.9.2", + "stackable-kafka-shared", + "stackable-operator", + "tokio", + "tracing", +] + [[package]] name = "stackable-kafka-operator" version = "0.0.0-dev" @@ -3160,12 +3187,22 @@ dependencies = [ "serde", "serde_json", "snafu 0.9.2", + "stackable-kafka-shared", "stackable-operator", "strum", "tokio", "tracing", ] +[[package]] +name = "stackable-kafka-shared" +version = "0.0.0-dev" +dependencies = [ + "rstest", + "stackable-operator", + "strum", +] + [[package]] name = "stackable-operator" version = "0.122.0" @@ -3195,6 +3232,7 @@ dependencies = [ "serde_json", "sha2", "snafu 0.9.2", + "stackable-agent", "stackable-operator-derive", "stackable-shared", "stackable-telemetry", diff --git a/Cargo.nix b/Cargo.nix index b9c95c789..87fad5d3f 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -38,29 +38,37 @@ rec { # "public" attributes that we attempt to keep stable with new versions of crate2nix. # - rootCrate = rec { - packageId = "stackable-kafka-operator"; - # Use this attribute to refer to the derivation building your root crate package. - # You can override the features with rootCrate.build.override { features = [ "default" "feature1" ... ]; }. - build = internal.buildRustCrateWithFeatures { - inherit packageId; - }; - - # Debug support which might change between releases. - # File a bug if you depend on any for non-debug work! - debug = internal.debugCrate { inherit packageId; }; - }; # Refer your crate build derivation by name here. # You can override the features with # workspaceMembers."${crateName}".build.override { features = [ "default" "feature1" ... ]; }. workspaceMembers = { + "stackable-kafka-agent" = rec { + packageId = "stackable-kafka-agent"; + build = internal.buildRustCrateWithFeatures { + packageId = "stackable-kafka-agent"; + }; + + # Debug support which might change between releases. + # File a bug if you depend on any for non-debug work! + debug = internal.debugCrate { inherit packageId; }; + }; "stackable-kafka-operator" = rec { packageId = "stackable-kafka-operator"; build = internal.buildRustCrateWithFeatures { packageId = "stackable-kafka-operator"; }; + # Debug support which might change between releases. + # File a bug if you depend on any for non-debug work! + debug = internal.debugCrate { inherit packageId; }; + }; + "stackable-kafka-shared" = rec { + packageId = "stackable-kafka-shared"; + build = internal.buildRustCrateWithFeatures { + packageId = "stackable-kafka-shared"; + }; + # Debug support which might change between releases. # File a bug if you depend on any for non-debug work! debug = internal.debugCrate { inherit packageId; }; @@ -10599,6 +10607,50 @@ rec { "std" = [ "alloc" ]; }; }; + "stackable-agent" = rec { + crateName = "stackable-agent"; + version = "0.1.0"; + edition = "2024"; + workspace_member = null; + src = pkgs.fetchgit { + url = "https://github.com/stackabletech/operator-rs.git"; + rev = "b0b8ca31a79da03b3fa20507b709390d2647efe4"; + sha256 = "1yfmjslv9s111zc4zfqzi2aankchiw8yjxx2zlc17wgayqrbmxrj"; + }; + libName = "stackable_agent"; + license = "Apache-2.0"; + authors = [ + "Stackable GmbH " + ]; + dependencies = [ + { + name = "k8s-openapi"; + packageId = "k8s-openapi"; + usesDefaultFeatures = false; + features = [ "schemars" "v1_36" ]; + } + { + name = "schemars"; + packageId = "schemars"; + features = [ "url2" ]; + } + { + name = "serde"; + packageId = "serde"; + features = [ "derive" ]; + } + { + name = "stackable-versioned"; + packageId = "stackable-versioned"; + } + { + name = "strum"; + packageId = "strum"; + features = [ "derive" ]; + } + ]; + + }; "stackable-certs" = rec { crateName = "stackable-certs"; version = "0.5.0"; @@ -10703,6 +10755,67 @@ rec { }; resolvedDefaultFeatures = [ "default" "rustls" ]; }; + "stackable-kafka-agent" = rec { + crateName = "stackable-kafka-agent"; + version = "0.0.0-dev"; + edition = "2024"; + crateBin = [ + { + name = "stackable-kafka-agent"; + path = "src/main.rs"; + requiredFeatures = [ ]; + } + ]; + src = lib.cleanSourceWith { filter = sourceFilter; src = ./rust/agent-binary; }; + license = "OSL-3.0"; + authors = [ + "Stackable GmbH " + ]; + dependencies = [ + { + name = "anyhow"; + packageId = "anyhow"; + } + { + name = "axum"; + packageId = "axum"; + } + { + name = "clap"; + packageId = "clap"; + } + { + name = "snafu"; + packageId = "snafu 0.9.2"; + } + { + name = "stackable-kafka-shared"; + packageId = "stackable-kafka-shared"; + } + { + name = "stackable-operator"; + packageId = "stackable-operator"; + features = [ "agent" "crds" "webhook" ]; + } + { + name = "tokio"; + packageId = "tokio"; + features = [ "full" ]; + } + { + name = "tracing"; + packageId = "tracing"; + } + ]; + buildDependencies = [ + { + name = "built"; + packageId = "built"; + features = [ "chrono" "git2" ]; + } + ]; + + }; "stackable-kafka-operator" = rec { crateName = "stackable-kafka-operator"; version = "0.0.0-dev"; @@ -10757,10 +10870,14 @@ rec { name = "snafu"; packageId = "snafu 0.9.2"; } + { + name = "stackable-kafka-shared"; + packageId = "stackable-kafka-shared"; + } { name = "stackable-operator"; packageId = "stackable-operator"; - features = [ "crds" "webhook" ]; + features = [ "agent" "crds" "webhook" ]; } { name = "strum"; @@ -10791,6 +10908,36 @@ rec { } ]; + }; + "stackable-kafka-shared" = rec { + crateName = "stackable-kafka-shared"; + version = "0.0.0-dev"; + edition = "2024"; + src = lib.cleanSourceWith { filter = sourceFilter; src = ./rust/shared; }; + libName = "stackable_kafka_shared"; + license = "OSL-3.0"; + authors = [ + "Stackable GmbH " + ]; + dependencies = [ + { + name = "stackable-operator"; + packageId = "stackable-operator"; + features = [ "agent" "crds" "webhook" ]; + } + { + name = "strum"; + packageId = "strum"; + features = [ "derive" ]; + } + ]; + devDependencies = [ + { + name = "rstest"; + packageId = "rstest"; + } + ]; + }; "stackable-operator" = rec { crateName = "stackable-operator"; @@ -10916,6 +11063,11 @@ rec { name = "snafu"; packageId = "snafu 0.9.2"; } + { + name = "stackable-agent"; + packageId = "stackable-agent"; + optional = true; + } { name = "stackable-operator-derive"; packageId = "stackable-operator-derive"; @@ -10995,7 +11147,7 @@ rec { "time" = [ "stackable-shared/time" ]; "webhook" = [ "dep:stackable-webhook" ]; }; - resolvedDefaultFeatures = [ "crds" "default" "webhook" ]; + resolvedDefaultFeatures = [ "agent" "crds" "default" "webhook" ]; }; "stackable-operator-derive" = rec { crateName = "stackable-operator-derive"; diff --git a/Cargo.toml b/Cargo.toml index 022e3bffd..c05f43ef2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["rust/operator-binary"] +members = ["rust/operator-binary", "rust/agent-binary", "rust/shared"] resolver = "2" [workspace.package] @@ -10,9 +10,11 @@ edition = "2024" repository = "https://github.com/stackabletech/kafka-operator" [workspace.dependencies] -stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.122.0", features = ["crds", "webhook"] } +stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.122.0", features = ["agent", "crds", "webhook"] } +stackable-kafka-shared = { path = "rust/shared" } anyhow = "1.0" +axum = "0.8" built = { version = "0.8", features = ["chrono", "git2"] } clap = "4.6" const_format = "0.2" diff --git a/crate-hashes.json b/crate-hashes.json index e59dc4661..aa6c876cf 100644 --- a/crate-hashes.json +++ b/crate-hashes.json @@ -1,5 +1,6 @@ { "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.122.0#k8s-version@0.1.5": "1yfmjslv9s111zc4zfqzi2aankchiw8yjxx2zlc17wgayqrbmxrj", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.122.0#stackable-agent@0.1.0": "1yfmjslv9s111zc4zfqzi2aankchiw8yjxx2zlc17wgayqrbmxrj", "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.122.0#stackable-certs@0.5.0": "1yfmjslv9s111zc4zfqzi2aankchiw8yjxx2zlc17wgayqrbmxrj", "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.122.0#stackable-operator-derive@0.3.2": "1yfmjslv9s111zc4zfqzi2aankchiw8yjxx2zlc17wgayqrbmxrj", "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.122.0#stackable-operator@0.122.0": "1yfmjslv9s111zc4zfqzi2aankchiw8yjxx2zlc17wgayqrbmxrj", diff --git a/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml b/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml new file mode 100644 index 000000000..fb3f2995e --- /dev/null +++ b/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml @@ -0,0 +1,21 @@ +--- +# Agent ClusterRole: bound (via per KafkaCluster RoleBinding) to the ServiceAccount that the +# agent pod runs as. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: {{ include "kafka-operator.name" . }}-agent-clusterrole + labels: + {{- include "kafka-operator.labels" . | nindent 4 }} +rules: +{{ if .Capabilities.APIVersions.Has "security.openshift.io/v1" }} + # On OpenShift, the agent pod must be allowed to use the nonroot-v2 SCC + - apiGroups: + - security.openshift.io + resources: + - securitycontextconstraints + resourceNames: + - nonroot-v2 + verbs: + - use +{{ end }} diff --git a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml index 59a34e6ce..aafed503c 100644 --- a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml @@ -41,7 +41,7 @@ rules: - list - patch - watch - # Required to bind the product ClusterRole to the per-cluster ServiceAccount. + # Required to bind the product and agent ClusterRoles to the per-cluster ServiceAccounts. - apiGroups: - rbac.authorization.k8s.io resources: @@ -50,6 +50,7 @@ rules: - bind resourceNames: - {{ include "kafka-operator.name" . }}-clusterrole + - {{ include "kafka-operator.name" . }}-agent-clusterrole # StatefulSet created per role group (broker, KRaft controller). Applied via # SSA, tracked for orphan cleanup, and owned by the controller. - apiGroups: @@ -63,6 +64,18 @@ rules: - list - patch - watch + # Deployment of the agent. + - apiGroups: + - apps + resources: + - deployments + verbs: + - create + - delete + - get + - list + - patch + - watch # PodDisruptionBudget created per role group. Applied via SSA and tracked for orphan cleanup, and owned by the controller. - apiGroups: - policy diff --git a/extra/crds.yaml b/extra/crds.yaml index 8b954ae6b..48821787b 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -1024,6 +1024,60 @@ spec: nullable: true pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$ type: string + managementAccess: + description: |- + Management access of the Stackable Data Platform to this Kafka cluster, used to manage + resources like `KafkaTopic`s. The credential must be authorized in Kafka by the user. + nullable: true + properties: + authentication: + description: How the Stackable Data Platform authenticates to Kafka. + properties: + userProvided: + description: A credential provided by the user. + oneOf: + - required: + - anonymous + - required: + - tls + properties: + anonymous: + description: 'No credential: the product does not require authentication.' + type: object + tls: + description: 'Source of a TLS client certificate: a secret-operator SecretClass or a static Secret.' + oneOf: + - required: + - secretClassName + - required: + - secretName + properties: + secretClassName: + description: A SecretClass providing the certificate. + maxLength: 253 + minLength: 1 + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$ + type: string + secretName: + description: |- + A static Secret holding the certificate in the keys `tls.crt` and `tls.key` (PEM), e.g. as + a Secret of type `kubernetes.io/tls`. + maxLength: 253 + minLength: 1 + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$ + type: string + type: object + type: object + required: + - userProvided + type: object + enabled: + description: Whether management access is enabled. + type: boolean + required: + - authentication + - enabled + type: object metadataManager: description: |- Metadata manager to use for the Kafka cluster. diff --git a/rust/agent-binary/Cargo.toml b/rust/agent-binary/Cargo.toml new file mode 100644 index 000000000..700eece71 --- /dev/null +++ b/rust/agent-binary/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "stackable-kafka-agent" +description = "Stackable Agent for Apache Kafka" +version.workspace = true +authors.workspace = true +license.workspace = true +edition.workspace = true +repository.workspace = true +publish = false + +[dependencies] +stackable-kafka-shared.workspace = true +stackable-operator.workspace = true + +anyhow.workspace = true +axum.workspace = true +clap.workspace = true +snafu.workspace = true +tokio.workspace = true +tracing.workspace = true + +[build-dependencies] +built.workspace = true + +[[bin]] +name = "stackable-kafka-agent" +path = "src/main.rs" diff --git a/rust/agent-binary/build.rs b/rust/agent-binary/build.rs new file mode 100644 index 000000000..fa809bfd1 --- /dev/null +++ b/rust/agent-binary/build.rs @@ -0,0 +1,3 @@ +fn main() { + built::write_built_file().unwrap(); +} diff --git a/rust/agent-binary/src/health.rs b/rust/agent-binary/src/health.rs new file mode 100644 index 000000000..13a79ef6f --- /dev/null +++ b/rust/agent-binary/src/health.rs @@ -0,0 +1,58 @@ +//! The health endpoints of the agent, which the probes of its Deployment call. + +use std::{ + future::Future, + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::Arc, +}; + +use axum::{ + Router, + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, + routing::get, +}; +use snafu::{ResultExt, Snafu}; +use stackable_kafka_shared::agent::{HEALTH_PORT, LIVENESS_PATH, READINESS_PATH}; +use stackable_operator::webhook::health::HealthCheckRegistry; +use tokio::net::TcpListener; +use tracing::info; + +const SOCKET_ADDRESS: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), HEALTH_PORT); + +#[derive(Debug, Snafu)] +pub enum Error { + #[snafu(display("failed to listen on {address} for the health endpoints"))] + Bind { + source: std::io::Error, + address: SocketAddr, + }, + + #[snafu(display("failed to serve the health endpoints"))] + Serve { source: std::io::Error }, +} + +/// Serves the endpoints until `shutdown` completes. Liveness only shows that the agent answers, +/// readiness is decided by the checks of the registry. +pub async fn serve( + readiness: Arc, + shutdown: impl Future + Send + 'static, +) -> Result<(), Error> { + let router = Router::new() + .route(LIVENESS_PATH, get(StatusCode::OK)) + .route(READINESS_PATH, get(ready)) + .with_state(readiness); + let listener = TcpListener::bind(SOCKET_ADDRESS).await.context(BindSnafu { + address: SOCKET_ADDRESS, + })?; + info!(address = %SOCKET_ADDRESS, "Serving the health endpoints"); + axum::serve(listener, router) + .with_graceful_shutdown(shutdown) + .await + .context(ServeSnafu) +} + +async fn ready(State(readiness): State>) -> Response { + (&*readiness).into_response() +} diff --git a/rust/agent-binary/src/main.rs b/rust/agent-binary/src/main.rs new file mode 100644 index 000000000..2341559f1 --- /dev/null +++ b/rust/agent-binary/src/main.rs @@ -0,0 +1,51 @@ +//! The Stackable Kafka agent: a per-cluster controller that reconciles in-cluster resources like `KafkaTopic`s. + +use std::sync::Arc; + +use clap::Parser; +use stackable_operator::{ + cli::{Command, CommonOptions}, + telemetry::Tracing, + utils::signal::SignalWatcher, + webhook::health::HealthCheckRegistry, +}; + +mod health; + +mod built_info { + include!(concat!(env!("OUT_DIR"), "/built.rs")); +} + +#[derive(clap::Parser)] +#[clap(about, author)] +struct Opts { + #[clap(subcommand)] + cmd: Command, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + match Opts::parse().cmd { + Command::Crd => anyhow::bail!("this agent has no CRDs, they are owned by the operator"), + Command::Run(CommonOptions { telemetry, .. }) => { + let _tracing_guard = Tracing::pre_configured(built_info::PKG_NAME, telemetry).init()?; + + tracing::info!( + built_info.pkg_version = built_info::PKG_VERSION, + built_info.git_version = built_info::GIT_VERSION, + built_info.target = built_info::TARGET, + built_info.built_time_utc = built_info::BUILT_TIME_UTC, + built_info.rustc_version = built_info::RUSTC_VERSION, + "Starting {description}", + description = built_info::PKG_DESCRIPTION + ); + + let readiness = Arc::new(HealthCheckRegistry::new()); + + let sigterm_watcher = SignalWatcher::sigterm()?; + health::serve(readiness, sigterm_watcher.handle()).await?; + } + } + + Ok(()) +} diff --git a/rust/operator-binary/Cargo.toml b/rust/operator-binary/Cargo.toml index 38e9e1e18..ca41d66db 100644 --- a/rust/operator-binary/Cargo.toml +++ b/rust/operator-binary/Cargo.toml @@ -9,6 +9,7 @@ repository.workspace = true publish = false [dependencies] +stackable-kafka-shared.workspace = true stackable-operator.workspace = true indoc.workspace = true diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index dd3176ecf..1c349de28 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -15,6 +15,7 @@ use std::{ use const_format::concatcp; use snafu::{ResultExt, Snafu}; +use stackable_kafka_shared::agent::ClientSecurity; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, @@ -25,9 +26,9 @@ use stackable_operator::{ resources::{NoRuntimeLimits, Resources}, }, constant, - crd::listener, + crd::{listener, management_access::v1alpha1::TlsClientCredential}, k8s_openapi::api::{ - apps::v1::StatefulSet, + apps::v1::{Deployment, StatefulSet}, core::v1::{ConfigMap, Service, ServiceAccount}, policy::v1::PodDisruptionBudget, rbac::v1::RoleBinding, @@ -35,17 +36,21 @@ use stackable_operator::{ kube::{ Resource, api::{DynamicObject, ObjectMeta}, - core::{DeserializeGuard, error_boundary}, + core::{DeserializeGuard, Selector, error_boundary}, runtime::{controller::Action, reflector::ObjectRef}, }, logging::controller::ReconcilerError, shared::time::Duration, v2::{ HasName, HasUid, NameIsValidLabelValue, + agent_utils::AgentResourceNames, + kvp::label, role_group_utils::ResourceNames, role_utils, types::{ - kubernetes::{ConfigMapName, ListenerClassName, ListenerName, NamespaceName, Uid}, + kubernetes::{ + ConfigMapName, ListenerClassName, ListenerName, NamespaceName, SecretClassName, Uid, + }, operator::{ClusterName, ControllerName, OperatorName, ProductName, ProductVersion}, }, }, @@ -85,6 +90,11 @@ constant!(PRODUCT_NAME: ProductName = APP_NAME); constant!(OPERATOR_NAME: OperatorName = KAFKA_OPERATOR_NAME); constant!(CONTROLLER_NAME: ControllerName = KAFKA_CONTROLLER_NAME); +/// Selector matching the agents of all KafkaClusters, e.g. to watch them. +pub fn all_agents_selector() -> Selector { + Selector::from_iter(BTreeMap::from(label::all_agents_selector(&PRODUCT_NAME))) +} + #[derive(Snafu, Debug)] pub enum PodDescriptorsError { #[snafu(display( @@ -117,6 +127,7 @@ pub struct Applied; /// bootstrap [`Listener`](listener)s. pub struct KubernetesResources { pub stateful_sets: Vec, + pub deployments: Vec, pub services: Vec, pub listeners: Vec, pub config_maps: Vec, @@ -162,6 +173,7 @@ pub struct ValidatedCluster { /// address-less around the first reconcile runs; the listener-operator populates the ingress /// addresses and the `Listener` watch triggers a new run once it does. pub bootstrap_listeners: Vec, + pub agent_config: Option, } /// The number of pods a role group runs when its `replicas` is unset. @@ -286,6 +298,14 @@ impl ValidatedCluster { } } + /// Type-safe names for the resources of the agent. + pub fn agent_resource_names(&self) -> AgentResourceNames { + AgentResourceNames { + cluster_name: self.name.clone(), + product_name: PRODUCT_NAME.clone(), + } + } + /// Type-safe names for the resources of a given role group. pub(crate) fn role_group_resource_names( &self, @@ -367,6 +387,34 @@ impl ValidatedClusterConfig { } } +/// The configuration of the agent deployed with the cluster. +pub struct ValidatedAgentConfig { + pub image: String, + pub connection: AgentConnection, +} + +#[derive(Debug, PartialEq)] +pub enum AgentConnection { + Plaintext, + Tls { + server_ca_secret_class: SecretClassName, + }, + MutualTls { + server_ca_secret_class: SecretClassName, + client_credential: TlsClientCredential, + }, +} + +impl From<&AgentConnection> for ClientSecurity { + fn from(connection: &AgentConnection) -> Self { + match connection { + AgentConnection::Plaintext => Self::Plaintext, + AgentConnection::Tls { .. } => Self::Tls, + AgentConnection::MutualTls { .. } => Self::MutualTls, + } + } +} + /// Per-role configuration extracted during validation. /// /// Resolved from the raw [`v1alpha1::KafkaCluster`] spec during validation so the reconcile loop @@ -450,6 +498,7 @@ pub type ControllerRoleGroupConfig = role_utils::RoleGroupConfig< pub struct Ctx { pub client: stackable_operator::client::Client, pub operator_environment: OperatorEnvironmentOptions, + pub operator_image: Option, } #[derive(Snafu, Debug, EnumDiscriminants)] @@ -519,9 +568,13 @@ pub async fn reconcile_kafka( .context(DereferenceSnafu)?; // validate (no client required) - let validated_cluster = - validate::validate(kafka, dereferenced_objects, &ctx.operator_environment) - .context(ValidateClusterSnafu)?; + let validated_cluster = validate::validate( + kafka, + dereferenced_objects, + &ctx.operator_environment, + ctx.operator_image.as_deref(), + ) + .context(ValidateClusterSnafu)?; tracing::debug!( kerberos_enabled = validated_cluster.cluster_config.kafka_security.has_kerberos_enabled(), @@ -567,24 +620,31 @@ pub fn error_policy( #[cfg(test)] pub(crate) mod test_support { - use std::collections::BTreeMap; + use std::{collections::BTreeMap, str::FromStr}; use stackable_operator::{ + builder::meta::ObjectMetaBuilder, cli::OperatorEnvironmentOptions, commons::networking::DomainName, - crd::{authentication::core, listener}, + crd::{ + authentication::{core, kerberos, tls}, + listener, + }, utils::{cluster_info::KubernetesClusterInfo, yaml_from_str_singleton_map}, + v2::types::kubernetes::SecretClassName, }; use super::{ BrokerRoleGroupConfig, ControllerRoleGroupConfig, ValidatedCluster, - dereference::DereferencedObjects, validate::validate, + dereference::DereferencedObjects, security::ValidatedKafkaSecurity, validate::validate, }; use crate::crd::{ authentication::ResolvedAuthenticationClasses, authorization::KafkaAuthorizationConfig, v1alpha1, }; + pub const OPERATOR_IMAGE: &str = "oci.stackable.tech/sdp/kafka-operator:0.0.0-dev"; + /// The expected `app.kubernetes.io/version` label value for the given product version. /// /// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main @@ -601,6 +661,96 @@ pub(crate) mod test_support { yaml_from_str_singleton_map(yaml).expect("invalid test KafkaCluster YAML") } + fn tls_auth_class() -> core::v1alpha1::AuthenticationClass { + core::v1alpha1::AuthenticationClass { + metadata: ObjectMetaBuilder::new().name("tls-auth").build(), + spec: core::v1alpha1::AuthenticationClassSpec { + provider: core::v1alpha1::AuthenticationClassProvider::Tls( + tls::v1alpha1::AuthenticationProvider { + client_cert_secret_class: Some("client-auth-secret-class".to_string()), + }, + ), + }, + } + } + + fn kerberos_auth_class() -> core::v1alpha1::AuthenticationClass { + core::v1alpha1::AuthenticationClass { + metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(), + spec: core::v1alpha1::AuthenticationClassSpec { + provider: core::v1alpha1::AuthenticationClassProvider::Kerberos( + kerberos::v1alpha1::AuthenticationProvider { + kerberos_secret_class: "kerberos-secret-class".to_string(), + }, + ), + }, + } + } + + fn no_auth() -> ResolvedAuthenticationClasses { + ResolvedAuthenticationClasses::new(vec![]) + } + + /// Plaintext: no TLS, no authentication, no OPA. + pub fn plaintext() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + no_auth(), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + None, + None, + ) + } + + /// Server TLS only (encryption without client authentication). + pub fn server_tls() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + no_auth(), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + Some("tls".parse().unwrap()), + None, + ) + } + + /// Mutual TLS (client-certificate authentication). + pub fn client_auth_tls() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![tls_auth_class()]), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + Some("tls".parse().unwrap()), + None, + ) + } + + /// Kerberos, which also requires server and internal TLS. + pub fn kerberos() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![kerberos_auth_class()]), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + Some("tls".parse().unwrap()), + None, + ) + } + + /// Internal TLS only (broker/controller encryption without client TLS). + pub fn internal_tls() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + no_auth(), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + None, + None, + ) + } + + /// OPA authorization with a TLS truststore (internal + server TLS also enabled). + pub fn opa() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + no_auth(), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + Some("tls".parse().unwrap()), + Some("opa-tls".parse().unwrap()), + ) + } + fn cluster_info() -> KubernetesClusterInfo { KubernetesClusterInfo { cluster_domain: DomainName::try_from("cluster.local").expect("valid domain"), @@ -711,6 +861,7 @@ pub(crate) mod test_support { bootstrap_listeners: Vec::new(), }, &operator_environment(), + Some(OPERATOR_IMAGE), ) .expect("validate should succeed for the test fixture") } @@ -719,6 +870,13 @@ pub(crate) mod test_support { /// the result. Used for tests asserting on a specific validation failure. pub fn validate_err( kafka: &v1alpha1::KafkaCluster, + ) -> Result { + validate_with_operator_image(kafka, Some(OPERATOR_IMAGE)) + } + + pub fn validate_with_operator_image( + kafka: &v1alpha1::KafkaCluster, + operator_image: Option<&str>, ) -> Result { validate( kafka, @@ -729,6 +887,7 @@ pub(crate) mod test_support { bootstrap_listeners: Vec::new(), }, &operator_environment(), + operator_image, ) } } @@ -940,6 +1099,7 @@ spec: {} operator_service_name: "kafka-operator".to_owned(), image_repository: "oci.stackable.tech/sdp".to_owned(), }, + operator_image: None, }); reconcile_kafka(Arc::new(kafka), ctx).await diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs index 1ce080f87..5f740ecc4 100644 --- a/rust/operator-binary/src/controller/apply.rs +++ b/rust/operator-binary/src/controller/apply.rs @@ -78,6 +78,7 @@ impl<'a> Applier<'a> { // compile here instead of silently never being applied. let KubernetesResources { stateful_sets, + deployments, services, listeners, config_maps, @@ -97,6 +98,7 @@ impl<'a> Applier<'a> { let config_maps = self.add_resources(config_maps).await?; let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; let stateful_sets = self.add_resources(stateful_sets).await?; + let deployments = self.add_resources(deployments).await?; self.cluster_resources .delete_orphaned_resources(self.client) @@ -105,6 +107,7 @@ impl<'a> Applier<'a> { Ok(KubernetesResources { stateful_sets, + deployments, services, listeners, config_maps, diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 6b7879081..fd39567e2 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -397,7 +397,7 @@ mod tests { use indoc::indoc; use super::*; - use crate::controller::build::security::tests::{kerberos, plaintext}; + use crate::controller::test_support::{kerberos, plaintext}; #[test] fn controller_command_does_not_export_a_realm_without_kerberos() { diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 59cc807b2..30664d8ab 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -20,6 +20,9 @@ use crate::{ RoleName, ValidatedCluster, build::{ resource::{ + agent::{ + build_agent_deployment, build_agent_role_binding, build_agent_service_account, + }, discovery::build_discovery_configmap, pdb::build_pdb, rbac::{build_role_binding, build_service_account}, @@ -147,14 +150,24 @@ pub fn build(cluster: &ValidatedCluster) -> Result config_maps.push(build_discovery_configmap(cluster).context(DiscoveryConfigMapSnafu)?); + let mut deployments = vec![]; + let mut service_accounts = vec![build_service_account(cluster)]; + let mut role_bindings = vec![build_role_binding(cluster)]; + if let Some(agent_config) = &cluster.agent_config { + deployments.push(build_agent_deployment(cluster, agent_config)); + service_accounts.push(build_agent_service_account(cluster)); + role_bindings.push(build_agent_role_binding(cluster)); + } + Ok(KubernetesResources { stateful_sets, + deployments, services, listeners, config_maps, pod_disruption_budgets, - service_accounts: vec![build_service_account(cluster)], - role_bindings: vec![build_role_binding(cluster)], + service_accounts, + role_bindings, status: PhantomData, }) } @@ -169,6 +182,16 @@ pub(crate) fn recommended_labels_for_cluster_resources(cluster: &ValidatedCluste ) } +pub(crate) fn recommended_labels_for_agent_resources(cluster: &ValidatedCluster) -> Labels { + label::recommended_labels_for_agent_resources( + &cluster.name, + &PRODUCT_NAME, + &cluster.product_version, + &OPERATOR_NAME, + &CONTROLLER_NAME, + ) +} + pub(crate) fn recommended_labels_for_role_resources( cluster: &ValidatedCluster, role_name: &RoleName, @@ -225,6 +248,10 @@ pub(crate) fn recommended_labels_for_unversioned_role_group_resources( ) } +pub(crate) fn agent_selector(cluster: &ValidatedCluster) -> Labels { + label::agent_selector(&cluster.name, &PRODUCT_NAME) +} + /// Selector labels matching the pods of a role group. pub(crate) fn role_group_selector( cluster: &ValidatedCluster, @@ -555,6 +582,8 @@ mod tests { sorted_names(&resources.role_bindings), ["simple-kafka-rolebinding"] ); + // No agent without management access. + assert!(resources.deployments.is_empty()); } #[test] diff --git a/rust/operator-binary/src/controller/build/properties/listener.rs b/rust/operator-binary/src/controller/build/properties/listener.rs index acc632558..2cb533769 100644 --- a/rust/operator-binary/src/controller/build/properties/listener.rs +++ b/rust/operator-binary/src/controller/build/properties/listener.rs @@ -6,7 +6,13 @@ use std::collections::BTreeMap; -use stackable_operator::{commons::networking::DomainName, v2::types::kubernetes::NamespaceName}; +use stackable_operator::{ + commons::networking::DomainName, + v2::types::{ + common::Port, + kubernetes::{NamespaceName, ServiceName}, + }, +}; use crate::{ controller::{RoleGroupName, ValidatedCluster, security::ValidatedKafkaSecurity}, @@ -151,6 +157,17 @@ pub(crate) fn pod_fqdn( format!("${{env:POD_NAME}}.{sts_service_name}.{namespace}.svc.{cluster_domain}") } +/// `..svc.:`, which resolves to the Pods of a headless +/// Service. +pub(crate) fn headless_service_address( + headless_service_name: &ServiceName, + namespace: &NamespaceName, + cluster_domain: &DomainName, + port: &Port, +) -> String { + format!("{headless_service_name}.{namespace}.svc.{cluster_domain}:{port}") +} + #[cfg(test)] mod tests { use stackable_operator::{ diff --git a/rust/operator-binary/src/controller/build/resource/agent.rs b/rust/operator-binary/src/controller/build/resource/agent.rs new file mode 100644 index 000000000..f1fd3a84a --- /dev/null +++ b/rust/operator-binary/src/controller/build/resource/agent.rs @@ -0,0 +1,511 @@ +//! Builds the resources of the agent deployed alongside the cluster. + +use std::str::FromStr; + +use stackable_kafka_shared::agent::{ + ClientSecurity, ENV_BOOTSTRAP_SERVERS, ENV_CLIENT_SECURITY, ENV_CLUSTER_NAME, + ENV_CLUSTER_NAMESPACE, HEALTH_PORT, KAFKA_TLS_DIR, LIVENESS_PATH, READINESS_PATH, +}; +use stackable_operator::{ + builder::{ + meta::ObjectMetaBuilder, + pod::{ + PodBuilder, + container::ContainerBuilder, + probe::ProbeBuilder, + resources::ResourceRequirementsBuilder, + security::PodSecurityContextBuilder, + volume::{SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder}, + }, + }, + commons::secret_class::SecretClassVolumeProvisionParts, + constant, + k8s_openapi::{ + api::{ + apps::v1::{Deployment, DeploymentSpec, DeploymentStrategy}, + core::v1::{ + EnvVar, EnvVarSource, HTTPGetAction, ObjectFieldSelector, Probe, ServiceAccount, + }, + rbac::v1::RoleBinding, + }, + apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, + }, + shared::time::Duration, + v2::{ + builder::{ + meta::ownerreference_from_resource, + pod::container::{EnvVarName, EnvVarSet, new_container_builder}, + }, + rbac::{self, RbacResourceNames}, + types::kubernetes::{ContainerName, SecretClassName, VolumeName}, + }, +}; + +use crate::{ + controller::{ + AgentConnection, ValidatedAgentConfig, ValidatedCluster, + build::{ + agent_selector, properties::listener::headless_service_address, + recommended_labels_for_agent_resources, + }, + }, + crd::role::KafkaRole, +}; + +constant!(AGENT_CONTAINER_NAME: ContainerName = "agent"); +const HEALTH_PORT_NAME: &str = "health"; +constant!(KAFKA_TLS_VOLUME_NAME: VolumeName = "tls-kafka"); +constant!(KUBERNETES_NODE_NAME: EnvVarName = "KUBERNETES_NODE_NAME"); +constant!(KUBERNETES_CLUSTER_DOMAIN: EnvVarName = "KUBERNETES_CLUSTER_DOMAIN"); + +pub fn build_agent_deployment( + cluster: &ValidatedCluster, + agent_config: &ValidatedAgentConfig, +) -> Deployment { + let resource_names = cluster.agent_resource_names(); + let mut container_builder_agent = new_container_builder(&AGENT_CONTAINER_NAME); + let mut pod_builder = PodBuilder::new(); + + match &agent_config.connection { + AgentConnection::Plaintext => {} + AgentConnection::Tls { + server_ca_secret_class, + } => add_kafka_tls_volume_and_mount( + server_ca_secret_class, + &mut pod_builder, + &mut container_builder_agent, + ), + AgentConnection::MutualTls { + server_ca_secret_class, + client_credential, + } => { + add_kafka_tls_volume_and_mount( + server_ca_secret_class, + &mut pod_builder, + &mut container_builder_agent, + ); + client_credential + .add_volumes_and_mounts(&mut pod_builder, vec![&mut container_builder_agent]) + .expect( + "the agent Pod has no other volume or mount for the TLS client certificate", + ); + } + } + + container_builder_agent + .image(&agent_config.image) + .command(vec!["stackable-kafka-agent".to_owned()]) + .args(vec!["run".to_owned()]) + .add_env_vars(agent_env_vars(cluster, agent_config)) + .add_container_port(HEALTH_PORT_NAME, HEALTH_PORT.into()) + .readiness_probe(http_probe(READINESS_PATH, Duration::from_secs(10))) + .liveness_probe(http_probe(LIVENESS_PATH, Duration::from_secs(60))) + .resources( + ResourceRequirementsBuilder::new() + .with_cpu_request("100m") + .with_cpu_limit("500m") + .with_memory_request("128Mi") + .with_memory_limit("128Mi") + .build(), + ); + + pod_builder + .metadata( + ObjectMetaBuilder::new() + .with_labels(recommended_labels_for_agent_resources(cluster)) + .build(), + ) + .image_pull_secrets_from_product_image(&cluster.image) + .add_container(container_builder_agent.build()) + .service_account_name(resource_names.service_account_name().to_string()) + .security_context( + PodSecurityContextBuilder::with_stackable_defaults() + .fs_group(1000) + .build(), + ); + + Deployment { + metadata: ObjectMetaBuilder::new() + .name_and_namespace(cluster) + .name(resource_names.deployment_name().to_string()) + .ownerreference(ownerreference_from_resource(cluster, None, Some(true))) + .with_labels(recommended_labels_for_agent_resources(cluster)) + .build(), + spec: Some(DeploymentSpec { + replicas: Some(1), + // Never run two agents at the same time, as for the operator itself. + strategy: Some(DeploymentStrategy { + type_: Some("Recreate".to_owned()), + ..DeploymentStrategy::default() + }), + selector: LabelSelector { + match_labels: Some(agent_selector(cluster).into()), + ..LabelSelector::default() + }, + template: pod_builder.build_template(), + ..DeploymentSpec::default() + }), + status: None, + } +} + +fn http_probe(path: &str, period: Duration) -> Probe { + ProbeBuilder::http_get(HTTPGetAction { + path: Some(path.to_owned()), + port: IntOrString::String(HEALTH_PORT_NAME.to_owned()), + ..HTTPGetAction::default() + }) + .with_period(period) + .with_failure_threshold(3) + .build() + .expect("the probe period is a constant that fits into an i32") +} + +pub fn build_agent_service_account(cluster: &ValidatedCluster) -> ServiceAccount { + rbac::build_service_account( + cluster, + &cluster.agent_resource_names(), + recommended_labels_for_agent_resources(cluster), + ) +} + +/// Binds the agent ServiceAccount to the agent ClusterRole deployed by the Helm chart. +pub fn build_agent_role_binding(cluster: &ValidatedCluster) -> RoleBinding { + rbac::build_role_binding( + cluster, + &cluster.agent_resource_names(), + recommended_labels_for_agent_resources(cluster), + ) +} + +fn agent_env_vars(cluster: &ValidatedCluster, agent_config: &ValidatedAgentConfig) -> EnvVarSet { + let node_name_env_var = EnvVar { + name: KUBERNETES_NODE_NAME.to_string(), + value: None, + value_from: Some(EnvVarSource { + field_ref: Some(ObjectFieldSelector { + field_path: "spec.nodeName".to_owned(), + ..ObjectFieldSelector::default() + }), + ..EnvVarSource::default() + }), + }; + + EnvVarSet::new() + .with_value(&ENV_BOOTSTRAP_SERVERS, bootstrap_servers(cluster)) + .with_value(&ENV_CLUSTER_NAME, cluster.name.to_string()) + .with_value(&ENV_CLUSTER_NAMESPACE, cluster.namespace.to_string()) + .with_value( + &ENV_CLIENT_SECURITY, + ClientSecurity::from(&agent_config.connection).to_string(), + ) + // Saves the agent from looking the domain up via the kubelet, which needs extra RBAC. + .with_value( + &KUBERNETES_CLUSTER_DOMAIN, + cluster.cluster_domain.to_string(), + ) + .with_env_var(node_name_env_var) + .expect("the env var name is a valid EnvVarName") +} + +/// The headless Services of the broker role groups, which resolve to the broker Pods. +/// +/// They keep the traffic within the Kubernetes cluster regardless of the ListenerClass. The pod +/// scope of the broker certificates puts them into the SANs, so hostname verification passes. The +/// agent stays on the bootstrap connection, so it never follows the advertised addresses. +fn bootstrap_servers(cluster: &ValidatedCluster) -> String { + let client_port = cluster.cluster_config.kafka_security.client_port(); + cluster + .broker_role_group_configs + .keys() + .map(|role_group_name| { + headless_service_address( + &cluster + .role_group_resource_names(&KafkaRole::Broker, role_group_name) + .headless_service_name(), + &cluster.namespace, + &cluster.cluster_domain, + &client_port, + ) + }) + .collect::>() + .join(",") +} + +fn add_kafka_tls_volume_and_mount( + secret_class: &SecretClassName, + pod_builder: &mut PodBuilder, + container_builder: &mut ContainerBuilder, +) { + pod_builder + .add_volume( + VolumeBuilder::new(&*KAFKA_TLS_VOLUME_NAME) + .ephemeral( + SecretOperatorVolumeSourceBuilder::new( + secret_class, + SecretClassVolumeProvisionParts::Public, + ) + .with_format(SecretFormat::TlsPem) + .build() + .expect( + "the annotations are built from a valid SecretClass name and constant \ + values", + ), + ) + .build(), + ) + .expect("the volume name is statically defined and must not collide"); + container_builder + .add_volume_mount(&*KAFKA_TLS_VOLUME_NAME, KAFKA_TLS_DIR) + .expect("the mount path is statically defined and must not collide"); +} + +#[cfg(test)] +mod tests { + use serde_json::{Value, json}; + + use super::*; + use crate::controller::test_support::{ + OPERATOR_IMAGE, app_version_label, minimal_kafka, validated_cluster, + }; + + const TLS_SECRET_CLASS: &str = + "{enabled: true, authentication: {userProvided: {tls: {secretClassName: tls}}}}"; + + fn cluster(management_access: &str) -> ValidatedCluster { + cluster_with_image("{productVersion: 3.9.2}", management_access) + } + + fn cluster_with_image(image: &str, management_access: &str) -> ValidatedCluster { + let kafka = minimal_kafka(&format!( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: {image} + clusterConfig: + metadataManager: kraft + managementAccess: {management_access} + controllers: + roleGroups: + default: + replicas: 1 + brokers: + roleGroups: + default: + replicas: 1 + "# + )); + validated_cluster(&kafka) + } + + fn agent_deployment(cluster: &ValidatedCluster) -> Value { + let agent_config = cluster + .agent_config + .as_ref() + .expect("management access is enabled"); + serde_json::to_value(build_agent_deployment(cluster, agent_config)) + .expect("must be serializable") + } + + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *AGENT_CONTAINER_NAME; + let _ = *KAFKA_TLS_VOLUME_NAME; + let _ = *KUBERNETES_NODE_NAME; + let _ = *KUBERNETES_CLUSTER_DOMAIN; + } + + #[test] + fn agent_config_is_only_resolved_if_management_access_is_enabled() { + assert!( + cluster( + "{enabled: false, authentication: {userProvided: {tls: {secretClassName: tls}}}}" + ) + .agent_config + .is_none() + ); + + let cluster = cluster(TLS_SECRET_CLASS); + let agent_config = cluster.agent_config.expect("management access is enabled"); + assert_eq!(agent_config.image, OPERATOR_IMAGE); + } + + #[test] + fn test_deployment() { + let deployment = agent_deployment(&cluster(TLS_SECRET_CLASS)); + + assert_eq!(deployment["metadata"]["name"], "simple-kafka-agent"); + assert_eq!(deployment["spec"]["replicas"], 1); + assert_eq!(deployment["spec"]["strategy"]["type"], "Recreate"); + // secret-operator volumes are only readable by the fsGroup. + assert_eq!( + deployment["spec"]["template"]["spec"]["securityContext"]["fsGroup"], + 1000 + ); + // The selector must not match the broker or controller Pods. + assert_eq!( + deployment["spec"]["selector"]["matchLabels"], + json!({ + "app.kubernetes.io/component": "kafka-agent", + "app.kubernetes.io/instance": "simple-kafka", + "app.kubernetes.io/name": "kafka" + }) + ); + assert_eq!( + deployment["spec"]["template"]["metadata"]["labels"], + json!({ + "app.kubernetes.io/component": "kafka-agent", + "app.kubernetes.io/instance": "simple-kafka", + "app.kubernetes.io/managed-by": "kafka.stackable.tech_kafkacluster", + "app.kubernetes.io/name": "kafka", + "app.kubernetes.io/version": app_version_label("3.9.2"), + "stackable.tech/vendor": "Stackable" + }) + ); + + let pod_spec = &deployment["spec"]["template"]["spec"]; + assert_eq!(pod_spec["serviceAccountName"], "simple-kafka-agent"); + assert_eq!(pod_spec["containers"][0]["image"], OPERATOR_IMAGE); + assert_eq!( + pod_spec["containers"][0]["command"], + json!(["stackable-kafka-agent"]) + ); + assert_eq!(pod_spec["containers"][0]["args"], json!(["run"])); + assert_eq!(pod_spec["containers"][0]["ports"][0]["name"], "health"); + assert_eq!(pod_spec["containers"][0]["ports"][0]["containerPort"], 8080); + assert_eq!( + pod_spec["containers"][0]["readinessProbe"]["httpGet"], + json!({"path": "/readyz", "port": "health"}) + ); + assert_eq!( + pod_spec["containers"][0]["livenessProbe"]["httpGet"], + json!({"path": "/healthz", "port": "health"}) + ); + assert_eq!( + pod_spec["containers"][0]["volumeMounts"], + json!([ + {"mountPath": "/stackable/secrets/tls-kafka", "name": "tls-kafka"}, + {"mountPath": "/stackable/secrets/tls-management-access", "name": "tls-management-access"} + ]) + ); + assert_eq!( + pod_spec["volumes"][0]["ephemeral"]["volumeClaimTemplate"]["metadata"]["annotations"], + json!({ + "secrets.stackable.tech/class": "tls", + "secrets.stackable.tech/format": "tls-pem", + "secrets.stackable.tech/provision-parts": "public" + }) + ); + assert_eq!(pod_spec["volumes"][1]["name"], "tls-management-access"); + + assert_eq!( + pod_spec["containers"][0]["env"], + json!([ + { + "name": "KAFKA_BOOTSTRAP_SERVERS", + "value": "simple-kafka-broker-default-headless.default.svc.cluster.local:9093" + }, + {"name": "KAFKA_CLIENT_SECURITY", "value": "mutual-tls"}, + {"name": "KAFKA_CLUSTER_NAME", "value": "simple-kafka"}, + {"name": "KAFKA_CLUSTER_NAMESPACE", "value": "default"}, + {"name": "KUBERNETES_CLUSTER_DOMAIN", "value": "cluster.local"}, + { + "name": "KUBERNETES_NODE_NAME", + "valueFrom": {"fieldRef": {"fieldPath": "spec.nodeName"}} + } + ]) + ); + } + + #[test] + fn test_deployment_with_static_secret() { + let deployment = agent_deployment(&cluster( + "{enabled: true, authentication: {userProvided: {tls: {secretName: my-cert}}}}", + )); + + assert_eq!( + deployment["spec"]["template"]["spec"]["volumes"][1], + json!({"name": "tls-management-access", "secret": {"optional": false, "secretName": "my-cert"}}) + ); + } + + #[test] + fn deployment_with_anonymous_authentication_verifies_brokers_without_credential() { + let deployment = agent_deployment(&cluster( + "{enabled: true, authentication: {userProvided: {anonymous: {}}}}", + )); + + let pod_spec = &deployment["spec"]["template"]["spec"]; + assert_eq!(pod_spec["volumes"][0]["name"], "tls-kafka"); + assert_eq!(pod_spec["volumes"][1], Value::Null); + assert_eq!( + pod_spec["containers"][0]["env"][1], + json!({"name": "KAFKA_CLIENT_SECURITY", "value": "tls"}) + ); + } + + #[test] + fn test_deployment_uses_product_image_pull_secrets() { + let deployment = agent_deployment(&cluster_with_image( + "{productVersion: 3.9.2, pullSecrets: [{name: regcred}]}", + TLS_SECRET_CLASS, + )); + + assert_eq!( + deployment["spec"]["template"]["spec"]["imagePullSecrets"], + json!([{"name": "regcred"}]) + ); + } + + #[test] + fn test_role_binding() { + let cluster = cluster(TLS_SECRET_CLASS); + + assert_eq!( + json!({ + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "RoleBinding", + "metadata": { + "labels": { + "app.kubernetes.io/component": "kafka-agent", + "app.kubernetes.io/instance": "simple-kafka", + "app.kubernetes.io/managed-by": "kafka.stackable.tech_kafkacluster", + "app.kubernetes.io/name": "kafka", + "app.kubernetes.io/version": app_version_label("3.9.2"), + "stackable.tech/vendor": "Stackable" + }, + "name": "simple-kafka-agent", + "namespace": "default", + "ownerReferences": [ + { + "apiVersion": "kafka.stackable.tech/v1alpha1", + "controller": true, + "kind": "KafkaCluster", + "name": "simple-kafka", + "uid": "12345678-1234-1234-1234-123456789012" + } + ] + }, + "roleRef": { + "apiGroup": "rbac.authorization.k8s.io", + "kind": "ClusterRole", + "name": "kafka-agent-clusterrole" + }, + "subjects": [ + { + "kind": "ServiceAccount", + "name": "simple-kafka-agent", + "namespace": "default" + } + ] + }), + serde_json::to_value(build_agent_role_binding(&cluster)).expect("must be serializable") + ); + } +} diff --git a/rust/operator-binary/src/controller/build/resource/mod.rs b/rust/operator-binary/src/controller/build/resource/mod.rs index 7f458714f..4acba6265 100644 --- a/rust/operator-binary/src/controller/build/resource/mod.rs +++ b/rust/operator-binary/src/controller/build/resource/mod.rs @@ -1,5 +1,6 @@ //! Builders that assemble Kubernetes resources for kafka rolegroups. +pub mod agent; pub mod config_map; pub mod discovery; pub mod listener; diff --git a/rust/operator-binary/src/controller/build/role_group/controller.rs b/rust/operator-binary/src/controller/build/role_group/controller.rs index af391e36f..3368dd11f 100644 --- a/rust/operator-binary/src/controller/build/role_group/controller.rs +++ b/rust/operator-binary/src/controller/build/role_group/controller.rs @@ -510,7 +510,7 @@ mod tests { #[test] fn quorum_manager_container_env_can_be_overridden() { let resolved_product_image = test_resolved_product_image(); - let kafka_security = crate::controller::build::security::tests::kerberos(); + let kafka_security = crate::controller::test_support::kerberos(); assert!( kafka_security.has_kerberos_enabled(), "sanity check failed: the fixture must have Kerberos enabled so that KRB5_CONFIG \ diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index dccf549ee..141f5a23e 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -13,7 +13,6 @@ use stackable_operator::{ }, commons::secret_class::SecretClassVolumeProvisionParts, constant, - crd::authentication::core, k8s_openapi::api::core::v1::Volume, shared::time::Duration, v2::types::kubernetes::VolumeName, @@ -306,7 +305,7 @@ pub fn mount_broker_tls_volumes( let mut volumes = Vec::new(); // add tls (server or client authentication volumes) if required - if let Some(tls_server_secret_class) = tls_secret_class(security) { + if let Some(tls_server_secret_class) = security.tls_secret_class() { // We have to mount tls pem files for kcat's readiness-probe command (the mount can be // used directly) volumes.push(create_kcat_tls_volume( @@ -626,19 +625,6 @@ pub fn controller_config_settings( config } -/// Returns the `SecretClass` provided in a `AuthenticationClass` for TLS. -fn tls_secret_class(security: &ValidatedKafkaSecurity) -> Option<&str> { - security - .tls_client_authentication_class() - .and_then(|auth_class| match &auth_class.spec.provider { - core::v1alpha1::AuthenticationClassProvider::Tls(tls) => { - tls.client_cert_secret_class.as_deref() - } - _ => None, - }) - .or_else(|| security.tls_server_secret_class()) -} - /// Creates ephemeral volumes to mount the `SecretClass` into the Pods for kcat client fn create_kcat_tls_volume( volume_name: &VolumeName, @@ -730,107 +716,13 @@ fn kcat_client_sasl_ssl(cert_directory: &str, service_name: &str) -> Vec } #[cfg(test)] -pub(crate) mod tests { - use std::{collections::BTreeMap, str::FromStr}; - - use stackable_operator::{ - builder::meta::ObjectMetaBuilder, - crd::authentication::{core, kerberos, tls}, - v2::types::kubernetes::SecretClassName, - }; +mod tests { + use std::collections::BTreeMap; use super::*; - use crate::crd::authentication::ResolvedAuthenticationClasses; - - fn tls_auth_class() -> core::v1alpha1::AuthenticationClass { - core::v1alpha1::AuthenticationClass { - metadata: ObjectMetaBuilder::new().name("tls-auth").build(), - spec: core::v1alpha1::AuthenticationClassSpec { - provider: core::v1alpha1::AuthenticationClassProvider::Tls( - tls::v1alpha1::AuthenticationProvider { - client_cert_secret_class: Some("client-auth-secret-class".to_string()), - }, - ), - }, - } - } - - fn kerberos_auth_class() -> core::v1alpha1::AuthenticationClass { - core::v1alpha1::AuthenticationClass { - metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(), - spec: core::v1alpha1::AuthenticationClassSpec { - provider: core::v1alpha1::AuthenticationClassProvider::Kerberos( - kerberos::v1alpha1::AuthenticationProvider { - kerberos_secret_class: "kerberos-secret-class".to_string(), - }, - ), - }, - } - } - - fn no_auth() -> ResolvedAuthenticationClasses { - ResolvedAuthenticationClasses::new(vec![]) - } - - /// Plaintext: no TLS, no authentication, no OPA. - pub(crate) fn plaintext() -> ValidatedKafkaSecurity { - ValidatedKafkaSecurity::new( - no_auth(), - SecretClassName::from_str("tls").expect("tls secret class name is valid"), - None, - None, - ) - } - - /// Server TLS only (encryption without client authentication). - fn server_tls() -> ValidatedKafkaSecurity { - ValidatedKafkaSecurity::new( - no_auth(), - SecretClassName::from_str("tls").expect("tls secret class name is valid"), - Some("tls".parse().unwrap()), - None, - ) - } - - /// Mutual TLS (client-certificate authentication). - fn client_auth_tls() -> ValidatedKafkaSecurity { - ValidatedKafkaSecurity::new( - ResolvedAuthenticationClasses::new(vec![tls_auth_class()]), - SecretClassName::from_str("tls").expect("tls secret class name is valid"), - Some("tls".parse().unwrap()), - None, - ) - } - - /// Kerberos, which also requires server and internal TLS. - pub(crate) fn kerberos() -> ValidatedKafkaSecurity { - ValidatedKafkaSecurity::new( - ResolvedAuthenticationClasses::new(vec![kerberos_auth_class()]), - SecretClassName::from_str("tls").expect("tls secret class name is valid"), - Some("tls".parse().unwrap()), - None, - ) - } - - /// Internal TLS only (broker/controller encryption without client TLS). - fn internal_tls() -> ValidatedKafkaSecurity { - ValidatedKafkaSecurity::new( - no_auth(), - SecretClassName::from_str("tls").expect("tls secret class name is valid"), - None, - None, - ) - } - - /// OPA authorization with a TLS truststore (internal + server TLS also enabled). - fn opa() -> ValidatedKafkaSecurity { - ValidatedKafkaSecurity::new( - no_auth(), - SecretClassName::from_str("tls").expect("tls secret class name is valid"), - Some("tls".parse().unwrap()), - Some("opa-tls".parse().unwrap()), - ) - } + use crate::controller::test_support::{ + client_auth_tls, internal_tls, kerberos, opa, plaintext, server_tls, + }; fn as_str_vec(commands: &[String]) -> Vec<&str> { commands.iter().map(String::as_str).collect() diff --git a/rust/operator-binary/src/controller/security.rs b/rust/operator-binary/src/controller/security.rs index 90bd62178..a97c732f2 100644 --- a/rust/operator-binary/src/controller/security.rs +++ b/rust/operator-binary/src/controller/security.rs @@ -101,6 +101,18 @@ impl ValidatedKafkaSecurity { .get_tls_authentication_class() } + /// The `SecretClass` of a TLS `AuthenticationClass`, otherwise the server `SecretClass`. + pub fn tls_secret_class(&self) -> Option<&str> { + self.tls_client_authentication_class() + .and_then(|auth_class| match &auth_class.spec.provider { + core::v1alpha1::AuthenticationClassProvider::Tls(tls) => { + tls.client_cert_secret_class.as_deref() + } + _ => None, + }) + .or_else(|| self.tls_server_secret_class()) + } + /// Retrieve the internal `SecretClass`. Internal (inter-broker) TLS is mandatory, so this is /// always set (defaults to `tls`). pub fn tls_internal_secret_class(&self) -> &str { diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 634a21b52..4ca0743c3 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -6,7 +6,7 @@ use std::{collections::BTreeMap, str::FromStr}; use serde::Serialize; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu, ensure}; use stackable_operator::{ cli::OperatorEnvironmentOptions, commons::product_image_selection, @@ -24,14 +24,15 @@ use stackable_operator::{ validate_logging_configuration_for_container, }, role_utils::{JavaCommonConfig, Role, RoleGroup, RoleGroupConfig, with_validated_config}, - types::kubernetes::ConfigMapName, + types::kubernetes::{ConfigMapName, SecretClassName}, }, }; use crate::{ controller::{ - BrokerRoleGroupConfig, ControllerRoleGroupConfig, RoleGroupName, ValidatedBrokerConfig, - ValidatedCluster, ValidatedClusterConfig, ValidatedControllerConfig, ValidatedRoleConfig, + AgentConnection, BrokerRoleGroupConfig, ControllerRoleGroupConfig, RoleGroupName, + ValidatedAgentConfig, ValidatedBrokerConfig, ValidatedCluster, ValidatedClusterConfig, + ValidatedControllerConfig, ValidatedRoleConfig, dereference::DereferencedObjects, replicas_or_default, security::{self, ValidatedKafkaSecurity}, @@ -131,6 +132,32 @@ pub enum Error { to join without it" ))] MissingKraftControllerRole, + + #[snafu(display( + "management access requires the operator image, which also contains the agent; set the \ + OPERATOR_IMAGE environment variable of the operator" + ))] + ManagementAccessRequiresOperatorImage, + + #[snafu(display("management access is only supported in KRaft mode"))] + ManagementAccessRequiresKraft, + + #[snafu(display("management access does not support clusters with Kerberos authentication"))] + ManagementAccessWithKerberos, + + #[snafu(display( + "the cluster requires client authentication, but management access is anonymous" + ))] + AnonymousManagementAccessWithClientAuthentication, + + #[snafu(display("TLS management access requires the cluster to use TLS"))] + TlsManagementAccessRequiresTls, + + #[snafu(display("the SecretClass {secret_class:?} of the client listener is invalid"))] + ParseClientListenerSecretClass { + source: stackable_operator::v2::macros::attributed_string_type::Error, + secret_class: String, + }, } /// Validated logging configuration for a Kafka role group's Kafka and (optional) Vector @@ -215,6 +242,7 @@ pub fn validate( kafka: &v1alpha1::KafkaCluster, dereferenced_objects: DereferencedObjects, operator_environment: &OperatorEnvironmentOptions, + operator_image: Option<&str>, ) -> Result { let image = kafka .spec @@ -342,6 +370,29 @@ pub fn validate( .cluster_domain .clone(); + let agent_config = kafka + .spec + .cluster_config + .management_access + .as_ref() + .filter(|management_access| management_access.enabled) + .map(|management_access| { + ensure!( + metadata_manager == crate::crd::MetadataManager::KRaft, + ManagementAccessRequiresKraftSnafu + ); + Ok(ValidatedAgentConfig { + image: operator_image + .context(ManagementAccessRequiresOperatorImageSnafu)? + .to_owned(), + connection: validate_agent_connection( + &management_access.authentication.user_provided, + &kafka_security, + )?, + }) + }) + .transpose()?; + Ok(ValidatedCluster { metadata: ValidatedCluster::object_meta(&name, &namespace, &uid), name, @@ -366,6 +417,7 @@ pub fn validate( controller_config, controller_role_group_configs, bootstrap_listeners: dereferenced_objects.bootstrap_listeners, + agent_config, }) } @@ -558,16 +610,159 @@ fn inject_cluster_id(env_overrides: EnvVarSet, cluster_id: Option<&str>) -> EnvV } } +/// Determines how the agent connects to the cluster, and rejects management access authentication +/// config that cannot connect to it. +fn validate_agent_connection( + authentication: &v1alpha1::UserProvidedAuthentication, + security: &ValidatedKafkaSecurity, +) -> Result { + ensure!( + !security.has_kerberos_enabled(), + ManagementAccessWithKerberosSnafu + ); + + let server_ca_secret_class = security + .tls_secret_class() + .map(|secret_class| { + SecretClassName::from_str(secret_class) + .context(ParseClientListenerSecretClassSnafu { secret_class }) + }) + .transpose()?; + + match (authentication, server_ca_secret_class) { + (v1alpha1::UserProvidedAuthentication::Anonymous(_), server_ca_secret_class) => { + ensure!( + security.tls_client_authentication_class().is_none(), + AnonymousManagementAccessWithClientAuthenticationSnafu + ); + Ok(match server_ca_secret_class { + None => AgentConnection::Plaintext, + Some(server_ca_secret_class) => AgentConnection::Tls { + server_ca_secret_class, + }, + }) + } + (v1alpha1::UserProvidedAuthentication::Tls(_), None) => { + TlsManagementAccessRequiresTlsSnafu.fail() + } + ( + v1alpha1::UserProvidedAuthentication::Tls(client_credential), + Some(server_ca_secret_class), + ) => Ok(AgentConnection::MutualTls { + server_ca_secret_class, + client_credential: client_credential.clone(), + }), + } +} + #[cfg(test)] mod tests { - use stackable_operator::v2::builder::pod::container::EnvVarSet; + use std::str::FromStr; + + use rstest::rstest; + use stackable_operator::{ + crd::management_access::v1alpha1::{Anonymous, TlsClientCredential}, + v2::{ + builder::pod::container::EnvVarSet, + types::kubernetes::{SecretClassName, SecretName}, + }, + }; - use super::{Error, KAFKA_CLUSTER_ID, inject_cluster_id}; - use crate::controller::test_support::{ - app_version_label, broker_role_group_config, controller_role_group_config, minimal_kafka, - validated_cluster, + use super::{Error, KAFKA_CLUSTER_ID, inject_cluster_id, validate_agent_connection}; + use crate::{ + controller::{ + AgentConnection, + security::ValidatedKafkaSecurity, + test_support::{ + app_version_label, broker_role_group_config, client_auth_tls, + controller_role_group_config, kerberos, minimal_kafka, plaintext, server_tls, + validate_with_operator_image, validated_cluster, + }, + }, + crd::v1alpha1::UserProvidedAuthentication, }; + fn anonymous() -> UserProvidedAuthentication { + UserProvidedAuthentication::Anonymous(Anonymous::default()) + } + + fn agent_cert() -> TlsClientCredential { + TlsClientCredential::SecretName( + SecretName::from_str("agent-cert").expect("the Secret name should be valid"), + ) + } + + fn tls() -> UserProvidedAuthentication { + UserProvidedAuthentication::Tls(agent_cert()) + } + + fn tls_secret_class(name: &str) -> SecretClassName { + SecretClassName::from_str(name).expect("the SecretClass name should be valid") + } + + #[rstest] + #[case::plaintext_anonymous(plaintext(), anonymous(), AgentConnection::Plaintext)] + #[case::server_tls_anonymous( + server_tls(), + anonymous(), + AgentConnection::Tls { server_ca_secret_class: tls_secret_class("tls") } + )] + #[case::server_tls_with_credential( + server_tls(), + tls(), + AgentConnection::MutualTls { + server_ca_secret_class: tls_secret_class("tls"), + client_credential: agent_cert(), + } + )] + #[case::client_auth_with_credential( + client_auth_tls(), + tls(), + AgentConnection::MutualTls { + server_ca_secret_class: tls_secret_class("client-auth-secret-class"), + client_credential: agent_cert(), + } + )] + fn agent_connection( + #[case] security: ValidatedKafkaSecurity, + #[case] authentication: UserProvidedAuthentication, + #[case] expected: AgentConnection, + ) { + let connection = validate_agent_connection(&authentication, &security) + .expect("the management access should be valid"); + + assert_eq!(expected, connection); + } + + #[rstest] + #[case::anonymous(anonymous())] + #[case::tls(tls())] + fn reject_management_access_with_kerberos(#[case] authentication: UserProvidedAuthentication) { + let error = validate_agent_connection(&authentication, &kerberos()) + .expect_err("the management access should be invalid"); + + assert!(matches!(error, Error::ManagementAccessWithKerberos)); + } + + #[test] + fn reject_anonymous_management_access_with_client_authentication() { + let error = validate_agent_connection(&anonymous(), &client_auth_tls()) + .expect_err("the management access should be invalid"); + + assert!(matches!( + error, + Error::AnonymousManagementAccessWithClientAuthentication + )); + } + + #[test] + fn reject_tls_management_access_without_tls() { + let error = validate_agent_connection(&tls(), &plaintext()) + .expect_err("the management access should be invalid"); + + assert!(matches!(error, Error::TlsManagementAccessRequiresTls)); + } + #[test] fn test_constants() { // Test that dereferencing the constants does not panic. @@ -679,6 +874,89 @@ mod tests { } } + #[rstest::rstest] + #[case::enabled(true)] + #[case::disabled(false)] + fn management_access_requires_the_operator_image(#[case] enabled: bool) { + let kafka = minimal_kafka(&format!( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + metadataManager: kraft + managementAccess: + enabled: {enabled} + authentication: + userProvided: + tls: + secretClassName: tls + controllers: + roleGroups: + default: + replicas: 1 + brokers: + roleGroups: + default: + replicas: 1 + "# + )); + + let result = validate_with_operator_image(&kafka, None); + + if enabled { + assert!( + matches!(result, Err(Error::ManagementAccessRequiresOperatorImage)), + "expected ManagementAccessRequiresOperatorImage, got: {:?}", + result.err() + ); + } else { + assert!(result.is_ok(), "expected Ok, got: {:?}", result.err()); + } + } + + #[test] + fn zookeeper_mode_rejects_management_access() { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + zookeeperConfigMapName: zk-discovery + managementAccess: + enabled: true + authentication: + userProvided: + anonymous: {} + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + + let result = crate::controller::test_support::validate_err(&kafka); + + assert!( + matches!(result, Err(Error::ManagementAccessRequiresKraft)), + "expected ManagementAccessRequiresKraft, got: {:?}", + result.err() + ); + } + #[test] fn injects_cluster_id_when_absent() { let env = inject_cluster_id(EnvVarSet::new(), Some("my-id")); diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 388ed5452..5035c1213 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -18,6 +18,7 @@ use stackable_operator::{ }, config::merge::Merge, constant, + crd::management_access::v1alpha1::{Anonymous, TlsClientCredential}, deep_merger::ObjectOverrides, kube::{CustomResource, runtime::reflector::ObjectRef}, role_utils::GenericRoleConfig, @@ -241,6 +242,37 @@ pub mod versioned { /// #[serde(skip_serializing_if = "Option::is_none")] pub broker_id_pod_config_map_name: Option, + + // no doc - docs in KafkaManagementAccess struct. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub management_access: Option, + } + + /// Management access of the Stackable Data Platform to this Kafka cluster, used to manage + /// resources like `KafkaTopic`s. The credential must be authorized in Kafka by the user. + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct KafkaManagementAccess { + /// Whether management access is enabled. + pub enabled: bool, + + /// How the Stackable Data Platform authenticates to Kafka. + pub authentication: v1alpha1::KafkaManagementAccessAuthentication, + } + + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct KafkaManagementAccessAuthentication { + // no doc - docs in UserProvidedAuthentication enum. + pub user_provided: v1alpha1::UserProvidedAuthentication, + } + + /// A credential provided by the user. + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub enum UserProvidedAuthentication { + Anonymous(Anonymous), + Tls(TlsClientCredential), } #[derive(Clone, Debug, Default, Deserialize, JsonSchema, Merge, PartialEq, Serialize)] @@ -274,6 +306,7 @@ impl Default for v1alpha1::KafkaClusterConfig { zookeeper_config_map_name: None, metadata_manager: None, broker_id_pod_config_map_name: None, + management_access: None, } } } diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 7cc27ba43..7a0a9c5f9 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -14,7 +14,7 @@ use stackable_operator::{ crd::listener, eos::EndOfSupportChecker, k8s_openapi::api::{ - apps::v1::StatefulSet, + apps::v1::{Deployment, StatefulSet}, core::v1::{ConfigMap, Service, ServiceAccount}, policy::v1::PodDisruptionBudget, rbac::v1::RoleBinding, @@ -37,7 +37,7 @@ use stackable_operator::{ }; use crate::{ - controller::KAFKA_FULL_CONTROLLER_NAME, + controller::{KAFKA_FULL_CONTROLLER_NAME, all_agents_selector}, crd::{KAFKA_OPERATOR_NAME, KafkaCluster, KafkaClusterVersion, v1alpha1}, webhooks::conversion::create_webhook_server, }; @@ -68,6 +68,11 @@ struct Opts { struct KafkaRun { #[clap(flatten)] common: RunArguments, + + /// The image of this operator, which also contains the agent. Required to enable management + /// access on a KafkaCluster. + #[arg(long, env)] + operator_image: Option, } #[tokio::main] @@ -84,7 +89,7 @@ async fn main() -> anyhow::Result<()> { maintenance, common, }, - .. + operator_image, }) => { // NOTE (@NickLarsenNZ): Before stackable-telemetry was used: // - The console log level was set by `KAFKA_OPERATOR_LOG`, and is now `CONSOLE_LOG` (when using Tracing::pre_configured). @@ -179,6 +184,11 @@ async fn main() -> anyhow::Result<()> { watch_namespace.get_api::>(&client), watcher::Config::default(), ) + .owns( + watch_namespace.get_api::>(&client), + // The agent is the only Deployment + watcher::Config::default().labels_from(&all_agents_selector()), + ) .watches( watch_namespace.get_api::>(&client), watcher::Config::default(), @@ -197,6 +207,7 @@ async fn main() -> anyhow::Result<()> { Arc::new(controller::Ctx { client: client.clone(), operator_environment, + operator_image, }), ) // We can let the reporting happen in the background diff --git a/rust/shared/Cargo.toml b/rust/shared/Cargo.toml new file mode 100644 index 000000000..a26223f1c --- /dev/null +++ b/rust/shared/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "stackable-kafka-shared" +description = "Types and contracts shared by the Stackable operator and agent for Apache Kafka" +version.workspace = true +authors.workspace = true +license.workspace = true +edition.workspace = true +repository.workspace = true +publish = false + +[dependencies] +stackable-operator.workspace = true + +strum.workspace = true + +[dev-dependencies] +rstest.workspace = true diff --git a/rust/shared/src/agent.rs b/rust/shared/src/agent.rs new file mode 100644 index 000000000..802ebc951 --- /dev/null +++ b/rust/shared/src/agent.rs @@ -0,0 +1,56 @@ +//! Paths and environment variables through which the operator configures the agent. + +use std::str::FromStr; + +use stackable_operator::{constant, v2::builder::pod::container::EnvVarName}; +use strum::{Display, EnumString}; + +/// Directory of the CA certificate the Kafka brokers are verified with. +pub const KAFKA_TLS_DIR: &str = "/stackable/secrets/tls-kafka"; + +/// Port of the HTTP health endpoints, which the probes of the agent Deployment call. +pub const HEALTH_PORT: u16 = 8080; +pub const LIVENESS_PATH: &str = "/healthz"; +pub const READINESS_PATH: &str = "/readyz"; + +constant!(pub ENV_CLUSTER_NAME: EnvVarName = "KAFKA_CLUSTER_NAME"); +constant!(pub ENV_CLUSTER_NAMESPACE: EnvVarName = "KAFKA_CLUSTER_NAMESPACE"); +constant!(pub ENV_CLIENT_SECURITY: EnvVarName = "KAFKA_CLIENT_SECURITY"); +// Comma-separated `host:port` list +constant!(pub ENV_BOOTSTRAP_SERVERS: EnvVarName = "KAFKA_BOOTSTRAP_SERVERS"); + +/// How the agent connects to the brokers. +#[derive(Clone, Copy, Debug, Display, EnumString, Eq, PartialEq)] +#[strum(serialize_all = "kebab-case")] +pub enum ClientSecurity { + Plaintext, + Tls, + MutualTls, +} + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use rstest::rstest; + + use super::*; + + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *ENV_CLUSTER_NAME; + let _ = *ENV_CLUSTER_NAMESPACE; + let _ = *ENV_CLIENT_SECURITY; + let _ = *ENV_BOOTSTRAP_SERVERS; + } + + #[rstest] + #[case::plaintext(ClientSecurity::Plaintext, "plaintext")] + #[case::tls(ClientSecurity::Tls, "tls")] + #[case::mutual_tls(ClientSecurity::MutualTls, "mutual-tls")] + fn roundtrip(#[case] security: ClientSecurity, #[case] value: &str) { + assert_eq!(value, security.to_string()); + assert_eq!(Ok(security), ClientSecurity::from_str(value)); + } +} diff --git a/rust/shared/src/lib.rs b/rust/shared/src/lib.rs new file mode 100644 index 000000000..f17bc55db --- /dev/null +++ b/rust/shared/src/lib.rs @@ -0,0 +1 @@ +pub mod agent;