test: Adapt the integration tests to Vector 0.58.0

CHANGELOG.md

@@ -30,7 +30,7 @@ All notable changes to this project will be documented in this file.
   to ensure only one voter is added/removed at a time and thus keep the quorum healthy ([#1010]).
 - Internal operator refactoring: introduce a build() step in the reconciler that
   assembles all relevant Kubernetes resources before anything is applied ([#985]).
-- Bump stackable-operator to 0.118.0 ([#994], [#1011], [#1021]).
+- Bump stackable-operator to 0.121.0 ([#994], [#1011], [#1021], [#1029], [#1043]).
 - The RBAC ServiceAccount and RoleBinding are now built with the operator-rs `v2::rbac`
   functions and carry the full set of recommended labels ([#990]).
 - BREAKING: The `brokers` role is now required by the CRD; a KafkaCluster without it was
@@ -48,10 +48,9 @@ All notable changes to this project will be documented in this file.
   operator upgrade, delete each broker StatefulSet so that the operator immediately recreates it
   with the new labels ([#1011]).
 - Make operations infallible where dependent on static inputs ([#1017]).
-- Bump stackable-operator to 0.119.0 ([#1029]).
-- Bump stackable-operator to 0.121.0 and remove `serde_yaml` ([#1043]).
+- Remove `serde_yaml` ([#1043]).
 - All pods now set `securityContext.seccompProfile.type` to `RuntimeDefault` ([#1043]).
-- Support Vector 0.58.0 ([#1039], [#1044]).
+- Support Vector 0.58.0 ([#1039], [#1044], [#1046]).
 - test: Update OPA to 1.20.2 in integration tests and docs example ([#1042]).
 
 ### Fixed
@@ -92,6 +91,7 @@ All notable changes to this project will be documented in this file.
 [#1042]: https://github.com/stackabletech/kafka-operator/pull/1042
 [#1043]: https://github.com/stackabletech/kafka-operator/pull/1043
 [#1044]: https://github.com/stackabletech/kafka-operator/pull/1044
+[#1046]: https://github.com/stackabletech/kafka-operator/pull/1046
 
 ## [26.7.0] - 2026-07-21
 

tests/templates/kuttl/logging/kafka-vector-aggregator-values.yaml.j2

@@ -64,7 +64,9 @@ customConfig:
     aggregator:
       inputs: [vector]
       type: vector
-      address: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+      routing:
+        endpoints:
+          - {{ lookup('env', 'VECTOR_AGGREGATOR') }}
       buffer:
         # Avoid back pressure from VECTOR_AGGREGATOR. The test should
         # not fail if the aggregator is not available.

tests/templates/kuttl/opa/vector-aggregator-values.yaml.j2

@@ -72,7 +72,9 @@ customConfig:
     aggregator:
       inputs: [vector]
       type: vector
-      address: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+      routing:
+        endpoints:
+          - {{ lookup('env', 'VECTOR_AGGREGATOR') }}
       buffer:
         # Avoid back pressure from VECTOR_AGGREGATOR. The test should
         # not fail if the aggregator is not available.

tests/templates/kuttl/smoke-kraft/kafka-vector-aggregator-values.yaml.j2

@@ -96,7 +96,9 @@ customConfig:
     aggregator:
       inputs: [vector]
       type: vector
-      address: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+      routing:
+        endpoints:
+          - {{ lookup('env', 'VECTOR_AGGREGATOR') }}
       buffer:
         # Avoid back pressure from VECTOR_AGGREGATOR. The test should
         # not fail if the aggregator is not available.

tests/templates/kuttl/smoke-kraft/test_log_aggregation.py

@@ -1,45 +1,40 @@
-#!/usr/bin/env python3
-import requests
+import json
+import subprocess
 
 
 def check_sent_events():
-    response = requests.post(
-        "http://kafka-vector-aggregator:8686/graphql",
-        json={
-            "query": """
-                {
-                    transforms(first:100) {
-                        nodes {
-                            componentId
-                            metrics {
-                                sentEventsTotal {
-                                    sentEventsTotal
-                                }
-                            }
-                        }
-                    }
-                }
-            """
-        },
+    response = subprocess.run(
+        [
+            "grpcurl",
+            "-plaintext",
+            "-d",
+            '{"limit": 100}',
+            "kafka-vector-aggregator:8686",
+            "vector.observability.v1.ObservabilityService/GetComponents",
+        ],
+        capture_output=True,
+        text=True,
+        check=True,  # Raise a CalledProcessError if non-zero return
+        timeout=20,  # seconds
     )
+    result = json.loads(response.stdout)
+    components = result.get("components", [])
+    transforms = [
+        c for c in components if c.get("componentType") == "COMPONENT_TYPE_TRANSFORM"
+    ]
 
-    assert response.status_code == 200, (
-        "Cannot access the API of the vector aggregator."
-    )
-
-    result = response.json()
+    assert len(transforms) > 0, "No transform components found"
 
-    transforms = result["data"]["transforms"]["nodes"]
     for transform in transforms:
         sentEvents = transform["metrics"]["sentEventsTotal"]
         componentId = transform["componentId"]
 
         if componentId == "filteredInvalidEvents":
-            assert sentEvents is None or sentEvents["sentEventsTotal"] == 0, (
+            assert sentEvents is None or int(sentEvents) == 0, (
                 "Invalid log events were sent."
             )
         else:
-            assert sentEvents is not None and sentEvents["sentEventsTotal"] > 0, (
+            assert sentEvents is not None and int(sentEvents) > 0, (
                 f'No events were sent in "{componentId}".'
             )