OpenTelemetry Mutiny

bom/application/pom.xml

@@ -2692,6 +2692,11 @@
                 <artifactId>quarkus-opentelemetry</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.quarkus</groupId>
+                <artifactId>quarkus-opentelemetry-async-mutiny</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.quarkus</groupId>
                 <artifactId>quarkus-opentelemetry-exporter-jaeger-deployment</artifactId>

core/deployment/src/main/java/io/quarkus/deployment/Feature.java

@@ -72,6 +72,7 @@ public enum Feature {
     OIDC_TOKEN_PROPAGATION_REACTIVE,
     OPENSHIFT_CLIENT,
     OPENTELEMETRY,
+    OPENTELEMETRY_MUTINY_ASYNC,
     OPENTELEMETRY_JAEGER_EXPORTER,
     OPENTELEMETRY_OTLP_EXPORTER,
     PICOCLI,

devtools/bom-descriptor-json/pom.xml

@@ -1592,6 +1592,19 @@
                         </exclusion>
                     </exclusions>
                 </dependency>
+                <dependency>
+                    <groupId>io.quarkus</groupId>
+                    <artifactId>quarkus-opentelemetry-async-mutiny</artifactId>
+                    <version>${project.version}</version>
+                    <type>pom</type>
+                    <scope>test</scope>
+                    <exclusions>
+                        <exclusion>
+                            <groupId>*</groupId>
+                            <artifactId>*</artifactId>
+                        </exclusion>
+                    </exclusions>
+                </dependency>
                 <dependency>
                     <groupId>io.quarkus</groupId>
                     <artifactId>quarkus-opentelemetry-exporter-jaeger</artifactId>

docs/pom.xml

@@ -1558,6 +1558,19 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-opentelemetry-async-mutiny-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-opentelemetry-exporter-jaeger-deployment</artifactId>

extensions/opentelemetry/opentelemetry-async-mutiny/deployment/pom.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.quarkus</groupId>
+        <artifactId>quarkus-opentelemetry-async-mutiny-parent</artifactId>
+        <version>999-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>quarkus-opentelemetry-async-mutiny-deployment</artifactId>
+    <name>Quarkus - OpenTelemetry Async - Mutiny - Deployment</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-opentelemetry-deployment</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-opentelemetry-async-mutiny</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-grpc-common-deployment</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5-internal</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>io.quarkus</groupId>
+                            <artifactId>quarkus-extension-processor</artifactId>
+                            <version>${project.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

extensions/opentelemetry/opentelemetry-async-mutiny/deployment/src/main/java/io/quarkus/opentelemetry/async/mutiny/deployment/MutinyAsyncProcessor.java

@@ -0,0 +1,61 @@
+package io.quarkus.opentelemetry.async.mutiny.deployment;
+
+import java.util.function.BooleanSupplier;
+
+import io.quarkus.deployment.Feature;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
+import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem;
+import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
+import io.quarkus.opentelemetry.async.mutiny.runtime.MutinyAsyncConfig;
+import io.quarkus.opentelemetry.async.mutiny.runtime.tracing.MutinyAsyncRecorder;
+import io.quarkus.opentelemetry.async.mutiny.runtime.tracing.OpenTelemetryMultiInterceptor;
+import io.quarkus.opentelemetry.async.mutiny.runtime.tracing.OpenTelemetryUniInterceptor;
+
+public class MutinyAsyncProcessor {
+
+    static class MutinyAsyncEnabled implements BooleanSupplier {
+        MutinyAsyncConfig.MutinyAsyncBuildConfig mutinyAsyncConfig;
+
+        public boolean getAsBoolean() {
+            return mutinyAsyncConfig.enabled;
+        }
+    }
+
+    @BuildStep(onlyIf = MutinyAsyncEnabled.class)
+    FeatureBuildItem feature() {
+        return new FeatureBuildItem(Feature.OPENTELEMETRY_MUTINY_ASYNC);
+    }
+
+    @BuildStep(onlyIf = MutinyAsyncEnabled.class)
+    @Record(ExecutionTime.RUNTIME_INIT)
+    void registerAsyncStrategy(final MutinyAsyncConfig.MutinyAsyncRuntimeConfig runtimeConfig,
+            final MutinyAsyncRecorder recorder) {
+        recorder.registerAsyncStrategy(runtimeConfig);
+    }
+
+    @BuildStep(onlyIf = MutinyAsyncEnabled.class)
+    void registerUniInterceptor(
+            final BuildProducer<NativeImageResourceBuildItem> resource,
+            final BuildProducer<ReflectiveClassBuildItem> reflectiveClass) {
+
+        resource.produce(new NativeImageResourceBuildItem(
+                "META-INF/services/io.smallrye.mutiny.infrastructure.UniInterceptor"));
+        reflectiveClass
+                .produce(new ReflectiveClassBuildItem(true, true, OpenTelemetryUniInterceptor.class));
+    }
+
+    @BuildStep(onlyIf = MutinyAsyncEnabled.class)
+    void registerMultiInterceptor(
+            final BuildProducer<NativeImageResourceBuildItem> resource,
+            final BuildProducer<ReflectiveClassBuildItem> reflectiveClass) {
+
+        resource.produce(new NativeImageResourceBuildItem(
+                "META-INF/services/io.smallrye.mutiny.infrastructure.MultiInterceptor"));
+        reflectiveClass
+                .produce(new ReflectiveClassBuildItem(true, true, OpenTelemetryMultiInterceptor.class));
+    }
+}

extensions/opentelemetry/opentelemetry-async-mutiny/deployment/src/test/java/io/quarkus/opentelemetry/async/mutiny/deployment/MutinyAsyncDisabledTest.java

@@ -0,0 +1,32 @@
+package io.quarkus.opentelemetry.async.mutiny.deployment;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.opentelemetry.instrumentation.api.annotation.support.async.AsyncOperationEndStrategies;
+import io.quarkus.opentelemetry.async.mutiny.runtime.tracing.TracingMulti;
+import io.quarkus.opentelemetry.async.mutiny.runtime.tracing.TracingUni;
+import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+
+public class MutinyAsyncDisabledTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .withEmptyApplication()
+            .overrideConfigKey("quarkus.opentelemetry.tracer.async.mutiny.enabled", "false");
+
+    @Test
+    void testAsyncStrategies() {
+        assertThat(AsyncOperationEndStrategies.instance().resolveStrategy(Uni.class)).isNull();
+        assertThat(AsyncOperationEndStrategies.instance().resolveStrategy(Multi.class)).isNull();
+
+        // TODO: Not sure how much sense this makes. In the other test I was forced to register the interceptors
+        //  manually. So when I do not do that here then this is true for sure.
+        assertThat(Uni.createFrom().item("test")).isNotInstanceOf(TracingUni.class);
+        assertThat(Multi.createFrom().item("test")).isNotInstanceOf(TracingMulti.class);
+    }
+}

extensions/opentelemetry/opentelemetry-async-mutiny/deployment/src/test/java/io/quarkus/opentelemetry/async/mutiny/deployment/TestSpanExporter.java

@@ -0,0 +1,66 @@
+package io.quarkus.opentelemetry.async.mutiny.deployment;
+
+import static java.util.Comparator.comparingLong;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Collectors;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import io.quarkus.arc.Unremovable;
+
+@Unremovable
+@ApplicationScoped
+public class TestSpanExporter implements SpanExporter {
+    private final List<SpanData> finishedSpanItems = new CopyOnWriteArrayList<>();
+    private volatile boolean isStopped = false;
+
+    /**
+     * Careful when retrieving the list of finished spans. There is a chance when the response is already sent to the
+     * client and Vert.x still writing the end of the spans. This means that a response is available to assert from the
+     * test side but not all spans may be available yet. For this reason, this method requires the number of expected
+     * spans.
+     */
+    public List<SpanData> getFinishedSpanItems(int spanCount) {
+        assertSpanCount(spanCount);
+        return finishedSpanItems.stream().sorted(comparingLong(SpanData::getStartEpochNanos).reversed())
+                .collect(Collectors.toList());
+    }
+
+    public void assertSpanCount(int spanCount) {
+        await().atMost(30, SECONDS).untilAsserted(() -> assertEquals(spanCount, finishedSpanItems.size()));
+    }
+
+    public void reset() {
+        finishedSpanItems.clear();
+    }
+
+    @Override
+    public CompletableResultCode export(Collection<SpanData> spans) {
+        if (isStopped) {
+            return CompletableResultCode.ofFailure();
+        }
+        finishedSpanItems.addAll(spans);
+        return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+        return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+        finishedSpanItems.clear();
+        isStopped = true;
+        return CompletableResultCode.ofSuccess();
+    }
+}