added listener dirs and service

Author: adwk67

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:

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,20 @@
 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},
@@ -36,8 +43,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 +117,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),
+            // TODO should be unversioned
+            &self.recommended_labels(),
+        )
+        .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 +148,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()
         };
 
@@ -271,6 +302,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 +373,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>,

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("cluster-internal".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)]