diff --git a/CHANGELOG.md b/CHANGELOG.md index f8aba730..46a3e3f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file. ### Added +- Add `spec.clusterConfig.lineage` to emit [OpenLineage](https://openlineage.io/) lineage events ([#914]). - Support floating tags for product images via the new `spec.image.stackableVersionPolicy` field ([#944]). - Add `/ready` endpoint to the operator Deployment, which reports the CRD installation status ([#949]). @@ -70,6 +71,7 @@ All notable changes to this project will be documented in this file. [#909]: https://github.com/stackabletech/trino-operator/pull/909 [#913]: https://github.com/stackabletech/trino-operator/pull/913 +[#914]: https://github.com/stackabletech/trino-operator/pull/914 [#918]: https://github.com/stackabletech/trino-operator/pull/918 [#923]: https://github.com/stackabletech/trino-operator/pull/923 [#924]: https://github.com/stackabletech/trino-operator/pull/924 diff --git a/Cargo.lock b/Cargo.lock index f880ffff..fa578e38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -402,9 +402,9 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "combine" -version = "4.6.8" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "memchr", @@ -602,9 +602,9 @@ dependencies = [ [[package]] name = "defmt" -version = "1.1.1" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1" +checksum = "e4a0795ac740cb54d7b3775c9c17345f6aaecf884c01733b39d6017738e0e3e1" dependencies = [ "bitflags 1.3.2", "defmt-macros", @@ -612,21 +612,21 @@ dependencies = [ [[package]] name = "defmt-macros" -version = "1.1.1" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8" +checksum = "840bb056bd99c495c9188198a258ab787c79f306d864db3e6e8451442d6d8366" dependencies = [ "defmt-parser", "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.9", ] [[package]] name = "defmt-parser" -version = "1.0.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e" +checksum = "71958f69fb2de856c5dd51ac9377e986ca1aa5919f931bf5395a807277f889a0" dependencies = [ "thiserror 2.0.21", ] @@ -2234,9 +2234,9 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" [[package]] name = "pest" -version = "2.9.3" +version = "2.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b568374ba38b33a6c627141f891faf16902b08d2db26b8ede1bcb0a15b1919fa" +checksum = "fd47a460a9cbc3b3429c7e22bf0d76653e779319bae45a499ca010c9a990bb23" dependencies = [ "memchr", "psm", @@ -2246,9 +2246,9 @@ dependencies = [ [[package]] name = "pest_derive" -version = "2.9.3" +version = "2.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b66e184b924cebaaff20ab2256ca52f12332d528a39aa76553b5d96f92aacf7f" +checksum = "203e46697708f7519ab1608393cf89239cafb46b8e6d08a8ffe359359b11ef44" dependencies = [ "pest", "pest_generator", @@ -2256,9 +2256,9 @@ dependencies = [ [[package]] name = "pest_generator" -version = "2.9.3" +version = "2.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a87478d267e4de54a626af9754f2f0f58e927aac6ed0575fe89bc05ad6851694" +checksum = "e84ae6340a458d3d6cfc73cf328cf3cfe90b2484cd0de25eb66aa954615ed4d5" dependencies = [ "pest", "pest_meta", @@ -2269,9 +2269,9 @@ dependencies = [ [[package]] name = "pest_meta" -version = "2.9.3" +version = "2.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f986f248b4241ac359b831f6139aaa34e03b08a37b6caf7e201a33f95c869e1" +checksum = "b57f92f32b3dbce6598fd8f5cdf1f06a6072570b8103fa21728fa9036923e710" dependencies = [ "pest", ] diff --git a/Cargo.nix b/Cargo.nix index 5a3008a9..b118afef 100644 --- a/Cargo.nix +++ b/Cargo.nix @@ -1247,9 +1247,9 @@ rec { }; "combine" = rec { crateName = "combine"; - version = "4.6.8"; + version = "4.6.9"; edition = "2018"; - sha256 = "0ppwzwdmszpan9ybx1myc6ldg5zih2sazf9idckdxrh9gn9j1hyg"; + sha256 = "14izj50knw6rgl6ky3903k43pnl6qjgyw4d30n6ljdb4jykfnmj3"; authors = [ "Markus Westerlind " ]; @@ -1830,10 +1830,10 @@ rec { }; "defmt" = rec { crateName = "defmt"; - version = "1.1.1"; + version = "1.2.0"; edition = "2021"; links = "defmt"; - sha256 = "1lc8xlfj700xqjmvp7n9hhc1czgpaqkq960iqw6d5fwk9zz3p5g2"; + sha256 = "1qg3w0w7f0fn74xp60aci37swsjz6hbrqp3pngbm9js0qxd7k874"; authors = [ "The Knurling-rs developers" ]; @@ -1854,9 +1854,9 @@ rec { }; "defmt-macros" = rec { crateName = "defmt-macros"; - version = "1.1.1"; + version = "1.2.0"; edition = "2021"; - sha256 = "1s2zkcbaj1l306ph1n1gsfm6vzc2sah4acl1qc6pw4x2ghpcgnds"; + sha256 = "0rl3dlnl8lc4dqzdnr6q0vrpjz3qmdca5641334rbi4rpmbb02w4"; procMacro = true; libName = "defmt_macros"; authors = [ @@ -1877,7 +1877,7 @@ rec { } { name = "syn"; - packageId = "syn 2.0.119"; + packageId = "syn 3.0.9"; features = [ "full" "extra-traits" ]; } ]; @@ -1886,9 +1886,9 @@ rec { }; "defmt-parser" = rec { crateName = "defmt-parser"; - version = "1.0.0"; + version = "1.1.0"; edition = "2021"; - sha256 = "0gpfky9sssil5qfaix5wxcwiqk7snszhl5gq3vcwkrxjncs07mhh"; + sha256 = "1849z1vp502s77sip4wzj6jimjl6x5vr7b2ivp2mds1dzdlqz5bi"; libName = "defmt_parser"; authors = [ "The Knurling-rs developers" @@ -7417,9 +7417,9 @@ rec { }; "pest" = rec { crateName = "pest"; - version = "2.9.3"; + version = "2.9.4"; edition = "2021"; - sha256 = "1yhr35ds3c5ww7nvh9nvs842p40nmwgqj7ql4z3accwbld5kfs5m"; + sha256 = "08xvj2lwj450ki4mmr5s369pfgk5fq6vy8kyki1b7hybm5ha8izx"; authors = [ "Dragoș Tiselice " ]; @@ -7456,9 +7456,9 @@ rec { }; "pest_derive" = rec { crateName = "pest_derive"; - version = "2.9.3"; + version = "2.9.4"; edition = "2021"; - sha256 = "0zygma96zndmadjsg6m353ak48ziab55c8mb43zsmsscj95ihvmn"; + sha256 = "0i7g26dkang3zyl0hvcfdfsaz713i77r70v0n6d53xq8fxllcgi0"; procMacro = true; authors = [ "Dragoș Tiselice " @@ -7484,9 +7484,9 @@ rec { }; "pest_generator" = rec { crateName = "pest_generator"; - version = "2.9.3"; + version = "2.9.4"; edition = "2021"; - sha256 = "150nhpb5mh4vx1gmgl3fmix953pmy3r595xg4sk59pp4cz97hx58"; + sha256 = "1mflbrhm9abanrgf43fdhhj0psfgyf635kvkzin3v3a518sfcjp8"; authors = [ "Dragoș Tiselice " ]; @@ -7522,9 +7522,9 @@ rec { }; "pest_meta" = rec { crateName = "pest_meta"; - version = "2.9.3"; + version = "2.9.4"; edition = "2021"; - sha256 = "1qb9r2akz8q1wbvwmdipiaq06km3m8wn27w3kcssqha2icj6z62g"; + sha256 = "04774dlh7acgf8hzl0w11dbp4q3ay3qwvxfqixcydg1x5grr4zxm"; authors = [ "Dragoș Tiselice " ]; diff --git a/Cargo.toml b/Cargo.toml index 27dd6191..e420fd8b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,5 +30,8 @@ tracing = "0.1" yaml_serde = "0.10" [patch."https://github.com/stackabletech/operator-rs.git"] -# stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main"} +# 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 = "main" } # stackable-operator = { path = "../operator-rs/crates/stackable-operator" } diff --git a/deploy/helm/trino-operator/templates/clusterrole-operator.yaml b/deploy/helm/trino-operator/templates/clusterrole-operator.yaml index a8b08535..a6813bab 100644 --- a/deploy/helm/trino-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/trino-operator/templates/clusterrole-operator.yaml @@ -171,3 +171,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 00000000..f0cdd871 --- /dev/null +++ b/docs/modules/trino/pages/usage-guide/openlineage.adoc @@ -0,0 +1,149 @@ += 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 lineage events for the queries run on the cluster. +The `spec.clusterConfig.lineage` block in the TrinoCluster custom resource configures the OpenLineage event listener and its connection to a backend. + +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. + +[IMPORTANT] +==== +By default, `SELECT` queries do *not* emit any lineage events, so a cluster that only reads data produces no lineage at all. +To also capture reads, set `openlineage-event-listener.trino.include-query-types` as shown in <>. +==== + +== Enabling OpenLineage + +OpenLineage is configured through `spec.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: + http: + host: marquez + port: 5000 + # path: /api/v1/lineage # <4> + # tls: + # verification: + # server: + # caCert: + # secretClass: marquez-ca + # namespace: trino-lineage # <3> + # jobNameFormat: "$QUERY_ID" # <5> + # datasetNamespaceUri: https://trino-prod # <6> + ... +---- +<1> Adding the `lineage` block enables OpenLineage event emission. Omit it entirely to leave OpenLineage off (the default). +<2> Required. 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. +<3> Optional. The OpenLineage namespace events are reported under. Defaults to `default`. +<4> Optional. URL path of the endpoint events are POSTed to. Defaults to `/api/v1/lineage`. +<5> Optional. Sets the OpenLineage job name format. Accepts an arbitrary string with optional `$QUERY_ID`, `$USER`, `$SOURCE` and `$CLIENT_IP` substitution variables. 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}^]. +<6> Optional. Sets `openlineage-event-listener.trino.uri`, from which Trino derives the namespace of every dataset it reports. The URI is only an identifier and is never connected to. Defaults to `https://.` when empty or unset. Set it explicitly to keep dataset identities stable across Kubernetes clusters or namespace moves. + +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: + http: + host: marquez + port: 5000 +---- + +[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: + http: + 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 `openlineage-event-listener.transport.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.transport.timeout: "10s" +---- + +Event listeners run on the coordinator, so `event-listener.properties` overrides only take effect on the coordinator role. + +[#include-select-queries] +=== Emitting lineage for `SELECT` queries + +`openlineage-event-listener.trino.include-query-types` controls which query types emit lineage events. +It replaces the default list, so repeat the default types when adding `SELECT`: + +[source,yaml] +---- +spec: + coordinators: + configOverrides: + event-listener.properties: + openlineage-event-listener.trino.include-query-types: "SELECT,INSERT,DELETE,MERGE,UPDATE,ALTER_TABLE_EXECUTE,DATA_DEFINITION" +---- + +Every `SELECT` then produces a `START` and a `COMPLETE` (or `FAIL`) event, which can noticeably increase the load on the lineage backend for clusters serving many interactive or dashboard queries. diff --git a/docs/modules/trino/partials/nav.adoc b/docs/modules/trino/partials/nav.adoc index d90607af..35c80472 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 901063b6..24470ddc 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -889,6 +889,128 @@ 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/). + oneOf: + - required: + - http + properties: + http: + description: Publish events over HTTP(S) to an OpenLineage backend such as Marquez. + 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: 'Hostname or IP address of the OpenLineage backend without any protocol or port. For example: `marquez`.' + type: string + path: + default: /api/v1/lineage + description: URL path of the endpoint lineage events are sent to. + type: string + port: + description: 'Port the OpenLineage backend listens on. For example: `5000`.' + format: uint16 + 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 + type: object + reference: + type: string + type: object + datasetNamespaceUri: + default: '' + description: |- + URI identifying this Trino cluster in the emitted lineage. Trino replaces its scheme + with `trino` and uses the result as the namespace of every dataset (table) it reports, + e.g. `trino://my-trino`. It is never connected to, so it does not have to be reachable. + Defaults to `https://.` when empty or unset. + type: string + jobNameFormat: + default: $QUERY_ID + description: |- + Format of the OpenLineage job name emitted for each query. Accepts an arbitrary string + with optional `$QUERY_ID`, `$USER`, `$SOURCE` and `$CLIENT_IP` substitution variables. + type: string + namespace: + default: default + description: The OpenLineage namespace lineage is reported under. + type: string + required: + - connection + type: object tls: default: internalSecretClass: tls @@ -1447,6 +1569,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 @@ -2056,6 +2184,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 @@ -2764,6 +2898,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 @@ -3365,6 +3505,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 00000000..352a420f --- /dev/null +++ b/rust/operator-binary/src/config/lineage.rs @@ -0,0 +1,323 @@ +//! 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, +//! job name format 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`, +//! `OpenLineageConfig`) 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::OpenLineageTransport}, + k8s_openapi::api::core::v1::{SecretVolumeSource, Volume, VolumeMount}, +}; + +use crate::{ + controller::build::command, + crd::{OPENLINEAGE_AUTH_SECRET_KEY, STACKABLE_CLIENT_TLS_DIR, v1alpha1::TrinoLineageConfig}, +}; + +/// 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 endpoint path key, set from the connection's `path`. +pub const OPENLINEAGE_TRANSPORT_ENDPOINT_KEY: &str = + "openlineage-event-listener.transport.endpoint"; +/// 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.jobNameFormat`, which defaults to `$QUERY_ID`. +pub const OPENLINEAGE_JOB_NAME_FORMAT_KEY: &str = "openlineage-event-listener.job.name-format"; +/// The URI identifying this Trino cluster in emitted lineage. Trino derives the dataset namespace +/// from it. Set from `spec.clusterConfig.lineage.datasetNamespaceUri`, see +/// [`dataset_namespace_uri`] for the default. +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 the OpenLineage HTTP transport URL"))] + TransportUrl { + 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: &TrinoLineageConfig, + cluster_name: &str, + 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 + .open_lineage + .connection + .clone() + .resolve(client, namespace) + .await + .context(ResolveConnectionSnafu)?; + + let OpenLineageTransport::Http(http) = &connection.transport; + + 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(), + ); + // Trino takes the backend origin and the endpoint path as separate properties. + let transport_url = http.url().context(TransportUrlSnafu)?; + properties.insert( + OPENLINEAGE_TRANSPORT_URL_KEY.to_string(), + transport_url.origin().ascii_serialization(), + ); + properties.insert( + OPENLINEAGE_TRANSPORT_ENDPOINT_KEY.to_string(), + http.path.clone(), + ); + properties.insert( + OPENLINEAGE_NAMESPACE_KEY.to_string(), + lineage.open_lineage.namespace.clone(), + ); + properties.insert( + OPENLINEAGE_JOB_NAME_FORMAT_KEY.to_string(), + lineage.job_name_format.clone(), + ); + properties.insert( + OPENLINEAGE_TRINO_URI_KEY.to_string(), + dataset_namespace_uri(lineage, cluster_name, namespace), + ); + + // 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 there is no CA to trust). + let (tls_volumes, tls_mounts) = http + .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(&http.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) = &http.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(), + } +} + +/// The `trino.uri` Trino derives the dataset namespace from: the user-provided +/// `datasetNamespaceUri`, or `https://.` when it is empty (the CRD +/// default). +/// +/// The default includes the Kubernetes namespace so that equally named TrinoClusters in different +/// namespaces (e.g. `dev` and `prod`) do not report the same datasets. It is host-only and does not +/// depend on role groups or TLS, so it stays stable as long as the TrinoCluster is not renamed or +/// moved. +fn dataset_namespace_uri( + lineage: &TrinoLineageConfig, + cluster_name: &str, + namespace: &str, +) -> String { + if lineage.dataset_namespace_uri.is_empty() { + format!("https://{cluster_name}.{namespace}") + } else { + lineage.dataset_namespace_uri.clone() + } +} + +#[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 }), + } + } + + fn lineage_config(yaml: &str) -> TrinoLineageConfig { + yaml_serde::from_str(yaml).expect("illegal test input") + } + + #[test] + fn dataset_namespace_uri_defaults_to_cluster_name_and_namespace() { + let lineage = lineage_config("connection: {reference: marquez}"); + assert_eq!( + dataset_namespace_uri(&lineage, "simple-trino", "prod"), + "https://simple-trino.prod" + ); + } + + #[test] + fn empty_dataset_namespace_uri_falls_back_to_default() { + let lineage = lineage_config("{connection: {reference: marquez}, datasetNamespaceUri: ''}"); + assert_eq!( + dataset_namespace_uri(&lineage, "simple-trino", "prod"), + "https://simple-trino.prod" + ); + } + + #[test] + fn dataset_namespace_uri_uses_user_value() { + let lineage = lineage_config( + "{connection: {reference: marquez}, datasetNamespaceUri: 'https://trino-prod:443'}", + ); + assert_eq!( + dataset_namespace_uri(&lineage, "simple-trino", "prod"), + "https://trino-prod:443" + ); + } + + #[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() { + // `verification.none` still uses https, but there is no CA to import. + 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 152d5080..f3a70f9d 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 6484e6d1..79a81a83 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 e2b6e3ed..9ce08bf5 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#" @@ -394,6 +413,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 00000000..77fc98b8 --- /dev/null +++ b/rust/operator-binary/src/controller/build/properties/event_listener_properties.rs @@ -0,0 +1,169 @@ +//! 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 settings resolved in [`crate::config::lineage`] and merges any user +//! `event-listener.properties` overrides (highest precedence). + +use std::collections::BTreeMap; + +use crate::{ + controller::{TrinoRoleGroupConfig, ValidatedCluster}, + crd::TrinoRole, +}; + +/// 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, +) -> 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 { + props.extend(lineage.properties.clone()); + } + + // User overrides (highest precedence). + props.extend(rg.config_overrides.event_listener_properties.clone()); + + props +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + 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, + OPENLINEAGE_TRINO_URI_KEY, ResolvedLineageConfig, + }, + controller::{ + ValidatedCluster, + build::properties::test_support::{MINIMAL_TRINO_YAML, empty_derefs}, + }, + crd::TrinoRole, + }; + + /// 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()), + ( + OPENLINEAGE_TRINO_URI_KEY.to_string(), + "https://simple-trino.dev".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); + 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); + 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); + + 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"); + assert_eq!( + props.get(OPENLINEAGE_TRINO_URI_KEY).unwrap(), + "https://simple-trino.dev" + ); + // 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); + + 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 9cb93013..855680a4 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 1a2863a5..c77e6a71 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -17,8 +17,8 @@ use crate::{ object_meta, 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, }, recommended_labels_for_role_group_resources, }, @@ -157,7 +157,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); + 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.numeric_product_version, @@ -167,7 +178,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 6cd36c48..a12bc55f 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -312,7 +312,13 @@ pub fn build_rolegroup_statefulset( &mut cb_trino, ) .context(InvalidAuthenticationConfigSnafu)?; - add_derived_volumes_and_mounts(cluster, &mut pod_builder, &mut cb_prepare, &mut cb_trino)?; + add_derived_volumes_and_mounts( + cluster, + trino_role, + &mut pod_builder, + &mut cb_prepare, + &mut cb_trino, + )?; let mut prepare_args = vec![]; if let ValidatedContainerLogConfigChoice::Automatic(log_config) = @@ -347,6 +353,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![ @@ -742,10 +755,11 @@ fn add_tls_volumes_and_mounts( } /// Adds the volumes and volume mounts whose names derive from user-supplied resources: the -/// catalogs, fault-tolerant execution and client spooling. They may collide with each other or -/// with the operator's own volumes, so the adds stay fallible. +/// catalogs, fault-tolerant execution, client spooling and OpenLineage. They may collide with each +/// other or with the operator's own volumes, so the adds stay fallible. fn add_derived_volumes_and_mounts( cluster: &ValidatedCluster, + trino_role: &TrinoRole, pod_builder: &mut PodBuilder, cb_prepare: &mut ContainerBuilder, cb_trino: &mut ContainerBuilder, @@ -788,6 +802,22 @@ fn add_derived_volumes_and_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 fe7af645..8b6d9f4c 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -8,7 +8,10 @@ use std::str::FromStr; use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ client::Client, - kube::runtime::reflector::{Lookup, ObjectRef}, + kube::{ + ResourceExt, + runtime::reflector::{Lookup, ObjectRef}, + }, v2::controller_utils::get_namespace, }; @@ -18,6 +21,7 @@ use crate::{ config::{ client_protocol::{self, ResolvedClientProtocolConfig}, fault_tolerant_execution::{self, ResolvedFaultTolerantExecutionConfig}, + lineage::{self, ResolvedLineageConfig}, }, crd::{ authentication::{ResolvedAuthenticationClassRef, resolve_authentication_classes}, @@ -60,6 +64,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, @@ -82,6 +89,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. @@ -155,6 +163,20 @@ 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, + &trino.name_any(), + client, + namespace.as_ref(), + ) + .await + .context(OpenLineageConfigurationSnafu)?, + ), + None => None, + }; + Ok(DereferencedObjects { resolved_authentication_classes, catalog_definitions, @@ -162,6 +184,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 27505faf..681fc462 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -34,6 +34,7 @@ use crate::{ config::{ client_protocol::ResolvedClientProtocolConfig, fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig, + lineage::ResolvedLineageConfig, }, crd::{TrinoRole, catalog::TrinoCatalogName, discovery::TrinoPodRef, v1alpha1}, trino_controller::PRODUCT_NAME, @@ -116,6 +117,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, } @@ -472,6 +474,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 c2cec4b9..b1518646 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -268,6 +268,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, }; @@ -386,6 +387,7 @@ pub(crate) fn merged_role_group_config( trino_opa_config: None, resolved_fte_config: None, resolved_client_protocol_config: None, + resolved_lineage_config: None, }; let role_groups = match trino_role { TrinoRole::Coordinator => { @@ -427,6 +429,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 188d36ad..f00cd6d0 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}, constant, - crd::authentication::core, + crd::{authentication::core, openlineage}, deep_merger::ObjectOverrides, k8s_openapi::apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::LabelSelector}, kube::{CustomResource, ResourceExt}, @@ -93,6 +93,10 @@ constant!(pub ENV_INTERNAL_SECRET: EnvVarName = "INTERNAL_SECRET"); constant!(pub INTERNAL_SECRET_SECRET_KEY: SecretKey = "INTERNAL_SECRET"); constant!(pub ENV_SPOOLING_SECRET: EnvVarName = "SPOOLING_SECRET"); constant!(pub SPOOLING_SECRET_SECRET_KEY: SecretKey = "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 constant!(TLS_DEFAULT_SECRET_CLASS: SecretClassName = "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) @@ -293,6 +306,26 @@ pub mod versioned { pub vector_aggregator_config_map_name: Option, } + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] + #[serde(rename_all = "camelCase")] + pub struct TrinoLineageConfig { + // no doc - it's in the struct. + #[serde(flatten)] + pub open_lineage: openlineage::v1alpha1::OpenLineageConfig, + + /// Format of the OpenLineage job name emitted for each query. Accepts an arbitrary string + /// with optional `$QUERY_ID`, `$USER`, `$SOURCE` and `$CLIENT_IP` substitution variables. + #[serde(default = "TrinoLineageConfig::default_job_name_format")] + pub job_name_format: String, + + /// URI identifying this Trino cluster in the emitted lineage. Trino replaces its scheme + /// with `trino` and uses the result as the namespace of every dataset (table) it reports, + /// e.g. `trino://my-trino`. It is never connected to, so it does not have to be reachable. + /// Defaults to `https://.` when empty or unset. + #[serde(default)] + pub dataset_namespace_uri: String, + } + #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub enum TrinoAuthorization { @@ -362,6 +395,12 @@ pub mod versioned { } } +impl v1alpha1::TrinoLineageConfig { + pub fn default_job_name_format() -> String { + "$QUERY_ID".to_string() + } +} + impl v1alpha1::TrinoAuthorizationOpaConfig { pub fn enabled_column_masking_default() -> bool { true @@ -858,6 +897,55 @@ mod tests { assert_eq!(server_secret_class(&trino), Some("simple-trino-server-tls")); } + #[test] + fn lineage_config_is_flattened() { + let input = r#" + apiVersion: trino.stackable.tech/v1alpha1 + kind: TrinoCluster + metadata: + name: simple-trino + spec: + image: + productVersion: "481" + coordinators: + roleGroups: + default: + replicas: 1 + workers: + roleGroups: + default: + replicas: 1 + clusterConfig: + catalogLabelSelector: {} + lineage: + connection: + reference: marquez + namespace: trino-lineage + jobNameFormat: trino-$QUERY_ID + datasetNamespaceUri: https://trino-prod + "#; + let trino: v1alpha1::TrinoCluster = + yaml_serde::from_str(input).expect("illegal test input"); + let lineage = trino + .spec + .cluster_config + .lineage + .expect("lineage is configured"); + assert_eq!(lineage.open_lineage.namespace, "trino-lineage"); + assert_eq!(lineage.job_name_format, "trino-$QUERY_ID"); + assert_eq!(lineage.dataset_namespace_uri, "https://trino-prod"); + } + + #[test] + fn lineage_job_name_format_defaults_to_query_id() { + let lineage: v1alpha1::TrinoLineageConfig = yaml_serde::from_str(indoc::indoc! {" + connection: + reference: marquez + "}) + .expect("illegal test input"); + assert_eq!(lineage.job_name_format, "$QUERY_ID"); + } + impl RoundtripTestData for v1alpha1::TrinoClusterSpec { fn roundtrip_test_data() -> Vec { stackable_operator::utils::yaml_from_str_singleton_map(indoc::indoc! {r#" @@ -901,6 +989,12 @@ mod tests { - s3://exchange-bucket/ connection: reference: minio + lineage: + connection: + reference: marquez + namespace: trino-lineage + jobNameFormat: trino-$QUERY_ID + datasetNamespaceUri: https://trino-prod vectorAggregatorConfigMapName: vector-aggregator-discovery coordinators: config: diff --git a/rust/operator-binary/src/trino_controller.rs b/rust/operator-binary/src/trino_controller.rs index 7a202f8b..5f0cfe6f 100644 --- a/rust/operator-binary/src/trino_controller.rs +++ b/rust/operator-binary/src/trino_controller.rs @@ -246,6 +246,7 @@ mod tests { trino_opa_config, resolved_fte_config, resolved_client_protocol_config, + resolved_lineage_config: None, }; let operator_env = OperatorEnvironmentOptions { @@ -474,6 +475,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 00000000..2d6a0df5 --- /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 00000000..67185acf --- /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 00000000..9cbf0351 --- /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 00000000..8f5f9b3e --- /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 00000000..8ec0cc6b --- /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 00000000..2bdb3ccd --- /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 00000000..da1a5508 --- /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 00000000..ae1b71ce --- /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 00000000..32cef43a --- /dev/null +++ b/tests/templates/kuttl/openlineage/02_trino.yaml.j2 @@ -0,0 +1,92 @@ +{% 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: + http: + # 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). + jobNameFormat: "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 00000000..4709ac4f --- /dev/null +++ b/tests/templates/kuttl/openlineage/03-assert.yaml @@ -0,0 +1,31 @@ +--- +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 "openlineage-event-listener.transport.endpoint=/api/v1/lineage" + echo "$CM" | grep -q "event-listener.config-files=/stackable/rwconfig/event-listener.properties" + # jobNameFormat maps to the OpenLineage job name format. + echo "$CM" | grep -q "openlineage-event-listener.job.name-format=trino-" + # datasetNamespaceUri defaults to https://.. Java properties escape `:`, + # so drop backslashes before matching. + echo "$CM" | tr -d '\\' | grep -q "openlineage-event-listener.trino.uri=https://trino-openlineage.$NAMESPACE" + # 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 00000000..ff0d29e6 --- /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 00000000..4d4b4d9f --- /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 00000000..b6bd3a44 --- /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 00000000..2506144c --- /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 10eaba2f..bf25953d 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -71,6 +71,14 @@ dimensions: values: - "false" - "true" + - name: openlineage-use-tls + values: + - "false" + - "true" + - name: openlineage-use-auth + values: + - "false" + - "true" - name: openshift values: - "false" @@ -128,6 +136,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