added shutdown test & hbase-entrypoint.sh

Author: razvan

Cargo.nix

@@ -18,7 +18,11 @@ use stackable_operator::{
         fragment::{self, Fragment, ValidationError},
         merge::{Atomic, Merge},
     },
-    k8s_openapi::{api::core::v1::EnvVar, apimachinery::pkg::api::resource::Quantity, DeepMerge},
+    k8s_openapi::{
+        api::core::v1::{EnvVar, EnvVarSource, ObjectFieldSelector},
+        apimachinery::pkg::api::resource::Quantity,
+        DeepMerge,
+    },
     kube::{runtime::reflector::ObjectRef, CustomResource, ResourceExt},
     product_config_utils::Configuration,
     product_logging::{self, spec::Logging},
@@ -1051,6 +1055,14 @@ impl HbaseCluster {
         }
     }
 
+    pub fn service_port(&self, role: &HbaseRole) -> u16 {
+        match role {
+            HbaseRole::Master => HBASE_MASTER_PORT,
+            HbaseRole::RegionServer => HBASE_REGIONSERVER_PORT,
+            _ => HBASE_REST_PORT,
+        }
+    }
+
     /// Name of the port used by the Web UI, which depends on HTTPS usage
     fn ui_port_name(&self) -> String {
         if self.has_https_enabled() {
@@ -1064,20 +1076,36 @@ impl HbaseCluster {
 
 pub fn merged_env(rolegroup_config: Option<&BTreeMap<String, String>>) -> Vec<EnvVar> {
     let merged_env: Vec<EnvVar> = if let Some(rolegroup_config) = rolegroup_config {
-        let env_vars_from_config: BTreeMap<String, EnvVar> = rolegroup_config
+        let mut env_vars_from_config: Vec<EnvVar> = rolegroup_config
             .iter()
-            .map(|(env_name, env_value)| {
-                (
-                    env_name.clone(),
-                    EnvVar {
-                        name: env_name.clone(),
-                        value: Some(env_value.to_owned()),
-                        value_from: None,
-                    },
-                )
+            .map(|(env_name, env_value)| EnvVar {
+                name: env_name.clone(),
+                value: Some(env_value.to_owned()),
+                value_from: None,
             })
             .collect();
-        env_vars_from_config.into_values().collect()
+
+        // Needed by the hbase-entrypoint.sh script
+        env_vars_from_config.push(EnvVar {
+            name: "NAMESPACE".to_string(),
+            value: None,
+            value_from: Some(EnvVarSource {
+                field_ref: Some(ObjectFieldSelector {
+                    field_path: "metadata.namespace".to_string(),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            }),
+        });
+
+        // Needed by the hbase-entrypoint.sh script
+        env_vars_from_config.push(EnvVar {
+            name: "REGION_MOVER_OPTS".to_string(),
+            value: None,
+            value_from: None,
+        });
+
+        env_vars_from_config
     } else {
         vec![]
     };

rust/crd/src/lib.rs

@@ -110,7 +110,7 @@ handle_term_signal()
     if [ "${term_child_pid}" ]; then
         if [ -n "$PRE_SHUTDOWN_COMMAND" ]; then
             echo "Start pre-shutdown command: $PRE_SHUTDOWN_COMMAND"
-            source "$PRE_SHUTDOWN_COMMAND"
+            $(${PRE_SHUTDOWN_COMMAND})
             echo "Done pre-shutdown command"
         fi
         kill -TERM "${term_child_pid}" 2>/dev/null
@@ -853,38 +853,12 @@ fn build_rolegroup_statefulset(
     let mut hbase_container = ContainerBuilder::new("hbase").expect("ContainerBuilder not created");
     hbase_container
         .image_from_product_image(resolved_product_image)
-        .command(vec![
-            "/bin/bash".to_string(),
-            "-x".to_string(),
-            "-euo".to_string(),
-            "pipefail".to_string(),
-            "-c".to_string(),
+        .command(vec!["/stackable/hbase/bin/hbase-entrypoint.sh".to_string()])
+        .args(vec![
+            hbase_role.cli_role_name(),
+            rolegroup_ref.object_name(),
+            hbase.service_port(hbase_role).to_string(),
         ])
-        .args(vec![formatdoc! {"
-            mkdir -p {CONFIG_DIR_NAME}
-            cp {HDFS_DISCOVERY_TMP_DIR}/hdfs-site.xml {CONFIG_DIR_NAME}
-            cp {HDFS_DISCOVERY_TMP_DIR}/core-site.xml {CONFIG_DIR_NAME}
-            cp {HBASE_CONFIG_TMP_DIR}/* {CONFIG_DIR_NAME}
-            cp {HBASE_LOG_CONFIG_TMP_DIR}/{log4j_properties_file_name} {CONFIG_DIR_NAME}
-
-            {kerberos_container_start_commands}
-
-            PRE_SHUTDOWN_COMMAND={pre_shutdown_command}
-            {HBASE_BASH_TRAP_FUNCTIONS}
-            {remove_vector_shutdown_file_command}
-            prepare_signal_handlers
-            bin/hbase {hbase_role_name_in_command} start &
-            wait_for_termination $!
-            {create_vector_shutdown_file_command}
-            ",
-            pre_shutdown_command=config.pre_shutdown_command(),
-            hbase_role_name_in_command = hbase_role.cli_role_name(),
-            kerberos_container_start_commands = kerberos_container_start_commands(hbase),
-            remove_vector_shutdown_file_command =
-                remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
-            create_vector_shutdown_file_command =
-                create_vector_shutdown_file_command(STACKABLE_LOG_DIR),
-        }])
         .add_env_vars(merged_env)
         .add_volume_mount("hbase-config", HBASE_CONFIG_TMP_DIR)
         .add_volume_mount("hdfs-discovery", HDFS_DISCOVERY_TMP_DIR)

rust/operator-binary/src/hbase_controller.rs

@@ -95,11 +95,18 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
     parser.add_argument(
         "--operator",
         help="Patch operator version in release.yaml. Format <operator>=<version>",
-        nargs="*",
+        action="append",
         type=cli_parse_operator_args,
         default=[],
     )
 
+    parser.add_argument(
+        "--skip-operator",
+        help="Skip given operator(s) when installing a release.",
+        action="append",
+        default=[],
+    )
+
     parser.add_argument(
         "--test",
         help="Kuttl test to run.",
@@ -138,7 +145,7 @@ def cli_parse_operator_args(args: str) -> tuple[str, str]:
             f"Invalid operator argument: {args}. Must be in format <operator>=<version>"
         )
     op, version = args.split("=", maxsplit=1)
-    return (op, version)
+    return op, version
 
 
 def cli_log_level(cli_arg: str) -> int:
@@ -179,11 +186,13 @@ def have_requirements() -> None:
 
 @contextlib.contextmanager
 def release_file(
-    operators: list[tuple[str, str]] = [],
+    operators: list[tuple[str, str]], skip_ops: list[str]
 ) -> collections.abc.Generator[str, None, None]:
-    """Patch release.yaml with operator versions if needed.
+    """Generate a (possibly modified) copy of the release.yaml file.
 
-    If no --operator is set, the default release file is used.
+    Operator versions passed as --operator take precedence over the release.yaml contents.
+
+    Operators passed as --skip-operator are excluded from the resulting release.yaml contents.
 
     If an invalid operator name is provided (i.e. one that doesn't exist in the
     original release file), a TestRunnerException is raised.
@@ -194,36 +203,60 @@ def release_file(
 
     def _patch():
         release_file = os.path.join("tests", "release.yaml")
-        # Make a copy so we can mutate it without affecting the original
-        ops_copy = operators.copy()
+        # A marker to validate that all ops were patched
         patched_release = []
         with open(release_file, "r") as f:
+            patched_ops = []
             patch_version = ""
             for line in f:
                 if patch_version:
                     line = re.sub(":.+$", f": {patch_version}", line)
                     patch_version = ""
                 else:
-                    for op, version in ops_copy:
+                    for op, version in operators:
                         if op in line:
                             patch_version = version
-                            ops_copy.remove((op, version))  # found an operator to patch
+                            patched_ops.append(op)
                             break
-                patched_release.append(line)
-        if ops_copy:
-            # Some --operator args were not found in the release file. This is
-            # most likely a typo and CI pipelines should terminate early in such
-            # cases.
+                patched_release.append(line.rstrip("\n"))
+
+            # Sanity test that cli didn't contain garbage that is silently discarded
+            ops_not_patched = set([op for op, _ in operators]) - set(patched_ops)
+            if ops_not_patched:
+                logging.error(
+                    f"Patched operators [{', '.join(ops_not_patched)}] not found in {release_file}"
+                )
+                raise TestRunnerException()
+
+        # Filter out skip operators
+        release_contents = []
+        skip = False
+        valid_skip_ops = []
+        for line in patched_release:
+            if skip:
+                skip = False
+                continue
+            for op in skip_ops:
+                if op in line:
+                    skip = True
+                    valid_skip_ops.append(op)
+                    break
+            else:
+                release_contents.append(line)
+        # Sanity test that cli didn't contain garbage that is silently discarded
+        ops_not_skipped = set(skip_ops) - set(valid_skip_ops)
+        if ops_not_skipped:
             logging.error(
-                f"Operators {', '.join([op for op, _ in ops_copy])} not found in {release_file}"
+                f"Skipped operators [{', '.join(ops_not_skipped)}] not found in {release_file}"
             )
             raise TestRunnerException()
+
         with tempfile.NamedTemporaryFile(
             mode="w",
             delete=False,
             prefix="patched",
         ) as f:
-            pcontents = "".join(patched_release)
+            pcontents = "\n".join(release_contents)
             logging.debug(f"Writing patched release to {f.name}: {pcontents}\n")
             f.write(pcontents)
             return f.name
@@ -353,7 +386,7 @@ def main(argv) -> int:
         logging.basicConfig(encoding="utf-8", level=opts.log_level)
         have_requirements()
         gen_tests(opts.test_suite)
-        with release_file(opts.operator) as f:
+        with release_file(opts.operator, opts.skip_operator) as f:
             maybe_install_release(opts.skip_release, f)
         if opts.skip_tests:
             logging.info("Skip running tests.")

scripts/run-tests

@@ -0,0 +1,11 @@
+---
+apiVersion: v1
+kind: LimitRange
+metadata:
+  name: limit-request-ratio
+spec:
+  limits:
+    - type: "Container"
+      maxLimitRequestRatio:
+        cpu: 5
+        memory: 1

tests/templates/kuttl/shutdown/00-limit-range.yaml

@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}

tests/templates/kuttl/shutdown/00-patch-ns.yaml.j2

@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}

tests/templates/kuttl/shutdown/01-assert.yaml.j2

@@ -0,0 +1,9 @@
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+{% endif %}

tests/templates/kuttl/shutdown/01-install-vector-aggregator-discovery-configmap.yaml.j2

@@ -0,0 +1,14 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: install-zk
+timeout: 600
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-zk-server-default
+status:
+  readyReplicas: 1
+  replicas: 1

tests/templates/kuttl/shutdown/10-assert.yaml

@@ -0,0 +1,29 @@
+---
+apiVersion: zookeeper.stackable.tech/v1alpha1
+kind: ZookeeperCluster
+metadata:
+  name: test-zk
+spec:
+  image:
+    productVersion: "{{ test_scenario['values']['zookeeper-latest'] }}"
+    pullPolicy: IfNotPresent
+  clusterConfig:
+    listenerClass: "cluster-internal"
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+    vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  servers:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+    roleGroups:
+      default:
+        replicas: 1
+---
+apiVersion: zookeeper.stackable.tech/v1alpha1
+kind: ZookeeperZnode
+metadata:
+  name: test-znode
+spec:
+  clusterRef:
+    name: test-zk