node_config.rs

//! Configuration of an OpenSearch node

use std::{iter, str::FromStr};

use serde_json::json;
use stackable_operator::{
    builder::pod::container::FieldPathEnvVar,
    commons::networking::DomainName,
    constant,
    k8s_openapi::DeepMerge,
    v2::{
        builder::pod::container::{EnvVarName, EnvVarSet},
        config_overrides::JsonConfigOverrides,
        product_logging::framework::STACKABLE_LOG_DIR,
        role_group_utils::{self, ResourceNames},
        types::{kubernetes::ServiceName, operator::RoleGroupName},
    },
};

use super::ValidatedCluster;
use crate::{
    controller::{
        OpenSearchRoleGroupConfig, ValidatedNodeRole,
        build::role_group_builder::RoleGroupSecurityMode, replicas,
    },
    crd::v1alpha1,
};

/// The main configuration file of OpenSearch
pub const CONFIGURATION_FILE_OPENSEARCH_YML: &str = "opensearch.yml";

/// The cluster name.
/// Type: string
const CONFIG_OPTION_CLUSTER_NAME: &str = "cluster.name";

/// The list of hosts that perform discovery when a node is started.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_DISCOVERY_SEED_HOSTS: &str = "discovery.seed_hosts";

/// By default, OpenSearch forms a multi-node cluster. Set `discovery.type` to `single-node` to
/// form a single-node cluster.
/// Type: string
const CONFIG_OPTION_DISCOVERY_TYPE: &str = "discovery.type";

/// Specifies an address or addresses that an OpenSearch node publishes to other nodes for HTTP
/// communication.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_HTTP_PUBLISH_HOST: &str = "http.publish_host";

/// A list of cluster-manager-eligible nodes used to bootstrap the cluster.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_INITIAL_CLUSTER_MANAGER_NODES: &str = "cluster.initial_cluster_manager_nodes";

/// Binds an OpenSearch node to an address.
/// Type: string
const CONFIG_OPTION_NETWORK_HOST: &str = "network.host";

/// Specifies an address or addresses that an OpenSearch node publishes to other nodes in the
/// cluster so that they can connect to it.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_NETWORK_PUBLISH_HOST: &str = "network.publish_host";

/// The custom node attribute "role-group"
/// Type: string
const CONFIG_OPTION_NODE_ATTR_ROLE_GROUP: &str = "node.attr.role-group";

/// A descriptive name for the node.
/// Type: string
const CONFIG_OPTION_NODE_NAME: &str = "node.name";

/// Defines one or more roles for an OpenSearch node.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_NODE_ROLES: &str = "node.roles";

/// Defines the path for the logs
/// OpenSearch grants the required access rights, see
/// <https://github.com/opensearch-project/OpenSearch/blob/3.4.0/server/src/main/java/org/opensearch/bootstrap/Security.java#L369>
/// The permissions "write" and "delete" are required for the log file rollover.
/// Type: string
const CONFIG_OPTION_PATH_LOGS: &str = "path.logs";

/// If this is set to true, the OpenSearch security plugin will automatically initialize the
/// configuration index with the files in the config directory if the index does not exist.
/// Type: boolean
const CONFIG_OPTION_PLUGINS_SECURITY_ALLOW_DEFAULT_INIT_SECURITYINDEX: &str =
    "plugins.security.allow_default_init_securityindex";

/// Defines the DNs of certificates to which admin privileges should be assigned.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_PLUGINS_SECURITY_AUTHCZ_ADMIN_DN: &str = "plugins.security.authcz.admin_dn";

/// Whether to disable the security plugin
/// Type: boolean
const CONFIG_OPTION_PLUGINS_SECURITY_DISABLED: &str = "plugins.security.disabled";

/// Specifies a list of distinguished names (DNs) that denote the other nodes in the cluster.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_PLUGINS_SECURITY_NODES_DN: &str = "plugins.security.nodes_dn";

/// Whether to enable TLS on the REST layer. If enabled, only HTTPS is allowed.
/// Type: boolean
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_ENABLED: &str = "plugins.security.ssl.http.enabled";

/// Path to the cert PEM file used for TLS on the HTTP PORT.
/// type: string
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_PEMCERT_FILEPATH: &str =
    "plugins.security.ssl.http.pemcert_filepath";

/// Path to the key PEM file used for TLS on the HTTP PORT.
/// type: string
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_PEMKEY_FILEPATH: &str =
    "plugins.security.ssl.http.pemkey_filepath";

