fix integration test

Author: razvan

Cargo.nix

@@ -15,6 +15,8 @@ When configured, the operator:
 
 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.openLineage.connection`, which either inlines an OpenLineage connection or references a standalone `OpenLineageConnection` resource by name:
@@ -63,6 +65,8 @@ spec:
 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.<namespace>.svc.cluster.local`), otherwise the TLS handshake fails and no events are delivered.
+
 [#authentication]
 == Authentication
 

crate-hashes.json

@@ -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.open_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::open_lineage::ResolvedOpenLineageConfig,
+        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_open_lineage() -> ValidatedCluster {
+        let mut derefs = empty_derefs();
+        derefs.resolved_open_lineage_config = Some(ResolvedOpenLineageConfig {
+            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_open_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_open_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_open_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);

docs/modules/trino/pages/usage-guide/openlineage.adoc

@@ -27,11 +27,11 @@ metadata:
   name: trino-openlineage
 spec:
   image:
-{% if test_scenario['values']['trino'].find(",") > 0 %}
-    custom: "{{ test_scenario['values']['trino'].split(',')[1] }}"
-    productVersion: "{{ test_scenario['values']['trino'].split(',')[0] }}"
+{% 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'] }}"
+    productVersion: "{{ test_scenario['values']['trino-latest'] }}"
 {% endif %}
     pullPolicy: IfNotPresent
   clusterConfig:
@@ -41,7 +41,9 @@ spec:
     openLineage:
       connection:
         inline:
-          host: openlineage-receiver
+          # Use the full service FQDN: with TLS the backend's secret-operator cert only carries the
+          # `openlineage-receiver.<ns>.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:
@@ -83,3 +85,16 @@ metadata:
 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: {}

rust/operator-binary/src/controller/build/properties/config_properties.rs

@@ -4,12 +4,14 @@ kind: TestAssert
 timeout: 300
 commands:
   # The coordinator ConfigMap must carry a generated event-listener.properties selecting the
-  # OpenLineage plugin and pointing at the receiver.
+  # 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"
   # 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"

tests/templates/kuttl/openlineage/02_trino.yaml.j2

@@ -1,5 +1,10 @@
 #!/usr/bin/env python
-"""Run a query against Trino to make the OpenLineage event listener emit lineage events."""
+"""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
 
@@ -29,9 +34,14 @@ def get_connection(coordinator):
 
     conn = get_connection(args["coordinator"])
     cursor = conn.cursor()
-    # Reads the built-in tpch dataset; the OpenLineage listener reports `tpch.tiny.nation` as an
-    # input dataset in the emitted (START + COMPLETE) query events.
-    cursor.execute("SELECT COUNT(*) FROM tpch.tiny.nation")
+
+    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 tpch.tiny.nation row count: {result[0]}"
-    print("query executed, OpenLineage events should have been emitted")
+    assert result[0] == 25, f"unexpected CTAS row count: {result[0]}"
+    print("CTAS executed, OpenLineage events should have been emitted")

tests/templates/kuttl/openlineage/03-assert.yaml

@@ -9,7 +9,7 @@ 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 queried tpch dataset ...
-      kubectl exec -n "$NAMESPACE" "$POD" -- sh -c 'cat /tmp/lineage/* 2>/dev/null | grep -qi "nation"'
+      # 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"'