diff --git a/CHANGELOG.md b/CHANGELOG.md index bbb19dac..e5312d3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file. ### Added +- Add support for emitting [OpenLineage](https://openlineage.io/) events from `SparkApplications` via `spec.lineage` ([#726]). - Support floating tags for product images via the new `spec.image.stackableVersionPolicy` field ([#773]). - The history server and Spark Connect server StatefulSets now carry the @@ -42,6 +43,8 @@ All notable changes to this project will be documented in this file. - BREAKING (behaviour): Keys of the `spark-env.sh` `configOverrides` must be valid shell identifiers (matching `[a-zA-Z_][a-zA-Z0-9_]*`) and are now rejected if they are not ([#761]). - Make operations infallible where dependent on static inputs ([#766], [#769]). +- Bump stackable-operator to 0.119.0, 0.120.0 ([#779], [#726]). +- test: Bump vector-aggregator to 0.58.0 ([#784]). - Support Vector 0.58.0 ([#784], [#788]). ### Fixed @@ -77,6 +80,7 @@ All notable changes to this project will be documented in this file. node's image cache ([#764]). [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721 +[#726]: https://github.com/stackabletech/spark-k8s-operator/pull/726 [#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727 [#732]: https://github.com/stackabletech/spark-k8s-operator/pull/732 [#744]: https://github.com/stackabletech/spark-k8s-operator/pull/744 diff --git a/Cargo.lock b/Cargo.lock index 9f425ed2..f4639310 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -402,9 +402,9 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "combine" -version = "4.6.8" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "memchr", diff --git a/Cargo.nix b/Cargo.nix index 75f29a0b..c83f48b2 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -1247,9 +1247,9 @@ rec { }; "combine" = rec { crateName = "combine"; - version = "4.6.8"; + version = "4.6.9"; edition = "2018"; - sha256 = "0ppwzwdmszpan9ybx1myc6ldg5zih2sazf9idckdxrh9gn9j1hyg"; + sha256 = "14izj50knw6rgl6ky3903k43pnl6qjgyw4d30n6ljdb4jykfnmj3"; authors = [ "Markus Westerlind " ]; diff --git a/deploy/helm/spark-k8s-operator/templates/roles.yaml b/deploy/helm/spark-k8s-operator/templates/roles.yaml index 2d9f3b56..c2d575fd 100644 --- a/deploy/helm/spark-k8s-operator/templates/roles.yaml +++ b/deploy/helm/spark-k8s-operator/templates/roles.yaml @@ -176,6 +176,15 @@ rules: - get - list - watch + # OpenLineage backend connections referenced by SparkApplications. + - apiGroups: + - lineage.stackable.tech + resources: + - openlineageconnections + verbs: + - get + - list + - watch # Required for managing how the History Server and Connect Server are exposed # outside of the cluster. The history and connect controllers watch Listeners via # .owns(Listener). 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..d80c3746 --- /dev/null +++ b/docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc @@ -0,0 +1,148 @@ += 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 configured, 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 +* 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 + +The backend is configured through the `lineage.connection`, which either inlines an OpenLineage +connection or references a standalone `OpenLineageConnection` resource by name. +The operator points the transport at `://:`: + +[source,yaml] +---- +apiVersion: spark.stackable.tech/v1alpha1 +kind: SparkApplication +metadata: + name: orders-by-nation +spec: + lineage: # <1> + connection: # <2> + inline: + http: + host: marquez + port: 5000 + # path: /api/v1/lineage + # tls: + # verification: + # server: + # caCert: + # secretClass: marquez-ca + # namespace: orders-lineage # <3> + # appName: orders-by-nation # <4> + ... +---- +<1> Adding the `lineage` block enables OpenLineage event emission. Omit it entirely to leave OpenLineage off (the default), in which case the operator emits nothing OpenLineage-related and behaviour is unchanged. +<2> The OpenLineage backend connection. Use `inline` (shown here) to define the connection directly, + or `reference: ` to point at an `OpenLineageConnection` resource in the same namespace. + Required. The connection selects a transport; `http` is currently the only one supported. +<3> Optional. The OpenLineage namespace events are reported under. + Defaults to `default`. +<4> Optional. A stable OpenLineage application (job) name, see <>. + If not set, it is resolved from `sparkConf` or the resource name. + +The `OpenLineageConnection` resource (`lineage.stackable.tech`) is a reusable definition of a +connection to an OpenLineage backend, provided by the Stackable operator library. Reference it to +share one backend definition across multiple applications: + +[source,yaml] +---- +apiVersion: lineage.stackable.tech/v1alpha1 +kind: OpenLineageConnection +metadata: + name: marquez +spec: + http: + host: marquez + port: 5000 +---- + +The transport scheme is `https` when `tls.verification.server` is configured, otherwise `http`. +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. + +[WARNING] +==== +Disabling TLS server verification (`tls.verification: none`) is not supported for OpenLineage. +The OpenLineage HTTP transport has no option to skip certificate verification, so if the backend presents a certificate that the driver JVM does not trust, the TLS handshake fails. +The Spark application keeps running, but the lineage events are dropped, and the only trace is an error in the driver log. + +If the backend uses a certificate signed by a private CA, configure `tls.verification.server.caCert` as shown above instead. +==== + +You can still override any of the injected values — including the full `spark.openlineage.transport.url` — through `sparkConf`. + +[#authentication] +== Authentication + +Backends that require authentication (for example DataHub, or a Marquez instance behind a gateway) expect a bearer token on each request. +Configure this by setting `credentialsSecretName` on the connection to the name of a Secret holding the token: + +[source,yaml] +---- +apiVersion: spark.stackable.tech/v1alpha1 +kind: SparkApplication +metadata: + name: orders-by-nation +spec: + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + credentialsSecretName: openlineage-token # <1> + ... +--- +apiVersion: v1 +kind: Secret +metadata: + name: openlineage-token + namespace: default # <2> +stringData: + apiKey: "s3cr3t-bearer-token" # <3> +---- +<1> Name of the Secret holding the token. If omitted, no authentication is used. The Secret must exist in the same namespace as the `SparkApplication`. +<2> The Secret must be in the workload's namespace so it can be projected into the driver. +<3> The token must be stored under the fixed key `apiKey`. The operator never reads the Secret; it is injected into the driver's OpenLineage listener via a `secretKeyRef` environment variable, so the token never appears in the spark-submit arguments or the pod spec. + +The operator translates this into the OpenLineage HTTP transport's `api_key` (bearer) authentication. +Combine it with the `tls` block above when the backend also needs TLS server verification. + +[NOTE] +==== +When `credentialsSecretName` is set, the operator delivers the *entire* transport configuration +(type, URL and auth) to the driver through `OPENLINEAGE__*` environment variables rather than +`spark.openlineage.transport.*` `--conf` properties. +This is required: OpenLineage resolves the `transport` from a single source, so a transport defined +via `sparkConf` would discard the environment-provided authentication. +Consequently, do **not** override `spark.openlineage.transport.*` via `sparkConf` when using +`credentialsSecretName` — doing so drops the credentials and the backend will reject the events. +==== + +[#job-name] +== Job name + +The OpenLineage job name is resolved in the following order: + +. `lineage.appName`, if not empty. +. `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 `lineage.appName` 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 100ffbc3..2e685f6f 100644 --- a/docs/modules/spark-k8s/partials/nav.adoc +++ b/docs/modules/spark-k8s/partials/nav.adoc @@ -10,6 +10,7 @@ ** xref:spark-k8s:usage-guide/security/index.adoc[] *** xref:spark-k8s:usage-guide/security/encryption-in-transit.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 5ff4fee4..0c771bb2 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -1753,6 +1753,124 @@ spec: type: object x-kubernetes-preserve-unknown-fields: true type: object + lineage: + 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). + The backend connection is either inlined or references an `OpenLineageConnection` resource. + See the [OpenLineage usage guide](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/openlineage). + nullable: true + properties: + appName: + description: |- + A stable OpenLineage application (job) name. Setting this prevents fragmented run + history. If not set (or empty), the name resolves to `spark.app.name` from `sparkConf` + and then to the name of the `SparkApplication`. + nullable: true + type: string + connection: + description: |- + The OpenLineage backend connection, either inlined or referencing an + `OpenLineageConnection` resource by name. + oneOf: + - required: + - inline + - required: + - reference + properties: + inline: + description: |- + OpenLineage connection definition as a resource. + Learn more about [OpenLineage](https://openlineage.io/). + oneOf: + - required: + - http + properties: + http: + description: Publish events over HTTP(S) to an OpenLineage backend such as Marquez. + properties: + credentialsSecretName: + description: |- + Name of a Secret containing the API key used to authenticate against the OpenLineage + backend. The API key must be stored under the key `apiKey`. The Secret must be located in + the same namespace as the workload using this connection. If not specified, no + authentication is used. + nullable: true + type: string + host: + description: 'Hostname or IP address of the OpenLineage backend without any protocol or port. For example: `marquez`.' + type: string + path: + default: /api/v1/lineage + description: URL path of the endpoint lineage events are sent to. + type: string + port: + description: 'Port the OpenLineage backend listens on. For example: `5000`.' + format: uint16 + maximum: 65535.0 + minimum: 0.0 + type: integer + tls: + description: Use a TLS connection. If not specified no TLS will be used. + nullable: true + properties: + verification: + description: The verification method used to verify the certificates of the server and/or the client. + oneOf: + - required: + - none + - required: + - server + properties: + none: + description: Use TLS but don't verify certificates. + type: object + server: + description: Use TLS and a CA certificate to verify the server. + properties: + caCert: + description: CA cert to verify the server. + oneOf: + - required: + - webPki + - required: + - secretClass + properties: + secretClass: + description: |- + Name of the [SecretClass](https://docs.stackable.tech/home/nightly/secret-operator/secretclass) which will provide the CA certificate. + Note that a SecretClass does not need to have a key but can also work with just a CA certificate, + so if you got provided with a CA cert but don't have access to the key you can still use this method. + type: string + webPki: + description: |- + Use TLS and the CA certificates trusted by the common web browsers to verify the server. + This can be useful when you e.g. use public AWS S3 or other public available services. + type: object + type: object + required: + - caCert + type: object + type: object + required: + - verification + type: object + required: + - host + - port + type: object + type: object + reference: + type: string + type: object + namespace: + default: default + description: The OpenLineage namespace lineage is reported under. + type: string + required: + - connection + type: object logFileDirectory: description: The log file directory definition used by the Spark history server. nullable: true @@ -6467,6 +6585,124 @@ spec: type: object x-kubernetes-preserve-unknown-fields: true type: object + lineage: + 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). + The backend connection is either inlined or references an `OpenLineageConnection` resource. + See the [OpenLineage usage guide](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/openlineage). + nullable: true + properties: + appName: + description: |- + A stable OpenLineage application (job) name. Setting this prevents fragmented run + history. If not set (or empty), the name resolves to `spark.app.name` from `sparkConf` + and then to the name of the `SparkApplication`. + nullable: true + type: string + connection: + description: |- + The OpenLineage backend connection, either inlined or referencing an + `OpenLineageConnection` resource by name. + oneOf: + - required: + - inline + - required: + - reference + properties: + inline: + description: |- + OpenLineage connection definition as a resource. + Learn more about [OpenLineage](https://openlineage.io/). + oneOf: + - required: + - http + properties: + http: + description: Publish events over HTTP(S) to an OpenLineage backend such as Marquez. + properties: + credentialsSecretName: + description: |- + Name of a Secret containing the API key used to authenticate against the OpenLineage + backend. The API key must be stored under the key `apiKey`. The Secret must be located in + the same namespace as the workload using this connection. If not specified, no + authentication is used. + nullable: true + type: string + host: + description: 'Hostname or IP address of the OpenLineage backend without any protocol or port. For example: `marquez`.' + type: string + path: + default: /api/v1/lineage + description: URL path of the endpoint lineage events are sent to. + type: string + port: + description: 'Port the OpenLineage backend listens on. For example: `5000`.' + format: uint16 + maximum: 65535.0 + minimum: 0.0 + type: integer + tls: + description: Use a TLS connection. If not specified no TLS will be used. + nullable: true + properties: + verification: + description: The verification method used to verify the certificates of the server and/or the client. + oneOf: + - required: + - none + - required: + - server + properties: + none: + description: Use TLS but don't verify certificates. + type: object + server: + description: Use TLS and a CA certificate to verify the server. + properties: + caCert: + description: CA cert to verify the server. + oneOf: + - required: + - webPki + - required: + - secretClass + properties: + secretClass: + description: |- + Name of the [SecretClass](https://docs.stackable.tech/home/nightly/secret-operator/secretclass) which will provide the CA certificate. + Note that a SecretClass does not need to have a key but can also work with just a CA certificate, + so if you got provided with a CA cert but don't have access to the key you can still use this method. + type: string + webPki: + description: |- + Use TLS and the CA certificates trusted by the common web browsers to verify the server. + This can be useful when you e.g. use public AWS S3 or other public available services. + type: object + type: object + required: + - caCert + type: object + type: object + required: + - verification + type: object + required: + - host + - port + type: object + type: object + reference: + type: string + type: object + namespace: + default: default + description: The OpenLineage namespace lineage is reported under. + type: string + required: + - connection + type: object logFileDirectory: description: The log file directory definition used by the Spark history server. nullable: true diff --git a/rust/operator-binary/src/config/jvm.rs b/rust/operator-binary/src/config/jvm.rs index b8568669..0c368fa4 100644 --- a/rust/operator-binary/src/config/jvm.rs +++ b/rust/operator-binary/src/config/jvm.rs @@ -1,9 +1,9 @@ -use stackable_operator::crd::s3; +use stackable_operator::crd::{openlineage, s3}; use crate::crd::{ constants::{ - JVM_SECURITY_PROPERTIES_FILE, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE, - VOLUME_MOUNT_PATH_CONFIG, + JVM_SECURITY_PROPERTIES_FILE, OPENLINEAGE_ADD_OPENS, STACKABLE_TLS_STORE_PASSWORD, + STACKABLE_TRUST_STORE, VOLUME_MOUNT_PATH_CONFIG, }, logdir::ResolvedLogDir, tlscerts::tls_secret_names, @@ -20,6 +20,7 @@ pub fn construct_extra_java_options( spark_application: &SparkApplication, s3_conn: &Option, log_dir: &Option, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> (String, String) { // 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. @@ -28,7 +29,7 @@ pub fn construct_extra_java_options( "-Djava.security.properties={VOLUME_MOUNT_PATH_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}"), @@ -36,10 +37,15 @@ pub fn construct_extra_java_options( ]); } + // OpenLineage needs `java.base/java.security` opened to the unnamed module, otherwise the + // driver throws a non-fatal `InaccessibleObjectException` and silently degrades + // extension-interface lineage. Added to both driver and executor. + if spark_application.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 - // unit tests below. + // above. let driver = match &spark_application.spec.driver { Some(driver) => driver .product_specific_common_config @@ -61,6 +67,8 @@ pub fn construct_extra_java_options( #[cfg(test)] mod tests { + use rstest::rstest; + use super::*; #[test] @@ -79,8 +87,9 @@ mod tests { let spark_app: SparkApplication = stackable_operator::utils::yaml_from_str_singleton_map(input).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, None); assert_eq!( driver_extra_java_options, @@ -103,7 +112,7 @@ mod tests { mode: cluster mainApplicationFile: test.py sparkImage: - productVersion: 1.2.3 + productVersion: 4.1.2 driver: jvmArgumentOverrides: add: @@ -118,8 +127,9 @@ mod tests { let spark_app: SparkApplication = stackable_operator::utils::yaml_from_str_singleton_map(input).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, None); assert_eq!( driver_extra_java_options, @@ -130,4 +140,64 @@ mod tests { "-Dhttps.proxyHost=from-executor" ); } + + 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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + "#; + + 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 on every Spark version when OpenLineage is enabled, and never when + /// OpenLineage is absent. + #[rstest] + #[case::enabled_spark_3(OPENLINEAGE_ENABLED, "3.5.8", true)] + #[case::enabled_spark_4(OPENLINEAGE_ENABLED, "4.1.2", true)] + #[case::absent_spark_4(OPENLINEAGE_ABSENT, "4.1.2", false)] + fn test_openlineage_add_opens( + #[case] yaml_template: &str, + #[case] product_version: &str, + #[case] expect_add_opens: bool, + ) { + let spark_app = stackable_operator::utils::yaml_from_str_singleton_map( + &yaml_template.replace("PLACEHOLDER", product_version), + ) + .unwrap(); + let (driver_extra_java_options, executor_extra_java_options) = + construct_extra_java_options(&spark_app, &None, &None, None); + + 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 ca12f300..f667d74b 100644 --- a/rust/operator-binary/src/crd/constants.rs +++ b/rust/operator-binary/src/crd/constants.rs @@ -115,6 +115,62 @@ pub const DEFAULT_LISTENER_CLASS: &str = "cluster-internal"; 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"; + +/// `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. +/// +/// This is a stable, version- and Scala-independent symlink the image maintains (mirroring the +/// `jmx_prometheus_javaagent.jar` pattern): each Spark image bakes the correct +/// `openlineage-spark_-.jar` for its build (Scala 2.13 for Spark 4.x, Scala 2.12 for +/// Spark 3.5.x) 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"; + +/// Java module-system flag OpenLineage requires: 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"; + +/// Env vars delivering the OpenLineage HTTP transport config via the OpenLineage Java client's +/// `OPENLINEAGE__` env-var configuration (`__` separates the config hierarchy). +/// +/// When an `AuthenticationClass` is configured, the operator delivers the **whole** transport this +/// way rather than via `spark.openlineage.transport.*` `--conf`. This is required, not cosmetic: +/// OpenLineage resolves `transport` from a single source, so a `--conf`-defined transport would +/// drop the env-provided `auth` sub-tree entirely (verified against openlineage-spark 1.51.0). +/// Delivering the token as a `secretKeyRef` env var also keeps it out of the spark-submit args. +pub const OPENLINEAGE_TRANSPORT_TYPE_ENV: &str = "OPENLINEAGE__TRANSPORT__TYPE"; + +/// Env var carrying the OpenLineage HTTP transport URL. See [`OPENLINEAGE_TRANSPORT_TYPE_ENV`]. +pub const OPENLINEAGE_TRANSPORT_URL_ENV: &str = "OPENLINEAGE__TRANSPORT__URL"; + +/// Env var carrying the OpenLineage HTTP transport endpoint path, set from the connection's `path`. +/// See [`OPENLINEAGE_TRANSPORT_TYPE_ENV`]. +pub const OPENLINEAGE_TRANSPORT_ENDPOINT_ENV: &str = "OPENLINEAGE__TRANSPORT__ENDPOINT"; + +/// The OpenLineage transport type the operator uses (the HTTP transport). +pub const OPENLINEAGE_TRANSPORT_TYPE_HTTP: &str = "http"; + +/// Env var selecting the OpenLineage HTTP transport auth type. See [`OPENLINEAGE_TRANSPORT_TYPE_ENV`]. +pub const OPENLINEAGE_AUTH_TYPE_ENV: &str = "OPENLINEAGE__TRANSPORT__AUTH__TYPE"; + +/// Env var carrying the OpenLineage HTTP transport bearer token. Sourced from the referenced +/// Secret via `secretKeyRef` (key [`OPENLINEAGE_AUTH_SECRET_KEY`]), so the operator never reads +/// the token and it stays out of the Job/pod spec. +pub const OPENLINEAGE_AUTH_API_KEY_ENV: &str = "OPENLINEAGE__TRANSPORT__AUTH__API_KEY"; + +/// The only OpenLineage HTTP transport auth type the operator supports: a static bearer token. +pub const OPENLINEAGE_AUTH_TYPE_API_KEY: &str = "api_key"; + +/// Fixed key that must hold the bearer token inside the Secret referenced by the Static +/// `AuthenticationClass`. +pub const OPENLINEAGE_AUTH_SECRET_KEY: &str = "apiKey"; + /// 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 1759d1f8..f1477f10 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -26,7 +26,7 @@ use stackable_operator::{ merge::Merge, }, constant, - crd::s3, + crd::{openlineage, s3}, k8s_openapi::{ api::core::v1::{EmptyDirVolumeSource, EnvVar, PodTemplateSpec, Volume, VolumeMount}, apimachinery::pkg::api::resource::Quantity, @@ -55,6 +55,7 @@ use crate::{ SubmitConfig, SubmitConfigFragment, VolumeMounts, }, }, + lineage::{append_conf_csv, openlineage_transport_env_vars, transport_base_url}, }; pub mod affinity; @@ -73,7 +74,7 @@ pub enum Error { #[snafu(display("object has no namespace associated"))] NoNamespace, - #[snafu(display("object has no name"))] + #[snafu(display("object has no name"), visibility(pub(crate)))] ObjectHasNoName, #[snafu(display("failed to convert java heap config to unit [{unit}]"))] @@ -262,6 +263,29 @@ 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). + /// The backend connection is either inlined or references an `OpenLineageConnection` resource. + /// See the [OpenLineage usage guide](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/openlineage). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lineage: Option, + } + + /// OpenLineage configuration of a `SparkApplication`: the shared OpenLineage config, plus + /// Spark specific settings. + #[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct SparkLineageConfig { + #[serde(flatten)] + pub open_lineage: openlineage::v1alpha1::OpenLineageConfig, + + /// A stable OpenLineage application (job) name. Setting this prevents fragmented run + /// history. If not set (or empty), the name resolves to `spark.app.name` from `sparkConf` + /// and then to the name of the `SparkApplication`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub app_name: Option, } #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize, Merge)] @@ -324,6 +348,7 @@ impl v1alpha1::SparkApplication { logdir: &Option, log_config_map: Option<&str>, requested_secret_lifetime: &Duration, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> Result, 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. @@ -404,7 +429,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) @@ -457,6 +482,7 @@ impl v1alpha1::SparkApplication { &self, s3conn: &Option, logdir: &Option, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> Vec { let mut tmpl_mounts = vec![ VolumeMount { @@ -471,7 +497,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: @@ -497,6 +524,7 @@ impl v1alpha1::SparkApplication { s3conn: &Option, logdir: &Option, logging_enabled: bool, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> Vec { if self.spec.image.is_some() { mounts.push(VolumeMount { @@ -555,7 +583,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(), @@ -574,28 +602,36 @@ impl v1alpha1::SparkApplication { mounts } + /// True when OpenLineage emission is configured for this application. OpenLineage is off exactly + /// when the `lineage` block is absent — its presence is the enable switch. + pub fn lineage_enabled(&self) -> bool { + self.spec.lineage.is_some() + } + pub fn build_command( &self, s3conn: &Option, log_dir: &Option, spark_image: &ResolvedProductImage, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> Result, Error> { // mandatory properties let mode = &self.spec.mode; 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!( @@ -677,7 +713,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, 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}\""), @@ -756,9 +792,68 @@ 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(lineage) = &self.spec.lineage { + // The backend connection is resolved (inline or referenced) before `build_command`. + // The transport type is always `http` (the OpenLineage HTTP transport); the URL scheme + // it points at is `https` when the connection configures TLS server verification. + // + // When authentication is configured (the connection sets `credentialsSecretName`), the + // transport is delivered entirely via `OPENLINEAGE__` env vars instead (see `env`) — it + // MUST come from a single source, or OpenLineage drops the env-provided `auth`. So skip + // the `--conf` transport in that case. + if let Some(connection) = open_lineage_conn + && crate::lineage::http_transport(connection) + .credentials_secret_name + .is_none() + { + let http = crate::lineage::http_transport(connection); + submit_conf.insert( + "spark.openlineage.transport.type".to_string(), + "http".to_string(), + ); + submit_conf.insert( + "spark.openlineage.transport.url".to_string(), + transport_base_url(http), + ); + submit_conf.insert( + "spark.openlineage.transport.endpoint".to_string(), + http.path.clone(), + ); + } + + submit_conf.insert( + "spark.openlineage.namespace".to_string(), + lineage.open_lineage.namespace.clone(), + ); + + // Stable job name — fixes the intermittent `unknown` bug (see the usage guide). + submit_conf.insert( + "spark.openlineage.appName".to_string(), + self.resolved_lineage_app_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.lineage_enabled() { + append_conf_csv( + &mut submit_conf, + "spark.extraListeners", + OPENLINEAGE_LISTENER_CLASS, + ); + // Reference the stable symlink the image maintains; it points at the correct + // Scala/version build for this image (2.13 on Spark 4.x, 2.12 on Spark 3.5.x), so the + // operator needs to track neither the jar version nor its Scala suffix. + 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}\"")); @@ -782,12 +877,16 @@ impl v1alpha1::SparkApplication { /// The base environment for the submit, driver and executor containers: the user-defined /// `spec.env` entries plus the environment variables set by the operator. /// + /// Driver-only variables (the OpenLineage transport) are added separately by + /// [`Self::with_openlineage_transport_env`]. + /// /// The role-specific `envOverrides` are merged on top in [`Self::merged_env`], so they take /// precedence over everything set here. pub fn env( &self, s3conn: &Option, logdir: &Option, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> Result { // The CRD accepts raw `EnvVar` objects in `spec.env`, so the names are only validated // here. @@ -823,7 +922,7 @@ impl v1alpha1::SparkApplication { format!("$SPARK_HOME/python:{VOLUME_MOUNT_PATH_REQ}:$PYTHONPATH"), ); } - if tlscerts::tls_secret_names(s3conn, logdir).is_some() { + if tlscerts::tls_secret_names(s3conn, logdir, open_lineage_conn).is_some() { env = env.with_value( &STACKABLE_TLS_STORE_PASSWORD_ENV, STACKABLE_TLS_STORE_PASSWORD, @@ -832,6 +931,33 @@ impl v1alpha1::SparkApplication { Ok(env) } + /// Adds the OpenLineage HTTP transport environment variables to `env`, the environment of the + /// containers that run the driver JVM. + /// + /// The listener only runs in the driver JVM, so executors do not need the transport, + /// and must not get the token. + pub fn with_openlineage_transport_env( + &self, + mut env: EnvVarSet, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, + ) -> Result { + if let Some(connection) = open_lineage_conn { + let http = crate::lineage::http_transport(connection); + if let Some(secret_name) = &http.credentials_secret_name { + for env_var in openlineage_transport_env_vars( + &transport_base_url(http), + &http.path, + secret_name, + ) { + env = env + .with_env_var(env_var) + .context(ParseSparkEnvVarNameSnafu)?; + } + } + } + Ok(env) + } + pub fn submit_config(&self) -> Result { if let Some(CommonConfiguration { mut config, .. }) = self.spec.job.clone() { config.merge(&SubmitConfig::default_config()); @@ -1622,7 +1748,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 { @@ -1654,6 +1780,564 @@ spec: assert_eq!(got, expected); } + /// Builds a `SparkApplication` from YAML and returns the assembled spark-submit command string. + /// + /// Test fixtures always use an inline OpenLineage connection, so the connection is extracted + /// directly (no async `resolve()` against a k8s API) and passed to `build_command` — exactly the + /// resolved value the controller would compute in `dereference`. + fn build_command_with_openlineage(yaml: &str, product_version: &str) -> String { + use stackable_operator::crd::openlineage::v1alpha1::InlineConnectionOrReference; + + // `singleton_map_recursive` so externally-tagged enums (the OpenLineage `connection`) use + // the `inline:`/`reference:` map form that k8s/JSON uses, rather than serde_yaml's `!tag`. + let spark_application: v1alpha1::SparkApplication = + stackable_operator::utils::yaml_from_str_singleton_map(yaml).unwrap(); + let open_lineage_conn = + spark_application + .spec + .lineage + .as_ref() + .map(|job| match &job.open_lineage.connection { + InlineConnectionOrReference::Inline(spec) => spec.clone(), + InlineConnectionOrReference::Reference(name) => { + panic!( + "test fixtures must use an inline connection, got reference {name:?}" + ) + } + }); + spark_application + .build_command( + &None, + &None, + &test_product_image(&spark_application, product_version), + open_lineage_conn.as_ref(), + ) + .unwrap() + .join(" ") + } + + /// Resolves the product image of the application, reporting `product_version` as the resolved + /// Spark version (the YAML fixtures use a placeholder version). + fn test_product_image( + spark_application: &v1alpha1::SparkApplication, + product_version: &str, + ) -> ResolvedProductImage { + let mut image = spark_application + .spec + .spark_image + .resolve( + "spark-k8s", + "oci.example.org/sdp", + &"0.0.0-dev" + .parse() + .expect("static semantic version must parse"), + ) + .expect("the product image resolves"); + image.product_version = product_version.to_string(); + image + } + + const MINIMAL_APP: &str = indoc! {r#" + --- + apiVersion: spark.stackable.tech/v1alpha1 + kind: SparkApplication + metadata: + name: test + namespace: default + spec: + mode: cluster + mainApplicationFile: test.py + sparkImage: + productVersion: 1.2.3 + "#}; + + fn test_openlineage_connection( + credentials_secret_name: Option, + ) -> openlineage::ResolvedOpenLineageConnection { + openlineage::v1alpha1::OpenLineageConnectionSpec { + transport: openlineage::v1alpha1::OpenLineageTransport::Http( + openlineage::v1alpha1::HttpTransport { + host: "marquez".parse().expect("valid host name"), + port: 5000, + path: openlineage::v1alpha1::HttpTransport::DEFAULT_PATH.to_string(), + tls: stackable_operator::commons::tls_verification::TlsClientDetails { + tls: None, + }, + credentials_secret_name, + }, + ), + } + } + + #[test] + fn test_openlineage_full_transport_delivered_via_env_when_authenticated() { + let spark_application = + yaml_serde::from_str::(MINIMAL_APP).unwrap(); + let connection = test_openlineage_connection(Some("ol-token".to_string())); + + let base_env = spark_application + .env(&None, &None, Some(&connection)) + .unwrap(); + // Shared by all roles, so it must not carry the transport or the token reference. + let shared: Vec = base_env.clone().into(); + assert!( + !shared.iter().any(|v| v.name == OPENLINEAGE_AUTH_API_KEY_ENV + || v.name == OPENLINEAGE_TRANSPORT_URL_ENV), + "OpenLineage transport env vars must not be in the env shared with executors" + ); + + let env: Vec = spark_application + .with_openlineage_transport_env(base_env, Some(&connection)) + .unwrap() + .into(); + let value = |name| { + env.iter() + .find(|v| v.name == name) + .and_then(|v| v.value.as_deref()) + }; + + // The whole transport is delivered via env (not --conf) so OpenLineage keeps the auth. + assert_eq!(value(OPENLINEAGE_TRANSPORT_TYPE_ENV), Some("http")); + assert_eq!( + value(OPENLINEAGE_TRANSPORT_URL_ENV), + Some("http://marquez:5000") + ); + assert_eq!( + value(OPENLINEAGE_AUTH_TYPE_ENV), + Some(OPENLINEAGE_AUTH_TYPE_API_KEY) + ); + + let key_var = env + .iter() + .find(|v| v.name == OPENLINEAGE_AUTH_API_KEY_ENV) + .expect("auth api key env var present"); + assert!( + key_var.value.is_none(), + "the token must never be a literal env value" + ); + let selector = key_var + .value_from + .as_ref() + .and_then(|source| source.secret_key_ref.as_ref()) + .expect("api key sourced from a secretKeyRef"); + assert_eq!(selector.name, "ol-token"); + assert_eq!(selector.key, OPENLINEAGE_AUTH_SECRET_KEY); + } + + #[test] + fn test_no_openlineage_transport_env_vars_when_unauthenticated() { + let spark_application = + yaml_serde::from_str::(MINIMAL_APP).unwrap(); + let connection = test_openlineage_connection(None); + + // A connection but no credentialsSecretName: transport stays in --conf, no OL env vars. + let env: Vec = spark_application + .with_openlineage_transport_env( + spark_application + .env(&None, &None, Some(&connection)) + .unwrap(), + Some(&connection), + ) + .unwrap() + .into(); + + for name in [ + OPENLINEAGE_TRANSPORT_TYPE_ENV, + OPENLINEAGE_TRANSPORT_URL_ENV, + OPENLINEAGE_AUTH_TYPE_ENV, + OPENLINEAGE_AUTH_API_KEY_ENV, + ] { + assert!( + !env.iter().any(|v| v.name == name), + "unexpected OpenLineage env var {name} when unauthenticated" + ); + } + } + + #[test] + fn test_openlineage_authenticated_omits_transport_conf() { + // With auth present the transport moves to env vars, so build_command must NOT emit the + // `spark.openlineage.transport.*` --conf (that would make OpenLineage drop the env auth). + // namespace / appName / listener stay in --conf (OPENLINEAGE_ENABLED_APP has the block). + let spark_application: v1alpha1::SparkApplication = + stackable_operator::utils::yaml_from_str_singleton_map(OPENLINEAGE_ENABLED_APP) + .unwrap(); + let connection = test_openlineage_connection(Some("ol-token".to_string())); + + let command = spark_application + .build_command( + &None, + &None, + &test_product_image(&spark_application, "3.5.8"), + Some(&connection), + ) + .unwrap() + .join(" "); + + assert!( + !command.contains("spark.openlineage.transport.type"), + "transport.type must not be in --conf when authenticated" + ); + assert!( + !command.contains("spark.openlineage.transport.url"), + "transport.url must not be in --conf when authenticated" + ); + assert!(command.contains(r#"--conf "spark.openlineage.namespace="#)); + assert!(command.contains(r#"--conf "spark.openlineage.appName="#)); + assert!(command.contains(&format!( + r#"--conf "spark.extraListeners={OPENLINEAGE_LISTENER_CLASS}""# + ))); + } + + 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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + "#}; + + #[test] + fn test_openlineage_injects_conf_when_enabled() { + let command = build_command_with_openlineage(OPENLINEAGE_ENABLED_APP, "4.1.2"); + + assert!(command.contains(r#"--conf "spark.openlineage.transport.type=http""#)); + assert!( + command.contains(r#"--conf "spark.openlineage.transport.url=http://marquez:5000""#) + ); + // Namespace comes from the CRD, which defaults it to `default`. + assert!(command.contains(r#"--conf "spark.openlineage.namespace=default""#)); + // The endpoint path comes from the connection, defaulting to /api/v1/lineage. + assert!( + command.contains(r#"--conf "spark.openlineage.transport.endpoint=/api/v1/lineage""#) + ); + // appName falls back to metadata.name (no jobName / 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}""# + ))); + // The jar is referenced via the stable, Scala/version-independent image symlink. + 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" + ); + } + + #[test] + fn test_openlineage_stable_jar_uri_and_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 emits `--add-opens` as on Spark 4.x. + let command = build_command_with_openlineage(OPENLINEAGE_ENABLED_APP, "3.5.8"); + + assert!(command.contains(&format!( + r#"--conf "spark.jars={OPENLINEAGE_JAR_LOCAL_URI}""# + ))); + assert_eq!(command.matches(OPENLINEAGE_ADD_OPENS).count(), 2); + // 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""#)); + } + + #[test] + fn test_openlineage_injects_nothing_when_absent() { + // No `lineage` block → OpenLineage is off (its absence is the disable switch). + 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 + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2"); + + 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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + sparkConf: + spark.extraListeners: com.example.CustomListener + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2"); + + 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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + sparkConf: + spark.openlineage.transport.url: http://custom:1234 + spark.openlineage.namespace: custom-ns + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2"); + + 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_uses_https_when_tls_verification_set() { + 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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + tls: + verification: + server: + caCert: + secretClass: marquez-ca + "#}; + let command = build_command_with_openlineage(yaml, "4.1.2"); + + assert!( + command.contains(r#"--conf "spark.openlineage.transport.url=https://marquez:5000""#) + ); + } + + /// 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 + lineage: + connection: + inline: + http: + 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::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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + sparkConf: + spark.app.name: spark-conf-name + "#}, + "spark-conf-name" + )] + #[case::lineage_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 + lineage: + appName: lineage-name + connection: + inline: + http: + host: marquez + port: 5000 + sparkConf: + spark.app.name: spark-conf-name + "#}, + "lineage-name" + )] + #[case::empty_lineage_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 + lineage: + appName: "" + connection: + inline: + http: + host: marquez + port: 5000 + sparkConf: + spark.app.name: spark-conf-name + "#}, + "spark-conf-name" + )] + #[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 + lineage: + connection: + inline: + http: + host: marquez + port: 5000 + "#}, + "metadata-name" + )] + fn test_openlineage_app_name_resolution(#[case] yaml: &str, #[case] expected_name: &str) { + let spark_application: v1alpha1::SparkApplication = + stackable_operator::utils::yaml_from_str_singleton_map(yaml).unwrap(); + let resolved = spark_application.resolved_lineage_app_name().unwrap(); + + assert_eq!(resolved, expected_name); + } + #[rstest] #[case::from_spark_image(None, "IfNotPresent")] #[case::spark_conf_wins(Some("Never"), "Never")] @@ -1699,7 +2383,7 @@ spec: .expect("the product image resolves"); let command = spark_application - .build_command(&None, &None, &resolved_product_image) + .build_command(&None, &None, &resolved_product_image, None) .expect("the submit command can be built") .join(" "); diff --git a/rust/operator-binary/src/crd/roles.rs b/rust/operator-binary/src/crd/roles.rs index f8428971..031832ef 100644 --- a/rust/operator-binary/src/crd/roles.rs +++ b/rust/operator-binary/src/crd/roles.rs @@ -29,7 +29,7 @@ use stackable_operator::{ merge::{Atomic, Merge}, }, constant, - crd::s3, + crd::{openlineage, s3}, k8s_openapi::{api::core::v1::VolumeMount, apimachinery::pkg::api::resource::Quantity}, product_logging::{self, spec::Logging}, schemars::{self, JsonSchema}, @@ -207,9 +207,16 @@ impl RoleConfig { spark_application: &v1alpha1::SparkApplication, s3conn: &Option, logdir: &Option, + open_lineage_conn: Option<&openlineage::ResolvedOpenLineageConnection>, ) -> Vec { 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, + ) } } diff --git a/rust/operator-binary/src/crd/template_merger.rs b/rust/operator-binary/src/crd/template_merger.rs index 8adcc4d9..f7389e27 100644 --- a/rust/operator-binary/src/crd/template_merger.rs +++ b/rust/operator-binary/src/crd/template_merger.rs @@ -58,6 +58,11 @@ pub fn deep_merge(base: &SparkApplication, overlay: &SparkApplication) -> SparkA .vector_aggregator_config_map_name .clone() .or_else(|| base.spec.vector_aggregator_config_map_name.clone()), + lineage: overlay + .spec + .lineage + .clone() + .or_else(|| base.spec.lineage.clone()), s3connection: overlay .spec .s3connection diff --git a/rust/operator-binary/src/crd/tlscerts.rs b/rust/operator-binary/src/crd/tlscerts.rs index a31bda07..630ca8da 100644 --- a/rust/operator-binary/src/crd/tlscerts.rs +++ b/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), + }), + }), + } = &crate::lineage::http_transport(conn).tls + { + return Some(secret_name); + } + + None +} + pub fn tls_secret_names<'a>( s3conn: &'a Option, logdir: &'a Option, + open_lineage_conn: Option<&'a openlineage::ResolvedOpenLineageConnection>, ) -> Option> { // 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 { diff --git a/rust/operator-binary/src/lineage.rs b/rust/operator-binary/src/lineage.rs new file mode 100644 index 00000000..cb55ca69 --- /dev/null +++ b/rust/operator-binary/src/lineage.rs @@ -0,0 +1,188 @@ +//! Operator-side OpenLineage helpers for [`SparkApplication`]. +//! +//! The reusable OpenLineage types (`OpenLineageConnectionSpec`, `InlineConnectionOrReference`, +//! `OpenLineageConfig`) live in the `stackable_operator::crd::openlineage` library module. This module +//! only holds the bits that depend on the operator's own `SparkApplication` (job-name resolution) +//! plus a spark-submit conf helper. +//! +//! [`SparkApplication`]: crate::crd::v1alpha1::SparkApplication + +use std::collections::BTreeMap; + +use snafu::OptionExt; +use stackable_operator::{ + crd::openlineage::{ + ResolvedOpenLineageConnection, + v1alpha1::{HttpTransport, OpenLineageTransport}, + }, + k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector}, +}; + +use crate::crd::{ + Error, ObjectHasNoNameSnafu, + constants::{ + OPENLINEAGE_AUTH_API_KEY_ENV, OPENLINEAGE_AUTH_SECRET_KEY, OPENLINEAGE_AUTH_TYPE_API_KEY, + OPENLINEAGE_AUTH_TYPE_ENV, OPENLINEAGE_TRANSPORT_ENDPOINT_ENV, + OPENLINEAGE_TRANSPORT_TYPE_ENV, OPENLINEAGE_TRANSPORT_TYPE_HTTP, + OPENLINEAGE_TRANSPORT_URL_ENV, + }, + v1alpha1, +}; + +/// Returns the HTTP transport of a resolved OpenLineage connection. +/// +/// `http` is currently the only [`OpenLineageTransport`] variant, so this cannot fail. When further +/// transports (for example Apache Kafka) are added upstream, this stops compiling and every caller +/// has to decide what the new transport means for Spark. +pub(crate) fn http_transport(conn: &ResolvedOpenLineageConnection) -> &HttpTransport { + let OpenLineageTransport::Http(http) = &conn.transport; + http +} + +/// The base URL (scheme, host and port, without path) of the OpenLineage HTTP transport, as expected +/// by `spark.openlineage.transport.url`. The path is passed separately as the endpoint. +pub(crate) fn transport_base_url(http: &HttpTransport) -> String { + format!( + "{scheme}://{host}:{port}", + scheme = if http.tls.uses_tls() { "https" } else { "http" }, + host = http.host, + port = http.port, + ) +} + +/// 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`). +pub(crate) 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()); + } + } +} + +/// Builds the driver env vars that deliver the **entire** OpenLineage HTTP transport — type, URL and +/// bearer-token auth — via the OpenLineage Java client's `OPENLINEAGE__` env-var configuration. +/// +/// This is used only when `credentialsSecretName` is configured. The whole transport must come from +/// one source: OpenLineage resolves `transport` as a unit, so if `spark.openlineage.transport.type` +/// or `.url` were set via `--conf`, the transport would be taken entirely from SparkConf and the +/// env-provided `auth` sub-tree would be silently dropped (verified against openlineage-spark +/// 1.51.0). Delivering the token as a `secretKeyRef` also keeps it out of the spark-submit `--conf` +/// args and the Job/pod spec — the operator never reads it. +pub(crate) fn openlineage_transport_env_vars( + transport_url: &str, + endpoint: &str, + secret_name: &str, +) -> Vec { + let literal = |name: &str, value: &str| EnvVar { + name: name.to_string(), + value: Some(value.to_string()), + value_from: None, + }; + + vec![ + literal( + OPENLINEAGE_TRANSPORT_TYPE_ENV, + OPENLINEAGE_TRANSPORT_TYPE_HTTP, + ), + literal(OPENLINEAGE_TRANSPORT_URL_ENV, transport_url), + literal(OPENLINEAGE_TRANSPORT_ENDPOINT_ENV, endpoint), + literal(OPENLINEAGE_AUTH_TYPE_ENV, OPENLINEAGE_AUTH_TYPE_API_KEY), + EnvVar { + name: OPENLINEAGE_AUTH_API_KEY_ENV.to_string(), + value: None, + value_from: Some(EnvVarSource { + secret_key_ref: Some(SecretKeySelector { + name: secret_name.to_string(), + key: OPENLINEAGE_AUTH_SECRET_KEY.to_string(), + optional: None, + }), + ..EnvVarSource::default() + }), + }, + ] +} + +impl v1alpha1::SparkApplication { + /// Resolves the stable OpenLineage job/app name, in priority order: + /// 1. `lineage.appName`, when set and not empty, else + /// 2. `spark.app.name` from `sparkConf`, else + /// 3. `metadata.name`. + /// + /// Always yields a non-blank name — which is exactly what fixes the intermittent `unknown` bug. + pub fn resolved_lineage_app_name(&self) -> Result { + if let Some(app_name) = self + .spec + .lineage + .as_ref() + .and_then(|lineage| lineage.app_name.as_ref()) + .filter(|app_name| !app_name.is_empty()) + { + return Ok(app_name.clone()); + } + + if let Some(app_name) = self.spec.spark_conf.get("spark.app.name") { + return Ok(app_name.clone()); + } + + self.metadata.name.clone().context(ObjectHasNoNameSnafu) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transport_env_vars_carry_full_transport_with_token_secret_ref() { + let vars = openlineage_transport_env_vars( + "https://marquez:5000", + HttpTransport::DEFAULT_PATH, + "my-secret", + ); + + let get = |name: &str| { + vars.iter() + .find(|v| v.name == name) + .unwrap_or_else(|| panic!("env var {name} present")) + }; + + // The whole transport is delivered via env so OpenLineage keeps the auth sub-tree. + assert_eq!( + get(OPENLINEAGE_TRANSPORT_TYPE_ENV).value.as_deref(), + Some(OPENLINEAGE_TRANSPORT_TYPE_HTTP) + ); + assert_eq!( + get(OPENLINEAGE_TRANSPORT_URL_ENV).value.as_deref(), + Some("https://marquez:5000") + ); + assert_eq!( + get(OPENLINEAGE_TRANSPORT_ENDPOINT_ENV).value.as_deref(), + Some(HttpTransport::DEFAULT_PATH) + ); + assert_eq!( + get(OPENLINEAGE_AUTH_TYPE_ENV).value.as_deref(), + Some(OPENLINEAGE_AUTH_TYPE_API_KEY) + ); + + let key_var = get(OPENLINEAGE_AUTH_API_KEY_ENV); + assert!( + key_var.value.is_none(), + "the token must never be delivered as a literal value" + ); + let selector = key_var + .value_from + .as_ref() + .and_then(|source| source.secret_key_ref.as_ref()) + .expect("api key sourced from a secretKeyRef"); + assert_eq!(selector.name, "my-secret"); + assert_eq!(selector.key, OPENLINEAGE_AUTH_SECRET_KEY); + } +} diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 2007a80f..9687ea83 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -55,6 +55,7 @@ mod config; mod connect; mod crd; mod history; +mod lineage; mod pod_driver_controller; mod product_logging; mod spark_k8s_controller; @@ -83,6 +84,7 @@ struct Opts { pub struct Ctx { pub client: stackable_operator::client::Client, pub operator_environment: OperatorEnvironmentOptions, + pub event_recorder: Recorder, } #[tokio::main] @@ -167,11 +169,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 { @@ -179,6 +176,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), @@ -255,11 +260,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 { @@ -267,6 +267,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::>( @@ -336,11 +342,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 { @@ -348,6 +349,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/build/mod.rs b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs index e7af9379..3a5d6a7c 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs @@ -56,6 +56,7 @@ type Result = std::result::Result; pub fn build(validated: &ValidatedSparkApplication) -> Result> { let spark_application = &validated.spark_application; let opt_s3conn = &validated.cluster_config.s3_connection; + let opt_open_lineage_conn = validated.cluster_config.open_lineage_connection.as_ref(); let logdir = &validated.cluster_config.log_dir; let resolved_product_image = &validated.resolved_product_image; @@ -63,7 +64,13 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result Result Result Result { // The static truststore mount is added first; the certificate mounts below embed the // user-supplied SecretClass names and therefore stay fallible. @@ -259,6 +260,7 @@ 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 mut cb = new_container_builder(SparkContainer::Spark.name()); @@ -297,7 +299,12 @@ pub(crate) fn pod_template( cb.add_volume_mount(VOLUME_MOUNT_NAME_CONFIG.as_ref(), VOLUME_MOUNT_PATH_CONFIG) .expect("The mount paths are statically defined and there should be no duplicates.") // These mounts include the user-supplied `volumeMounts`, so this add stays fallible. - .add_volume_mounts(config.volume_mounts(spark_application, s3conn, logdir)) + .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()) @@ -480,6 +487,7 @@ mod tests { let validated = validate( DereferencedSparkApplication { + open_lineage_connection: None, spark_application, resolved_template_refs: Vec::new(), s3_connection: Some(s3_connection), @@ -541,6 +549,7 @@ mod tests { let validated = validate( DereferencedSparkApplication { + open_lineage_connection: None, spark_application, resolved_template_refs: Vec::new(), s3_connection: None, @@ -560,7 +569,7 @@ mod tests { .expect("the driver config resolves"); let env = validated .spark_application - .env(&None, &None) + .env(&None, &None, None) .expect("the base environment can be built"); let service_account = ServiceAccount { metadata: ObjectMeta { @@ -613,6 +622,7 @@ mod tests { validate( DereferencedSparkApplication { + open_lineage_connection: None, spark_application, resolved_template_refs: Vec::new(), s3_connection, @@ -670,7 +680,7 @@ mod tests { .expect("the driver config resolves"); let env = validated .spark_application - .env(&None, &None) + .env(&None, &None, None) .expect("the base environment can be built"); let result = pod_template( @@ -729,7 +739,7 @@ mod tests { .expect("the driver config resolves"); let env = validated .spark_application - .env(s3conn, &None) + .env(s3conn, &None, None) .expect("the base environment can be built"); let requested_secret_lifetime = driver_config .requested_secret_lifetime @@ -737,7 +747,7 @@ mod tests { let volumes = validated .spark_application - .volumes(s3conn, &None, None, &requested_secret_lifetime) + .volumes(s3conn, &None, None, &requested_secret_lifetime, None) .expect("the SparkApplication volumes can be built"); let volume_names: Vec<&str> = volumes.iter().map(|volume| volume.name.as_str()).collect(); assert!(volume_names.contains(&"s3-creds"), "{volume_names:?}"); diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs index bf30b482..3aa88747 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs @@ -73,6 +73,7 @@ pub(crate) fn pod_template_config_map( 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 cm_name = spark_application.pod_template_config_map_name(role.clone()); let log_config_map = if let Some(ContainerLogConfig { @@ -96,6 +97,7 @@ pub(crate) fn pod_template_config_map( logdir, Some(&log_config_map), &requested_secret_lifetime, + open_lineage_conn, ) .context(CreateVolumesSnafu)?; volumes.push( diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs index 2616648b..abebea1f 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs @@ -61,6 +61,7 @@ pub(crate) fn spark_job( let spark_image = &validated.resolved_product_image; 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 mut cb = new_container_builder(SparkContainer::SparkSubmit.name()); // The SPARK_SUBMIT_OPTS env var is used to configure the JVM settings of the spark-submit job. @@ -70,7 +71,7 @@ pub(crate) fn spark_job( let mut spark_submit_opts_env = vec![format!( "-Djava.security.properties={VOLUME_MOUNT_PATH_CONFIG}/{JVM_SECURITY_PROPERTIES_FILE}" )]; - if tlscerts::tls_secret_names(s3conn, logdir).is_some() { + if tlscerts::tls_secret_names(s3conn, logdir, open_lineage_conn).is_some() { spark_submit_opts_env.push(format!( "-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12" )); @@ -102,7 +103,11 @@ pub(crate) fn spark_job( .add_volume_mount(VOLUME_MOUNT_NAME_CONFIG.as_ref(), VOLUME_MOUNT_PATH_CONFIG) .expect("The mount paths are statically defined and there should be no duplicates.") // These mounts include the user-supplied `volumeMounts`, so this add stays fallible. - .add_volume_mounts(spark_application.spark_job_volume_mounts(s3conn, logdir)) + .add_volume_mounts(spark_application.spark_job_volume_mounts( + s3conn, + logdir, + open_lineage_conn, + )) .context(AddVolumeMountSnafu)? .add_env_vars(merged_env); @@ -126,7 +131,13 @@ pub(crate) fn spark_job( .context(MissingSecretLifetimeSnafu)?; volumes.extend( spark_application - .volumes(s3conn, logdir, None, &requested_secret_lifetime) + .volumes( + s3conn, + logdir, + None, + &requested_secret_lifetime, + open_lineage_conn, + ) .context(CreateVolumesSnafu)?, ); @@ -230,6 +241,7 @@ mod tests { let validated = validate( DereferencedSparkApplication { + open_lineage_connection: None, spark_application, resolved_template_refs: Vec::new(), s3_connection: None, @@ -249,7 +261,7 @@ mod tests { .expect("the submit config resolves"); let env = validated .spark_application - .env(&None, &None) + .env(&None, &None, None) .expect("the base environment can be built"); let service_account = ServiceAccount { metadata: ObjectMeta { diff --git a/rust/operator-binary/src/spark_k8s_controller/dereference.rs b/rust/operator-binary/src/spark_k8s_controller/dereference.rs index aa68ea14..789952b3 100644 --- a/rust/operator-binary/src/spark_k8s_controller/dereference.rs +++ b/rust/operator-binary/src/spark_k8s_controller/dereference.rs @@ -5,7 +5,10 @@ //! Synchronous validation belongs in the sibling [`super::validate`] module. use snafu::{ResultExt, Snafu}; -use stackable_operator::{client::Client, crd::s3}; +use stackable_operator::{ + client::Client, + crd::{openlineage, s3}, +}; use crate::crd::{ logdir::ResolvedLogDir, @@ -23,6 +26,11 @@ pub enum Error { source: stackable_operator::crd::s3::v1alpha1::ConnectionError, }, + #[snafu(display("failed to resolve OpenLineage connection"))] + ResolveOpenLineageConnection { + source: stackable_operator::crd::openlineage::v1alpha1::OpenLineageError, + }, + #[snafu(display("failed to resolve log directory"))] LogDir { source: crate::crd::logdir::Error }, @@ -40,6 +48,9 @@ pub struct DereferencedSparkApplication { pub resolved_template_refs: Vec, /// Resolved S3 connection, if `spec.s3connection` is set. pub s3_connection: Option, + /// Resolved OpenLineage backend connection, if `spec.lineage` is set. Its + /// `credentialsSecretName` (when set) carries the bearer-token Secret for authentication. + pub open_lineage_connection: Option, /// Resolved log directory, if `spec.log_file_directory` is set. pub log_dir: Option, } @@ -73,7 +84,20 @@ pub async fn dereference( None => None, }; - // 3. Log directory (also pulls S3Bucket + TLS secret internally). + // 3. OpenLineage connection (inline or referenced `OpenLineageConnection`). + let open_lineage_connection = match merged_app.spec.lineage.as_ref() { + Some(job) => Some( + job.open_lineage + .connection + .clone() + .resolve(client, namespace) + .await + .context(ResolveOpenLineageConnectionSnafu)?, + ), + None => None, + }; + + // 4. Log directory (also pulls S3Bucket + TLS secret internally). let log_dir = match merged_app.spec.log_file_directory.as_ref() { Some(log_file_dir) => Some( ResolvedLogDir::resolve(log_file_dir, merged_app.metadata.namespace.clone(), client) @@ -87,6 +111,7 @@ pub async fn dereference( spark_application: merged_app, resolved_template_refs, s3_connection, + open_lineage_connection, log_dir, }) } diff --git a/rust/operator-binary/src/spark_k8s_controller/validate.rs b/rust/operator-binary/src/spark_k8s_controller/validate.rs index 9c76fcb5..7118c7e1 100644 --- a/rust/operator-binary/src/spark_k8s_controller/validate.rs +++ b/rust/operator-binary/src/spark_k8s_controller/validate.rs @@ -13,7 +13,7 @@ use stackable_operator::{ tls_verification::TlsVerification, }, constant, - crd::s3, + crd::{openlineage, s3}, k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta, kube::Resource, v2::{ @@ -102,6 +102,7 @@ pub struct ValidatedSparkApplication { pub struct ValidatedClusterConfig { pub resolved_template_refs: Vec, pub s3_connection: Option, + pub open_lineage_connection: Option, pub log_dir: Option, } @@ -197,6 +198,7 @@ pub fn validate( cluster_config: ValidatedClusterConfig { resolved_template_refs: dereferenced.resolved_template_refs, s3_connection: dereferenced.s3_connection, + open_lineage_connection: dereferenced.open_lineage_connection, log_dir: dereferenced.log_dir, }, }) @@ -258,6 +260,7 @@ mod tests { resolved_template_refs: Vec::new(), s3_connection: None, log_dir: None, + open_lineage_connection: None, }, &OperatorEnvironmentOptions { operator_namespace: "stackable-operators".to_string(), diff --git a/rust/operator-binary/src/test_support.rs b/rust/operator-binary/src/test_support.rs index 657e170f..3b13819f 100644 --- a/rust/operator-binary/src/test_support.rs +++ b/rust/operator-binary/src/test_support.rs @@ -7,7 +7,14 @@ use stackable_operator::{ cli::OperatorEnvironmentOptions, client::Client, commons::networking::DomainName, - kube::{Client as KubeClient, Config, core::DeserializeGuard, runtime::controller::Action}, + kube::{ + Client as KubeClient, Config, + core::DeserializeGuard, + runtime::{ + controller::Action, + events::{Recorder, Reporter}, + }, + }, utils::cluster_info::KubernetesClusterInfo, }; @@ -47,12 +54,13 @@ where .build() .expect("current-thread tokio runtime") .block_on(async { + let kube_client = KubeClient::try_from(Config::new( + "http://127.0.0.1:1".parse().expect("valid static URI"), + )) + .expect("client from static config"); let ctx = Arc::new(Ctx { client: Client::new( - KubeClient::try_from(Config::new( - "http://127.0.0.1:1".parse().expect("valid static URI"), - )) - .expect("client from static config"), + kube_client.clone(), None, "default".to_owned(), KubernetesClusterInfo { @@ -65,6 +73,13 @@ where operator_service_name: "spark-k8s-operator".to_owned(), image_repository: "oci.stackable.tech/sdp".to_owned(), }, + event_recorder: Recorder::new( + kube_client, + Reporter { + controller: "test".to_owned(), + instance: None, + }, + ), }); reconcile(Arc::new(object), ctx).await diff --git a/tests/templates/kuttl/openlineage/00-assert.yaml b/tests/templates/kuttl/openlineage/00-assert.yaml new file mode 100644 index 00000000..5baf8caa --- /dev/null +++ b/tests/templates/kuttl/openlineage/00-assert.yaml @@ -0,0 +1,9 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 900 +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: integration-tests-sa diff --git a/tests/templates/kuttl/openlineage/00-patch-ns.yaml.j2 b/tests/templates/kuttl/openlineage/00-patch-ns.yaml.j2 new file mode 100644 index 00000000..67185acf --- /dev/null +++ b/tests/templates/kuttl/openlineage/00-patch-ns.yaml.j2 @@ -0,0 +1,9 @@ +{% if test_scenario['values']['openshift'] == 'true' %} +# see https://github.com/stackabletech/issues/issues/566 +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}' + timeout: 120 +{% endif %} diff --git a/tests/templates/kuttl/openlineage/00-serviceaccount.yaml.j2 b/tests/templates/kuttl/openlineage/00-serviceaccount.yaml.j2 new file mode 100644 index 00000000..9cbf0351 --- /dev/null +++ b/tests/templates/kuttl/openlineage/00-serviceaccount.yaml.j2 @@ -0,0 +1,29 @@ +--- +kind: Role +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: use-integration-tests-scc +rules: +{% if test_scenario['values']['openshift'] == "true" %} + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + resourceNames: ["privileged"] + verbs: ["use"] +{% endif %} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: integration-tests-sa +--- +kind: RoleBinding +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: use-integration-tests-scc +subjects: + - kind: ServiceAccount + name: integration-tests-sa +roleRef: + kind: Role + name: use-integration-tests-scc + apiGroup: rbac.authorization.k8s.io diff --git a/tests/templates/kuttl/openlineage/01-assert.yaml.j2 b/tests/templates/kuttl/openlineage/01-assert.yaml.j2 new file mode 100644 index 00000000..50b1d4c3 --- /dev/null +++ b/tests/templates/kuttl/openlineage/01-assert.yaml.j2 @@ -0,0 +1,10 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +{% endif %} diff --git a/tests/templates/kuttl/openlineage/01-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/openlineage/01-install-vector-aggregator-discovery-configmap.yaml.j2 new file mode 100644 index 00000000..2d6a0df5 --- /dev/null +++ b/tests/templates/kuttl/openlineage/01-install-vector-aggregator-discovery-configmap.yaml.j2 @@ -0,0 +1,9 @@ +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +data: + ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }} +{% endif %} diff --git a/tests/templates/kuttl/openlineage/10-assert.yaml b/tests/templates/kuttl/openlineage/10-assert.yaml new file mode 100644 index 00000000..188a11e8 --- /dev/null +++ b/tests/templates/kuttl/openlineage/10-assert.yaml @@ -0,0 +1,12 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 300 +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: openlineage-receiver +status: + availableReplicas: 1 + readyReplicas: 1 diff --git a/tests/templates/kuttl/openlineage/10-install-receiver.yaml.j2 b/tests/templates/kuttl/openlineage/10-install-receiver.yaml.j2 new file mode 100644 index 00000000..c3390131 --- /dev/null +++ b/tests/templates/kuttl/openlineage/10-install-receiver.yaml.j2 @@ -0,0 +1,7 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # `envsubst` injects the kuttl-assigned $NAMESPACE (only $NAMESPACE, so the nginx config's own + # `$` tokens are left untouched) into the SecretClass name / cert scope. + - script: envsubst '$NAMESPACE' < 10_receiver.yaml | kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/openlineage/10_receiver.yaml.j2 b/tests/templates/kuttl/openlineage/10_receiver.yaml.j2 new file mode 100644 index 00000000..d880244c --- /dev/null +++ b/tests/templates/kuttl/openlineage/10_receiver.yaml.j2 @@ -0,0 +1,163 @@ +# Lightweight OpenLineage receiver. +# +# Instead of a full Marquez, this is a plain nginx that accepts the OpenLineage HTTP transport's +# `POST /api/v1/lineage` requests and persists each request body as a flat file. The assertion in +# `30-assert.yaml` then greps those files to prove events were delivered (and, in the TLS scenario, +# that the driver trusted the secret-operator-issued server cert end-to-end). +# +# nginx cannot write a request body to disk on a `return` (the body is discarded), so the receive +# location sets `client_body_in_file_only on` and `proxy_pass`es to a local sink server; nginx then +# buffers the full body to a file (kept, not cleaned) before forwarding. Bodies land in +# `/tmp/lineage/`. +# +# `$NAMESPACE` is substituted by the `10-install-receiver` step via `envsubst` (kuttl assigns the +# namespace at run time; beku leaves the literal in place). +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} +--- +# A genuine autoTls SecretClass: the secret operator generates the CA and issues the nginx server +# cert. Suffixed with the (cluster-scoped) namespace to avoid collisions across parallel scenarios. +apiVersion: secrets.stackable.tech/v1alpha1 +kind: SecretClass +metadata: + name: openlineage-tls-$NAMESPACE +spec: + backend: + autoTls: + ca: + autoGenerate: true + secret: + name: openlineage-tls-ca + namespace: $NAMESPACE +{% endif %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: openlineage-receiver-config +data: + nginx.conf: | + worker_processes 1; + error_log /dev/stderr notice; + pid /tmp/nginx.pid; + events { worker_connections 1024; } + http { + access_log /dev/stdout; + client_body_temp_path /tmp/nginx-body; + proxy_temp_path /tmp/nginx-proxy; + fastcgi_temp_path /tmp/nginx-fastcgi; + uwsgi_temp_path /tmp/nginx-uwsgi; + scgi_temp_path /tmp/nginx-scgi; + + # Internal sink: acknowledges the forwarded lineage event. + server { + listen 127.0.0.1:8080; + location / { return 201; } + } + + server { + listen 5000{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} ssl{% endif %}; +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + ssl_certificate /stackable/tls/tls.crt; + ssl_certificate_key /stackable/tls/tls.key; +{% endif %} + client_max_body_size 16m; + + # Persist each OpenLineage event body to /tmp/lineage/, then acknowledge via the sink. + location = /api/v1/lineage { +{% if test_scenario['values']['openlineage-use-auth'] == 'true' %} + # Require the bearer token the operator wires in from the credentialsSecretName Secret. + # Without a valid token nginx returns 401 and never persists a body, so the 30-assert + # event checks only pass if the driver actually authenticated. + if ($http_authorization != "Bearer openlineage-test-token") { return 401; } +{% endif %} + client_body_in_file_only on; + client_body_temp_path /tmp/lineage; + proxy_pass http://127.0.0.1:8080; + } + + location / { return 204; } + } + } +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: openlineage-receiver + labels: + app: openlineage-receiver +spec: + replicas: 1 + selector: + matchLabels: + app: openlineage-receiver + template: + metadata: + labels: + app: openlineage-receiver + spec: +{% if test_scenario['values']['openshift'] != 'true' %} + # On plain Kubernetes, make mounted emptyDirs writable by the nginx-unprivileged group (101). + # On OpenShift this is omitted so the restricted SCC can assign an fsGroup from the project range. + securityContext: + fsGroup: 101 +{% endif %} + containers: + - name: nginx + image: docker.io/nginxinc/nginx-unprivileged:1.27-alpine + ports: + - containerPort: 5000 + volumeMounts: + - name: config + mountPath: /etc/nginx/nginx.conf + subPath: nginx.conf + - name: tmp + mountPath: /tmp +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + - name: tls + mountPath: /stackable/tls +{% endif %} + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 128Mi + volumes: + - name: config + configMap: + name: openlineage-receiver-config + - name: tmp + emptyDir: {} +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + - name: tls + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: openlineage-tls-$NAMESPACE + # `service` scope puts openlineage-receiver..svc.cluster.local in the cert SANs, + # which is exactly the host the Spark driver dials (see the SparkApplication). + secrets.stackable.tech/scope: service=openlineage-receiver + secrets.stackable.tech/format: tls-pem + spec: + accessModes: ["ReadWriteOnce"] + resources: + requests: + storage: "1" + storageClassName: secrets.stackable.tech +{% endif %} +--- +apiVersion: v1 +kind: Service +metadata: + name: openlineage-receiver + labels: + app: openlineage-receiver +spec: + selector: + app: openlineage-receiver + ports: + - name: http + port: 5000 + targetPort: 5000 diff --git a/tests/templates/kuttl/openlineage/20-assert.yaml b/tests/templates/kuttl/openlineage/20-assert.yaml new file mode 100644 index 00000000..ac15a9a5 --- /dev/null +++ b/tests/templates/kuttl/openlineage/20-assert.yaml @@ -0,0 +1,11 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 900 +--- +apiVersion: spark.stackable.tech/v1alpha1 +kind: SparkApplication +metadata: + name: openlineage +status: + phase: Succeeded diff --git a/tests/templates/kuttl/openlineage/20-deploy-spark-app.yaml.j2 b/tests/templates/kuttl/openlineage/20-deploy-spark-app.yaml.j2 new file mode 100644 index 00000000..6519d8a4 --- /dev/null +++ b/tests/templates/kuttl/openlineage/20-deploy-spark-app.yaml.j2 @@ -0,0 +1,7 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # `envsubst '$NAMESPACE'` fills in the connection host FQDN (and the TLS secretClass name) while + # leaving any other `$` tokens in the PySpark script untouched. + - script: envsubst '$NAMESPACE' < 20_spark-app.yaml | kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/openlineage/20_spark-app.yaml.j2 b/tests/templates/kuttl/openlineage/20_spark-app.yaml.j2 new file mode 100644 index 00000000..26207ba8 --- /dev/null +++ b/tests/templates/kuttl/openlineage/20_spark-app.yaml.j2 @@ -0,0 +1,98 @@ +# The SparkApplication under test plus its PySpark job. +# +# The job reads/writes a small DataFrame so the OpenLineage Spark listener emits START/COMPLETE run +# events. The operator injects the listener, the jar, and the `spark.openlineage.*` config from the +# `lineage` block below; the events are POSTed to the nginx receiver. +# +# `$NAMESPACE` is substituted by the `20-deploy-spark-app` step so the connection host is the fully +# qualified Service DNS name — which is what the secret-operator `service` scope puts in the cert +# SANs, so TLS hostname verification on the driver succeeds. +{% if test_scenario['values']['openlineage-use-auth'] == 'true' %} +--- +# Token Secret for the auth scenario. The Secret is namespaced and must live in the workload +# namespace; the operator wires its `apiKey` entry into the driver as the OpenLineage bearer token +# via a secretKeyRef env var (referenced by `credentialsSecretName` on the connection). +apiVersion: v1 +kind: Secret +metadata: + name: openlineage-token +stringData: + apiKey: "openlineage-test-token" +{% endif %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: openlineage-job +data: + openlineage-job.py: | + from pyspark.sql import SparkSession + + spark = SparkSession.builder.appName("openlineage-test-app").getOrCreate() + + df = spark.createDataFrame( + [(1, "a"), (2, "b"), (2, "c"), (3, "d")], + ["id", "grp"], + ) + # An aggregation + collect triggers a SQL execution, which makes the OpenLineage listener emit + # START/COMPLETE run events. We deliberately avoid a filesystem write: in cluster mode the driver + # and executors have separate local filesystems, so a write to a local path would fail the commit. + result = df.groupBy("grp").count().collect() + print(f"aggregation result: {result}") + + spark.stop() +--- +apiVersion: spark.stackable.tech/v1alpha1 +kind: SparkApplication +metadata: + name: openlineage +spec: +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + sparkImage: +{% if test_scenario['values']['spark'].find(",") > 0 %} + custom: "{{ test_scenario['values']['spark'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['spark'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['spark'] }}" +{% endif %} + pullPolicy: IfNotPresent + mode: cluster + mainApplicationFile: "local:///stackable/spark/jobs/openlineage-job.py" + lineage: + appName: openlineage-test-app + connection: + inline: + http: + host: openlineage-receiver.$NAMESPACE.svc.cluster.local + port: 5000 +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + tls: + verification: + server: + caCert: + secretClass: openlineage-tls-$NAMESPACE +{% endif %} +{% if test_scenario['values']['openlineage-use-auth'] == 'true' %} + credentialsSecretName: openlineage-token +{% endif %} + driver: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + volumeMounts: + - name: script + mountPath: /stackable/spark/jobs + executor: + replicas: 1 + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + volumeMounts: + - name: script + mountPath: /stackable/spark/jobs + volumes: + - name: script + configMap: + name: openlineage-job diff --git a/tests/templates/kuttl/openlineage/30-assert.yaml b/tests/templates/kuttl/openlineage/30-assert.yaml new file mode 100644 index 00000000..0120b54b --- /dev/null +++ b/tests/templates/kuttl/openlineage/30-assert.yaml @@ -0,0 +1,14 @@ +--- +# Verify the OpenLineage events actually reached the receiver. In the TLS scenario this also proves +# the driver trusted the secret-operator-issued cert end-to-end (a failed handshake would mean no +# body files were written). +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 120 +commands: + - script: | + POD=$(kubectl get -n "$NAMESPACE" pod -l app=openlineage-receiver -o jsonpath='{.items[0].metadata.name}') + # At least one lineage event body was persisted, mentioning our job name ... + kubectl exec -n "$NAMESPACE" "$POD" -- sh -c 'cat /tmp/lineage/* 2>/dev/null | grep -q "openlineage-test-app"' + # ... and a COMPLETE run event was received (the job finished and reported lineage). + kubectl exec -n "$NAMESPACE" "$POD" -- sh -c 'cat /tmp/lineage/* 2>/dev/null | grep -q "COMPLETE"' diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 9247d55c..5cdcf23d 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -70,6 +70,14 @@ dimensions: values: - "false" - "true" + - name: openlineage-use-tls + values: + - "false" + - "true" + - name: openlineage-use-auth + values: + - "false" + - "true" tests: - name: smoke dimensions: @@ -169,6 +177,12 @@ tests: - kerberos-realm - openshift - s3-use-tls + - name: openlineage + dimensions: + - spark + - openlineage-use-tls + - openlineage-use-auth + - openshift suites: - name: nightly @@ -177,6 +191,8 @@ suites: - expr: last - name: s3-use-tls expr: "true" + - name: openlineage-use-tls + expr: "true" - name: smoke-latest select: - smoke @@ -191,3 +207,5 @@ suites: expr: "true" - name: s3-use-tls expr: "true" + - name: openlineage-use-tls + expr: "true"