diff --git a/sdk-tests/pom.xml b/sdk-tests/pom.xml index b274bf9907..467e7ce06c 100644 --- a/sdk-tests/pom.xml +++ b/sdk-tests/pom.xml @@ -30,6 +30,7 @@ ${project.basedir}/proto 1.33.1 3.13.0 + 0.14.0 @@ -63,6 +64,21 @@ protoc-jar-maven-plugin 3.10.1 + + io.opentelemetry + opentelemetry-sdk + ${opentelemetry.version} + + + io.opentelemetry + opentelemetry-exporter-logging + ${opentelemetry.version} + + + io.opentelemetry + opentelemetry-exporter-zipkin + ${opentelemetry.version} + io.dapr dapr-sdk @@ -105,6 +121,12 @@ 2.3.5.RELEASE test + + com.jayway.jsonpath + json-path + 2.4.0 + test + com.bettercloud diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/OpenTelemetry.java b/sdk-tests/src/test/java/io/dapr/it/tracing/OpenTelemetry.java new file mode 100644 index 0000000000..7b8c7a77c6 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/OpenTelemetry.java @@ -0,0 +1,100 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.context.propagation.TextMapPropagator; +import io.opentelemetry.exporter.zipkin.ZipkinSpanExporter; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; +import org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.io.IOException; +import java.net.Socket; +import java.util.HashMap; +import java.util.Map; + +public class OpenTelemetry { + + private static final int ZIPKIN_PORT = 9411; + + private static final String ENDPOINT_V2_SPANS = "/api/v2/spans"; + + /** + * Creates an opentelemetry instance. + * @param serviceName Name of the service in Zipkin + * @return OpenTelemetry. + */ + public static io.opentelemetry.api.OpenTelemetry createOpenTelemetry(String serviceName) { + // Only exports to Zipkin if it is up. Otherwise, ignore it. + // This is helpful to avoid exceptions for examples that do not require Zipkin. + if (isZipkinUp()) { + String httpUrl = String.format("http://localhost:%d", ZIPKIN_PORT); + ZipkinSpanExporter zipkinExporter = + ZipkinSpanExporter.builder() + .setEndpoint(httpUrl + ENDPOINT_V2_SPANS) + .setServiceName(serviceName) + .build(); + + SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(zipkinExporter)) + .build(); + + return OpenTelemetrySdk.builder() + .setTracerProvider(sdkTracerProvider) + .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance())) + .buildAndRegisterGlobal(); + } else { + System.out.println("WARNING: Zipkin is not available."); + } + + return null; + } + + /** + * Converts current OpenTelemetry's context into Reactor's context. + * @return Reactor's context. + */ + public static reactor.util.context.Context getReactorContext() { + return getReactorContext(Context.current()); + } + + /** + * Converts given OpenTelemetry's context into Reactor's context. + * @param context OpenTelemetry's context. + * @return Reactor's context. + */ + public static reactor.util.context.Context getReactorContext(Context context) { + Map map = new HashMap<>(); + TextMapPropagator.Setter> setter = + (carrier, key, value) -> map.put(key, value); + + GlobalOpenTelemetry.getPropagators().getTextMapPropagator().inject(context, map, setter); + reactor.util.context.Context reactorContext = reactor.util.context.Context.empty(); + for (Map.Entry entry : map.entrySet()) { + reactorContext = reactorContext.put(entry.getKey(), entry.getValue()); + } + return reactorContext; + } + + private static boolean isZipkinUp() { + try (Socket ignored = new Socket("localhost", ZIPKIN_PORT)) { + return true; + } catch (IOException ignored) { + return false; + } + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java b/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java new file mode 100644 index 0000000000..a50b806d0f --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/Validation.java @@ -0,0 +1,70 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing; + +import com.jayway.jsonpath.DocumentContext; +import com.jayway.jsonpath.JsonPath; +import net.minidev.json.JSONArray; +import okhttp3.HttpUrl; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * Class used to verify that traces are present as expected. + * + * Checks for the main span and then checks for its child span for `sleep` API call. + */ +public final class Validation { + + private static final OkHttpClient HTTP_CLIENT = new OkHttpClient(); + + /** + * JSON Path for main span Id. + */ + public static final String JSONPATH_MAIN_SPAN_ID = "$..[?(@.name == \"%s\")]['id']"; + + /** + * JSON Path for child span Id where duration is greater than 1s. + */ + public static final String JSONPATH_SLEEP_SPAN_ID = + "$..[?(@.parentId=='%s' && @.duration > 1000000 && @.name=='%s')]['id']"; + + public static void validate(String spanName, String sleepSpanName) throws Exception { + // Must wait for some time to make sure Zipkin receives all spans. + Thread.sleep(5000); + HttpUrl.Builder urlBuilder = new HttpUrl.Builder(); + urlBuilder.scheme("http") + .host("localhost") + .port(9411) + .addPathSegments("api/v2/traces"); + Request.Builder requestBuilder = new Request.Builder() + .url(urlBuilder.build()); + requestBuilder.method("GET", null); + + Request request = requestBuilder.build(); + + Response response = HTTP_CLIENT.newCall(request).execute(); + DocumentContext documentContext = JsonPath.parse(response.body().string()); + String mainSpanId = readOne(documentContext, String.format(JSONPATH_MAIN_SPAN_ID, spanName)).toString(); + assertNotNull(mainSpanId); + + String sleepSpanId = readOne(documentContext, String.format(JSONPATH_SLEEP_SPAN_ID, mainSpanId, sleepSpanName)) + .toString(); + assertNotNull(sleepSpanId); + } + + private static Object readOne(DocumentContext documentContext, String path) { + JSONArray arr = documentContext.read(path); + assertTrue(arr.size() > 0); + + return arr.get(0); + } + +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java b/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java new file mode 100644 index 0000000000..d47d012767 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/Service.java @@ -0,0 +1,133 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing.grpc; + +import com.google.protobuf.Any; +import io.dapr.v1.AppCallbackGrpc; +import io.dapr.v1.CommonProtos; +import io.grpc.Server; +import io.grpc.ServerBuilder; +import io.grpc.stub.StreamObserver; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequest; +import static io.dapr.it.MethodInvokeServiceProtos.DeleteMessageResponse; +import static io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest; +import static io.dapr.it.MethodInvokeServiceProtos.GetMessagesResponse; +import static io.dapr.it.MethodInvokeServiceProtos.PostMessageRequest; +import static io.dapr.it.MethodInvokeServiceProtos.PostMessageResponse; +import static io.dapr.it.MethodInvokeServiceProtos.SleepRequest; +import static io.dapr.it.MethodInvokeServiceProtos.SleepResponse; + +public class Service { + + public static final String SUCCESS_MESSAGE = "application discovered on port "; + + /** + * Server mode: class that encapsulates all server-side logic for Grpc. + */ + private static class MyDaprService extends AppCallbackGrpc.AppCallbackImplBase { + + /** + * Server mode: Grpc server. + */ + private Server server; + + /** + * Server mode: starts listening on given port. + * + * @param port Port to listen on. + * @throws IOException Errors while trying to start service. + */ + private void start(int port) throws IOException { + this.server = ServerBuilder + .forPort(port) + .addService(this) + .build() + .start(); + System.out.printf("Server: started listening on port %d\n", port); + + // Now we handle ctrl+c (or any other JVM shutdown) + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("Server: shutting down gracefully ..."); + MyDaprService.this.server.shutdown(); + System.out.println("Server: Bye."); + })); + } + + /** + * Server mode: waits for shutdown trigger. + * + * @throws InterruptedException Propagated interrupted exception. + */ + private void awaitTermination() throws InterruptedException { + if (this.server != null) { + this.server.awaitTermination(); + } + } + + /** + * Server mode: this is the Dapr method to receive Invoke operations via Grpc. + * + * @param request Dapr envelope request, + * @param responseObserver Dapr envelope response. + */ + @Override + public void onInvoke(CommonProtos.InvokeRequest request, + StreamObserver responseObserver) { + try { + if ("sleepOverGRPC".equals(request.getMethod())) { + SleepRequest req = SleepRequest.parseFrom(request.getData().getValue().toByteArray()); + + SleepResponse res = this.sleep(req); + + CommonProtos.InvokeResponse.Builder responseBuilder = CommonProtos.InvokeResponse.newBuilder(); + responseBuilder.setData(Any.pack(res)); + responseObserver.onNext(responseBuilder.build()); + } + } catch (Exception e) { + responseObserver.onError(e); + } finally { + responseObserver.onCompleted(); + } + } + + public SleepResponse sleep(SleepRequest request) { + if (request.getSeconds() < 0) { + throw new IllegalArgumentException("Sleep time cannot be negative."); + } + + try { + Thread.sleep(request.getSeconds() * 1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + new RuntimeException(e); + } + + // Now respond with current timestamp. + return SleepResponse.newBuilder().build(); + } + } + + /** + * This is the main method of this app. + * @param args The port to listen on. + * @throws Exception An Exception. + */ + public static void main(String[] args) throws Exception { + int port = Integer.parseInt(args[0]); + + System.out.printf("Service starting on port %d ...\n", port); + + final MyDaprService service = new MyDaprService(); + service.start(port); + service.awaitTermination(); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/TracingIT.java b/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/TracingIT.java new file mode 100644 index 0000000000..a40874f92e --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/grpc/TracingIT.java @@ -0,0 +1,91 @@ +package io.dapr.it.tracing.grpc; + +import io.dapr.client.DaprApiProtocol; +import io.dapr.client.DaprClient; +import io.dapr.client.DaprClientBuilder; +import io.dapr.client.domain.HttpExtension; +import io.dapr.it.BaseIT; +import io.dapr.it.DaprRun; +import io.dapr.it.tracing.Validation; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import java.util.Arrays; +import java.util.Collection; +import java.util.UUID; + +import static io.dapr.it.MethodInvokeServiceProtos.SleepRequest; +import static io.dapr.it.tracing.OpenTelemetry.createOpenTelemetry; +import static io.dapr.it.tracing.OpenTelemetry.getReactorContext; +import static org.junit.runners.Parameterized.Parameter; +import static org.junit.runners.Parameterized.Parameters; + +@RunWith(Parameterized.class) +public class TracingIT extends BaseIT { + + /** + * Parameters for this test. + * Param #1: useGrpc. + * @return Collection of parameter tuples. + */ + @Parameters + public static Collection data() { + return Arrays.asList(new Object[][] { { false }, { true } }); + } + + /** + * Run of a Dapr application. + */ + private DaprRun daprRun = null; + + @Parameter + public boolean useGrpc; + + @Before + public void init() throws Exception { + daprRun = startDaprApp( + TracingIT.class.getSimpleName(), + Service.SUCCESS_MESSAGE, + Service.class, + DaprApiProtocol.GRPC, // appProtocol + 60000); + + if (this.useGrpc) { + daprRun.switchToGRPC(); + } else { + daprRun.switchToHTTP(); + } + + // Wait since service might be ready even after port is available. + Thread.sleep(2000); + } + + @Test + public void testInvoke() throws Exception { + final OpenTelemetry openTelemetry = createOpenTelemetry("service over grpc"); + final Tracer tracer = openTelemetry.getTracer("grpc integration test tracer"); + + final String spanName = UUID.randomUUID().toString(); + Span span = tracer.spanBuilder(spanName).setSpanKind(Span.Kind.CLIENT).startSpan(); + + try (DaprClient client = new DaprClientBuilder().build()) { + try (Scope scope = span.makeCurrent()) { + SleepRequest req = SleepRequest.newBuilder().setSeconds(1).build(); + client.invokeMethod(daprRun.getAppName(), "sleepOverGRPC", req.toByteArray(), HttpExtension.POST) + .subscriberContext(getReactorContext()) + .block(); + } + } + span.end(); + OpenTelemetrySdk.getGlobalTracerManagement().shutdown(); + + Validation.validate(spanName, "calllocal/tracingit_service/sleepovergrpc"); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/Controller.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/Controller.java new file mode 100644 index 0000000000..c5364963f6 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/Controller.java @@ -0,0 +1,20 @@ +package io.dapr.it.tracing.http; + +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +/** + * SpringBoot Controller to handle input binding. + */ +@RestController +public class Controller { + + @PostMapping(path = "/sleep") + public void sleep(@RequestBody int seconds) throws InterruptedException { + if (seconds < 0) { + throw new IllegalArgumentException("Sleep time cannot be negative."); + } + Thread.sleep(seconds * 1000); + } +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java new file mode 100644 index 0000000000..cd28293864 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryConfig.java @@ -0,0 +1,35 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing.http; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Tracer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; +import org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +@EnableAutoConfiguration(exclude = {DataSourceAutoConfiguration.class, HibernateJpaAutoConfiguration.class}) +public class OpenTelemetryConfig { + + public static final String TRACER_NAME = "integration testing tracer"; + + public static final String SERVICE_NAME = "integration testing service over http"; + + @Bean + public OpenTelemetry initOpenTelemetry() { + return io.dapr.it.tracing.OpenTelemetry.createOpenTelemetry(SERVICE_NAME); + } + + @Bean + public Tracer initTracer(@Autowired OpenTelemetry openTelemetry) { + return openTelemetry.getTracer(TRACER_NAME); + } + +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryInterceptor.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryInterceptor.java new file mode 100644 index 0000000000..5cd8a472f5 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryInterceptor.java @@ -0,0 +1,64 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing.http; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.propagation.TextMapPropagator; +import org.jetbrains.annotations.Nullable; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.HandlerInterceptor; +import org.springframework.web.servlet.ModelAndView; + +import javax.servlet.DispatcherType; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; +import java.util.Collections; + +@Component +public class OpenTelemetryInterceptor implements HandlerInterceptor { + + @Autowired + private OpenTelemetry openTelemetry; + + private static final TextMapPropagator.Getter HTTP_SERVLET_REQUEST_GETTER = + new TextMapPropagator.Getter<>() { + @Override + public Iterable keys(HttpServletRequest carrier) { + return Collections.list(carrier.getHeaderNames()); + } + + @Nullable + @Override + public String get(@Nullable HttpServletRequest carrier, String key) { + return carrier.getHeader(key); + } + }; + + @Override + public boolean preHandle( + HttpServletRequest request, HttpServletResponse response, Object handler) { + final TextMapPropagator textFormat = openTelemetry.getPropagators().getTextMapPropagator(); + // preHandle is called twice for asynchronous request. For more information, read: + // https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/web/servlet/AsyncHandlerInterceptor.html + if (request.getDispatcherType() == DispatcherType.ASYNC) { + return true; + } + + Context context = textFormat.extract(Context.current(), request, HTTP_SERVLET_REQUEST_GETTER); + request.setAttribute("opentelemetry-context", context); + return true; + } + + @Override + public void postHandle( + HttpServletRequest request, HttpServletResponse response, Object handler, + ModelAndView modelAndView) { + // There is no global context to be changed in post handle since it is done in preHandle on a new call. + } + +} diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryInterceptorConfig.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryInterceptorConfig.java new file mode 100644 index 0000000000..a7de394024 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/OpenTelemetryInterceptorConfig.java @@ -0,0 +1,23 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing.http; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.config.annotation.InterceptorRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurationSupport; + +@Component +public class OpenTelemetryInterceptorConfig extends WebMvcConfigurationSupport { + + @Autowired + OpenTelemetryInterceptor interceptor; + + @Override + public void addInterceptors(InterceptorRegistry registry) { + registry.addInterceptor(interceptor); + } +} \ No newline at end of file diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java new file mode 100644 index 0000000000..de20876119 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/Service.java @@ -0,0 +1,39 @@ +/* + * Copyright (c) Microsoft Corporation and Dapr Contributors. + * Licensed under the MIT License. + */ + +package io.dapr.it.tracing.http; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + + +/** + * Service for subscriber. + */ +@SpringBootApplication +public class Service { + + public static final String SUCCESS_MESSAGE = "Completed initialization in"; + + public static void main(String[] args) { + int port = Integer.parseInt(args[0]); + + System.out.printf("Service starting on port %d ...\n", port); + + // Start Dapr's callback endpoint. + start(port); + } + + /** + * Starts Dapr's callback in a given port. + * + * @param port Port to listen to. + */ + private static void start(int port) { + SpringApplication app = new SpringApplication(Service.class); + app.run(String.format("--server.port=%d", port)); + } + +} \ No newline at end of file diff --git a/sdk-tests/src/test/java/io/dapr/it/tracing/http/TracingIT.java b/sdk-tests/src/test/java/io/dapr/it/tracing/http/TracingIT.java new file mode 100644 index 0000000000..4752be1d55 --- /dev/null +++ b/sdk-tests/src/test/java/io/dapr/it/tracing/http/TracingIT.java @@ -0,0 +1,89 @@ +package io.dapr.it.tracing.http; + +import io.dapr.client.DaprClient; +import io.dapr.client.DaprClientBuilder; +import io.dapr.client.domain.HttpExtension; +import io.dapr.it.BaseIT; +import io.dapr.it.DaprRun; +import io.dapr.it.tracing.Validation; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import java.util.Arrays; +import java.util.Collection; +import java.util.UUID; + +import static io.dapr.it.tracing.OpenTelemetry.createOpenTelemetry; +import static io.dapr.it.tracing.OpenTelemetry.getReactorContext; +import static org.junit.runners.Parameterized.Parameter; +import static org.junit.runners.Parameterized.Parameters; + +@RunWith(Parameterized.class) +public class TracingIT extends BaseIT { + + /** + * Parameters for this test. + * Param #1: useGrpc. + * @return Collection of parameter tuples. + */ + @Parameters + public static Collection data() { + return Arrays.asList(new Object[][] { { false }, { true } }); + } + + /** + * Run of a Dapr application. + */ + private DaprRun daprRun = null; + + @Parameter + public boolean useGrpc; + + @Before + public void init() throws Exception { + daprRun = startDaprApp( + TracingIT.class.getSimpleName(), + Service.SUCCESS_MESSAGE, + Service.class, + true, + 30000); + + if (this.useGrpc) { + daprRun.switchToGRPC(); + } else { + daprRun.switchToHTTP(); + } + + // Wait since service might be ready even after port is available. + Thread.sleep(2000); + } + + @Test + public void testInvoke() throws Exception { + final OpenTelemetry openTelemetry = createOpenTelemetry(OpenTelemetryConfig.SERVICE_NAME); + final Tracer tracer = openTelemetry.getTracer(OpenTelemetryConfig.TRACER_NAME); + + final String spanName = UUID.randomUUID().toString(); + Span span = tracer.spanBuilder(spanName).setSpanKind(Span.Kind.CLIENT).startSpan(); + + try (DaprClient client = new DaprClientBuilder().build()) { + try (Scope scope = span.makeCurrent()) { + client.invokeMethod(daprRun.getAppName(), "sleep", 1, HttpExtension.POST) + .subscriberContext(getReactorContext()) + .block(); + } + } + span.end(); + OpenTelemetrySdk.getGlobalTracerManagement().shutdown(); + + Validation.validate(spanName, "calllocal/tracingit_service/sleep"); + } + +}