Skip to content

Commit 4414ac9

Browse files
FrawlessFrawless
authored
Scrape Kafka metrics from pods only from Kafka and KE (#39)
Signed-off-by: Jakub Stejskal <xstejs24@gmail.com>
1 parent dece0db commit 4414ac9

11 files changed

Lines changed: 772 additions & 5 deletions

File tree

‎common/src/main/java/io/streamshub/mcp/common/service/metrics/PodScrapingMetricsProvider.java‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,10 @@ public List<MetricSample> queryMetrics(final MetricsQueryParams params) {
6464
Set<String> metricFilter = params.metricNames() != null
6565
? new HashSet<>(params.metricNames()) : null;
6666

67+
LOG.debugf("Scraping %d pod target(s), metric filter: %s",
68+
targets.size(),
69+
metricFilter != null ? metricFilter : "none");
70+
6771
List<MetricSample> allSamples = new ArrayList<>();
6872

6973
for (PodTarget target : targets) {
@@ -78,21 +82,26 @@ public List<MetricSample> queryMetrics(final MetricsQueryParams params) {
7882
String body = kubernetesClient.raw(proxyUrl);
7983
if (body != null && !body.isEmpty()) {
8084
List<MetricSample> samples = PrometheusTextParser.parse(body, metricFilter);
85+
LOG.debugf("Scraped %d sample(s) from pod %s/%s",
86+
samples.size(), target.namespace(), target.podName());
8187
for (MetricSample sample : samples) {
8288
allSamples.add(MetricSample.of(sample.name(),
8389
MetricLabelFilter.filterLabels(sample.labels()),
8490
sample.value(), sample.timestamp()));
8591
}
8692
} else {
87-
LOG.warnf("Empty metrics response from pod %s/%s",
93+
LOG.debugf("Empty metrics response from pod %s/%s",
8894
target.namespace(), target.podName());
8995
}
9096
} catch (Exception e) {
91-
LOG.warnf("Error scraping metrics from pod %s/%s: %s",
97+
LOG.debugf("Error scraping metrics from pod %s/%s: %s",
9298
target.namespace(), target.podName(), e.getMessage());
9399
}
94100
}
95101

102+
LOG.debugf("Scraping complete: collected %d total sample(s) from %d target(s)",
103+
allSamples.size(), targets.size());
104+
96105
return allSamples;
97106
}
98107

‎strimzi-mcp/src/main/java/io/streamshub/mcp/strimzi/config/StrimziConstants.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,17 @@ public static final class ComponentTypes {
7373
*/
7474
public static final String KAFKA = "kafka";
7575

76+
/**
77+
* Component type value for Kafka Exporter components.
78+
*/
79+
public static final String KAFKA_EXPORTER = "kafka-exporter";
80+
81+
/**
82+
* Component types representing Kafka broker pods.
83+
*/
84+
public static final java.util.Set<String> BROKER_TYPES =
85+
java.util.Set.of(KAFKA);
86+
7687
private ComponentTypes() {
7788
}
7889
}

‎strimzi-mcp/src/main/java/io/streamshub/mcp/strimzi/config/StrimziToolsPrompts.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,14 @@ public final class StrimziToolsPrompts {
109109
+ " Defaults to 'reconciliation' if no category"
110110
+ " or metric names are provided.";
111111

112+
/**
113+
* Kafka Exporter metrics category parameter description.
114+
*/
115+
public static final String EXPORTER_METRICS_CATEGORY_DESC =
116+
"Metric category: 'consumer_lag', 'partitions',"
117+
+ " or 'resources'."
118+
+ " Defaults to 'consumer_lag' if omitted.";
119+
112120
/**
113121
* Metric names parameter description.
114122
*/

‎strimzi-mcp/src/main/java/io/streamshub/mcp/strimzi/config/metrics/KafkaExporterMetricCategories.java‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
/*
2+
* Copyright StreamsHub authors.
3+
* License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
4+
*/
5+
package io.streamshub.mcp.strimzi.config.metrics;
6+
7+
import java.util.List;
8+
import java.util.Map;
9+
import java.util.Set;
10+
import java.util.stream.Collectors;
11+
12+
/**
13+
* Curated metric name categories for Kafka Exporter metrics.
14+
* Maps human-friendly category names to lists of Prometheus metric names,
15+
* and provides interpretation guides for each category.
16+
*/
17+
public final class KafkaExporterMetricCategories {
18+
19+
private static final Map<String, List<String>> CATEGORIES = Map.of(
20+
"consumer_lag", List.of(
21+
"kafka_consumergroup_current_offset",
22+
"kafka_consumergroup_lag",
23+
"kafka_consumergroup_lag_seconds"
24+
),
25+
"partitions", List.of(
26+
"kafka_topic_partitions",
27+
"kafka_topic_partition_current_offset",
28+
"kafka_topic_partition_oldest_offset",
29+
"kafka_topic_partition_in_sync_replica",
30+
"kafka_topic_partition_under_replicated_partition",
31+
"kafka_topic_partition_replicas"
32+
),
33+
"resources", List.of(
34+
"jvm_memory_used_bytes",
35+
"jvm_memory_max_bytes",
36+
"jvm_gc_collection_seconds_count",
37+
"jvm_gc_collection_seconds_sum",
38+
"process_cpu_seconds_total",
39+
"jvm_threads_current"
40+
)
41+
);
42+
43+
private static final Map<String, String> DESCRIPTIONS = Map.of(
44+
"consumer_lag",
45+
"**[HIGH - CONSUMER GROUP LAG]**\n\n"
46+
+ "kafka_consumergroup_lag: Number of messages a consumer group is behind the latest offset. "
47+
+ "**THRESHOLDS**: 0 = fully caught up, <1000 = healthy, 1000-10000 = monitor, "
48+
+ ">10000 = consumers falling behind (scale consumers or investigate slowness). "
49+
+ "Sustained growth = consumers cannot keep up with producer throughput.\n\n"
50+
+ "kafka_consumergroup_lag_seconds: Estimated time in seconds for the consumer group "
51+
+ "to catch up. >60s = significant delay, >300s = investigate consumer health.\n\n"
52+
+ "kafka_consumergroup_current_offset: Current committed offset per consumer group/partition. "
53+
+ "Stalled offset = consumer is stuck or dead. Compare with partition end offset to calculate lag.",
54+
"partitions",
55+
"**[HIGH - PARTITION HEALTH]**\n\n"
56+
+ "kafka_topic_partition_under_replicated_partition: Partitions where ISR count < replica count. "
57+
+ "Should be 0. >0 means replicas are lagging — check broker health and disk I/O. "
58+
+ "**TIME-SENSITIVE**: Transient during rolling restarts, persistent = data loss risk.\n\n"
59+
+ "kafka_topic_partition_in_sync_replica: Number of in-sync replicas per partition. "
60+
+ "Should equal kafka_topic_partition_replicas. Drop below min.insync.replicas = "
61+
+ "producers with acks=all will fail.\n\n"
62+
+ "kafka_topic_partition_replicas: Configured replica count per partition.\n\n"
63+
+ "**[MEDIUM - PARTITION OFFSETS]**\n\n"
64+
+ "kafka_topic_partition_current_offset: Latest offset (end of log) per partition. "
65+
+ "Rate of change = write throughput per partition.\n\n"
66+
+ "kafka_topic_partition_oldest_offset: Earliest available offset per partition. "
67+
+ "Gap between oldest and current = retained data volume. "
68+
+ "Oldest offset advancing = log segments being cleaned/compacted.\n\n"
69+
+ "kafka_topic_partitions: Number of partitions per topic.",
70+
"resources",
71+
MetricsDescriptions.jvmDescription("get_kafka_cluster_pods",
72+
"exporter scraping failures or missing metrics")
73+
);
74+
75+
private KafkaExporterMetricCategories() {
76+
// Utility class — no instantiation
77+
}
78+
79+
/**
80+
* Resolves a category name to its list of metric names.
81+
*
82+
* @param category the category name (case-insensitive)
83+
* @return the list of metric names, or an empty list if the category is unknown
84+
*/
85+
public static List<String> resolve(final String category) {
86+
if (category == null) {
87+
return List.of();
88+
}
89+
return CATEGORIES.getOrDefault(category.toLowerCase(java.util.Locale.ROOT), List.of());
90+
}
91+
92+
/**
93+
* Returns all available category names.
94+
*
95+
* @return the set of category names
96+
*/
97+
public static Set<String> allCategories() {
98+
return CATEGORIES.keySet();
99+
}
100+
101+
/**
102+
* Returns an interpretation guide for the given categories.
103+
*
104+
* @param categories the category names to get descriptions for
105+
* @return a combined interpretation guide, or null if no categories match
106+
*/
107+
public static String interpretation(final List<String> categories) {
108+
if (categories == null || categories.isEmpty()) {
109+
return null;
110+
}
111+
String result = categories.stream()
112+
.map(c -> c.toLowerCase(java.util.Locale.ROOT))
113+
.filter(DESCRIPTIONS::containsKey)
114+
.map(DESCRIPTIONS::get)
115+
.collect(Collectors.joining("\n\n"));
116+
return result.isEmpty() ? null : result;
117+
}
118+
}

‎strimzi-mcp/src/main/java/io/streamshub/mcp/strimzi/dto/metrics/KafkaExporterMetricsResponse.java‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright StreamsHub authors.
3+
* License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
4+
*/
5+
package io.streamshub.mcp.strimzi.dto.metrics;
6+
7+
import com.fasterxml.jackson.annotation.JsonInclude;
8+
import com.fasterxml.jackson.annotation.JsonProperty;
9+
import io.streamshub.mcp.common.dto.metrics.MetricSample;
10+
import io.streamshub.mcp.common.dto.metrics.MetricTimeSeries;
11+
12+
import java.time.Instant;
13+
import java.util.List;
14+
15+
/**
16+
* Response containing metrics data from a Kafka Exporter.
17+
* For instant queries, metrics are returned as a flat list in {@code metrics}.
18+
* For range queries, metrics are grouped into compact time series in {@code timeSeries}.
19+
*
20+
* @param clusterName the Kafka cluster name
21+
* @param namespace the Kubernetes namespace
22+
* @param provider the metrics provider used
23+
* @param categories the metric categories requested
24+
* @param metricCount the number of distinct metric names in the response
25+
* @param sampleCount the total number of metric samples
26+
* @param metrics the flat list of metric samples (instant queries)
27+
* @param timeSeries the grouped time series (range queries)
28+
* @param interpretation brief guide for interpreting the returned metrics
29+
* @param timestamp the time this result was generated
30+
* @param message a human-readable summary of the result
31+
*/
32+
@JsonInclude(JsonInclude.Include.NON_NULL)
33+
public record KafkaExporterMetricsResponse(
34+
@JsonProperty("cluster_name") String clusterName,
35+
@JsonProperty("namespace") String namespace,
36+
@JsonProperty("provider") String provider,
37+
@JsonProperty("categories") List<String> categories,
38+
@JsonProperty("metric_count") long metricCount,
39+
@JsonProperty("sample_count") int sampleCount,
40+
@JsonProperty("metrics") List<MetricSample> metrics,
41+
@JsonProperty("time_series") List<MetricTimeSeries> timeSeries,
42+
@JsonProperty("interpretation") String interpretation,
43+
@JsonProperty("timestamp") Instant timestamp,
44+
@JsonProperty("message") String message
45+
) {
46+
47+
/**
48+
* Creates a response with metric data. If samples have timestamps (range query),
49+
* groups them into compact time series. Otherwise returns a flat metrics list.
50+
*
51+
* @param clusterName the Kafka cluster name
52+
* @param namespace the Kubernetes namespace
53+
* @param provider the metrics provider name
54+
* @param categories the requested categories
55+
* @param samples the metric samples
56+
* @param interpretation brief guide for interpreting the returned metrics
57+
* @return a response with the metric data
58+
*/
59+
public static KafkaExporterMetricsResponse of(final String clusterName, final String namespace,
60+
final String provider, final List<String> categories,
61+
final List<MetricSample> samples,
62+
final String interpretation) {
63+
long metricCount = samples.stream()
64+
.map(MetricSample::name)
65+
.distinct()
66+
.count();
67+
68+
String msg = String.format("Retrieved %d samples across %d metrics from Kafka Exporter for cluster '%s'",
69+
samples.size(), metricCount, clusterName);
70+
71+
boolean isRangeData = !samples.isEmpty() && samples.getFirst().timestamp() != null;
72+
73+
if (isRangeData) {
74+
List<MetricTimeSeries> series = MetricTimeSeries.fromSamples(samples);
75+
return new KafkaExporterMetricsResponse(clusterName, namespace, provider, categories,
76+
metricCount, samples.size(), null, series, interpretation, Instant.now(), msg);
77+
}
78+
79+
return new KafkaExporterMetricsResponse(clusterName, namespace, provider, categories,
80+
metricCount, samples.size(), samples, null, interpretation, Instant.now(), msg);
81+
}
82+
83+
/**
84+
* Creates an empty response when no metrics are available.
85+
*
86+
* @param clusterName the Kafka cluster name
87+
* @param namespace the Kubernetes namespace
88+
* @param message descriptive message explaining why no metrics are available
89+
* @return an empty response
90+
*/
91+
public static KafkaExporterMetricsResponse empty(final String clusterName, final String namespace,
92+
final String message) {
93+
return new KafkaExporterMetricsResponse(clusterName, namespace, null, List.of(),
94+
0, 0, List.of(), null, null, Instant.now(), message);
95+
}
96+
}

0 commit comments

Comments
 (0)