Support specifying Service type

Author: sbernauer

CHANGELOG.md

@@ -12,6 +12,10 @@ All notable changes to this project will be documented in this file.
 
 ### Changed
 
+- Support specifying Service type.
+ This enables us to later switch non-breaking to using `ListenerClasses` for the exposure of Services.
+ So far only `external-unstable` is supported, which re-uses the current code to create NodePorts
+ and calculate the reachable address to advertise ([#XXX]).
 - `operator-rs` `0.30.1` -> `0.39.0` ([#545], [#574]).
 - Bumped stackable versions to "23.4.0-rc1" ([#545]).
 - Bumped kafka stackable versions to "23.4.0-rc2" ([#547]).

deploy/helm/kafka-operator/crds/crds.yaml

@@ -1423,6 +1423,15 @@ spec:
  - configMapName
  type: object
  type: object
+ listenerClass:
+ default: external-unstable
+ description: |-
+ In the future this setting will control, which ListenerClass <https://docs.stackable.tech/home/stable/listener-operator/listenerclass.html> will be used to expose the service. Currently only a subset of the ListenerClasses are supported by choosing the type of the created Services by looking at the ListenerClass name specified, In a future release support for custom ListenerClasses will be introduced without a breaking change:
+
+ * external-unstable: Use a NodePort service
+ enum:
+ - external-unstable
+ type: string
  tls:
  default:
  internalSecretClass: tls

rust/crd/src/lib.rs

@@ -116,6 +116,32 @@ pub struct KafkaClusterConfig {
  pub vector_aggregator_config_map_name: Option<String>,
  /// ZooKeeper discovery config map name.
  pub zookeeper_config_map_name: String,
+ /// In the future this setting will control, which ListenerClass <https://docs.stackable.tech/home/stable/listener-operator/listenerclass.html>
+ /// will be used to expose the service.
+ /// Currently only a subset of the ListenerClasses are supported by choosing the type of the created Services
+ /// by looking at the ListenerClass name specified,
+ /// In a future release support for custom ListenerClasses will be introduced without a breaking change:
+ ///
+ /// * external-unstable: Use a NodePort service
+ #[serde(default)]
+ pub listener_class: CurrentlySupportedListenerClasses,
+}
+
+// TODO: Temporary solution until listener-operator is finished
+#[derive(Clone, Debug, Default, Display, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum CurrentlySupportedListenerClasses {
+ #[default]
+ #[serde(rename = "external-unstable")]
+ ExternalUnstable,
+}
+
+impl CurrentlySupportedListenerClasses {
+ pub fn k8s_service_type(&self) -> String {
+ match self {
+ CurrentlySupportedListenerClasses::ExternalUnstable => "NodePort".to_string(),
+ }
+ }
 }
 
 impl KafkaCluster {

rust/operator/src/kafka_controller.rs

@@ -598,6 +598,8 @@ fn build_broker_rolegroup_service(
  .with_label("prometheus.io/scrape", "true")
  .build(),
  spec: Some(ServiceSpec {
+ // Internal communication does not need to be exposed
+ type_: Some("ClusterIP".to_string()),
  cluster_ip: Some("None".to_string()),
  ports: Some(service_ports(kafka_security)),
  selector: Some(role_group_selector_labels(