/// Path to the trusted CAs PEM file used for TLS on the HTTP PORT.
/// type: string
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_PEMTRUSTEDCAS_FILEPATH: &str =
    "plugins.security.ssl.http.pemtrustedcas_filepath";

/// Whether to enable TLS on internal node-to-node communication using the transport port.
/// type: boolean
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_ENABLED: &str =
    "plugins.security.ssl.transport.enabled";

/// Path to the cert PEM file used for TLS on the transport PORT.
/// type: string
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_PEMCERT_FILEPATH: &str =
    "plugins.security.ssl.transport.pemcert_filepath";

/// Path to the key PEM file used for TLS on the transport PORT.
/// type: string
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_PEMKEY_FILEPATH: &str =
    "plugins.security.ssl.transport.pemkey_filepath";

/// Path to the trusted CAs PEM file used for TLS on the transport PORT.
/// type: string
const CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_PEMTRUSTEDCAS_FILEPATH: &str =
    "plugins.security.ssl.transport.pemtrustedcas_filepath";

/// Specifies an address or addresses that an OpenSearch node publishes to other nodes for
/// transport communication.
/// Type: (comma-separated) list of strings
const CONFIG_OPTION_TRANSPORT_PUBLISH_HOST: &str = "transport.publish_host";

const DEFAULT_OPENSEARCH_HOME: &str = "/stackable/opensearch";

constant!(ENV_VAR_NAME_DISCOVERY_SEED_HOSTS: EnvVarName = CONFIG_OPTION_DISCOVERY_SEED_HOSTS);
constant!(ENV_VAR_NAME_HTTP_PUBLISH_HOST: EnvVarName = CONFIG_OPTION_HTTP_PUBLISH_HOST);
constant!(ENV_VAR_NAME_INITIAL_CLUSTER_MANAGER_NODES: EnvVarName = CONFIG_OPTION_INITIAL_CLUSTER_MANAGER_NODES);
constant!(ENV_VAR_NAME_NETWORK_PUBLISH_HOST: EnvVarName = CONFIG_OPTION_NETWORK_PUBLISH_HOST);
constant!(ENV_VAR_NAME_NODE_NAME: EnvVarName = CONFIG_OPTION_NODE_NAME);
constant!(ENV_VAR_NAME_NODE_ROLES: EnvVarName = CONFIG_OPTION_NODE_ROLES);
constant!(ENV_VAR_NAME_OPENSEARCH_HOME: EnvVarName = "OPENSEARCH_HOME");
constant!(ENV_VAR_NAME_OPENSEARCH_PATH_CONF: EnvVarName = "OPENSEARCH_PATH_CONF");
constant!(ENV_VAR_NAME_POD_NAME: EnvVarName = "_POD_NAME");
constant!(ENV_VAR_NAME_TRANSPORT_PUBLISH_HOST: EnvVarName = CONFIG_OPTION_TRANSPORT_PUBLISH_HOST);

/// Configuration of an OpenSearch node based on the cluster and role-group configuration
pub struct NodeConfig {
    cluster: ValidatedCluster,
    role_group_name: RoleGroupName,
    role_group_config: OpenSearchRoleGroupConfig,
    role_group_security_mode: RoleGroupSecurityMode,
    pub seed_nodes_service_name: ServiceName,
    cluster_domain_name: DomainName,
    headless_service_name: ServiceName,
}

// Most functions are public because their configuration values could also be used in environment
// variables.
impl NodeConfig {
    pub fn new(
        cluster: ValidatedCluster,
        role_group_name: RoleGroupName,
        role_group_config: OpenSearchRoleGroupConfig,
        role_group_security_mode: RoleGroupSecurityMode,
        seed_nodes_service_name: ServiceName,
        cluster_domain_name: DomainName,
        headless_service_name: ServiceName,
    ) -> Self {
        Self {
            cluster,
            role_group_name,
            role_group_config,
            role_group_security_mode,
            seed_nodes_service_name,
            cluster_domain_name,
            headless_service_name,
        }
    }

    /// Creates the main OpenSearch configuration file in YAML format
    pub fn opensearch_config_file_content(&self) -> String {
        serde_yaml::to_string(&self.opensearch_config())
            .expect("serde_json::Value should always be serializable as a string of YAML")
    }

