Observability: Trace context propagation within DPF

CHANGELOG.md

@@ -97,6 +97,7 @@ in the detailed section referring to by linking pull requests or issues.
 * Http Deprovisioner Webhook endpoint (#1039)
 * Add performance test example and scheduled workflow (#1029)
 * Add basic authentication mechanism for DataManagement API (#981)
+* Trace context propagation in DPF (#1162)
 
 #### Changed
 

extensions/data-plane-transfer/data-plane-transfer-client/build.gradle.kts

@@ -19,6 +19,7 @@ plugins {
 val httpMockServer: String by project
 val jodahFailsafeVersion: String by project
 val okHttpVersion: String by project
+val openTelemetryVersion: String by project
 val faker: String by project
 
 dependencies {
@@ -30,6 +31,7 @@ dependencies {
 
     implementation("net.jodah:failsafe:${jodahFailsafeVersion}")
     implementation("com.squareup.okhttp3:okhttp:${okHttpVersion}")
+    implementation("io.opentelemetry:opentelemetry-extension-annotations:${openTelemetryVersion}")
 
     testImplementation("org.mock-server:mockserver-netty:${httpMockServer}:shaded")
     testImplementation("org.mock-server:mockserver-client-java:${httpMockServer}:shaded")

extensions/data-plane-transfer/data-plane-transfer-client/src/main/java/org/eclipse/dataspaceconnector/transfer/dataplane/client/EmbeddedDataPlaneTransferClient.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.dataspaceconnector.transfer.dataplane.client;
 
+import io.opentelemetry.extension.annotations.WithSpan;
 import org.eclipse.dataspaceconnector.dataplane.spi.manager.DataPlaneManager;
 import org.eclipse.dataspaceconnector.spi.response.ResponseStatus;
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
@@ -31,6 +32,7 @@ public EmbeddedDataPlaneTransferClient(DataPlaneManager dataPlaneManager) {
         this.dataPlaneManager = dataPlaneManager;
     }
 
+    @WithSpan
     @Override
     public StatusResult<Void> transfer(DataFlowRequest request) {
         var result = dataPlaneManager.validate(request);

extensions/data-plane-transfer/data-plane-transfer-client/src/main/java/org/eclipse/dataspaceconnector/transfer/dataplane/client/RemoteDataPlaneTransferClient.java

@@ -16,6 +16,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.opentelemetry.extension.annotations.WithSpan;
 import net.jodah.failsafe.Failsafe;
 import net.jodah.failsafe.RetryPolicy;
 import okhttp3.MediaType;
@@ -57,6 +58,7 @@ public RemoteDataPlaneTransferClient(OkHttpClient client, DataPlaneSelectorClien
         this.mapper = mapper;
     }
 
+    @WithSpan
     @Override
     public StatusResult<Void> transfer(DataFlowRequest request) {
         var instance = selectorClient.find(request.getSourceDataAddress(), request.getDestinationDataAddress(), selectorStrategy);

extensions/data-plane/data-plane-framework/build.gradle.kts

@@ -12,13 +12,17 @@
  *
  */
 
+val openTelemetryVersion: String by project
+
 plugins {
     `java-library`
 }
 
 dependencies {
+    api(project(":spi:core-spi"))
     api(project(":extensions:data-plane:data-plane-spi"))
     implementation(project(":common:util"))
+    implementation("io.opentelemetry:opentelemetry-extension-annotations:${openTelemetryVersion}")
     testImplementation(testFixtures(project(":launchers:junit")))
 }
 

extensions/data-plane/data-plane-framework/src/main/java/org/eclipse/dataspaceconnector/dataplane/framework/DataPlaneFrameworkExtension.java

@@ -96,6 +96,8 @@ public void initialize(ServiceExtensionContext context) {
         context.registerService(DataTransferExecutorServiceContainer.class, executorContainer);
 
         monitor = context.getMonitor();
+        var telemetry = context.getTelemetry();
+
         var queueCapacity = context.getSetting(QUEUE_CAPACITY, DEFAULT_QUEUE_CAPACITY);
         var workers = context.getSetting(WORKERS, DEFAULT_WORKERS);
         var waitTimeout = context.getSetting(WAIT_TIMEOUT, DEFAULT_WAIT_TIMEOUT);
@@ -108,7 +110,9 @@ public void initialize(ServiceExtensionContext context) {
                 .pipelineService(pipelineService)
                 .transferServiceRegistry(transferServiceRegistry)
                 .store(new InMemoryDataPlaneStore(IN_MEMORY_STORE_CAPACITY))
-                .monitor(monitor).build();
+                .monitor(monitor)
+                .telemetry(telemetry)
+                .build();
 
         context.registerService(DataPlaneManager.class, dataPlaneManager);
     }

extensions/data-plane/data-plane-framework/src/main/java/org/eclipse/dataspaceconnector/dataplane/framework/manager/DataPlaneManagerImpl.java

@@ -25,6 +25,7 @@
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
 import org.eclipse.dataspaceconnector.spi.result.Result;
 import org.eclipse.dataspaceconnector.spi.system.ExecutorInstrumentation;
+import org.eclipse.dataspaceconnector.spi.telemetry.Telemetry;
 import org.eclipse.dataspaceconnector.spi.types.domain.transfer.DataFlowRequest;
 
 import java.util.concurrent.ArrayBlockingQueue;
@@ -49,6 +50,7 @@ public class DataPlaneManagerImpl implements DataPlaneManager {
     private PipelineService pipelineService;
     private ExecutorInstrumentation executorInstrumentation;
     private Monitor monitor;
+    private Telemetry telemetry;
 
     private BlockingQueue<DataFlowRequest> queue;
     private ExecutorService executorService;
@@ -89,8 +91,12 @@ public Result<Boolean> validate(DataFlowRequest dataRequest) {
     }
 
     public void initiateTransfer(DataFlowRequest dataRequest) {
-        queue.add(dataRequest);
-        store.received(dataRequest.getProcessId());
+        // store current trace context in entity for request traceability
+        DataFlowRequest dataRequestWithTraceContext = dataRequest.toBuilder()
+                .traceContext(telemetry.getCurrentTraceContext())
+                .build();
+        queue.add(dataRequestWithTraceContext);
+        store.received(dataRequestWithTraceContext.getProcessId());
     }
 
     @Override
@@ -116,21 +122,9 @@ private void run() {
                 if (request == null) {
                     continue;
                 }
-                final var polledRequest = request;
+                // propagate trace context for request into the current thread
+                telemetry.contextPropagationMiddleware(this::processDataFlowRequest).accept(request);
 
-                var transferService = transferServiceRegistry.resolveTransferService(polledRequest);
-                if (transferService == null) {
-                    // Should not happen since resolving a transferService is part of payload validation
-                    // TODO persist error details
-                    store.completed(polledRequest.getProcessId());
-                } else {
-                    transferService.transfer(request).whenComplete((result, exception) -> {
-                        if (polledRequest.isTrackable()) {
-                            // TODO persist TransferResult or error details
-                            store.completed(polledRequest.getProcessId());
-                        }
-                    });
-                }
             } catch (InterruptedException e) {
                 Thread.interrupted();
                 active.set(false);
@@ -147,6 +141,22 @@ private void run() {
         }
     }
 
+    private void processDataFlowRequest(DataFlowRequest request) {
+        var transferService = transferServiceRegistry.resolveTransferService(request);
+        if (transferService == null) {
+            // Should not happen since resolving a transferService is part of payload validation
+            // TODO persist error details
+            store.completed(request.getProcessId());
+        } else {
+            transferService.transfer(request).whenComplete((result, exception) -> {
+                if (request.isTrackable()) {
+                    // TODO persist TransferResult or error details
+                    store.completed(request.getProcessId());
+                }
+            });
+        }
+    }
+
     public static class Builder {
         private DataPlaneManagerImpl manager;
 
@@ -174,6 +184,11 @@ public Builder monitor(Monitor monitor) {
             return this;
         }
 
+        public Builder telemetry(Telemetry telemetry) {
+            manager.telemetry = telemetry;
+            return this;
+        }
+
         public Builder queueCapacity(int capacity) {
             manager.queueCapacity = capacity;
             return this;
@@ -200,6 +215,7 @@ public DataPlaneManagerImpl build() {
 
         private Builder() {
             manager = new DataPlaneManagerImpl();
+            this.manager.telemetry = new Telemetry(); // default noop implementation
         }
     }
 

extensions/data-plane/data-plane-framework/src/main/java/org/eclipse/dataspaceconnector/dataplane/framework/pipeline/PipelineServiceImpl.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.dataspaceconnector.dataplane.framework.pipeline;
 
+import io.opentelemetry.extension.annotations.WithSpan;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.DataSink;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.DataSinkFactory;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.DataSource;
@@ -72,6 +73,7 @@ public Result<Boolean> validate(DataFlowRequest request) {
         return Result.success(true);
     }
 
+    @WithSpan
     @Override
     public CompletableFuture<StatusResult<Void>> transfer(DataFlowRequest request) {
         var sourceFactory = getSourceFactory(request);

extensions/data-plane/data-plane-framework/src/main/java/org/eclipse/dataspaceconnector/dataplane/framework/pipeline/PipelineServiceTransferServiceImpl.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.dataspaceconnector.dataplane.framework.pipeline;
 
+import io.opentelemetry.extension.annotations.WithSpan;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.PipelineService;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.TransferService;
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
@@ -43,6 +44,7 @@ public Result<Boolean> validate(DataFlowRequest request) {
         return pipelineService.validate(request);
     }
 
+    @WithSpan
     @Override
     public CompletableFuture<StatusResult<Void>> transfer(DataFlowRequest request) {
         return pipelineService.transfer(request);

extensions/data-plane/data-plane-spi/build.gradle.kts

@@ -13,6 +13,7 @@
  */
 
 val mockitoVersion: String by project
+val openTelemetryVersion: String by project
 
 plugins {
     `java-library`
@@ -21,6 +22,7 @@ plugins {
 dependencies {
     api(project(":spi:core-spi"))
     implementation(project(":common:util"))
+    implementation("io.opentelemetry:opentelemetry-extension-annotations:${openTelemetryVersion}")
 }
 
 

extensions/data-plane/data-plane-spi/src/main/java/org/eclipse/dataspaceconnector/dataplane/spi/pipeline/ParallelSink.java

@@ -14,15 +14,20 @@
 
 package org.eclipse.dataspaceconnector.dataplane.spi.pipeline;
 
+import io.opentelemetry.extension.annotations.WithSpan;
 import org.eclipse.dataspaceconnector.common.stream.PartitionIterator;
 import org.eclipse.dataspaceconnector.spi.monitor.Monitor;
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
 import org.eclipse.dataspaceconnector.spi.result.AbstractResult;
+import org.eclipse.dataspaceconnector.spi.telemetry.Telemetry;
+import org.eclipse.dataspaceconnector.spi.telemetry.TraceCarrier;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 
 import static java.util.concurrent.CompletableFuture.supplyAsync;
 import static java.util.stream.Collectors.toList;
@@ -37,12 +42,16 @@ public abstract class ParallelSink implements DataSink {
     protected int partitionSize = 5;
     protected ExecutorService executorService;
     protected Monitor monitor;
+    protected Telemetry telemetry;
 
+    @WithSpan
     @Override
     public CompletableFuture<StatusResult<Void>> transfer(DataSource source) {
         try (var partStream = source.openPartStream()) {
             var partitioned = PartitionIterator.streamOf(partStream, partitionSize);
-            var futures = partitioned.map(parts -> supplyAsync(() -> transferParts(parts), executorService)).collect(toList());
+            var traceCarrier = telemetry.getTraceCarrierWithCurrentContext();
+
+            var futures = partitioned.map(parts -> processPartsAsync(parts, traceCarrier)).collect(toList());
             return futures.stream()
                     .collect(asyncAllOf())
                     .thenApply(results -> results.stream()
@@ -57,13 +66,20 @@ public CompletableFuture<StatusResult<Void>> transfer(DataSource source) {
         }
     }
 
+    @NotNull
+    private CompletableFuture<StatusResult<Void>> processPartsAsync(List<DataSource.Part> parts, TraceCarrier traceCarrier) {
+        Supplier<StatusResult<Void>> supplier = () -> transferParts(parts);
+        return supplyAsync(telemetry.contextPropagationMiddleware(supplier, traceCarrier), executorService);
+    }
+
     protected abstract StatusResult<Void> transferParts(List<DataSource.Part> parts);
 
     protected abstract static class Builder<B extends Builder<B, T>, T extends ParallelSink> {
         protected T sink;
 
         protected Builder(T sink) {
             this.sink = sink;
+            this.sink.telemetry = new Telemetry(); // default noop implementation
         }
 
         public B requestId(String requestId) {
@@ -86,6 +102,11 @@ public B monitor(Monitor monitor) {
             return self();
         }
 
+        public B telemetry(Telemetry telemetry) {
+            sink.telemetry = telemetry;
+            return self();
+        }
+
         public T build() {
             Objects.requireNonNull(sink.requestId, "requestId");
             Objects.requireNonNull(sink.executorService, "executorService");

extensions/data-plane/data-plane-spi/src/test/java/org/eclipse/dataspaceconnector/dataplane/spi/pipeline/ParallelSinkTest.java

@@ -18,6 +18,7 @@
 import org.eclipse.dataspaceconnector.spi.monitor.Monitor;
 import org.eclipse.dataspaceconnector.spi.response.ResponseStatus;
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
+import org.eclipse.dataspaceconnector.spi.telemetry.Telemetry;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -48,6 +49,7 @@ class ParallelSinkTest {
     void setup() {
         fakeSink = new FakeParallelSink();
         fakeSink.monitor = monitor;
+        fakeSink.telemetry = new Telemetry(); // default noop implementation
         fakeSink.executorService = executor;
         fakeSink.requestId = UUID.randomUUID().toString();
     }

samples/04.0-file-transfer/transfer-file/build.gradle.kts

@@ -19,6 +19,7 @@ plugins {
 }
 
 val rsApi: String by project
+val openTelemetryVersion: String by project
 
 dependencies {
     api(project(":spi"))
@@ -28,6 +29,7 @@ dependencies {
     implementation(project(":extensions:data-plane-selector:selector-core"))
     implementation(project(":extensions:data-plane-selector:selector-store"))
     implementation(project(":extensions:data-plane:data-plane-framework"))
+    implementation("io.opentelemetry:opentelemetry-extension-annotations:${openTelemetryVersion}")
 
     implementation(project(":extensions:data-plane:data-plane-spi"))
     implementation(project(":extensions:in-memory:assetindex-memory"))

samples/04.0-file-transfer/transfer-file/src/main/java/org/eclipse/dataspaceconnector/extensions/api/FileTransferDataSink.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.dataspaceconnector.extensions.api;
 
+import io.opentelemetry.extension.annotations.WithSpan;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.DataSource;
 import org.eclipse.dataspaceconnector.dataplane.spi.pipeline.ParallelSink;
 import org.eclipse.dataspaceconnector.spi.response.StatusResult;
@@ -29,6 +30,7 @@
 class FileTransferDataSink extends ParallelSink {
     private File file;
 
+    @WithSpan
     @Override
     protected StatusResult<Void> transferParts(List<DataSource.Part> parts) {
         for (DataSource.Part part : parts) {

spi/core-spi/src/main/java/org/eclipse/dataspaceconnector/spi/telemetry/InMemoryTraceCarrier.java

@@ -0,0 +1,35 @@
+/*
+ *  Copyright (c) 2022 Microsoft Corporation
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Microsoft Corporation - Initial implementation
+ *
+ */
+
+package org.eclipse.dataspaceconnector.spi.telemetry;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Simple {@link TraceCarrier} to use in situations where no entity is persisted (e.g. asynchronous processing)
+ */
+class InMemoryTraceCarrier implements TraceCarrier {
+
+    private final Map<String, String> traceContext;
+
+    InMemoryTraceCarrier(Map<String, String> traceContext) {
+        this.traceContext = Collections.unmodifiableMap(traceContext);
+    }
+
+    @Override
+    public Map<String, String> getTraceContext() {
+        return traceContext;
+    }
+}