open-telemetry
/
opentelemetry-java-instrumentation
mirror of https://github.com/open-telemetry/opentelemetry-java-instrumentation.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix flaky kafka metrics test (#6511)

This commit is contained in:
Lauri Tulmin 2022-08-26 01:22:38 +03:00 committed by GitHub
parent dd752816b7
commit 33d2e40a9e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/java/io/opentelemetry/instrumentation/kafka/internal/OpenTelemetryMetricsReporterTest.java
View File

@ -273,18 +273,8 @@ class OpenTelemetryMetricsReporterTest {
"kafka.consumer.successful_reauthentication_total", "kafka.consumer.successful_reauthentication_total",
"kafka.consumer.time_between_poll_avg", "kafka.consumer.time_between_poll_avg",
"kafka.consumer.time_between_poll_max", "kafka.consumer.time_between_poll_max",
"kafka.consumer.incoming_byte_rate",
"kafka.consumer.incoming_byte_total",
"kafka.consumer.outgoing_byte_rate",
"kafka.consumer.outgoing_byte_total",
"kafka.consumer.request_latency_avg", "kafka.consumer.request_latency_avg",
"kafka.consumer.request_latency_max", "kafka.consumer.request_latency_max",
"kafka.consumer.request_rate",
"kafka.consumer.request_size_avg",
"kafka.consumer.request_size_max",
"kafka.consumer.request_total",
"kafka.consumer.response_rate",
"kafka.consumer.response_total",
"kafka.producer.batch_size_avg", "kafka.producer.batch_size_avg",
"kafka.producer.batch_size_max", "kafka.producer.batch_size_max",
"kafka.producer.batch_split_rate", "kafka.producer.batch_split_rate",
@ -350,27 +340,9 @@ class OpenTelemetryMetricsReporterTest {
"kafka.producer.successful_reauthentication_rate", "kafka.producer.successful_reauthentication_rate",
"kafka.producer.successful_reauthentication_total", "kafka.producer.successful_reauthentication_total",
"kafka.producer.waiting_threads", "kafka.producer.waiting_threads",
"kafka.producer.incoming_byte_rate",
"kafka.producer.incoming_byte_total",
"kafka.producer.outgoing_byte_rate",
"kafka.producer.outgoing_byte_total",
"kafka.producer.request_latency_avg",
"kafka.producer.request_latency_max",
"kafka.producer.request_rate",
"kafka.producer.request_size_avg",
"kafka.producer.request_size_max",
"kafka.producer.request_total",
"kafka.producer.response_rate",
"kafka.producer.response_total",
"kafka.producer.byte_rate", "kafka.producer.byte_rate",
"kafka.producer.byte_total", "kafka.producer.byte_total",
"kafka.producer.compression_rate", "kafka.producer.compression_rate"));
"kafka.producer.record_error_rate",
"kafka.producer.record_error_total",
"kafka.producer.record_retry_rate",
"kafka.producer.record_retry_total",
"kafka.producer.record_send_rate",
"kafka.producer.record_send_total"));
List<MetricData> metrics = testing.metrics(); List<MetricData> metrics = testing.metrics();
Set<String> metricNames = metrics.stream().map(MetricData::getName).collect(toSet()); Set<String> metricNames = metrics.stream().map(MetricData::getName).collect(toSet());

18
testing-common/src/main/java/io/opentelemetry/instrumentation/testing/LibraryTestRunner.java
View File

@ -17,6 +17,7 @@ import io.opentelemetry.sdk.logs.data.LogData;
import io.opentelemetry.sdk.metrics.SdkMeterProvider; import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.data.MetricData;
import io.opentelemetry.sdk.metrics.export.MetricReader;
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
import io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter;
import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
@ -40,6 +41,7 @@ public final class LibraryTestRunner extends InstrumentationTestRunner {
private static final OpenTelemetrySdk openTelemetry; private static final OpenTelemetrySdk openTelemetry;
private static final InMemorySpanExporter testSpanExporter; private static final InMemorySpanExporter testSpanExporter;
private static final InMemoryMetricExporter testMetricExporter; private static final InMemoryMetricExporter testMetricExporter;
private static final MetricReader metricReader;
private static boolean forceFlushCalled; private static boolean forceFlushCalled;
static { static {
@ -48,6 +50,13 @@ public final class LibraryTestRunner extends InstrumentationTestRunner {
testSpanExporter = InMemorySpanExporter.create(); testSpanExporter = InMemorySpanExporter.create();
testMetricExporter = InMemoryMetricExporter.create(AggregationTemporality.DELTA); testMetricExporter = InMemoryMetricExporter.create(AggregationTemporality.DELTA);
metricReader =
PeriodicMetricReader.builder(testMetricExporter)
// Set really long interval. We'll call forceFlush when we need the metrics
// instead of collecting them periodically.
.setInterval(Duration.ofNanos(Long.MAX_VALUE))
.build();
openTelemetry = openTelemetry =
OpenTelemetrySdk.builder() OpenTelemetrySdk.builder()
.setTracerProvider( .setTracerProvider(
@ -56,13 +65,7 @@ public final class LibraryTestRunner extends InstrumentationTestRunner {
.addSpanProcessor(SimpleSpanProcessor.create(LoggingSpanExporter.create())) .addSpanProcessor(SimpleSpanProcessor.create(LoggingSpanExporter.create()))
.addSpanProcessor(SimpleSpanProcessor.create(testSpanExporter)) .addSpanProcessor(SimpleSpanProcessor.create(testSpanExporter))
.build()) .build())
.setMeterProvider( .setMeterProvider(SdkMeterProvider.builder().registerMetricReader(metricReader).build())
SdkMeterProvider.builder()
.registerMetricReader(
PeriodicMetricReader.builder(testMetricExporter)
.setInterval(Duration.ofMillis(100))
.build())
.build())
.setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance())) .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
.buildAndRegisterGlobal(); .buildAndRegisterGlobal();
} }
@ -114,6 +117,7 @@ public final class LibraryTestRunner extends InstrumentationTestRunner {
@Override @Override
public List<MetricData> getExportedMetrics() { public List<MetricData> getExportedMetrics() {
metricReader.forceFlush().join(10, TimeUnit.SECONDS);
return testMetricExporter.getFinishedMetricItems(); return testMetricExporter.getFinishedMetricItems();
} }