Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 8 additions & 0 deletions crates/stackable-operator/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
91 changes: 91 additions & 0 deletions crates/stackable-operator/crds/OpenLineageConnection.yaml
Original file line number Diff line number Diff line change
@@ -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: {}
1 change: 1 addition & 0 deletions crates/stackable-operator/src/crd/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down
157 changes: 157 additions & 0 deletions crates/stackable-operator/src/crd/openlineage/mod.rs
Original file line number Diff line number Diff line change
@@ -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");
}
}
62 changes: 62 additions & 0 deletions crates/stackable-operator/src/crd/openlineage/v1alpha1_impl.rs
Original file line number Diff line number Diff line change
@@ -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)
}
}
}
}
2 changes: 2 additions & 0 deletions crates/xtask/src/crd/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use stackable_operator::{
Listener, ListenerClass, ListenerClassVersion, ListenerVersion, PodListeners,
PodListenersVersion,
},
openlineage::{OpenLineageConnection, OpenLineageConnectionVersion},
s3::{S3Bucket, S3BucketVersion, S3Connection, S3ConnectionVersion},
scaler::{Scaler, ScalerVersion},
},
Expand Down Expand Up @@ -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);
Expand Down