Skip to content

Commit 5291554

Browse files
maltesandermaltesander
committed
refactor: adapt logging to v2, add vector test
1 parent a7a7686 commit 5291554

12 files changed

Lines changed: 681 additions & 485 deletions

File tree

‎rust/operator-binary/src/controller.rs‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -347,11 +347,6 @@ pub struct ValidatedClusterConfig {
347347
/// The `ConfigMap` mapping pods to broker ids, if the user supplied one. Resolved from the
348348
/// raw spec during validation so the build steps never have to read it.
349349
pub broker_id_pod_config_map_name: Option<String>,
350-
351-
/// The discovery `ConfigMap` providing the Vector aggregator address, if Vector log
352-
/// aggregation is configured. Resolved from the raw spec during validation so the build
353-
/// steps never have to read it.
354-
pub vector_aggregator_config_map_name: Option<String>,
355350
}
356351

357352
impl ValidatedClusterConfig {
@@ -422,6 +417,8 @@ pub struct ValidatedRoleGroupConfig {
422417
pub pod_overrides: stackable_operator::k8s_openapi::api::core::v1::PodTemplateSpec,
423418
pub jvm_argument_overrides:
424419
stackable_operator::v2::jvm_argument_overrides::JvmArgumentOverrides,
420+
/// Validated logging configuration (derived from `config.logging` during validation).
421+
pub logging: validate::ValidatedLogging,
425422
}
426423

427424
pub struct Ctx {
@@ -621,14 +618,17 @@ pub async fn reconcile_kafka(
621618

622619
for (kafka_role, rg_map) in &validated_cluster.role_group_configs {
623620
for (rolegroup_name, validated_rg) in rg_map {
624-
// The Vector log-aggregation config still consumes a v1 `RoleGroupRef`; it is built
625-
// here and used only for that. All other identification uses the typed `kafka_role` /
626-
// `rolegroup_name` (and `ValidatedCluster::resource_names`).
621+
// `rolegroup_ref` is a v1 `RoleGroupRef` retained only for the error context of the
622+
// per-rolegroup apply calls below. All other identification uses the typed
623+
// `kafka_role` / `rolegroup_name` (and `ValidatedCluster::resource_names`).
627624
let rolegroup_ref = kafka.rolegroup_ref(kafka_role, rolegroup_name.to_string());
628-
let vector_config = build::properties::logging::build_vector_config(
629-
&rolegroup_ref,
630-
&validated_rg.config,
631-
);
625+
// The Vector agent config is the static `vector.yaml`, added to the rolegroup
626+
// ConfigMap only when the Vector agent is enabled (resolved during validation).
627+
let vector_config = validated_rg
628+
.logging
629+
.vector_container
630+
.is_some()
631+
.then(build::properties::product_logging::vector_config_file_content);
632632

633633
let rg_headless_service = build_rolegroup_headless_service(
634634
&validated_cluster,

‎rust/operator-binary/src/controller/build/properties/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
pub mod broker_properties;
44
pub mod controller_properties;
55
pub mod listener;
6-
pub mod logging;
6+
pub mod product_logging;
77
pub mod security_properties;
88

99
use crate::crd::{

rust/operator-binary/src/controller/build/properties/logging.rs renamed to rust/operator-binary/src/controller/build/properties/product_logging/mod.rs

Lines changed: 30 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,12 @@ use stackable_operator::{
99
self,
1010
spec::{ContainerLogConfig, ContainerLogConfigChoice},
1111
},
12-
role_utils::RoleGroupRef,
1312
};
1413

1514
use super::ConfigFileName;
1615
use crate::crd::{
1716
STACKABLE_LOG_DIR,
1817
role::{AnyConfig, broker::BrokerContainer, controller::ControllerContainer},
19-
v1alpha1,
2018
};
2119

2220
/// The maximum size of a single Kafka log file before it is rotated.
@@ -28,6 +26,18 @@ pub const MAX_KAFKA_LOG_FILES_SIZE: MemoryQuantity = MemoryQuantity {
2826
const KAFKA_LOG4J_FILE: &str = "kafka.log4j.xml";
2927
const KAFKA_LOG4J2_FILE: &str = "kafka.log4j2.xml";
3028

29+
/// The static, env-driven Vector agent configuration (`vector.yaml`).
30+
///
31+
/// The v2 [`vector_container`](stackable_operator::v2::product_logging::framework::vector_container)
32+
/// mounts this file and supplies its `${...}` values (`LOG_DIR`, `DATA_DIR`, `NAMESPACE`,
33+
/// `CLUSTER_NAME`, `ROLE_NAME`, `ROLE_GROUP_NAME`, `VECTOR_AGGREGATOR_ADDRESS`) as env vars.
34+
const VECTOR_CONFIG: &str = include_str!("vector.yaml");
35+
36+
/// Returns the Vector agent config (`vector.yaml`) content.
37+
pub fn vector_config_file_content() -> String {
38+
VECTOR_CONFIG.to_owned()
39+
}
40+
3141
const CONSOLE_CONVERSION_PATTERN_LOG4J: &str = "[%d] %p %m (%c)%n";
3242
const CONSOLE_CONVERSION_PATTERN_LOG4J2: &str = "%d{ISO8601} %p [%t] %c - %m%n";
3343

@@ -75,29 +85,6 @@ pub fn role_group_config_map_data(
7585
configs
7686
}
7787

78-
/// Builds the Vector agent config for a role group, or `None` when the Vector agent is disabled.
79-
///
80-
/// Takes a v1 [`RoleGroupRef`] because the upstream `create_vector_config` still requires one;
81-
/// this is the only remaining consumer of `RoleGroupRef` in the operator.
82-
pub fn build_vector_config(
83-
rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
84-
merged_config: &AnyConfig,
85-
) -> Option<String> {
86-
let vector_log_config = merged_config.vector_logging();
87-
let vector_log_config = if let ContainerLogConfig {
88-
choice: Some(ContainerLogConfigChoice::Automatic(log_config)),
89-
} = &*vector_log_config
90-
{
91-
Some(log_config)
92-
} else {
93-
None
94-
};
95-
96-
merged_config
97-
.vector_logging_enabled()
98-
.then(|| product_logging::framework::create_vector_config(rolegroup, vector_log_config))
99-
}
100-
10188
fn log4j_config_if_automatic(
10289
log_config: Option<Cow<ContainerLogConfig>>,
10390
container_name: impl Display,
@@ -147,3 +134,21 @@ fn log4j2_config_if_automatic(
147134
None
148135
}
149136
}
137+
138+
#[cfg(test)]
139+
mod tests {
140+
use super::*;
141+
142+
#[test]
143+
fn test_vector_config_file_content() {
144+
let content = vector_config_file_content();
145+
assert!(!content.is_empty());
146+
// The two Kafka log formats must be handled ...
147+
assert!(content.contains("files_log4j"));
148+
assert!(content.contains("files_log4j2"));
149+
// ... while the non-Kafka sources were removed.
150+
assert!(!content.contains("files_stdout"));
151+
assert!(!content.contains("files_tracing_rs"));
152+
assert!(!content.contains("files_opa_json"));
153+
}
154+
}

‎rust/operator-binary/src/controller/build/properties/product_logging/test-vector.sh‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
#!/usr/bin/env sh
2+
3+
DATA_DIR=/stackable/log/_vector-state \
4+
LOG_DIR=/stackable/log \
5+
NAMESPACE=default \
6+
CLUSTER_NAME=kafka \
7+
ROLE_NAME=broker \
8+
ROLE_GROUP_NAME=default \
9+
VECTOR_AGGREGATOR_ADDRESS=vector-aggregator \
10+
VECTOR_FILE_LOG_LEVEL=info \
11+
vector test vector.yaml vector-test.yaml

‎rust/operator-binary/src/controller/build/properties/product_logging/vector-test.yaml‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
# Run tests with `./test-vector.sh`
2+
#
3+
# A downside of these test cases is that they compare the whole event and that the message can
4+
# contain source code positions in vector.yaml, e.g. "function call error for \"parse_xml\" at
5+
# (584:643)". Please adapt the tests if you change VRL code in vector.yaml.
6+
---
7+
tests:
8+
- name: Test log4j XML log entry without throwable
9+
inputs:
10+
- type: log
11+
insert_at: processed_files_log4j
12+
log_fields:
13+
file: /stackable/log/kafka/kafka.log4j.xml
14+
message: >
15+
<log4j:event logger="kafka.server.KafkaServer"
16+
timestamp="1759397248582" level="INFO"
17+
thread="main"><log4j:message>started
18+
(kafka.server.KafkaServer)</log4j:message></log4j:event>
19+
pod: kafka-broker-default-0
20+
source_type: file
21+
timestamp: 2025-10-02T09:27:29.473487331Z
22+
outputs:
23+
- extract_from: extended_logs
24+
conditions:
25+
- type: vrl
26+
source: |
27+
expected_log_event = {
28+
"cluster": "kafka",
29+
"container": "kafka",
30+
"file": "kafka.log4j.xml",
31+
"level": "INFO",
32+
"logger": "kafka.server.KafkaServer",
33+
"message": "started (kafka.server.KafkaServer)",
34+
"namespace": "default",
35+
"pod": "kafka-broker-default-0",
36+
"role": "broker",
37+
"roleGroup": "default",
38+
"timestamp": t'2025-10-02T09:27:28.582Z'
39+
}
40+
41+
assert_eq!(expected_log_event, .)
42+
- name: Test log4j2 XML log entry without stacktrace
43+
inputs:
44+
- type: log
45+
insert_at: processed_files_log4j2
46+
log_fields:
47+
file: /stackable/log/kafka/kafka.log4j2.xml
48+
message: >
49+
<Event xmlns="http://logging.apache.org/log4j/2.0/events"
50+
timeMillis="1759397248582" level="INFO"
51+
loggerName="kafka.server.KafkaServer"
52+
thread="main"><Message>started
53+
(kafka.server.KafkaServer)</Message></Event>
54+
pod: kafka-broker-default-0
55+
source_type: file
56+
timestamp: 2025-10-02T09:27:29.473487331Z
57+
outputs:
58+
- extract_from: extended_logs
59+
conditions:
60+
- type: vrl
61+
source: |
62+
expected_log_event = {
63+
"cluster": "kafka",
64+
"container": "kafka",
65+
"file": "kafka.log4j2.xml",
66+
"level": "INFO",
67+
"logger": "kafka.server.KafkaServer",
68+
"message": "started (kafka.server.KafkaServer)",
69+
"namespace": "default",
70+
"pod": "kafka-broker-default-0",
71+
"role": "broker",
72+
"roleGroup": "default",
73+
"timestamp": t'2025-10-02T09:27:28.582Z'
74+
}
75+
76+
assert_eq!(expected_log_event, .)
77+
- name: Test Vector internal logs
78+
inputs:
79+
- type: log
80+
insert_at: filtered_logs_vector
81+
log_fields:
82+
arch: x86_64
83+
message: Vector has started.
84+
metadata:
85+
kind: event
86+
level: INFO
87+
module_path: vector::internal_events::process
88+
target: vector
89+
pid: 14
90+
pod: kafka-broker-default-0
91+
source_type: internal_logs
92+
timestamp: 2025-10-02T09:46:14.479381097Z
93+
version: 0.49.0
94+
outputs:
95+
- extract_from: extended_logs
96+
conditions:
97+
- type: vrl
98+
source: |
99+
expected_log_event = {
100+
"arch": "x86_64",
101+
"cluster": "kafka",
102+
"container": "vector",
103+
"level": "INFO",
104+
"logger": "vector::internal_events::process",
105+
"message": "Vector has started.",
106+
"namespace": "default",
107+
"pod": "kafka-broker-default-0",
108+
"role": "broker",
109+
"roleGroup": "default",
110+
"timestamp": "2025-10-02T09:46:14.479381097Z",
111+
"version": "0.49.0"
112+
}
113+
114+
assert_eq!(expected_log_event, .)
115+
- name: Test Vector internal log level filtering - INFO passes
116+
inputs:
117+
- type: log
118+
insert_at: filtered_logs_vector
119+
log_fields:
120+
metadata:
121+
level: INFO
122+
outputs:
123+
- extract_from: filtered_logs_vector
124+
conditions:
125+
- type: vrl
126+
source: |
127+
assert_eq!("INFO", .metadata.level)
128+
- name: Test Vector internal log level filtering - DEBUG dropped
129+
inputs:
130+
- type: log
131+
insert_at: filtered_logs_vector
132+
log_fields:
133+
metadata:
134+
level: DEBUG
135+
no_outputs_from:
136+
- filtered_logs_vector

0 commit comments

Comments
 (0)