Propagate trace context in RESTAPI runner()

- the runner() method is used in the aai-resources service
  to run the aai-core logic within a dedicated thread
- wrap this thread in a callable to propagate otel trace
  context to the thread

Issue-ID: AAI-4224
Change-Id: Ib94d614de0483577a1f895187a5e9d824711e99b
Signed-off-by: Fiete Ostkamp <fiete.ostkamp@telekom.de>

diff --git a/aai-core/pom.xml b/aai-core/pom.xml
index 3058693..e5a7399 100644 (file)
--- a/aai-core/pom.xml
+++ b/aai-core/pom.xml
@@ -262,6 +262,12 @@ limitations under the License.
                        <artifactId>micrometer-observation</artifactId>
     </dependency>
                <dependency>
+    <groupId>io.opentelemetry</groupId>
+                       <artifactId>opentelemetry-context</artifactId>
+                       <!-- <version>1.58.0</version> -->
+                       <scope>compile</scope>
+       </dependency>
+               <dependency>
                        <groupId>org.slf4j</groupId>
                        <artifactId>slf4j-api</artifactId>
                </dependency>

diff --git a/aai-core/src/main/java/org/onap/aai/restcore/RESTAPI.java b/aai-core/src/main/java/org/onap/aai/restcore/RESTAPI.java
index f4da017..3334e26 100644 (file)
--- a/aai-core/src/main/java/org/onap/aai/restcore/RESTAPI.java
+++ b/aai-core/src/main/java/org/onap/aai/restcore/RESTAPI.java
@@ -33,6 +33,8 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
 import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
@@ -350,9 +352,21 @@ public class RESTAPI {
             String timeoutByApp = AAIConfig.get(tba);
             String timeoutDefaultLimit = AAIConfig.get(tdl);
             String sourceOfTruth = headers.getRequestHeaders().getFirst("X-FromAppId");
+
             if (isTimeoutEnabled(sourceOfTruth, timeoutEnabled, timeoutByApp, timeoutDefaultLimit)) {
                 executor = Executors.newSingleThreadExecutor();
-                handler = executor.submit(c);
+
+                // Capture current context
+                Context context = Context.current();
+
+                // Wrap the callable to propagate context
+                Callable<Response> wrappedCallable = () -> {
+                    try (Scope scope = context.makeCurrent()) {
+                        return c.call();
+                    }
+                };
+
+                handler = executor.submit(wrappedCallable);
                 response = executeProcess(handler, sourceOfTruth, timeoutByApp, timeoutDefaultLimit, httpMethod,
                         headers, info);
             } else {