diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index 47e5df715..e208cf326 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -7,13 +7,17 @@ All notable changes to this project will be documented in this file. ### Added - [v2] Add `EnvVarSet::with_env_var` to add a given `EnvVar` to the set ([#1249]). - +- 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 environment variables into account ([#1249]). [#1249]: https://github.com/stackabletech/operator-rs/pull/1249 +[#1250]: https://github.com/stackabletech/operator-rs/pull/1250 ## [0.113.4] - 2026-07-09 diff --git a/crates/stackable-operator/crds/OpenLineageConnection.yaml b/crates/stackable-operator/crds/OpenLineageConnection.yaml new file mode 100644 index 000000000..14439f431 --- /dev/null +++ b/crates/stackable-operator/crds/OpenLineageConnection.yaml @@ -0,0 +1,99 @@ +--- +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: + 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: + 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..1dfb5442e --- /dev/null +++ b/crates/stackable-operator/src/crd/openlineage/mod.rs @@ -0,0 +1,170 @@ +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, + + /// 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 credentials_secret_name: Option, + } + + /// 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 + - host: marquez + port: 5000 + credentialsSecretName: openlineage-credentials + "}) + .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 }, + credentials_secret_name: 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 {}, + }), + }), + }, + credentials_secret_name: None, + }; + + 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 {}, + }), + }, + 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 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);