feat: add crd::openlineage module

Author: razvan

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

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: {}

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;
 

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<String>,
+
+        /// 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<String>,
+    }
+}
+
+#[cfg(test)]
+impl stackable_versioned::test_utils::RoundtripTestData for v1alpha1::OpenLineageConnectionSpec {
+    fn roundtrip_test_data() -> Vec<Self> {
+        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");
+    }
+}

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<crate::client::Error>,
+        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<ResolvedOpenLineageConnection, OpenLineageError> {
+        match self {
+            Self::Inline(inline) => Ok(inline),
+            Self::Reference(reference) => {
+                let connection_spec = client
+                    .get::<OpenLineageConnection>(&reference, namespace)
+                    .await
+                    .context(RetrieveOpenLineageConnectionSnafu {
+                        open_lineage_connection: reference,
+                    })?
+                    .spec;
+
+                Ok(connection_spec)
+            }
+        }
+    }
+}

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);