From 6fe3299f52f8c90859a8beecb70bda3fbbf43e09 Mon Sep 17 00:00:00 2001 From: jack-berg <34418638+jack-berg@users.noreply.github.com> Date: Mon, 25 Apr 2022 12:36:26 -0500 Subject: [PATCH] Add metric support for grpc (#5923) * Add metric support for grpc * Spotless --- .../grpc/v1_6/GrpcTelemetryBuilder.java | 8 +- .../grpc/v1_6/AbstractGrpcStreamingTest.java | 52 ++++ .../grpc/v1_6/AbstractGrpcTest.java | 252 ++++++++++++++++++ .../testing/LibraryTestRunner.java | 6 +- .../exporter/AgentTestingCustomizer.java | 12 +- .../exporter/AgentTestingExporterFactory.java | 3 + .../exporter/OtlpInMemoryMetricExporter.java | 2 +- 7 files changed, 326 insertions(+), 9 deletions(-) diff --git a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTelemetryBuilder.java b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTelemetryBuilder.java index 838cccf91cb4..e1b6e29eb6fd 100644 --- a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTelemetryBuilder.java +++ b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTelemetryBuilder.java @@ -16,7 +16,9 @@ import io.opentelemetry.instrumentation.api.instrumenter.net.NetClientAttributesExtractor; import io.opentelemetry.instrumentation.api.instrumenter.net.NetServerAttributesExtractor; import io.opentelemetry.instrumentation.api.instrumenter.rpc.RpcClientAttributesExtractor; +import io.opentelemetry.instrumentation.api.instrumenter.rpc.RpcClientMetrics; import io.opentelemetry.instrumentation.api.instrumenter.rpc.RpcServerAttributesExtractor; +import io.opentelemetry.instrumentation.api.instrumenter.rpc.RpcServerMetrics; import io.opentelemetry.instrumentation.grpc.v1_6.internal.GrpcNetClientAttributesGetter; import io.opentelemetry.instrumentation.grpc.v1_6.internal.GrpcNetServerAttributesGetter; import io.opentelemetry.semconv.trace.attributes.SemanticAttributes; @@ -126,11 +128,13 @@ public GrpcTelemetry build() { clientInstrumenterBuilder .addAttributesExtractor(RpcClientAttributesExtractor.create(rpcAttributesGetter)) - .addAttributesExtractor(NetClientAttributesExtractor.create(netClientAttributesGetter)); + .addAttributesExtractor(NetClientAttributesExtractor.create(netClientAttributesGetter)) + .addRequestMetrics(RpcClientMetrics.get()); serverInstrumenterBuilder .addAttributesExtractor(RpcServerAttributesExtractor.create(rpcAttributesGetter)) .addAttributesExtractor( - NetServerAttributesExtractor.create(new GrpcNetServerAttributesGetter())); + NetServerAttributesExtractor.create(new GrpcNetServerAttributesGetter())) + .addRequestMetrics(RpcServerMetrics.get()); if (peerService != null) { clientInstrumenterBuilder.addAttributesExtractor( diff --git a/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcStreamingTest.java b/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcStreamingTest.java index f0a6bd6b697a..8ed691382238 100644 --- a/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcStreamingTest.java +++ b/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcStreamingTest.java @@ -18,10 +18,12 @@ import io.grpc.ServerBuilder; import io.grpc.Status; import io.grpc.stub.StreamObserver; +import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; import io.opentelemetry.instrumentation.testing.util.ThrowingRunnable; import io.opentelemetry.sdk.testing.assertj.EventDataAssert; +import io.opentelemetry.sdk.testing.assertj.MetricAssertions; import io.opentelemetry.semconv.trace.attributes.SemanticAttributes; import java.util.ArrayList; import java.util.List; @@ -214,6 +216,56 @@ public void onCompleted() { SemanticAttributes.RPC_GRPC_STATUS_CODE, (long) Status.Code.OK.value())) .hasEventsSatisfyingExactly(events.toArray(new Consumer[0])))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.server.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .anySatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "Conversation") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.client.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .allSatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_PEER_NAME, "localhost") + .put( + SemanticAttributes.NET_PEER_PORT, + server.getPort()) + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "Conversation") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); } private ManagedChannel createChannel(Server server) throws Exception { diff --git a/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcTest.java b/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcTest.java index 62c8a64ad775..77db4075eb49 100644 --- a/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcTest.java +++ b/instrumentation/grpc-1.6/testing/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/AbstractGrpcTest.java @@ -40,10 +40,12 @@ import io.grpc.reflection.v1alpha.ServerReflectionRequest; import io.grpc.reflection.v1alpha.ServerReflectionResponse; import io.grpc.stub.StreamObserver; +import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; import io.opentelemetry.instrumentation.testing.util.ThrowingRunnable; +import io.opentelemetry.sdk.testing.assertj.MetricAssertions; import io.opentelemetry.sdk.trace.data.StatusData; import io.opentelemetry.semconv.trace.attributes.SemanticAttributes; import java.util.Queue; @@ -205,6 +207,56 @@ public void sayHello( "SENT"), entry( SemanticAttributes.MESSAGE_ID, 2L)))))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.server.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .anySatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.client.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .allSatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_PEER_NAME, "localhost") + .put( + SemanticAttributes.NET_PEER_PORT, + server.getPort()) + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); } @Test @@ -349,6 +401,56 @@ public void sayHello( span.hasName("child") .hasKind(SpanKind.INTERNAL) .hasParent(trace.getSpan(0)))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.server.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .anySatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.client.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .allSatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_PEER_NAME, "localhost") + .put( + SemanticAttributes.NET_PEER_PORT, + server.getPort()) + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); } @Test @@ -501,6 +603,56 @@ public void onCompleted() { span.hasName("child") .hasKind(SpanKind.INTERNAL) .hasParent(trace.getSpan(0)))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.server.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .anySatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.client.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .allSatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_PEER_NAME, "localhost") + .put( + SemanticAttributes.NET_PEER_PORT, + server.getPort()) + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); } @ParameterizedTest @@ -610,6 +762,56 @@ public void sayHello( span.hasException(status.getCause()); } }))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.server.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .anySatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.client.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .allSatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_PEER_NAME, "localhost") + .put( + SemanticAttributes.NET_PEER_PORT, + server.getPort()) + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); } @ParameterizedTest @@ -717,6 +919,56 @@ public void sayHello( entry(SemanticAttributes.MESSAGE_ID, 1L))); span.hasException(status.asRuntimeException()); }))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.server.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .anySatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); + testing() + .waitAndAssertMetrics( + "io.opentelemetry.grpc-1.6", + "rpc.client.duration", + metrics -> + metrics.anySatisfy( + metric -> + MetricAssertions.assertThat(metric) + .hasUnit("ms") + .hasDoubleHistogram() + .points() + .allSatisfy( + point -> + MetricAssertions.assertThat(point) + .hasAttributes( + Attributes.builder() + .put(SemanticAttributes.NET_PEER_NAME, "localhost") + .put( + SemanticAttributes.NET_PEER_PORT, + server.getPort()) + .put(SemanticAttributes.NET_TRANSPORT, "ip_tcp") + .put(SemanticAttributes.RPC_METHOD, "SayHello") + .put( + SemanticAttributes.RPC_SERVICE, + "example.Greeter") + .put(SemanticAttributes.RPC_SYSTEM, "grpc") + .build())))); } static class ErrorProvider implements ArgumentsProvider { diff --git a/testing-common/src/main/java/io/opentelemetry/instrumentation/testing/LibraryTestRunner.java b/testing-common/src/main/java/io/opentelemetry/instrumentation/testing/LibraryTestRunner.java index f170dae09e39..940c2acf89a2 100644 --- a/testing-common/src/main/java/io/opentelemetry/instrumentation/testing/LibraryTestRunner.java +++ b/testing-common/src/main/java/io/opentelemetry/instrumentation/testing/LibraryTestRunner.java @@ -15,6 +15,7 @@ import io.opentelemetry.sdk.common.CompletableResultCode; import io.opentelemetry.sdk.logs.data.LogData; import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter; @@ -28,6 +29,7 @@ import java.time.Duration; import java.util.Collections; import java.util.List; +import java.util.concurrent.TimeUnit; /** * An implementation of {@link InstrumentationTestRunner} that initializes OpenTelemetry SDK and @@ -44,7 +46,7 @@ public final class LibraryTestRunner extends InstrumentationTestRunner { GlobalOpenTelemetry.resetForTest(); testSpanExporter = InMemorySpanExporter.create(); - testMetricExporter = InMemoryMetricExporter.create(); + testMetricExporter = InMemoryMetricExporter.create(AggregationTemporality.DELTA); openTelemetry = OpenTelemetrySdk.builder() @@ -89,6 +91,8 @@ public void afterTestClass() {} @Override public void clearAllExportedData() { + // Flush meter provider to remove any lingering measurements + openTelemetry.getSdkMeterProvider().forceFlush().join(10, TimeUnit.SECONDS); testSpanExporter.reset(); testMetricExporter.reset(); forceFlushCalled = false; diff --git a/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingCustomizer.java b/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingCustomizer.java index 973d461bbd8f..001ae2a6caf9 100644 --- a/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingCustomizer.java +++ b/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingCustomizer.java @@ -8,6 +8,7 @@ import com.google.auto.service.AutoService; import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider; +import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; import java.time.Duration; @@ -19,6 +20,11 @@ public class AgentTestingCustomizer implements AutoConfigurationCustomizerProvid new AgentTestingSpanProcessor( SimpleSpanProcessor.create(AgentTestingExporterFactory.spanExporter)); + static final MetricReader metricReader = + PeriodicMetricReader.builder(AgentTestingExporterFactory.metricExporter) + .setInterval(Duration.ofMillis(100)) + .build(); + static void reset() { spanProcessor.forceFlushCalled = false; } @@ -29,10 +35,6 @@ public void customize(AutoConfigurationCustomizer autoConfigurationCustomizer) { (tracerProvider, config) -> tracerProvider.addSpanProcessor(spanProcessor)); autoConfigurationCustomizer.addMeterProviderCustomizer( - (meterProvider, config) -> - meterProvider.registerMetricReader( - PeriodicMetricReader.builder(AgentTestingExporterFactory.metricExporter) - .setInterval(Duration.ofMillis(100)) - .build())); + (meterProvider, config) -> meterProvider.registerMetricReader(metricReader)); } } diff --git a/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingExporterFactory.java b/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingExporterFactory.java index 4706f6489a0f..de369beb5b01 100644 --- a/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingExporterFactory.java +++ b/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/AgentTestingExporterFactory.java @@ -6,6 +6,7 @@ package io.opentelemetry.javaagent.testing.exporter; import java.util.List; +import java.util.concurrent.TimeUnit; public class AgentTestingExporterFactory { @@ -26,6 +27,8 @@ public static List getLogExportRequests() { } public static void reset() { + // Flush meter provider to remove any lingering measurements + AgentTestingCustomizer.metricReader.flush().join(10, TimeUnit.SECONDS); spanExporter.reset(); metricExporter.reset(); logExporter.reset(); diff --git a/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/OtlpInMemoryMetricExporter.java b/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/OtlpInMemoryMetricExporter.java index 7a4e0f511b7f..c3244e03f894 100644 --- a/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/OtlpInMemoryMetricExporter.java +++ b/testing/agent-exporter/src/main/java/io/opentelemetry/javaagent/testing/exporter/OtlpInMemoryMetricExporter.java @@ -34,7 +34,7 @@ void reset() { @Override public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { - return AggregationTemporality.CUMULATIVE; + return AggregationTemporality.DELTA; } @Override