    pub fn opensearch_config(&self) -> serde_json::Value {
        let mut config = self.static_opensearch_config();

        config.merge_from(self.tls_config());

        let overrides: JsonConfigOverrides = self
            .role_group_config
            .config_overrides
            .opensearch_yml
            .clone();

        overrides.apply(&config)
    }

    /// Creates the main OpenSearch configuration file as JSON map
    ///
    /// The file should only contain cluster-wide configuration options. Node-specific options
    /// should be defined as environment variables.
    pub fn static_opensearch_config(&self) -> serde_json::Value {
        let mut config = json!({
            CONFIG_OPTION_CLUSTER_NAME: self.cluster.name,
            // Bind to all interfaces because the IP address is not known in advance.
            CONFIG_OPTION_NETWORK_HOST: "0.0.0.0",
            CONFIG_OPTION_DISCOVERY_TYPE: self.discovery_type(),
            CONFIG_OPTION_PLUGINS_SECURITY_NODES_DN: json!(self.nodes_dn()),
            CONFIG_OPTION_NODE_ATTR_ROLE_GROUP: self.role_group_name,
            CONFIG_OPTION_PATH_LOGS: format!(
                "{STACKABLE_LOG_DIR}/{container}",
                container = v1alpha1::Container::OpenSearch.to_container_name()
            ),
        });

        config.merge_from(match self.role_group_security_mode {
            RoleGroupSecurityMode::Initializing { .. } => {
                json!({
                    CONFIG_OPTION_PLUGINS_SECURITY_ALLOW_DEFAULT_INIT_SECURITYINDEX: true
                })
            }
            RoleGroupSecurityMode::Managing { .. }
            | RoleGroupSecurityMode::Participating { .. } => {
                json!({
                    CONFIG_OPTION_PLUGINS_SECURITY_AUTHCZ_ADMIN_DN: self.super_admin_dn()
                })
            }
            RoleGroupSecurityMode::Disabled => {
                json!({
                    CONFIG_OPTION_PLUGINS_SECURITY_DISABLED: true
                })
            }
        });

        config
    }

    /// Returns the list of distinguished names (DNs) that denote the other nodes in the cluster.
    ///
    /// The list looks similar to:
    /// - DC=local,DC=cluster,DC=svc,DC=my-namespace,DC=opensearch-nodes-cluster-manager-headless,DC=opensearch-nodes-cluster-manager-*
    /// - DC=local,DC=cluster,DC=svc,DC=my-namespace,DC=opensearch-nodes-data-headless,DC=opensearch-nodes-data-*
    /// - CN=generated certificate for pod
    ///
    /// The entry "CN=generated certificate for pod" is still added to make the transition from
    /// SDP 26.3 to 26.7 possible.
    fn nodes_dn(&self) -> Vec<String> {
        self.cluster
            .role_group_configs
            .keys()
            .map(|role_group_name| {
                let resource_names = ResourceNames {
                    cluster_name: self.cluster.name.clone(),
                    role_name: ValidatedCluster::role_name(),
                    role_group_name: role_group_name.clone(),
                };

                self.cluster_domain_name
                    .split('.')
                    .rev()
                    .chain([
                        "svc",
                        self.cluster.namespace.as_ref(),
                        resource_names.headless_service_name().as_ref(),
                        &format!(
                            "{stateful_set_name}-*",
                            stateful_set_name = resource_names.stateful_set_name()
                        ),
                    ])
                    .map(|component| format!("DC={component}"))
                    .collect::<Vec<_>>()
                    .join(",")
            })
            // TODO Remove "CN=generated certificate for pod" after the release of SDP 26.7 and
            // adapt the comment of the function and the tests.
            //
            // tracked in https://github.com/stackabletech/opensearch-operator/issues/145
            .chain(iter::once("CN=generated certificate for pod".to_owned()))
            .collect()
    }

    /// Distinguished name (DN) of the super admin certificate
    pub fn super_admin_dn(&self) -> String {
        // The common name field is limited to 64 characters, see RFC 5280.
        format!("CN=update-security-config.{}", self.cluster.uid)
    }

