diff --git a/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/DaprClientAutoConfiguration.java b/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/DaprClientAutoConfiguration.java
index 941ae6ff22..6c2d44fa23 100644
--- a/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/DaprClientAutoConfiguration.java
+++ b/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/DaprClientAutoConfiguration.java
@@ -20,12 +20,16 @@
 import io.dapr.config.Properties;
 import io.dapr.workflows.client.DaprWorkflowClient;
 import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;
+import io.micrometer.observation.ObservationRegistry;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.trace.Tracer;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 
+import javax.annotation.Nullable;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -84,9 +88,12 @@ DaprClient daprClient(DaprClientBuilder daprClientBuilder) {
 
   @Bean
   @ConditionalOnMissingBean
-  DaprWorkflowClient daprWorkflowClient(DaprConnectionDetails daprConnectionDetails) {
+  DaprWorkflowClient daprWorkflowClient(DaprConnectionDetails daprConnectionDetails,
+                                        @Nullable ObservationRegistry observationRegistry,
+                                        @Nullable Tracer tracer,
+                                        @Nullable Meter meter) {
     Properties properties = createPropertiesFromConnectionDetails(daprConnectionDetails);
-    return new DaprWorkflowClient(properties);
+    return new DaprWorkflowClient(properties,  observationRegistry, tracer, meter);
   }
 
   @Bean
diff --git a/pom.xml b/pom.xml
index b45636c3e1..5717bb856a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -31,6 +31,8 @@
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
     <maven.compiler.release>11</maven.compiler.release>
+    <micrometer.version>1.16.0</micrometer.version>
+    <opentelemetry-api.version>1.56.0</opentelemetry-api.version>
     <maven.deploy.skip>true</maven.deploy.skip>
     <!--
         manually declare durabletask-client's jackson dependencies for workflows sdk
@@ -71,7 +73,7 @@
     <junit-vintage-engine.version>5.7.0</junit-vintage-engine.version>
     <junit-platform-console.version>1.7.0</junit-platform-console.version>
     <reactor.version>3.5.12</reactor.version>
-    <durabletask-client.version>1.5.10</durabletask-client.version>
+    <durabletask-client.version>1.5.11-SNAPSHOT</durabletask-client.version>
     <testcontainers-redis.version>2.2.2</testcontainers-redis.version>
     <slf4j.version>2.0.9</slf4j.version>
     <mockito.version>3.11.2</mockito.version>
@@ -127,6 +129,16 @@
         <artifactId>mockito-core</artifactId>
         <version>${mockito.version}</version>
       </dependency>
+      <dependency>
+        <groupId>io.micrometer</groupId>
+        <artifactId>micrometer-observation</artifactId>
+        <version>${micrometer.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>io.opentelemetry</groupId>
+        <artifactId>opentelemetry-api</artifactId>
+        <version>${opentelemetry-api.version}</version>
+      </dependency>
       <dependency>
         <groupId>io.projectreactor</groupId>
         <artifactId>reactor-test</artifactId>
diff --git a/sdk-workflows/pom.xml b/sdk-workflows/pom.xml
index 3afb38c40f..703e93a661 100644
--- a/sdk-workflows/pom.xml
+++ b/sdk-workflows/pom.xml
@@ -41,6 +41,14 @@
       <artifactId>junit-vintage-engine</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-observation</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-api</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.dapr</groupId>
       <artifactId>durabletask-client</artifactId>
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java b/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java
index 2391b8f635..17e3965e9f 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/WorkflowActivityContext.java
@@ -25,4 +25,6 @@ public interface WorkflowActivityContext {
 
   <T> T getInput(Class<T> targetType);
 
+  String getTraceParentId();
+
 }
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java
index 79725c0209..b5d9485e75 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java
@@ -22,10 +22,15 @@
 import io.dapr.utils.NetworkUtils;
 import io.dapr.workflows.Workflow;
 import io.dapr.workflows.internal.ApiTokenClientInterceptor;
+import io.dapr.workflows.internal.TraceParentClientInterceptor;
 import io.dapr.workflows.runtime.DefaultWorkflowInstanceStatus;
 import io.dapr.workflows.runtime.DefaultWorkflowState;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
+import io.micrometer.observation.ObservationRegistry;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
 
 import javax.annotation.Nullable;
 
@@ -41,6 +46,9 @@ public class DaprWorkflowClient implements AutoCloseable {
   private ClientInterceptor workflowApiTokenInterceptor;
   private DurableTaskClient innerClient;
   private ManagedChannel grpcChannel;
+  private ObservationRegistry observationRegistry;
+  private Tracer tracer;
+  private Meter meter;
 
   /**
    * Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel.
@@ -55,7 +63,25 @@ public DaprWorkflowClient() {
    * @param properties Properties for the GRPC Channel.
    */
   public DaprWorkflowClient(Properties properties) {
-    this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)));
+    this(properties, null, null, null);
+  }
+
+  /**
+   * Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel.
+   *
+   * @param properties Properties for the GRPC Channel.
+   * @param observationRegistry micrometer observation registry.
+   * @param tracer otel tracer for traces.
+   * @param meter otel meter for metrics.
+   */
+  public DaprWorkflowClient(Properties properties, ObservationRegistry observationRegistry,
+                            Tracer tracer, Meter meter) {
+    this(NetworkUtils.buildGrpcManagedChannel(properties,
+        new ApiTokenClientInterceptor(properties)), tracer);
+    this.observationRegistry = observationRegistry;
+    this.tracer = tracer;
+    this.meter = meter;
+
   }
 
   /**
@@ -63,8 +89,8 @@ public DaprWorkflowClient(Properties properties) {
    *
    * @param grpcChannel ManagedChannel for GRPC channel.
    */
-  private DaprWorkflowClient(ManagedChannel grpcChannel) {
-    this(createDurableTaskClient(grpcChannel), grpcChannel);
+  private DaprWorkflowClient(ManagedChannel grpcChannel, Tracer tracer) {
+    this(createDurableTaskClient(grpcChannel, tracer), grpcChannel);
   }
 
   /**
@@ -86,7 +112,17 @@ private DaprWorkflowClient(DurableTaskClient innerClient, ManagedChannel grpcCha
    * @return the randomly-generated instance ID for new Workflow instance.
    */
   public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz) {
-    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName());
+    if (tracer == null) {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName());
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.scheduleNewWorkflow")
+        .setAttribute("workflow", clazz.getName())
+        .startSpan();
+    try {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName());
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -98,7 +134,18 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz) {
    * @return the randomly-generated instance ID for new Workflow instance.
    */
   public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object input) {
-    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input);
+    if (tracer == null) {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.scheduleNewWorkflow")
+        .setAttribute("workflow", clazz.getName())
+        .setAttribute("input", input.toString())
+        .startSpan();
+    try {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -111,7 +158,19 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object in
    * @return the <code>instanceId</code> parameter value.
    */
   public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object input, String instanceId) {
-    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input, instanceId);
+    if (tracer == null) {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input, instanceId);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.scheduleNewWorkflow")
+        .setAttribute("workflow", clazz.getName())
+        .setAttribute("input", input.toString())
+        .setAttribute("instanceId", instanceId)
+        .startSpan();
+    try {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), input, instanceId);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -123,10 +182,22 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, Object in
    * @return the <code>instanceId</code> parameter value.
    */
   public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, NewWorkflowOptions options) {
-    NewOrchestrationInstanceOptions orchestrationInstanceOptions = fromNewWorkflowOptions(options);
+    if (tracer == null) {
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(), options);
+    }
 
