diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 81646132f..7c0c2ce7c 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -1063,4 +1063,25 @@ Here is an example of using custom state stores with functional style described } ---- -These state stores can be then accessed by the applications directly. \ No newline at end of file +These state stores can be then accessed by the applications directly. + +==== Accessing Kafka Streams Metrics + +Spring Cloud Stream Kafka Streams binder provides a basic mechanism for accessing Kafka Streams metrics exported through a MircoMeter `MeterRegistry`. +Kafka Streams metrics that are available through `KafkaStreams#metrics()` are exported to this meter registry by the binder. +The metrics exported are from the consumers, producers, admin-client and the stream itself. + +The metrics exported by the binder are exported with the format of metrics group name followed by a dot and then the actual metric name. +All dashes in the original metric information is replaced with dots. + +For e.g. the metric name `network-io-total` from the metric group `consumer-metrics` is available in the micrometer registry as `consumer.metrics.network.io.total`. +Similarly, the metric `commit-total` from `stream-metrics` is available as `stream.metrics.commit.total`. + +You can either programmatically access the Micrometer `MeterRegistry` in the application and then iterate through the available gauges or use Spring Boot actuator to access the metrics through a REST endpoint. +When accessing through the Boot actuator endpoint, make sure to add `metrics` to the property `management.endpoints.web.exposure.include`. +Then you can access `/acutator/metrics` to get a list of all the available metrics which then can be individually accessed through the same URL (`/actuator/metrics/`). + +Anything beyond the info level metrics available through `KafkaStreams#metrics()`, (for e.g. the debugging level metrics) are still only available through JMX after you set the `metrics.recording.level` to `DEBUG`. +Kafka Streams, by default, set this level to `INFO`. +https://kafka.apache.org/documentation/#kafka_streams_monitoring[Please see this section] from Kafka Streams documentation for more details. +In a future release, binder may support exporting these DEBUG level metrics as well through Micrometer. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java new file mode 100644 index 000000000..4e860a436 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderMetrics.java @@ -0,0 +1,96 @@ +/* + * Copyright 2019-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import java.util.Map; +import java.util.function.ToDoubleFunction; + +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.binder.MeterBinder; +import org.apache.kafka.common.Metric; +import org.apache.kafka.common.MetricName; +import org.apache.kafka.streams.KafkaStreams; + +/** + * Kafka Streams binder metrics implementation that exports the metrics available + * through {@link KafkaStreams#metrics()} into a micrometer {@link io.micrometer.core.instrument.MeterRegistry}. + * + * @author Soby Chacko + * @since 3.0.0 + */ +public class KafkaStreamsBinderMetrics { + + private KafkaStreams kafkaStreams; + + private final MeterRegistry meterRegistry; + + private MeterBinder meterBinder; + + public KafkaStreamsBinderMetrics(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + + public void bindTo(MeterRegistry meterRegistry) { + if (this.meterBinder == null) { + this.meterBinder = new MeterBinder() { + @Override + @SuppressWarnings("unchecked") + public void bindTo(MeterRegistry registry) { + if (KafkaStreamsBinderMetrics.this.kafkaStreams != null) { + final Map metrics = KafkaStreamsBinderMetrics.this.kafkaStreams.metrics(); + + for (Map.Entry metric : metrics.entrySet()) { + final Gauge.Builder builder = + Gauge.builder(sanitize(metric.getKey().group() + "." + metric.getKey().name()), this, + toDoubleFunction(metric.getValue())); + final Map tags = metric.getKey().tags(); + for (Map.Entry tag : tags.entrySet()) { + builder.tag(tag.getKey(), tag.getValue()); + } + builder.description(metric.getKey().description()) + .register(meterRegistry); + } + } + } + + ToDoubleFunction toDoubleFunction(Metric metric) { + return (o) -> { + if (metric.metricValue() instanceof Number) { + return (Double) metric.metricValue(); + } + else { + return 0.0; + } + }; + } + }; + } + this.meterBinder.bindTo(this.meterRegistry); + } + + public void addMetrics(KafkaStreams kafkaStreams) { + synchronized (KafkaStreamsBinderMetrics.this) { + this.kafkaStreams = kafkaStreams; + this.bindTo(this.meterRegistry); + } + } + + private static String sanitize(String value) { + return value.replaceAll("-", "."); + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 1d4b874eb..b4fbe48b4 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -23,6 +23,7 @@ import java.util.Map; import java.util.Properties; import java.util.stream.Collectors; +import io.micrometer.core.instrument.MeterRegistry; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; @@ -32,6 +33,7 @@ import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -48,8 +50,11 @@ import org.springframework.cloud.stream.config.BinderProperties; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.function.StreamFunctionProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.Environment; import org.springframework.core.env.MapPropertySource; @@ -67,6 +72,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Gary Russell */ +@Configuration @EnableConfigurationProperties(KafkaStreamsExtendedBindingProperties.class) @ConditionalOnBean(BindingService.class) @AutoConfigureAfter(BindingServiceConfiguration.class) @@ -338,8 +344,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration { } @Bean - public KafkaStreamsRegistry kafkaStreamsRegistry() { - return new KafkaStreamsRegistry(); + public KafkaStreamsRegistry kafkaStreamsRegistry(KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics) { + return new KafkaStreamsRegistry(kafkaStreamsBinderMetrics); } @Bean @@ -364,4 +370,35 @@ public class KafkaStreamsBinderSupportAutoConfiguration { cleanupConfig.getIfUnique(), streamFunctionProperties, kafkaStreamsBinderConfigurationProperties); } + + + @Configuration + @ConditionalOnMissingBean(value = KafkaStreamsBinderMetrics.class, name = "outerContext") + @ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry") + protected class KafkaStreamsBinderMetricsConfiguration { + + @Bean + @ConditionalOnBean(MeterRegistry.class) + @ConditionalOnMissingBean(KafkaStreamsBinderMetrics.class) + public KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics( + MeterRegistry meterRegistry) { + + return new KafkaStreamsBinderMetrics(meterRegistry); + } + } + + @Configuration + @ConditionalOnBean(name = "outerContext") + @ConditionalOnMissingBean(KafkaStreamsBinderMetrics.class) + @ConditionalOnClass(name = "io.micrometer.core.instrument.MeterRegistry") + protected class KafkaStreamsBinderMetricsConfigurationWithMultiBinder { + + @Bean + public KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics(ConfigurableApplicationContext context) { + + MeterRegistry meterRegistry = context.getBean("outerContext", ApplicationContext.class) + .getBean(MeterRegistry.class); + return new KafkaStreamsBinderMetrics(meterRegistry); + } + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java index 644b3c353..0799b89a9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsRegistry.java @@ -29,6 +29,12 @@ import org.apache.kafka.streams.KafkaStreams; */ class KafkaStreamsRegistry { + private final KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics; + + KafkaStreamsRegistry(KafkaStreamsBinderMetrics kafkaStreamsBinderMetrics) { + this.kafkaStreamsBinderMetrics = kafkaStreamsBinderMetrics; + } + private final Set kafkaStreams = new HashSet<>(); Set getKafkaStreams() { @@ -40,6 +46,7 @@ class KafkaStreamsRegistry { * @param kafkaStreams {@link KafkaStreams} object created in the application */ void registerKafkaStreams(KafkaStreams kafkaStreams) { + this.kafkaStreamsBinderMetrics.addMetrics(kafkaStreams); this.kafkaStreams.add(kafkaStreams); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java index c0d435ae2..1b34befe6 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java @@ -95,7 +95,8 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { public void testStateStoreRetrievalRetry() { KafkaStreams mock = Mockito.mock(KafkaStreams.class); - KafkaStreamsRegistry kafkaStreamsRegistry = new KafkaStreamsRegistry(); + KafkaStreamsBinderMetrics mockMetrics = Mockito.mock(KafkaStreamsBinderMetrics.class); + KafkaStreamsRegistry kafkaStreamsRegistry = new KafkaStreamsRegistry(mockMetrics); kafkaStreamsRegistry.registerKafkaStreams(mock); KafkaStreamsBinderConfigurationProperties binderConfigurationProperties = new KafkaStreamsBinderConfigurationProperties(new KafkaProperties()); @@ -140,8 +141,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } } - private void receiveAndValidateFoo(ConfigurableApplicationContext context) - throws Exception { + private void receiveAndValidateFoo(ConfigurableApplicationContext context) { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index 0d0492e17..58f9a3d5a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -21,6 +21,7 @@ import java.util.Date; import java.util.Map; import java.util.function.Function; +import io.micrometer.core.instrument.MeterRegistry; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -35,9 +36,11 @@ import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @@ -92,6 +95,8 @@ public class KafkaStreamsBinderWordCountFunctionTests { "=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { receiveAndValidate("words", "counts"); + final MeterRegistry meterRegistry = context.getBean(MeterRegistry.class); + assertThat(meterRegistry.get("stream.metrics.commit.total").gauge().value()).isEqualTo(1.0); } } @@ -183,6 +188,9 @@ public class KafkaStreamsBinderWordCountFunctionTests { @EnableAutoConfiguration public static class WordCountProcessorApplication { + @Autowired + InteractiveQueryService interactiveQueryService; + @Bean public Function, KStream> process() {