Skip to content

Commit cce51bc

Browse files
jeanbisuttijeanbisutti
authored
Fix "java - monitor-opentelemetry-exporter - tests" pipeline (Azure#44522)
1 parent 5806027 commit cce51bc

File tree

4 files changed

+159
-27
lines changed

4 files changed

+159
-27
lines changed

‎sdk/monitor/azure-monitor-opentelemetry-exporter/pom.xml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@
4242
<!-- Configures the Java 9+ run to perform the required module exports, opens, and reads that are necessary for testing but shouldn't be part of the module-info. -->
4343
<javaModulesSurefireArgLine>
4444
--add-opens com.azure.monitor.opentelemetry.exporter/com.azure.monitor.opentelemetry.exporter=ALL-UNNAMED
45+
--add-exports com.azure.core/com.azure.core.implementation.util=ALL-UNNAMED
46+
--add-reads com.azure.monitor.opentelemetry.autoconfigure=com.azure.core.tracing.opentelemetry
4547
</javaModulesSurefireArgLine>
4648

4749
<checkstyle.suppressionsLocation>checkstyle-suppressions.xml</checkstyle.suppressionsLocation>

‎sdk/monitor/azure-monitor-opentelemetry-exporter/src/test/java/com/azure/monitor/opentelemetry/exporter/AppConfigurationExporterIntegrationTest.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,15 @@
33

44
package com.azure.monitor.opentelemetry.exporter;
55

6+
import com.azure.core.credential.TokenCredential;
67
import com.azure.core.http.HttpPipelineCallContext;
78
import com.azure.core.http.HttpPipelineNextPolicy;
89
import com.azure.core.http.HttpResponse;
910
import com.azure.core.http.policy.HttpLogDetailLevel;
1011
import com.azure.core.http.policy.HttpLogOptions;
1112
import com.azure.core.http.policy.HttpPipelinePolicy;
1213
import com.azure.core.test.annotation.LiveOnly;
14+
import com.azure.core.util.Configuration;
1315
import com.azure.core.util.Context;
1416
import com.azure.core.util.FluxUtil;
1517
import com.azure.data.appconfiguration.ConfigurationClient;
@@ -37,6 +39,15 @@
3739

3840
@LiveOnly
3941
public class AppConfigurationExporterIntegrationTest extends MonitorExporterClientTestBase {
42+
43+
private TokenCredential credential;
44+
45+
@Override
46+
public void beforeTest() {
47+
super.beforeTest();
48+
credential = TokenCredentialUtil.getTestTokenCredential(interceptorManager);
49+
}
50+
4051
@Test
4152
public void setConfigurationTest() throws InterruptedException {
4253
CountDownLatch exporterCountDown = new CountDownLatch(1);
@@ -87,8 +98,10 @@ public void testDisableTracing() throws InterruptedException {
8798
assertTrue(exporterCountDown.await(60, TimeUnit.SECONDS));
8899
}
89100

90-
private static ConfigurationClient getConfigurationClient() {
91-
return new ConfigurationClientBuilder().connectionString(System.getenv("AZURE_APPCONFIG_CONNECTION_STRING"))
101+
private ConfigurationClient getConfigurationClient() {
102+
String endPoint = Configuration.getGlobalConfiguration().get("AZURE_APPCONFIG_ENDPOINT");
103+
return new ConfigurationClientBuilder().credential(credential)
104+
.endpoint(endPoint)
92105
.httpLogOptions(new HttpLogOptions().setLogLevel(HttpLogDetailLevel.BODY_AND_HEADERS))
93106
.buildClient();
94107
}

‎sdk/monitor/azure-monitor-opentelemetry-exporter/src/test/java/com/azure/monitor/opentelemetry/exporter/EventHubsExporterIntegrationTest.java‎

Lines changed: 84 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,18 @@
33

44
package com.azure.monitor.opentelemetry.exporter;
55

6+
import com.azure.core.credential.TokenCredential;
7+
import com.azure.core.http.HttpPipelineCallContext;
8+
import com.azure.core.http.HttpPipelineNextPolicy;
9+
import com.azure.core.http.HttpPipelineNextSyncPolicy;
10+
import com.azure.core.http.HttpResponse;
611
import com.azure.core.http.policy.HttpPipelinePolicy;
712
import com.azure.core.test.annotation.LiveOnly;
13+
import com.azure.core.tracing.opentelemetry.OpenTelemetryTracingOptions;
14+
import com.azure.core.util.ClientOptions;
815
import com.azure.core.util.FluxUtil;
16+
import com.azure.core.util.logging.ClientLogger;
17+
import com.azure.core.util.Configuration;
918
import com.azure.messaging.eventhubs.EventData;
1019
import com.azure.messaging.eventhubs.EventHubClientBuilder;
1120
import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
@@ -14,17 +23,25 @@
1423
import com.azure.messaging.eventhubs.LoadBalancingStrategy;
1524
import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
1625
import com.azure.messaging.eventhubs.models.CreateBatchOptions;
26+
import com.azure.monitor.opentelemetry.exporter.implementation.localstorage.LocalStorageTelemetryPipelineListener;
27+
import com.azure.monitor.opentelemetry.exporter.implementation.models.MonitorDomain;
28+
import com.azure.monitor.opentelemetry.exporter.implementation.models.RemoteDependencyData;
29+
import com.azure.monitor.opentelemetry.exporter.implementation.models.TelemetryItem;
1730
import com.azure.monitor.opentelemetry.exporter.implementation.utils.TestUtils;
1831
import com.azure.storage.blob.BlobContainerAsyncClient;
1932
import com.azure.storage.blob.BlobContainerClientBuilder;
33+
import io.opentelemetry.api.OpenTelemetry;
2034
import io.opentelemetry.api.trace.Span;
2135
import io.opentelemetry.api.trace.Tracer;
2236
import io.opentelemetry.context.Scope;
2337
import org.junit.jupiter.api.Disabled;
2438
import org.junit.jupiter.api.Test;
2539
import reactor.core.publisher.Mono;
40+
import reactor.test.StepVerifier;
2641

2742
import java.nio.charset.StandardCharsets;
43+
import java.time.Duration;
44+
import java.util.List;
2845
import java.util.concurrent.CountDownLatch;
2946
import java.util.concurrent.TimeUnit;
3047

@@ -37,38 +54,80 @@ public class EventHubsExporterIntegrationTest extends MonitorExporterClientTestB
3754
private static final String STORAGE_CONNECTION_STRING = System.getenv("STORAGE_CONNECTION_STRING");
3855
private static final String CONTAINER_NAME = System.getenv("STORAGE_CONTAINER_NAME");
3956

57+
private static final ClientLogger LOGGER = new ClientLogger(EventHubsExporterIntegrationTest.class);
58+
59+
private TokenCredential credential;
60+
61+
@Override
62+
public void beforeTest() {
63+
super.beforeTest();
64+
credential = TokenCredentialUtil.getTestTokenCredential(interceptorManager);
65+
}
66+
4067
@Test
68+
@SuppressWarnings("try")
4169
public void producerTest() throws InterruptedException {
70+
String ehNamespace = Configuration.getGlobalConfiguration().get("AZURE_EVENTHUBS_FULLY_QUALIFIED_DOMAIN_NAME");
71+
String ehName = Configuration.getGlobalConfiguration().get("AZURE_EVENTHUBS_EVENT_HUB_NAME");
72+
4273
CountDownLatch exporterCountDown = new CountDownLatch(2);
4374
String spanName = "event-hubs-producer-testing";
44-
HttpPipelinePolicy validationPolicy = (context, next) -> {
45-
Mono<String> asyncString = FluxUtil.collectBytesInByteBufferStream(context.getHttpRequest().getBody())
46-
.map(bytes -> new String(bytes, StandardCharsets.UTF_8));
47-
asyncString.subscribe(value -> {
48-
if (value.contains(spanName)) {
49-
exporterCountDown.countDown();
50-
}
51-
if (value.contains("EventHubs.send")) {
52-
exporterCountDown.countDown();
75+
HttpPipelinePolicy validationPolicy = new HttpPipelinePolicy() {
76+
@Override
77+
public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineNextPolicy next) {
78+
checkTelemetry(context);
79+
return next.process();
80+
}
81+
82+
@Override
83+
public HttpResponse processSync(HttpPipelineCallContext context, HttpPipelineNextSyncPolicy next) {
84+
checkTelemetry(context);
85+
return next.processSync();
86+
}
87+
88+
private void checkTelemetry(HttpPipelineCallContext context) {
89+
byte[] asyncBytes = LocalStorageTelemetryPipelineListener
90+
.ungzip(context.getHttpRequest().getBodyAsBinaryData().toBytes());
91+
List<TelemetryItem> telemetryItems = TestUtils.deserialize(asyncBytes);
92+
93+
for (TelemetryItem telemetryItem : telemetryItems) {
94+
MonitorDomain monitorDomain = telemetryItem.getData().getBaseData();
95+
RemoteDependencyData remoteDependencyData = TestUtils.toRemoteDependencyData(monitorDomain);
96+
String remoteDependencyName = remoteDependencyData.getName();
97+
if (remoteDependencyName.contains(spanName)) {
98+
exporterCountDown.countDown();
99+
LOGGER.info("Count down " + spanName);
100+
} else if (("send " + ehName).equals(remoteDependencyName)) {
101+
exporterCountDown.countDown();
102+
LOGGER.info("Count down eventHubs send");
103+
} else {
104+
LOGGER.info("remoteDependencyName = " + remoteDependencyName);
105+
}
53106
}
54-
});
55-
return next.process();
107+
}
56108
};
57-
Tracer tracer = TestUtils.createOpenTelemetrySdk(getHttpPipeline(validationPolicy)).getTracer("Sample");
58-
EventHubProducerAsyncClient producer
59-
= new EventHubClientBuilder().connectionString(CONNECTION_STRING).buildAsyncProducerClient();
60-
Span span = tracer.spanBuilder(spanName).startSpan();
61-
Scope scope = span.makeCurrent();
62-
try {
63-
producer.createBatch().flatMap(batch -> {
64-
batch.tryAdd(new EventData("test event"));
65-
return producer.send(batch);
66-
}).subscribe();
67-
} finally {
68-
span.end();
69-
scope.close();
109+
110+
OpenTelemetry otel = TestUtils.createOpenTelemetrySdk(getHttpPipeline(validationPolicy));
111+
Tracer tracer = otel.getTracer("Sample");
112+
113+
try (EventHubProducerAsyncClient producer = new EventHubClientBuilder().credential(credential)
114+
.fullyQualifiedNamespace(ehNamespace)
115+
.eventHubName(ehName)
116+
.clientOptions(
117+
new ClientOptions().setTracingOptions(new OpenTelemetryTracingOptions().setOpenTelemetry(otel)))
118+
.buildAsyncProducerClient()) {
119+
120+
Span span = tracer.spanBuilder(spanName).startSpan();
121+
try (Scope scope = span.makeCurrent()) {
122+
StepVerifier.create(producer.createBatch().flatMap(batch -> {
123+
batch.tryAdd(new EventData("test event"));
124+
return producer.send(batch);
125+
})).expectComplete().verify(Duration.ofSeconds(60));
126+
} finally {
127+
span.end();
128+
}
70129
}
71-
assertTrue(exporterCountDown.await(5, TimeUnit.SECONDS));
130+
assertTrue(exporterCountDown.await(20, TimeUnit.SECONDS));
72131
}
73132

74133
@Disabled("Processor integration tests require separate consumer group to not have partition contention in CI - https://github.com/Azure/azure-sdk-for-java/issues/23567")

‎sdk/monitor/azure-monitor-opentelemetry-exporter/src/test/java/com/azure/monitor/opentelemetry/exporter/TokenCredentialUtil.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
package com.azure.monitor.opentelemetry.exporter;
5+
6+
import com.azure.core.credential.TokenCredential;
7+
import com.azure.core.test.InterceptorManager;
8+
import com.azure.core.test.utils.MockTokenCredential;
9+
import com.azure.core.util.Configuration;
10+
import com.azure.core.util.CoreUtils;
11+
import com.azure.identity.AzurePipelinesCredentialBuilder;
12+
import com.azure.identity.DefaultAzureCredentialBuilder;
13+
import reactor.core.publisher.Mono;
14+
import reactor.core.scheduler.Schedulers;
15+
16+
class TokenCredentialUtil {
17+
18+
/**
19+
* Gets a token credential for use in tests.
20+
* @param interceptorManager the interceptor manager
21+
* @return the TokenCredential
22+
*/
23+
static TokenCredential getTestTokenCredential(InterceptorManager interceptorManager) {
24+
if (interceptorManager.isLiveMode()) {
25+
return getPipelineCredential();
26+
} else if (interceptorManager.isRecordMode()) {
27+
return new DefaultAzureCredentialBuilder().build();
28+
} else {
29+
return new MockTokenCredential();
30+
}
31+
}
32+
33+
private static TokenCredential getPipelineCredential() {
34+
final String serviceConnectionId = getPropertyValue("AZURESUBSCRIPTION_SERVICE_CONNECTION_ID");
35+
final String clientId = getPropertyValue("AZURESUBSCRIPTION_CLIENT_ID");
36+
final String tenantId = getPropertyValue("AZURESUBSCRIPTION_TENANT_ID");
37+
final String systemAccessToken = getPropertyValue("SYSTEM_ACCESSTOKEN");
38+
39+
if (CoreUtils.isNullOrEmpty(serviceConnectionId)
40+
|| CoreUtils.isNullOrEmpty(clientId)
41+
|| CoreUtils.isNullOrEmpty(tenantId)
42+
|| CoreUtils.isNullOrEmpty(systemAccessToken)) {
43+
return null;
44+
}
45+
46+
TokenCredential cred = new AzurePipelinesCredentialBuilder().systemAccessToken(systemAccessToken)
47+
.clientId(clientId)
48+
.tenantId(tenantId)
49+
.serviceConnectionId(serviceConnectionId)
50+
.build();
51+
52+
return request -> Mono.defer(() -> cred.getToken(request)).subscribeOn(Schedulers.boundedElastic());
53+
}
54+
55+
private static String getPropertyValue(String propertyName) {
56+
return Configuration.getGlobalConfiguration().get(propertyName, System.getenv(propertyName));
57+
}
58+
}

0 commit comments

Comments
 (0)