feat: openlineage.tls now also installs the server CA if necesary

Author: razvan

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 built from the `host` and `port` fields), 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 built from the `host` and `port` fields), a stable job name, and the required `--add-opens` JVM flag. When the connection configures TLS server verification against a `secretClass` CA, that certificate is mounted into the driver and added to its trust store ([#XXXX]).
 
 ### Fixed
 

docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc

@@ -62,7 +62,9 @@ spec:
 ----
 
 The transport scheme is `https` when `tls.verification.server` is configured, otherwise `http`.
-Explicit certificate handling beyond selecting the scheme will be added later.
+When the connection verifies the server against a `secretClass` CA (as in the commented `tls` block
+above), the operator mounts that SecretClass certificate into the driver and adds it to the driver's
+trust store, so the OpenLineage listener trusts the backend's certificate.
 
 You can still override any of the injected values — including the full `spark.openlineage.transport.url` — through `sparkConf`.
 

rust/operator-binary/src/config/jvm.rs

@@ -1,4 +1,4 @@
-use stackable_operator::crd::s3;
+use stackable_operator::crd::{openlineage, s3};
 
 use crate::crd::{
     Error,
@@ -26,6 +26,7 @@ pub fn construct_extra_java_options(
     s3_conn: &Option<s3::v1alpha1::ConnectionSpec>,
     log_dir: &Option<ResolvedLogDir>,
     product_version: &str,
+    open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>,
 ) -> 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.
@@ -34,7 +35,7 @@ pub fn construct_extra_java_options(
         "-Djava.security.properties={VOLUME_MOUNT_PATH_LOG_CONFIG}/{JVM_SECURITY_PROPERTIES_FILE}"
     )];
 
-    if tls_secret_names(s3_conn, log_dir).is_some() {
+    if tls_secret_names(s3_conn, log_dir, open_lineage_conn).is_some() {
         jvm_args.extend([
             format!("-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12"),
             format!("-Djavax.net.ssl.trustStorePassword={STACKABLE_TLS_STORE_PASSWORD}"),
@@ -102,7 +103,7 @@ mod tests {
         "#,
         );
         let (driver_extra_java_options, executor_extra_java_options) =
-            construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap();
+            construct_extra_java_options(&spark_app, &None, &None, "4.1.2", None).unwrap();
 
         assert_eq!(
             driver_extra_java_options,
@@ -140,7 +141,7 @@ mod tests {
         "#,
         );
         let (driver_extra_java_options, executor_extra_java_options) =
-            construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap();
+            construct_extra_java_options(&spark_app, &None, &None, "4.1.2", None).unwrap();
 
         assert_eq!(
             driver_extra_java_options,
@@ -194,7 +195,7 @@ mod tests {
     ) {
         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, product_version).unwrap();
+            construct_extra_java_options(&spark_app, &None, &None, product_version, None).unwrap();
 
         assert_eq!(
             driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS),

rust/operator-binary/src/crd/mod.rs

@@ -337,6 +337,7 @@ impl v1alpha1::SparkApplication {
         logdir: &Option<ResolvedLogDir>,
         log_config_map: Option<&str>,
         requested_secret_lifetime: &Duration,
+        open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>,
     ) -> Result<Vec<Volume>, Error> {
         // Collect the volumes in a map keyed by name to avoid duplicates.
         // Duplicates can happen when the the S3 credentials volume and the history server log directory use the same secret class.
@@ -417,7 +418,7 @@ impl v1alpha1::SparkApplication {
                     .build(),
             );
         }
-        if let Some(cert_secrets) = tlscerts::tls_secret_names(s3conn, logdir) {
+        if let Some(cert_secrets) = tlscerts::tls_secret_names(s3conn, logdir, open_lineage_conn) {
             result.insert(
                 STACKABLE_TRUST_STORE_NAME.to_string(),
                 VolumeBuilder::new(STACKABLE_TRUST_STORE_NAME)
@@ -470,6 +471,7 @@ impl v1alpha1::SparkApplication {
         &self,
         s3conn: &Option<s3::v1alpha1::ConnectionSpec>,
         logdir: &Option<ResolvedLogDir>,
+        open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>,
     ) -> Vec<VolumeMount> {
         let mut tmpl_mounts = vec![
             VolumeMount {
@@ -484,7 +486,8 @@ impl v1alpha1::SparkApplication {
             },
         ];
 
-        tmpl_mounts = self.add_common_volume_mounts(tmpl_mounts, s3conn, logdir, false);
+        tmpl_mounts =
+            self.add_common_volume_mounts(tmpl_mounts, s3conn, logdir, false, open_lineage_conn);
 
         if let Some(CommonConfiguration {
             config:
@@ -510,6 +513,7 @@ impl v1alpha1::SparkApplication {
         s3conn: &Option<s3::v1alpha1::ConnectionSpec>,
         logdir: &Option<ResolvedLogDir>,
         logging_enabled: bool,
+        open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>,
     ) -> Vec<VolumeMount> {
         if self.spec.image.is_some() {
             mounts.push(VolumeMount {
@@ -568,7 +572,7 @@ impl v1alpha1::SparkApplication {
                 ..VolumeMount::default()
             });
         }
-        if let Some(cert_secrets) = tlscerts::tls_secret_names(s3conn, logdir) {
+        if let Some(cert_secrets) = tlscerts::tls_secret_names(s3conn, logdir, open_lineage_conn) {
             mounts.push(VolumeMount {
                 name: STACKABLE_TRUST_STORE_NAME.into(),
                 mount_path: STACKABLE_TRUST_STORE.into(),
@@ -606,17 +610,18 @@ impl v1alpha1::SparkApplication {
         let name = self.metadata.name.clone().context(ObjectHasNoNameSnafu)?;
 
         // Commands needed to build the p12 trust store from the secret class certs configured for
-        // S3 connections.
-        let build_truststore_commands = match tlscerts::tls_secret_names(s3conn, log_dir) {
-            Some(cert_secrets) => {
-                let mut build_truststore_str =
-                    vec![tlscerts::convert_system_trust_store_to_pkcs12()];
-                build_truststore_str
-                    .extend(cert_secrets.into_iter().map(tlscerts::import_truststore));
-                format!("{};", build_truststore_str.join(" && "))
-            }
-            None => "".to_string(),
-        };
+        // S3 connections, the log directory, and the OpenLineage backend connection.
+        let build_truststore_commands =
+            match tlscerts::tls_secret_names(s3conn, log_dir, open_lineage_conn) {
+                Some(cert_secrets) => {
+                    let mut build_truststore_str =
+                        vec![tlscerts::convert_system_trust_store_to_pkcs12()];
+                    build_truststore_str
+                        .extend(cert_secrets.into_iter().map(tlscerts::import_truststore));
+                    format!("{};", build_truststore_str.join(" && "))
+                }
+                None => "".to_string(),
+            };
 
         let mut submit_cmd = vec![
             format!(
@@ -698,7 +703,13 @@ impl v1alpha1::SparkApplication {
         }
 
         let (driver_extra_java_options, executor_extra_java_options) =
-            construct_extra_java_options(self, s3conn, log_dir, product_version)?;
+            construct_extra_java_options(
+                self,
+                s3conn,
+                log_dir,
+                product_version,
+                open_lineage_conn,
+            )?;
         submit_cmd.extend(vec![
             format!("--conf spark.driver.extraJavaOptions=\"{driver_extra_java_options}\""),
             format!("--conf spark.executor.extraJavaOptions=\"{executor_extra_java_options}\""),
@@ -845,6 +856,7 @@ impl v1alpha1::SparkApplication {
         &self,
         s3conn: &Option<s3::v1alpha1::ConnectionSpec>,
         logdir: &Option<ResolvedLogDir>,
+        open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>,
     ) -> Vec<EnvVar> {
         let mut e: Vec<EnvVar> = self.spec.env.clone();
 
@@ -877,7 +889,7 @@ impl v1alpha1::SparkApplication {
                 value_from: None,
             });
         }
-        if tlscerts::tls_secret_names(s3conn, logdir).is_some() {
+        if tlscerts::tls_secret_names(s3conn, logdir, open_lineage_conn).is_some() {
             e.push(EnvVar {
                 name: "STACKABLE_TLS_STORE_PASSWORD".to_string(),
                 value: Some(STACKABLE_TLS_STORE_PASSWORD.to_string()),
@@ -1627,7 +1639,7 @@ spec:
         "#})
         .unwrap();
 
-        let got = spark_application.spark_job_volume_mounts(&None, &None);
+        let got = spark_application.spark_job_volume_mounts(&None, &None, None);
 
         let expected = vec![
             VolumeMount {
@@ -1875,6 +1887,63 @@ spec:
         );
     }
 
+    /// The OpenLineage connection's `caCert.secretClass` must be wired into the driver truststore,
+    /// so the OpenLineage listener's HTTPS POST to the backend trusts the secret-operator cert.
+    /// Without this the `https` scheme above is unusable (handshake fails against the default JVM
+    /// truststore).
+    #[test]
+    fn test_openlineage_tls_secret_class_wired_into_truststore() {
+        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:
+                connection:
+                  inline:
+                    host: marquez
+                    port: 5000
+                    tls:
+                      verification:
+                        server:
+                          caCert:
+                            secretClass: marquez-ca
+        "#};
+        let command = build_command_with_openlineage(yaml, "4.1.2");
+
+        // The mounted secret-operator truststore for `marquez-ca` is imported into the shared
+        // PKCS12 truststore.
+        assert!(command.contains(&format!(
+            "{STACKABLE_MOUNT_PATH_TLS}/marquez-ca/truststore.p12"
+        )));
+        // The driver + executor JVMs are pointed at that truststore.
+        assert_eq!(
+            command
+                .matches(&format!(
+                    "-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12"
+                ))
+                .count(),
+            2,
+        );
+    }
+
+    /// Without an OpenLineage TLS `secretClass` (and no S3/logdir), no truststore is built and the
+    /// driver JVM gets no truststore options.
+    #[test]
+    fn test_openlineage_without_tls_has_no_truststore() {
+        let command = build_command_with_openlineage(OPENLINEAGE_ENABLED_APP, "4.1.2");
+
+        assert!(!command.contains("-Djavax.net.ssl.trustStore="));
+        assert!(!command.contains(STACKABLE_MOUNT_PATH_TLS));
+    }
+
     #[rstest]
     #[case::explicit_app_name(
         indoc! {r#"

rust/operator-binary/src/crd/roles.rs

@@ -28,7 +28,7 @@ use stackable_operator::{
         fragment::Fragment,
         merge::{Atomic, Merge},
     },
-    crd::s3,
+    crd::{openlineage, s3},
     k8s_openapi::{api::core::v1::VolumeMount, apimachinery::pkg::api::resource::Quantity},
     product_logging::{self, spec::Logging},
     schemars::{self, JsonSchema},
@@ -167,9 +167,16 @@ impl RoleConfig {
         spark_application: &v1alpha1::SparkApplication,
         s3conn: &Option<s3::v1alpha1::ConnectionSpec>,
         logdir: &Option<ResolvedLogDir>,
+        open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>,
     ) -> Vec<VolumeMount> {
         let volume_mounts = self.volume_mounts.clone().into();
-        spark_application.add_common_volume_mounts(volume_mounts, s3conn, logdir, true)
+        spark_application.add_common_volume_mounts(
+            volume_mounts,
+            s3conn,
+            logdir,
+            true,
+            open_lineage_conn,
+        )
     }
 }
 

rust/operator-binary/src/crd/tlscerts.rs

@@ -4,7 +4,7 @@ use stackable_operator::{
     commons::tls_verification::{
         CaCert, Tls, TlsClientDetails, TlsServerVerification, TlsVerification,
     },
-    crd::s3,
+    crd::{openlineage, s3},
 };
 
 use crate::crd::{
@@ -33,9 +33,32 @@ pub fn tls_secret_name(s3conn: &s3::v1alpha1::ConnectionSpec) -> Option<&str> {
     None
 }
 
+/// Extracts the SecretClass name backing the CA cert for a resolved OpenLineage connection, if the
+/// connection verifies the server's TLS certificate against a SecretClass CA. Mirrors
+/// [`tls_secret_name`] for S3 connections — both carry a flattened [`TlsClientDetails`].
+pub fn openlineage_tls_secret_name(
+    conn: &openlineage::ResolvedOpenLineageConnection,
+) -> Option<&str> {
+    if let TlsClientDetails {
+        tls:
+            Some(Tls {
+                verification:
+                    TlsVerification::Server(TlsServerVerification {
+                        ca_cert: CaCert::SecretClass(secret_name),
+                    }),
+            }),
+    } = &conn.tls
+    {
+        return Some(secret_name);
+    }
+
+    None
+}
+
 pub fn tls_secret_names<'a>(
     s3conn: &'a Option<s3::v1alpha1::ConnectionSpec>,
     logdir: &'a Option<ResolvedLogDir>,
+    open_lineage_conn: Option<&'a openlineage::ResolvedOpenLineageConnection>,
 ) -> Option<Vec<&'a str>> {
     // Ensure there are no duplicate secret names.
     let mut names = BTreeSet::new();
@@ -49,6 +72,10 @@ pub fn tls_secret_names<'a>(
     {
         names.insert(secret_name);
     }
+
+    if let Some(secret_name) = open_lineage_conn.and_then(openlineage_tls_secret_name) {
+        names.insert(secret_name);
+    }
     if names.is_empty() {
         None
     } else {

rust/operator-binary/src/spark_k8s_controller.rs

@@ -166,7 +166,7 @@ pub async fn reconcile(
         .await
         .context(ApplyRoleBindingSnafu)?;
 
-    let env_vars = spark_application.env(opt_s3conn, logdir);
+    let env_vars = spark_application.env(opt_s3conn, logdir, opt_open_lineage_conn);
 
     let driver_config = spark_application
         .driver_config()

rust/operator-binary/src/spark_k8s_controller/build/pod.rs

@@ -51,6 +51,7 @@ fn init_containers(
     let spark_application = &validated.spark_application;
     let s3conn = &validated.cluster_config.s3_connection;
     let logdir = &validated.cluster_config.log_dir;
+    let open_lineage_conn = validated.cluster_config.open_lineage_connection.as_ref();
     let spark_image = &validated.resolved_product_image;
     let mut jcb = new_container_builder(&SparkContainer::Job.to_container_name());
     let job_container = match &spark_application.spec.image {
@@ -153,7 +154,7 @@ fn init_containers(
     let mut tcb = new_container_builder(&SparkContainer::Tls.to_container_name());
     let mut args = Vec::new();
 
-    let tls_container = match tlscerts::tls_secret_names(s3conn, logdir) {
+    let tls_container = match tlscerts::tls_secret_names(s3conn, logdir, open_lineage_conn) {
         Some(cert_secrets) => {
             args.push(tlscerts::convert_system_trust_store_to_pkcs12());
             for cert_secret in cert_secrets {
@@ -207,16 +208,22 @@ pub(crate) fn pod_template(
     let spark_application = &validated.spark_application;
     let s3conn = &validated.cluster_config.s3_connection;
     let logdir = &validated.cluster_config.log_dir;
+    let open_lineage_conn = validated.cluster_config.open_lineage_connection.as_ref();
     let spark_image = &validated.resolved_product_image;
     let container_name = SparkContainer::Spark.to_string();
     let mut cb = new_container_builder(&SparkContainer::Spark.to_container_name());
     let merged_env = spark_application.merged_env(role.clone(), env);
 
-    cb.add_volume_mounts(config.volume_mounts(spark_application, s3conn, logdir))
-        .context(AddVolumeMountSnafu)?
-        .add_env_vars(merged_env)
-        .resources(config.resources.clone().into())
-        .image_from_product_image(spark_image);
+    cb.add_volume_mounts(config.volume_mounts(
+        spark_application,
+        s3conn,
+        logdir,
+        open_lineage_conn,
+    ))
+    .context(AddVolumeMountSnafu)?
+    .add_env_vars(merged_env)
+    .resources(config.resources.clone().into())
+    .image_from_product_image(spark_image);
 
     if config.logging.enable_vector_agent {
         cb.add_env_var(