JMX Scraper: Kafka server, producer and consumer YAMLs and integration tests added #1670
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,146 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.contrib.jmxscraper.target_systems.kafka; | ||
|
|
||
| import static io.opentelemetry.contrib.jmxscraper.assertions.DataPointAttributes.attribute; | ||
| import static io.opentelemetry.contrib.jmxscraper.assertions.DataPointAttributes.attributeGroup; | ||
| import static io.opentelemetry.contrib.jmxscraper.assertions.DataPointAttributes.attributeWithAnyValue; | ||
| import static io.opentelemetry.contrib.jmxscraper.target_systems.kafka.KafkaContainerFactory.createKafkaConsumerContainer; | ||
| import static io.opentelemetry.contrib.jmxscraper.target_systems.kafka.KafkaContainerFactory.createKafkaContainer; | ||
| import static io.opentelemetry.contrib.jmxscraper.target_systems.kafka.KafkaContainerFactory.createKafkaProducerContainer; | ||
| import static io.opentelemetry.contrib.jmxscraper.target_systems.kafka.KafkaContainerFactory.createZookeeperContainer; | ||
|
|
||
| import io.opentelemetry.contrib.jmxscraper.JmxScraperContainer; | ||
| import io.opentelemetry.contrib.jmxscraper.target_systems.MetricsVerifier; | ||
| import io.opentelemetry.contrib.jmxscraper.target_systems.TargetSystemIntegrationTest; | ||
| import java.nio.file.Path; | ||
| import java.util.Arrays; | ||
| import java.util.Collection; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.testcontainers.containers.GenericContainer; | ||
| import org.testcontainers.containers.output.Slf4jLogConsumer; | ||
| import org.testcontainers.containers.wait.strategy.Wait; | ||
|
|
||
| public class KafkaConsumerIntegrationTest extends TargetSystemIntegrationTest { | ||
|
|
||
| @Override | ||
| protected Collection<GenericContainer<?>> createPrerequisiteContainers() { | ||
| GenericContainer<?> zookeeper = | ||
| createZookeeperContainer() | ||
| .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("zookeeper"))) | ||
| .withNetworkAliases("zookeeper"); | ||
|
|
||
| GenericContainer<?> kafka = | ||
| createKafkaContainer() | ||
| .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("kafka"))) | ||
| .withNetworkAliases("kafka") | ||
| .dependsOn(zookeeper); | ||
|
|
||
| GenericContainer<?> kafkaProducer = | ||
| createKafkaProducerContainer() | ||
| .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("kafka-producer"))) | ||
| .withNetworkAliases("kafka-producer") | ||
| .dependsOn(kafka); | ||
|
|
||
| return Arrays.asList(zookeeper, kafka, kafkaProducer); | ||
| } | ||
|
|
||
| @Override | ||
| protected GenericContainer<?> createTargetContainer(int jmxPort) { | ||
| return createKafkaConsumerContainer() | ||
| .withEnv("JMX_PORT", Integer.toString(jmxPort)) | ||
| .withExposedPorts(jmxPort) | ||
| .waitingFor(Wait.forListeningPorts(jmxPort)); | ||
| } | ||
|
|
||
| @Override | ||
| protected JmxScraperContainer customizeScraperContainer( | ||
| JmxScraperContainer scraper, GenericContainer<?> target, Path tempDir) { | ||
| return scraper.withTargetSystem("kafka-consumer"); | ||
| } | ||
|
|
||
| @Override | ||
| protected MetricsVerifier createMetricsVerifier() { | ||
| return MetricsVerifier.create() | ||
| .add( | ||
| "kafka.consumer.fetch-rate", | ||
| metric -> | ||
| metric | ||
| .hasDescription("The number of fetch requests for all topics per second") | ||
| .hasUnit("{request}") | ||
| .isGauge() | ||
| .hasDataPointsWithOneAttribute( | ||
| attributeWithAnyValue("client.id"))) // changed to follow semconv | ||
robsunday marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| .add( | ||
| "kafka.consumer.records-lag-max", | ||
| metric -> | ||
| metric | ||
| .hasDescription("Number of messages the consumer lags behind the producer") | ||
| .hasUnit("{message}") | ||
| .isGauge() | ||
| .hasDataPointsWithOneAttribute(attributeWithAnyValue("client.id"))) | ||
| .add( | ||
| "kafka.consumer.total.bytes-consumed-rate", | ||
| metric -> | ||
| metric | ||
| .hasDescription( | ||
| "The average number of bytes consumed for all topics per second") | ||
| .hasUnit("By") | ||
|
There was a problem hiding this comment. [for reviewer] Changed from 'by' to follow semconv |
||
| .isGauge() | ||
| .hasDataPointsWithOneAttribute(attributeWithAnyValue("client.id"))) | ||
| .add( | ||
| "kafka.consumer.total.fetch-size-avg", | ||
| metric -> | ||
| metric | ||
| .hasDescription( | ||
| "The average number of bytes fetched per request for all topics") | ||
| .hasUnit("By") | ||
| .isGauge() | ||
| .hasDataPointsWithOneAttribute(attributeWithAnyValue("client.id"))) | ||
| .add( | ||
| "kafka.consumer.total.records-consumed-rate", | ||
| metric -> | ||
| metric | ||
| .hasDescription( | ||
| "The average number of records consumed for all topics per second") | ||
| .hasUnit("{record}") | ||
|
There was a problem hiding this comment. [for reviewer] Multiple units in this file have been changed from "1" to annotated form to follow semconv |
||
| .isGauge() | ||
| .hasDataPointsWithOneAttribute(attributeWithAnyValue("client.id"))) | ||
| .add( | ||
| "kafka.consumer.bytes-consumed-rate", | ||
| metric -> | ||
| metric | ||
| .hasDescription("The average number of bytes consumed per second") | ||
| .hasUnit("By") | ||
| .isGauge() | ||
| .hasDataPointsWithAttributes( | ||
| attributeGroup( | ||
| attributeWithAnyValue("client.id"), | ||
| attribute("topic", "test-topic-1")))) | ||
| .add( | ||
| "kafka.consumer.fetch-size-avg", | ||
| metric -> | ||
| metric | ||
| .hasDescription("The average number of bytes fetched per request") | ||
| .hasUnit("By") | ||
| .isGauge() | ||
| .hasDataPointsWithAttributes( | ||
| attributeGroup( | ||
| attributeWithAnyValue("client.id"), | ||
| attribute("topic", "test-topic-1")))) | ||
| .add( | ||
| "kafka.consumer.records-consumed-rate", | ||
| metric -> | ||
| metric | ||
| .hasDescription("The average number of records consumed per second") | ||
| .hasUnit("{record}") | ||
| .isGauge() | ||
| .hasDataPointsWithAttributes( | ||
| attributeGroup( | ||
| attributeWithAnyValue("client.id"), | ||
| attribute("topic", "test-topic-1")))); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.contrib.jmxscraper.target_systems.kafka; | ||
|
|
||
| import java.time.Duration; | ||
| import org.testcontainers.containers.GenericContainer; | ||
| import org.testcontainers.containers.wait.strategy.Wait; | ||
|
|
||
| public class KafkaContainerFactory { | ||
| private static final int KAFKA_PORT = 9092; | ||
| private static final String KAFKA_BROKER = "kafka:" + KAFKA_PORT; | ||
| private static final String KAFKA_DOCKER_IMAGE = "bitnami/kafka:2.8.1"; | ||
|
|
||
| private KafkaContainerFactory() {} | ||
|
|
||
| public static GenericContainer<?> createZookeeperContainer() { | ||
| return new GenericContainer<>("zookeeper:3.5") | ||
| .withStartupTimeout(Duration.ofMinutes(2)) | ||
| .waitingFor(Wait.forListeningPort()); | ||
| } | ||
|
|
||
| public static GenericContainer<?> createKafkaContainer() { | ||
| return new GenericContainer<>(KAFKA_DOCKER_IMAGE) | ||
| .withEnv("KAFKA_CFG_ZOOKEEPER_CONNECT", "zookeeper:2181") | ||
| .withEnv("ALLOW_PLAINTEXT_LISTENER", "yes") // Removed in 3.5.1 | ||
| .withStartupTimeout(Duration.ofMinutes(2)) | ||
| .withExposedPorts(KAFKA_PORT) | ||
| // .waitingFor(Wait.forListeningPorts(KAFKA_PORT)); | ||
robsunday marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| .waitingFor( | ||
| Wait.forLogMessage(".*KafkaServer.*started \\(kafka.server.KafkaServer\\).*", 1)); | ||
| } | ||
|
|
||
| public static GenericContainer<?> createKafkaProducerContainer() { | ||
| return new GenericContainer<>(KAFKA_DOCKER_IMAGE) | ||
| // .withCopyFileToContainer( | ||
| // MountableFile.forClasspathResource("kafka-producer.sh"), | ||
| // "/usr/bin/kafka-producer.sh") | ||
| // .withCommand("/usr/bin/kafka-producer.sh") | ||
robsunday marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| .withCommand( | ||
| "sh", | ||
| "-c", | ||
| "echo 'Sending messages to test-topic-1'; " | ||
| + "i=1; while true; do echo \"Message $i\"; sleep .25; i=$((i+1)); done | /opt/bitnami/kafka/bin/kafka-console-producer.sh --bootstrap-server " | ||
|
There was a problem hiding this comment. [for reviewer] Topic is automatically created when messages are sent to it - no need to manually create it |
||
| + KAFKA_BROKER | ||
| + " --topic test-topic-1;") | ||
| .withStartupTimeout(Duration.ofMinutes(2)) | ||
| .waitingFor(Wait.forLogMessage(".*Welcome to the Bitnami kafka container.*", 1)); | ||
| } | ||
|
|
||
| public static GenericContainer<?> createKafkaConsumerContainer() { | ||
| return new GenericContainer<>(KAFKA_DOCKER_IMAGE) | ||
| .withCommand( | ||
| "kafka-console-consumer.sh", | ||
| "--bootstrap-server", | ||
| KAFKA_BROKER, | ||
| "--whitelist", | ||
| "test-topic-.*", | ||
| "--max-messages", | ||
| "100") | ||
| .withStartupTimeout(Duration.ofMinutes(2)) | ||
| .waitingFor(Wait.forListeningPort()); | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[for reviewer] I rearranged containers shutdown sequence to be in reverse order than starting to avoid unnecessary errors in the logs