From 1315e2135486e1512737729a85d5d31e10222ef5 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 17 Jul 2026 09:42:21 +0200 Subject: [PATCH 01/17] feat: add crd::openlineage module --- crates/stackable-operator/CHANGELOG.md | 8 + .../crds/OpenLineageConnection.yaml | 91 ++++++++++ crates/stackable-operator/src/crd/mod.rs | 1 + .../src/crd/openlineage/mod.rs | 157 ++++++++++++++++++ .../src/crd/openlineage/v1alpha1_impl.rs | 62 +++++++ crates/xtask/src/crd/mod.rs | 2 + 6 files changed, 321 insertions(+) create mode 100644 crates/stackable-operator/crds/OpenLineageConnection.yaml create mode 100644 crates/stackable-operator/src/crd/openlineage/mod.rs create mode 100644 crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index 8f50b4a44..387025091 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -4,6 +4,14 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- Add `crd::openlineage` module with the `OpenLineageConnection` CRD (a reusable connection to an + OpenLineage backend), an `InlineConnectionOrReference` wrapper with `resolve()`, and an embeddable + `OpenLineageJob` type for operators ([#XXXX]). + +[#XXXX]: https://github.com/stackabletech/operator-rs/pull/XXXX + ## [0.113.4] - 2026-07-09 ### Changed diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml new file mode 100644 index 000000000..d7a9e5abd --- /dev/null +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -0,0 +1,91 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: openlineageconnections.openlineage.stackable.tech +spec: + group: openlineage.stackable.tech + names: + categories: [] + kind: OpenLineageConnection + plural: openlineageconnections + shortNames: [] + singular: openlineageconnection + scope: Namespaced + versions: + - additionalPrinterColumns: [] + name: v1alpha1 + schema: + openAPIV3Schema: + description: A reusable definition of a connection to an OpenLineage backend. + properties: + spec: + description: |- + OpenLineage connection definition as a resource. + Learn more about [OpenLineage](https://openlineage.io/). + properties: + host: + description: 'Host of the OpenLineage backend without any protocol or port. For example: `marquez`.' + 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 + required: + - spec + title: OpenLineageConnection + type: object + served: true + storage: true + subresources: {} diff --git a/crates/stackable-operator/src/crd/mod.rs b/crates/stackable-operator/src/crd/mod.rs index be7ddad02..d80d75c34 100644 --- a/crates/stackable-operator/src/crd/mod.rs +++ b/crates/stackable-operator/src/crd/mod.rs @@ -7,6 +7,7 @@ use serde::{Deserialize, Serialize}; pub mod authentication; pub mod git_sync; pub mod listener; +pub mod openlineage; pub mod s3; pub mod scaler; diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs new file mode 100644 index 000000000..bcd589452 --- /dev/null +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -0,0 +1,157 @@ +use kube::CustomResource; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +use crate::{commons::tls_verification::TlsClientDetails, versioned::versioned}; + +mod v1alpha1_impl; + +// FIXME (@Techassi): This should be versioned as well, but the macro cannot +// handle new-type structs yet. +/// Use this type in your operator! +pub type ResolvedOpenLineageConnection = v1alpha1::OpenLineageConnectionSpec; + +#[versioned( + version(name = "v1alpha1"), + crates( + kube_core = "kube::core", + k8s_openapi = "k8s_openapi", + schemars = "schemars", + ) +)] +pub mod versioned { + pub mod v1alpha1 { + pub use v1alpha1_impl::OpenLineageError; + } + + /// OpenLineage connection definition as a resource. + /// Learn more about [OpenLineage](https://openlineage.io/). + #[versioned(crd( + group = "openlineage.stackable.tech", + kind = "OpenLineageConnection", + plural = "openlineageconnections", + doc = "A reusable definition of a connection to an OpenLineage backend.", + namespaced + ))] + #[derive(CustomResource, Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct OpenLineageConnectionSpec { + /// Host of the OpenLineage backend without any protocol or port. For example: `marquez`. + pub host: String, + + /// Port the OpenLineage backend listens on. For example: `5000`. + pub port: u16, + + /// Use a TLS connection. If not specified no TLS will be used. + /// When TLS server verification is configured, the transport uses `https` instead of `http`. + #[serde(flatten)] + pub tls: TlsClientDetails, + } + + /// An OpenLineage connection, either inlined or referenced by the name of an + /// [`OpenLineageConnection`] resource in the same namespace. + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + // TODO: This probably should be serde(untagged), but this would be a breaking change + pub enum InlineConnectionOrReference { + Inline(OpenLineageConnectionSpec), + Reference(String), + } + + /// OpenLineage lineage-emission configuration for a single workload/application. + /// + /// Embed this in an operator's workload spec to enable OpenLineage for that workload. + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct OpenLineageJob { + /// The OpenLineage backend connection, either inlined or referencing an + /// `OpenLineageConnection` resource by name. + pub connection: InlineConnectionOrReference, + + /// The OpenLineage namespace lineage is reported under. + /// If unset, operators typically default to the workload's Kubernetes namespace. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub namespace: Option, + + /// A stable OpenLineage job/application name. Setting this prevents fragmented run history. + /// If unset, operators resolve a name from workload-specific configuration. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub app_name: Option, + } +} + +#[cfg(test)] +impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineageConnectionSpec { + fn roundtrip_test_data() -> Vec { + crate::utils::yaml_from_str_singleton_map(indoc::indoc! {" + - host: marquez + port: 5000 + - host: marquez + port: 5000 + tls: + verification: + none: {} + - host: marquez + port: 5000 + tls: + verification: + server: + caCert: + secretClass: openlineage-cert + "}) + .expect("Failed to parse OpenLineageConnectionSpec YAML") + } +} + +#[cfg(test)] +mod tests { + use crate::{ + commons::tls_verification::{ + CaCert, Tls, TlsClientDetails, TlsServerVerification, TlsVerification, + }, + crd::openlineage::v1alpha1::OpenLineageConnectionSpec, + }; + + #[test] + fn http_transport_url_without_tls() { + let connection = OpenLineageConnectionSpec { + host: "marquez".to_string(), + port: 5000, + tls: TlsClientDetails { tls: None }, + }; + + assert_eq!(connection.transport_url(), "http://marquez:5000"); + } + + #[test] + fn https_transport_url_with_server_verification() { + let connection = OpenLineageConnectionSpec { + host: "marquez".to_string(), + port: 5000, + tls: TlsClientDetails { + tls: Some(Tls { + verification: TlsVerification::Server(TlsServerVerification { + ca_cert: CaCert::WebPki {}, + }), + }), + }, + }; + + assert_eq!(connection.transport_url(), "https://marquez:5000"); + } + + #[test] + fn http_transport_url_without_verification() { + let connection = OpenLineageConnectionSpec { + host: "marquez".to_string(), + port: 5000, + tls: TlsClientDetails { + tls: Some(Tls { + verification: TlsVerification::None {}, + }), + }, + }; + + assert_eq!(connection.transport_url(), "http://marquez:5000"); + } +} diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs new file mode 100644 index 000000000..f76173085 --- /dev/null +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -0,0 +1,62 @@ +use snafu::{ResultExt as _, Snafu}; + +use crate::{ + client::Client, + crd::openlineage::{ + ResolvedOpenLineageConnection, + v1alpha1::{InlineConnectionOrReference, OpenLineageConnection, OpenLineageConnectionSpec}, + }, +}; + +#[derive(Debug, Snafu)] +pub enum OpenLineageError { + #[snafu(display("failed to retrieve OpenLineage connection '{open_lineage_connection}'"))] + RetrieveOpenLineageConnection { + #[snafu(source(from(crate::client::Error, Box::new)))] + source: Box, + open_lineage_connection: String, + }, +} + +impl OpenLineageConnectionSpec { + /// Build the OpenLineage transport URL from this connection. + /// + /// The scheme is `https` when TLS server verification is configured + /// (`tls.verification.server`), otherwise `http`. + pub fn transport_url(&self) -> String { + let scheme = if self.tls.uses_tls_verification() { + "https" + } else { + "http" + }; + + format!( + "{scheme}://{host}:{port}", + host = self.host, + port = self.port + ) + } +} + +impl InlineConnectionOrReference { + pub async fn resolve( + self, + client: &Client, + namespace: &str, + ) -> Result { + match self { + Self::Inline(inline) => Ok(inline), + Self::Reference(reference) => { + let connection_spec = client + .get::(&reference, namespace) + .await + .context(RetrieveOpenLineageConnectionSnafu { + open_lineage_connection: reference, + })? + .spec; + + Ok(connection_spec) + } + } + } +} diff --git a/crates/xtask/src/crd/mod.rs b/crates/xtask/src/crd/mod.rs index f3ed2caa0..7cb127306 100644 --- a/crates/xtask/src/crd/mod.rs +++ b/crates/xtask/src/crd/mod.rs @@ -9,6 +9,7 @@ use stackable_operator::{ Listener, ListenerClass, ListenerClassVersion, ListenerVersion, PodListeners, PodListenersVersion, }, + openlineage::{OpenLineageConnection, OpenLineageConnectionVersion}, s3::{S3Bucket, S3BucketVersion, S3Connection, S3ConnectionVersion}, scaler::{Scaler, ScalerVersion}, }, @@ -75,6 +76,7 @@ pub fn generate_preview() -> Result<(), Error> { write_crd!(path, AuthenticationClass, V1Alpha1); write_crd!(path, Listener, V1Alpha1); write_crd!(path, ListenerClass, V1Alpha1); + write_crd!(path, OpenLineageConnection, V1Alpha1); write_crd!(path, PodListeners, V1Alpha1); write_crd!(path, S3Bucket, V1Alpha1); write_crd!(path, S3Connection, V1Alpha1); From bdfb9367dc0d125358a76df05bdb02c05c404290 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 17 Jul 2026 09:47:41 +0200 Subject: [PATCH 02/17] attempt to fix cargo deny lint by updating "spin" version --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c6c504fd2..56e176500 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3632,9 +3632,9 @@ dependencies = [ [[package]] name = "spin" -version = "0.9.8" +version = "0.9.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" [[package]] name = "spki" From 8b1b821d6913d31a589cf7db5d089d982b220fb1 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:35:52 +0200 Subject: [PATCH 03/17] add optional authentication class reference --- .../crds/OpenLineageConnection.yaml | 9 +++++ .../src/crd/openlineage/mod.rs | 14 +++++++ .../src/crd/openlineage/v1alpha1_impl.rs | 39 +++++++++++++++++-- 3 files changed, 59 insertions(+), 3 deletions(-) diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index d7a9e5abd..375a06f60 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -24,6 +24,15 @@ spec: OpenLineage connection definition as a resource. Learn more about [OpenLineage](https://openlineage.io/). properties: + authenticationClassRef: + description: |- + Name of an [`AuthenticationClass`](https://docs.stackable.tech/home/nightly/concepts/authentication) used + to authenticate against the OpenLineage backend. The `AuthenticationClass` is cluster-scoped + and referenced by name; it is resolved at runtime via + [`OpenLineageConnectionSpec::resolve_authentication_class`]. If not specified, no + authentication is used. + nullable: true + type: string host: description: 'Host of the OpenLineage backend without any protocol or port. For example: `marquez`.' type: string diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index bcd589452..9f7f39666 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -46,6 +46,14 @@ pub mod versioned { /// When TLS server verification is configured, the transport uses `https` instead of `http`. #[serde(flatten)] pub tls: TlsClientDetails, + + /// Name of an [`AuthenticationClass`](DOCS_BASE_URL_PLACEHOLDER/concepts/authentication) used + /// to authenticate against the OpenLineage backend. The `AuthenticationClass` is cluster-scoped + /// and referenced by name; it is resolved at runtime via + /// [`OpenLineageConnectionSpec::resolve_authentication_class`]. If not specified, no + /// authentication is used. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub authentication_class_ref: Option, } /// An OpenLineage connection, either inlined or referenced by the name of an @@ -98,6 +106,9 @@ impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineag server: caCert: secretClass: openlineage-cert + - host: marquez + port: 5000 + authenticationClassRef: openlineage-auth "}) .expect("Failed to parse OpenLineageConnectionSpec YAML") } @@ -118,6 +129,7 @@ mod tests { host: "marquez".to_string(), port: 5000, tls: TlsClientDetails { tls: None }, + authentication_class_ref: None, }; assert_eq!(connection.transport_url(), "http://marquez:5000"); @@ -135,6 +147,7 @@ mod tests { }), }), }, + authentication_class_ref: None, }; assert_eq!(connection.transport_url(), "https://marquez:5000"); @@ -150,6 +163,7 @@ mod tests { verification: TlsVerification::None {}, }), }, + authentication_class_ref: None, }; assert_eq!(connection.transport_url(), "http://marquez:5000"); diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index f76173085..7a42af009 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -2,9 +2,14 @@ use snafu::{ResultExt as _, Snafu}; use crate::{ client::Client, - crd::openlineage::{ - ResolvedOpenLineageConnection, - v1alpha1::{InlineConnectionOrReference, OpenLineageConnection, OpenLineageConnectionSpec}, + crd::{ + authentication::core::v1alpha1::AuthenticationClass, + openlineage::{ + ResolvedOpenLineageConnection, + v1alpha1::{ + InlineConnectionOrReference, OpenLineageConnection, OpenLineageConnectionSpec, + }, + }, }, }; @@ -16,6 +21,13 @@ pub enum OpenLineageError { source: Box, open_lineage_connection: String, }, + + #[snafu(display("failed to retrieve AuthenticationClass '{authentication_class}'"))] + RetrieveAuthenticationClass { + #[snafu(source(from(crate::client::Error, Box::new)))] + source: Box, + authentication_class: String, + }, } impl OpenLineageConnectionSpec { @@ -36,6 +48,27 @@ impl OpenLineageConnectionSpec { port = self.port ) } + + /// Resolves the [`AuthenticationClass`] referenced by this connection, if any. + /// + /// Returns `Ok(None)` when no `authenticationClassRef` is configured. The `AuthenticationClass` + /// is cluster-scoped, so no namespace is required. + pub async fn resolve_authentication_class( + &self, + client: &Client, + ) -> Result, OpenLineageError> { + let Some(authentication_class_ref) = &self.authentication_class_ref else { + return Ok(None); + }; + + let resolved = AuthenticationClass::resolve(client, authentication_class_ref) + .await + .context(RetrieveAuthenticationClassSnafu { + authentication_class: authentication_class_ref.clone(), + })?; + + Ok(Some(resolved)) + } } impl InlineConnectionOrReference { From 01535bc1acb05e8a28f098d40bf5bef3105e075e Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 21 Jul 2026 15:16:31 +0200 Subject: [PATCH 04/17] refactor(openlineage): replace authenticationClassRef with credentialsSecretName AuthenticationClass is a server-side identity abstraction and does not fit outbound OpenLineage client->backend authentication, which is a static bearer token. Replace `authentication_class_ref` on OpenLineageConnectionSpec with an optional `credentials_secret_name` (a Secret holding the api key under `apiKey`) and drop `resolve_authentication_class`. See https://github.com/stackabletech/decisions/issues/90 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../crds/OpenLineageConnection.yaml | 9 ++--- .../src/crd/openlineage/mod.rs | 17 ++++---- .../src/crd/openlineage/v1alpha1_impl.rs | 39 ++----------------- 3 files changed, 15 insertions(+), 50 deletions(-) diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index 375a06f60..14439f431 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -24,12 +24,11 @@ spec: OpenLineage connection definition as a resource. Learn more about [OpenLineage](https://openlineage.io/). properties: - authenticationClassRef: + credentialsSecretName: description: |- - Name of an [`AuthenticationClass`](https://docs.stackable.tech/home/nightly/concepts/authentication) used - to authenticate against the OpenLineage backend. The `AuthenticationClass` is cluster-scoped - and referenced by name; it is resolved at runtime via - [`OpenLineageConnectionSpec::resolve_authentication_class`]. If not specified, no + 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 diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 9f7f39666..1dfb5442e 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -47,13 +47,12 @@ pub mod versioned { #[serde(flatten)] pub tls: TlsClientDetails, - /// Name of an [`AuthenticationClass`](DOCS_BASE_URL_PLACEHOLDER/concepts/authentication) used - /// to authenticate against the OpenLineage backend. The `AuthenticationClass` is cluster-scoped - /// and referenced by name; it is resolved at runtime via - /// [`OpenLineageConnectionSpec::resolve_authentication_class`]. If not specified, no + /// 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. #[serde(default, skip_serializing_if = "Option::is_none")] - pub authentication_class_ref: Option, + pub credentials_secret_name: Option, } /// An OpenLineage connection, either inlined or referenced by the name of an @@ -108,7 +107,7 @@ impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineag secretClass: openlineage-cert - host: marquez port: 5000 - authenticationClassRef: openlineage-auth + credentialsSecretName: openlineage-credentials "}) .expect("Failed to parse OpenLineageConnectionSpec YAML") } @@ -129,7 +128,7 @@ mod tests { host: "marquez".to_string(), port: 5000, tls: TlsClientDetails { tls: None }, - authentication_class_ref: None, + credentials_secret_name: None, }; assert_eq!(connection.transport_url(), "http://marquez:5000"); @@ -147,7 +146,7 @@ mod tests { }), }), }, - authentication_class_ref: None, + credentials_secret_name: None, }; assert_eq!(connection.transport_url(), "https://marquez:5000"); @@ -163,7 +162,7 @@ mod tests { verification: TlsVerification::None {}, }), }, - authentication_class_ref: None, + credentials_secret_name: None, }; assert_eq!(connection.transport_url(), "http://marquez:5000"); diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index 7a42af009..f76173085 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -2,14 +2,9 @@ use snafu::{ResultExt as _, Snafu}; use crate::{ client::Client, - crd::{ - authentication::core::v1alpha1::AuthenticationClass, - openlineage::{ - ResolvedOpenLineageConnection, - v1alpha1::{ - InlineConnectionOrReference, OpenLineageConnection, OpenLineageConnectionSpec, - }, - }, + crd::openlineage::{ + ResolvedOpenLineageConnection, + v1alpha1::{InlineConnectionOrReference, OpenLineageConnection, OpenLineageConnectionSpec}, }, }; @@ -21,13 +16,6 @@ pub enum OpenLineageError { source: Box, open_lineage_connection: String, }, - - #[snafu(display("failed to retrieve AuthenticationClass '{authentication_class}'"))] - RetrieveAuthenticationClass { - #[snafu(source(from(crate::client::Error, Box::new)))] - source: Box, - authentication_class: String, - }, } impl OpenLineageConnectionSpec { @@ -48,27 +36,6 @@ impl OpenLineageConnectionSpec { port = self.port ) } - - /// Resolves the [`AuthenticationClass`] referenced by this connection, if any. - /// - /// Returns `Ok(None)` when no `authenticationClassRef` is configured. The `AuthenticationClass` - /// is cluster-scoped, so no namespace is required. - pub async fn resolve_authentication_class( - &self, - client: &Client, - ) -> Result, OpenLineageError> { - let Some(authentication_class_ref) = &self.authentication_class_ref else { - return Ok(None); - }; - - let resolved = AuthenticationClass::resolve(client, authentication_class_ref) - .await - .context(RetrieveAuthenticationClassSnafu { - authentication_class: authentication_class_ref.clone(), - })?; - - Ok(Some(resolved)) - } } impl InlineConnectionOrReference { From b209765b96869a9948b8ce154bed48dc3eed6491 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 21 Jul 2026 16:43:56 +0200 Subject: [PATCH 05/17] update changelog with pr number --- crates/stackable-operator/CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index 387025091..eee4c6179 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -8,9 +8,9 @@ All notable changes to this project will be documented in this file. - Add `crd::openlineage` module with the `OpenLineageConnection` CRD (a reusable connection to an OpenLineage backend), an `InlineConnectionOrReference` wrapper with `resolve()`, and an embeddable - `OpenLineageJob` type for operators ([#XXXX]). + `OpenLineageJob` type for operators ([#1250]). -[#XXXX]: https://github.com/stackabletech/operator-rs/pull/XXXX +[#1250]: https://github.com/stackabletech/operator-rs/pull/XXXX ## [0.113.4] - 2026-07-09 From 69d113461c187e15ec5dc779602af02c78610887 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:09:11 +0200 Subject: [PATCH 06/17] fix markdown lint --- crates/stackable-operator/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index e208cf326..c86688483 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -10,7 +10,7 @@ All notable changes to this project will be documented in this file. - Add `crd::openlineage` module with the `OpenLineageConnection` CRD (a reusable connection to an OpenLineage backend), an `InlineConnectionOrReference` wrapper with `resolve()`, and an embeddable `OpenLineageJob` type for operators ([#1250]). - + ### Changed - [v2] BREAKING: Converting an `EnvVarSet` into a `Vec` takes dependencies between From 97f484c29ea77ba62a48fd6256bf69802a954a9b Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 24 Jul 2026 12:36:37 +0200 Subject: [PATCH 07/17] refactor(openlineage): rename OpenLineageJob.app_name to job_name and move CRD to lineage.stackable.tech Rename the embeddable OpenLineageJob field `app_name` to `job_name` (serialized `appName` -> `jobName`) and move the OpenLineageConnection CRD from the `openlineage.stackable.tech` API group to `lineage.stackable.tech`. Regenerated crds/OpenLineageConnection.yaml. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/stackable-operator/CHANGELOG.md | 5 +++-- crates/stackable-operator/crds/OpenLineageConnection.yaml | 4 ++-- crates/stackable-operator/src/crd/openlineage/mod.rs | 6 +++--- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index 4dbfa6e26..b020f566d 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -7,8 +7,9 @@ All notable changes to this project will be documented in this file. ### Added - Add `crd::openlineage` module with the `OpenLineageConnection` CRD (a reusable connection to an - OpenLineage backend), an `InlineConnectionOrReference` wrapper with `resolve()`, and an embeddable - `OpenLineageJob` type for operators ([#1250]). + OpenLineage backend, in the `lineage.stackable.tech` API group), an `InlineConnectionOrReference` + wrapper with `resolve()`, and an embeddable `OpenLineageJob` type (with a `jobName` field) for + operators ([#1250]). [#1250]: https://github.com/stackabletech/operator-rs/pull/1250 diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index 14439f431..ce1b89792 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -2,9 +2,9 @@ apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: - name: openlineageconnections.openlineage.stackable.tech + name: openlineageconnections.lineage.stackable.tech spec: - group: openlineage.stackable.tech + group: lineage.stackable.tech names: categories: [] kind: OpenLineageConnection diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 1dfb5442e..56dcaa2fa 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -27,7 +27,7 @@ pub mod versioned { /// OpenLineage connection definition as a resource. /// Learn more about [OpenLineage](https://openlineage.io/). #[versioned(crd( - group = "openlineage.stackable.tech", + group = "lineage.stackable.tech", kind = "OpenLineageConnection", plural = "openlineageconnections", doc = "A reusable definition of a connection to an OpenLineage backend.", @@ -80,10 +80,10 @@ pub mod versioned { #[serde(default, skip_serializing_if = "Option::is_none")] pub namespace: Option, - /// A stable OpenLineage job/application name. Setting this prevents fragmented run history. + /// A stable OpenLineage job name. Setting this prevents fragmented run history. /// If unset, operators resolve a name from workload-specific configuration. #[serde(default, skip_serializing_if = "Option::is_none")] - pub app_name: Option, + pub job_name: Option, } } From 59a36ec258c2d39c2f56adff17fb4a96306e4af1 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:44:12 +0200 Subject: [PATCH 08/17] refactor: rename OpenLineageJob -> OpenLineageConfig and make namespace required. --- crates/stackable-operator/CHANGELOG.md | 2 +- .../stackable-operator/src/crd/openlineage/mod.rs | 7 +++---- .../src/crd/openlineage/v1alpha1_impl.rs | 14 +++++++++++++- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index b020f566d..4e1043548 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -8,7 +8,7 @@ All notable changes to this project will be documented in this file. - Add `crd::openlineage` module with the `OpenLineageConnection` CRD (a reusable connection to an OpenLineage backend, in the `lineage.stackable.tech` API group), an `InlineConnectionOrReference` - wrapper with `resolve()`, and an embeddable `OpenLineageJob` type (with a `jobName` field) for + wrapper with `resolve()`, and an embeddable `OpenLineageConfig` type (with a `jobName` field) for operators ([#1250]). [#1250]: https://github.com/stackabletech/operator-rs/pull/1250 diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 56dcaa2fa..15a254bd1 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -70,15 +70,14 @@ pub mod versioned { /// Embed this in an operator's workload spec to enable OpenLineage for that workload. #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] - pub struct OpenLineageJob { + pub struct OpenLineageConfig { /// The OpenLineage backend connection, either inlined or referencing an /// `OpenLineageConnection` resource by name. pub connection: InlineConnectionOrReference, /// The OpenLineage namespace lineage is reported under. - /// If unset, operators typically default to the workload's Kubernetes namespace. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub namespace: Option, + #[serde(default = "v1alpha1::OpenLineageConfig::default_namespace")] + pub namespace: String, /// A stable OpenLineage job name. Setting this prevents fragmented run history. /// If unset, operators resolve a name from workload-specific configuration. diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index f76173085..5656b6140 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -4,7 +4,10 @@ use crate::{ client::Client, crd::openlineage::{ ResolvedOpenLineageConnection, - v1alpha1::{InlineConnectionOrReference, OpenLineageConnection, OpenLineageConnectionSpec}, + v1alpha1::{ + InlineConnectionOrReference, OpenLineageConfig, OpenLineageConnection, + OpenLineageConnectionSpec, + }, }, }; @@ -38,6 +41,15 @@ impl OpenLineageConnectionSpec { } } +impl OpenLineageConfig { + /// Having it as `const &str` as well, so we don't always allocate a [`String`] just for comparisons + pub const DEFAULT_NAMESPACE: &str = "default"; + + pub(super) fn default_namespace() -> String { + Self::DEFAULT_NAMESPACE.to_string() + } +} + impl InlineConnectionOrReference { pub async fn resolve( self, From b97aa3c42bc5cb2305696bde34823fc5efd7accc Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:33:06 +0200 Subject: [PATCH 09/17] add OpenLineageTransport enum with a http variant to be prepared for future sinks (like kafka) --- crates/stackable-operator/CHANGELOG.md | 3 + .../crds/OpenLineageConnection.yaml | 117 ++++++++++-------- .../src/crd/openlineage/mod.rs | 70 +++++++---- .../src/crd/openlineage/v1alpha1_impl.rs | 7 +- 4 files changed, 114 insertions(+), 83 deletions(-) diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index 4e1043548..15fdf972f 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -10,6 +10,9 @@ All notable changes to this project will be documented in this file. OpenLineage backend, in the `lineage.stackable.tech` API group), an `InlineConnectionOrReference` wrapper with `resolve()`, and an embeddable `OpenLineageConfig` type (with a `jobName` field) for operators ([#1250]). + The connection spec selects one of the `OpenLineageTransport` variants, mirroring the transport + types of the OpenLineage client libraries. Currently only `http` (HTTP(S), with optional TLS and + API key) is supported. [#1250]: https://github.com/stackabletech/operator-rs/pull/1250 diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index ce1b89792..9e5ce5acf 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -23,72 +23,79 @@ spec: description: |- OpenLineage connection definition as a resource. Learn more about [OpenLineage](https://openlineage.io/). + oneOf: + - required: + - http 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: 'Host of the OpenLineage backend without any protocol or port. For example: `marquez`.' - 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 + http: + description: Publish events over HTTP(S) to an OpenLineage backend such as Marquez. properties: - verification: - description: The verification method used to verify the certificates of the server and/or the client. - oneOf: - - required: - - none - - required: - - server + 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: 'Host of the OpenLineage backend without any protocol or port. For example: `marquez`.' + 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: - none: - description: Use TLS but don't verify certificates. - type: object - server: - description: Use TLS and a CA certificate to verify the server. + verification: + description: The verification method used to verify the certificates of the server and/or the client. + oneOf: + - required: + - none + - required: + - server properties: - caCert: - description: CA cert to verify the server. - oneOf: - - required: - - webPki - - required: - - secretClass + none: + description: Use TLS but don't verify certificates. + type: object + server: + description: Use TLS and a CA certificate to verify the server. 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. + 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 - required: - - caCert type: object + required: + - verification type: object required: - - verification + - host + - port type: object - required: - - host - - port type: object required: - spec diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 15a254bd1..db5678673 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -36,6 +36,24 @@ pub mod versioned { #[derive(CustomResource, Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct OpenLineageConnectionSpec { + /// The transport used to publish lineage events. + #[serde(flatten)] + pub transport: OpenLineageTransport, + } + + /// The transport used to publish OpenLineage events. Mirrors the transport types of the + /// OpenLineage client libraries. Exactly one transport must be specified. + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub enum OpenLineageTransport { + /// Publish events over HTTP(S) to an OpenLineage backend such as Marquez. + Http(HttpTransport), + } + + /// Publish events over HTTP(S) to an OpenLineage backend such as Marquez. + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct HttpTransport { /// Host of the OpenLineage backend without any protocol or port. For example: `marquez`. pub host: String, @@ -90,23 +108,27 @@ pub mod versioned { impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineageConnectionSpec { fn roundtrip_test_data() -> Vec { crate::utils::yaml_from_str_singleton_map(indoc::indoc! {" - - host: marquez - port: 5000 - - host: marquez - port: 5000 - tls: - verification: - none: {} - - host: marquez - port: 5000 - tls: - verification: - server: - caCert: - secretClass: openlineage-cert - - host: marquez - port: 5000 - credentialsSecretName: openlineage-credentials + - http: + host: marquez + port: 5000 + - http: + host: marquez + port: 5000 + tls: + verification: + none: {} + - http: + host: marquez + port: 5000 + tls: + verification: + server: + caCert: + secretClass: openlineage-cert + - http: + host: marquez + port: 5000 + credentialsSecretName: openlineage-credentials "}) .expect("Failed to parse OpenLineageConnectionSpec YAML") } @@ -118,24 +140,24 @@ mod tests { commons::tls_verification::{ CaCert, Tls, TlsClientDetails, TlsServerVerification, TlsVerification, }, - crd::openlineage::v1alpha1::OpenLineageConnectionSpec, + crd::openlineage::v1alpha1::HttpTransport, }; #[test] fn http_transport_url_without_tls() { - let connection = OpenLineageConnectionSpec { + let transport = HttpTransport { host: "marquez".to_string(), port: 5000, tls: TlsClientDetails { tls: None }, credentials_secret_name: None, }; - assert_eq!(connection.transport_url(), "http://marquez:5000"); + assert_eq!(transport.transport_url(), "http://marquez:5000"); } #[test] fn https_transport_url_with_server_verification() { - let connection = OpenLineageConnectionSpec { + let transport = HttpTransport { host: "marquez".to_string(), port: 5000, tls: TlsClientDetails { @@ -148,12 +170,12 @@ mod tests { credentials_secret_name: None, }; - assert_eq!(connection.transport_url(), "https://marquez:5000"); + assert_eq!(transport.transport_url(), "https://marquez:5000"); } #[test] fn http_transport_url_without_verification() { - let connection = OpenLineageConnectionSpec { + let transport = HttpTransport { host: "marquez".to_string(), port: 5000, tls: TlsClientDetails { @@ -164,6 +186,6 @@ mod tests { credentials_secret_name: None, }; - assert_eq!(connection.transport_url(), "http://marquez:5000"); + assert_eq!(transport.transport_url(), "http://marquez:5000"); } } diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index 5656b6140..19f2ba4b9 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -5,8 +5,7 @@ use crate::{ crd::openlineage::{ ResolvedOpenLineageConnection, v1alpha1::{ - InlineConnectionOrReference, OpenLineageConfig, OpenLineageConnection, - OpenLineageConnectionSpec, + HttpTransport, InlineConnectionOrReference, OpenLineageConfig, OpenLineageConnection, }, }, }; @@ -21,8 +20,8 @@ pub enum OpenLineageError { }, } -impl OpenLineageConnectionSpec { - /// Build the OpenLineage transport URL from this connection. +impl HttpTransport { + /// Build the OpenLineage transport URL from this transport. /// /// The scheme is `https` when TLS server verification is configured /// (`tls.verification.server`), otherwise `http`. From b7ea4d4d6a02f7db701a8483734d45a52dbdcb37 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:43:58 +0200 Subject: [PATCH 10/17] add HttpTransport path field --- .../crds/OpenLineageConnection.yaml | 4 ++++ crates/stackable-operator/src/crd/openlineage/mod.rs | 11 +++++++++++ .../src/crd/openlineage/v1alpha1_impl.rs | 7 +++++++ 3 files changed, 22 insertions(+) diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index 9e5ce5acf..f6180145b 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -41,6 +41,10 @@ spec: host: description: 'Host 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 diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index db5678673..288bacbf8 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -60,6 +60,10 @@ pub mod versioned { /// Port the OpenLineage backend listens on. For example: `5000`. pub port: u16, + /// URL path of the endpoint lineage events are sent to. + #[serde(default = "v1alpha1::HttpTransport::default_path")] + pub path: String, + /// Use a TLS connection. If not specified no TLS will be used. /// When TLS server verification is configured, the transport uses `https` instead of `http`. #[serde(flatten)] @@ -129,6 +133,10 @@ impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineag host: marquez port: 5000 credentialsSecretName: openlineage-credentials + - http: + host: marquez + port: 5000 + path: /custom/lineage/endpoint "}) .expect("Failed to parse OpenLineageConnectionSpec YAML") } @@ -148,6 +156,7 @@ mod tests { let transport = HttpTransport { host: "marquez".to_string(), port: 5000, + path: HttpTransport::default_path(), tls: TlsClientDetails { tls: None }, credentials_secret_name: None, }; @@ -160,6 +169,7 @@ mod tests { let transport = HttpTransport { host: "marquez".to_string(), port: 5000, + path: HttpTransport::default_path(), tls: TlsClientDetails { tls: Some(Tls { verification: TlsVerification::Server(TlsServerVerification { @@ -178,6 +188,7 @@ mod tests { let transport = HttpTransport { host: "marquez".to_string(), port: 5000, + path: HttpTransport::default_path(), tls: TlsClientDetails { tls: Some(Tls { verification: TlsVerification::None {}, diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index 19f2ba4b9..9ee45559e 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -21,6 +21,13 @@ pub enum OpenLineageError { } impl HttpTransport { + /// Having it as `const &str` as well, so we don't always allocate a [`String`] just for comparisons + pub const DEFAULT_PATH: &str = "/api/v1/lineage"; + + pub(super) fn default_path() -> String { + Self::DEFAULT_PATH.to_string() + } + /// Build the OpenLineage transport URL from this transport. /// /// The scheme is `https` when TLS server verification is configured From a44b28402c7d708855d7fab541a1738deeca0b90 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:28:57 +0200 Subject: [PATCH 11/17] decision: remove job_name field --- crates/stackable-operator/crds/OpenLineageConnection.yaml | 6 +----- crates/stackable-operator/src/crd/openlineage/mod.rs | 5 ----- 2 files changed, 1 insertion(+), 10 deletions(-) diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index f6180145b..795b4dd57 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -6,15 +6,12 @@ metadata: spec: group: lineage.stackable.tech names: - categories: [] kind: OpenLineageConnection plural: openlineageconnections - shortNames: [] singular: openlineageconnection scope: Namespaced versions: - - additionalPrinterColumns: [] - name: v1alpha1 + - name: v1alpha1 schema: openAPIV3Schema: description: A reusable definition of a connection to an OpenLineage backend. @@ -107,4 +104,3 @@ spec: type: object served: true storage: true - subresources: {} diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 288bacbf8..9bbe0a17a 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -100,11 +100,6 @@ pub mod versioned { /// The OpenLineage namespace lineage is reported under. #[serde(default = "v1alpha1::OpenLineageConfig::default_namespace")] pub namespace: String, - - /// A stable OpenLineage job name. Setting this prevents fragmented run history. - /// If unset, operators resolve a name from workload-specific configuration. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub job_name: Option, } } From 55f9a1f233c6921bb2a1a47f0193333890f35519 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Thu, 1 Oct 2026 14:03:40 +0200 Subject: [PATCH 12/17] fix cargo deny lint --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 02573c11a..db3144c69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5215,9 +5215,9 @@ dependencies = [ [[package]] name = "yoke-derive" -version = "0.8.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33811428bee40dbceb6d545e95754741d17a6aef9a4849f0fd62e2ba4f412a78" +checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71" dependencies = [ "proc-macro2", "quote", From c784a7d2705d8cea130e7158e9bb857f2cacc542 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:22:30 +0200 Subject: [PATCH 13/17] review feedback: use HostName instead of String --- .../crds/OpenLineageConnection.yaml | 2 +- .../src/crd/openlineage/mod.rs | 22 ++++++++++++++----- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml index 795b4dd57..e2bbe1113 100644 --- a/crates/stackable-operator/crds/OpenLineageConnection.yaml +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -36,7 +36,7 @@ spec: nullable: true type: string host: - description: 'Host of the OpenLineage backend without any protocol or port. For example: `marquez`.' + description: 'Hostname or IP address of the OpenLineage backend without any protocol or port. For example: `marquez`.' type: string path: default: /api/v1/lineage diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 9bbe0a17a..2eb195f61 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -2,7 +2,10 @@ use kube::CustomResource; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use crate::{commons::tls_verification::TlsClientDetails, versioned::versioned}; +use crate::{ + commons::{networking::HostName, tls_verification::TlsClientDetails}, + versioned::versioned, +}; mod v1alpha1_impl; @@ -20,6 +23,7 @@ pub type ResolvedOpenLineageConnection = v1alpha1::OpenLineageConnectionSpec; ) )] pub mod versioned { + pub mod v1alpha1 { pub use v1alpha1_impl::OpenLineageError; } @@ -54,8 +58,8 @@ pub mod versioned { #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct HttpTransport { - /// Host of the OpenLineage backend without any protocol or port. For example: `marquez`. - pub host: String, + /// Hostname or IP address of the OpenLineage backend without any protocol or port. For example: `marquez`. + pub host: HostName, /// Port the OpenLineage backend listens on. For example: `5000`. pub port: u16, @@ -149,7 +153,9 @@ mod tests { #[test] fn http_transport_url_without_tls() { let transport = HttpTransport { - host: "marquez".to_string(), + host: "marquez" + .parse() + .expect("cannot parse [marquez] as host name"), port: 5000, path: HttpTransport::default_path(), tls: TlsClientDetails { tls: None }, @@ -162,7 +168,9 @@ mod tests { #[test] fn https_transport_url_with_server_verification() { let transport = HttpTransport { - host: "marquez".to_string(), + host: "marquez" + .parse() + .expect("cannot parse [marquez] as host name"), port: 5000, path: HttpTransport::default_path(), tls: TlsClientDetails { @@ -181,7 +189,9 @@ mod tests { #[test] fn http_transport_url_without_verification() { let transport = HttpTransport { - host: "marquez".to_string(), + host: "marquez" + .parse() + .expect("cannot parse [marquez] as host name"), port: 5000, path: HttpTransport::default_path(), tls: TlsClientDetails { From 1cdbb857e229fd87d64b070ed20f7e3a314f6d77 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 5 Oct 2026 15:04:51 +0200 Subject: [PATCH 14/17] review feedback: prefer url::Url to String --- .../src/crd/openlineage/mod.rs | 40 +++++++++++++++++-- .../src/crd/openlineage/v1alpha1_impl.rs | 26 ++++++++++-- 2 files changed, 59 insertions(+), 7 deletions(-) diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 2eb195f61..8ff605cba 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -162,7 +162,10 @@ mod tests { credentials_secret_name: None, }; - assert_eq!(transport.transport_url(), "http://marquez:5000"); + assert_eq!( + transport.url().expect("valid http transport url").as_str(), + "http://marquez:5000/" + ); } #[test] @@ -183,7 +186,10 @@ mod tests { credentials_secret_name: None, }; - assert_eq!(transport.transport_url(), "https://marquez:5000"); + assert_eq!( + transport.url().expect("valid https transport url").as_str(), + "https://marquez:5000/" + ); } #[test] @@ -202,6 +208,34 @@ mod tests { credentials_secret_name: None, }; - assert_eq!(transport.transport_url(), "http://marquez:5000"); + assert_eq!( + transport.url().expect("valid https transport url").as_str(), + "http://marquez:5000/" + ); + } + + #[test] + fn http_transport_url_with_path() { + let transport = HttpTransport { + host: "marquez" + .parse() + .expect("cannot parse [marquez] as host name"), + port: 5000, + path: HttpTransport::default_path(), + tls: TlsClientDetails { + tls: Some(Tls { + verification: TlsVerification::None {}, + }), + }, + credentials_secret_name: None, + }; + + assert_eq!( + transport + .url_with_path() + .expect("valid https transport url") + .as_str(), + "http://marquez:5000/api/v1/lineage" + ); } } diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index 9ee45559e..57410e148 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -12,6 +12,11 @@ use crate::{ #[derive(Debug, Snafu)] pub enum OpenLineageError { + #[snafu(display("failed to build OpenLineage URL from endpoint '{endpoint}'"))] + HttpTransportUrl { + source: url::ParseError, + endpoint: String, + }, #[snafu(display("failed to retrieve OpenLineage connection '{open_lineage_connection}'"))] RetrieveOpenLineageConnection { #[snafu(source(from(crate::client::Error, Box::new)))] @@ -28,22 +33,35 @@ impl HttpTransport { Self::DEFAULT_PATH.to_string() } - /// Build the OpenLineage transport URL from this transport. + /// Build the OpenLineage transport URL (without path) from this transport. /// /// The scheme is `https` when TLS server verification is configured /// (`tls.verification.server`), otherwise `http`. - pub fn transport_url(&self) -> String { + pub fn url(&self) -> Result { let scheme = if self.tls.uses_tls_verification() { "https" } else { "http" }; - format!( + let endpoint = format!( "{scheme}://{host}:{port}", host = self.host, port = self.port - ) + ); + + url::Url::parse(&endpoint).context(HttpTransportUrlSnafu { endpoint }) + } + + /// Build the OpenLineage transport URL (with path) from this transport. + pub fn url_with_path(&self) -> Result { + match self.url() { + Ok(mut target) => { + target.set_path(&self.path); + Ok(target) + } + Err(err) => Err(err), + } } } From 16e0b88cce11c614bac9bbbf3e1b1e928f1595cb Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 5 Oct 2026 15:37:09 +0200 Subject: [PATCH 15/17] review feedback: derive http scheme from tls flag (not tls verification) --- .../src/crd/openlineage/mod.rs | 26 ++----------------- .../src/crd/openlineage/v1alpha1_impl.rs | 9 +------ 2 files changed, 3 insertions(+), 32 deletions(-) diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 8ff605cba..1e7a37b0a 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -169,7 +169,7 @@ mod tests { } #[test] - fn https_transport_url_with_server_verification() { + fn https_transport_url_with_tls() { let transport = HttpTransport { host: "marquez" .parse() @@ -192,28 +192,6 @@ mod tests { ); } - #[test] - fn http_transport_url_without_verification() { - let transport = HttpTransport { - host: "marquez" - .parse() - .expect("cannot parse [marquez] as host name"), - port: 5000, - path: HttpTransport::default_path(), - tls: TlsClientDetails { - tls: Some(Tls { - verification: TlsVerification::None {}, - }), - }, - credentials_secret_name: None, - }; - - assert_eq!( - transport.url().expect("valid https transport url").as_str(), - "http://marquez:5000/" - ); - } - #[test] fn http_transport_url_with_path() { let transport = HttpTransport { @@ -235,7 +213,7 @@ mod tests { .url_with_path() .expect("valid https transport url") .as_str(), - "http://marquez:5000/api/v1/lineage" + "https://marquez:5000/api/v1/lineage" ); } } diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index 57410e148..d0d0949a4 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -34,15 +34,8 @@ impl HttpTransport { } /// Build the OpenLineage transport URL (without path) from this transport. - /// - /// The scheme is `https` when TLS server verification is configured - /// (`tls.verification.server`), otherwise `http`. pub fn url(&self) -> Result { - let scheme = if self.tls.uses_tls_verification() { - "https" - } else { - "http" - }; + let scheme = if self.tls.uses_tls() { "https" } else { "http" }; let endpoint = format!( "{scheme}://{host}:{port}", From e300c183c1e008b92b4171d42f2cc9ba8f69d97d Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Mon, 5 Oct 2026 15:50:59 +0200 Subject: [PATCH 16/17] review feedback: rename url functions --- .../src/crd/openlineage/mod.rs | 21 +++++++++++-------- .../src/crd/openlineage/v1alpha1_impl.rs | 6 +++--- 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 1e7a37b0a..0ecc333af 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -151,7 +151,7 @@ mod tests { }; #[test] - fn http_transport_url_without_tls() { + fn http_transport_url_without_path_without_tls() { let transport = HttpTransport { host: "marquez" .parse() @@ -163,13 +163,16 @@ mod tests { }; assert_eq!( - transport.url().expect("valid http transport url").as_str(), + transport + .url_without_path() + .expect("valid http transport url") + .as_str(), "http://marquez:5000/" ); } #[test] - fn https_transport_url_with_tls() { + fn https_transport_url_without_path_with_tls() { let transport = HttpTransport { host: "marquez" .parse() @@ -187,13 +190,16 @@ mod tests { }; assert_eq!( - transport.url().expect("valid https transport url").as_str(), + transport + .url_without_path() + .expect("valid https transport url") + .as_str(), "https://marquez:5000/" ); } #[test] - fn http_transport_url_with_path() { + fn http_transport_url() { let transport = HttpTransport { host: "marquez" .parse() @@ -209,10 +215,7 @@ mod tests { }; assert_eq!( - transport - .url_with_path() - .expect("valid https transport url") - .as_str(), + transport.url().expect("valid https transport url").as_str(), "https://marquez:5000/api/v1/lineage" ); } diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index d0d0949a4..3f1c07c03 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -34,7 +34,7 @@ impl HttpTransport { } /// Build the OpenLineage transport URL (without path) from this transport. - pub fn url(&self) -> Result { + pub fn url_without_path(&self) -> Result { let scheme = if self.tls.uses_tls() { "https" } else { "http" }; let endpoint = format!( @@ -47,8 +47,8 @@ impl HttpTransport { } /// Build the OpenLineage transport URL (with path) from this transport. - pub fn url_with_path(&self) -> Result { - match self.url() { + pub fn url(&self) -> Result { + match self.url_without_path() { Ok(mut target) => { target.set_path(&self.path); Ok(target) From 40cb063f37d830057b55de8390f964e77d091be7 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Tue, 6 Oct 2026 10:13:45 +0200 Subject: [PATCH 17/17] review feedback: transport only has url(). Clients need to deal with it. --- .../src/crd/openlineage/mod.rs | 38 ++++++------------- .../src/crd/openlineage/v1alpha1_impl.rs | 23 +++-------- 2 files changed, 18 insertions(+), 43 deletions(-) diff --git a/crates/stackable-operator/src/crd/openlineage/mod.rs b/crates/stackable-operator/src/crd/openlineage/mod.rs index 0ecc333af..1ab686cae 100644 --- a/crates/stackable-operator/src/crd/openlineage/mod.rs +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -144,14 +144,12 @@ impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineag #[cfg(test)] mod tests { use crate::{ - commons::tls_verification::{ - CaCert, Tls, TlsClientDetails, TlsServerVerification, TlsVerification, - }, + commons::tls_verification::{Tls, TlsClientDetails, TlsVerification}, crd::openlineage::v1alpha1::HttpTransport, }; #[test] - fn http_transport_url_without_path_without_tls() { + fn http_transport_url_without_tls() { let transport = HttpTransport { host: "marquez" .parse() @@ -163,16 +161,13 @@ mod tests { }; assert_eq!( - transport - .url_without_path() - .expect("valid http transport url") - .as_str(), - "http://marquez:5000/" + transport.url().expect("valid http transport url").as_str(), + "http://marquez:5000/api/v1/lineage" ); } #[test] - fn https_transport_url_without_path_with_tls() { + fn https_transport_url_with_tls() { let transport = HttpTransport { host: "marquez" .parse() @@ -181,42 +176,33 @@ mod tests { path: HttpTransport::default_path(), tls: TlsClientDetails { tls: Some(Tls { - verification: TlsVerification::Server(TlsServerVerification { - ca_cert: CaCert::WebPki {}, - }), + verification: TlsVerification::None {}, }), }, credentials_secret_name: None, }; assert_eq!( - transport - .url_without_path() - .expect("valid https transport url") - .as_str(), - "https://marquez:5000/" + transport.url().expect("valid https transport url").as_str(), + "https://marquez:5000/api/v1/lineage" ); } #[test] - fn http_transport_url() { + fn http_transport_with_custom_path() { let transport = HttpTransport { host: "marquez" .parse() .expect("cannot parse [marquez] as host name"), port: 5000, - path: HttpTransport::default_path(), - tls: TlsClientDetails { - tls: Some(Tls { - verification: TlsVerification::None {}, - }), - }, + path: "/custom/api/v1/lineage".to_string(), + tls: TlsClientDetails { tls: None }, credentials_secret_name: None, }; assert_eq!( transport.url().expect("valid https transport url").as_str(), - "https://marquez:5000/api/v1/lineage" + "http://marquez:5000/custom/api/v1/lineage" ); } } diff --git a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs index 3f1c07c03..3b139ed49 100644 --- a/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs +++ b/crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs @@ -33,29 +33,18 @@ impl HttpTransport { Self::DEFAULT_PATH.to_string() } - /// Build the OpenLineage transport URL (without path) from this transport. - pub fn url_without_path(&self) -> Result { - let scheme = if self.tls.uses_tls() { "https" } else { "http" }; - + /// Build the OpenLineage transport URL (including path) from this transport. + pub fn url(&self) -> Result { let endpoint = format!( - "{scheme}://{host}:{port}", + "{scheme}://{host}:{port}/{path}", + scheme = if self.tls.uses_tls() { "https" } else { "http" }, host = self.host, - port = self.port + port = self.port, + path = self.path.strip_prefix("/").unwrap_or(&self.path), ); url::Url::parse(&endpoint).context(HttpTransportUrlSnafu { endpoint }) } - - /// Build the OpenLineage transport URL (with path) from this transport. - pub fn url(&self) -> Result { - match self.url_without_path() { - Ok(mut target) => { - target.set_path(&self.path); - Ok(target) - } - Err(err) => Err(err), - } - } } impl OpenLineageConfig {