Skip to content
Draft
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
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ All notable changes to this project will be documented in this file.
that their Pods are automatically rolled when a mounted ConfigMap or Secret changes ([#754]).
- Add `/ready` endpoint to the operator Deployment, which reports the CRD installation status ([#779]).
- Add support for Spark 4.x and HBase connectors ([#797]).
- The logs of the `containerdebug` process are now collected by the Vector agent and forwarded to
the Vector aggregator ([#802]).

### Changed

Expand Down Expand Up @@ -88,6 +90,10 @@ All notable changes to this project will be documented in this file.
executors. With the default `Always`, these containers pull on every start instead of using the
node's image cache ([#764]).
- The submit Job and the history server no longer print S3 credentials to the container log ([#796]).
- The `containerdebug` process writes its log file again. The operator set
`CONTAINERDEBUG_LOG_DIRECTORY`, which containerdebug stopped reading in its version 0.2.0 when it
moved to `stackable-telemetry`; the directory is now set via `FILE_LOG_DIRECTORY`. The file is
rotated hourly and at most two files are kept, because rotation is opt-in ([#802]).

[#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721
[#726]: https://github.com/stackabletech/spark-k8s-operator/pull/726
Expand Down Expand Up @@ -116,6 +122,7 @@ All notable changes to this project will be documented in this file.
[#795]: https://github.com/stackabletech/spark-k8s-operator/pull/795
[#796]: https://github.com/stackabletech/spark-k8s-operator/pull/796
[#797]: https://github.com/stackabletech/spark-k8s-operator/pull/797
[#802]: https://github.com/stackabletech/spark-k8s-operator/pull/802

## [26.7.0] - 2026-07-21

Expand Down
6 changes: 3 additions & 3 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

13 changes: 9 additions & 4 deletions Cargo.nix

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 8 additions & 0 deletions docs/modules/spark-k8s/pages/usage-guide/logging.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,14 @@ It is the user's responsibility to install and configure the vector aggregator,

NOTE: Only logs produced by the application's driver and executors are collected. Logs produced by `spark-submit` are discarded.

NOTE: The logs of the xref:concepts:observability/containerdebug.adoc[`containerdebug` process] are collected as well, from the driver and executor Pods and from the history server.
They are labelled `container: containerdebug`, although no such Kubernetes container exists.
The process runs in the background of the `spark` and `spark-history` containers.
It also runs in the `spark-submit` Pod, whose logs are discarded as described above.
The label comes from `/stackable/log/containerdebug`, the directory which the operator sets with the `FILE_LOG_DIRECTORY` environment variable and from which the Vector agent derives the container name.
Overriding `FILE_LOG_DIRECTORY` with `envOverrides` therefore changes how these logs are aggregated.
The agent only collects files one level below `/stackable/log`, so a sibling directory such as `/stackable/log/spark` relabels the events instead of dropping them, while any deeper or shallower path stops the aggregation.

== History server

The following snippet shows how to configure log aggregation for the history server:
Expand Down
1 change: 1 addition & 0 deletions rust/operator-binary/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ yaml_serde.workspace = true
[dev-dependencies]
indoc.workspace = true
rstest.workspace = true
stackable-operator = { workspace = true, features = ["test-support"] }

[build-dependencies]
built.workspace = true
48 changes: 21 additions & 27 deletions rust/operator-binary/src/connect/controller/build/executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,10 +38,10 @@ use crate::{
s3,
},
crd::constants::{
CONTAINERDEBUG_LOG_DIRECTORY, JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE,
MAX_SPARK_LOG_FILES_SIZE, METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE,
VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG,
VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE,
METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE, VOLUME_MOUNT_NAME_CONFIG,
VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG,
VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
},
product_logging,
};
Expand Down Expand Up @@ -93,20 +93,28 @@ pub fn executor_pod_template(
let config = &validated.executor_config;
let resolved_product_image = &validated.resolved_product_image;
let resolved_s3 = &validated.cluster_config.resolved_s3;
let container_env = executor_env(&validated.executor_overrides.env_overrides);
// The operator sets no environment variables on the executor container, so the user's
// `envOverrides` are all there is. `FILE_LOG_DIRECTORY` in particular would have no effect,
// because `containerdebug` is not started in the executor containers.
let container_env = validated.executor_overrides.env_overrides.clone();

let (s3_volumes, s3_volume_mounts) = resolved_s3
.volumes_and_mounts()
.context(BuildS3VolumesAndMountsSnafu)?;

let mut container = new_container_builder(SparkConnectContainer::Spark.name());
container
.add_env_vars(container_env)
.add_volume_mount(VOLUME_MOUNT_NAME_CONFIG.as_ref(), VOLUME_MOUNT_PATH_CONFIG)
.expect("The mount paths are statically defined and there should be no duplicates.")
.add_volume_mount(VOLUME_MOUNT_NAME_LOG.as_ref(), VOLUME_MOUNT_PATH_LOG)
.expect("The mount paths are statically defined and there should be no duplicates.");

// `add_env_vars` creates the `env` list even for an empty set, which would show up as
// `env: []` in the rendered pod template.
if container_env.iter().next().is_some() {
container.add_env_vars(container_env);
}

let metadata = ObjectMetaBuilder::new()
.with_labels(recommended_labels_for_role_resources(
validated,
Expand Down Expand Up @@ -213,19 +221,6 @@ pub fn executor_pod_template(
Ok(result)
}

/// The environment variables of the executor container.
///
/// The user's `envOverrides` are merged in last so that they override any operator-set
/// environment variable.
fn executor_env(env_overrides: &EnvVarSet) -> EnvVarSet {
EnvVarSet::new()
.with_value(
&CONTAINERDEBUG_LOG_DIRECTORY,
format!("{VOLUME_MOUNT_PATH_LOG}/containerdebug"),
)
.merge(env_overrides.clone())
}

pub(crate) fn executor_properties(
validated: &ValidatedSparkConnectServer,
) -> Result<BTreeMap<String, Option<String>>, Error> {
Expand Down Expand Up @@ -384,6 +379,7 @@ mod tests {
use stackable_operator::{
commons::product_image_selection::PullPolicy,
k8s_openapi::{api::core::v1::EnvVar, apimachinery::pkg::apis::meta::v1::ObjectMeta},
v2::builder::pod::container::EnvVarName,
};

use super::*;
Expand Down Expand Up @@ -430,15 +426,13 @@ mod tests {
);
}

/// `envOverrides` must be applied after all operator-set environment variables, so a user
/// override replaces the operator-set value instead of duplicating it or being ignored.
/// The user's `envOverrides` must reach the executor container exactly once. The operator
/// sets no environment variables there, so there is no operator-set value to replace.
#[test]
fn env_overrides_override_operator_set_env_vars() {
fn env_overrides_are_applied_to_the_executor_container() {
let mut validated = minimal_validated_cluster();
validated.executor_overrides.env_overrides = EnvVarSet::new().with_value(
&"CONTAINERDEBUG_LOG_DIRECTORY"
.parse()
.expect("valid env var name"),
&EnvVarName::from_str_unsafe("FILE_LOG_DIRECTORY"),
"/custom/log/dir",
);

Expand Down Expand Up @@ -466,10 +460,10 @@ mod tests {

let matching: Vec<&EnvVar> = env
.iter()
.filter(|env_var| env_var.name == "CONTAINERDEBUG_LOG_DIRECTORY")
.filter(|env_var| env_var.name == "FILE_LOG_DIRECTORY")
.collect();

// The override must replace the operator-set value, not duplicate it.
// Exactly once, so a future operator-set value cannot end up duplicated next to it.
assert_eq!(matching.len(), 1);
assert_eq!(matching[0].value.as_deref(), Some("/custom/log/dir"));
}
Expand Down
27 changes: 18 additions & 9 deletions rust/operator-binary/src/connect/controller/build/server.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use std::{collections::BTreeMap, str::FromStr};

use const_format::concatcp;
use indoc::formatdoc;
use snafu::{OptionExt, ResultExt, Snafu};
use stackable_operator::{
Expand Down Expand Up @@ -60,12 +61,12 @@ use crate::{
},
crd::{
constants::{
CONTAINERDEBUG_LOG_DIRECTORY, JVM_SECURITY_PROPERTIES_FILE, LISTENER_VOLUME_DIR,
LISTENER_VOLUME_NAME, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE,
METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME,
SPARK_NO_DAEMONIZE, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG,
VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG,
VOLUME_MOUNT_PATH_LOG_CONFIG,
FILE_LOG_DIRECTORY, FILE_LOG_MAX_FILES, FILE_LOG_ROTATION_PERIOD,
JVM_SECURITY_PROPERTIES_FILE, LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME,
LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE, METRICS_PROPERTIES_FILE,
POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME, SPARK_NO_DAEMONIZE,
VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG,
VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
},
listener_ext,
},
Expand Down Expand Up @@ -207,6 +208,9 @@ pub(crate) fn build_stateful_set(
VolumeBuilder::new(VOLUME_MOUNT_NAME_LOG.as_ref())
.with_empty_dir(
None::<String>,
// Only the Spark log files are budgeted. The `containerdebug` files in this
// volume are rotated by time, not by size, so they have no budget; raising
// `FILE_LOG_LEVEL` via `envOverrides` can therefore evict the Pod.
Some(calculate_log_volume_size_limit(&[MAX_SPARK_LOG_FILES_SIZE])),
)
.build(),
Expand Down Expand Up @@ -363,6 +367,7 @@ pub(crate) fn build_stateful_set(
#[allow(clippy::result_large_err)]
pub(crate) fn command_args(user_args: &[String]) -> Vec<String> {
let mut command = formatdoc! { "
mkdir --parents \"$FILE_LOG_DIRECTORY\"
containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &

cp {VOLUME_MOUNT_PATH_CONFIG}/{SPARK_DEFAULTS_FILE_NAME} /tmp/spark.properties
Expand All @@ -387,9 +392,12 @@ pub(crate) fn command_args(user_args: &[String]) -> Vec<String> {
fn env(env_overrides: &EnvVarSet) -> EnvVarSet {
EnvVarSet::new()
.with_value(
&CONTAINERDEBUG_LOG_DIRECTORY,
format!("{VOLUME_MOUNT_PATH_LOG}/containerdebug"),
&FILE_LOG_DIRECTORY,
concatcp!(VOLUME_MOUNT_PATH_LOG, "/containerdebug"),
)
// Rotation is opt-in, so without these the file would grow unbounded.
.with_value(&FILE_LOG_ROTATION_PERIOD, "hourly")
.with_value(&FILE_LOG_MAX_FILES, "2")
.with_value(&SPARK_NO_DAEMONIZE, "true")
.merge(env_overrides.clone())
}
Expand Down Expand Up @@ -545,6 +553,7 @@ mod tests {
use stackable_operator::{
commons::product_image_selection::PullPolicy,
k8s_openapi::{api::core::v1::EnvVar, apimachinery::pkg::apis::meta::v1::ObjectMeta},
v2::builder::pod::container::EnvVarName,
};

use super::*;
Expand Down Expand Up @@ -600,7 +609,7 @@ mod tests {
fn env_overrides_override_operator_set_env_vars() {
let mut validated = minimal_validated_cluster();
validated.server_overrides.env_overrides = EnvVarSet::new().with_value(
&"SPARK_NO_DAEMONIZE".parse().expect("valid env var name"),
&EnvVarName::from_str_unsafe("SPARK_NO_DAEMONIZE"),
"overridden",
);

Expand Down
14 changes: 10 additions & 4 deletions rust/operator-binary/src/crd/constants.rs
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,13 @@ pub const VOLUME_MOUNT_PATH_LOG_CONFIG: &str = "/stackable/log_config";
constant!(pub VOLUME_MOUNT_NAME_LOG: VolumeName = "log");
pub const VOLUME_MOUNT_PATH_LOG: &str = "/stackable/log";

// Tells the `containerdebug` process running in the background of a product container where to
// write its tracing information.
constant!(pub CONTAINERDEBUG_LOG_DIRECTORY: EnvVarName = "CONTAINERDEBUG_LOG_DIRECTORY");
// Let the `containerdebug` process write its tracing information to a file. These are the generic
// `stackable-telemetry` variables and apply to the whole product container, so any other
// telemetry-based process in it would log into the same directory and the Vector agent would label
// its events with `container: containerdebug`.
constant!(pub FILE_LOG_DIRECTORY: EnvVarName = "FILE_LOG_DIRECTORY");
constant!(pub FILE_LOG_MAX_FILES: EnvVarName = "FILE_LOG_MAX_FILES");
constant!(pub FILE_LOG_ROTATION_PERIOD: EnvVarName = "FILE_LOG_ROTATION_PERIOD");
// Prevents the history/connect server from detaching itself from the start script, which would
// terminate the Pod immediately.
constant!(pub SPARK_NO_DAEMONIZE: EnvVarName = "SPARK_NO_DAEMONIZE");
Expand Down Expand Up @@ -193,7 +197,9 @@ mod tests {
#[test]
fn test_constants() {
// Test that dereferencing the constants does not panic.
let _ = *CONTAINERDEBUG_LOG_DIRECTORY;
let _ = *FILE_LOG_DIRECTORY;
let _ = *FILE_LOG_MAX_FILES;
let _ = *FILE_LOG_ROTATION_PERIOD;
let _ = *LISTENER_VOLUME_NAME;
let _ = *SPARK_NO_DAEMONIZE;
let _ = *VOLUME_MOUNT_NAME_CONFIG;
Expand Down
3 changes: 1 addition & 2 deletions rust/operator-binary/src/crd/history.rs
Original file line number Diff line number Diff line change
Expand Up @@ -317,7 +317,6 @@ fn default_listener_class() -> ListenerClassName {

#[cfg(test)]
mod test {
use std::str::FromStr;

use indoc::indoc;
use stackable_operator::{
Expand Down Expand Up @@ -420,7 +419,7 @@ mod test {

assert_eq!(
env_overrides
.get(&EnvVarName::from_str("TEST_SPARK_HIST_VAR").expect("valid env var name"))
.get(&EnvVarName::from_str_unsafe("TEST_SPARK_HIST_VAR"))
.and_then(|env_var| env_var.value.clone()),
Some("ROLEGROUP".to_string())
);
Expand Down
Loading
Loading