feat: Add listener support

CHANGELOG.md

@@ -16,5 +16,7 @@ All notable changes to this project will be documented in this file.
   - Resources (CPU, memory, storage)
   - PodDisruptionBudgets
   - Replicas
+- Add Listener support ([#17]).
 
 [#10]: https://github.com/stackabletech/opensearch-operator/pull/10
+[#17]: https://github.com/stackabletech/opensearch-operator/pull/17

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

@@ -133,6 +133,10 @@ spec:
                           description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                           nullable: true
                           type: string
+                        listenerClass:
+                          description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+                          nullable: true
+                          type: string
                         nodeRoles:
                           items:
                             enum:
@@ -325,6 +329,10 @@ spec:
                                 description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                                 nullable: true
                                 type: string
+                              listenerClass:
+                                description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+                                nullable: true
+                                type: string
                               nodeRoles:
                                 items:
                                   enum:

deploy/helm/opensearch-operator/templates/roles.yaml

@@ -70,6 +70,17 @@ rules:
       - customresourcedefinitions
     verbs:
       - get
+  - apiGroups:
+      - listeners.stackable.tech
+    resources:
+      - listeners
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+      - create
+      - delete
   - apiGroups:
       - events.k8s.io
     resources:

docs/modules/opensearch/pages/usage-guide/listenerclass.adoc

@@ -0,0 +1,17 @@
+= Service exposition with ListenerClasses
+:description: Configure OpenSearch service exposure with ListenerClasses: cluster-internal, external-unstable, or external-stable.
+
+The operator deploys a xref:listener-operator:listener.adoc[Listener] for OpenSearch role-groups.
+The listener defaults to only being accessible from within the Kubernetes cluster, but this can be changed by setting `.spec.nodes.roleGroups.{role-group-name}.config.listenerClass`:
+
+[source,yaml]
+----
+spec:
+  nodes:
+    roleGroups:
+      cluster-manager:
+        config:
+          listenerClass: external-stable
+----
+<1> Specify a ListenerClass, such as `external-stable`, `external-unstable`, or `cluster-internal` (the default setting is `cluster-internal`) at role-group level.
+This can be set for all role-groups individually.

docs/modules/opensearch/partials/nav.adoc

@@ -2,6 +2,7 @@
 ** xref:opensearch:getting_started/installation.adoc[]
 ** xref:opensearch:getting_started/first_steps.adoc[]
 * xref:opensearch:usage-guide/index.adoc[]
+** xref:opensearch:usage-guide/listenerclass.adoc[]
 ** xref:opensearch:usage-guide/node-roles.adoc[]
 ** xref:opensearch:usage-guide/storage-resource-configuration.adoc[]
 ** xref:opensearch:usage-guide/configuration-environment-overrides.adoc[]

rust/operator-binary/src/controller.rs

@@ -6,6 +6,7 @@ use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     cluster_resources::ClusterResourceApplyStrategy,
     commons::{affinity::StackableAffinity, product_image_selection::ProductImage},
+    crd::listener::v1alpha1::Listener,
     k8s_openapi::api::{
         apps::v1::StatefulSet,
         core::v1::{ConfigMap, Service, ServiceAccount},
@@ -111,6 +112,7 @@ pub struct ValidatedOpenSearchConfig {
     pub node_roles: NodeRoles,
     pub resources: stackable_operator::commons::resources::Resources<v1alpha1::StorageConfig>,
     pub termination_grace_period_seconds: i64,
+    pub listener_class: String,
 }
 
 // validated and converted to validated and safe types
@@ -275,6 +277,7 @@ struct Applied;
 struct KubernetesResources<T> {
     stateful_sets: Vec<StatefulSet>,
     services: Vec<Service>,
+    listeners: Vec<Listener>,
     config_maps: Vec<ConfigMap>,
     service_accounts: Vec<ServiceAccount>,
     role_bindings: Vec<RoleBinding>,

rust/operator-binary/src/controller/apply.rs

@@ -62,6 +62,8 @@ impl<'a> Applier<'a> {
 
         let services = self.add_resources(resources.services).await?;
 
+        let listeners = self.add_resources(resources.listeners).await?;
+
         let config_maps = self.add_resources(resources.config_maps).await?;
 
         let service_accounts = self.add_resources(resources.service_accounts).await?;
@@ -78,6 +80,7 @@ impl<'a> Applier<'a> {
         Ok(KubernetesResources {
             stateful_sets,
             services,
+            listeners,
             config_maps,
             service_accounts,
             role_bindings,

rust/operator-binary/src/controller/build.rs

@@ -12,13 +12,15 @@ pub fn build(names: &ContextNames, cluster: ValidatedCluster) -> KubernetesResou
     let mut config_maps = vec![];
     let mut stateful_sets = vec![];
     let mut services = vec![];
+    let mut listeners = vec![];
 
     let role_builder = RoleBuilder::new(cluster.clone(), names);
 
     for role_group_builder in role_builder.role_group_builders() {
         config_maps.push(role_group_builder.build_config_map());
         stateful_sets.push(role_group_builder.build_stateful_set());
         services.push(role_group_builder.build_headless_service());
+        listeners.push(role_group_builder.build_listener());
     }
 
     let cluster_manager_service = role_builder.build_cluster_manager_service();
@@ -33,6 +35,7 @@ pub fn build(names: &ContextNames, cluster: ValidatedCluster) -> KubernetesResou
     KubernetesResources {
         stateful_sets,
         services,
+        listeners,
         config_maps,
         service_accounts,
         role_bindings,

rust/operator-binary/src/controller/build/node_config.rs

@@ -286,6 +286,7 @@ mod tests {
                 node_roles: NodeRoles::default(),
                 resources: Resources::default(),
                 termination_grace_period_seconds: 30,
+                listener_class: "cluster-internal".to_string(),
             },
             config_overrides: HashMap::default(),
             env_overrides: [("TEST".to_owned(), "value".to_owned())].into(),

rust/operator-binary/src/controller/build/role_group_builder.rs

@@ -1,13 +1,22 @@
+use std::str::FromStr;
+
 use stackable_operator::{
-    builder::{meta::ObjectMetaBuilder, pod::container::ContainerBuilder},
+    builder::{
+        meta::ObjectMetaBuilder,
+        pod::{
+            container::ContainerBuilder,
+            volume::{ListenerOperatorVolumeSourceBuilder, ListenerReference},
+        },
+    },
+    crd::listener::{self},
     k8s_openapi::{
         DeepMerge,
         api::{
             apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
                 Affinity, ConfigMap, ConfigMapVolumeSource, Container, ContainerPort,
-                PodSecurityContext, PodSpec, PodTemplateSpec, Probe, Service, ServicePort,
-                ServiceSpec, TCPSocketAction, Volume, VolumeMount,
+                PersistentVolumeClaim, PodSecurityContext, PodSpec, PodTemplateSpec, Probe,
+                Service, ServicePort, ServiceSpec, TCPSocketAction, Volume, VolumeMount,
             },
         },
         apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
@@ -21,7 +30,7 @@ use crate::{
     controller::{ContextNames, OpenSearchRoleGroupConfig, ValidatedCluster},
     crd::v1alpha1,
     framework::{
-        RoleGroupName,
+        ProductVersion, RoleGroupName,
         builder::meta::ownerreference_from_resource,
         kvp::label::{recommended_labels, role_group_selector, role_selector},
         role_group_utils::ResourceNames,
@@ -36,8 +45,11 @@ pub const TRANSPORT_PORT: u16 = 9300;
 const CONFIG_VOLUME_NAME: &str = "config";
 const DATA_VOLUME_NAME: &str = "data";
 
+const LISTENER_VOLUME_NAME: &str = "listener";
+const LISTENER_VOLUME_DIR: &str = "/stackable/listener";
+
 // Path in opensearchproject/opensearch:3.0.0
-const OPENSEARCH_BASE_PATH: &str = "/usr/share/opensearch";
+const OPENSEARCH_BASE_PATH: &str = "/stackable/opensearch/";
 
 pub struct RoleGroupBuilder<'a> {
     service_account_name: String,
@@ -107,6 +119,27 @@ impl<'a> RoleGroupBuilder<'a> {
             .data
             .build_pvc(DATA_VOLUME_NAME, Some(vec!["ReadWriteOnce"]));
 
+        let listener_group_name = self.resource_names.listener_service_name();
+
+        // Listener endpoints for the all rolegroups will use persistent
+        // volumes so that load balancers can hard-code the target
+        // addresses. This will be the case even when no class is set (and
+        // the value defaults to cluster-internal) as the address should
+        // still be consistent.
+        let listener_volume_claim_template = ListenerOperatorVolumeSourceBuilder::new(
+            &ListenerReference::ListenerName(listener_group_name),
+            &self
+                .recommended_labels(ProductVersion::from_str("none").expect("version is supplied")),
+        )
+        .expect("should be a listener group name")
+        .build_pvc(LISTENER_VOLUME_NAME.to_string())
+        .expect("should be a valid annotation");
+
+        let pvcs: Option<Vec<PersistentVolumeClaim>> = Some(vec![
+            data_volume_claim_template,
+            listener_volume_claim_template,
+        ]);
+
         let spec = StatefulSetSpec {
             // Order does not matter for OpenSearch
             pod_management_policy: Some("Parallel".to_string()),
@@ -117,7 +150,7 @@ impl<'a> RoleGroupBuilder<'a> {
             },
             service_name: Some(self.resource_names.headless_service_name()),
             template,
-            volume_claim_templates: Some(vec![data_volume_claim_template]),
+            volume_claim_templates: pvcs,
             ..StatefulSetSpec::default()
         };
 
@@ -135,7 +168,7 @@ impl<'a> RoleGroupBuilder<'a> {
         }
 
         let metadata = ObjectMetaBuilder::new()
-            .with_labels(self.recommended_labels())
+            .with_labels(self.recommended_labels(self.cluster.product_version.clone()))
             .with_labels(node_role_labels)
             .build();
 
@@ -271,6 +304,11 @@ impl<'a> RoleGroupBuilder<'a> {
                     name: DATA_VOLUME_NAME.to_owned(),
                     ..VolumeMount::default()
                 },
+                VolumeMount {
+                    mount_path: LISTENER_VOLUME_DIR.to_owned(),
+                    name: LISTENER_VOLUME_NAME.to_owned(),
+                    ..VolumeMount::default()
+                },
             ])
             .expect("The mount paths are statically defined and there should be no duplicates.")
             .add_container_ports(vec![
@@ -337,6 +375,33 @@ impl<'a> RoleGroupBuilder<'a> {
         }
     }
 
+    pub fn build_listener(&self) -> listener::v1alpha1::Listener {
+        let metadata =
+            self.common_metadata(self.resource_names.listener_service_name(), Labels::new());
+
+        let listener_class = self.role_group_config.config.listener_class.to_owned();
+
+        listener::v1alpha1::Listener {
+            metadata,
+            spec: listener::v1alpha1::ListenerSpec {
+                class_name: Some(listener_class),
+                ports: Some(self.listener_ports()),
+                ..listener::v1alpha1::ListenerSpec::default()
+            },
+            status: None,
+        }
+    }
+
+    /// We only use the http port here and intentionally omit
+    /// the metrics one.
+    fn listener_ports(&self) -> Vec<listener::v1alpha1::ListenerPort> {
+        vec![listener::v1alpha1::ListenerPort {
+            name: HTTP_PORT_NAME.to_string(),
+            port: HTTP_PORT.into(),
+            protocol: Some("TCP".to_string()),
+        }]
+    }
+
     fn common_metadata(
         &self,
         resource_name: impl Into<String>,
@@ -350,16 +415,16 @@ impl<'a> RoleGroupBuilder<'a> {
                 None,
                 Some(true),
             ))
-            .with_labels(self.recommended_labels())
+            .with_labels(self.recommended_labels(self.cluster.product_version.clone()))
             .with_labels(extra_labels)
             .build()
     }
 
-    fn recommended_labels(&self) -> Labels {
+    fn recommended_labels(&self, product_version: ProductVersion) -> Labels {
         recommended_labels(
             &self.cluster,
             &self.context_names.product_name,
-            &self.cluster.product_version,
+            &product_version,
             &self.context_names.operator_name,
             &self.context_names.controller_name,
             &ValidatedCluster::role_name(),

rust/operator-binary/src/controller/validate.rs

@@ -122,6 +122,7 @@ fn validate_role_group_config(
         node_roles: merged_role_group.config.config.node_roles,
         resources: merged_role_group.config.config.resources,
         termination_grace_period_seconds,
+        listener_class: merged_role_group.config.config.listener_class,
     };
 
     Ok(RoleGroupConfig {

rust/operator-binary/src/crd/mod.rs

@@ -30,6 +30,8 @@ use crate::framework::{
     role_utils::GenericProductSpecificCommonConfig,
 };
 
+const DEFAULT_LISTENER_CLASS: &str = "cluster-internal";
+
 #[versioned(version(name = "v1alpha1"))]
 pub mod versioned {
 
@@ -130,6 +132,10 @@ pub mod versioned {
 
         #[fragment_attrs(serde(default))]
         pub resources: Resources<StorageConfig>,
+
+        /// This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+        #[serde(default = "default_listener_class")]
+        pub listener_class: String,
     }
 
     #[derive(Clone, Debug, Default, JsonSchema, PartialEq, Fragment)]
@@ -162,6 +168,10 @@ pub mod versioned {
     }
 }
 
+fn default_listener_class() -> String {
+    DEFAULT_LISTENER_CLASS.to_string()
+}
+
 impl HasStatusCondition for v1alpha1::OpenSearchCluster {
     fn conditions(&self) -> Vec<ClusterCondition> {
         match &self.status {
@@ -232,6 +242,7 @@ impl v1alpha1::OpenSearchConfig {
                     },
                 },
             },
+            listener_class: Some(DEFAULT_LISTENER_CLASS.to_string()),
         }
     }
 }

rust/operator-binary/src/framework/role_group_utils.rs

@@ -65,6 +65,16 @@ impl ResourceNames {
 
         format!("{}{SUFFIX}", self.qualified_role_group_name())
     }
+
+    pub fn listener_service_name(&self) -> String {
+        // Compile-time check
+        const _: () = assert!(
+            ResourceNames::MAX_QUALIFIED_ROLE_GROUP_NAME_LENGTH <= MAX_OBJECT_NAME_LENGTH,
+            "The listener name `<cluster_name>-<role_name>-<role_group_name>` must not exceed 253 characters."
+        );
+
+        self.qualified_role_group_name()
+    }
 }
 
 #[cfg(test)]

tests/templates/kuttl/external-access/00-patch-ns.yaml

@@ -0,0 +1,15 @@
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: |
+      kubectl patch namespace $NAMESPACE --patch='
+        {
+          "metadata": {
+            "labels": {
+              "pod-security.kubernetes.io/enforce": "privileged"
+            }
+          }
+        }'
+    timeout: 120