asafm commented on code in PR #22010: URL: https://github.com/apache/pulsar/pull/22010#discussion_r1481052604
########## pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/PulsarBrokerOpenTelemetry.java: ########## @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.broker.stats; + +import io.opentelemetry.api.metrics.Meter; +import java.io.Closeable; +import lombok.Getter; +import org.apache.pulsar.PulsarVersion; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.common.stats.OpenTelemetryService; + +public class PulsarBrokerOpenTelemetry implements Closeable { + + public static final String SERVICE_NAME = "pulsar-broker"; + private static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.pulsar.broker"; Review Comment: Since you're using it in a single place, and it's private, I would inline it ########## pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/stats/PulsarWorkerOpenTelemetry.java: ########## @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.functions.worker.stats; + +import io.opentelemetry.api.metrics.Meter; +import java.io.Closeable; +import lombok.Getter; +import org.apache.pulsar.PulsarVersion; +import org.apache.pulsar.common.stats.OpenTelemetryService; +import org.apache.pulsar.functions.worker.WorkerConfig; + +public class PulsarWorkerOpenTelemetry implements Closeable { + + public static final String SERVICE_NAME = "pulsar-function-worker"; + private static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.pulsar.function_worker"; Review Comment: Same as the previous comment. Used only in 1 place - I would inline it. ########## pulsar-otel-integration/src/main/java/org/apache/pulsar/common/stats/OpenTelemetryService.java: ########## @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.common.stats; + +import static com.google.common.base.Preconditions.checkArgument; +import com.google.common.annotations.VisibleForTesting; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.resources.Resource; +import java.io.Closeable; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.Consumer; +import lombok.Builder; +import org.apache.commons.lang3.StringUtils; + +/** + * Provides a common OpenTelemetry service for Pulsar components to use. Responsible for instantiating the OpenTelemetry + * SDK with a set of override properties. Once initialized, furnishes access to OpenTelemetry. + */ +public class OpenTelemetryService implements Closeable { + + public static final AttributeKey<String> CLUSTER_ATTRIBUTE = AttributeKey.stringKey("pulsar.cluster"); + public static final AttributeKey<String> SERVICE_NAME_ATTRIBUTE = AttributeKey.stringKey("service.name"); Review Comment: I searched, and it seems auto-generated classes contain all the agreed-upon attribute names (i.e., semantic conventions) as constants. You can find them [here](https://github.com/open-telemetry/semantic-conventions-java/blob/main/src/main/java/io/opentelemetry/semconv/ResourceAttributes.java). I think it's better to use them. This is both for the `service.name` and `service.version`. Regarding cluster and in the future topic attributes. Maybe we should place those in an API designed to be public API. Suppose we place those at `PulsarOtelAttributes`. When Plugin authors in the future will use `OpenTelemetry`, we can guide them to use this. They won't need to know `OpenTelemetryService` as it is an internal detail of Pulsar. ########## pulsar-proxy/src/main/java/org/apache/pulsar/proxy/stats/PulsarProxyOpenTelemetry.java: ########## @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.proxy.stats; + +import io.opentelemetry.api.metrics.Meter; +import java.io.Closeable; +import lombok.Getter; +import org.apache.pulsar.PulsarVersion; +import org.apache.pulsar.common.stats.OpenTelemetryService; +import org.apache.pulsar.proxy.server.ProxyConfiguration; + +public class PulsarProxyOpenTelemetry implements Closeable { + + public static final String SERVICE_NAME = "pulsar-proxy"; + private static final String INSTRUMENTATION_SCOPE_NAME = "org.apache.pulsar.proxy"; Review Comment: Inline ########## pulsar-broker/pom.xml: ########## @@ -470,6 +472,11 @@ <version>${project.version}</version> </dependency> + <dependency> + <groupId>io.rest-assured</groupId> Review Comment: Where are you using this new dependency in this PR in `pulsar-broker`? ########## pulsar-otel-integration/src/test/java/org/apache/pulsar/common/stats/MetricDataMatcher.java: ########## @@ -80,6 +80,17 @@ private boolean matchesDataAttributes(MetricData md) { private boolean matchesAttributes(Attributes actual, Attributes expected) { // Return true iff all attribute pairs in expected are a subset of those in actual. Allows tests to specify // just the attributes they care about, instead of exhaustively having to list all of them. + if (true) { Review Comment: IntelliJ didn't highlight this? ########## pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/PulsarWorkerOpenTelemetry.java: ########## @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.functions.instance.stats; + +import io.opentelemetry.api.metrics.Meter; +import java.io.Closeable; +import lombok.Getter; +import org.apache.pulsar.common.stats.OpenTelemetryService; +import org.apache.pulsar.functions.instance.InstanceConfig; + +public class PulsarWorkerOpenTelemetry implements Closeable { + + private final OpenTelemetryService openTelemetryService; + + @Getter + private final Meter meter; + + public PulsarWorkerOpenTelemetry(InstanceConfig instanceConfig) { + openTelemetryService = OpenTelemetryService.builder().clusterName(instanceConfig.getClusterName()).build(); Review Comment: Regarding 1: Suppose people run in production, with say 2 broker nodes. They send OTLP to collector and from there Prometheus scrapes it. Do we know if this path works because we can differentiate the metrics of the 2 brokers? If so, on which field? I'm asking since I'm not sure how the collector knows the hostname. ########## pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/PulsarWorkerService.java: ########## @@ -188,6 +190,7 @@ public void init(WorkerConfig workerConfig, this.statsUpdater = Executors .newSingleThreadScheduledExecutor(new DefaultThreadFactory("worker-stats-updater")); this.metricsGenerator = new MetricsGenerator(this.statsUpdater, workerConfig); + this.openTelemetry = new PulsarWorkerOpenTelemetry(workerConfig); Review Comment: Shouldn't this also be closed at `stop()`? ########## pulsar-otel-integration/src/main/java/org/apache/pulsar/common/stats/OpenTelemetryService.java: ########## @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.common.stats; + +import static com.google.common.base.Preconditions.checkArgument; +import com.google.common.annotations.VisibleForTesting; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.resources.Resource; +import java.io.Closeable; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.Consumer; +import lombok.Builder; +import org.apache.commons.lang3.StringUtils; + +/** + * Provides a common OpenTelemetry service for Pulsar components to use. Responsible for instantiating the OpenTelemetry + * SDK with a set of override properties. Once initialized, furnishes access to OpenTelemetry. + */ +public class OpenTelemetryService implements Closeable { + + public static final AttributeKey<String> CLUSTER_ATTRIBUTE = AttributeKey.stringKey("pulsar.cluster"); + public static final AttributeKey<String> SERVICE_NAME_ATTRIBUTE = AttributeKey.stringKey("service.name"); + public static final AttributeKey<String> SERVICE_VERSION_ATTRIBUTE = AttributeKey.stringKey("service.version"); + public static final String OTEL_SDK_DISABLED = "otel.sdk.disabled"; Review Comment: Aside from testing, do we need this to be public? Do we expect the Pulsar developers to use this constant? Maybe package access is enough? ########## pulsar-otel-integration/src/test/java/org/apache/pulsar/common/stats/OpenTelemetryServiceTest.java: ########## @@ -0,0 +1,234 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.common.stats; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.LongCounterBuilder; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.MetricDataType; +import io.opentelemetry.sdk.metrics.export.MetricReader; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import lombok.Cleanup; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class OpenTelemetryServiceTest { + + private OpenTelemetryService openTelemetryService; + private InMemoryMetricReader reader; + private Meter meter; + + @BeforeMethod + public void setup() throws Exception { + reader = InMemoryMetricReader.create(); + openTelemetryService = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(reader, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"))). + clusterName("openTelemetryServiceTestCluster"). + build(); + meter = openTelemetryService.getOpenTelemetry().getMeter("openTelemetryServiceTestInstrument"); + } + + @AfterMethod + public void teardown() throws Exception { + openTelemetryService.close(); + reader.close(); + } + + // Customizes the SDK builder to include the MetricReader and extra properties for testing purposes. + private static Consumer<AutoConfigurationCustomizer> getAutoConfigurationCustomizer(MetricReader extraReader, + Pair<String, String>... extraProperties) { + return autoConfigurationCustomizer -> { + if (extraReader != null) { + autoConfigurationCustomizer.addMeterProviderCustomizer( + (sdkMeterProviderBuilder, __) -> sdkMeterProviderBuilder.registerMetricReader(extraReader)); + } + var extraPropertiesMap = + Arrays.stream(extraProperties).collect(Collectors.toMap(Pair::getLeft, Pair::getRight)); + autoConfigurationCustomizer.addPropertiesSupplier(() -> extraPropertiesMap); + }; + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testClusterNameCannotBeNull() throws Exception { + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder().build(); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testClusterNameCannotBeEmpty() throws Exception { + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder().clusterName(StringUtils.EMPTY).build(); + } + + @Test + public void testIsClusterNameSet() throws Exception { + @Cleanup + InMemoryMetricReader reader = InMemoryMetricReader.create(); + + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(reader, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"))). + clusterName("testCluster"). + build(); + + Predicate<MetricData> predicate = MetricDataMatcher.builder(). + resourceAttribute(Attributes.of(AttributeKey.stringKey("pulsar.cluster"), "testCluster")). + build(); + + Collection<MetricData> metricData = reader.collectAllMetrics(); + assertTrue(metricData.stream().anyMatch(predicate)); + } + + @Test + public void testIsServiceNameAndVersionSet() throws Exception { + @Cleanup + var reader = InMemoryMetricReader.create(); + + @Cleanup + var ots = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(reader, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"))). + clusterName("testServiceNameAndVersion"). + serviceName("openTelemetryServiceTestService"). + serviceVersion("1.0.0"). + build(); + + var predicate = MetricDataMatcher.builder(). + resourceAttribute(Attributes.of( + AttributeKey.stringKey("pulsar.cluster"), "testServiceNameAndVersion", + AttributeKey.stringKey("service.name"), "openTelemetryServiceTestService", + AttributeKey.stringKey("service.version"), "1.0.0")). + build(); + + var metricData = reader.collectAllMetrics(); + assertTrue(metricData.stream().anyMatch(predicate)); + } + + @Test + public void testIsInstrumentationNameSetOnMeter() throws Exception { + Meter meter = openTelemetryService.getOpenTelemetry().getMeter("testInstrumentationScope"); + meter.counterBuilder("dummyCounter").build().add(1); + MetricDataMatcher predicate = MetricDataMatcher.builder(). + name("dummyCounter"). + instrumentationScopeInfo(InstrumentationScopeInfo.create("testInstrumentationScope")). + build(); + Collection<MetricData> metricData = reader.collectAllMetrics(); + assertTrue(metricData.stream().anyMatch(predicate)); + } + + @Test + public void testMetricCardinalityIsSet() throws Exception { + var prometheusExporterPort = 9464; + @Cleanup + var ots = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(null, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"), Review Comment: 1. This comment is relevant to all places we use Pair: How about we do this: ``` autoConfigurationCustomizer(getAutoConfigurationCustomizer(null, Map.of( OpenTelemetryService.OTEL_SDK_DISABLED, "false", "otel.metrics.exporter", "prometheus", "otel.exporter.prometheus.port", Integer.toString(prometheusExporterPort), "otel.metric.export.interval", "100")) ``` It's available since Java 9, so it should work. I remembered this when I saw the `var` :) ########## tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/OpenTelemetryCollectorContainer.java: ########## @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.tests.integration.containers; + +import java.time.Duration; +import org.apache.http.HttpStatus; +import org.testcontainers.containers.wait.strategy.HttpWaitStrategy; +import org.testcontainers.utility.MountableFile; + +public class OpenTelemetryCollectorContainer extends ChaosContainer<OpenTelemetryCollectorContainer> { + + private static final String IMAGE_NAME = "otel/opentelemetry-collector-contrib:latest"; + private static final String NAME = "otel-collector"; + + public static final int PROMETHEUS_EXPORTER_PORT = 8889; + private static final int OTLP_RECEIVER_PORT = 4317; + private static final int ZPAGES_PORT = 55679; + + public OpenTelemetryCollectorContainer(String clusterName) { + super(clusterName, IMAGE_NAME); + } + + @Override + protected void configure() { + super.configure(); + + this.withCopyFileToContainer( + MountableFile.forClasspathResource("containers/otel-collector-config.yaml", 0644), + "/etc/otel-collector-config.yaml") + .withCommand("--config=/etc/otel-collector-config.yaml") + .withExposedPorts(OTLP_RECEIVER_PORT, PROMETHEUS_EXPORTER_PORT, ZPAGES_PORT) + .withCreateContainerCmdModifier(createContainerCmd -> { + createContainerCmd.withHostName(NAME); Review Comment: I mean, why do you need the line `withHostName(NAME)` ########## pulsar-otel-integration/src/main/java/org/apache/pulsar/common/stats/OpenTelemetryService.java: ########## @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.common.stats; + +import static com.google.common.base.Preconditions.checkArgument; +import com.google.common.annotations.VisibleForTesting; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.resources.Resource; +import java.io.Closeable; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.Consumer; +import lombok.Builder; +import org.apache.commons.lang3.StringUtils; + +/** + * Provides a common OpenTelemetry service for Pulsar components to use. Responsible for instantiating the OpenTelemetry + * SDK with a set of override properties. Once initialized, furnishes access to OpenTelemetry. + */ +public class OpenTelemetryService implements Closeable { + + public static final AttributeKey<String> CLUSTER_ATTRIBUTE = AttributeKey.stringKey("pulsar.cluster"); + public static final AttributeKey<String> SERVICE_NAME_ATTRIBUTE = AttributeKey.stringKey("service.name"); + public static final AttributeKey<String> SERVICE_VERSION_ATTRIBUTE = AttributeKey.stringKey("service.version"); + public static final String OTEL_SDK_DISABLED = "otel.sdk.disabled"; + private static final String MAX_CARDINALITY_LIMIT_KEY = "otel.experimental.metrics.cardinality.limit"; + public static final int MAX_CARDINALITY_LIMIT = 10000; + + private final OpenTelemetrySdk openTelemetrySdk; + + @Builder + public OpenTelemetryService(String clusterName, + String serviceName, + String serviceVersion, + // Allows customizing the SDK builder; for testing purposes only. + @VisibleForTesting Consumer<AutoConfigurationCustomizer> autoConfigurationCustomizer) { Review Comment: I find this confusing: 1. AutoConfigurationCustomizer is used (currently) only via SPI. Seeing that we're using it, for our own purpose is confusing, and might confuse future Pulsar developers. 2. A consumer of AutoConfigurationCustomizer is very confusing. If it were only `AutoConfigurationCustomizer`, it would make more sense IMO. I suggest `Consumer<AutoConfiguredOpenTelemetrySdkBuilder>`. ########## pulsar-otel-integration/src/test/java/org/apache/pulsar/common/stats/OpenTelemetryServiceTest.java: ########## @@ -0,0 +1,234 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.common.stats; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.LongCounterBuilder; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.MetricDataType; +import io.opentelemetry.sdk.metrics.export.MetricReader; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import lombok.Cleanup; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class OpenTelemetryServiceTest { + + private OpenTelemetryService openTelemetryService; + private InMemoryMetricReader reader; + private Meter meter; + + @BeforeMethod + public void setup() throws Exception { + reader = InMemoryMetricReader.create(); + openTelemetryService = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(reader, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"))). + clusterName("openTelemetryServiceTestCluster"). + build(); + meter = openTelemetryService.getOpenTelemetry().getMeter("openTelemetryServiceTestInstrument"); + } + + @AfterMethod + public void teardown() throws Exception { + openTelemetryService.close(); + reader.close(); + } + + // Customizes the SDK builder to include the MetricReader and extra properties for testing purposes. + private static Consumer<AutoConfigurationCustomizer> getAutoConfigurationCustomizer(MetricReader extraReader, + Pair<String, String>... extraProperties) { + return autoConfigurationCustomizer -> { + if (extraReader != null) { + autoConfigurationCustomizer.addMeterProviderCustomizer( + (sdkMeterProviderBuilder, __) -> sdkMeterProviderBuilder.registerMetricReader(extraReader)); + } + var extraPropertiesMap = + Arrays.stream(extraProperties).collect(Collectors.toMap(Pair::getLeft, Pair::getRight)); + autoConfigurationCustomizer.addPropertiesSupplier(() -> extraPropertiesMap); + }; + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testClusterNameCannotBeNull() throws Exception { + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder().build(); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testClusterNameCannotBeEmpty() throws Exception { + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder().clusterName(StringUtils.EMPTY).build(); + } + + @Test + public void testIsClusterNameSet() throws Exception { + @Cleanup + InMemoryMetricReader reader = InMemoryMetricReader.create(); + + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(reader, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"))). + clusterName("testCluster"). + build(); + + Predicate<MetricData> predicate = MetricDataMatcher.builder(). + resourceAttribute(Attributes.of(AttributeKey.stringKey("pulsar.cluster"), "testCluster")). + build(); + + Collection<MetricData> metricData = reader.collectAllMetrics(); + assertTrue(metricData.stream().anyMatch(predicate)); + } + + @Test + public void testIsServiceNameAndVersionSet() throws Exception { + @Cleanup + var reader = InMemoryMetricReader.create(); + + @Cleanup + var ots = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(reader, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"))). + clusterName("testServiceNameAndVersion"). + serviceName("openTelemetryServiceTestService"). + serviceVersion("1.0.0"). + build(); + + var predicate = MetricDataMatcher.builder(). + resourceAttribute(Attributes.of( + AttributeKey.stringKey("pulsar.cluster"), "testServiceNameAndVersion", + AttributeKey.stringKey("service.name"), "openTelemetryServiceTestService", + AttributeKey.stringKey("service.version"), "1.0.0")). + build(); + + var metricData = reader.collectAllMetrics(); + assertTrue(metricData.stream().anyMatch(predicate)); + } + + @Test + public void testIsInstrumentationNameSetOnMeter() throws Exception { + Meter meter = openTelemetryService.getOpenTelemetry().getMeter("testInstrumentationScope"); + meter.counterBuilder("dummyCounter").build().add(1); + MetricDataMatcher predicate = MetricDataMatcher.builder(). + name("dummyCounter"). + instrumentationScopeInfo(InstrumentationScopeInfo.create("testInstrumentationScope")). + build(); + Collection<MetricData> metricData = reader.collectAllMetrics(); + assertTrue(metricData.stream().anyMatch(predicate)); + } + + @Test + public void testMetricCardinalityIsSet() throws Exception { + var prometheusExporterPort = 9464; + @Cleanup + var ots = OpenTelemetryService.builder(). + autoConfigurationCustomizer(getAutoConfigurationCustomizer(null, + Pair.of(OpenTelemetryService.OTEL_SDK_DISABLED, "false"), + Pair.of("otel.metrics.exporter", "prometheus"), + Pair.of("otel.exporter.prometheus.port", Integer.toString(prometheusExporterPort)), + Pair.of("otel.metric.export.interval", "100"))). Review Comment: This is relevant only to periodic metric reader hence not required here. ########## pulsar-otel-metrics-provider/src/test/java/org/apache/pulsar/common/stats/OpenTelemetryServiceTest.java: ########## @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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.apache.pulsar.common.stats; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.DoubleCounter; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; +import io.opentelemetry.sdk.common.InstrumentationScopeInfo; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.MetricDataType; +import io.opentelemetry.sdk.metrics.export.MetricReader; +import io.opentelemetry.sdk.metrics.internal.SdkMeterProviderUtil; +import io.opentelemetry.sdk.metrics.internal.state.MetricStorage; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.util.Collection; +import java.util.function.Predicate; +import lombok.Cleanup; +import org.apache.commons.lang3.StringUtils; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +public class OpenTelemetryServiceTest { + + private OpenTelemetryService openTelemetryService; + private InMemoryMetricReader reader; + private Meter meter; + + @BeforeMethod + public void setup() throws Exception { + reader = InMemoryMetricReader.create(); + openTelemetryService = OpenTelemetryService.builder(). + sdkBuilder(getSdkBuilder(reader)). + clusterName("openTelemetryServiceTestCluster"). + extraProperty(OpenTelemetryService.OTEL_SDK_DISABLED, "false"). + build(); + meter = openTelemetryService.getMeter("openTelemetryServiceTestInstrument"); + } + + @AfterMethod + public void teardown() throws Exception { + openTelemetryService.close(); + reader.close(); + } + + // Overrides the default sdkBuilder to include the InMemoryMetricReader for testing purposes. + private static AutoConfiguredOpenTelemetrySdkBuilder getSdkBuilder(MetricReader extraReader) { + return AutoConfiguredOpenTelemetrySdk.builder(). + addMeterProviderCustomizer((sdkMeterProviderBuilder, configProperties) -> { + SdkMeterProviderUtil.registerMetricReaderWithCardinalitySelector( + sdkMeterProviderBuilder, extraReader, + // Override the max cardinality limit for this extra reader. + instrumentType -> OpenTelemetryService.MAX_CARDINALITY_LIMIT + 1); + return sdkMeterProviderBuilder; + }); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testClusterNameCannotBeNull() throws Exception { + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder().build(); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testClusterNameCannotBeEmpty() throws Exception { + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder().clusterName(StringUtils.EMPTY).build(); + } + + @Test + public void testIsClusterNameSet() throws Exception { + @Cleanup + InMemoryMetricReader reader = InMemoryMetricReader.create(); + + @Cleanup + OpenTelemetryService ots = OpenTelemetryService.builder(). + sdkBuilder(getSdkBuilder(reader)). + clusterName("testCluster"). + extraProperty(OpenTelemetryService.OTEL_SDK_DISABLED, "false"). + build(); + + Predicate<MetricData> predicate = MetricDataMatcher.builder(). + resourceAttribute(Attributes.of(AttributeKey.stringKey("pulsar.cluster"), "testCluster")). + build(); + + Collection<MetricData> metricData = reader.collectAllMetrics(); + Assert.assertTrue(metricData.stream().anyMatch(predicate)); Review Comment: [AssertJ](https://assertj.github.io/doc/) biggest advantage is its error messages. They are better than the JUnit Assert ones. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: commits-unsubscr...@pulsar.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
