feat: run containerdebug in the background (#508)

* feat: run containerdebug in the background

* various fixes

* reorganize the smoke test files

* update changelog

* fix containerdebug log dir

* successfully ran tests

* reintroduce the CONTAINERDEBUG_LOG_DIRECTORY env var

* less verbose test output

* ensure vol mount exists and remove env duplicates

* Update rust/crd/src/lib.rs

Co-authored-by: Natalie Klestrup Röijezon <[email protected]>

---------

Co-authored-by: Natalie Klestrup Röijezon <[email protected]>

Author: razvan

CHANGELOG.md

@@ -8,8 +8,10 @@ All notable changes to this project will be documented in this file.
 
 - The lifetime of auto generated TLS certificates is now configurable with the role and roleGroup
   config property `requestedSecretLifetime`. This helps reducing frequent Pod restarts ([#501]).
+- Run a `containerdebug` process in the background of each Spark container to collect debugging information ([#508]).
 
 [#501]: https://github.com/stackabletech/spark-k8s-operator/pull/501
+[#508]: https://github.com/stackabletech/spark-k8s-operator/pull/508
 
 ## [24.11.0] - 2024-11-18
 

rust/crd/src/history.rs

@@ -242,6 +242,16 @@ impl SparkHistoryServer {
         let mut vars: BTreeMap<String, EnvVar> = BTreeMap::new();
         let role_env_overrides = &self.role().config.env_overrides;
 
+        // Needed by the `containerdebug` running in the background of the history container
+        // to log it's tracing information to.
+        vars.insert(
+            "CONTAINERDEBUG_LOG_DIRECTORY".to_string(),
+            EnvVar {
+                name: "CONTAINERDEBUG_LOG_DIRECTORY".to_string(),
+                value: Some(format!("{VOLUME_MOUNT_PATH_LOG}/containerdebug")),
+                value_from: None,
+            },
+        );
         // This env var prevents the history server from detaching itself from the
         // start script because this leads to the Pod terminating immediately.
         vars.insert(

rust/crd/src/lib.rs

@@ -326,18 +326,18 @@ impl SparkApplication {
                     .with_config_map(log_config_map)
                     .build(),
             );
-
-            result.push(
-                VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG)
-                    .with_empty_dir(
-                        None::<String>,
-                        Some(product_logging::framework::calculate_log_volume_size_limit(
-                            &[MAX_SPARK_LOG_FILES_SIZE, MAX_INIT_LOG_FILES_SIZE],
-                        )),
-                    )
-                    .build(),
-            );
         }
+        // This volume is also used by the containerdebug process so it must always be there.
+        result.push(
+            VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG)
+                .with_empty_dir(
+                    None::<String>,
+                    Some(product_logging::framework::calculate_log_volume_size_limit(
+                        &[MAX_SPARK_LOG_FILES_SIZE, MAX_INIT_LOG_FILES_SIZE],
+                    )),
+                )
+                .build(),
+        );
 
         if !self.packages().is_empty() {
             result.push(
@@ -466,14 +466,16 @@ impl SparkApplication {
                 mount_path: VOLUME_MOUNT_PATH_LOG_CONFIG.into(),
                 ..VolumeMount::default()
             });
-
-            mounts.push(VolumeMount {
-                name: VOLUME_MOUNT_NAME_LOG.into(),
-                mount_path: VOLUME_MOUNT_PATH_LOG.into(),
-                ..VolumeMount::default()
-            });
         }
 
+        // This is used at least by the containerdebug process.
+        // The volume is always there.
+        mounts.push(VolumeMount {
+            name: VOLUME_MOUNT_NAME_LOG.into(),
+            mount_path: VOLUME_MOUNT_PATH_LOG.into(),
+            ..VolumeMount::default()
+        });
+
         if !self.packages().is_empty() {
             mounts.push(VolumeMount {
                 name: VOLUME_MOUNT_NAME_IVY2.into(),
@@ -527,9 +529,7 @@ impl SparkApplication {
         let mode = &self.spec.mode;
         let name = self.metadata.name.clone().context(ObjectHasNoNameSnafu)?;
 
-        let mut submit_cmd: Vec<String> = vec![];
-
-        submit_cmd.extend(vec![
+        let mut submit_cmd = vec![
             "/stackable/spark/bin/spark-submit".to_string(),
             "--verbose".to_string(),
             "--master k8s://https://${KUBERNETES_SERVICE_HOST}:${KUBERNETES_SERVICE_PORT_HTTPS}".to_string(),
@@ -547,7 +547,7 @@ impl SparkApplication {
             format!("--conf spark.driver.extraClassPath=/stackable/spark/extra-jars/*"),
             format!("--conf spark.executor.defaultJavaOptions=-Dlog4j.configurationFile={VOLUME_MOUNT_PATH_LOG_CONFIG}/{LOG4J2_CONFIG_FILE}"),
             format!("--conf spark.executor.extraClassPath=/stackable/spark/extra-jars/*"),
-        ]);
+        ];
 
         // See https://spark.apache.org/docs/latest/running-on-kubernetes.html#dependency-management
         // for possible S3 related properties
@@ -676,7 +676,10 @@ impl SparkApplication {
 
         submit_cmd.extend(self.spec.args.clone());
 
-        Ok(submit_cmd)
+        Ok(vec![
+            format!("containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &"),
+            submit_cmd.join(" "),
+        ])
     }
 
     pub fn env(
@@ -685,6 +688,27 @@ impl SparkApplication {
         logdir: &Option<ResolvedLogDir>,
     ) -> Vec<EnvVar> {
         let mut e: Vec<EnvVar> = self.spec.env.clone();
+
+        // These env variables enable the `containerdebug` process in driver and executor pods.
+        // More precisely, this process runs in the background of every `spark` container.
+        // - `CONTAINERDEBUG_LOG_DIRECTORY` - is the location where tracing information from the process
+        // is written. This directory is created by the process itself.
+        // - `_STACKABLE_PRE_HOOK` - is evaluated by the entrypoint script (run-spark.sh) in the Spark images
+        // before the actual JVM process is started. The result of this evaluation is that the
+        // `containerdebug` process is executed in the background.
+        e.extend(vec![
+            EnvVar {
+                name: "CONTAINERDEBUG_LOG_DIRECTORY".into(),
+                value: Some(format!("{VOLUME_MOUNT_PATH_LOG}/containerdebug")),
+                value_from: None,
+            },
+            EnvVar {
+                name: "_STACKABLE_PRE_HOOK".into(),
+                value: Some(format!( "containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &")),
+                value_from: None,
+            },
+        ]);
+
         if self.requirements().is_some() {
             e.push(EnvVar {
                 name: "PYTHONPATH".to_string(),
@@ -1385,6 +1409,12 @@ mod tests {
                 name: "executor-pod-template".into(),
                 ..VolumeMount::default()
             },
+            VolumeMount {
+                mount_path: "/stackable/log".into(),
+                mount_propagation: None,
+                name: "log".into(),
+                ..VolumeMount::default()
+            },
             VolumeMount {
                 mount_path: "/kerberos".into(),
                 mount_propagation: None,

rust/operator-binary/src/history/history_controller.rs

@@ -527,7 +527,13 @@ fn build_stateful_set(
         .context(InvalidContainerNameSnafu)?
         .image_from_product_image(resolved_product_image)
         .resources(merged_config.resources.clone().into())
-        .command(vec!["/bin/bash".to_string()])
+        .command(vec![
+            "/bin/bash".to_string(),
+            "-x".to_string(),
+            "-euo".to_string(),
+            "pipefail".to_string(),
+            "-c".to_string(),
+        ])
         .args(command_args(log_dir))
         .add_container_port("http", 18080)
         .add_container_port("metrics", METRICS_PORT.into())
@@ -751,10 +757,10 @@ fn command_args(logdir: &ResolvedLogDir) -> Vec<String> {
     }
 
     command.extend(vec![
+        format!("containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &"),
         format!("/stackable/spark/sbin/start-history-server.sh --properties-file {VOLUME_MOUNT_PATH_CONFIG}/{SPARK_DEFAULTS_FILE_NAME}"),
     ]);
-
-    vec![String::from("-c"), command.join(" && ")]
+    vec![command.join("\n")]
 }
 
 fn labels<'a, T>(

rust/operator-binary/src/spark_k8s_controller.rs

@@ -466,8 +466,14 @@ fn init_containers(
 
             Some(
                 jcb.image(job_image)
-                    .command(vec!["/bin/bash".to_string(), "-c".to_string()])
-                    .args(vec![args.join(" && ")])
+                    .command(vec![
+                        "/bin/bash".to_string(),
+                        "-x".to_string(),
+                        "-euo".to_string(),
+                        "pipefail".to_string(),
+                        "-c".to_string(),
+                    ])
+                    .args(vec![args.join("\n")])
                     .add_volume_mount(VOLUME_MOUNT_NAME_JOB, VOLUME_MOUNT_PATH_JOB)
                     .context(AddVolumeMountSnafu)?
                     .add_volume_mount(VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_PATH_LOG)
@@ -509,8 +515,14 @@ fn init_containers(
             ));
 
             rcb.image(&spark_image.image)
-                .command(vec!["/bin/bash".to_string(), "-c".to_string()])
-                .args(vec![args.join(" && ")])
+                .command(vec![
+                    "/bin/bash".to_string(),
+                    "-x".to_string(),
+                    "-euo".to_string(),
+                    "pipefail".to_string(),
+                    "-c".to_string(),
+                ])
+                .args(vec![args.join("\n")])
                 .add_volume_mount(VOLUME_MOUNT_NAME_REQ, VOLUME_MOUNT_PATH_REQ)
                 .context(AddVolumeMountSnafu)?
                 .add_volume_mount(VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_PATH_LOG)
@@ -549,8 +561,14 @@ fn init_containers(
             }
             Some(
                 tcb.image(&spark_image.image)
-                    .command(vec!["/bin/bash".to_string(), "-c".to_string()])
-                    .args(vec![args.join(" && ")])
+                    .command(vec![
+                        "/bin/bash".to_string(),
+                        "-x".to_string(),
+                        "-euo".to_string(),
+                        "pipefail".to_string(),
+                        "-c".to_string(),
+                    ])
+                    .args(vec![args.join("\n")])
                     .add_volume_mount(STACKABLE_TRUST_STORE_NAME, STACKABLE_TRUST_STORE)
                     .context(AddVolumeMountSnafu)?
                     .resources(
@@ -858,12 +876,17 @@ fn spark_job(
     let mut cb = ContainerBuilder::new(&SparkContainer::SparkSubmit.to_string())
         .context(IllegalContainerNameSnafu)?;
 
-    let args = [job_commands.join(" ")];
     let merged_env = spark_application.merged_env(SparkApplicationRole::Submit, env);
 
     cb.image_from_product_image(spark_image)
-        .command(vec!["/bin/bash".to_string(), "-c".to_string()])
-        .args(vec![args.join(" && ")])
+        .command(vec![
+            "/bin/bash".to_string(),
+            "-x".to_string(),
+            "-euo".to_string(),
+            "pipefail".to_string(),
+            "-c".to_string(),
+        ])
+        .args(vec![job_commands.join("\n")])
         .resources(job_config.resources.clone().into())
         .add_volume_mounts(spark_application.spark_job_volume_mounts(s3conn, logdir))
         .context(AddVolumeMountSnafu)?

tests/templates/kuttl/smoke/03-assert.yaml

@@ -1,13 +1,18 @@
 ---
 apiVersion: kuttl.dev/v1beta1
 kind: TestAssert
+timeout: 900
+---
+apiVersion: apps/v1
+kind: Deployment
 metadata:
-  name: history-api-check
-timeout: 180
+  name: test-minio
+status:
+  readyReplicas: 1
 ---
-apiVersion: batch/v1
-kind: Job
+apiVersion: apps/v1
+kind: Deployment
 metadata:
-  name: history-api-check
+  name: eventlog-minio
 status:
-  succeeded: 1
+  readyReplicas: 1