diff --git a/CHANGELOG.md b/CHANGELOG.md index 2177753fb..c0111b371 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,12 +4,17 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- Add `spec.clusterConfig.lineage` to emit [OpenLineage](https://openlineage.io/) lineage events ([#914]). + ### Changed - Internal operator refactoring: introduce a build() step in the reconciler that assembles all relevant Kubernetes resources before anything is applied ([#909]). [#909]: https://github.com/stackabletech/trino-operator/pull/909 +[#914]: https://github.com/stackabletech/trino-operator/pull/914 ## [26.7.0] - 2026-07-21 diff --git a/Cargo.lock b/Cargo.lock index 93dc350f6..b05f9b076 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1607,7 +1607,7 @@ dependencies = [ [[package]] name = "k8s-version" version = "0.1.3" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "darling", "regex", @@ -2992,9 +2992,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" @@ -3015,7 +3015,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" [[package]] name = "stackable-certs" version = "0.4.1" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "const-oid", "ecdsa", @@ -3038,8 +3038,8 @@ dependencies = [ [[package]] name = "stackable-operator" -version = "0.113.3" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +version = "0.113.4" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "base64", "clap", @@ -3084,7 +3084,7 @@ dependencies = [ [[package]] name = "stackable-operator-derive" version = "0.3.1" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "darling", "proc-macro2", @@ -3095,7 +3095,7 @@ dependencies = [ [[package]] name = "stackable-shared" version = "0.1.2" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "jiff", "k8s-openapi", @@ -3112,7 +3112,7 @@ dependencies = [ [[package]] name = "stackable-telemetry" version = "0.6.5" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "axum", "clap", @@ -3158,7 +3158,7 @@ dependencies = [ [[package]] name = "stackable-versioned" version = "0.11.1" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "kube", "schemars", @@ -3172,7 +3172,7 @@ dependencies = [ [[package]] name = "stackable-versioned-macros" version = "0.11.1" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "convert_case", "convert_case_extras", @@ -3190,7 +3190,7 @@ dependencies = [ [[package]] name = "stackable-webhook" version = "0.9.2" -source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#013bbf43f7006a4ddfc08a147f68441ed88b462b" +source = "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#01535bc1acb05e8a28f098d40bf5bef3105e075e" dependencies = [ "arc-swap", "async-trait", diff --git a/Cargo.nix b/Cargo.nix index 5a7818be9..364455cec 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -5107,9 +5107,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "k8s_version"; authors = [ @@ -9817,9 +9817,9 @@ rec { }; "spin" = rec { crateName = "spin"; - version = "0.9.8"; + version = "0.9.9"; edition = "2015"; - sha256 = "0rvam5r0p3a6qhc18scqpvpgb3ckzyqxpgdfyjnghh8ja7byi039"; + sha256 = "03psal0vh1xdxp7agphw09p7kf50v3bj1zshijq1s5bkdd7jcqrp"; authors = [ "Mathijs van de Nes " "John Ericson " @@ -9891,9 +9891,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "stackable_certs"; authors = [ @@ -9990,13 +9990,13 @@ rec { }; "stackable-operator" = rec { crateName = "stackable-operator"; - version = "0.113.3"; + version = "0.113.4"; edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "stackable_operator"; authors = [ @@ -10193,9 +10193,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; procMacro = true; libName = "stackable_operator_derive"; @@ -10228,9 +10228,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "stackable_shared"; authors = [ @@ -10309,9 +10309,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "stackable_telemetry"; authors = [ @@ -10518,9 +10518,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "stackable_versioned"; authors = [ @@ -10568,9 +10568,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; procMacro = true; libName = "stackable_versioned_macros"; @@ -10636,9 +10636,9 @@ rec { edition = "2024"; workspace_member = null; src = pkgs.fetchgit { - url = "https://github.com/stackabletech/operator-rs.git"; - rev = "013bbf43f7006a4ddfc08a147f68441ed88b462b"; - sha256 = "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr"; + url = "https://github.com/stackabletech//operator-rs.git"; + rev = "01535bc1acb05e8a28f098d40bf5bef3105e075e"; + sha256 = "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2"; }; libName = "stackable_webhook"; authors = [ diff --git a/Cargo.toml b/Cargo.toml index dbac40e92..9a28055d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,5 +29,9 @@ tokio = { version = "1.52", features = ["full"] } tracing = "0.1" [patch."https://github.com/stackabletech/operator-rs.git"] +# TODO: Switch back to a release tag once operator-rs#1250 (the `crd::openlineage` +# module) has merged and a new stackable-operator release has been published. +# See . +stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "feat/openlineage-crd" } # stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "smooth-operator"} # stackable-operator = { path = "../operator-rs/crates/stackable-operator" } diff --git a/crate-hashes.json b/crate-hashes.json index 77adf52af..2eab374aa 100644 --- a/crate-hashes.json +++ b/crate-hashes.json @@ -1,12 +1,12 @@ { - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#k8s-version@0.1.3": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-certs@0.4.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator-derive@0.3.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-operator@0.113.3": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-shared@0.1.2": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-telemetry@0.6.5": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned-macros@0.11.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-versioned@0.11.1": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", - "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.113.3#stackable-webhook@0.9.2": "1ab5qmawljwbll6b44gw3md57s9kjiy5p16akz6rv52j6fdxjpgr", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#k8s-version@0.1.3": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-certs@0.4.1": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-operator-derive@0.3.1": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-operator@0.113.4": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-shared@0.1.2": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-telemetry@0.6.5": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-versioned-macros@0.11.1": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-versioned@0.11.1": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", + "git+https://github.com/stackabletech//operator-rs.git?branch=feat%2Fopenlineage-crd#stackable-webhook@0.9.2": "1irgyw99cp8vyw14l414aargkdvh5s43vxxldxlf3jr4q7fxv8h2", "git+https://github.com/stackabletech/product-config.git?tag=0.8.0#product-config@0.8.0": "1dz70kapm2wdqcr7ndyjji0lhsl98bsq95gnb2lw487wf6yr7987" } \ No newline at end of file diff --git a/deploy/helm/trino-operator/templates/clusterrole-operator.yaml b/deploy/helm/trino-operator/templates/clusterrole-operator.yaml index 6e14e6969..677f9962c 100644 --- a/deploy/helm/trino-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/trino-operator/templates/clusterrole-operator.yaml @@ -176,3 +176,12 @@ rules: - get - list - watch + # Read OpenLineage connection configuration referenced in the TrinoCluster spec. + - apiGroups: + - lineage.stackable.tech + resources: + - openlineageconnections + verbs: + - get + - list + - watch diff --git a/docs/modules/trino/pages/usage-guide/openlineage.adoc b/docs/modules/trino/pages/usage-guide/openlineage.adoc new file mode 100644 index 000000000..1f5fcee0f --- /dev/null +++ b/docs/modules/trino/pages/usage-guide/openlineage.adoc @@ -0,0 +1,121 @@ += OpenLineage +:description: Emit OpenLineage data lineage events from Trino queries to an OpenLineage-compatible backend such as Marquez. +:keywords: OpenLineage, data lineage, event listener, Marquez, Trino + +https://openlineage.io/[OpenLineage] is an open standard for data lineage collection. +Trino ships an https://trino.io/docs/current/admin/event-listeners-openlineage.html[OpenLineage event listener] as a core plugin, which emits a lineage event for every executed query. +The Trino operator can configure this listener automatically, so lineage from all queries run on the cluster is sent to an OpenLineage-compatible backend such as https://marquezproject.github.io/marquez/[Marquez], without any manual configuration. + +When configured, the operator: + +* writes an `event-listener.properties` file on the *coordinator* selecting the OpenLineage plugin and its HTTP transport, +* points the transport at the backend endpoint (`://:`), +* records this cluster's `trino.uri` and a lineage namespace, and +* (optionally) imports a backend CA certificate into the coordinator trust store and injects a bearer token for authenticated backends. + +The event listener runs on the coordinator only; workers are not affected. + +NOTE: The Trino OpenLineage plugin only emits events for data-modifying query types (`INSERT`, `DELETE`, `MERGE`, `UPDATE`, `ALTER_TABLE_EXECUTE` and `DATA_DEFINITION`) by default; plain `SELECT` queries are filtered out. Adjust `openlineage-event-listener.trino.include-query-types` via config overrides (see below) to change this. + +== Enabling OpenLineage + +OpenLineage is configured through `clusterConfig.lineage.connection`, which either inlines an OpenLineage connection or references a standalone `OpenLineageConnection` resource by name: + +[source,yaml] +---- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCluster +metadata: + name: trino +spec: + clusterConfig: + lineage: # <1> + connection: # <2> + inline: + host: marquez + port: 5000 + # tls: + # verification: + # server: + # caCert: + # secretClass: marquez-ca + # namespace: trino-lineage # <3> + # jobName: "$QUERY_ID" # <4> + ... +---- +<1> Adding the `lineage` block enables OpenLineage event emission. Omit it entirely to leave OpenLineage off (the default). +<2> The OpenLineage backend connection. Use `inline` (shown here) to define the connection directly, or `reference: ` to point at an `OpenLineageConnection` resource in the same namespace. Required. +<3> Optional. The OpenLineage namespace events are reported under. Defaults to the cluster's Kubernetes namespace. +<4> Optional. Sets the OpenLineage job name format (`openlineage-event-listener.job.name-format`). Accepts an arbitrary string with optional `$QUERY_ID`, `$USER`, `$SOURCE` and `$CLIENT_IP` substitution variables. When unset, Trino defaults to `$QUERY_ID`. See the link:https://trino.io/docs/current/admin/event-listeners-openlineage.html[Trino OpenLineage event listener documentation {external-link-icon}^]. + +The `OpenLineageConnection` resource (`lineage.stackable.tech`) is a reusable definition of a connection to an OpenLineage backend, provided by the Stackable operator library. +Reference it to share one backend definition across multiple clusters: + +[source,yaml] +---- +apiVersion: lineage.stackable.tech/v1alpha1 +kind: OpenLineageConnection +metadata: + name: marquez +spec: + host: marquez + port: 5000 +---- + +The transport scheme is `https` when `tls.verification.server` is configured, otherwise `http`. +When the connection verifies the server against a `secretClass` CA (as in the commented `tls` block above), the operator mounts that SecretClass certificate into the coordinator and adds it to the trust store, so the OpenLineage listener trusts the backend's certificate. + +IMPORTANT: With TLS server verification, the `host` must match a Subject Alternative Name in the backend's certificate. Use the exact hostname the certificate was issued for (for an in-cluster backend, typically its full service FQDN such as `marquez..svc.cluster.local`), otherwise the TLS handshake fails and no events are delivered. + +[#authentication] +== Authentication + +Backends that require authentication (for example DataHub, or a Marquez instance behind a gateway) expect a bearer token on each request. +Configure this by setting `credentialsSecretName` on the connection to the name of a Secret holding the token: + +[source,yaml] +---- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCluster +metadata: + name: trino +spec: + clusterConfig: + lineage: + connection: + inline: + host: marquez + port: 5000 + credentialsSecretName: openlineage-token # <1> + ... +--- +apiVersion: v1 +kind: Secret +metadata: + name: openlineage-token + namespace: default # <2> +stringData: + apiKey: "s3cr3t-bearer-token" # <3> +---- +<1> Name of the Secret holding the token. If omitted, no authentication is used. The Secret must exist in the same namespace as the `TrinoCluster`. +<2> The Secret must be in the cluster's namespace so it can be projected into the coordinator. +<3> The token must be stored under the fixed key `apiKey`. The token is mounted as a file into the coordinator and resolved into the event listener configuration at startup, so it never appears in the ConfigMap or the pod's environment. + +The operator translates this into the OpenLineage HTTP transport's `api-key` (bearer) authentication. +Combine it with the `tls` block above when the backend also needs TLS server verification. + +== Overriding the generated configuration + +The generated `event-listener.properties` can be extended or overridden per role group through xref:usage-guide/overrides.adoc[config overrides], using the `event-listener.properties` file name. +Overrides are merged last and therefore take precedence over the operator-generated values: + +[source,yaml] +---- +spec: + coordinators: + configOverrides: + event-listener.properties: + openlineage-event-listener.trino.uri: "https://trino.example.com:8443" +---- + +Event listeners run on the coordinator, so `event-listener.properties` overrides only take effect on the coordinator role. diff --git a/docs/modules/trino/partials/nav.adoc b/docs/modules/trino/partials/nav.adoc index d90607af9..35c804728 100644 --- a/docs/modules/trino/partials/nav.adoc +++ b/docs/modules/trino/partials/nav.adoc @@ -11,6 +11,7 @@ ** xref:trino:usage-guide/s3.adoc[] ** xref:trino:usage-guide/security.adoc[] ** xref:trino:usage-guide/monitoring.adoc[] +** xref:trino:usage-guide/openlineage.adoc[] ** xref:trino:usage-guide/log_aggregation.adoc[] ** xref:trino:usage-guide/OpenTelemetry.adoc[] ** xref:trino:usage-guide/query.adoc[] diff --git a/extra/crds.yaml b/extra/crds.yaml index 9ceb2b2a1..b27e1cdbc 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -891,6 +891,111 @@ spec: - exchangeManager type: object type: object + lineage: + description: |- + Emit [OpenLineage](https://openlineage.io/) lineage events for the queries run on this + Trino cluster. + nullable: true + properties: + connection: + description: |- + The OpenLineage backend connection, either inlined or referencing an + `OpenLineageConnection` resource by name. + oneOf: + - required: + - inline + - required: + - reference + properties: + inline: + description: |- + OpenLineage connection definition as a resource. + Learn more about [OpenLineage](https://openlineage.io/). + 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 + reference: + type: string + type: object + jobName: + description: |- + A stable OpenLineage job/application name. Setting this prevents fragmented run history. + If unset, operators resolve a name from workload-specific configuration. + nullable: true + type: string + namespace: + description: |- + The OpenLineage namespace lineage is reported under. + If unset, operators typically default to the workload's Kubernetes namespace. + nullable: true + type: string + required: + - connection + type: object tls: default: internalSecretClass: tls @@ -1450,6 +1555,12 @@ spec: default: {} description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. type: object + event-listener.properties: + additionalProperties: + type: string + default: {} + description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. + type: object exchange-manager.properties: additionalProperties: type: string @@ -2058,6 +2169,12 @@ spec: default: {} description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. type: object + event-listener.properties: + additionalProperties: + type: string + default: {} + description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. + type: object exchange-manager.properties: additionalProperties: type: string @@ -2730,6 +2847,12 @@ spec: default: {} description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. type: object + event-listener.properties: + additionalProperties: + type: string + default: {} + description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. + type: object exchange-manager.properties: additionalProperties: type: string @@ -3330,6 +3453,12 @@ spec: default: {} description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. type: object + event-listener.properties: + additionalProperties: + type: string + default: {} + description: Flat key-value overrides for `*.properties`, Hadoop XML, etc. + type: object exchange-manager.properties: additionalProperties: type: string diff --git a/rust/operator-binary/src/config/lineage.rs b/rust/operator-binary/src/config/lineage.rs new file mode 100644 index 000000000..9b23760ab --- /dev/null +++ b/rust/operator-binary/src/config/lineage.rs @@ -0,0 +1,257 @@ +//! Resolves `spec.clusterConfig.lineage` into everything the coordinator needs to run the +//! Trino [OpenLineage](https://openlineage.io/) event listener. +//! +//! Trino ships the OpenLineage event listener as a **core** plugin, so nothing needs to be added +//! to the image; the listener runs on the **coordinator only**. This module produces: +//! +//! * the connection-derived `event-listener.properties` settings (transport type/URL, namespace +//! and — when authentication is configured — the bearer-token reference), +//! * the Kubernetes [`Volume`]s/[`VolumeMount`]s for the backend CA certificate and the token +//! Secret, and +//! * the init-container commands that import a `SecretClass` CA into the client truststore. +//! +//! The reusable OpenLineage CRD types (`OpenLineageConnectionSpec`, `InlineConnectionOrReference`, +//! `OpenLineageJob`) live in [`stackable_operator::crd::openlineage`]; this module only holds the +//! Trino-specific wiring around them. + +use std::collections::BTreeMap; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + commons::tls_verification::{ + CaCert, TlsClientDetails, TlsClientDetailsError, TlsServerVerification, TlsVerification, + }, + crd::openlineage::{self, v1alpha1::OpenLineageJob}, + k8s_openapi::api::core::v1::{SecretVolumeSource, Volume, VolumeMount}, +}; + +use crate::{ + controller::build::command, + crd::{OPENLINEAGE_AUTH_SECRET_KEY, STACKABLE_CLIENT_TLS_DIR}, +}; + +/// Directory the OpenLineage bearer-token Secret is mounted at on the coordinator. Referenced from +/// the `api-key` property via Trino's `${file:...}` secret placeholder (resolved at startup by +/// `config-utils template`, see [`crate::controller::build::command::container_trino_args`]). +const STACKABLE_OPENLINEAGE_AUTH_DIR: &str = "/stackable/openlineage_auth"; + +/// Name of the token Secret volume mounted on the coordinator. +const OPENLINEAGE_AUTH_VOLUME_NAME: &str = "openlineage-auth"; + +// --- `event-listener.properties` keys, see the OpenLineage usage guide --- +// The Trino OpenLineage event listener uses the fixed `openlineage-event-listener.` property prefix +// regardless of the file name. `event-listener.name=openlineage` selects the plugin. + +/// Selects the event listener plugin. +pub const EVENT_LISTENER_NAME_KEY: &str = "event-listener.name"; +/// The value selecting the OpenLineage event listener plugin. +pub const EVENT_LISTENER_NAME_OPENLINEAGE: &str = "openlineage"; +/// OpenLineage transport type key (we always use the HTTP transport). +pub const OPENLINEAGE_TRANSPORT_TYPE_KEY: &str = "openlineage-event-listener.transport.type"; +/// The OpenLineage HTTP transport type value. +pub const OPENLINEAGE_TRANSPORT_TYPE_HTTP: &str = "HTTP"; +/// OpenLineage HTTP transport base URL key (scheme/host/port of the backend). +pub const OPENLINEAGE_TRANSPORT_URL_KEY: &str = "openlineage-event-listener.transport.url"; +/// OpenLineage HTTP transport bearer-token (API key) key. Only set when authentication is +/// configured; the value is a `${file:...}` reference so the token never enters the ConfigMap. +pub const OPENLINEAGE_TRANSPORT_API_KEY_KEY: &str = "openlineage-event-listener.transport.api-key"; +/// The OpenLineage namespace lineage is reported under. +pub const OPENLINEAGE_NAMESPACE_KEY: &str = "openlineage-event-listener.namespace"; +/// Format for the emitted OpenLineage job name. Set from `spec.clusterConfig.lineage.jobName`. +/// Accepts an arbitrary string with optional `$QUERY_ID`, `$USER`, `$SOURCE` and `$CLIENT_IP` +/// substitution variables (Trino defaults to `$QUERY_ID` when unset). +pub const OPENLINEAGE_JOB_NAME_FORMAT_KEY: &str = "openlineage-event-listener.job.name-format"; +/// The URI identifying this Trino cluster in emitted lineage. Computed from the coordinator service +/// at ConfigMap-build time (see [`crate::controller::build::properties::event_listener_properties`]). +pub const OPENLINEAGE_TRINO_URI_KEY: &str = "openlineage-event-listener.trino.uri"; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to resolve the OpenLineage connection"))] + ResolveConnection { + source: openlineage::v1alpha1::OpenLineageError, + }, + + #[snafu(display( + "failed to build volumes and mounts for the OpenLineage backend TLS CA certificate" + ))] + TlsVolumesAndMounts { source: TlsClientDetailsError }, +} + +/// Everything the coordinator needs to run the OpenLineage event listener, resolved from +/// `spec.clusterConfig.lineage` during the dereference step. +#[derive(Clone, Debug)] +pub struct ResolvedLineageConfig { + /// Connection-derived `event-listener.properties` entries. The cluster-dependent `trino.uri` + /// is added later by the properties builder. + pub properties: BTreeMap, + + /// Volumes for the backend CA certificate and (when authenticated) the token Secret. + pub volumes: Vec, + + /// Volume mounts matching [`Self::volumes`], to be added to the coordinator container. + pub volume_mounts: Vec, + + /// Commands run in the coordinator prepare container to import a `SecretClass` CA into the + /// client truststore. + pub init_container_extra_start_commands: Vec, +} + +impl ResolvedLineageConfig { + /// Resolves the OpenLineage connection (inline or referenced), backend TLS trust and (optional) + /// authentication into the coordinator-side configuration. + pub async fn from_config( + lineage: &OpenLineageJob, + client: &Client, + namespace: &str, + ) -> Result { + let mut properties = BTreeMap::new(); + let mut volumes = Vec::new(); + let mut volume_mounts = Vec::new(); + let mut init_container_extra_start_commands = Vec::new(); + + let connection = lineage + .connection + .clone() + .resolve(client, namespace) + .await + .context(ResolveConnectionSnafu)?; + + properties.insert( + EVENT_LISTENER_NAME_KEY.to_string(), + EVENT_LISTENER_NAME_OPENLINEAGE.to_string(), + ); + properties.insert( + OPENLINEAGE_TRANSPORT_TYPE_KEY.to_string(), + OPENLINEAGE_TRANSPORT_TYPE_HTTP.to_string(), + ); + properties.insert( + OPENLINEAGE_TRANSPORT_URL_KEY.to_string(), + connection.transport_url(), + ); + // Default the OpenLineage namespace to the workload's Kubernetes namespace. + properties.insert( + OPENLINEAGE_NAMESPACE_KEY.to_string(), + lineage + .namespace + .clone() + .unwrap_or_else(|| namespace.to_string()), + ); + + // The stable OpenLineage job name format. When unset, Trino defaults to `$QUERY_ID`. + if let Some(job_name) = &lineage.job_name { + properties.insert( + OPENLINEAGE_JOB_NAME_FORMAT_KEY.to_string(), + job_name.clone(), + ); + } + + // Backend TLS: mount and import a `SecretClass` CA into the client truststore. WebPKI and + // no verification need nothing (WebPKI is trusted via the system bundle already seeded into + // the truststore; without server verification the URL is plain `http`). + let (tls_volumes, tls_mounts) = connection + .tls + .volumes_and_mounts() + .context(TlsVolumesAndMountsSnafu)?; + volumes.extend(tls_volumes); + volume_mounts.extend(tls_mounts); + init_container_extra_start_commands + .extend(openlineage_tls_truststore_commands(&connection.tls)); + + // Authentication: when the connection sets `credentialsSecretName`, mount that Secret and + // reference the token via a `${file:...}` placeholder so it is resolved at startup and never + // lands in the ConfigMap. The token must be stored under the `apiKey` Secret key. + if let Some(secret_name) = &connection.credentials_secret_name { + volumes.push(Volume { + name: OPENLINEAGE_AUTH_VOLUME_NAME.to_string(), + secret: Some(SecretVolumeSource { + secret_name: Some(secret_name.clone()), + ..SecretVolumeSource::default() + }), + ..Volume::default() + }); + volume_mounts.push(VolumeMount { + name: OPENLINEAGE_AUTH_VOLUME_NAME.to_string(), + mount_path: STACKABLE_OPENLINEAGE_AUTH_DIR.to_string(), + read_only: Some(true), + ..VolumeMount::default() + }); + properties.insert( + OPENLINEAGE_TRANSPORT_API_KEY_KEY.to_string(), + format!( + "${{file:UTF-8:{STACKABLE_OPENLINEAGE_AUTH_DIR}/{OPENLINEAGE_AUTH_SECRET_KEY}}}" + ), + ); + } + + Ok(Self { + properties, + volumes, + volume_mounts, + init_container_extra_start_commands, + }) + } +} + +/// Init-container commands that add the backend's `SecretClass` CA certificate to the client +/// truststore (which is the JVM default truststore, see `config::jvm`). Returns an empty list when +/// no import is needed: no TLS, `verification.none` (plain `http`), or WebPKI verification (trusted +/// via the seeded system bundle). +fn openlineage_tls_truststore_commands(tls: &TlsClientDetails) -> Vec { + match tls.tls.as_ref().map(|tls| &tls.verification) { + Some(TlsVerification::Server(TlsServerVerification { + ca_cert: CaCert::SecretClass(_), + })) => tls + .tls_ca_cert_mount_path() + .map(|ca_cert| command::add_cert_to_truststore(&ca_cert, STACKABLE_CLIENT_TLS_DIR)) + .unwrap_or_default(), + _ => Vec::new(), + } +} + +#[cfg(test)] +mod tests { + use stackable_operator::commons::tls_verification::Tls; + + use super::*; + + fn tls_details(verification: Option) -> TlsClientDetails { + TlsClientDetails { + tls: verification.map(|verification| Tls { verification }), + } + } + + #[test] + fn no_tls_yields_no_truststore_commands() { + assert!( + openlineage_tls_truststore_commands(&tls_details(None)).is_empty(), + "plain http needs no truststore setup" + ); + } + + #[test] + fn verification_none_yields_no_truststore_commands() { + // For OpenLineage `verification.none` means no server verification, i.e. plain http. + assert!( + openlineage_tls_truststore_commands(&tls_details(Some(TlsVerification::None {}))) + .is_empty() + ); + } + + #[test] + fn webpki_verification_yields_no_truststore_commands() { + let details = tls_details(Some(TlsVerification::Server(TlsServerVerification { + ca_cert: CaCert::WebPki {}, + }))); + assert!(openlineage_tls_truststore_commands(&details).is_empty()); + } + + #[test] + fn secret_class_verification_yields_truststore_commands() { + let details = tls_details(Some(TlsVerification::Server(TlsServerVerification { + ca_cert: CaCert::SecretClass("openlineage-tls".to_string()), + }))); + assert!(!openlineage_tls_truststore_commands(&details).is_empty()); + } +} diff --git a/rust/operator-binary/src/config/mod.rs b/rust/operator-binary/src/config/mod.rs index 152d50804..f3a70f9d4 100644 --- a/rust/operator-binary/src/config/mod.rs +++ b/rust/operator-binary/src/config/mod.rs @@ -1,4 +1,5 @@ pub mod client_protocol; pub mod fault_tolerant_execution; pub mod jvm; +pub mod lineage; pub mod s3; diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index cb667ae26..9a77029bf 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -132,6 +132,17 @@ pub fn container_trino_args( ) )); + // Resolve the OpenLineage bearer token (if any) in the coordinator's event listener config. + // The file only exists on the coordinator; `${file:...}` placeholders are not resolved by Trino + // for `event-listener.properties` (trinodb/trino#8245), so we template it here at startup. + args.push(format!( + "test -f {rw_event_listener_config_file} && config-utils template {rw_event_listener_config_file}", + rw_event_listener_config_file = format!( + "{RW_CONFIG_DIR_NAME}/{event_listener}", + event_listener = ConfigFileName::EventListener + ) + )); + args.push("set -x".to_string()); // Start command diff --git a/rust/operator-binary/src/controller/build/properties/config_properties.rs b/rust/operator-binary/src/controller/build/properties/config_properties.rs index ca4a5298c..7ed38c5ac 100644 --- a/rust/operator-binary/src/controller/build/properties/config_properties.rs +++ b/rust/operator-binary/src/controller/build/properties/config_properties.rs @@ -6,11 +6,11 @@ use snafu::Snafu; use stackable_operator::{memory::BinaryMultiple, utils::cluster_info::KubernetesClusterInfo}; use crate::{ - controller::{TrinoRoleGroupConfig, ValidatedCluster}, + controller::{TrinoRoleGroupConfig, ValidatedCluster, build::properties::ConfigFileName}, crd::{ Container, ENV_INTERNAL_SECRET, HTTP_PORT, HTTPS_PORT, MAX_TRINO_LOG_FILES_SIZE, - STACKABLE_INTERNAL_TLS_DIR, STACKABLE_LOG_DIR, STACKABLE_SERVER_TLS_DIR, - STACKABLE_TLS_STORE_PASSWORD, TrinoRole, + RW_CONFIG_DIR_NAME, STACKABLE_INTERNAL_TLS_DIR, STACKABLE_LOG_DIR, + STACKABLE_SERVER_TLS_DIR, STACKABLE_TLS_STORE_PASSWORD, TrinoRole, discovery::{TrinoDiscovery, TrinoDiscoveryProtocol}, }, }; @@ -21,6 +21,9 @@ const HTTP_SERVER_LOG_ENABLED: &str = "http-server.log.enabled"; // config.properties const COORDINATOR: &str = "coordinator"; const DISCOVERY_URI: &str = "discovery.uri"; +// Registers event listener config files. Trino does not load `event-listener.properties` from +// `--etc-dir`; the file must be listed here explicitly (see trinodb/trino `EventListenerManager`). +const EVENT_LISTENER_CONFIG_FILES: &str = "event-listener.config-files"; const HTTP_SERVER_HTTP_PORT: &str = "http-server.http.port"; const QUERY_MAX_MEMORY: &str = "query.max-memory"; const QUERY_MAX_MEMORY_PER_NODE: &str = "query.max-memory-per-node"; @@ -96,6 +99,19 @@ pub fn build( (role == TrinoRole::Coordinator).to_string(), ); + // Register the OpenLineage event listener (coordinator only). The `event-listener.properties` + // file itself is built by `event_listener_properties` and lands in the read-write config dir; + // without this registration Trino would never load it (it does not scan `--etc-dir`). + if role == TrinoRole::Coordinator && cluster.cluster_config.lineage.is_some() { + props.insert( + EVENT_LISTENER_CONFIG_FILES.to_string(), + format!( + "{RW_CONFIG_DIR_NAME}/{event_listener}", + event_listener = ConfigFileName::EventListener + ), + ); + } + // Trino's own JSON logging output. props.insert(LOG_FORMAT.to_string(), "json".to_string()); props.insert( @@ -249,9 +265,12 @@ pub fn build( #[cfg(test)] mod tests { use super::*; - use crate::controller::build::properties::test_support::{ - MINIMAL_TRINO_YAML, file_auth_class, validated_cluster_from_yaml, - validated_cluster_from_yaml_with_auth, + use crate::{ + config::lineage::ResolvedLineageConfig, + controller::build::properties::test_support::{ + MINIMAL_TRINO_YAML, empty_derefs, file_auth_class, validated_cluster_from_yaml, + validated_cluster_from_yaml_with_auth, validated_cluster_from_yaml_with_derefs, + }, }; const SERVER_TLS_ONLY_YAML: &str = r#" @@ -347,6 +366,60 @@ mod tests { ); } + fn cluster_with_lineage() -> ValidatedCluster { + let mut derefs = empty_derefs(); + derefs.resolved_lineage_config = Some(ResolvedLineageConfig { + properties: BTreeMap::new(), + volumes: Vec::new(), + volume_mounts: Vec::new(), + init_container_extra_start_commands: Vec::new(), + }); + validated_cluster_from_yaml_with_derefs(MINIMAL_TRINO_YAML, derefs) + } + + #[test] + fn coordinator_registers_openlineage_event_listener_config_file() { + let cluster = cluster_with_lineage(); + let props = build( + &cluster, + TrinoRole::Coordinator, + &rg(&cluster, &TrinoRole::Coordinator), + &cluster_info(), + ) + .unwrap(); + // Trino only loads the event listener when it is registered here. + assert_eq!( + props.get(EVENT_LISTENER_CONFIG_FILES).map(String::as_str), + Some("/stackable/rwconfig/event-listener.properties") + ); + } + + #[test] + fn worker_does_not_register_event_listener_config_file() { + let cluster = cluster_with_lineage(); + let props = build( + &cluster, + TrinoRole::Worker, + &rg(&cluster, &TrinoRole::Worker), + &cluster_info(), + ) + .unwrap(); + assert!(!props.contains_key(EVENT_LISTENER_CONFIG_FILES)); + } + + #[test] + fn no_event_listener_config_file_without_lineage() { + let cluster = validated_cluster_from_yaml(MINIMAL_TRINO_YAML); + let props = build( + &cluster, + TrinoRole::Coordinator, + &rg(&cluster, &TrinoRole::Coordinator), + &cluster_info(), + ) + .unwrap(); + assert!(!props.contains_key(EVENT_LISTENER_CONFIG_FILES)); + } + #[test] fn server_tls_only_uses_server_keystore_dir_and_http_discovery() { let cluster = validated_cluster_from_yaml(SERVER_TLS_ONLY_YAML); diff --git a/rust/operator-binary/src/controller/build/properties/event_listener_properties.rs b/rust/operator-binary/src/controller/build/properties/event_listener_properties.rs new file mode 100644 index 000000000..d70f844bf --- /dev/null +++ b/rust/operator-binary/src/controller/build/properties/event_listener_properties.rs @@ -0,0 +1,199 @@ +//! Builder for the coordinator's `event-listener.properties` (the Trino OpenLineage event listener). +//! +//! The OpenLineage event listener runs on the **coordinator only**, so this builder returns an +//! empty map for every other role and the caller omits the file from those ConfigMaps. For the +//! coordinator it emits the connection-derived settings resolved in [`crate::config::lineage`], +//! adds the cluster-facing `trino.uri`, and finally merges any user `event-listener.properties` +//! overrides (highest precedence). + +use std::collections::BTreeMap; + +use stackable_operator::utils::cluster_info::KubernetesClusterInfo; + +use crate::{ + config::lineage::OPENLINEAGE_TRINO_URI_KEY, + controller::{TrinoRoleGroupConfig, ValidatedCluster}, + crd::{ + TrinoRole, + discovery::{TrinoDiscovery, TrinoDiscoveryProtocol}, + }, +}; + +/// Build the `event-listener.properties` key/value pairs. +/// +/// Returns an empty map when OpenLineage is not configured and no user overrides are provided (and +/// always for non-coordinator roles). Callers should omit the file from the ConfigMap in that case. +pub fn build( + cluster: &ValidatedCluster, + role: TrinoRole, + rg: &TrinoRoleGroupConfig, + cluster_info: &KubernetesClusterInfo, +) -> BTreeMap { + let mut props = BTreeMap::new(); + + // Event listeners only run on the coordinator. + if role != TrinoRole::Coordinator { + return props; + } + + if let Some(lineage) = &cluster.cluster_config.lineage { + // Connection-derived settings (transport type/URL, namespace and optional api-key). + props.extend(lineage.properties.clone()); + + // The URI identifying this Trino cluster in emitted lineage. Uses the same coordinator + // address as `discovery.uri`; the scheme follows the client-facing TLS setting. + if let Some(coordinator_ref) = cluster.cluster_config.coordinator_pod_refs.first() { + let protocol = if cluster.tls_enabled() { + TrinoDiscoveryProtocol::Https + } else { + TrinoDiscoveryProtocol::Http + }; + let discovery = TrinoDiscovery::new(coordinator_ref, protocol); + props.insert( + OPENLINEAGE_TRINO_URI_KEY.to_string(), + discovery.discovery_uri(cluster_info), + ); + } + } + + // User overrides (highest precedence). + props.extend(rg.config_overrides.event_listener_properties.clone()); + + props +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use stackable_operator::utils::cluster_info::KubernetesClusterInfo; + + use super::*; + use crate::{ + config::lineage::{ + EVENT_LISTENER_NAME_KEY, OPENLINEAGE_NAMESPACE_KEY, OPENLINEAGE_TRANSPORT_API_KEY_KEY, + OPENLINEAGE_TRANSPORT_TYPE_KEY, OPENLINEAGE_TRANSPORT_URL_KEY, ResolvedLineageConfig, + }, + controller::{ + ValidatedCluster, + build::properties::test_support::{MINIMAL_TRINO_YAML, empty_derefs}, + }, + crd::TrinoRole, + }; + + fn cluster_info() -> KubernetesClusterInfo { + KubernetesClusterInfo { + cluster_domain: "cluster.local".parse().unwrap(), + } + } + + /// A resolved OpenLineage config as `config::lineage` would produce it for an inline + /// `http://marquez:5000` connection, optionally with a bearer-token api-key reference. + fn resolved_lineage(with_auth: bool) -> ResolvedLineageConfig { + let mut properties = BTreeMap::from([ + ( + EVENT_LISTENER_NAME_KEY.to_string(), + "openlineage".to_string(), + ), + ( + OPENLINEAGE_TRANSPORT_TYPE_KEY.to_string(), + "HTTP".to_string(), + ), + ( + OPENLINEAGE_TRANSPORT_URL_KEY.to_string(), + "http://marquez:5000".to_string(), + ), + (OPENLINEAGE_NAMESPACE_KEY.to_string(), "default".to_string()), + ]); + if with_auth { + properties.insert( + OPENLINEAGE_TRANSPORT_API_KEY_KEY.to_string(), + "${file:UTF-8:/stackable/openlineage_auth/apiKey}".to_string(), + ); + } + ResolvedLineageConfig { + properties, + volumes: Vec::new(), + volume_mounts: Vec::new(), + init_container_extra_start_commands: Vec::new(), + } + } + + fn cluster_with_lineage(with_auth: bool) -> ValidatedCluster { + let mut derefs = empty_derefs(); + derefs.resolved_lineage_config = Some(resolved_lineage(with_auth)); + crate::controller::build::properties::test_support::validated_cluster_from_yaml_with_derefs( + MINIMAL_TRINO_YAML, + derefs, + ) + } + + fn coordinator_rg(cluster: &ValidatedCluster) -> TrinoRoleGroupConfig { + cluster.role_group_configs[&TrinoRole::Coordinator] + .values() + .next() + .unwrap() + .clone() + } + + #[test] + fn worker_role_renders_empty() { + let cluster = cluster_with_lineage(false); + // Reuse the coordinator role group config; the role argument alone must gate emission. + let rg = coordinator_rg(&cluster); + let props = build(&cluster, TrinoRole::Worker, &rg, &cluster_info()); + assert!( + props.is_empty(), + "event listeners must not be configured on workers" + ); + } + + #[test] + fn coordinator_without_lineage_renders_empty() { + let cluster = + crate::controller::build::properties::test_support::validated_cluster_from_yaml( + MINIMAL_TRINO_YAML, + ); + let rg = coordinator_rg(&cluster); + let props = build(&cluster, TrinoRole::Coordinator, &rg, &cluster_info()); + assert!(props.is_empty()); + } + + #[test] + fn coordinator_emits_listener_transport_and_trino_uri() { + let cluster = cluster_with_lineage(false); + let rg = coordinator_rg(&cluster); + let props = build(&cluster, TrinoRole::Coordinator, &rg, &cluster_info()); + + assert_eq!(props.get(EVENT_LISTENER_NAME_KEY).unwrap(), "openlineage"); + assert_eq!(props.get(OPENLINEAGE_TRANSPORT_TYPE_KEY).unwrap(), "HTTP"); + assert_eq!( + props.get(OPENLINEAGE_TRANSPORT_URL_KEY).unwrap(), + "http://marquez:5000" + ); + assert_eq!(props.get(OPENLINEAGE_NAMESPACE_KEY).unwrap(), "default"); + // The default Trino cluster enables server TLS, so the recorded Trino URI is https. + let trino_uri = props + .get(OPENLINEAGE_TRINO_URI_KEY) + .expect("trino.uri is set"); + assert!( + trino_uri.starts_with("https://") && trino_uri.contains("coordinator"), + "trino.uri should be the coordinator address, got: {trino_uri}" + ); + // No auth configured -> no api-key. + assert!(!props.contains_key(OPENLINEAGE_TRANSPORT_API_KEY_KEY)); + } + + #[test] + fn coordinator_with_auth_emits_api_key_file_reference() { + let cluster = cluster_with_lineage(true); + let rg = coordinator_rg(&cluster); + let props = build(&cluster, TrinoRole::Coordinator, &rg, &cluster_info()); + + let api_key = props.get(OPENLINEAGE_TRANSPORT_API_KEY_KEY).unwrap(); + assert!( + api_key.starts_with("${file:UTF-8:"), + "the token must be referenced from a file, never inlined: {api_key}" + ); + } +} diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs index e461f02f7..e0840da74 100644 --- a/rust/operator-binary/src/controller/build/properties/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/mod.rs @@ -6,6 +6,7 @@ pub mod access_control_properties; pub mod config_properties; +pub mod event_listener_properties; pub mod exchange_manager_properties; pub mod log_properties; pub mod logging; @@ -31,6 +32,8 @@ pub enum ConfigFileName { ExchangeManager, #[strum(serialize = "spooling-manager.properties")] SpoolingManager, + #[strum(serialize = "event-listener.properties")] + EventListener, } #[cfg(test)] @@ -51,6 +54,7 @@ pub(crate) mod test_support { trino_opa_config: None, resolved_fte_config: None, resolved_client_protocol_config: None, + resolved_lineage_config: None, } } diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index 7476f9f09..a3ee09636 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -15,8 +15,8 @@ use crate::{ RoleGroupName, ValidatedCluster, build::properties::{ ConfigFileName, access_control_properties, config_properties, - exchange_manager_properties, log_properties, node_properties, product_logging, - security_properties, spooling_manager_properties, + event_listener_properties, exchange_manager_properties, log_properties, + node_properties, product_logging, security_properties, spooling_manager_properties, }, }, crd::TrinoRole, @@ -161,7 +161,18 @@ pub fn build_rolegroup_config_map( ); } - // 8. jvm.config. The role + role-group `jvmArgumentOverrides` were already merged in the + // 8. event-listener.properties (optional, coordinator only — the OpenLineage event listener). + let el = event_listener_properties::build(cluster, role.clone(), rg, cluster_info); + if !el.is_empty() { + data.insert( + ConfigFileName::EventListener.to_string(), + to_java_properties_string(el.iter()).with_context(|_| WritePropertiesSnafu { + file: ConfigFileName::EventListener.to_string(), + })?, + ); + } + + // 9. jvm.config. The role + role-group `jvmArgumentOverrides` were already merged in the // validate step and are carried by `product_specific_common_config`. let jvm_config = jvm::jvm_config( cluster.product_version, @@ -171,7 +182,7 @@ pub fn build_rolegroup_config_map( .context(BuildJvmConfigSnafu)?; data.insert(JVM_CONFIG.to_string(), jvm_config); - // 9. Vector agent config (`vector.yaml`) if the Vector agent is enabled. The file is templated + // 10. Vector agent config (`vector.yaml`) if the Vector agent is enabled. The file is templated // with environment variables injected by the Vector container at runtime. if rg.config.logging.enable_vector_agent { data.insert( diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 206686304..f39be8378 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -270,6 +270,13 @@ pub fn build_rolegroup_statefulset( )); } + // Add the OpenLineage backend CA certificate to the truststore if configured (coordinator only). + if trino_role == &TrinoRole::Coordinator + && let Some(resolved_lineage) = &cluster.cluster_config.lineage + { + prepare_args.extend(resolved_lineage.init_container_extra_start_commands.clone()); + } + let container_prepare = cb_prepare .image_from_product_image(resolved_product_image) .command(vec![ @@ -780,5 +787,21 @@ fn tls_volume_mounts( .context(AddVolumeSnafu)?; } + // OpenLineage backend CA certificate and (when authenticated) the bearer-token Secret. The + // event listener runs on the coordinator only, so these are mounted there only. + if trino_role == &TrinoRole::Coordinator + && let Some(resolved_lineage) = &cluster.cluster_config.lineage + { + cb_prepare + .add_volume_mounts(resolved_lineage.volume_mounts.clone()) + .context(AddVolumeMountSnafu)?; + cb_trino + .add_volume_mounts(resolved_lineage.volume_mounts.clone()) + .context(AddVolumeMountSnafu)?; + pod_builder + .add_volumes(resolved_lineage.volumes.clone()) + .context(AddVolumeSnafu)?; + } + Ok(()) } diff --git a/rust/operator-binary/src/controller/dereference.rs b/rust/operator-binary/src/controller/dereference.rs index 93a8d0502..1231af12e 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -18,6 +18,7 @@ use crate::{ config::{ client_protocol::{self, ResolvedClientProtocolConfig}, fault_tolerant_execution::{self, ResolvedFaultTolerantExecutionConfig}, + lineage::{self, ResolvedLineageConfig}, }, crd::{ authentication::{ResolvedAuthenticationClassRef, resolve_authentication_classes}, @@ -66,6 +67,9 @@ pub enum Error { #[snafu(display("failed to resolve client protocol configuration"))] ClientProtocolConfiguration { source: client_protocol::Error }, + #[snafu(display("failed to resolve OpenLineage configuration"))] + OpenLineageConfiguration { source: lineage::Error }, + #[snafu(display("invalid OpaConfig"))] InvalidOpaConfig { source: stackable_operator::commons::opa::Error, @@ -88,6 +92,7 @@ pub struct DereferencedObjects { pub trino_opa_config: Option, pub resolved_fte_config: Option, pub resolved_client_protocol_config: Option, + pub resolved_lineage_config: Option, } /// Fetches all Kubernetes objects referenced from the [`v1alpha1::TrinoCluster`] spec. @@ -171,6 +176,15 @@ pub async fn dereference( None => None, }; + let resolved_lineage_config = match trino.spec.cluster_config.lineage.as_ref() { + Some(lineage) => Some( + ResolvedLineageConfig::from_config(lineage, client, namespace.as_ref()) + .await + .context(OpenLineageConfigurationSnafu)?, + ), + None => None, + }; + Ok(DereferencedObjects { resolved_authentication_classes, catalog_definitions, @@ -178,6 +192,7 @@ pub async fn dereference( trino_opa_config, resolved_fte_config, resolved_client_protocol_config, + resolved_lineage_config, }) } diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 7e00d944c..d73704dce 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -38,6 +38,7 @@ use crate::{ config::{ client_protocol::ResolvedClientProtocolConfig, fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig, + lineage::ResolvedLineageConfig, }, crd::{APP_NAME, TrinoRole, catalog::TrinoCatalogName, discovery::TrinoPodRef, v1alpha1}, trino_controller::{CONTROLLER_NAME, OPERATOR_NAME}, @@ -87,6 +88,7 @@ pub struct ValidatedClusterConfig { pub authorization: Option, pub fault_tolerant_execution: Option, pub client_protocol: Option, + pub lineage: Option, pub coordinator_pod_refs: Vec, pub catalogs: BTreeMap, } @@ -436,6 +438,7 @@ pub(crate) fn validated_cluster() -> ValidatedCluster { trino_opa_config: None, resolved_fte_config: None, resolved_client_protocol_config: None, + resolved_lineage_config: None, }; let operator_env = OperatorEnvironmentOptions { operator_namespace: "stackable-operators".to_string(), diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 21d82b904..e3f9b2da3 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -298,6 +298,7 @@ pub fn validate( authorization: dereferenced_objects.trino_opa_config.clone(), fault_tolerant_execution: dereferenced_objects.resolved_fte_config.clone(), client_protocol: dereferenced_objects.resolved_client_protocol_config.clone(), + lineage: dereferenced_objects.resolved_lineage_config.clone(), coordinator_pod_refs: trino.coordinator_pods(&namespace).collect(), catalogs, }; @@ -399,6 +400,7 @@ mod tests { trino_opa_config: None, resolved_fte_config: None, resolved_client_protocol_config: None, + resolved_lineage_config: None, } } diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 5661b0023..57e1c3e18 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -22,7 +22,7 @@ use stackable_operator::{ }, }, config::{fragment::Fragment, merge::Merge}, - crd::authentication::core, + crd::{authentication::core, openlineage}, deep_merger::ObjectOverrides, k8s_openapi::apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::LabelSelector}, kube::{CustomResource, ResourceExt}, @@ -85,6 +85,10 @@ pub const STACKABLE_TLS_STORE_PASSWORD: &str = "changeit"; // secret vars pub const ENV_INTERNAL_SECRET: &str = "INTERNAL_SECRET"; pub const ENV_SPOOLING_SECRET: &str = "SPOOLING_SECRET"; +// OpenLineage +/// Fixed key that must hold the OpenLineage HTTP transport bearer token inside the Secret named by +/// `credentialsSecretName` on the connection used in `spec.clusterConfig.lineage`. +pub const OPENLINEAGE_AUTH_SECRET_KEY: &str = "apiKey"; // TLS const TLS_DEFAULT_SECRET_CLASS: &str = "tls"; // Listener @@ -205,6 +209,10 @@ pub mod versioned { // File name defined in [`crate::controller::build::properties::ConfigFileName`] #[serde(default, rename = "spooling-manager.properties")] pub spooling_manager_properties: KeyValueConfigOverrides, + + // File name defined in [`crate::controller::build::properties::ConfigFileName`] + #[serde(default, rename = "event-listener.properties")] + pub event_listener_properties: KeyValueConfigOverrides, } #[derive(Clone, Debug, Default, Fragment, JsonSchema, PartialEq)] @@ -285,6 +293,11 @@ pub mod versioned { #[serde(skip_serializing_if = "Option::is_none")] pub client_protocol: Option, + /// Emit [OpenLineage](https://openlineage.io/) lineage events for the queries run on this + /// Trino cluster. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lineage: Option, + /// Name of the Vector aggregator [discovery ConfigMap](DOCS_BASE_URL_PLACEHOLDER/concepts/service_discovery). /// It must contain the key `ADDRESS` with the address of the Vector aggregator. /// Follow the [logging tutorial](DOCS_BASE_URL_PLACEHOLDER/tutorials/logging-vector-aggregator) diff --git a/rust/operator-binary/src/trino_controller.rs b/rust/operator-binary/src/trino_controller.rs index 8c9741eb4..eb27944b9 100644 --- a/rust/operator-binary/src/trino_controller.rs +++ b/rust/operator-binary/src/trino_controller.rs @@ -371,6 +371,7 @@ mod tests { trino_opa_config, resolved_fte_config, resolved_client_protocol_config, + resolved_lineage_config: None, }; let operator_env = OperatorEnvironmentOptions { @@ -602,6 +603,7 @@ mod tests { trino_opa_config: None, resolved_fte_config: None, resolved_client_protocol_config: None, + resolved_lineage_config: None, }; let operator_env = OperatorEnvironmentOptions { operator_namespace: "stackable-operators".to_string(), diff --git a/tests/templates/kuttl/openlineage/00-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/openlineage/00-install-vector-aggregator-discovery-configmap.yaml.j2 new file mode 100644 index 000000000..2d6a0df5f --- /dev/null +++ b/tests/templates/kuttl/openlineage/00-install-vector-aggregator-discovery-configmap.yaml.j2 @@ -0,0 +1,9 @@ +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +data: + ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }} +{% endif %} diff --git a/tests/templates/kuttl/openlineage/00-patch-ns.yaml.j2 b/tests/templates/kuttl/openlineage/00-patch-ns.yaml.j2 new file mode 100644 index 000000000..67185acfd --- /dev/null +++ b/tests/templates/kuttl/openlineage/00-patch-ns.yaml.j2 @@ -0,0 +1,9 @@ +{% if test_scenario['values']['openshift'] == 'true' %} +# see https://github.com/stackabletech/issues/issues/566 +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}' + timeout: 120 +{% endif %} diff --git a/tests/templates/kuttl/openlineage/00-rbac.yaml.j2 b/tests/templates/kuttl/openlineage/00-rbac.yaml.j2 new file mode 100644 index 000000000..9cbf03511 --- /dev/null +++ b/tests/templates/kuttl/openlineage/00-rbac.yaml.j2 @@ -0,0 +1,29 @@ +--- +kind: Role +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: use-integration-tests-scc +rules: +{% if test_scenario['values']['openshift'] == "true" %} + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + resourceNames: ["privileged"] + verbs: ["use"] +{% endif %} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: integration-tests-sa +--- +kind: RoleBinding +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: use-integration-tests-scc +subjects: + - kind: ServiceAccount + name: integration-tests-sa +roleRef: + kind: Role + name: use-integration-tests-scc + apiGroup: rbac.authorization.k8s.io diff --git a/tests/templates/kuttl/openlineage/01-assert.yaml b/tests/templates/kuttl/openlineage/01-assert.yaml new file mode 100644 index 000000000..8f5f9b3e8 --- /dev/null +++ b/tests/templates/kuttl/openlineage/01-assert.yaml @@ -0,0 +1,12 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 300 +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: openlineage-receiver +status: + readyReplicas: 1 + replicas: 1 diff --git a/tests/templates/kuttl/openlineage/01-install-receiver.yaml.j2 b/tests/templates/kuttl/openlineage/01-install-receiver.yaml.j2 new file mode 100644 index 000000000..8ec0cc6b0 --- /dev/null +++ b/tests/templates/kuttl/openlineage/01-install-receiver.yaml.j2 @@ -0,0 +1,7 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # `envsubst` injects the kuttl-assigned $NAMESPACE (only $NAMESPACE, so the nginx config's own + # `$` tokens are left untouched) into the SecretClass name / cert scope. + - script: envsubst '$NAMESPACE' < 01_receiver.yaml | kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/openlineage/01_receiver.yaml.j2 b/tests/templates/kuttl/openlineage/01_receiver.yaml.j2 new file mode 100644 index 000000000..2bdb3ccd9 --- /dev/null +++ b/tests/templates/kuttl/openlineage/01_receiver.yaml.j2 @@ -0,0 +1,163 @@ +# Lightweight OpenLineage receiver. +# +# Instead of a full Marquez, this is a plain nginx that accepts the OpenLineage HTTP transport's +# `POST /api/v1/lineage` requests and persists each request body as a flat file. The assertion in +# `05-assert.yaml` then greps those files to prove events were delivered (and, in the TLS scenario, +# that the coordinator trusted the secret-operator-issued server cert end-to-end). +# +# nginx cannot write a request body to disk on a `return` (the body is discarded), so the receive +# location sets `client_body_in_file_only on` and `proxy_pass`es to a local sink server; nginx then +# buffers the full body to a file (kept, not cleaned) before forwarding. Bodies land in +# `/tmp/lineage/`. +# +# `$NAMESPACE` is substituted by the `01-install-receiver` step via `envsubst` (kuttl assigns the +# namespace at run time; beku leaves the literal in place). +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} +--- +# A genuine autoTls SecretClass: the secret operator generates the CA and issues the nginx server +# cert. Suffixed with the (cluster-scoped) namespace to avoid collisions across parallel scenarios. +apiVersion: secrets.stackable.tech/v1alpha1 +kind: SecretClass +metadata: + name: openlineage-tls-$NAMESPACE +spec: + backend: + autoTls: + ca: + autoGenerate: true + secret: + name: openlineage-tls-ca + namespace: $NAMESPACE +{% endif %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: openlineage-receiver-config +data: + nginx.conf: | + worker_processes 1; + error_log /dev/stderr notice; + pid /tmp/nginx.pid; + events { worker_connections 1024; } + http { + access_log /dev/stdout; + client_body_temp_path /tmp/nginx-body; + proxy_temp_path /tmp/nginx-proxy; + fastcgi_temp_path /tmp/nginx-fastcgi; + uwsgi_temp_path /tmp/nginx-uwsgi; + scgi_temp_path /tmp/nginx-scgi; + + # Internal sink: acknowledges the forwarded lineage event. + server { + listen 127.0.0.1:8080; + location / { return 201; } + } + + server { + listen 5000{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} ssl{% endif %}; +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + ssl_certificate /stackable/tls/tls.crt; + ssl_certificate_key /stackable/tls/tls.key; +{% endif %} + client_max_body_size 16m; + + # Persist each OpenLineage event body to /tmp/lineage/, then acknowledge via the sink. + location = /api/v1/lineage { +{% if test_scenario['values']['openlineage-use-auth'] == 'true' %} + # Require the bearer token the operator wires in from the credentialsSecretName Secret. + # Without a valid token nginx returns 401 and never persists a body, so the 05-assert + # event checks only pass if the coordinator actually authenticated. + if ($http_authorization != "Bearer openlineage-test-token") { return 401; } +{% endif %} + client_body_in_file_only on; + client_body_temp_path /tmp/lineage; + proxy_pass http://127.0.0.1:8080; + } + + location / { return 204; } + } + } +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: openlineage-receiver + labels: + app: openlineage-receiver +spec: + replicas: 1 + selector: + matchLabels: + app: openlineage-receiver + template: + metadata: + labels: + app: openlineage-receiver + spec: +{% if test_scenario['values']['openshift'] != 'true' %} + # On plain Kubernetes, make mounted emptyDirs writable by the nginx-unprivileged group (101). + # On OpenShift this is omitted so the restricted SCC can assign an fsGroup from the project range. + securityContext: + fsGroup: 101 +{% endif %} + containers: + - name: nginx + image: docker.io/nginxinc/nginx-unprivileged:1.27-alpine + ports: + - containerPort: 5000 + volumeMounts: + - name: config + mountPath: /etc/nginx/nginx.conf + subPath: nginx.conf + - name: tmp + mountPath: /tmp +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + - name: tls + mountPath: /stackable/tls +{% endif %} + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 128Mi + volumes: + - name: config + configMap: + name: openlineage-receiver-config + - name: tmp + emptyDir: {} +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + - name: tls + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: openlineage-tls-$NAMESPACE + # `service` scope puts openlineage-receiver..svc.cluster.local in the cert SANs, + # which is exactly the host the Trino coordinator dials (see the TrinoCluster). + secrets.stackable.tech/scope: service=openlineage-receiver + secrets.stackable.tech/format: tls-pem + spec: + accessModes: ["ReadWriteOnce"] + resources: + requests: + storage: "1" + storageClassName: secrets.stackable.tech +{% endif %} +--- +apiVersion: v1 +kind: Service +metadata: + name: openlineage-receiver + labels: + app: openlineage-receiver +spec: + selector: + app: openlineage-receiver + ports: + - name: http + port: 5000 + targetPort: 5000 diff --git a/tests/templates/kuttl/openlineage/02-assert.yaml b/tests/templates/kuttl/openlineage/02-assert.yaml new file mode 100644 index 000000000..da1a5508a --- /dev/null +++ b/tests/templates/kuttl/openlineage/02-assert.yaml @@ -0,0 +1,25 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: trino-openlineage-coordinator-default +status: + readyReplicas: 1 + replicas: 1 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: trino-openlineage-worker-default +status: + readyReplicas: 1 + replicas: 1 +--- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCatalog +metadata: + name: tpch diff --git a/tests/templates/kuttl/openlineage/02-install-trino.yaml.j2 b/tests/templates/kuttl/openlineage/02-install-trino.yaml.j2 new file mode 100644 index 000000000..ae1b71cec --- /dev/null +++ b/tests/templates/kuttl/openlineage/02-install-trino.yaml.j2 @@ -0,0 +1,8 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # `envsubst` injects the kuttl-assigned $NAMESPACE into the (cluster-scoped) SecretClass / + # references, so parallel scenarios do not collide. + - script: envsubst '$NAMESPACE' < 02_trino.yaml | kubectl apply -n $NAMESPACE -f - + timeout: 120 diff --git a/tests/templates/kuttl/openlineage/02_trino.yaml.j2 b/tests/templates/kuttl/openlineage/02_trino.yaml.j2 new file mode 100644 index 000000000..00dc5dfcb --- /dev/null +++ b/tests/templates/kuttl/openlineage/02_trino.yaml.j2 @@ -0,0 +1,91 @@ +{% if test_scenario['values']['openlineage-use-auth'] == 'true' %} +--- +apiVersion: v1 +kind: Secret +metadata: + name: openlineage-token +stringData: + # The operator resolves this into the OpenLineage HTTP transport `api-key`; the nginx receiver + # requires exactly `Authorization: Bearer openlineage-test-token`. + apiKey: "openlineage-test-token" +{% endif %} +--- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCluster +metadata: + name: trino-openlineage +spec: + image: +{% if test_scenario['values']['trino-latest'].find(",") > 0 %} + custom: "{{ test_scenario['values']['trino-latest'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['trino-latest'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['trino-latest'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: + catalogLabelSelector: + matchLabels: + trino: trino-openlineage + lineage: + connection: + inline: + # Use the full service FQDN: with TLS the backend's secret-operator cert only carries the + # `openlineage-receiver..svc.cluster.local` SAN, so a shorter host fails verification. + host: openlineage-receiver.$NAMESPACE.svc.cluster.local + port: 5000 +{% if test_scenario['values']['openlineage-use-tls'] == 'true' %} + tls: + verification: + server: + caCert: + secretClass: openlineage-tls-$NAMESPACE +{% endif %} +{% if test_scenario['values']['openlineage-use-auth'] == 'true' %} + credentialsSecretName: openlineage-token +{% endif %} + # Sets openlineage-event-listener.job.name-format (asserted in 03-assert.yaml). + jobName: "trino-$QUERY_ID" +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + coordinators: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + roleGroups: + default: + replicas: 1 + config: {} + workers: + config: + gracefulShutdownTimeout: 5s # Let the test run faster + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + roleGroups: + default: + replicas: 1 + config: {} +--- +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCatalog +metadata: + name: tpch + labels: + trino: trino-openlineage +spec: + connector: + tpch: {} +--- +# A writable target so the test can run a CREATE TABLE AS SELECT. The OpenLineage event listener +# only emits for data-modifying query types (INSERT/DATA_DEFINITION/...); a plain SELECT is filtered +# out by the plugin's default `include-query-types`. +apiVersion: trino.stackable.tech/v1alpha1 +kind: TrinoCatalog +metadata: + name: blackhole + labels: + trino: trino-openlineage +spec: + connector: + blackHole: {} diff --git a/tests/templates/kuttl/openlineage/03-assert.yaml b/tests/templates/kuttl/openlineage/03-assert.yaml new file mode 100644 index 000000000..318e0ea1c --- /dev/null +++ b/tests/templates/kuttl/openlineage/03-assert.yaml @@ -0,0 +1,27 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 300 +commands: + # The coordinator ConfigMap must carry a generated event-listener.properties selecting the + # OpenLineage plugin and pointing at the receiver, and config.properties must register that file + # (Trino does not load it from --etc-dir otherwise). + - script: | + CM=$(kubectl get -n "$NAMESPACE" configmap trino-openlineage-coordinator-default -o yaml) + echo "$CM" | grep -q "event-listener.properties" + echo "$CM" | grep -q "event-listener.name=openlineage" + echo "$CM" | grep -q "openlineage-event-listener.transport.url" + echo "$CM" | grep -q "event-listener.config-files=/stackable/rwconfig/event-listener.properties" + # jobName maps to the OpenLineage job name format. + echo "$CM" | grep -q "openlineage-event-listener.job.name-format=trino-" + # Workers must NOT get an event listener configured. + - script: | + ! kubectl get -n "$NAMESPACE" configmap trino-openlineage-worker-default -o yaml | grep -q "event-listener.properties" +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: trino-test-helper +status: + readyReplicas: 1 + replicas: 1 diff --git a/tests/templates/kuttl/openlineage/03-install-test-helper.yaml b/tests/templates/kuttl/openlineage/03-install-test-helper.yaml new file mode 100644 index 000000000..ff0d29e60 --- /dev/null +++ b/tests/templates/kuttl/openlineage/03-install-test-helper.yaml @@ -0,0 +1,29 @@ +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: trino-test-helper + labels: + app: trino-test-helper +spec: + replicas: 1 + selector: + matchLabels: + app: trino-test-helper + template: + metadata: + labels: + app: trino-test-helper + spec: + serviceAccount: integration-tests-sa + containers: + - name: trino-test-helper + image: oci.stackable.tech/sdp/testing-tools/trino:0.3.0-stackable0.0.0-dev + command: ["sleep", "infinity"] + resources: + requests: + cpu: "250m" + memory: "64Mi" + limits: + cpu: "250m" + memory: "64Mi" diff --git a/tests/templates/kuttl/openlineage/04-run-query.yaml b/tests/templates/kuttl/openlineage/04-run-query.yaml new file mode 100644 index 000000000..4d4b4d9ff --- /dev/null +++ b/tests/templates/kuttl/openlineage/04-run-query.yaml @@ -0,0 +1,7 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: kubectl cp -n $NAMESPACE 04_emit_lineage.py trino-test-helper-0:/tmp/ + - script: kubectl exec -n $NAMESPACE trino-test-helper-0 -- python /tmp/04_emit_lineage.py -c trino-openlineage-coordinator + timeout: 120 diff --git a/tests/templates/kuttl/openlineage/04_emit_lineage.py b/tests/templates/kuttl/openlineage/04_emit_lineage.py new file mode 100644 index 000000000..b6bd3a448 --- /dev/null +++ b/tests/templates/kuttl/openlineage/04_emit_lineage.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python +"""Run a data-modifying query against Trino so the OpenLineage event listener emits lineage events. + +A plain SELECT is filtered out by the plugin's default `include-query-types`; a CREATE TABLE AS +SELECT (INSERT / DATA_DEFINITION) is emitted, with `tpch.tiny.nation` as the input dataset and the +blackhole table as the output dataset. +""" + +import argparse + +import trino + + +def get_connection(coordinator): + return trino.dbapi.connect( + host=coordinator, + port=8443, + user="test", + http_scheme="https", + verify=False, + session_properties={"query_max_execution_time": "60s"}, + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument( + "-c", + "--coordinator", + required=True, + help="Trino Coordinator Host to connect to", + ) + args = vars(parser.parse_args()) + + conn = get_connection(args["coordinator"]) + cursor = conn.cursor() + + cursor.execute("CREATE SCHEMA IF NOT EXISTS blackhole.lineage_test") + cursor.fetchall() + cursor.execute("DROP TABLE IF EXISTS blackhole.lineage_test.nation_copy") + cursor.fetchall() + cursor.execute( + "CREATE TABLE blackhole.lineage_test.nation_copy AS SELECT * FROM tpch.tiny.nation" + ) + result = cursor.fetchone() + assert result[0] == 25, f"unexpected CTAS row count: {result[0]}" + print("CTAS executed, OpenLineage events should have been emitted") diff --git a/tests/templates/kuttl/openlineage/05-assert.yaml b/tests/templates/kuttl/openlineage/05-assert.yaml new file mode 100644 index 000000000..2506144c6 --- /dev/null +++ b/tests/templates/kuttl/openlineage/05-assert.yaml @@ -0,0 +1,15 @@ +--- +# Verify the OpenLineage events actually reached the receiver. In the TLS scenario this also proves +# the coordinator trusted the secret-operator-issued cert end-to-end (a failed handshake would mean +# no body files were written); in the auth scenario it proves the coordinator sent the bearer token +# (without it nginx returns 401 and persists nothing). +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 180 +commands: + - script: | + POD=$(kubectl get -n "$NAMESPACE" pod -l app=openlineage-receiver -o jsonpath='{.items[0].metadata.name}') + # At least one lineage event body was persisted, referencing the CTAS output dataset ... + kubectl exec -n "$NAMESPACE" "$POD" -- sh -c 'cat /tmp/lineage/* 2>/dev/null | grep -q "nation_copy"' + # ... and a COMPLETE run event was received (the query finished and reported lineage). + kubectl exec -n "$NAMESPACE" "$POD" -- sh -c 'cat /tmp/lineage/* 2>/dev/null | grep -q "COMPLETE"' diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index c765a62d8..9a76dfdaf 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -68,6 +68,14 @@ dimensions: values: - "false" - "true" + - name: openlineage-use-tls + values: + - "false" + - "true" + - name: openlineage-use-auth + values: + - "false" + - "true" - name: openshift values: - "false" @@ -125,6 +133,12 @@ tests: # Long term we should test all versions - trino-latest - openshift + - name: openlineage + dimensions: + - trino-latest + - openlineage-use-tls + - openlineage-use-auth + - openshift - name: listener dimensions: - trino