-    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(),
-        orchestrationInstanceOptions);
+    Span span = tracer.spanBuilder("dapr.workflow.scheduleNewWorkflow")
+        .setAttribute("workflow", clazz.getName())
+        .setAttribute("options", options.toString())
+        .startSpan();
+    try {
+      NewOrchestrationInstanceOptions orchestrationInstanceOptions = fromNewWorkflowOptions(options);
+
+      return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName(),
+          orchestrationInstanceOptions);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -136,7 +207,18 @@ public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz, NewWorkfl
    * @param reason             reason for suspending the workflow instance.
    */
   public void suspendWorkflow(String workflowInstanceId, @Nullable String reason) {
-    this.innerClient.suspendInstance(workflowInstanceId, reason);
+    if (tracer == null) {
+      this.innerClient.suspendInstance(workflowInstanceId, reason);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.suspendWorkflow")
+        .setAttribute("instanceId", workflowInstanceId)
+        .setAttribute("reason", reason)
+        .startSpan();
+    try {
+      this.innerClient.suspendInstance(workflowInstanceId, reason);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -146,7 +228,18 @@ public void suspendWorkflow(String workflowInstanceId, @Nullable String reason)
    * @param reason             reason for resuming the workflow instance.
    */
   public void resumeWorkflow(String workflowInstanceId, @Nullable String reason) {
-    this.innerClient.resumeInstance(workflowInstanceId, reason);
+    if (tracer == null) {
+      this.innerClient.suspendInstance(workflowInstanceId, reason);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.resumeWorkflow")
+        .setAttribute("instanceId", workflowInstanceId)
+        .setAttribute("reason", reason)
+        .startSpan();
+    try {
+      this.innerClient.resumeInstance(workflowInstanceId, reason);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -156,7 +249,18 @@ public void resumeWorkflow(String workflowInstanceId, @Nullable String reason) {
    * @param output             the optional output to set for the terminated orchestration instance.
    */
   public void terminateWorkflow(String workflowInstanceId, @Nullable Object output) {
-    this.innerClient.terminate(workflowInstanceId, output);
+    if (tracer == null) {
+      this.innerClient.terminate(workflowInstanceId, output);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.terminateWorkflow")
+        .setAttribute("instanceId", workflowInstanceId)
+        .setAttribute("output", output.toString())
+        .startSpan();
+    try {
+      this.innerClient.terminate(workflowInstanceId, output);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -171,9 +275,20 @@ public void terminateWorkflow(String workflowInstanceId, @Nullable Object output
   @Nullable
   @Deprecated(forRemoval = true)
   public WorkflowInstanceStatus getInstanceState(String instanceId, boolean getInputsAndOutputs) {
-    OrchestrationMetadata metadata = this.innerClient.getInstanceMetadata(instanceId, getInputsAndOutputs);
-
-    return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    if (tracer == null) {
+      OrchestrationMetadata metadata = this.innerClient.getInstanceMetadata(instanceId, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.getInstanceState")
+        .setAttribute("instanceId", instanceId)
+        .setAttribute("fetchInputOutput", getInputsAndOutputs)
+        .startSpan();
+    try {
+      OrchestrationMetadata metadata = this.innerClient.getInstanceMetadata(instanceId, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -186,9 +301,20 @@ public WorkflowInstanceStatus getInstanceState(String instanceId, boolean getInp
    */
   @Nullable
   public WorkflowState getWorkflowState(String instanceId, boolean getInputsAndOutputs) {
-    OrchestrationMetadata metadata = this.innerClient.getInstanceMetadata(instanceId, getInputsAndOutputs);
-
-    return metadata == null ? null : new DefaultWorkflowState(metadata);
+    if (tracer == null) {
+      OrchestrationMetadata metadata = this.innerClient.getInstanceMetadata(instanceId, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowState(metadata);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.getWorkflowState")
+        .setAttribute("instanceId", instanceId)
+        .setAttribute("fetchInputOutput", getInputsAndOutputs)
+        .startSpan();
+    try {
+      OrchestrationMetadata metadata = this.innerClient.getInstanceMetadata(instanceId, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowState(metadata);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -214,9 +340,20 @@ public WorkflowState getWorkflowState(String instanceId, boolean getInputsAndOut
   public WorkflowInstanceStatus waitForInstanceStart(String instanceId, Duration timeout, boolean getInputsAndOutputs)
       throws TimeoutException {
 
-    OrchestrationMetadata metadata = this.innerClient.waitForInstanceStart(instanceId, timeout, getInputsAndOutputs);
-
-    return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    if (tracer == null) {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceStart(instanceId, timeout, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.waitForInstanceStart")
+        .setAttribute("instanceId", instanceId)
+        .setAttribute("fetchInputOutput", getInputsAndOutputs)
+        .startSpan();
+    try {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceStart(instanceId, timeout, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    } finally {
+      span.end();
+    }
   }
 
 
@@ -241,9 +378,20 @@ public WorkflowInstanceStatus waitForInstanceStart(String instanceId, Duration t
   public WorkflowState waitForWorkflowStart(String instanceId, Duration timeout, boolean getInputsAndOutputs)
       throws TimeoutException {
 
-    OrchestrationMetadata metadata = this.innerClient.waitForInstanceStart(instanceId, timeout, getInputsAndOutputs);
-
-    return metadata == null ? null : new DefaultWorkflowState(metadata);
+    if (tracer == null) {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceStart(instanceId, timeout, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowState(metadata);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.waitForWorkflowStart")
+        .setAttribute("instanceId", instanceId)
+        .setAttribute("fetchInputOutput", getInputsAndOutputs)
+        .startSpan();
+    try {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceStart(instanceId, timeout, getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowState(metadata);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -271,9 +419,22 @@ public WorkflowState waitForWorkflowStart(String instanceId, Duration timeout, b
   public WorkflowInstanceStatus waitForInstanceCompletion(String instanceId, Duration timeout,
                                                           boolean getInputsAndOutputs) throws TimeoutException {
 
-    OrchestrationMetadata metadata = this.innerClient.waitForInstanceCompletion(instanceId, timeout,
-        getInputsAndOutputs);
-    return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    if (tracer == null) {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceCompletion(instanceId, timeout,
+          getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.waitForWorkflowStart")
+        .setAttribute("instanceId", instanceId)
+        .setAttribute("fetchInputOutput", getInputsAndOutputs)
+        .startSpan();
+    try {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceCompletion(instanceId, timeout,
+          getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowInstanceStatus(metadata);
+    } finally {
+      span.end();
+    }
   }
 
 
@@ -300,9 +461,22 @@ public WorkflowInstanceStatus waitForInstanceCompletion(String instanceId, Durat
   public WorkflowState waitForWorkflowCompletion(String instanceId, Duration timeout,
                                                           boolean getInputsAndOutputs) throws TimeoutException {
 
-    OrchestrationMetadata metadata = this.innerClient.waitForInstanceCompletion(instanceId, timeout,
-        getInputsAndOutputs);
-    return metadata == null ? null : new DefaultWorkflowState(metadata);
+    if (tracer == null) {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceCompletion(instanceId, timeout,
+          getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowState(metadata);
+    }
+    Span span = tracer.spanBuilder("dapr.workflow.waitForWorkflowCompletion")
+        .setAttribute("instanceId", instanceId)
+        .setAttribute("fetchInputOutput", getInputsAndOutputs)
+        .startSpan();
+    try {
+      OrchestrationMetadata metadata = this.innerClient.waitForInstanceCompletion(instanceId, timeout,
+          getInputsAndOutputs);
+      return metadata == null ? null : new DefaultWorkflowState(metadata);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -313,7 +487,19 @@ public WorkflowState waitForWorkflowCompletion(String instanceId, Duration timeo
    * @param eventPayload       The serializable data payload to include with the event.
    */
   public void raiseEvent(String workflowInstanceId, String eventName, Object eventPayload) {
-    this.innerClient.raiseEvent(workflowInstanceId, eventName, eventPayload);
+    if (tracer == null) {
+      this.innerClient.raiseEvent(workflowInstanceId, eventName, eventPayload);
+    }
+
+    Span span = tracer.spanBuilder("dapr.workflow.raiseEvent")
+        .setAttribute("eventName", eventName)
+        .setAttribute("payload", eventPayload.toString())
+        .startSpan();
+    try {
+      this.innerClient.raiseEvent(workflowInstanceId, eventName, eventPayload);
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -325,13 +511,27 @@ public void raiseEvent(String workflowInstanceId, String eventName, Object event
    */
   @Deprecated(forRemoval = true)
   public boolean purgeInstance(String workflowInstanceId) {
-    PurgeResult result = this.innerClient.purgeInstance(workflowInstanceId);
+    if (tracer == null) {
+      PurgeResult result = this.innerClient.purgeInstance(workflowInstanceId);
 
-    if (result != null) {
-      return result.getDeletedInstanceCount() > 0;
+      if (result != null) {
+        return result.getDeletedInstanceCount() > 0;
+      }
     }
+    Span span = tracer.spanBuilder("dapr.workflow.purgeInstance")
+        .setAttribute("instanceId", workflowInstanceId)
+        .startSpan();
+    try {
+      PurgeResult result = this.innerClient.purgeInstance(workflowInstanceId);
+
+      if (result != null) {
+        return result.getDeletedInstanceCount() > 0;
+      }
 
-    return false;
+      return false;
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -341,13 +541,29 @@ public boolean purgeInstance(String workflowInstanceId) {
    * @return Return true if the workflow state was found and purged successfully otherwise false.
    */
   public boolean purgeWorkflow(String workflowInstanceId) {
-    PurgeResult result = this.innerClient.purgeInstance(workflowInstanceId);
+    if (tracer == null) {
+      PurgeResult result = this.innerClient.purgeInstance(workflowInstanceId);
+
+      if (result != null) {
+        return result.getDeletedInstanceCount() > 0;
+      }
 
-    if (result != null) {
-      return result.getDeletedInstanceCount() > 0;
+      return false;
     }
+    Span span = tracer.spanBuilder("dapr.workflow.purgeWorkflow")
+        .setAttribute("instanceId", workflowInstanceId)
+        .startSpan();
+    try {
+      PurgeResult result = this.innerClient.purgeInstance(workflowInstanceId);
+
+      if (result != null) {
+        return result.getDeletedInstanceCount() > 0;
+      }
 
-    return false;
+      return false;
+    } finally {
+      span.end();
+    }
   }
 
   /**
@@ -373,9 +589,10 @@ public void close() throws InterruptedException {
    * @param grpcChannel ManagedChannel for GRPC.
    * @return a new instance of a DurableTaskClient with a GRPC channel.
    */
-  private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel) {
+  private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel, Tracer tracer) {
     return new DurableTaskGrpcClientBuilder()
         .grpcChannel(grpcChannel)
+        .tracer(tracer)
         .build();
   }
 
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/internal/TraceParentClientInterceptor.java b/sdk-workflows/src/main/java/io/dapr/workflows/internal/TraceParentClientInterceptor.java
new file mode 100644
index 0000000000..9eafd7148c
--- /dev/null
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/internal/TraceParentClientInterceptor.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright 2023 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package io.dapr.workflows.internal;
+
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+
+public class TraceParentClientInterceptor implements ClientInterceptor {
+
+  private Tracer tracer;
+
+  public TraceParentClientInterceptor(Tracer tracer) {
+    this.tracer = tracer;
+  }
+
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+            MethodDescriptor<ReqT, RespT> methodDescriptor,
+            CallOptions options,
+            Channel channel) {
+    // TBD: do we need timeout in workflow client?
+    ClientCall<ReqT, RespT> clientCall = channel.newCall(methodDescriptor, options);
+    return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(clientCall) {
+
+        @Override
+        public void start(final Listener<RespT> responseListener, final Metadata metadata) {
+          if (tracer == null) {
+            super.start(responseListener, metadata);
+          }
+          Span span = tracer.spanBuilder("dapr.workflow.grpc")
+              .setAttribute("method", methodDescriptor.getFullMethodName())
+              .setAttribute("options", options.toString())
+              .setAttribute("metadata", metadata.toString())
+              .startSpan();
+          try {
+            String traceId = span.getSpanContext().getTraceId();
+            String traceState = span.getSpanContext().getTraceState().toString();
+            byte[] traceIdBytes = span.getSpanContext().getTraceIdBytes();
+            metadata.put(Metadata.Key.of("traceparent", Metadata.ASCII_STRING_MARSHALLER), traceId);
+            metadata.put(Metadata.Key.of("tracestate", Metadata.ASCII_STRING_MARSHALLER), traceState);
+            metadata.put(Metadata.Key.of("grpc-trace-bin", Metadata.BINARY_BYTE_MARSHALLER), traceIdBytes);
+            System.out.println("Trace Id in interceptor: " + traceId);
+            System.out.println("Trace State in interceptor: " + traceState);
+            System.out.println("Metadata at the interceptor: " + metadata);
+            super.start(responseListener, metadata);
+          } finally {
+            span.end();
+          }
+        }
+    };
+  }
+}
+
diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java
index d6f935e8f8..07e4b3ead4 100644
--- a/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java
+++ b/sdk-workflows/src/main/java/io/dapr/workflows/runtime/DefaultWorkflowActivityContext.java
@@ -101,4 +101,9 @@ public <T> T getInput(Class<T> targetType) {
   public String getTaskExecutionId() {
     return this.innerContext.getTaskExecutionId();
   }
+
+  @Override
+  public String getTraceParentId() {
+    return this.innerContext.getTraceParentId();
+  }
 }