    pub fn tls_config(&self) -> serde_json::Value {
        let mut config = json!({});

        let opensearch_path_conf = self.opensearch_path_conf();

        if self
            .role_group_security_mode
            .tls_internal_secret_class()
            .is_some()
        {
            config.merge_from(json!({
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_ENABLED: true,
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_PEMCERT_FILEPATH: format!("{opensearch_path_conf}/tls/internal/tls.crt"),
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_PEMKEY_FILEPATH: format!("{opensearch_path_conf}/tls/internal/tls.key"),
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_TRANSPORT_PEMTRUSTEDCAS_FILEPATH: format!("{opensearch_path_conf}/tls/internal/ca.crt"),
            }));
        }

        if self
            .role_group_security_mode
            .tls_server_secret_class()
            .is_some()
        {
            config.merge_from(json!({
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_ENABLED: true,
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_PEMCERT_FILEPATH: format!("{opensearch_path_conf}/tls/server/tls.crt"),
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_PEMKEY_FILEPATH: format!("{opensearch_path_conf}/tls/server/tls.key"),
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_PEMTRUSTEDCAS_FILEPATH: format!("{opensearch_path_conf}/tls/server/ca.crt"),
            }));
        } else {
            config.merge_from(json!({
                CONFIG_OPTION_PLUGINS_SECURITY_SSL_HTTP_ENABLED: false
            }));
        }

        config
    }

    /// Creates environment variables for the OpenSearch configurations
    ///
    /// The environment variables should only contain node-specific configuration options.
    /// Cluster-wide options should be added to the configuration file.
    pub fn environment_variables(&self) -> EnvVarSet {
        let fqdn = format!(
            "$(_POD_NAME).{}.{}.svc.{}",
            self.headless_service_name, self.cluster.namespace, self.cluster_domain_name
        );

        let mut env_vars = EnvVarSet::new()
            .with_field_path(
                // Prefix with an underscore, so that it occurs before the other environment
                // variables which depend on it.
                &ENV_VAR_NAME_POD_NAME,
                &FieldPathEnvVar::Name,
            )
            // Set the OpenSearch node name to the Pod name.
            // The node name is used e.g. for INITIAL_CLUSTER_MANAGER_NODES.
            .with_field_path(
                &ENV_VAR_NAME_NODE_NAME,
                &FieldPathEnvVar::Name,
            )
            .with_value(
                &ENV_VAR_NAME_NETWORK_PUBLISH_HOST,
                &fqdn,
            )
            .with_value(
                &ENV_VAR_NAME_TRANSPORT_PUBLISH_HOST,
                &fqdn,
            )
            .with_value(
                &ENV_VAR_NAME_HTTP_PUBLISH_HOST,
                &fqdn,
            )
            .with_value(
                &ENV_VAR_NAME_DISCOVERY_SEED_HOSTS,
                format!(
                    "{}.{}.svc.{}",
                    self.seed_nodes_service_name, self.cluster.namespace, self.cluster_domain_name
                ),
            )
            .with_value(
                &ENV_VAR_NAME_NODE_ROLES,
                Self::to_comma_separated_list(
                    &self
                        .role_group_config
                        .config
                        .node_roles
                        .iter()
                        .map(|node_role| format!("{node_role}"))
                        .collect::<Vec<_>>(),
                )
                .expect("Node roles cannot contain commas, therefore creating a comma-separated list is safe."),
            );

        if let Some(initial_cluster_manager_nodes) = self.initial_cluster_manager_nodes() {
            env_vars = env_vars.with_value(
                &ENV_VAR_NAME_INITIAL_CLUSTER_MANAGER_NODES,
                initial_cluster_manager_nodes,
            );
        }

        env_vars.merge(self.role_group_config.env_overrides.clone())
    }

    /// Configuration for `discovery.type`
    ///
    /// "zen" is the default if `discovery.type` is not set.
    /// It is nevertheless explicitly set here.
    /// see <https://github.com/opensearch-project/OpenSearch/blob/3.0.0/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java#L88-L89>
    ///
    /// "single-node" disables the bootstrap checks, like validating the JVM and discovery
    /// configurations.
    pub fn discovery_type(&self) -> String {
        if self.cluster.is_single_node() {
            "single-node".to_owned()
        } else {
            "zen".to_owned()
        }
    }

