Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,17 @@ All notable changes to this project will be documented in this file.
- Deprecate support for Kafka `3.7.2` ([#892]).
- BREAKING: The `<cluster>-<role>-<rolegroup>` rolegroup service was replaced with a `<cluster>-<role>-<rolegroup>-headless`
and `<cluster>-<role>-<rolegroup>-metrics` rolegroup service ([#897]).
- Small cleanups and updates ([#900])
- remove the metrics port from services that don't need it
- use the new `server.yaml` for jmx configuration
- update metrics tests
- update monitoring doc

[#889]: https://github.com/stackabletech/kafka-operator/pull/889
[#890]: https://github.com/stackabletech/kafka-operator/pull/890
[#892]: https://github.com/stackabletech/kafka-operator/pull/892
[#897]: https://github.com/stackabletech/kafka-operator/pull/897
[#900]: https://github.com/stackabletech/kafka-operator/pull/900

## [25.7.0] - 2025-07-23

Expand Down
4 changes: 3 additions & 1 deletion docs/modules/kafka/pages/usage-guide/monitoring.adoc
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
= Monitoring
:description: The managed Kafka instances are automatically configured to export Prometheus metrics.

The managed Kafka instances are automatically configured to export Prometheus metrics.
The operator sets up all Kafka server instances with a JMX exporter agent.
The operator also sets up a dedicated metrics service for each role group.
The name of this service follows the schema `<stacklet name>-<role name>-<group name>-metrics`.
See xref:operators:monitoring.adoc[] for more details.
6 changes: 3 additions & 3 deletions rust/operator-binary/src/config/jvm.rs
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ fn construct_jvm_args<ConfigFragment>(
format!("-Xms{java_heap}"),
format!("-Djava.security.properties={STACKABLE_CONFIG_DIR}/{JVM_SECURITY_PROPERTIES_FILE}"),
format!(
"-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/broker.yaml"
"-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/server.yaml"
),
];

Expand Down Expand Up @@ -130,7 +130,7 @@ mod tests {
assert_eq!(
non_heap_jvm_args,
"-Djava.security.properties=/stackable/config/security.properties \
-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/broker.yaml"
-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/server.yaml"
);
assert_eq!(heap_jvm_args, "-Xmx1638m -Xms1638m");
}
Expand Down Expand Up @@ -177,7 +177,7 @@ mod tests {
assert_eq!(
non_heap_jvm_args,
"-Djava.security.properties=/stackable/config/security.properties \
-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/broker.yaml \
-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/server.yaml \
-Dhttps.proxyHost=proxy.my.corp \
-Djava.net.preferIPv4Stack=true \
-Dhttps.proxyPort=1234"
Expand Down
24 changes: 10 additions & 14 deletions rust/operator-binary/src/crd/security.rs
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,16 @@ impl KafkaTlsSecurity {
KafkaListenerName::Bootstrap.listener_ssl_truststore_type(),
"PKCS12".to_string(),
);
config.insert("sasl.enabled.mechanisms".to_string(), "GSSAPI".to_string());
config.insert(
"sasl.kerberos.service.name".to_string(),
KafkaRole::Broker.kerberos_service_name().to_string(),
);
config.insert(
"sasl.mechanism.inter.broker.protocol".to_string(),
"GSSAPI".to_string(),
);
tracing::debug!("Kerberos configs added: [{:#?}]", config);
}

// Internal TLS
Expand Down Expand Up @@ -545,20 +555,6 @@ impl KafkaTlsSecurity {
);
}

// Kerberos
if self.has_kerberos_enabled() {
config.insert("sasl.enabled.mechanisms".to_string(), "GSSAPI".to_string());
config.insert(
"sasl.kerberos.service.name".to_string(),
KafkaRole::Broker.kerberos_service_name().to_string(),
);
config.insert(
"sasl.mechanism.inter.broker.protocol".to_string(),
"GSSAPI".to_string(),
);
tracing::debug!("Kerberos configs added: [{:#?}]", config);
}

// common
config.insert(
Self::INTER_BROKER_LISTENER_NAME.to_string(),
Expand Down
33 changes: 12 additions & 21 deletions rust/operator-binary/src/resource/listener.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,7 @@ use stackable_operator::{
};

use crate::{
crd::{
METRICS_PORT, METRICS_PORT_NAME, role::broker::BrokerConfig, security::KafkaTlsSecurity,
v1alpha1,
},
crd::{role::broker::BrokerConfig, security::KafkaTlsSecurity, v1alpha1},
kafka_controller::KAFKA_CONTROLLER_NAME,
utils::build_recommended_labels,
};
Expand Down Expand Up @@ -53,33 +50,27 @@ pub fn build_broker_rolegroup_bootstrap_listener(
.build(),
spec: listener::v1alpha1::ListenerSpec {
class_name: Some(merged_config.bootstrap_listener_class.clone()),
ports: Some(listener_ports(kafka_security)),
ports: Some(bootstrap_listener_ports(kafka_security)),
..listener::v1alpha1::ListenerSpec::default()
},
status: None,
})
}

/// We only expose client HTTP / HTTPS and Metrics ports.
fn listener_ports(kafka_security: &KafkaTlsSecurity) -> Vec<listener::v1alpha1::ListenerPort> {
let mut ports = vec![
fn bootstrap_listener_ports(
kafka_security: &KafkaTlsSecurity,
) -> Vec<listener::v1alpha1::ListenerPort> {
vec![if kafka_security.has_kerberos_enabled() {
listener::v1alpha1::ListenerPort {
name: METRICS_PORT_NAME.to_string(),
port: METRICS_PORT.into(),
name: kafka_security.bootstrap_port_name().to_string(),
port: kafka_security.bootstrap_port().into(),
protocol: Some("TCP".to_string()),
},
}
} else {
listener::v1alpha1::ListenerPort {
name: kafka_security.client_port_name().to_string(),
port: kafka_security.client_port().into(),
protocol: Some("TCP".to_string()),
},
];
if kafka_security.has_kerberos_enabled() {
ports.push(listener::v1alpha1::ListenerPort {
name: kafka_security.bootstrap_port_name().to_string(),
port: kafka_security.bootstrap_port().into(),
protocol: Some("TCP".to_string()),
});
}
ports
}
}]
}
16 changes: 8 additions & 8 deletions tests/templates/kuttl/smoke-kraft/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,19 +3,19 @@
import requests

if __name__ == "__main__":
result = 0

LOG_LEVEL = "DEBUG" # if args.debug else 'INFO'
logging.basicConfig(
level=LOG_LEVEL,
format="%(asctime)s %(levelname)s: %(message)s",
stream=sys.stdout,
)

http_code = requests.get(
"http://test-kafka-broker-default-metrics:9606"
).status_code
if http_code != 200:
result = 1
response = requests.get("http://test-kafka-broker-default-metrics:9606/metrics")

assert response.status_code == 200, (
f"Expected HTTP return code 200 from the metrics endpoint but got [{response.status_code}]"
)

sys.exit(result)
assert "jmx_scrape_error" in response.text, (
"Expected metric [jmx_scrape_error] not found"
)
16 changes: 8 additions & 8 deletions tests/templates/kuttl/smoke/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,19 +3,19 @@
import requests

if __name__ == "__main__":
result = 0

LOG_LEVEL = "DEBUG" # if args.debug else 'INFO'
logging.basicConfig(
level=LOG_LEVEL,
format="%(asctime)s %(levelname)s: %(message)s",
stream=sys.stdout,
)

http_code = requests.get(
"http://test-kafka-broker-default-metrics:9606/metrics"
).status_code
if http_code != 200:
result = 1
response = requests.get("http://test-kafka-broker-default-metrics:9606/metrics")

assert response.status_code == 200, (
f"Expected HTTP return code 200 from the metrics endpoint but got [{response.status_code}]"
)

sys.exit(result)
assert "jmx_scrape_error" in response.text, (
"Expected metric [jmx_scrape_error] not found"
)