diff --git a/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilderTest.java b/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilderTest.java index 4e9e4615898..eabaf83ed29 100644 --- a/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilderTest.java +++ b/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilderTest.java @@ -24,6 +24,7 @@ import com.linecorp.armeria.testing.junit5.server.ServerExtension; import io.opentelemetry.api.GlobalOpenTelemetry; import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.LongCounter; import io.opentelemetry.api.metrics.LongCounterBuilder; import io.opentelemetry.api.metrics.Meter; @@ -38,6 +39,7 @@ import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector; import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector; import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; +import io.opentelemetry.sdk.metrics.internal.data.ImmutableDoublePointData; import io.opentelemetry.sdk.metrics.internal.data.ImmutableGaugeData; import io.opentelemetry.sdk.resources.Resource; import java.util.Collection; @@ -60,7 +62,8 @@ class OtlpHttpMetricExporterBuilderTest { "test", "test", "test", - ImmutableGaugeData.empty())); + ImmutableGaugeData.create( + singleton(ImmutableDoublePointData.create(0, 1, Attributes.empty(), 1.0))))); @RegisterExtension private static final ServerExtension server = diff --git a/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/metrics/OtlpGrpcMetricExporterBuilderTest.java b/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/metrics/OtlpGrpcMetricExporterBuilderTest.java index ea0d2dc4d19..a86d9562d6e 100644 --- a/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/metrics/OtlpGrpcMetricExporterBuilderTest.java +++ b/exporters/otlp/all/src/test/java/io/opentelemetry/exporter/otlp/metrics/OtlpGrpcMetricExporterBuilderTest.java @@ -25,6 +25,7 @@ import com.linecorp.armeria.testing.junit5.server.ServerExtension; import io.opentelemetry.api.GlobalOpenTelemetry; import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.LongCounter; import io.opentelemetry.api.metrics.LongCounterBuilder; import io.opentelemetry.api.metrics.Meter; @@ -40,6 +41,7 @@ import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector; import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector; import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; +import io.opentelemetry.sdk.metrics.internal.data.ImmutableDoublePointData; import io.opentelemetry.sdk.metrics.internal.data.ImmutableGaugeData; import io.opentelemetry.sdk.resources.Resource; import java.util.Collection; @@ -64,7 +66,8 @@ class OtlpGrpcMetricExporterBuilderTest { "test", "test", "test", - ImmutableGaugeData.empty())); + ImmutableGaugeData.create( + singleton(ImmutableDoublePointData.create(0, 1, Attributes.empty(), 1.0))))); @RegisterExtension private static final ServerExtension server = diff --git a/exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshaler.java b/exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshaler.java index 42c48b6a690..ca14bd0a24f 100644 --- a/exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshaler.java +++ b/exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshaler.java @@ -37,6 +37,10 @@ public MemoryMode getMemoryMode() { } public CompletableResultCode export(Collection metrics) { + int pointCount = 0; + for (MetricData metric : metrics) { + pointCount += metric.getData().getPoints().size(); + } if (memoryMode == MemoryMode.REUSABLE_DATA) { LowAllocationMetricsRequestMarshaler marshaler = marshalerPool.poll(); if (marshaler == null) { @@ -45,7 +49,7 @@ public CompletableResultCode export(Collection metrics) { LowAllocationMetricsRequestMarshaler exportMarshaler = marshaler; exportMarshaler.initialize(metrics); return doExport - .apply(exportMarshaler, metrics.size()) + .apply(exportMarshaler, pointCount) .whenComplete(() -> { exportMarshaler.reset(); marshalerPool.add(exportMarshaler); @@ -53,6 +57,6 @@ public CompletableResultCode export(Collection metrics) { } // MemoryMode == MemoryMode.IMMUTABLE_DATA MetricsRequestMarshaler request = MetricsRequestMarshaler.create(metrics); - return doExport.apply(request, metrics.size()); + return doExport.apply(request, pointCount); } } diff --git a/exporters/otlp/common/src/test/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshalerTest.java b/exporters/otlp/common/src/test/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshalerTest.java new file mode 100644 index 00000000000..5a27bd3a048 --- /dev/null +++ b/exporters/otlp/common/src/test/java/io/opentelemetry/exporter/internal/otlp/metrics/MetricReusableDataMarshalerTest.java @@ -0,0 +1,175 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.exporter.internal.otlp.metrics; + +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.exporter.internal.metrics.ExporterInstrumentation; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.common.InternalTelemetryVersion; +import io.opentelemetry.sdk.common.export.MemoryMode; +import io.opentelemetry.sdk.common.internal.ComponentId; +import io.opentelemetry.sdk.common.internal.StandardComponentId; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.LongPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.internal.data.ImmutableGaugeData; +import io.opentelemetry.sdk.metrics.internal.data.ImmutableLongPointData; +import io.opentelemetry.sdk.metrics.internal.data.ImmutableMetricData; +import io.opentelemetry.sdk.resources.Resource; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.net.URI; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +class MetricReusableDataMarshalerTest { + + @ParameterizedTest + @MethodSource("exportCases") + void countsDataPoints( + InternalTelemetryVersion telemetryVersion, MemoryMode memoryMode, boolean successful) { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + try (SdkMeterProvider provider = + SdkMeterProvider.builder().registerMetricReader(reader).build()) { + ExporterInstrumentation instrumentation = + new ExporterInstrumentation( + telemetryVersion, + () -> provider, + ComponentId.generateLazy(StandardComponentId.ExporterType.OTLP_GRPC_METRIC_EXPORTER), + URI.create("http://localhost:4317")); + AtomicReference pending = new AtomicReference<>(); + MetricReusableDataMarshaler marshaler = + new MetricReusableDataMarshaler( + memoryMode, + (request, count) -> { + ExporterInstrumentation.Recording recording = + instrumentation.startRecordingExport(count); + CompletableResultCode result = new CompletableResultCode(); + pending.set(result); + return result.whenComplete(() -> { + if (result.isSuccess()) { + recording.finishSuccessful(); + } else { + recording.finishFailed("test_failure"); + } + }); + }); + long exported = 0; + int[] expectedPointCounts = {5, 0, 0, 1}; + int batchIndex = 0; + for (List batch : + Arrays.asList( + Arrays.asList(metric("first", 3), metric("second", 2), metric("empty", 0)), + Collections.singletonList(metric("empty", 0)), + Collections.emptyList(), + Collections.singletonList(metric("next", 1)))) { + int points = expectedPointCounts[batchIndex++]; + CompletableResultCode result = marshaler.export(batch); + assertThat(result.isDone()).isFalse(); + if (telemetryVersion == InternalTelemetryVersion.LEGACY) { + assertLegacyMetricValue( + reader, + "otlp.exporter.seen", + exported + points, + Attributes.builder().put("type", "metric").build()); + if (batchIndex == 1) { + assertThat(reader.collectAllMetrics()) + .noneMatch(metric -> metric.getName().equals("otlp.exporter.exported")); + } else { + assertLegacyMetricValue( + reader, + "otlp.exporter.exported", + exported, + Attributes.builder().put("type", "metric").put("success", successful).build()); + } + } else { + assertMetricValue(reader, "otel.sdk.exporter.metric_data_point.inflight", points); + } + if (successful) { + pending.get().succeed(); + } else { + pending.get().fail(); + } + assertThat(result.isDone()).isTrue(); + assertThat(result.isSuccess()).isEqualTo(successful); + exported += points; + if (telemetryVersion == InternalTelemetryVersion.LEGACY) { + assertLegacyMetricValue( + reader, + "otlp.exporter.seen", + exported, + Attributes.builder().put("type", "metric").build()); + assertLegacyMetricValue( + reader, + "otlp.exporter.exported", + exported, + Attributes.builder().put("type", "metric").put("success", successful).build()); + } else { + assertMetricValue(reader, "otel.sdk.exporter.metric_data_point.inflight", 0); + assertMetricValue(reader, "otel.sdk.exporter.metric_data_point.exported", exported); + } + } + } + } + + private static void assertMetricValue(InMemoryMetricReader reader, String name, long expected) { + assertThat(reader.collectAllMetrics()) + .filteredOn(metric -> metric.getName().equals(name)) + .singleElement() + .satisfies(metric -> + assertThat(metric.getLongSumData().getPoints()) + .singleElement() + .extracting(LongPointData::getValue) + .isEqualTo(expected)); + } + + private static void assertLegacyMetricValue( + InMemoryMetricReader reader, String name, long expected, Attributes attributes) { + assertThat(reader.collectAllMetrics()) + .filteredOn(metric -> metric.getName().equals(name)) + .singleElement() + .satisfies(metric -> + assertThat(metric) + .hasLongSumSatisfying(sum -> + sum.hasPointsSatisfying( + point -> point.hasValue(expected).hasAttributes(attributes)))); + } + + private static MetricData metric(String name, int count) { + List points = new ArrayList<>(); + for (int i = 0; i < count; i++) { + points.add(ImmutableLongPointData.create(0, 1, Attributes.builder().put("id", i).build(), i)); + } + return ImmutableMetricData.createLongGauge( + Resource.empty(), + InstrumentationScopeInfo.create("test"), + name, + "", + "1", + ImmutableGaugeData.create(points)); + } + + private static Stream exportCases() { + return Stream.of(InternalTelemetryVersion.LEGACY, InternalTelemetryVersion.LATEST) + .flatMap(version -> + Arrays.stream(MemoryMode.values()) + .flatMap(mode -> + Stream.of( + Arguments.argumentSet( + version + " " + mode + " success", version, mode, true), + Arguments.argumentSet( + version + " " + mode + " failure", version, mode, false)))); + } +}