    /// Configuration for `cluster.initial_cluster_manager_nodes`
    ///
    /// Returns the node names of the initial cluster-manager nodes if
    /// * this is a multi-node cluster and
    /// * this node has the cluster-manager node role.
    ///
    /// Please read the following sections for an explanation of these restrictions.
    ///
    /// This configuration setting replaces the setting `cluster.initial_master_nodes`, see
    /// <https://github.com/opensearch-project/OpenSearch/blob/3.4.0/server/src/main/java/org/opensearch/cluster/coordination/ClusterBootstrapService.java#L79-L93>.
    ///
    /// This setting is required on nodes with the cluster-manager node role on a multi-node
    /// cluster. Otherwise the bootstrapping of the cluster fails and all pods report:
    /// > Wait for cluster to be available ...
    ///
    /// This setting must not be set on a single-node cluster, because otherwise the following
    /// error is thrown:
    ///  > setting [cluster.initial_cluster_manager_nodes] is not allowed when [discovery.type] is set to [single-node]
    ///
    /// see <https://github.com/opensearch-project/OpenSearch/blob/3.4.0/server/src/main/java/org/opensearch/cluster/coordination/ClusterBootstrapService.java#L126-L136>
    ///
    /// This setting does not seem to have an effect on nodes without the cluster-manager node
    /// role. However, as it is recommended (see the Elasticsearch documentation below) to not set
    /// it on master-ineligible nodes, it is not set.
    ///
    /// This setting seems to be ignored when the cluster has already formed. It is recommended in
    /// the Elasticsearch documentation to remove it once the cluster has formed, but as it is hard
    /// to determine if the bootstrapping was successfully completed, this setting is still set.
    /// Adding a new cluster-manager node and updating this setting also seems to be okay.
    ///
    /// # OpenSearch documentation
    ///
    /// > This setting is required when bootstrapping a cluster for the first time and should
    /// > contain the node names (as defined by `node.name`) of the initial cluster-manager-eligible
    /// > nodes. This list should be empty for nodes joining an existing cluster.
    ///
    /// see <https://docs.opensearch.org/3.3/install-and-configure/configuring-opensearch/discovery-gateway-settings/#static-discovery-settings>
    ///
    /// # Elasticsearch documentation
    ///
    /// The documentation for Elasticsearch is more detailed and contains the following
    /// notes:
    /// * Remove this setting once the cluster has formed, and never set it again for this cluster.
    /// * Do not configure this setting on master-ineligible nodes.
    /// * Do not configure this setting on nodes joining an existing cluster.
    /// * Do not configure this setting on nodes which are restarting.
    /// * Do not configure this setting when performing a full-cluster restart.
    ///
    /// see <https://www.elastic.co/docs/reference/elasticsearch/configuration-reference/discovery-cluster-formation-settings>
    ///
    /// # Implementation in the OpenSearch Helm chart
    ///
    /// The OpenSearch Helm chart sets this setting on master nodes on multi-node clusters, see
    /// see </home/sigi/projects/stackable/workspace/opensearch-operator/target/doc/stackable_opensearch_operator/index.html>.
    fn initial_cluster_manager_nodes(&self) -> Option<String> {
        if self.cluster.is_single_node()
            || !self
                .role_group_config
                .config
                .node_roles
                .contains(&ValidatedNodeRole::ClusterManager)
        {
            None
        } else {
            let cluster_manager_configs = self
                .cluster
                .role_group_configs_filtered_by_node_role(&ValidatedNodeRole::ClusterManager);

            // This setting requires node names as set in NODE_NAME.
            // The node names are set to the pod names with
            // `valueFrom.fieldRef.fieldPath: metadata.name`, so it is okay to calculate the pod
            // names here and use them as node names.
            let mut pod_names = vec![];
            for (role_group_name, role_group_config) in cluster_manager_configs {
                let role_group_resource_names = role_group_utils::ResourceNames {
                    cluster_name: self.cluster.name.clone(),
                    role_name: ValidatedCluster::role_name(),
                    role_group_name,
                };

                pod_names.extend(
                    (0..replicas(&role_group_config))
                        .map(|i| format!("{}-{i}", role_group_resource_names.stateful_set_name())),
                );
            }

            Some(Self::to_comma_separated_list(&pod_names).expect("Pod names cannot contain commas, therefore creating a comma-separated list is safe."))
        }
    }

    /// Return content of the `OPENSEARCH_HOME` environment variable from envOverrides or default to `DEFAULT_OPENSEARCH_HOME`
    pub fn opensearch_home(&self) -> String {
        self.environment_variables()
            .get(&ENV_VAR_NAME_OPENSEARCH_HOME)
            .and_then(|env_var| env_var.value.clone())
            .unwrap_or(DEFAULT_OPENSEARCH_HOME.to_owned())
    }

    /// Return content of the `OPENSEARCH_PATH_CONF` environment variable from envOverrides or default to `OPENSEARCH_HOME/config`
    pub fn opensearch_path_conf(&self) -> String {
        let opensearch_home = self.opensearch_home();
        self.environment_variables()
            .get(&ENV_VAR_NAME_OPENSEARCH_PATH_CONF)
            .and_then(|env_var| env_var.value.clone())
            .unwrap_or(format!("{opensearch_home}/config"))
    }

