Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file.

### Added

- Add a `containerDebug.enabled` opt-out for SparkApplication submit, driver, and executor containers ([#730]).
- Add support for emitting [OpenLineage](https://openlineage.io/) events from `SparkApplications` via `spec.lineage` ([#726]).
- Support for Spark `4.1.3` and `4.2.0` ([#790]).
- Support floating tags for product images via the new `spec.image.stackableVersionPolicy` field
Expand Down Expand Up @@ -90,6 +91,7 @@ All notable changes to this project will be documented in this file.
- The submit Job and the history server no longer print S3 credentials to the container log ([#796]).

[#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721
[#730]: https://github.com/stackabletech/spark-k8s-operator/issues/730
[#726]: https://github.com/stackabletech/spark-k8s-operator/pull/726
[#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727
[#732]: https://github.com/stackabletech/spark-k8s-operator/pull/732
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,3 +23,25 @@ Application driver pods expose Prometheus metrics at the following endpoints:
* `/metrics/executors/prometheus` for executor instances.

These endpoints are available on the same port as the Spark UI, which is 4040 by default.

== Container diagnostics

The `containerdebug` process runs by default in the submit, driver, and executor containers. To
disable it for a SparkApplication, set `spec.containerDebug.enabled` to `false`:

[source,yaml]
----
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: spark-example
spec:
mode: cluster
mainApplicationFile: local:///path/to/application.py
sparkImage:
productVersion: 3.5.8
containerDebug:
enabled: false
----

Omitting `containerDebug` or `enabled` keeps the process enabled.
18 changes: 18 additions & 0 deletions extra/crds.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,15 @@ spec:
items:
type: string
type: array
containerDebug:
description: Configure the containerdebug process for this application. It is enabled by default.
nullable: true
properties:
enabled:
default: true
description: Whether to run containerdebug. Defaults to `true`.
type: boolean
type: object
deps:
default:
excludePackages: []
Expand Down Expand Up @@ -4865,6 +4874,15 @@ spec:
items:
type: string
type: array
containerDebug:
description: Configure the containerdebug process for this application. It is enabled by default.
nullable: true
properties:
enabled:
default: true
description: Whether to run containerdebug. Defaults to `true`.
type: boolean
type: object
deps:
default:
excludePackages: []
Expand Down
95 changes: 73 additions & 22 deletions rust/operator-binary/src/crd/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,10 @@ pub mod versioned {
#[serde(skip_serializing_if = "Option::is_none")]
pub vector_aggregator_config_map_name: Option<ConfigMapName>,

/// Configure the containerdebug process for this application. It is enabled by default.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub container_debug: Option<ContainerDebugConfig>,

/// The job builds a spark-submit command, complete with arguments and referenced dependencies
/// such as templates, and passes it on to Spark.
/// The reason this property uses its own type (SubmitConfigFragment) is because logging is not
Expand Down Expand Up @@ -273,6 +277,19 @@ pub mod versioned {
pub lineage: Option<SparkLineageConfig>,
}

/// Configuration for the containerdebug process.
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ContainerDebugConfig {
/// Whether to run containerdebug. Defaults to `true`.
#[serde(default = "default_true")]
pub enabled: bool,
}

fn default_true() -> bool {
true
}

/// OpenLineage configuration of a `SparkApplication`: the shared OpenLineage config, plus
/// Spark specific settings.
#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
Expand Down Expand Up @@ -867,16 +884,29 @@ impl v1alpha1::SparkApplication {

submit_cmd.extend(self.spec.args.clone());

Ok(vec![
format!(
let mut commands = Vec::new();
if self.container_debug_enabled() {
commands.push(format!(
"containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &"
),
));
}
commands.extend([
build_truststore_commands,
// xtrace prints expanded credentials, `--verbose` prints the config redacted
"set +x".to_string(),
"echo 'Running \"spark-submit [...]\", which is not shown as it can contain credentials'".to_string(),
submit_cmd.join(" "),
])
]);

Ok(commands)
}

fn container_debug_enabled(&self) -> bool {
self.spec
.container_debug
.as_ref()
.map(|config| config.enabled)
.unwrap_or(true)
}

/// The base environment for the submit, driver and executor containers: the user-defined
Expand All @@ -902,24 +932,20 @@ impl v1alpha1::SparkApplication {
.context(ParseSparkEnvVarNameSnafu)?;
}

// 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.
env = env
.with_value(
&CONTAINERDEBUG_LOG_DIRECTORY,
format!("{VOLUME_MOUNT_PATH_LOG}/containerdebug"),
)
.with_value(
&STACKABLE_PRE_HOOK,
format!(
"containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &"
),
);
if self.container_debug_enabled() {
// These env variables enable the `containerdebug` process in driver and executor pods.
env = env
.with_value(
&CONTAINERDEBUG_LOG_DIRECTORY,
format!("{VOLUME_MOUNT_PATH_LOG}/containerdebug"),
)
.with_value(
&STACKABLE_PRE_HOOK,
format!(
"containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &"
),
);
}

if self.requirements().is_some() {
env = env.with_value(
Expand Down Expand Up @@ -2088,6 +2114,29 @@ spec:
assert!(!command.contains("spark.jars="));
}

#[test]
fn test_containerdebug_is_enabled_by_default_and_can_be_disabled() {
let default_command = build_command_with_openlineage(MINIMAL_APP, "4.1.2");
assert!(default_command.contains("containerdebug --output="));

let disabled_yaml = indoc! {r#"
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: test
namespace: default
spec:
mode: cluster
mainApplicationFile: test.py
sparkImage:
productVersion: 1.2.3
containerDebug:
enabled: false
"#};
let disabled_command = build_command_with_openlineage(disabled_yaml, "4.1.2");
assert!(!disabled_command.contains("containerdebug"));
}

#[test]
fn test_openlineage_appends_to_existing_extra_listeners() {
let yaml = indoc! {r#"
Expand Down Expand Up @@ -2467,6 +2516,8 @@ spec:
mainClass: org.apache.spark.examples.SparkPi
mainApplicationFile: /stackable/spark/examples/jars/spark-examples.jar
vectorAggregatorConfigMapName: vector-aggregator-discovery
containerDebug:
enabled: false
args:
- "--verbose"
- "--input"
Expand Down
47 changes: 47 additions & 0 deletions rust/operator-binary/src/crd/template_merger.rs
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,11 @@ pub fn deep_merge(base: &SparkApplication, overlay: &SparkApplication) -> SparkA
.vector_aggregator_config_map_name
.clone()
.or_else(|| base.spec.vector_aggregator_config_map_name.clone()),
container_debug: overlay
.spec
.container_debug
.clone()
.or_else(|| base.spec.container_debug.clone()),
lineage: overlay
.spec
.lineage
Expand Down Expand Up @@ -826,6 +831,48 @@ mod tests {
);
}

#[test]
fn test_deep_merge_container_debug() {
let base = yaml_serde::from_str::<crate::crd::v1alpha1::SparkApplication>(indoc! {r#"
---
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: base-app
spec:
mode: cluster
mainApplicationFile: base.py
sparkImage:
productVersion: "3.5.8"
containerDebug:
enabled: true
"#})
.unwrap();

let overlay = yaml_serde::from_str::<crate::crd::v1alpha1::SparkApplication>(indoc! {r#"
---
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: overlay-app
spec:
mode: cluster
mainApplicationFile: overlay.py
sparkImage:
productVersion: "4.1.0"
containerDebug:
enabled: false
"#})
.unwrap();

let merged = deep_merge(&base, &overlay);

assert_eq!(
merged.spec.container_debug.map(|config| config.enabled),
Some(false)
);
}

#[test]
fn test_deep_merge_pod_overrides() {
let base = yaml_serde::from_str::<crate::crd::v1alpha1::SparkApplication>(indoc! {r#"
Expand Down
60 changes: 60 additions & 0 deletions rust/operator-binary/src/spark_k8s_controller/build/pod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -610,6 +610,66 @@ mod tests {
assert_eq!(matching[0].value.as_deref(), Some("/custom/log/dir"));
}

#[test]
fn containerdebug_can_be_disabled_for_driver_pods() {
let yaml = indoc! {r#"
apiVersion: spark.stackable.tech/v1alpha1
kind: SparkApplication
metadata:
name: spark-example
namespace: default
spec:
mode: cluster
mainApplicationFile: test.py
sparkImage:
productVersion: 1.2.3
containerDebug:
enabled: false
"#};
let validated = validated_from_yaml(yaml, None);
let driver_config = validated
.spark_application
.driver_config()
.expect("the driver config resolves");
let env = validated
.spark_application
.env(&None, &None, None)
.expect("the base environment can be built");

let template = pod_template(
&validated,
SparkApplicationRole::Driver,
&driver_config,
&[],
&env,
&test_service_account(),
)
.expect("the driver pod template can be built");

let env = template
.spec
.expect("the pod template has a spec")
.containers
.iter()
.find(|container| container.name == "spark")
.expect("the spark container exists")
.env
.as_ref()
.expect("the spark container has env vars");

assert!(
!env.iter()
.any(|env_var| env_var.name == "CONTAINERDEBUG_LOG_DIRECTORY")
);
let pre_hook = env
.iter()
.find(|env_var| env_var.name == "_STACKABLE_PRE_HOOK")
.and_then(|env_var| env_var.value.as_deref())
.expect("the Spark environment bootstrap hook remains configured");
assert!(!pre_hook.contains("containerdebug"));
assert!(pre_hook.contains("spark-env.sh"));
}

/// Runs the real validate step against the given SparkApplication YAML, optionally with a
/// dereferenced S3 connection.
fn validated_from_yaml(
Expand Down