From 739b49a9ae9ab15eebdcc92ed52a3a811de0a296 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Thu, 9 Jul 2026 16:33:53 +0200 Subject: [PATCH 1/6] feat: add basic openlineage support --- CHANGELOG.md | 2 + .../pages/usage-guide/openlineage.adoc | 73 +++ docs/modules/spark-k8s/partials/nav.adoc | 1 + extra/crds.yaml | 72 +++ rust/operator-binary/src/config/jvm.rs | 42 +- rust/operator-binary/src/crd/constants.rs | 24 + rust/operator-binary/src/crd/mod.rs | 422 +++++++++++++++++- .../src/crd/template_merger.rs | 5 + rust/operator-binary/src/main.rs | 36 +- .../src/spark_k8s_controller.rs | 118 ++++- 10 files changed, 773 insertions(+), 22 deletions(-) create mode 100644 docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc diff --git a/CHANGELOG.md b/CHANGELOG.md index e198c570..f9405469 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ All notable changes to this project will be documented in this file. - BREAKING: Add required CLI argument and env var to set the image repository used to construct final product image names: `IMAGE_REPOSITORY` (`--image-repository`), eg. `oci.example.org/my/namespace` ([#684]). - Support for Spark `4.1.2` ([#708]) +- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap), a stable job name, and the required `--add-opens` JVM flag ([#XXXX]). ### Fixed @@ -44,6 +45,7 @@ All notable changes to this project will be documented in this file. [#705]: https://github.com/stackabletech/spark-k8s-operator/pull/705 [#708]: https://github.com/stackabletech/spark-k8s-operator/pull/708 [#716]: https://github.com/stackabletech/spark-k8s-operator/pull/716 +[#XXXX]: https://github.com/stackabletech/spark-k8s-operator/pull/XXXX ## [26.3.0] - 2026-03-16 diff --git a/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc new file mode 100644 index 00000000..569311d5 --- /dev/null +++ b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc @@ -0,0 +1,73 @@ += OpenLineage + +https://openlineage.io/[OpenLineage] is an open standard for data lineage collection. +The Spark operator can automatically emit lineage events for a `SparkApplication` to an OpenLineage-compatible backend such as https://marquezproject.github.io/marquez/[Marquez], without any manual `sparkConf` wiring. + +When enabled, the operator injects everything required to make the https://openlineage.io/docs/integrations/spark/[OpenLineage Spark listener] work: + +* the OpenLineage Spark listener (appended to `spark.extraListeners`, so any listener you configure yourself is preserved), +* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg), +* an HTTP transport pointing at the backend endpoint, +* a stable lineage namespace and job name (see <>), and +* the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors. + +== Enabling OpenLineage + +The backend endpoint is not configured on the `SparkApplication` directly. +Instead, mirroring the xref:usage-guide/logging.adoc[vector aggregator discovery], the operator reads it from a discovery ConfigMap referenced by `configMapName`: + +[source,yaml] +---- +apiVersion: spark.stackable.tech/v1alpha1 +kind: SparkApplication +metadata: + name: orders-by-nation +spec: + openLineage: + enabled: true # <1> + configMapName: marquez-discovery # <2> + # namespace: orders-lineage # <3> + # appName: orders-by-nation # <4> + ... +---- +<1> Enable OpenLineage event emission. Defaults to `false`, in which case the operator emits nothing OpenLineage-related and behaviour is unchanged. +<2> Name of a discovery ConfigMap holding the backend endpoint (see below). + Must be in the same namespace as the `SparkApplication`. +<3> Optional. The OpenLineage namespace events are reported under. + Defaults to the application's Kubernetes namespace. +<4> Optional. The stable OpenLineage job name (see <>). + +The discovery ConfigMap holds the backend base URL under the `ADDRESS` key: + +.Example discovery ConfigMap +[source,yaml] +---- +apiVersion: v1 +kind: ConfigMap +metadata: + name: marquez-discovery +data: + ADDRESS: http://marquez:5000 # <1> +---- +<1> Base URL of the OpenLineage backend, for example the Marquez API service. + +Alternatively, you can set the endpoint directly through `sparkConf` (`spark.openlineage.transport.url`). +If `enabled: true` and no endpoint can be resolved from either the discovery ConfigMap or `sparkConf`, reconciliation fails with a clear error rather than emitting a config that silently drops events. + +[#job-name] +== Job name + +The OpenLineage job name is resolved in the following order: + +. `spec.openLineage.appName`, if set. +. `spark.app.name` from `sparkConf`, if set. +. `metadata.name` as a last resort. + +The last case additionally emits a Kubernetes warning event. +Falling back to `metadata.name` is discouraged: if that name carries a timestamp or other run-specific suffix (as generated names often do), every run becomes a separate job in the backend and the run history fragments. +Set `spec.openLineage.appName` (or `spark.app.name`) to a stable value to keep all runs of the same logical job together. + +== Overriding injected configuration + +Every injected value is a default that goes into the submit configuration *before* your xref:usage-guide/overrides.adoc[`sparkConf`] is merged, so any key you set yourself wins. +`spark.extraListeners` and `spark.jars` are the exception: the operator *appends* to any value you provide (comma-merged) rather than overwriting it, so your own listeners and jars are kept alongside OpenLineage's. diff --git a/docs/modules/spark-k8s/partials/nav.adoc b/docs/modules/spark-k8s/partials/nav.adoc index 71f2278e..815c58dd 100644 --- a/docs/modules/spark-k8s/partials/nav.adoc +++ b/docs/modules/spark-k8s/partials/nav.adoc @@ -9,6 +9,7 @@ ** xref:spark-k8s:usage-guide/app_templates.adoc[] ** xref:spark-k8s:usage-guide/security.adoc[] ** xref:spark-k8s:usage-guide/logging.adoc[] +** xref:spark-k8s:usage-guide/openlineage.adoc[] ** xref:spark-k8s:usage-guide/history-server.adoc[] ** xref:spark-k8s:usage-guide/spark-connect.adoc[] ** xref:spark-k8s:usage-guide/examples.adoc[] diff --git a/extra/crds.yaml b/extra/crds.yaml index ab28308f..56f60237 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -1957,6 +1957,42 @@ spec: - cluster - client type: string + openLineage: + description: |- + Emit [OpenLineage](https://openlineage.io/) lineage events for this application. + The OpenLineage Spark listener runs on the driver and describes the whole application, + so this is application-scoped config (not per driver/executor role). + See the [OpenLineage usage guide](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/openlineage). + nullable: true + properties: + appName: + description: |- + A stable OpenLineage job/application name. Setting this prevents fragmented run history + (and the intermittent `unknown` job-name bug). If unset, the operator resolves it from + `spark.app.name`, falling back to `metadata.name` (with a warning event). + nullable: true + type: string + configMapName: + description: |- + Name of the OpenLineage backend [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery). + It must contain the key `ADDRESS` with the base URL of the OpenLineage backend + (e.g. `http://marquez:5000`). Mirrors the `vectorAggregatorConfigMapName` field on this CRD. + maxLength: 253 + minLength: 1 + nullable: true + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$ + type: string + enabled: + default: false + description: Enable OpenLineage event emission. Defaults to `false` (nothing is injected). + type: boolean + namespace: + description: |- + The OpenLineage namespace lineage is reported under. + Defaults to the application's Kubernetes namespace (`metadata.namespace`). + nullable: true + type: string + type: object s3connection: description: |- Configure an S3 connection that the SparkApplication has access to. @@ -6563,6 +6599,42 @@ spec: - cluster - client type: string + openLineage: + description: |- + Emit [OpenLineage](https://openlineage.io/) lineage events for this application. + The OpenLineage Spark listener runs on the driver and describes the whole application, + so this is application-scoped config (not per driver/executor role). + See the [OpenLineage usage guide](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/openlineage). + nullable: true + properties: + appName: + description: |- + A stable OpenLineage job/application name. Setting this prevents fragmented run history + (and the intermittent `unknown` job-name bug). If unset, the operator resolves it from + `spark.app.name`, falling back to `metadata.name` (with a warning event). + nullable: true + type: string + configMapName: + description: |- + Name of the OpenLineage backend [discovery ConfigMap](https://docs.stackable.tech/home/nightly/concepts/service_discovery). + It must contain the key `ADDRESS` with the base URL of the OpenLineage backend + (e.g. `http://marquez:5000`). Mirrors the `vectorAggregatorConfigMapName` field on this CRD. + maxLength: 253 + minLength: 1 + nullable: true + pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$ + type: string + enabled: + default: false + description: Enable OpenLineage event emission. Defaults to `false` (nothing is injected). + type: boolean + namespace: + description: |- + The OpenLineage namespace lineage is reported under. + Defaults to the application's Kubernetes namespace (`metadata.namespace`). + nullable: true + type: string + type: object s3connection: description: |- Configure an S3 connection that the SparkApplication has access to. diff --git a/rust/operator-binary/src/config/jvm.rs b/rust/operator-binary/src/config/jvm.rs index cc5388b8..21f19eff 100644 --- a/rust/operator-binary/src/config/jvm.rs +++ b/rust/operator-binary/src/config/jvm.rs @@ -2,8 +2,8 @@ use stackable_operator::crd::s3; use crate::crd::{ constants::{ - JVM_SECURITY_PROPERTIES_FILE, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE, - VOLUME_MOUNT_PATH_LOG_CONFIG, + JVM_SECURITY_PROPERTIES_FILE, OPENLINEAGE_ADD_OPENS, STACKABLE_TLS_STORE_PASSWORD, + STACKABLE_TRUST_STORE, VOLUME_MOUNT_PATH_LOG_CONFIG, }, logdir::ResolvedLogDir, tlscerts::tls_secret_names, @@ -36,6 +36,18 @@ pub fn construct_extra_java_options( ]); } + // OpenLineage on Spark 4.x needs `java.base/java.security` opened to the unnamed module, + // otherwise the driver throws a non-fatal `InaccessibleObjectException` and silently degrades + // extension-interface lineage (MVP §7). Added to both driver and executor. + if spark_application + .spec + .open_lineage + .as_ref() + .is_some_and(|open_lineage| open_lineage.enabled) + { + jvm_args.push(OPENLINEAGE_ADD_OPENS.to_string()); + } + // The role's `jvmArgumentOverrides` are applied on top of the operator-generated arguments // above. Note this is not purely additive: a role may also remove or replace operator-set // arguments (e.g. a `removeRegex` dropping the `-Djava.security.properties` default) — see the @@ -132,4 +144,30 @@ mod tests { "-Dhttps.proxyHost=from-executor" ); } + + #[test] + fn test_construct_jvm_arguments_openlineage_add_opens() { + let input = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + "#; + + let deserializer = serde_yaml::Deserializer::from_str(input); + let spark_app: SparkApplication = + serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + let (driver_extra_java_options, executor_extra_java_options) = + construct_extra_java_options(&spark_app, &None, &None); + + assert!(driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS)); + assert!(executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS)); + } } diff --git a/rust/operator-binary/src/crd/constants.rs b/rust/operator-binary/src/crd/constants.rs index efef0f14..50b5cf85 100644 --- a/rust/operator-binary/src/crd/constants.rs +++ b/rust/operator-binary/src/crd/constants.rs @@ -103,6 +103,30 @@ pub const DEFAULT_LISTENER_CLASS: &str = "cluster-internal"; pub const DEFAULT_SUBMIT_JOB_RETRY_ON_FAILURE_COUNT: u16 = 0; +// --- OpenLineage (see the OpenLineage usage guide) --- + +/// The OpenLineage Spark listener, appended to `spark.extraListeners` when OpenLineage is enabled. +pub const OPENLINEAGE_LISTENER_CLASS: &str = "io.openlineage.spark.agent.OpenLineageSparkListener"; + +/// `local://` URI of the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so +/// it shares the (child) classloader with `--packages` connectors. Delivering it this way — rather +/// than on `extraClassPath` (system classloader) — is what lets OpenLineage's connector probes +/// succeed; see the classpath discussion in the OpenLineage usage guide. +/// +/// IMPORTANT: the version MUST stay in sync with the `openlineage-spark-version` build-argument in +/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar at this path). +pub const OPENLINEAGE_JAR_LOCAL_URI: &str = + "local:///stackable/spark/openlineage/openlineage-spark_2.13-1.51.0.jar"; + +/// The key the OpenLineage discovery ConfigMap must hold, carrying the backend base URL. Mirrors the +/// `ADDRESS` contract of the existing `vectorAggregatorConfigMapName` discovery ConfigMap. +pub const OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY: &str = "ADDRESS"; + +/// Java module-system flag OpenLineage requires on Spark 4.x: without it the driver throws a +/// non-fatal `InaccessibleObjectException` and silently degrades extension-interface lineage. +/// Appended to both driver and executor `extraJavaOptions`. +pub const OPENLINEAGE_ADD_OPENS: &str = "--add-opens java.base/java.security=ALL-UNNAMED"; + /// The JVM `security.properties` entries the operator sets by default (DNS cache TTLs). pub fn default_jvm_security_properties() -> BTreeMap { [ diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index a456b1d9..53ef35db 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -9,7 +9,7 @@ use constants::*; use history::LogFileDirectorySpec; use logdir::ResolvedLogDir; use serde::{Deserialize, Serialize}; -use snafu::{OptionExt, ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu, ensure}; use stackable_operator::{ builder::pod::volume::{ SecretFormat, SecretOperatorVolumeSourceBuilder, SecretOperatorVolumeSourceBuilderError, @@ -114,6 +114,13 @@ pub enum Error { #[snafu(display("failed to configure log directory"))] ConfigureLogDir { source: logdir::Error }, + + #[snafu(display( + "OpenLineage is enabled but no backend endpoint could be resolved: set \ + `spec.openLineage.configMapName` (a discovery ConfigMap holding the `ADDRESS` key) or \ + `spark.openlineage.transport.url` in `sparkConf`" + ))] + MissingOpenLineageEndpoint, } pub type SparkApplicationJobRoleType = @@ -248,6 +255,43 @@ pub mod versioned { /// The log file directory definition used by the Spark history server. #[serde(default, skip_serializing_if = "Option::is_none")] pub log_file_directory: Option, + + /// Emit [OpenLineage](https://openlineage.io/) lineage events for this application. + /// The OpenLineage Spark listener runs on the driver and describes the whole application, + /// so this is application-scoped config (not per driver/executor role). + /// See the [OpenLineage usage guide](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/openlineage). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub open_lineage: Option, + } + + /// OpenLineage lineage emission for a [`SparkApplication`]. + /// + /// When enabled, the operator injects the OpenLineage Spark listener, points its HTTP transport + /// at the backend resolved from the discovery ConfigMap, and sets a stable job name. All injected + /// values are defaults: they can be overridden via `sparkConf`. + #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct OpenLineageSpec { + /// Enable OpenLineage event emission. Defaults to `false` (nothing is injected). + #[serde(default)] + pub enabled: bool, + + /// The OpenLineage namespace lineage is reported under. + /// Defaults to the application's Kubernetes namespace (`metadata.namespace`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub namespace: Option, + + /// A stable OpenLineage job/application name. Setting this prevents fragmented run history + /// (and the intermittent `unknown` job-name bug). If unset, the operator resolves it from + /// `spark.app.name`, falling back to `metadata.name` (with a warning event). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub app_name: Option, + + /// Name of the OpenLineage backend [discovery ConfigMap](DOCS_BASE_URL_PLACEHOLDER/concepts/service_discovery). + /// It must contain the key `ADDRESS` with the base URL of the OpenLineage backend + /// (e.g. `http://marquez:5000`). Mirrors the `vectorAggregatorConfigMapName` field on this CRD. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_map_name: Option, } #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize, Merge)] @@ -260,6 +304,35 @@ pub mod versioned { } } +/// The resolved OpenLineage job/app name and where it came from. +/// See [`v1alpha1::SparkApplication::resolved_openlineage_app_name`]. +#[derive(Clone, Debug, PartialEq)] +pub struct ResolvedOpenLineageAppName { + /// The resolved, non-blank name written to `spark.openlineage.appName`. + pub name: String, + /// `true` when the name fell back to `metadata.name`; the controller then emits a warning event + /// because a per-run-unique name (e.g. an orchestrator-generated `-` suffix) fragments + /// backend run history. + pub from_metadata_name: bool, +} + +/// Appends `value` to a comma-separated `--conf` value in `submit_conf`, preserving any existing +/// (e.g. user-supplied) entries and skipping `value` if it is already present. Used for the +/// OpenLineage keys that must accumulate rather than clobber (`spark.extraListeners`, `spark.jars`). +fn append_conf_csv(submit_conf: &mut BTreeMap, key: &str, value: &str) { + match submit_conf.get_mut(key) { + Some(existing) if !existing.is_empty() => { + if !existing.split(',').any(|entry| entry.trim() == value) { + existing.push(','); + existing.push_str(value); + } + } + _ => { + submit_conf.insert(key.to_string(), value.to_string()); + } + } +} + impl v1alpha1::SparkApplication { /// Returns if this [`SparkApplication`] has already created a Kubernetes Job doing the actual `spark-submit`. /// @@ -561,6 +634,7 @@ impl v1alpha1::SparkApplication { s3conn: &Option, log_dir: &Option, spark_image: &str, + openlineage_endpoint: Option<&str>, ) -> Result, Error> { // mandatory properties let mode = &self.spec.mode; @@ -733,9 +807,66 @@ impl v1alpha1::SparkApplication { submit_cmd.push(format!("--conf spark.jars.ivy={VOLUME_MOUNT_PATH_IVY2}")) } + // OpenLineage: inject the transport + job-name config that can be overridden by the user's + // `sparkConf` (so it goes in BEFORE the merge below). The two append-merge keys + // (`spark.jars`, `spark.extraListeners`) are handled AFTER the merge so a user value is + // combined with — not clobbered by — ours. See the OpenLineage usage guide. + if let Some(open_lineage) = &self.spec.open_lineage + && open_lineage.enabled + { + // Fail fast rather than emit a transport that silently drops every event. + let has_url_override = self + .spec + .spark_conf + .contains_key("spark.openlineage.transport.url"); + ensure!( + openlineage_endpoint.is_some() || has_url_override, + MissingOpenLineageEndpointSnafu + ); + + submit_conf.insert( + "spark.openlineage.transport.type".to_string(), + "http".to_string(), + ); + if let Some(endpoint) = openlineage_endpoint { + submit_conf.insert( + "spark.openlineage.transport.url".to_string(), + endpoint.to_string(), + ); + } + + let namespace = match &open_lineage.namespace { + Some(namespace) => namespace.clone(), + None => self.metadata.namespace.clone().context(NoNamespaceSnafu)?, + }; + submit_conf.insert("spark.openlineage.namespace".to_string(), namespace); + + // Stable job name — fixes the intermittent `unknown` bug (see the usage guide). + submit_conf.insert( + "spark.openlineage.appName".to_string(), + self.resolved_openlineage_app_name()?.name, + ); + } + // conf arguments: these should follow - and thus override - values set from resource limits above submit_conf.extend(self.spec.spark_conf.clone()); + // OpenLineage append-merge keys: combine our values with any user-provided ones (already + // merged into `submit_conf` above) so neither clobbers the other. + if self + .spec + .open_lineage + .as_ref() + .is_some_and(|open_lineage| open_lineage.enabled) + { + append_conf_csv( + &mut submit_conf, + "spark.extraListeners", + OPENLINEAGE_LISTENER_CLASS, + ); + append_conf_csv(&mut submit_conf, "spark.jars", OPENLINEAGE_JAR_LOCAL_URI); + } + // ...before being added to the command collection for (key, value) in submit_conf { submit_cmd.push(format!("--conf \"{key}={value}\"")); @@ -756,6 +887,39 @@ impl v1alpha1::SparkApplication { Ok(vec![submit_cmd.join(" ")]) } + /// Resolves the stable OpenLineage job/app name and its provenance (MVP §5), in priority order: + /// 1. `spec.openLineage.appName`, else + /// 2. `spark.app.name` from `sparkConf`, else + /// 3. `metadata.name` (a fallback the controller flags with a warning event, because a + /// per-run-unique name would fragment backend run history). + /// + /// Always yields a non-blank name — which is exactly what fixes the intermittent `unknown` bug. + pub fn resolved_openlineage_app_name(&self) -> Result { + if let Some(app_name) = self + .spec + .open_lineage + .as_ref() + .and_then(|open_lineage| open_lineage.app_name.clone()) + { + return Ok(ResolvedOpenLineageAppName { + name: app_name, + from_metadata_name: false, + }); + } + + if let Some(app_name) = self.spec.spark_conf.get("spark.app.name") { + return Ok(ResolvedOpenLineageAppName { + name: app_name.clone(), + from_metadata_name: false, + }); + } + + Ok(ResolvedOpenLineageAppName { + name: self.metadata.name.clone().context(ObjectHasNoNameSnafu)?, + from_metadata_name: true, + }) + } + pub fn env( &self, s3conn: &Option, @@ -1574,6 +1738,262 @@ spec: assert_eq!(got, expected); } + /// Builds a `SparkApplication` from YAML and returns the assembled spark-submit command string, + /// with `openlineage_endpoint` standing in for the value the controller resolves from the + /// discovery ConfigMap. + fn build_command_with_openlineage(yaml: &str, openlineage_endpoint: Option<&str>) -> String { + let spark_application = serde_yaml::from_str::(yaml).unwrap(); + spark_application + .build_command(&None, &None, "test-image", openlineage_endpoint) + .unwrap() + .join(" ") + } + + const OPENLINEAGE_ENABLED_APP: &str = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + "#}; + + #[test] + fn test_openlineage_injects_conf_when_enabled() { + let command = + build_command_with_openlineage(OPENLINEAGE_ENABLED_APP, Some("http://marquez:5000")); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.type=http""#)); + assert!( + command.contains(r#"--conf "spark.openlineage.transport.url=http://marquez:5000""#) + ); + // Namespace defaults to metadata.namespace. + assert!(command.contains(r#"--conf "spark.openlineage.namespace=default""#)); + // appName falls back to metadata.name (no appName / spark.app.name set). + assert!(command.contains(r#"--conf "spark.openlineage.appName=spark-examples""#)); + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# + ))); + assert!(command.contains(&format!( + r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# + ))); + // --add-opens reaches both driver and executor. + assert_eq!( + command.matches(OPENLINEAGE_ADD_OPENS).count(), + 2, + "expected --add-opens on both driver and executor extraJavaOptions" + ); + } + + #[rstest] + #[case::absent(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + "#})] + #[case::disabled(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: false + "#})] + fn test_openlineage_injects_nothing_when_absent_or_disabled(#[case] yaml: &str) { + // Endpoint is supplied to prove it is ignored, not that it is simply missing. + let command = build_command_with_openlineage(yaml, Some("http://marquez:5000")); + + assert!(!command.contains("openlineage")); + assert!(!command.contains("OpenLineageSparkListener")); + assert!(!command.contains("--add-opens")); + assert!(!command.contains("spark.jars=")); + } + + #[test] + fn test_openlineage_appends_to_existing_extra_listeners() { + let yaml = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.extraListeners: com.example.CustomListener + "#}; + let command = build_command_with_openlineage(yaml, Some("http://marquez:5000")); + + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners=com.example.CustomListener,{OPENLINEAGE_LISTENER_CLASS}""# + ))); + } + + #[test] + fn test_openlineage_conf_is_overridable_via_spark_conf() { + let yaml = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.openlineage.transport.url: http://custom:1234 + spark.openlineage.namespace: custom-ns + "#}; + let command = build_command_with_openlineage(yaml, Some("http://marquez:5000")); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.url=http://custom:1234""#)); + assert!(!command.contains("http://marquez:5000")); + assert!(command.contains(r#"--conf "spark.openlineage.namespace=custom-ns""#)); + } + + #[test] + fn test_openlineage_fails_fast_without_endpoint() { + // Enabled, but neither a resolved endpoint nor a sparkConf transport.url override. + let spark_application = + serde_yaml::from_str::(OPENLINEAGE_ENABLED_APP).unwrap(); + let result = spark_application.build_command(&None, &None, "test-image", None); + + assert!(matches!(result, Err(Error::MissingOpenLineageEndpoint))); + } + + #[test] + fn test_openlineage_endpoint_from_spark_conf_satisfies_fail_fast() { + // No resolved endpoint, but a sparkConf override → must NOT fail. + let yaml = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-examples + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.openlineage.transport.url: http://custom:1234 + "#}; + let command = build_command_with_openlineage(yaml, None); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.url=http://custom:1234""#)); + } + + #[rstest] + #[case::explicit_app_name( + indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: metadata-name + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + appName: explicit-name + sparkConf: + spark.app.name: spark-conf-name + "#}, + "explicit-name", + false + )] + #[case::spark_app_name( + indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: metadata-name + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + sparkConf: + spark.app.name: spark-conf-name + "#}, + "spark-conf-name", + false + )] + #[case::metadata_name_fallback( + indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: metadata-name + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + openLineage: + enabled: true + "#}, + "metadata-name", + true + )] + fn test_openlineage_app_name_resolution( + #[case] yaml: &str, + #[case] expected_name: &str, + #[case] expected_from_metadata_name: bool, + ) { + let spark_application = serde_yaml::from_str::(yaml).unwrap(); + let resolved = spark_application.resolved_openlineage_app_name().unwrap(); + + assert_eq!(resolved.name, expected_name); + assert_eq!(resolved.from_metadata_name, expected_from_metadata_name); + } + impl RoundtripTestData for v1alpha1::SparkApplicationSpec { fn roundtrip_test_data() -> Vec { stackable_operator::utils::yaml_from_str_singleton_map(indoc! {r#" diff --git a/rust/operator-binary/src/crd/template_merger.rs b/rust/operator-binary/src/crd/template_merger.rs index 7f8afbba..b12bff0f 100644 --- a/rust/operator-binary/src/crd/template_merger.rs +++ b/rust/operator-binary/src/crd/template_merger.rs @@ -55,6 +55,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()), + open_lineage: overlay + .spec + .open_lineage + .clone() + .or_else(|| base.spec.open_lineage.clone()), s3connection: overlay .spec .s3connection diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 9f14b776..c91b2290 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -70,6 +70,7 @@ struct Opts { pub struct Ctx { pub client: stackable_operator::client::Client, pub operator_environment: OperatorEnvironmentOptions, + pub event_recorder: Recorder, } #[tokio::main] @@ -136,11 +137,6 @@ async fn main() -> anyhow::Result<()> { .run(sigterm_watcher.handle()) .map_err(|err| anyhow!(err).context("failed to run webhook server")); - let ctx = Ctx { - client: client.clone(), - operator_environment: operator_environment.clone(), - }; - let spark_event_recorder = Arc::new(Recorder::new( client.as_kube_client(), Reporter { @@ -148,6 +144,14 @@ async fn main() -> anyhow::Result<()> { instance: None, }, )); + + let ctx = Ctx { + client: client.clone(), + operator_environment: operator_environment.clone(), + // Shares the recorder (and thus its event-aggregation cache) with the + // report_controller_reconciled reporting stream below; `Recorder` is `Clone`. + event_recorder: (*spark_event_recorder).clone(), + }; let app_controller = Controller::new( watch_namespace .get_api::>(&client), @@ -224,11 +228,6 @@ async fn main() -> anyhow::Result<()> { }, ).map(anyhow::Ok); - // Create new object because Ctx cannot be cloned - let ctx = Ctx { - client: client.clone(), - operator_environment: operator_environment.clone(), - }; let history_event_recorder = Arc::new(Recorder::new( client.as_kube_client(), Reporter { @@ -236,6 +235,12 @@ async fn main() -> anyhow::Result<()> { instance: None, }, )); + // Create new object because Ctx cannot be cloned + let ctx = Ctx { + client: client.clone(), + operator_environment: operator_environment.clone(), + event_recorder: (*history_event_recorder).clone(), + }; let history_controller = Controller::new( watch_namespace .get_api::>( @@ -289,11 +294,6 @@ async fn main() -> anyhow::Result<()> { .map(anyhow::Ok); // ============================== - // Create new object because Ctx cannot be cloned - let ctx = Ctx { - client: client.clone(), - operator_environment, - }; let connect_event_recorder = Arc::new(Recorder::new( client.as_kube_client(), Reporter { @@ -301,6 +301,12 @@ async fn main() -> anyhow::Result<()> { instance: None, }, )); + // Create new object because Ctx cannot be cloned + let ctx = Ctx { + client: client.clone(), + operator_environment, + event_recorder: (*connect_event_recorder).clone(), + }; let connect_controller = Controller::new( watch_namespace .get_api::>( diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs index a7a128b7..10867520 100644 --- a/rust/operator-binary/src/spark_k8s_controller.rs +++ b/rust/operator-binary/src/spark_k8s_controller.rs @@ -1,12 +1,16 @@ use std::sync::Arc; -use snafu::{ResultExt, Snafu}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ builder::{self}, + k8s_openapi::api::core::v1::ConfigMap, kube::{ - ResourceExt, + Resource, ResourceExt, core::{DeserializeGuard, error_boundary}, - runtime::controller::Action, + runtime::{ + controller::Action, + events::{Event, EventType, Recorder}, + }, }, logging::controller::ReconcilerError, shared::time::Duration, @@ -68,6 +72,18 @@ pub enum Error { #[snafu(display("vector agent is enabled but vector aggregator ConfigMap is missing"))] VectorAggregatorConfigMapMissing, + #[snafu(display("failed to resolve the OpenLineage discovery ConfigMap {name:?}"))] + ResolveOpenLineageConfigMap { + source: stackable_operator::client::Error, + name: String, + }, + + #[snafu(display( + "the OpenLineage discovery ConfigMap {name:?} does not contain the required key \ + {OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY:?}" + ))] + OpenLineageConfigMapMissingAddress { name: String }, + #[snafu(display("failed to validate the logging configuration"))] ValidateLoggingConfig { source: stackable_operator::v2::product_logging::framework::Error, @@ -223,8 +239,30 @@ pub async fn reconcile( .await .context(ApplyApplicationSnafu)?; + // Resolve the OpenLineage backend endpoint from the discovery ConfigMap (if configured) and + // warn if the job name falls back to `metadata.name` — both only matter when OpenLineage is + // enabled. + let openlineage_endpoint = resolve_openlineage_endpoint(client, spark_application).await?; + if spark_application + .spec + .open_lineage + .as_ref() + .is_some_and(|open_lineage| open_lineage.enabled) + && spark_application + .resolved_openlineage_app_name() + .context(BuildCommandSnafu)? + .from_metadata_name + { + publish_openlineage_app_name_warning(&ctx.event_recorder, spark_application).await; + } + let job_commands = spark_application - .build_command(opt_s3conn, logdir, &resolved_product_image.image) + .build_command( + opt_s3conn, + logdir, + &resolved_product_image.image, + openlineage_endpoint.as_deref(), + ) .context(BuildCommandSnafu)?; let submit_config = spark_application @@ -291,3 +329,75 @@ pub fn error_policy( _ => Action::requeue(*Duration::from_secs(5)), } } + +/// Resolves the OpenLineage backend base URL from the discovery ConfigMap referenced by +/// `spec.openLineage.configMapName` (key [`OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY`]). Returns `None` +/// when OpenLineage is disabled/absent or no ConfigMap is referenced — in the latter case a +/// `sparkConf` override is expected, which `build_command` enforces. +async fn resolve_openlineage_endpoint( + client: &stackable_operator::client::Client, + spark_application: &v1alpha1::SparkApplication, +) -> Result> { + let Some(open_lineage) = &spark_application.spec.open_lineage else { + return Ok(None); + }; + let (true, Some(config_map_name)) = (open_lineage.enabled, &open_lineage.config_map_name) + else { + return Ok(None); + }; + + let name = config_map_name.as_ref(); + let namespace = spark_application.namespace().unwrap_or_default(); + + let config_map = client.get::(name, &namespace).await.context( + ResolveOpenLineageConfigMapSnafu { + name: name.to_string(), + }, + )?; + + let address = config_map + .data + .as_ref() + .and_then(|data| data.get(OPENLINEAGE_CONFIG_MAP_ADDRESS_KEY)) + .cloned() + .context(OpenLineageConfigMapMissingAddressSnafu { + name: name.to_string(), + })?; + + Ok(Some(address)) +} + +/// Emits a Kubernetes warning event that the OpenLineage job name fell back to `metadata.name`, +/// which fragments backend run history when that name is unique per run. Event-publishing failures +/// are logged, not propagated — they must not fail reconciliation. +async fn publish_openlineage_app_name_warning( + event_recorder: &Recorder, + spark_application: &v1alpha1::SparkApplication, +) { + let name = spark_application.name_any(); + let publish_result = event_recorder + .publish( + &Event { + type_: EventType::Warning, + reason: "OpenLineageAppNameFallback".into(), + note: Some(format!( + "OpenLineage job name falls back to metadata.name ({name:?}) because neither \ + spec.openLineage.appName nor spark.app.name is set. If metadata.name is unique \ + per run (e.g. an orchestrator-generated - suffix), backend run \ + history will be fragmented into a new job per run. Set \ + spec.openLineage.appName to a stable value to avoid this." + )), + action: "ResolveOpenLineageAppName".into(), + secondary: None, + }, + &spark_application.object_ref(&()), + ) + .await; + + if let Err(error) = publish_result { + tracing::warn!( + ?error, + "failed to publish OpenLineage app-name fallback warning event" + ); + } +} From 43235f5af859a9163dbd6b3285a60aa6cba45b0d Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:29:24 +0200 Subject: [PATCH 2/6] tests: better testing of the hbase connectors (#717) * tests: better testing of the hbase connectors * bump hbase * update supported versions * update fixture --- .../partials/supported-versions.adoc | 2 +- .../10-deploy-spark-app.yaml.j2 | 56 ++++++++++++++++++- .../10-deploy-spark-app.yaml.j2 | 1 + .../pyspark-pi-driver-pod-template-data.json | 2 +- ...pyspark-pi-executor-pod-template-data.json | 2 +- .../pyspark-pi-job-template-spec.json | 2 +- tests/test-definition.yaml | 2 +- 7 files changed, 61 insertions(+), 6 deletions(-) diff --git a/docs/modules/spark-k8s/partials/supported-versions.adoc b/docs/modules/spark-k8s/partials/supported-versions.adoc index bd3cc078..494d69e6 100644 --- a/docs/modules/spark-k8s/partials/supported-versions.adoc +++ b/docs/modules/spark-k8s/partials/supported-versions.adoc @@ -5,7 +5,7 @@ - 4.1.2 (Hadoop 3.4.3, Scala 2.13, Python 3.12, Java 21) (LTS) - 4.1.1 (Hadoop 3.4.3, Scala 2.13, Python 3.12, Java 21) (Deprecated) -- 3.5.8 (Hadoop 3.4.2, Scala 2.12, Python 3.11, Java 17) (Deprecated) +- 3.5.8 (HBase 2.6.6, Hadoop 3.4.3, Scala 2.12, Python 3.11, Java 17) (Deprecated) Apache Spark 4.1.1 & 4.1.2 have the following known issues (as of July 2026): diff --git a/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 b/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 index 76245c96..87aa996d 100644 --- a/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 +++ b/tests/templates/kuttl/hbase-connector/10-deploy-spark-app.yaml.j2 @@ -77,7 +77,11 @@ data: spark = SparkSession.builder.appName("test-hbase").getOrCreate() df = spark.createDataFrame( - [("row1", "Hello, Stackable!")], + [ + ("row1", "Hello, Stackable!"), + ("row2", "Hello, HBase!"), + ("row3", "Hello, Spark!"), + ], "key: string, value: string" ) @@ -101,3 +105,53 @@ data: .option('catalog', catalog)\ .option('newtable', '5')\ .save() + + # Read the whole table back and register it as a temporary view so we can + # run SQL queries against it. + # + # hbase.spark.pushdown.columnfilter=false disables server-side predicate + # pushdown: the connector would otherwise ship a SparkSQLPushDownFilter to + # the RegionServers, which requires the hbase-spark connector classes on + # the RegionServer classpath (not present in the Stackable HBase image). + # With pushdown off, WHERE clauses are evaluated Spark-side over a full + # scan, which is all this test needs. + read_df = spark\ + .read\ + .format("org.apache.hadoop.hbase.spark")\ + .option('catalog', catalog)\ + .option('hbase.spark.pushdown.columnfilter', 'false')\ + .load() + read_df.createOrReplaceTempView("test_hbase") + + # Full scan: all rows that were written must be readable again. + all_rows = spark.sql("SELECT key, value FROM test_hbase").collect() + actual = {row["key"]: row["value"] for row in all_rows} + expected = { + "row1": "Hello, Stackable!", + "row2": "Hello, HBase!", + "row3": "Hello, Spark!", + } + assert actual == expected, f"full scan mismatch: {actual} != {expected}" + + # Row-key point lookup. + point = spark.sql("SELECT value FROM test_hbase WHERE key = 'row2'").collect() + assert len(point) == 1, f"point lookup returned {len(point)} rows, expected 1" + assert point[0]["value"] == "Hello, HBase!", f"unexpected value: {point[0]['value']}" + + # Range scan on the row key. + range_rows = spark.sql( + "SELECT key FROM test_hbase WHERE key >= 'row2' ORDER BY key" + ).collect() + assert [r["key"] for r in range_rows] == ["row2", "row3"], \ + f"unexpected range scan result: {[r['key'] for r in range_rows]}" + + # Filter on a non-rowkey column value. + like_rows = spark.sql( + "SELECT key FROM test_hbase WHERE value LIKE 'Hello, S%' ORDER BY key" + ).collect() + assert [r["key"] for r in like_rows] == ["row1", "row3"], \ + f"unexpected value filter result: {[r['key'] for r in like_rows]}" + + # Aggregation over the scanned rows. + count = spark.sql("SELECT COUNT(*) AS c FROM test_hbase").collect()[0]["c"] + assert count == 3, f"expected 3 rows, got {count}" diff --git a/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 b/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 index b2b9104d..9c4d46a7 100644 --- a/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 +++ b/tests/templates/kuttl/product-config-compat/10-deploy-spark-app.yaml.j2 @@ -5,6 +5,7 @@ metadata: name: pyspark-pi spec: sparkImage: + pullPolicy: IfNotPresent {% if test_scenario['values']['spark-regression'].find(",") > 0 %} custom: "{{ test_scenario['values']['spark-regression'].split(',')[1] }}" productVersion: "{{ test_scenario['values']['spark-regression'].split(',')[0] }}" diff --git a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json index c65e0c89..1e35fafb 100644 --- a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json +++ b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-driver-pod-template-data.json @@ -2,5 +2,5 @@ "log4j2.properties": "appenders = FILE, CONSOLE\n\nappender.CONSOLE.type = Console\nappender.CONSOLE.name = CONSOLE\nappender.CONSOLE.target = SYSTEM_ERR\nappender.CONSOLE.layout.type = PatternLayout\nappender.CONSOLE.layout.pattern = %d{ISO8601} %p [%t] %c - %m%n\nappender.CONSOLE.filter.threshold.type = ThresholdFilter\nappender.CONSOLE.filter.threshold.level = INFO\n\nappender.FILE.type = RollingFile\nappender.FILE.name = FILE\nappender.FILE.fileName = /stackable/log/spark/spark.log4j2.xml\nappender.FILE.filePattern = /stackable/log/spark/spark.log4j2.xml.%i\nappender.FILE.layout.type = XMLLayout\nappender.FILE.policies.type = Policies\nappender.FILE.policies.size.type = SizeBasedTriggeringPolicy\nappender.FILE.policies.size.size = 5MB\nappender.FILE.strategy.type = DefaultRolloverStrategy\nappender.FILE.strategy.max = 1\nappender.FILE.filter.threshold.type = ThresholdFilter\nappender.FILE.filter.threshold.level = INFO\n\n\nrootLogger.level=INFO\nrootLogger.appenderRefs = CONSOLE, FILE\nrootLogger.appenderRef.CONSOLE.ref = CONSOLE\nrootLogger.appenderRef.FILE.ref = FILE", "security.properties": "networkaddress.cache.negative.ttl=0\nnetworkaddress.cache.ttl=30\n", "spark-env.sh": "", - "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n prometheus.io/scrape: 'true'\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: Always\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: config\n" + "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n prometheus.io/scrape: 'true'\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: IfNotPresent\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-driver-pod-template\n name: config\n" } diff --git a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json index 03c53b63..096896c5 100644 --- a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json +++ b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-executor-pod-template-data.json @@ -2,5 +2,5 @@ "log4j2.properties": "appenders = FILE, CONSOLE\n\nappender.CONSOLE.type = Console\nappender.CONSOLE.name = CONSOLE\nappender.CONSOLE.target = SYSTEM_ERR\nappender.CONSOLE.layout.type = PatternLayout\nappender.CONSOLE.layout.pattern = %d{ISO8601} %p [%t] %c - %m%n\nappender.CONSOLE.filter.threshold.type = ThresholdFilter\nappender.CONSOLE.filter.threshold.level = INFO\n\nappender.FILE.type = RollingFile\nappender.FILE.name = FILE\nappender.FILE.fileName = /stackable/log/spark/spark.log4j2.xml\nappender.FILE.filePattern = /stackable/log/spark/spark.log4j2.xml.%i\nappender.FILE.layout.type = XMLLayout\nappender.FILE.policies.type = Policies\nappender.FILE.policies.size.type = SizeBasedTriggeringPolicy\nappender.FILE.policies.size.size = 5MB\nappender.FILE.strategy.type = DefaultRolloverStrategy\nappender.FILE.strategy.max = 1\nappender.FILE.filter.threshold.type = ThresholdFilter\nappender.FILE.filter.threshold.level = INFO\n\n\nrootLogger.level=INFO\nrootLogger.appenderRefs = CONSOLE, FILE\nrootLogger.appenderRef.CONSOLE.ref = CONSOLE\nrootLogger.appenderRef.FILE.ref = FILE", "security.properties": "networkaddress.cache.negative.ttl=0\nnetworkaddress.cache.ttl=30\n", "spark-env.sh": "", - "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: Always\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: config\n" + "template.yaml": "metadata:\n labels:\n app.kubernetes.io/component: spark\n app.kubernetes.io/instance: pyspark-pi\n app.kubernetes.io/managed-by: spark.stackable.tech_sparkapplication\n app.kubernetes.io/name: spark-k8s\n app.kubernetes.io/role-group: sparkapplication\n app.kubernetes.io/version: 3.5.8-stackable0.0.0-dev\n stackable.tech/vendor: Stackable\n name: spark\nspec:\n affinity: {}\n containers:\n - env:\n - name: CONTAINERDEBUG_LOG_DIRECTORY\n value: /stackable/log/containerdebug\n - name: _STACKABLE_PRE_HOOK\n value: containerdebug --output=/stackable/log/containerdebug-state.json --loop &\n image: oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev\n imagePullPolicy: IfNotPresent\n name: spark\n resources:\n limits:\n cpu: '2'\n memory: 1Gi\n requests:\n cpu: '1'\n memory: 1Gi\n volumeMounts:\n - mountPath: /stackable/log_config\n name: log-config\n - mountPath: /stackable/log\n name: log\n enableServiceLinks: false\n securityContext:\n fsGroup: 1000\n serviceAccountName: pyspark-pi\n volumes:\n - emptyDir:\n sizeLimit: 39Mi\n name: log\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: log-config\n - configMap:\n name: pyspark-pi-executor-pod-template\n name: config\n" } diff --git a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json index dc9ada45..22518c52 100644 --- a/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json +++ b/tests/templates/kuttl/product-config-compat/fixtures/pyspark-pi-job-template-spec.json @@ -31,7 +31,7 @@ } ], "image": "oci.stackable.tech/sdp/spark-k8s:3.5.8-stackable0.0.0-dev", - "imagePullPolicy": "Always", + "imagePullPolicy": "IfNotPresent", "name": "spark-submit", "resources": { "limits": { diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 005ca2a1..5d387b63 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -56,7 +56,7 @@ dimensions: - "CLUSTER.LOCAL" - name: hbase values: - - 2.6.4 + - 2.6.6 - name: hdfs-latest values: - 3.5.0 From c7313a8962074833326d573b96dd083565a3ad1a Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 10 Jul 2026 12:07:14 +0200 Subject: [PATCH 3/6] Revert "fix: make application templates namespaced (#694)" (#719) * Revert "fix: make application templates namespaced (#694)" This reverts commit b326be35a5e3c4168b1eea4931b5994f2c7d0e0d. * revert changelog * update nix crate hashes --- CHANGELOG.md | 2 - Cargo.nix | 18 ++--- crate-hashes.json | 18 ++--- .../pages/usage-guide/app_templates.adoc | 14 +--- extra/crds.yaml | 2 +- .../src/crd/template_merger.rs | 41 ------------ rust/operator-binary/src/crd/template_spec.rs | 67 +------------------ 7 files changed, 22 insertions(+), 140 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9405469..5954870a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,7 +31,6 @@ All notable changes to this project will be documented in this file. - Internal operator refactoring: introduce dereference() and validate() steps in the reconciler for spark application, spark connect and spark history server([#687]). - test: Bump vector-aggregator to 0.55.0, replace /graphql call with gRPC call ([#689]). - Fix the `SparkApplication` CRD description, which incorrectly described it as a "Spark cluster stacklet" rather than a Spark application ([#705]). -- BREAKING: make application templates namespaced instead of cluster wide objects ([#694]). [#674]: https://github.com/stackabletech/spark-k8s-operator/pull/674 [#679]: https://github.com/stackabletech/spark-k8s-operator/pull/679 @@ -40,7 +39,6 @@ All notable changes to this project will be documented in this file. [#687]: https://github.com/stackabletech/spark-k8s-operator/pull/687 [#689]: https://github.com/stackabletech/spark-k8s-operator/pull/689 [#692]: https://github.com/stackabletech/spark-k8s-operator/pull/692 -[#694]: https://github.com/stackabletech/spark-k8s-operator/pull/694 [#696]: https://github.com/stackabletech/spark-k8s-operator/pull/696 [#705]: https://github.com/stackabletech/spark-k8s-operator/pull/705 [#708]: https://github.com/stackabletech/spark-k8s-operator/pull/708 diff --git a/Cargo.nix b/Cargo.nix index fd10413f..4bb9f18e 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -5109,7 +5109,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "k8s_version"; authors = [ @@ -9893,7 +9893,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "stackable_certs"; authors = [ @@ -9996,7 +9996,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "stackable_operator"; authors = [ @@ -10195,7 +10195,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; procMacro = true; libName = "stackable_operator_derive"; @@ -10230,7 +10230,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "stackable_shared"; authors = [ @@ -10417,7 +10417,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "stackable_telemetry"; authors = [ @@ -10527,7 +10527,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "stackable_versioned"; authors = [ @@ -10577,7 +10577,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; procMacro = true; libName = "stackable_versioned_macros"; @@ -10645,7 +10645,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; + sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; }; libName = "stackable_webhook"; authors = [ diff --git a/crate-hashes.json b/crate-hashes.json index cd3917c3..77adf52a 100644 --- a/crate-hashes.json +++ b/crate-hashes.json @@ -1,12 +1,12 @@ { - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#k8s-version@0.1.3": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-certs@0.4.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator-derive@0.3.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator@0.113.3": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-shared@0.1.2": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-telemetry@0.6.5": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned-macros@0.11.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned@0.11.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-webhook@0.9.2": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#k8s-version@0.1.3": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-certs@0.4.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator-derive@0.3.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator@0.113.3": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-shared@0.1.2": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-telemetry@0.6.5": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned-macros@0.11.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned@0.11.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-webhook@0.9.2": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", "git+https://github.com/stackabletech/product-config.git?tag=0.8.0#product-config@0.8.0": "1dz70kapm2wdqcr7ndyjji0lhsl98bsq95gnb2lw487wf6yr7987" } \ No newline at end of file diff --git a/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc b/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc index be8ebb9f..67a289fb 100644 --- a/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc +++ b/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc @@ -5,23 +5,13 @@ Spark application templates are used to define reusable configurations for Spark When you have many applications with similar configurations, templates can help you avoid duplication by grouping common settings together. Application templates are available for the `v1alpha1` version of the SparkApplication custom resource and share the exact same structure as the SparkApplication resource, but with some differences in the way the operator handles them: -1. Application templates are namespace-scoped resources, just like Spark applications. - This means that a SparkApplication can only reference templates from its own namespace. +1. Application templates are cluster wide resources, while Spark application resources are namespace-scoped. + This means that application templates can be used across multiple namespaces, while Spark application resources are limited to the namespace they are created in. 2. Application templates are not reconciled by the operator, but must be referenced from a SparkApplication resource to be applied. This means that changes to an application template will not automatically trigger updates to SparkApplication resources that reference it. 3. An application can reference multiple application templates, and the settings from these templates will be merged together. The merging order of the templates is indicated by their index in the reference list. The application fields have the highest precedence and will override any conflicting settings from the templates. This allows you to have a base template with common settings and then override specific settings in the application resource as needed. 4. Application template references are immutable in the sense that once applied to an application they cannot be changed again. Currently templates are applied upon the creation of the application, and any changes to the template references after that will be ignored. 5. Application and template CRDs must have the exact same versions. Currently only `v1alpha1` is supported. -== Migrating from cluster-scoped templates - -IMPORTANT: Application templates used to be cluster wide resources when they were first released. This was a mistake. Many users do not have the access rights to create cluster scoped resources and so the templates are now namespace scoped. - -If you are migrating from older installations where templates were treated as cluster-wide resources, account for the following: - -1. Recreate each template in every namespace where SparkApplications use it. -2. Keep template names consistent per namespace if you want the same application annotations to continue working. -3. Cross-namespace template references are no longer resolved; templates and applications must be in the same namespace. -4. Update GitOps/automation manifests to create templates as namespace-targeted resources before reconciling dependent SparkApplications. == Examples Applications use `metadata.annotations` to reference application templates as shown below: diff --git a/extra/crds.yaml b/extra/crds.yaml index 56f60237..cba2c2a9 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -4657,7 +4657,7 @@ spec: shortNames: - sparkapptemplate singular: sparkapplicationtemplate - scope: Namespaced + scope: Cluster versions: - additionalPrinterColumns: [] name: v1alpha1 diff --git a/rust/operator-binary/src/crd/template_merger.rs b/rust/operator-binary/src/crd/template_merger.rs index b12bff0f..b4fc48dc 100644 --- a/rust/operator-binary/src/crd/template_merger.rs +++ b/rust/operator-binary/src/crd/template_merger.rs @@ -322,47 +322,6 @@ mod tests { assert_eq!(merged.spec.args, vec!["arg1", "arg2"]); } - #[test] - fn test_deep_merge_metadata_namespace_overlay_wins() { - let base = serde_yaml::from_str::(indoc! {r#" - --- - apiVersion: spark.stackable.tech/v1alpha1 - kind: SparkApplication - metadata: - name: base-app - namespace: template-namespace - spec: - mode: cluster - mainApplicationFile: base.py - sparkImage: - productVersion: "3.5.0" - "#}) - .unwrap(); - - let overlay = serde_yaml::from_str::(indoc! {r#" - --- - apiVersion: spark.stackable.tech/v1alpha1 - kind: SparkApplication - metadata: - name: overlay-app - namespace: app-namespace - spec: - mode: cluster - mainApplicationFile: overlay.py - sparkImage: - productVersion: "3.5.1" - "#}) - .unwrap(); - - let merged = deep_merge(&base, &overlay); - - assert_eq!( - merged.metadata.namespace, - Some("app-namespace".to_string()), - "overlay namespace should take precedence" - ); - } - #[test] fn test_deep_merge_spark_conf() { let base = serde_yaml::from_str::(indoc! {r#" diff --git a/rust/operator-binary/src/crd/template_spec.rs b/rust/operator-binary/src/crd/template_spec.rs index 7312abe3..da5fdd05 100644 --- a/rust/operator-binary/src/crd/template_spec.rs +++ b/rust/operator-binary/src/crd/template_spec.rs @@ -44,9 +44,6 @@ pub enum Error { template_name: String, source: stackable_operator::kube::Error, }, - - #[snafu(display("object has no namespace"))] - ObjectHasNoNamespace, } #[versioned( @@ -69,7 +66,6 @@ pub mod versioned { group = "spark.stackable.tech", plural = "sparkapptemplates", shortname = "sparkapptemplate", - namespaced ))] #[derive(Clone, CustomResource, Debug, Deserialize, JsonSchema, Serialize)] #[serde(rename_all = "camelCase")] @@ -267,10 +263,8 @@ pub(crate) async fn merge_application_templates( // In the future if we support additional strategies in addition to "enforce", // this list might not be identical to the one in `merge_template_options` // because some objects might be missing. - let namespace = spark_application_namespace(spark_application)?; let templates = resolve( client, - namespace, &merge_template_options.template_names, merge_template_options.apply_strategy, ) @@ -323,19 +317,8 @@ pub(crate) async fn merge_application_templates( Ok(default_result) } -fn spark_application_namespace( - spark_application: &super::v1alpha1::SparkApplication, -) -> Result<&str, Error> { - spark_application - .metadata - .namespace - .as_deref() - .ok_or(Error::ObjectHasNoNamespace) -} - async fn resolve( client: &stackable_operator::client::Client, - namespace: &str, template_names: &[String], apply_strategy: TemplateApplyStrategy, ) -> Result, Error> { @@ -343,8 +326,7 @@ async fn resolve( return Ok(vec![]); } - let templates_api = - Api::::namespaced(client.as_kube_client(), namespace); + let templates_api = Api::::all(client.as_kube_client()); let mut resolved_templates = Vec::new(); for template_name in template_names { let template_res = templates_api @@ -452,53 +434,6 @@ mod tests { assert!(options.template_names.is_empty()); } - #[test] - fn spark_application_namespace_returns_namespace() { - let spark_application = - serde_yaml::from_str::(indoc! {r#" - --- - apiVersion: spark.stackable.tech/v1alpha1 - kind: SparkApplication - metadata: - name: app-with-templates - namespace: default - spec: - mode: cluster - mainApplicationFile: local:///app.py - sparkImage: - productVersion: "3.5.8" - "#}) - .unwrap(); - - assert_eq!( - spark_application_namespace(&spark_application).unwrap(), - "default" - ); - } - - #[test] - fn spark_application_namespace_returns_error_when_missing() { - let spark_application = - serde_yaml::from_str::(indoc! {r#" - --- - apiVersion: spark.stackable.tech/v1alpha1 - kind: SparkApplication - metadata: - name: app-with-templates - spec: - mode: cluster - mainApplicationFile: local:///app.py - sparkImage: - productVersion: "3.5.8" - "#}) - .unwrap(); - - assert!(matches!( - spark_application_namespace(&spark_application), - Err(Error::ObjectHasNoNamespace) - )); - } - impl RoundtripTestData for v1alpha1::SparkApplicationTemplateSpec { fn roundtrip_test_data() -> Vec { // SparkApplicationTemplateSpec is just a wrapper around SparkApplicationSpec From 87da728b2933e946b540239089c7a3d5e8982e1e Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 10 Jul 2026 14:30:36 +0200 Subject: [PATCH 4/6] Reapply "fix: make application templates namespaced (#694)" (#719) (#720) * Reapply "fix: make application templates namespaced (#694)" (#719) This reverts commit e1b21e163ab5e1f68089b329a3ecf5922c2cf4cd. * update changelog --- CHANGELOG.md | 5 +- Cargo.nix | 18 ++--- crate-hashes.json | 18 ++--- .../pages/usage-guide/app_templates.adoc | 14 +++- extra/crds.yaml | 2 +- .../src/crd/template_merger.rs | 41 ++++++++++++ rust/operator-binary/src/crd/template_spec.rs | 67 ++++++++++++++++++- 7 files changed, 142 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5954870a..e1ab7be9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,7 @@ All notable changes to this project will be documented in this file. - Internal operator refactoring: introduce dereference() and validate() steps in the reconciler for spark application, spark connect and spark history server([#687]). - test: Bump vector-aggregator to 0.55.0, replace /graphql call with gRPC call ([#689]). - Fix the `SparkApplication` CRD description, which incorrectly described it as a "Spark cluster stacklet" rather than a Spark application ([#705]). +- BREAKING: make application templates namespaced instead of cluster wide objects ([#694]), reverted in [#719] and readded in [#720]. [#674]: https://github.com/stackabletech/spark-k8s-operator/pull/674 [#679]: https://github.com/stackabletech/spark-k8s-operator/pull/679 @@ -39,11 +40,13 @@ All notable changes to this project will be documented in this file. [#687]: https://github.com/stackabletech/spark-k8s-operator/pull/687 [#689]: https://github.com/stackabletech/spark-k8s-operator/pull/689 [#692]: https://github.com/stackabletech/spark-k8s-operator/pull/692 +[#694]: https://github.com/stackabletech/spark-k8s-operator/pull/694 [#696]: https://github.com/stackabletech/spark-k8s-operator/pull/696 [#705]: https://github.com/stackabletech/spark-k8s-operator/pull/705 [#708]: https://github.com/stackabletech/spark-k8s-operator/pull/708 [#716]: https://github.com/stackabletech/spark-k8s-operator/pull/716 -[#XXXX]: https://github.com/stackabletech/spark-k8s-operator/pull/XXXX +[#719]: https://github.com/stackabletech/spark-k8s-operator/pull/719 +[#720]: https://github.com/stackabletech/spark-k8s-operator/pull/720 ## [26.3.0] - 2026-03-16 diff --git a/Cargo.nix b/Cargo.nix index 4bb9f18e..fd10413f 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -5109,7 +5109,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "k8s_version"; authors = [ @@ -9893,7 +9893,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "stackable_certs"; authors = [ @@ -9996,7 +9996,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "stackable_operator"; authors = [ @@ -10195,7 +10195,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; procMacro = true; libName = "stackable_operator_derive"; @@ -10230,7 +10230,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "stackable_shared"; authors = [ @@ -10417,7 +10417,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "stackable_telemetry"; authors = [ @@ -10527,7 +10527,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "stackable_versioned"; authors = [ @@ -10577,7 +10577,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; procMacro = true; libName = "stackable_versioned_macros"; @@ -10645,7 +10645,7 @@ rec { src = pkgs.fetchgit { url = "https://github.com/stackabletech/operator-rs.git"; rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + sha256 = "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps"; }; libName = "stackable_webhook"; authors = [ diff --git a/crate-hashes.json b/crate-hashes.json index 77adf52a..cd3917c3 100644 --- a/crate-hashes.json +++ b/crate-hashes.json @@ -1,12 +1,12 @@ { - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#k8s-version@0.1.3": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-certs@0.4.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator-derive@0.3.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator@0.113.3": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-shared@0.1.2": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-telemetry@0.6.5": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned-macros@0.11.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned@0.11.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-webhook@0.9.2": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#k8s-version@0.1.3": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-certs@0.4.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator-derive@0.3.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator@0.113.3": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-shared@0.1.2": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-telemetry@0.6.5": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned-macros@0.11.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned@0.11.1": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", + "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-webhook@0.9.2": "054p2mcinq3x4iykmc72nhga7kapc8ihkx9mnzqq3qdm2bk2qcps", "git+https://github.com/stackabletech/product-config.git?tag=0.8.0#product-config@0.8.0": "1dz70kapm2wdqcr7ndyjji0lhsl98bsq95gnb2lw487wf6yr7987" } \ No newline at end of file diff --git a/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc b/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc index 67a289fb..be8ebb9f 100644 --- a/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc +++ b/docs/modules/spark-k8s/pages/usage-guide/app_templates.adoc @@ -5,13 +5,23 @@ Spark application templates are used to define reusable configurations for Spark When you have many applications with similar configurations, templates can help you avoid duplication by grouping common settings together. Application templates are available for the `v1alpha1` version of the SparkApplication custom resource and share the exact same structure as the SparkApplication resource, but with some differences in the way the operator handles them: -1. Application templates are cluster wide resources, while Spark application resources are namespace-scoped. - This means that application templates can be used across multiple namespaces, while Spark application resources are limited to the namespace they are created in. +1. Application templates are namespace-scoped resources, just like Spark applications. + This means that a SparkApplication can only reference templates from its own namespace. 2. Application templates are not reconciled by the operator, but must be referenced from a SparkApplication resource to be applied. This means that changes to an application template will not automatically trigger updates to SparkApplication resources that reference it. 3. An application can reference multiple application templates, and the settings from these templates will be merged together. The merging order of the templates is indicated by their index in the reference list. The application fields have the highest precedence and will override any conflicting settings from the templates. This allows you to have a base template with common settings and then override specific settings in the application resource as needed. 4. Application template references are immutable in the sense that once applied to an application they cannot be changed again. Currently templates are applied upon the creation of the application, and any changes to the template references after that will be ignored. 5. Application and template CRDs must have the exact same versions. Currently only `v1alpha1` is supported. +== Migrating from cluster-scoped templates + +IMPORTANT: Application templates used to be cluster wide resources when they were first released. This was a mistake. Many users do not have the access rights to create cluster scoped resources and so the templates are now namespace scoped. + +If you are migrating from older installations where templates were treated as cluster-wide resources, account for the following: + +1. Recreate each template in every namespace where SparkApplications use it. +2. Keep template names consistent per namespace if you want the same application annotations to continue working. +3. Cross-namespace template references are no longer resolved; templates and applications must be in the same namespace. +4. Update GitOps/automation manifests to create templates as namespace-targeted resources before reconciling dependent SparkApplications. == Examples Applications use `metadata.annotations` to reference application templates as shown below: diff --git a/extra/crds.yaml b/extra/crds.yaml index cba2c2a9..56f60237 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -4657,7 +4657,7 @@ spec: shortNames: - sparkapptemplate singular: sparkapplicationtemplate - scope: Cluster + scope: Namespaced versions: - additionalPrinterColumns: [] name: v1alpha1 diff --git a/rust/operator-binary/src/crd/template_merger.rs b/rust/operator-binary/src/crd/template_merger.rs index b4fc48dc..b12bff0f 100644 --- a/rust/operator-binary/src/crd/template_merger.rs +++ b/rust/operator-binary/src/crd/template_merger.rs @@ -322,6 +322,47 @@ mod tests { assert_eq!(merged.spec.args, vec!["arg1", "arg2"]); } + #[test] + fn test_deep_merge_metadata_namespace_overlay_wins() { + let base = serde_yaml::from_str::(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: base-app + namespace: template-namespace + spec: + mode: cluster + mainApplicationFile: base.py + sparkImage: + productVersion: "3.5.0" + "#}) + .unwrap(); + + let overlay = serde_yaml::from_str::(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: overlay-app + namespace: app-namespace + spec: + mode: cluster + mainApplicationFile: overlay.py + sparkImage: + productVersion: "3.5.1" + "#}) + .unwrap(); + + let merged = deep_merge(&base, &overlay); + + assert_eq!( + merged.metadata.namespace, + Some("app-namespace".to_string()), + "overlay namespace should take precedence" + ); + } + #[test] fn test_deep_merge_spark_conf() { let base = serde_yaml::from_str::(indoc! {r#" diff --git a/rust/operator-binary/src/crd/template_spec.rs b/rust/operator-binary/src/crd/template_spec.rs index da5fdd05..7312abe3 100644 --- a/rust/operator-binary/src/crd/template_spec.rs +++ b/rust/operator-binary/src/crd/template_spec.rs @@ -44,6 +44,9 @@ pub enum Error { template_name: String, source: stackable_operator::kube::Error, }, + + #[snafu(display("object has no namespace"))] + ObjectHasNoNamespace, } #[versioned( @@ -66,6 +69,7 @@ pub mod versioned { group = "spark.stackable.tech", plural = "sparkapptemplates", shortname = "sparkapptemplate", + namespaced ))] #[derive(Clone, CustomResource, Debug, Deserialize, JsonSchema, Serialize)] #[serde(rename_all = "camelCase")] @@ -263,8 +267,10 @@ pub(crate) async fn merge_application_templates( // In the future if we support additional strategies in addition to "enforce", // this list might not be identical to the one in `merge_template_options` // because some objects might be missing. + let namespace = spark_application_namespace(spark_application)?; let templates = resolve( client, + namespace, &merge_template_options.template_names, merge_template_options.apply_strategy, ) @@ -317,8 +323,19 @@ pub(crate) async fn merge_application_templates( Ok(default_result) } +fn spark_application_namespace( + spark_application: &super::v1alpha1::SparkApplication, +) -> Result<&str, Error> { + spark_application + .metadata + .namespace + .as_deref() + .ok_or(Error::ObjectHasNoNamespace) +} + async fn resolve( client: &stackable_operator::client::Client, + namespace: &str, template_names: &[String], apply_strategy: TemplateApplyStrategy, ) -> Result, Error> { @@ -326,7 +343,8 @@ async fn resolve( return Ok(vec![]); } - let templates_api = Api::::all(client.as_kube_client()); + let templates_api = + Api::::namespaced(client.as_kube_client(), namespace); let mut resolved_templates = Vec::new(); for template_name in template_names { let template_res = templates_api @@ -434,6 +452,53 @@ mod tests { assert!(options.template_names.is_empty()); } + #[test] + fn spark_application_namespace_returns_namespace() { + let spark_application = + serde_yaml::from_str::(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: app-with-templates + namespace: default + spec: + mode: cluster + mainApplicationFile: local:///app.py + sparkImage: + productVersion: "3.5.8" + "#}) + .unwrap(); + + assert_eq!( + spark_application_namespace(&spark_application).unwrap(), + "default" + ); + } + + #[test] + fn spark_application_namespace_returns_error_when_missing() { + let spark_application = + serde_yaml::from_str::(indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: app-with-templates + spec: + mode: cluster + mainApplicationFile: local:///app.py + sparkImage: + productVersion: "3.5.8" + "#}) + .unwrap(); + + assert!(matches!( + spark_application_namespace(&spark_application), + Err(Error::ObjectHasNoNamespace) + )); + } + impl RoundtripTestData for v1alpha1::SparkApplicationTemplateSpec { fn roundtrip_test_data() -> Vec { // SparkApplicationTemplateSpec is just a wrapper around SparkApplicationSpec From 6c5a3cb88a885c3faa512b902835a6cb5ae73431 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Thu, 16 Jul 2026 15:58:25 +0200 Subject: [PATCH 5/6] feat: support Spark 3.5.8 --- CHANGELOG.md | 2 +- .../pages/usage-guide/openlineage.adoc | 4 +- rust/operator-binary/src/config/jvm.rs | 141 ++++++++++++------ rust/operator-binary/src/crd/constants.rs | 18 ++- rust/operator-binary/src/crd/mod.rs | 124 ++++++++++++--- .../src/spark_k8s_controller.rs | 1 + 6 files changed, 216 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e1ab7be9..e2416911 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ All notable changes to this project will be documented in this file. - BREAKING: Add required CLI argument and env var to set the image repository used to construct final product image names: `IMAGE_REPOSITORY` (`--image-repository`), eg. `oci.example.org/my/namespace` ([#684]). - Support for Spark `4.1.2` ([#708]) -- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap), a stable job name, and the required `--add-opens` JVM flag ([#XXXX]). +- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap) and a stable job name. The Scala build of the listener jar and the `--add-opens java.base/java.security` JVM flag are selected by the Spark version: the `--add-opens` flag is emitted only on Spark 4.x (it is unnecessary on the Spark 3.5.x images and must not be set there) ([#XXXX]). ### Fixed diff --git a/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc index 569311d5..9974021c 100644 --- a/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc +++ b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc @@ -6,10 +6,10 @@ The Spark operator can automatically emit lineage events for a `SparkApplication When enabled, the operator injects everything required to make the https://openlineage.io/docs/integrations/spark/[OpenLineage Spark listener] work: * the OpenLineage Spark listener (appended to `spark.extraListeners`, so any listener you configure yourself is preserved), -* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg), +* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg). The operator selects the jar built for the image's Scala version automatically (Scala 2.13 for Spark 4.x, Scala 2.12 for Spark 3.5.x), * an HTTP transport pointing at the backend endpoint, * a stable lineage namespace and job name (see <>), and -* the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors. +* on Spark 4.x only, the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors. This flag is not added on the Spark 3.5.x images, where it is unnecessary (and, on their JVM, potentially harmful). == Enabling OpenLineage diff --git a/rust/operator-binary/src/config/jvm.rs b/rust/operator-binary/src/config/jvm.rs index 21f19eff..0c45972a 100644 --- a/rust/operator-binary/src/config/jvm.rs +++ b/rust/operator-binary/src/config/jvm.rs @@ -1,11 +1,13 @@ use stackable_operator::crd::s3; use crate::crd::{ + Error, constants::{ JVM_SECURITY_PROPERTIES_FILE, OPENLINEAGE_ADD_OPENS, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE, VOLUME_MOUNT_PATH_LOG_CONFIG, }, logdir::ResolvedLogDir, + spark_major_version, tlscerts::tls_secret_names, v1alpha1::SparkApplication, }; @@ -16,11 +18,15 @@ use crate::crd::{ /// /// Returns `(driver, executor)`: the operator-generated base arguments with the role's /// `jvmArgumentOverrides` applied on top. +/// +/// `product_version` is the resolved Spark product version (e.g. `4.1.2`); it gates the +/// version-specific OpenLineage `--add-opens` flag. pub fn construct_extra_java_options( spark_application: &SparkApplication, s3_conn: &Option, log_dir: &Option, -) -> (String, String) { + product_version: &str, +) -> Result<(String, String), Error> { // Note (@sbernauer): As of 2025-03-04, we did not set any heap related JVM arguments, so I // kept the implementation as is. We can always re-visit this as needed. @@ -39,12 +45,10 @@ pub fn construct_extra_java_options( // OpenLineage on Spark 4.x needs `java.base/java.security` opened to the unnamed module, // otherwise the driver throws a non-fatal `InaccessibleObjectException` and silently degrades // extension-interface lineage (MVP §7). Added to both driver and executor. - if spark_application - .spec - .open_lineage - .as_ref() - .is_some_and(|open_lineage| open_lineage.enabled) - { + // + // This is scoped to Spark 4.x: the flag is a no-op — and on some JVMs a startup error — on the + // JDK 11/17 Spark 3.x images the operator also ships, so it must not be emitted there. + if spark_application.openlineage_enabled() && spark_major_version(product_version)? >= 4 { jvm_args.push(OPENLINEAGE_ADD_OPENS.to_string()); } @@ -68,16 +72,24 @@ pub fn construct_extra_java_options( None => jvm_args.clone(), }; - (driver.join(" "), executor.join(" ")) + Ok((driver.join(" "), executor.join(" "))) } #[cfg(test)] mod tests { + use rstest::rstest; + use super::*; + fn spark_app_from_yaml(input: &str) -> SparkApplication { + let deserializer = serde_yaml::Deserializer::from_str(input); + serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap() + } + #[test] fn test_construct_jvm_arguments_defaults() { - let input = r#" + let spark_app = spark_app_from_yaml( + r#" apiVersion: spark.stackable.tech/v1alpha1 kind: SparkApplication metadata: @@ -86,14 +98,11 @@ mod tests { mode: cluster mainApplicationFile: test.py sparkImage: - productVersion: 1.2.3 - "#; - - let deserializer = serde_yaml::Deserializer::from_str(input); - let spark_app: SparkApplication = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + productVersion: 4.1.2 + "#, + ); let (driver_extra_java_options, executor_extra_java_options) = - construct_extra_java_options(&spark_app, &None, &None); + construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap(); assert_eq!( driver_extra_java_options, @@ -107,7 +116,8 @@ mod tests { #[test] fn test_construct_jvm_argument_overrides() { - let input = r#" + let spark_app = spark_app_from_yaml( + r#" apiVersion: spark.stackable.tech/v1alpha1 kind: SparkApplication metadata: @@ -116,7 +126,7 @@ mod tests { mode: cluster mainApplicationFile: test.py sparkImage: - productVersion: 1.2.3 + productVersion: 4.1.2 driver: jvmArgumentOverrides: add: @@ -127,13 +137,10 @@ mod tests { - -Dhttps.proxyHost=from-executor removeRegex: - -Djava.security.properties=.* - "#; - - let deserializer = serde_yaml::Deserializer::from_str(input); - let spark_app: SparkApplication = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + "#, + ); let (driver_extra_java_options, executor_extra_java_options) = - construct_extra_java_options(&spark_app, &None, &None); + construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap(); assert_eq!( driver_extra_java_options, @@ -145,29 +152,71 @@ mod tests { ); } - #[test] - fn test_construct_jvm_arguments_openlineage_add_opens() { - let input = r#" - apiVersion: spark.stackable.tech/v1alpha1 - kind: SparkApplication - metadata: - name: spark-example - spec: - mode: cluster - mainApplicationFile: test.py - sparkImage: - productVersion: 1.2.3 - openLineage: - enabled: true - "#; - - let deserializer = serde_yaml::Deserializer::from_str(input); - let spark_app: SparkApplication = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + const OPENLINEAGE_ENABLED: &str = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: PLACEHOLDER + openLineage: + enabled: true + "#; + + const OPENLINEAGE_DISABLED: &str = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: PLACEHOLDER + openLineage: + enabled: false + "#; + + const OPENLINEAGE_ABSENT: &str = r#" + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: spark-example + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: PLACEHOLDER + "#; + + /// `--add-opens` is emitted only on Spark 4.x with OpenLineage enabled — never on the Scala + /// 2.12 / JDK 17 Spark 3.5.x images, and never when OpenLineage is disabled or absent. + #[rstest] + #[case::enabled_spark_3(OPENLINEAGE_ENABLED, "3.5.8", false)] + #[case::enabled_spark_4(OPENLINEAGE_ENABLED, "4.1.2", true)] + #[case::disabled_spark_4(OPENLINEAGE_DISABLED, "4.1.2", false)] + #[case::absent_spark_4(OPENLINEAGE_ABSENT, "4.1.2", false)] + fn test_openlineage_add_opens_is_version_gated( + #[case] yaml_template: &str, + #[case] product_version: &str, + #[case] expect_add_opens: bool, + ) { + let spark_app = spark_app_from_yaml(&yaml_template.replace("PLACEHOLDER", product_version)); let (driver_extra_java_options, executor_extra_java_options) = - construct_extra_java_options(&spark_app, &None, &None); + construct_extra_java_options(&spark_app, &None, &None, product_version).unwrap(); - assert!(driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS)); - assert!(executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS)); + assert_eq!( + driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS), + expect_add_opens, + "driver --add-opens presence mismatch for Spark {product_version}" + ); + assert_eq!( + executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS), + expect_add_opens, + "executor --add-opens presence mismatch for Spark {product_version}" + ); } } diff --git a/rust/operator-binary/src/crd/constants.rs b/rust/operator-binary/src/crd/constants.rs index 50b5cf85..989ffa38 100644 --- a/rust/operator-binary/src/crd/constants.rs +++ b/rust/operator-binary/src/crd/constants.rs @@ -108,15 +108,25 @@ pub const DEFAULT_SUBMIT_JOB_RETRY_ON_FAILURE_COUNT: u16 = 0; /// The OpenLineage Spark listener, appended to `spark.extraListeners` when OpenLineage is enabled. pub const OPENLINEAGE_LISTENER_CLASS: &str = "io.openlineage.spark.agent.OpenLineageSparkListener"; +/// Version of the OpenLineage Spark listener jar baked into the Spark images. +/// +/// IMPORTANT: MUST stay in sync with the `openlineage-spark-version` build-argument in +/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar on disk). +pub const OPENLINEAGE_SPARK_JAR_VERSION: &str = "1.51.0"; + /// `local://` URI of the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so /// it shares the (child) classloader with `--packages` connectors. Delivering it this way — rather /// than on `extraClassPath` (system classloader) — is what lets OpenLineage's connector probes /// succeed; see the classpath discussion in the OpenLineage usage guide. /// -/// IMPORTANT: the version MUST stay in sync with the `openlineage-spark-version` build-argument in -/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar at this path). -pub const OPENLINEAGE_JAR_LOCAL_URI: &str = - "local:///stackable/spark/openlineage/openlineage-spark_2.13-1.51.0.jar"; +/// The jar's Scala suffix depends on the Spark image: Stackable ships Spark 4.x as Scala 2.13 and +/// Spark 3.5.x as Scala 2.12, so the operator selects the matching jar by Spark major version. +/// `scala_binary_version` is e.g. `"2.13"` or `"2.12"`. +pub fn openlineage_jar_local_uri(scala_binary_version: &str) -> String { + format!( + "local:///stackable/spark/openlineage/openlineage-spark_{scala_binary_version}-{OPENLINEAGE_SPARK_JAR_VERSION}.jar" + ) +} /// The key the OpenLineage discovery ConfigMap must hold, carrying the backend base URL. Mirrors the /// `ADDRESS` contract of the existing `vectorAggregatorConfigMapName` discovery ConfigMap. diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 53ef35db..4d2b4d67 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -121,6 +121,14 @@ pub enum Error { `spark.openlineage.transport.url` in `sparkConf`" ))] MissingOpenLineageEndpoint, + + #[snafu(display( + "failed to parse the Spark major version from the resolved product version {product_version:?}" + ))] + UnparseableSparkVersion { + source: std::num::ParseIntError, + product_version: String, + }, } pub type SparkApplicationJobRoleType = @@ -333,6 +341,21 @@ fn append_conf_csv(submit_conf: &mut BTreeMap, key: &str, value: } } +/// Parses the Spark major version from a resolved product version string such as `4.1.2` or +/// `3.5.8`. Used to gate version-specific OpenLineage wiring: the `--add-opens` JVM flag (only +/// needed on the JDK 17+/Spark 4.x line) and the Scala build of the baked OpenLineage jar +/// (Stackable ships Spark 4.x as Scala 2.13 and Spark 3.5.x as Scala 2.12). +pub(crate) fn spark_major_version(product_version: &str) -> Result { + product_version + .split('.') + .next() + .unwrap_or_default() + .parse::() + .context(UnparseableSparkVersionSnafu { + product_version: product_version.to_string(), + }) +} + impl v1alpha1::SparkApplication { /// Returns if this [`SparkApplication`] has already created a Kubernetes Job doing the actual `spark-submit`. /// @@ -629,11 +652,21 @@ impl v1alpha1::SparkApplication { mounts } + /// True when OpenLineage emission is configured and enabled. Off is expressible two ways: + /// the `openLineage` block absent, or present with `enabled: false`. + pub fn openlineage_enabled(&self) -> bool { + self.spec + .open_lineage + .as_ref() + .is_some_and(|open_lineage| open_lineage.enabled) + } + pub fn build_command( &self, s3conn: &Option, log_dir: &Option, spark_image: &str, + product_version: &str, openlineage_endpoint: Option<&str>, ) -> Result, Error> { // mandatory properties @@ -733,7 +766,7 @@ impl v1alpha1::SparkApplication { } let (driver_extra_java_options, executor_extra_java_options) = - construct_extra_java_options(self, s3conn, log_dir); + construct_extra_java_options(self, s3conn, log_dir, product_version)?; submit_cmd.extend(vec![ format!("--conf spark.driver.extraJavaOptions=\"{driver_extra_java_options}\""), format!("--conf spark.executor.extraJavaOptions=\"{executor_extra_java_options}\""), @@ -853,18 +886,26 @@ impl v1alpha1::SparkApplication { // OpenLineage append-merge keys: combine our values with any user-provided ones (already // merged into `submit_conf` above) so neither clobbers the other. - if self - .spec - .open_lineage - .as_ref() - .is_some_and(|open_lineage| open_lineage.enabled) - { + if self.openlineage_enabled() { append_conf_csv( &mut submit_conf, "spark.extraListeners", OPENLINEAGE_LISTENER_CLASS, ); - append_conf_csv(&mut submit_conf, "spark.jars", OPENLINEAGE_JAR_LOCAL_URI); + // Select the jar built for the image's Scala version: Stackable ships Spark 4.x as + // Scala 2.13 and Spark 3.5.x as Scala 2.12. Referencing the wrong build would leave a + // non-existent path in `spark.jars`, which Spark ignores silently (no listener, no + // events), so this must track the actual image. + let scala_binary_version = if spark_major_version(product_version)? >= 4 { + "2.13" + } else { + "2.12" + }; + append_conf_csv( + &mut submit_conf, + "spark.jars", + &openlineage_jar_local_uri(scala_binary_version), + ); } // ...before being added to the command collection @@ -1738,13 +1779,24 @@ spec: assert_eq!(got, expected); } - /// Builds a `SparkApplication` from YAML and returns the assembled spark-submit command string, - /// with `openlineage_endpoint` standing in for the value the controller resolves from the + /// Builds a `SparkApplication` from YAML and returns the assembled spark-submit command string. + /// `product_version` stands in for the resolved Spark product version (gates the version-specific + /// OpenLineage wiring) and `openlineage_endpoint` for the value the controller resolves from the /// discovery ConfigMap. - fn build_command_with_openlineage(yaml: &str, openlineage_endpoint: Option<&str>) -> String { + fn build_command_with_openlineage( + yaml: &str, + product_version: &str, + openlineage_endpoint: Option<&str>, + ) -> String { let spark_application = serde_yaml::from_str::(yaml).unwrap(); spark_application - .build_command(&None, &None, "test-image", openlineage_endpoint) + .build_command( + &None, + &None, + "test-image", + product_version, + openlineage_endpoint, + ) .unwrap() .join(" ") } @@ -1767,8 +1819,11 @@ spec: #[test] fn test_openlineage_injects_conf_when_enabled() { - let command = - build_command_with_openlineage(OPENLINEAGE_ENABLED_APP, Some("http://marquez:5000")); + let command = build_command_with_openlineage( + OPENLINEAGE_ENABLED_APP, + "4.1.2", + Some("http://marquez:5000"), + ); assert!(command.contains(r#"--conf "spark.openlineage.transport.type=http""#)); assert!( @@ -1781,10 +1836,12 @@ spec: assert!(command.contains(&format!( r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# ))); + // Spark 4.x is Scala 2.13, so the 2.13 build of the jar is referenced. assert!(command.contains(&format!( - r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# + r#"--conf "spark.jars={}""#, + openlineage_jar_local_uri("2.13") ))); - // --add-opens reaches both driver and executor. + // --add-opens reaches both driver and executor on Spark 4.x. assert_eq!( command.matches(OPENLINEAGE_ADD_OPENS).count(), 2, @@ -1792,6 +1849,31 @@ spec: ); } + #[test] + fn test_openlineage_selects_scala_2_12_jar_and_no_add_opens_on_spark_3() { + // On the Scala 2.12 / JDK 17 Spark 3.5.x images, the operator must reference the 2.12 build + // of the jar (the 2.13 path would not exist in the image) and must NOT emit `--add-opens`. + let command = build_command_with_openlineage( + OPENLINEAGE_ENABLED_APP, + "3.5.8", + Some("http://marquez:5000"), + ); + + assert!(command.contains(&format!( + r#"--conf "spark.jars={}""#, + openlineage_jar_local_uri("2.12") + ))); + assert!( + !command.contains(OPENLINEAGE_ADD_OPENS), + "--add-opens must not be emitted on Spark 3.x" + ); + // The rest of the OpenLineage wiring is still present on Spark 3.x. + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# + ))); + assert!(command.contains(r#"--conf "spark.openlineage.transport.type=http""#)); + } + #[rstest] #[case::absent(indoc! {r#" --- @@ -1823,7 +1905,7 @@ spec: "#})] fn test_openlineage_injects_nothing_when_absent_or_disabled(#[case] yaml: &str) { // Endpoint is supplied to prove it is ignored, not that it is simply missing. - let command = build_command_with_openlineage(yaml, Some("http://marquez:5000")); + let command = build_command_with_openlineage(yaml, "4.1.2", Some("http://marquez:5000")); assert!(!command.contains("openlineage")); assert!(!command.contains("OpenLineageSparkListener")); @@ -1850,7 +1932,7 @@ spec: sparkConf: spark.extraListeners: com.example.CustomListener "#}; - let command = build_command_with_openlineage(yaml, Some("http://marquez:5000")); + let command = build_command_with_openlineage(yaml, "4.1.2", Some("http://marquez:5000")); assert!(command.contains(&format!( r#"--conf "spark.extraListeners=com.example.CustomListener,{OPENLINEAGE_LISTENER_CLASS}""# @@ -1877,7 +1959,7 @@ spec: spark.openlineage.transport.url: http://custom:1234 spark.openlineage.namespace: custom-ns "#}; - let command = build_command_with_openlineage(yaml, Some("http://marquez:5000")); + let command = build_command_with_openlineage(yaml, "4.1.2", Some("http://marquez:5000")); assert!(command.contains(r#"--conf "spark.openlineage.transport.url=http://custom:1234""#)); assert!(!command.contains("http://marquez:5000")); @@ -1889,7 +1971,7 @@ spec: // Enabled, but neither a resolved endpoint nor a sparkConf transport.url override. let spark_application = serde_yaml::from_str::(OPENLINEAGE_ENABLED_APP).unwrap(); - let result = spark_application.build_command(&None, &None, "test-image", None); + let result = spark_application.build_command(&None, &None, "test-image", "4.1.2", None); assert!(matches!(result, Err(Error::MissingOpenLineageEndpoint))); } @@ -1914,7 +1996,7 @@ spec: sparkConf: spark.openlineage.transport.url: http://custom:1234 "#}; - let command = build_command_with_openlineage(yaml, None); + let command = build_command_with_openlineage(yaml, "4.1.2", None); assert!(command.contains(r#"--conf "spark.openlineage.transport.url=http://custom:1234""#)); } diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs index 10867520..c0f34c3f 100644 --- a/rust/operator-binary/src/spark_k8s_controller.rs +++ b/rust/operator-binary/src/spark_k8s_controller.rs @@ -261,6 +261,7 @@ pub async fn reconcile( opt_s3conn, logdir, &resolved_product_image.image, + &resolved_product_image.product_version, openlineage_endpoint.as_deref(), ) .context(BuildCommandSnafu)?; From 69d746aec9c270d20b1dec9bad556be1e9e9dc99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Thu, 16 Jul 2026 16:45:19 +0200 Subject: [PATCH 6/6] feat: harmonise OpenLineage jar name via a stable image symlink --- CHANGELOG.md | 2 +- .../pages/usage-guide/openlineage.adoc | 2 +- rust/operator-binary/src/crd/constants.rs | 20 ++++-------- rust/operator-binary/src/crd/mod.rs | 31 ++++++------------- 4 files changed, 17 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e2416911..88bc364a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ All notable changes to this project will be documented in this file. - BREAKING: Add required CLI argument and env var to set the image repository used to construct final product image names: `IMAGE_REPOSITORY` (`--image-repository`), eg. `oci.example.org/my/namespace` ([#684]). - Support for Spark `4.1.2` ([#708]) -- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap) and a stable job name. The Scala build of the listener jar and the `--add-opens java.base/java.security` JVM flag are selected by the Spark version: the `--add-opens` flag is emitted only on Spark 4.x (it is unnecessary on the Spark 3.5.x images and must not be set there) ([#XXXX]). +- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap) and a stable job name. The listener jar is referenced through a stable, version- and Scala-independent symlink maintained by the Spark image, so the operator tracks neither. The `--add-opens java.base/java.security` JVM flag is emitted only on Spark 4.x (it is unnecessary on the Spark 3.5.x images and must not be set there) ([#XXXX]). ### Fixed diff --git a/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc index 9974021c..e8aa52f3 100644 --- a/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc +++ b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc @@ -6,7 +6,7 @@ The Spark operator can automatically emit lineage events for a `SparkApplication When enabled, the operator injects everything required to make the https://openlineage.io/docs/integrations/spark/[OpenLineage Spark listener] work: * the OpenLineage Spark listener (appended to `spark.extraListeners`, so any listener you configure yourself is preserved), -* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg). The operator selects the jar built for the image's Scala version automatically (Scala 2.13 for Spark 4.x, Scala 2.12 for Spark 3.5.x), +* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg). The operator references a stable symlink maintained by the image, which points at the correct build for that image's Spark version (Scala 2.13 for Spark 4.x, Scala 2.12 for Spark 3.5.x), so neither the jar version nor its Scala suffix has to be tracked in the operator, * an HTTP transport pointing at the backend endpoint, * a stable lineage namespace and job name (see <>), and * on Spark 4.x only, the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors. This flag is not added on the Spark 3.5.x images, where it is unnecessary (and, on their JVM, potentially harmful). diff --git a/rust/operator-binary/src/crd/constants.rs b/rust/operator-binary/src/crd/constants.rs index 989ffa38..855fe87d 100644 --- a/rust/operator-binary/src/crd/constants.rs +++ b/rust/operator-binary/src/crd/constants.rs @@ -108,25 +108,17 @@ pub const DEFAULT_SUBMIT_JOB_RETRY_ON_FAILURE_COUNT: u16 = 0; /// The OpenLineage Spark listener, appended to `spark.extraListeners` when OpenLineage is enabled. pub const OPENLINEAGE_LISTENER_CLASS: &str = "io.openlineage.spark.agent.OpenLineageSparkListener"; -/// Version of the OpenLineage Spark listener jar baked into the Spark images. -/// -/// IMPORTANT: MUST stay in sync with the `openlineage-spark-version` build-argument in -/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar on disk). -pub const OPENLINEAGE_SPARK_JAR_VERSION: &str = "1.51.0"; - /// `local://` URI of the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so /// it shares the (child) classloader with `--packages` connectors. Delivering it this way — rather /// than on `extraClassPath` (system classloader) — is what lets OpenLineage's connector probes /// succeed; see the classpath discussion in the OpenLineage usage guide. /// -/// The jar's Scala suffix depends on the Spark image: Stackable ships Spark 4.x as Scala 2.13 and -/// Spark 3.5.x as Scala 2.12, so the operator selects the matching jar by Spark major version. -/// `scala_binary_version` is e.g. `"2.13"` or `"2.12"`. -pub fn openlineage_jar_local_uri(scala_binary_version: &str) -> String { - format!( - "local:///stackable/spark/openlineage/openlineage-spark_{scala_binary_version}-{OPENLINEAGE_SPARK_JAR_VERSION}.jar" - ) -} +/// This is a stable, version- and Scala-independent symlink the image maintains (mirroring the +/// `jmx_prometheus_javaagent.jar` pattern): the image bakes the correct `openlineage-spark_-.jar` +/// for its Spark build and symlinks it here. Referencing the symlink decouples the operator from the +/// jar's version and Scala suffix, so bumping either only touches `docker-images/spark-k8s`. +pub const OPENLINEAGE_JAR_LOCAL_URI: &str = + "local:///stackable/spark/openlineage/openlineage-spark.jar"; /// The key the OpenLineage discovery ConfigMap must hold, carrying the backend base URL. Mirrors the /// `ADDRESS` contract of the existing `vectorAggregatorConfigMapName` discovery ConfigMap. diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 4d2b4d67..b9da7e25 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -892,20 +892,9 @@ impl v1alpha1::SparkApplication { "spark.extraListeners", OPENLINEAGE_LISTENER_CLASS, ); - // Select the jar built for the image's Scala version: Stackable ships Spark 4.x as - // Scala 2.13 and Spark 3.5.x as Scala 2.12. Referencing the wrong build would leave a - // non-existent path in `spark.jars`, which Spark ignores silently (no listener, no - // events), so this must track the actual image. - let scala_binary_version = if spark_major_version(product_version)? >= 4 { - "2.13" - } else { - "2.12" - }; - append_conf_csv( - &mut submit_conf, - "spark.jars", - &openlineage_jar_local_uri(scala_binary_version), - ); + // Reference the stable symlink the image maintains; it points at the correct + // Scala/version build for this image, so the operator needs to know neither. + append_conf_csv(&mut submit_conf, "spark.jars", OPENLINEAGE_JAR_LOCAL_URI); } // ...before being added to the command collection @@ -1836,10 +1825,9 @@ spec: assert!(command.contains(&format!( r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# ))); - // Spark 4.x is Scala 2.13, so the 2.13 build of the jar is referenced. + // The jar is referenced via the stable, Scala/version-independent image symlink. assert!(command.contains(&format!( - r#"--conf "spark.jars={}""#, - openlineage_jar_local_uri("2.13") + r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# ))); // --add-opens reaches both driver and executor on Spark 4.x. assert_eq!( @@ -1850,9 +1838,9 @@ spec: } #[test] - fn test_openlineage_selects_scala_2_12_jar_and_no_add_opens_on_spark_3() { - // On the Scala 2.12 / JDK 17 Spark 3.5.x images, the operator must reference the 2.12 build - // of the jar (the 2.13 path would not exist in the image) and must NOT emit `--add-opens`. + fn test_openlineage_stable_jar_uri_and_no_add_opens_on_spark_3() { + // On the JDK 17 Spark 3.5.x images the operator references the same stable jar symlink (the + // image points it at the Scala 2.12 build) and must NOT emit `--add-opens`. let command = build_command_with_openlineage( OPENLINEAGE_ENABLED_APP, "3.5.8", @@ -1860,8 +1848,7 @@ spec: ); assert!(command.contains(&format!( - r#"--conf "spark.jars={}""#, - openlineage_jar_local_uri("2.12") + r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# ))); assert!( !command.contains(OPENLINEAGE_ADD_OPENS),