    fn to_comma_separated_list(values: &[String]) -> Option<String> {
        if values.iter().any(|value| value.contains(",")) {
            None
        } else if values.is_empty() {
            Some("[]".to_owned())
        } else {
            Some(values.join(","))
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::BTreeMap, str::FromStr};

    use pretty_assertions::assert_eq;
    use stackable_operator::{
        commons::{
            affinity::StackableAffinity,
            product_image_selection::{ProductImage, ResolvedProductImage},
            resources::Resources,
        },
        k8s_openapi::api::core::v1::PodTemplateSpec,
        kvp::LabelValue,
        product_logging::spec::AutomaticContainerLogConfig,
        shared::time::Duration,
        v2::{
            product_logging::framework::ValidatedContainerLogConfigChoice,
            role_utils::GenericCommonConfig,
            types::{
                kubernetes::{
                    ConfigMapKey, ConfigMapName, ListenerClassName, NamespaceName, SecretClassName,
                },
                operator::{ClusterName, ProductVersion},
            },
        },
    };
    use uuid::uuid;

    use super::*;
    use crate::{
        controller::{
            ValidatedLogging, ValidatedOpenSearchConfig, ValidatedOpenSearchConfigOverrides,
            ValidatedSecurity,
        },
        crd::v1alpha1,
    };

    struct TestConfig {
        replicas: u16,
        config_settings: serde_json::Value,
        env_vars: &'static [(&'static str, &'static str)],
    }

    impl Default for TestConfig {
        fn default() -> Self {
            Self {
                replicas: 3,
                config_settings: json!({}),
                env_vars: &[],
            }
        }
    }

    fn node_config(test_config: TestConfig) -> NodeConfig {
        let image: ProductImage = serde_json::from_str(r#"{"productVersion": "3.4.0"}"#)
            .expect("should be a valid ProductImage");

        let role_group_name = RoleGroupName::from_str_unsafe("data");

        let role_group_config = OpenSearchRoleGroupConfig {
            replicas: Some(test_config.replicas),
            config: ValidatedOpenSearchConfig {
                affinity: StackableAffinity::default(),
                discovery_service_exposed: true,
                listener_class: ListenerClassName::from_str_unsafe("cluster-internal"),
                logging: ValidatedLogging {
                    opensearch_container: ValidatedContainerLogConfigChoice::Automatic(
                        AutomaticContainerLogConfig::default(),
                    ),
                    vector_container: None,
                },
                node_roles: [
                    ValidatedNodeRole::ClusterManager,
                    ValidatedNodeRole::Data,
                    ValidatedNodeRole::Ingest,
                    ValidatedNodeRole::RemoteClusterClient,
                ]
                .into(),
                requested_secret_lifetime: Duration::from_str("1d")
                    .expect("should be a valid duration"),
                resources: Resources::default(),
                termination_grace_period_seconds: 30,
            },
            config_overrides: ValidatedOpenSearchConfigOverrides {
                opensearch_yml: JsonConfigOverrides::JsonMergePatch(test_config.config_settings),
            },
            env_overrides: EnvVarSet::new().with_values(
                test_config
                    .env_vars
                    .iter()
                    .map(|(k, v)| (EnvVarName::from_str_unsafe(k), *v)),
            ),
            cli_overrides: BTreeMap::default(),
            pod_overrides: PodTemplateSpec::default(),
            product_specific_common_config: GenericCommonConfig::default(),
        };

        let security_settings = v1alpha1::SecuritySettings {
            config: v1alpha1::SecuritySettingsFileType {
                managed_by: v1alpha1::SecuritySettingsFileTypeManagedBy::Operator,
                content: v1alpha1::SecuritySettingsFileTypeContent::ValueFrom(
                    v1alpha1::SecuritySettingsFileTypeContentValueFrom::ConfigMapKeyRef(
                        v1alpha1::ConfigMapKeyRef {
                            name: ConfigMapName::from_str_unsafe("security-config"),
                            key: ConfigMapKey::from_str_unsafe("config.yml"),
                        },
                    ),
                ),
            },
            ..v1alpha1::SecuritySettings::default()
        };
        let tls_server_secret_class = SecretClassName::from_str_unsafe("tls");
        let tls_internal_secret_class = SecretClassName::from_str_unsafe("tls");

        let validated_security = ValidatedSecurity::ManagedByOperator {
            managing_role_group: role_group_name.clone(),
            settings: security_settings.clone(),
            tls_server_secret_class: tls_server_secret_class.clone(),
            tls_internal_secret_class: tls_internal_secret_class.clone(),
        };

        let cluster = ValidatedCluster::new(
            ResolvedProductImage {
                product_version: "3.4.0".to_owned(),
                app_version_label_value: LabelValue::from_str("3.4.0-stackable0.0.0-dev")
                    .expect("should be a valid label value"),
                image: "oci.stackable.tech/sdp/opensearch:3.4.0-stackable0.0.0-dev".to_string(),
                image_pull_policy: "Always".to_owned(),
                pull_secrets: None,
            },
            ProductVersion::from_str_unsafe(image.product_version()),
            ClusterName::from_str_unsafe("my-opensearch-cluster"),
            NamespaceName::from_str_unsafe("default"),
            uuid!("0b1e30e6-326e-4c1a-868d-ad6598b49e8b"),
            v1alpha1::OpenSearchRoleConfig::default(),
            [(
                RoleGroupName::from_str_unsafe("default"),
                role_group_config.clone(),
            )]
            .into(),
            validated_security,
            vec![],
            None,
        );

        let role_group_security_config = RoleGroupSecurityMode::Managing {
            settings: security_settings.clone(),
            tls_server_secret_class: tls_server_secret_class.clone(),
            tls_internal_secret_class: tls_internal_secret_class.clone(),
        };

        NodeConfig::new(
            cluster,
            role_group_name,
            role_group_config,
            role_group_security_config,
            ServiceName::from_str_unsafe("my-opensearch-seed-nodes"),
            DomainName::from_str("cluster.local").expect("should be a valid domain name"),
            ServiceName::from_str_unsafe("my-opensearch-cluster-default-headless"),
        )
    }

    #[test]
    pub fn test_static_opensearch_config_file() {
        let node_config = node_config(TestConfig {
            config_settings: json!({"test": "value"}),
            ..TestConfig::default()
        });

        assert_eq!(
            concat!(
                "cluster.name: my-opensearch-cluster\n",
                "discovery.type: zen\n",
                "network.host: 0.0.0.0\n",
                "node.attr.role-group: data\n",
                "path.logs: /stackable/log/opensearch\n",
                "plugins.security.authcz.admin_dn: CN=update-security-config.0b1e30e6-326e-4c1a-868d-ad6598b49e8b\n",
                "plugins.security.nodes_dn:\n",
                "- DC=local,DC=cluster,DC=svc,DC=default,DC=my-opensearch-cluster-nodes-default-headless,DC=my-opensearch-cluster-nodes-default-*\n",
                "- CN=generated certificate for pod\n",
                "plugins.security.ssl.http.enabled: true\n",
                "plugins.security.ssl.http.pemcert_filepath: /stackable/opensearch/config/tls/server/tls.crt\n",
                "plugins.security.ssl.http.pemkey_filepath: /stackable/opensearch/config/tls/server/tls.key\n",
                "plugins.security.ssl.http.pemtrustedcas_filepath: /stackable/opensearch/config/tls/server/ca.crt\n",
                "plugins.security.ssl.transport.enabled: true\n",
                "plugins.security.ssl.transport.pemcert_filepath: /stackable/opensearch/config/tls/internal/tls.crt\n",
                "plugins.security.ssl.transport.pemkey_filepath: /stackable/opensearch/config/tls/internal/tls.key\n",
                "plugins.security.ssl.transport.pemtrustedcas_filepath: /stackable/opensearch/config/tls/internal/ca.crt\n",
                "test: value\n",
            )
            .to_owned(),
            node_config.opensearch_config_file_content()
        );
    }

    #[test]
    fn test_constants() {
        // Test that dereferencing the constants does not panic.
        let _ = *ENV_VAR_NAME_DISCOVERY_SEED_HOSTS;
        let _ = *ENV_VAR_NAME_HTTP_PUBLISH_HOST;
        let _ = *ENV_VAR_NAME_INITIAL_CLUSTER_MANAGER_NODES;
        let _ = *ENV_VAR_NAME_NETWORK_PUBLISH_HOST;
        let _ = *ENV_VAR_NAME_NODE_NAME;
        let _ = *ENV_VAR_NAME_NODE_ROLES;
        let _ = *ENV_VAR_NAME_OPENSEARCH_HOME;
        let _ = *ENV_VAR_NAME_OPENSEARCH_PATH_CONF;
        let _ = *ENV_VAR_NAME_POD_NAME;
        let _ = *ENV_VAR_NAME_TRANSPORT_PUBLISH_HOST;
    }

    #[test]
    pub fn test_super_admin_dn() {
        let node_config = node_config(TestConfig::default());

        let super_admin_dn = node_config.super_admin_dn();
        let parts: Vec<&str> = super_admin_dn.split("=").collect();

        assert_eq!(
            vec![
                "CN",
                "update-security-config.0b1e30e6-326e-4c1a-868d-ad6598b49e8b"
            ],
            parts
        );
        assert!(parts[1].len() <= 64);
    }

    #[test]
    pub fn test_environment_variables() {
        let node_config = node_config(TestConfig {
            replicas: 2,
            env_vars: &[("TEST", "value")],
            ..TestConfig::default()
        });

        assert_eq!(
            EnvVarSet::new()
                .with_value(&EnvVarName::from_str_unsafe("TEST"), "value")
                .with_field_path(
                    &EnvVarName::from_str_unsafe("_POD_NAME"),
                    &FieldPathEnvVar::Name
                )
                .with_value(
                    &EnvVarName::from_str_unsafe("cluster.initial_cluster_manager_nodes"),
                    "my-opensearch-cluster-nodes-default-0,my-opensearch-cluster-nodes-default-1",
                )
                .with_value(
                    &EnvVarName::from_str_unsafe("discovery.seed_hosts"),
                    "my-opensearch-seed-nodes.default.svc.cluster.local",
                )
                .with_value(
                    &EnvVarName::from_str_unsafe("http.publish_host"),
                    "$(_POD_NAME).my-opensearch-cluster-default-headless.default.svc.cluster.local",
                )
                .with_value(
                    &EnvVarName::from_str_unsafe("network.publish_host"),
                    "$(_POD_NAME).my-opensearch-cluster-default-headless.default.svc.cluster.local",
                )
                .with_field_path(
                    &EnvVarName::from_str_unsafe("node.name"),
                    &FieldPathEnvVar::Name
                )
                .with_value(
                    &EnvVarName::from_str_unsafe("node.roles"),
                    "cluster_manager,data,ingest,remote_cluster_client"
                )
                .with_value(
                    &EnvVarName::from_str_unsafe("transport.publish_host"),
                    "$(_POD_NAME).my-opensearch-cluster-default-headless.default.svc.cluster.local",
                ),
            node_config.environment_variables()
        );
    }

    #[test]
    pub fn test_discovery_type() {
        let node_config_single_node = node_config(TestConfig {
            replicas: 1,
            ..TestConfig::default()
        });

        let node_config_multiple_nodes = node_config(TestConfig {
            replicas: 2,
            ..TestConfig::default()
        });

        assert_eq!(
            "single-node".to_owned(),
            node_config_single_node.discovery_type()
        );
        assert_eq!(
            "zen".to_owned(),
            node_config_multiple_nodes.discovery_type()
        );
    }

    #[test]
    pub fn test_initial_cluster_manager_nodes() {
        let node_config_single_node = node_config(TestConfig {
            replicas: 1,
            ..TestConfig::default()
        });

        let node_config_multiple_nodes = node_config(TestConfig {
            replicas: 3,
            ..TestConfig::default()
        });

        assert_eq!(
            None,
            node_config_single_node.initial_cluster_manager_nodes()
        );
        assert_eq!(
            Some("my-opensearch-cluster-nodes-default-0,my-opensearch-cluster-nodes-default-1,my-opensearch-cluster-nodes-default-2".to_owned()),
            node_config_multiple_nodes.initial_cluster_manager_nodes()
        );
    }

    #[test]
    pub fn test_to_comma_separated_list() {
        assert_eq!(
            None,
            NodeConfig::to_comma_separated_list(&[
                "one".to_owned(),
                "two,three".to_owned(),
                "four".to_owned()
            ])
        );

        assert_eq!(
            Some("[]".to_owned()),
            NodeConfig::to_comma_separated_list(&[])
        );

        assert_eq!(
            Some("one,two,three".to_owned()),
            NodeConfig::to_comma_separated_list(&[
                "one".to_owned(),
                "two".to_owned(),
                "three".to_owned()
            ])
        );
    }
}