feat: Kafka agent

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,7 +10,8 @@ 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.121.0", features = ["crds", "webhook"] }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", branch = "feat/platform-access", features = ["agent", "crds", "webhook"] }
+stackable-kafka-shared = { path = "rust/shared" }
 
 anyhow = "1.0"
 built = { version = "0.8", features = ["chrono", "git2"] }

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 }}

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

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.

rust/agent-binary/Cargo.toml

@@ -0,0 +1,24 @@
+[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-operator.workspace = true
+
+anyhow.workspace = true
+clap.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+
+[build-dependencies]
+built.workspace = true
+
+[[bin]]
+name = "stackable-kafka-agent"
+path = "src/main.rs"

rust/agent-binary/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+    built::write_built_file().unwrap();
+}

rust/agent-binary/src/main.rs

@@ -0,0 +1,44 @@
+//! The Stackable Kafka agent: a per-cluster controller that reconciles in-cluster resources like `KafkaTopic`s.
+
+use clap::Parser;
+use stackable_operator::{
+    cli::{Command, CommonOptions},
+    telemetry::Tracing,
+    utils::signal::SignalWatcher,
+};
+
+mod built_info {
+    include!(concat!(env!("OUT_DIR"), "/built.rs"));
+}
+
+#[derive(clap::Parser)]
+#[clap(about, author)]
+struct Opts {
+    #[clap(subcommand)]
+    cmd: Command<CommonOptions>,
+}
+
+#[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 sigterm_watcher = SignalWatcher::sigterm()?;
+            sigterm_watcher.handle().await;
+        }
+    }
+
+    Ok(())
+}

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