io.micrometer
micrometer-observation-test
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 6b6dcac34b..9a763ba7c7 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
@@ -21,6 +21,8 @@
import io.dapr.spring.boot.properties.client.ClientPropertiesDaprConnectionDetails;
import io.dapr.spring.boot.properties.client.DaprClientProperties;
import io.dapr.spring.boot.properties.client.DaprConnectionDetails;
+import io.dapr.spring.observation.client.ObservationDaprClient;
+import io.dapr.spring.observation.client.ObservationDaprWorkflowClient;
import io.dapr.workflows.client.DaprWorkflowClient;
import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;
import io.micrometer.observation.ObservationRegistry;
diff --git a/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprClient.java b/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprClient.java
deleted file mode 100644
index 2529a65a95..0000000000
--- a/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprClient.java
+++ /dev/null
@@ -1,764 +0,0 @@
-/*
- * Copyright 2025 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.spring.boot.autoconfigure.client;
-
-import io.dapr.client.DaprClient;
-import io.dapr.client.domain.BulkPublishRequest;
-import io.dapr.client.domain.BulkPublishResponse;
-import io.dapr.client.domain.ConfigurationItem;
-import io.dapr.client.domain.DaprMetadata;
-import io.dapr.client.domain.DeleteJobRequest;
-import io.dapr.client.domain.DeleteStateRequest;
-import io.dapr.client.domain.ExecuteStateTransactionRequest;
-import io.dapr.client.domain.GetBulkSecretRequest;
-import io.dapr.client.domain.GetBulkStateRequest;
-import io.dapr.client.domain.GetConfigurationRequest;
-import io.dapr.client.domain.GetJobRequest;
-import io.dapr.client.domain.GetJobResponse;
-import io.dapr.client.domain.GetSecretRequest;
-import io.dapr.client.domain.GetStateRequest;
-import io.dapr.client.domain.HttpExtension;
-import io.dapr.client.domain.InvokeBindingRequest;
-import io.dapr.client.domain.InvokeMethodRequest;
-import io.dapr.client.domain.PublishEventRequest;
-import io.dapr.client.domain.SaveStateRequest;
-import io.dapr.client.domain.ScheduleJobRequest;
-import io.dapr.client.domain.State;
-import io.dapr.client.domain.StateOptions;
-import io.dapr.client.domain.SubscribeConfigurationRequest;
-import io.dapr.client.domain.SubscribeConfigurationResponse;
-import io.dapr.client.domain.TransactionalStateOperation;
-import io.dapr.client.domain.UnsubscribeConfigurationRequest;
-import io.dapr.client.domain.UnsubscribeConfigurationResponse;
-import io.dapr.utils.TypeRef;
-import io.grpc.Channel;
-import io.grpc.stub.AbstractStub;
-import io.micrometer.observation.Observation;
-import io.micrometer.observation.ObservationRegistry;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-/**
- * A {@link DaprClient} decorator that creates Micrometer Observation spans (bridged to OpenTelemetry)
- * for each non-deprecated method call. Consumers continue to use {@link DaprClient} as-is; no code
- * changes are required on their side.
- *
- * Deprecated methods are delegated directly without any observation.
- */
-public class ObservationDaprClient implements DaprClient {
-
- private final DaprClient delegate;
- private final ObservationRegistry observationRegistry;
-
- /**
- * Creates a new {@code ObservationDaprClient}.
- *
- * @param delegate the underlying {@link DaprClient} to delegate calls to
- * @param observationRegistry the Micrometer {@link ObservationRegistry} used to create spans
- */
- public ObservationDaprClient(DaprClient delegate, ObservationRegistry observationRegistry) {
- this.delegate = Objects.requireNonNull(delegate, "delegate must not be null");
- this.observationRegistry = Objects.requireNonNull(observationRegistry,
- "observationRegistry must not be null");
- }
-
- // -------------------------------------------------------------------------
- // Internal helpers
- // -------------------------------------------------------------------------
-
- private Mono observe(Observation obs, Supplier> monoSupplier) {
- return Mono.defer(() -> {
- obs.start();
- return monoSupplier.get()
- .doOnError(obs::error)
- .doFinally(signal -> obs.stop());
- });
- }
-
- private Flux observeFlux(Observation obs, Supplier> fluxSupplier) {
- return Flux.defer(() -> {
- obs.start();
- return fluxSupplier.get()
- .doOnError(obs::error)
- .doFinally(signal -> obs.stop());
- });
- }
-
- private static String safe(String value) {
- return value != null ? value : "";
- }
-
- private Observation observation(String name) {
- return Observation.createNotStarted(name, observationRegistry);
- }
-
- // -------------------------------------------------------------------------
- // Lifecycle
- // -------------------------------------------------------------------------
-
- @Override
- public Mono waitForSidecar(int timeoutInMilliseconds) {
- return observe(observation("dapr.client.wait_for_sidecar"),
- () -> delegate.waitForSidecar(timeoutInMilliseconds));
- }
-
- @Override
- public Mono shutdown() {
- return observe(observation("dapr.client.shutdown"),
- () -> delegate.shutdown());
- }
-
- @Override
- public void close() throws Exception {
- delegate.close();
- }
-
- // -------------------------------------------------------------------------
- // Pub/Sub
- // -------------------------------------------------------------------------
-
- @Override
- public Mono publishEvent(String pubsubName, String topicName, Object data) {
- return observe(
- observation("dapr.client.publish_event")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(pubsubName))
- .highCardinalityKeyValue("dapr.topic.name", safe(topicName)),
- () -> delegate.publishEvent(pubsubName, topicName, data));
- }
-
- @Override
- public Mono publishEvent(String pubsubName, String topicName, Object data,
- Map metadata) {
- return observe(
- observation("dapr.client.publish_event")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(pubsubName))
- .highCardinalityKeyValue("dapr.topic.name", safe(topicName)),
- () -> delegate.publishEvent(pubsubName, topicName, data, metadata));
- }
-
- @Override
- public Mono publishEvent(PublishEventRequest request) {
- return observe(
- observation("dapr.client.publish_event")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(request.getPubsubName()))
- .highCardinalityKeyValue("dapr.topic.name", safe(request.getTopic())),
- () -> delegate.publishEvent(request));
- }
-
- @Override
- public Mono> publishEvents(BulkPublishRequest request) {
- return observe(
- observation("dapr.client.publish_events")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(request.getPubsubName()))
- .highCardinalityKeyValue("dapr.topic.name", safe(request.getTopic())),
- () -> delegate.publishEvents(request));
- }
-
- @Override
- public Mono> publishEvents(String pubsubName, String topicName,
- String contentType, List events) {
- return observe(
- observation("dapr.client.publish_events")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(pubsubName))
- .highCardinalityKeyValue("dapr.topic.name", safe(topicName)),
- () -> delegate.publishEvents(pubsubName, topicName, contentType, events));
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public Mono> publishEvents(String pubsubName, String topicName,
- String contentType, T... events) {
- return observe(
- observation("dapr.client.publish_events")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(pubsubName))
- .highCardinalityKeyValue("dapr.topic.name", safe(topicName)),
- () -> delegate.publishEvents(pubsubName, topicName, contentType, events));
- }
-
- @Override
- public Mono> publishEvents(String pubsubName, String topicName,
- String contentType,
- Map requestMetadata,
- List events) {
- return observe(
- observation("dapr.client.publish_events")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(pubsubName))
- .highCardinalityKeyValue("dapr.topic.name", safe(topicName)),
- () -> delegate.publishEvents(pubsubName, topicName, contentType, requestMetadata, events));
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public Mono> publishEvents(String pubsubName, String topicName,
- String contentType,
- Map requestMetadata,
- T... events) {
- return observe(
- observation("dapr.client.publish_events")
- .highCardinalityKeyValue("dapr.pubsub.name", safe(pubsubName))
- .highCardinalityKeyValue("dapr.topic.name", safe(topicName)),
- () -> delegate.publishEvents(pubsubName, topicName, contentType, requestMetadata, events));
- }
-
- // -------------------------------------------------------------------------
- // Service Invocation — all deprecated, delegate directly
- // -------------------------------------------------------------------------
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, Object data,
- HttpExtension httpExtension, Map metadata,
- TypeRef type) {
- return delegate.invokeMethod(appId, methodName, data, httpExtension, metadata, type);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, Object request,
- HttpExtension httpExtension, Map metadata,
- Class clazz) {
- return delegate.invokeMethod(appId, methodName, request, httpExtension, metadata, clazz);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, Object request,
- HttpExtension httpExtension, TypeRef type) {
- return delegate.invokeMethod(appId, methodName, request, httpExtension, type);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, Object request,
- HttpExtension httpExtension, Class clazz) {
- return delegate.invokeMethod(appId, methodName, request, httpExtension, clazz);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, HttpExtension httpExtension,
- Map metadata, TypeRef type) {
- return delegate.invokeMethod(appId, methodName, httpExtension, metadata, type);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, HttpExtension httpExtension,
- Map metadata, Class clazz) {
- return delegate.invokeMethod(appId, methodName, httpExtension, metadata, clazz);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, Object request,
- HttpExtension httpExtension, Map metadata) {
- return delegate.invokeMethod(appId, methodName, request, httpExtension, metadata);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, Object request,
- HttpExtension httpExtension) {
- return delegate.invokeMethod(appId, methodName, request, httpExtension);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, HttpExtension httpExtension,
- Map metadata) {
- return delegate.invokeMethod(appId, methodName, httpExtension, metadata);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(String appId, String methodName, byte[] request,
- HttpExtension httpExtension, Map metadata) {
- return delegate.invokeMethod(appId, methodName, request, httpExtension, metadata);
- }
-
- @Override
- @Deprecated
- public Mono invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef type) {
- return delegate.invokeMethod(invokeMethodRequest, type);
- }
-
- // -------------------------------------------------------------------------
- // Bindings
- // -------------------------------------------------------------------------
-
- @Override
- public Mono invokeBinding(String bindingName, String operation, Object data) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(bindingName))
- .highCardinalityKeyValue("dapr.binding.operation", safe(operation)),
- () -> delegate.invokeBinding(bindingName, operation, data));
- }
-
- @Override
- public Mono invokeBinding(String bindingName, String operation, byte[] data,
- Map metadata) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(bindingName))
- .highCardinalityKeyValue("dapr.binding.operation", safe(operation)),
- () -> delegate.invokeBinding(bindingName, operation, data, metadata));
- }
-
- @Override
- public Mono invokeBinding(String bindingName, String operation, Object data,
- TypeRef type) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(bindingName))
- .highCardinalityKeyValue("dapr.binding.operation", safe(operation)),
- () -> delegate.invokeBinding(bindingName, operation, data, type));
- }
-
- @Override
- public Mono invokeBinding(String bindingName, String operation, Object data,
- Class clazz) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(bindingName))
- .highCardinalityKeyValue("dapr.binding.operation", safe(operation)),
- () -> delegate.invokeBinding(bindingName, operation, data, clazz));
- }
-
- @Override
- public Mono invokeBinding(String bindingName, String operation, Object data,
- Map metadata, TypeRef type) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(bindingName))
- .highCardinalityKeyValue("dapr.binding.operation", safe(operation)),
- () -> delegate.invokeBinding(bindingName, operation, data, metadata, type));
- }
-
- @Override
- public Mono invokeBinding(String bindingName, String operation, Object data,
- Map metadata, Class clazz) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(bindingName))
- .highCardinalityKeyValue("dapr.binding.operation", safe(operation)),
- () -> delegate.invokeBinding(bindingName, operation, data, metadata, clazz));
- }
-
- @Override
- public Mono invokeBinding(InvokeBindingRequest request) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(request.getName()))
- .highCardinalityKeyValue("dapr.binding.operation", safe(request.getOperation())),
- () -> delegate.invokeBinding(request));
- }
-
- @Override
- public Mono invokeBinding(InvokeBindingRequest request, TypeRef type) {
- return observe(
- observation("dapr.client.invoke_binding")
- .highCardinalityKeyValue("dapr.binding.name", safe(request.getName()))
- .highCardinalityKeyValue("dapr.binding.operation", safe(request.getOperation())),
- () -> delegate.invokeBinding(request, type));
- }
-
- // -------------------------------------------------------------------------
- // State Management
- // -------------------------------------------------------------------------
-
- @Override
- public Mono> getState(String storeName, State state, TypeRef type) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(state.getKey())),
- () -> delegate.getState(storeName, state, type));
- }
-
- @Override
- public Mono> getState(String storeName, State state, Class clazz) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(state.getKey())),
- () -> delegate.getState(storeName, state, clazz));
- }
-
- @Override
- public Mono> getState(String storeName, String key, TypeRef type) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.getState(storeName, key, type));
- }
-
- @Override
- public Mono> getState(String storeName, String key, Class clazz) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.getState(storeName, key, clazz));
- }
-
- @Override
- public Mono> getState(String storeName, String key, StateOptions options,
- TypeRef type) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.getState(storeName, key, options, type));
- }
-
- @Override
- public Mono> getState(String storeName, String key, StateOptions options,
- Class clazz) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.getState(storeName, key, options, clazz));
- }
-
- @Override
- public Mono> getState(GetStateRequest request, TypeRef type) {
- return observe(
- observation("dapr.client.get_state")
- .highCardinalityKeyValue("dapr.store.name", safe(request.getStoreName()))
- .highCardinalityKeyValue("dapr.state.key", safe(request.getKey())),
- () -> delegate.getState(request, type));
- }
-
- @Override
- public Mono>> getBulkState(String storeName, List keys,
- TypeRef type) {
- return observe(
- observation("dapr.client.get_bulk_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName)),
- () -> delegate.getBulkState(storeName, keys, type));
- }
-
- @Override
- public Mono>> getBulkState(String storeName, List keys,
- Class clazz) {
- return observe(
- observation("dapr.client.get_bulk_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName)),
- () -> delegate.getBulkState(storeName, keys, clazz));
- }
-
- @Override
- public Mono>> getBulkState(GetBulkStateRequest request, TypeRef type) {
- return observe(
- observation("dapr.client.get_bulk_state")
- .highCardinalityKeyValue("dapr.store.name", safe(request.getStoreName())),
- () -> delegate.getBulkState(request, type));
- }
-
- @Override
- public Mono executeStateTransaction(String storeName,
- List> operations) {
- return observe(
- observation("dapr.client.execute_state_transaction")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName)),
- () -> delegate.executeStateTransaction(storeName, operations));
- }
-
- @Override
- public Mono executeStateTransaction(ExecuteStateTransactionRequest request) {
- return observe(
- observation("dapr.client.execute_state_transaction")
- .highCardinalityKeyValue("dapr.store.name", safe(request.getStateStoreName())),
- () -> delegate.executeStateTransaction(request));
- }
-
- @Override
- public Mono saveBulkState(String storeName, List> states) {
- return observe(
- observation("dapr.client.save_bulk_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName)),
- () -> delegate.saveBulkState(storeName, states));
- }
-
- @Override
- public Mono saveBulkState(SaveStateRequest request) {
- return observe(
- observation("dapr.client.save_bulk_state")
- .highCardinalityKeyValue("dapr.store.name", safe(request.getStoreName())),
- () -> delegate.saveBulkState(request));
- }
-
- @Override
- public Mono saveState(String storeName, String key, Object value) {
- return observe(
- observation("dapr.client.save_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.saveState(storeName, key, value));
- }
-
- @Override
- public Mono saveState(String storeName, String key, String etag, Object value,
- StateOptions options) {
- return observe(
- observation("dapr.client.save_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.saveState(storeName, key, etag, value, options));
- }
-
- @Override
- public Mono saveState(String storeName, String key, String etag, Object value,
- Map meta, StateOptions options) {
- return observe(
- observation("dapr.client.save_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.saveState(storeName, key, etag, value, meta, options));
- }
-
- @Override
- public Mono deleteState(String storeName, String key) {
- return observe(
- observation("dapr.client.delete_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.deleteState(storeName, key));
- }
-
- @Override
- public Mono deleteState(String storeName, String key, String etag,
- StateOptions options) {
- return observe(
- observation("dapr.client.delete_state")
- .highCardinalityKeyValue("dapr.store.name", safe(storeName))
- .highCardinalityKeyValue("dapr.state.key", safe(key)),
- () -> delegate.deleteState(storeName, key, etag, options));
- }
-
- @Override
- public Mono deleteState(DeleteStateRequest request) {
- return observe(
- observation("dapr.client.delete_state")
- .highCardinalityKeyValue("dapr.store.name", safe(request.getStateStoreName()))
- .highCardinalityKeyValue("dapr.state.key", safe(request.getKey())),
- () -> delegate.deleteState(request));
- }
-
- // -------------------------------------------------------------------------
- // Secrets
- // -------------------------------------------------------------------------
-
- @Override
- public Mono> getSecret(String storeName, String secretName,
- Map metadata) {
- return observe(
- observation("dapr.client.get_secret")
- .highCardinalityKeyValue("dapr.secret.store", safe(storeName))
- .highCardinalityKeyValue("dapr.secret.name", safe(secretName)),
- () -> delegate.getSecret(storeName, secretName, metadata));
- }
-
- @Override
- public Mono> getSecret(String storeName, String secretName) {
- return observe(
- observation("dapr.client.get_secret")
- .highCardinalityKeyValue("dapr.secret.store", safe(storeName))
- .highCardinalityKeyValue("dapr.secret.name", safe(secretName)),
- () -> delegate.getSecret(storeName, secretName));
- }
-
- @Override
- public Mono> getSecret(GetSecretRequest request) {
- return observe(
- observation("dapr.client.get_secret")
- .highCardinalityKeyValue("dapr.secret.store", safe(request.getStoreName()))
- .highCardinalityKeyValue("dapr.secret.name", safe(request.getKey())),
- () -> delegate.getSecret(request));
- }
-
- @Override
- public Mono>> getBulkSecret(String storeName) {
- return observe(
- observation("dapr.client.get_bulk_secret")
- .highCardinalityKeyValue("dapr.secret.store", safe(storeName)),
- () -> delegate.getBulkSecret(storeName));
- }
-
- @Override
- public Mono>> getBulkSecret(String storeName,
- Map metadata) {
- return observe(
- observation("dapr.client.get_bulk_secret")
- .highCardinalityKeyValue("dapr.secret.store", safe(storeName)),
- () -> delegate.getBulkSecret(storeName, metadata));
- }
-
- @Override
- public Mono>> getBulkSecret(GetBulkSecretRequest request) {
- return observe(
- observation("dapr.client.get_bulk_secret")
- .highCardinalityKeyValue("dapr.secret.store", safe(request.getStoreName())),
- () -> delegate.getBulkSecret(request));
- }
-
- // -------------------------------------------------------------------------
- // Configuration
- // -------------------------------------------------------------------------
-
- @Override
- public Mono getConfiguration(String storeName, String key) {
- return observe(
- observation("dapr.client.get_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName))
- .highCardinalityKeyValue("dapr.configuration.key", safe(key)),
- () -> delegate.getConfiguration(storeName, key));
- }
-
- @Override
- public Mono getConfiguration(String storeName, String key,
- Map metadata) {
- return observe(
- observation("dapr.client.get_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName))
- .highCardinalityKeyValue("dapr.configuration.key", safe(key)),
- () -> delegate.getConfiguration(storeName, key, metadata));
- }
-
- @Override
- public Mono> getConfiguration(String storeName, String... keys) {
- return observe(
- observation("dapr.client.get_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName)),
- () -> delegate.getConfiguration(storeName, keys));
- }
-
- @Override
- public Mono> getConfiguration(String storeName, List keys,
- Map metadata) {
- return observe(
- observation("dapr.client.get_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName)),
- () -> delegate.getConfiguration(storeName, keys, metadata));
- }
-
- @Override
- public Mono> getConfiguration(GetConfigurationRequest request) {
- return observe(
- observation("dapr.client.get_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(request.getStoreName())),
- () -> delegate.getConfiguration(request));
- }
-
- @Override
- public Flux subscribeConfiguration(String storeName,
- String... keys) {
- return observeFlux(
- observation("dapr.client.subscribe_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName)),
- () -> delegate.subscribeConfiguration(storeName, keys));
- }
-
- @Override
- public Flux subscribeConfiguration(String storeName,
- List keys,
- Map metadata) {
- return observeFlux(
- observation("dapr.client.subscribe_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName)),
- () -> delegate.subscribeConfiguration(storeName, keys, metadata));
- }
-
- @Override
- public Flux subscribeConfiguration(
- SubscribeConfigurationRequest request) {
- return observeFlux(
- observation("dapr.client.subscribe_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(request.getStoreName())),
- () -> delegate.subscribeConfiguration(request));
- }
-
- @Override
- public Mono unsubscribeConfiguration(String id,
- String storeName) {
- return observe(
- observation("dapr.client.unsubscribe_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(storeName)),
- () -> delegate.unsubscribeConfiguration(id, storeName));
- }
-
- @Override
- public Mono unsubscribeConfiguration(
- UnsubscribeConfigurationRequest request) {
- return observe(
- observation("dapr.client.unsubscribe_configuration")
- .highCardinalityKeyValue("dapr.configuration.store", safe(request.getStoreName())),
- () -> delegate.unsubscribeConfiguration(request));
- }
-
- // -------------------------------------------------------------------------
- // gRPC Stub — no remote call at creation time, no observation needed
- // -------------------------------------------------------------------------
-
- @Override
- public > T newGrpcStub(String appId, Function stubBuilder) {
- return delegate.newGrpcStub(appId, stubBuilder);
- }
-
- // -------------------------------------------------------------------------
- // Metadata
- // -------------------------------------------------------------------------
-
- @Override
- public Mono getMetadata() {
- return observe(observation("dapr.client.get_metadata"), () -> delegate.getMetadata());
- }
-
- // -------------------------------------------------------------------------
- // Jobs
- // -------------------------------------------------------------------------
-
- @Override
- public Mono scheduleJob(ScheduleJobRequest scheduleJobRequest) {
- return observe(
- observation("dapr.client.schedule_job")
- .highCardinalityKeyValue("dapr.job.name", safe(scheduleJobRequest.getName())),
- () -> delegate.scheduleJob(scheduleJobRequest));
- }
-
- @Override
- public Mono getJob(GetJobRequest getJobRequest) {
- return observe(
- observation("dapr.client.get_job")
- .highCardinalityKeyValue("dapr.job.name", safe(getJobRequest.getName())),
- () -> delegate.getJob(getJobRequest));
- }
-
- @Override
- public Mono deleteJob(DeleteJobRequest deleteJobRequest) {
- return observe(
- observation("dapr.client.delete_job")
- .highCardinalityKeyValue("dapr.job.name", safe(deleteJobRequest.getName())),
- () -> delegate.deleteJob(deleteJobRequest));
- }
-}
diff --git a/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprWorkflowClient.java b/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprWorkflowClient.java
deleted file mode 100644
index 6ce4798a31..0000000000
--- a/dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprWorkflowClient.java
+++ /dev/null
@@ -1,295 +0,0 @@
-/*
- * Copyright 2025 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.spring.boot.autoconfigure.client;
-
-import io.dapr.config.Properties;
-import io.dapr.workflows.Workflow;
-import io.dapr.workflows.client.DaprWorkflowClient;
-import io.dapr.workflows.client.NewWorkflowOptions;
-import io.dapr.workflows.client.WorkflowState;
-import io.micrometer.observation.Observation;
-import io.micrometer.observation.ObservationRegistry;
-
-import javax.annotation.Nullable;
-import java.time.Duration;
-import java.util.Objects;
-import java.util.concurrent.TimeoutException;
-
-/**
- * A {@link DaprWorkflowClient} subclass that creates Micrometer Observation spans (bridged to
- * OpenTelemetry) for each non-deprecated method call.
- *
- * Because this class extends {@link DaprWorkflowClient}, consumers can keep injecting
- * {@code DaprWorkflowClient} without any code changes. Deprecated methods fall through to the
- * parent implementation without any observation.
- *
- *
Constructor note: calling {@code super(properties)} eagerly creates a gRPC
- * {@code ManagedChannel}, but the actual TCP connection is established lazily on the first RPC call,
- * so construction succeeds even when the Dapr sidecar is not yet available.
- */
-public class ObservationDaprWorkflowClient extends DaprWorkflowClient {
-
- private final ObservationRegistry observationRegistry;
-
- /**
- * Creates a new {@code ObservationDaprWorkflowClient}.
- *
- * @param properties connection properties for the underlying gRPC channel
- * @param observationRegistry the Micrometer {@link ObservationRegistry} used to create spans
- */
- public ObservationDaprWorkflowClient(Properties properties,
- ObservationRegistry observationRegistry) {
- super(properties);
- this.observationRegistry = Objects.requireNonNull(observationRegistry,
- "observationRegistry must not be null");
- }
-
- // -------------------------------------------------------------------------
- // Internal helpers
- // -------------------------------------------------------------------------
-
- private Observation observation(String name) {
- return Observation.createNotStarted(name, observationRegistry);
- }
-
- // -------------------------------------------------------------------------
- // scheduleNewWorkflow — only String-based "leaf" overloads are overridden.
- // Class-based overloads in the parent delegate to this.scheduleNewWorkflow(String, ...)
- // via dynamic dispatch, so they naturally pick up these observations.
- // -------------------------------------------------------------------------
-
- @Override
- public String scheduleNewWorkflow(String name) {
- Observation obs = observation("dapr.workflow.schedule")
- .highCardinalityKeyValue("dapr.workflow.name", name)
- .start();
- try {
- return super.scheduleNewWorkflow(name);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- @Override
- public String scheduleNewWorkflow(String name, Object input) {
- Observation obs = observation("dapr.workflow.schedule")
- .highCardinalityKeyValue("dapr.workflow.name", name)
- .start();
- try {
- return super.scheduleNewWorkflow(name, input);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- @Override
- public String scheduleNewWorkflow(String name, Object input,
- String instanceId) {
- Observation obs = observation("dapr.workflow.schedule")
- .highCardinalityKeyValue("dapr.workflow.name", name)
- .highCardinalityKeyValue("dapr.workflow.instance_id",
- instanceId != null ? instanceId : "")
- .start();
- try {
- return super.scheduleNewWorkflow(name, input, instanceId);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- @Override
- public String scheduleNewWorkflow(String name,
- NewWorkflowOptions options) {
- String instanceId = options != null && options.getInstanceId() != null
- ? options.getInstanceId() : "";
- Observation obs = observation("dapr.workflow.schedule")
- .highCardinalityKeyValue("dapr.workflow.name", name)
- .highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
- .start();
- try {
- return super.scheduleNewWorkflow(name, options);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- // -------------------------------------------------------------------------
- // Lifecycle operations
- // -------------------------------------------------------------------------
-
- @Override
- public void suspendWorkflow(String workflowInstanceId, @Nullable String reason) {
- Observation obs = observation("dapr.workflow.suspend")
- .highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
- .start();
- try {
- super.suspendWorkflow(workflowInstanceId, reason);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- @Override
- public void resumeWorkflow(String workflowInstanceId, @Nullable String reason) {
- Observation obs = observation("dapr.workflow.resume")
- .highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
- .start();
- try {
- super.resumeWorkflow(workflowInstanceId, reason);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- @Override
- public void terminateWorkflow(String workflowInstanceId, @Nullable Object output) {
- Observation obs = observation("dapr.workflow.terminate")
- .highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
- .start();
- try {
- super.terminateWorkflow(workflowInstanceId, output);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- // -------------------------------------------------------------------------
- // State queries
- // -------------------------------------------------------------------------
-
- @Override
- @Nullable
- public WorkflowState getWorkflowState(String instanceId, boolean getInputsAndOutputs) {
- Observation obs = observation("dapr.workflow.get_state")
- .highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
- .start();
- try {
- return super.getWorkflowState(instanceId, getInputsAndOutputs);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- // -------------------------------------------------------------------------
- // Waiting
- // -------------------------------------------------------------------------
-
- @Override
- @Nullable
- public WorkflowState waitForWorkflowStart(String instanceId, Duration timeout,
- boolean getInputsAndOutputs) throws TimeoutException {
- Observation obs = observation("dapr.workflow.wait_start")
- .highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
- .start();
- try {
- return super.waitForWorkflowStart(instanceId, timeout, getInputsAndOutputs);
- } catch (TimeoutException e) {
- obs.error(e);
- throw e;
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- @Override
- @Nullable
- public WorkflowState waitForWorkflowCompletion(String instanceId, Duration timeout,
- boolean getInputsAndOutputs)
- throws TimeoutException {
- Observation obs = observation("dapr.workflow.wait_completion")
- .highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
- .start();
- try {
- return super.waitForWorkflowCompletion(instanceId, timeout, getInputsAndOutputs);
- } catch (TimeoutException e) {
- obs.error(e);
- throw e;
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- // -------------------------------------------------------------------------
- // Events
- // -------------------------------------------------------------------------
-
- @Override
- public void raiseEvent(String workflowInstanceId, String eventName, Object eventPayload) {
- Observation obs = observation("dapr.workflow.raise_event")
- .highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
- .highCardinalityKeyValue("dapr.workflow.event_name", eventName)
- .start();
- try {
- super.raiseEvent(workflowInstanceId, eventName, eventPayload);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- // -------------------------------------------------------------------------
- // Cleanup
- // -------------------------------------------------------------------------
-
- @Override
- public boolean purgeWorkflow(String workflowInstanceId) {
- Observation obs = observation("dapr.workflow.purge")
- .highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
- .start();
- try {
- return super.purgeWorkflow(workflowInstanceId);
- } catch (RuntimeException e) {
- obs.error(e);
- throw e;
- } finally {
- obs.stop();
- }
- }
-
- // Deprecated methods (getInstanceState, waitForInstanceStart, waitForInstanceCompletion,
- // purgeInstance) are intentionally not overridden — they fall through to the parent
- // implementation without any observation.
-}
diff --git a/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/DaprClientObservationAutoConfigurationTest.java b/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/DaprClientObservationAutoConfigurationTest.java
index d73945b7c4..5141af8021 100644
--- a/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/DaprClientObservationAutoConfigurationTest.java
+++ b/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/DaprClientObservationAutoConfigurationTest.java
@@ -15,6 +15,8 @@
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
+import io.dapr.spring.observation.client.ObservationDaprClient;
+import io.dapr.spring.observation.client.ObservationDaprWorkflowClient;
import io.dapr.workflows.client.DaprWorkflowClient;
import io.micrometer.observation.ObservationRegistry;
import io.micrometer.observation.tck.TestObservationRegistry;
diff --git a/dapr-spring/dapr-spring-boot-observation/pom.xml b/dapr-spring/dapr-spring-boot-observation/pom.xml
new file mode 100644
index 0000000000..c78b98cc94
--- /dev/null
+++ b/dapr-spring/dapr-spring-boot-observation/pom.xml
@@ -0,0 +1,73 @@
+
+
+ 4.0.0
+
+ io.dapr.spring
+ dapr-spring-parent
+ 1.18.0-SNAPSHOT
+ ../pom.xml
+
+
+ dapr-spring-boot-observation
+ dapr-spring-boot-observation
+ Spring-agnostic Micrometer Observation decorators for DaprClient and DaprWorkflowClient, with OpenTelemetry trace propagation to the Dapr sidecar over gRPC.
+ jar
+
+
+
+ io.dapr
+ dapr-sdk
+
+
+ io.dapr
+ dapr-sdk-workflows
+
+
+ io.micrometer
+ micrometer-observation
+
+
+ io.opentelemetry
+ opentelemetry-api
+
+
+
+
+ io.micrometer
+ micrometer-observation-test
+ test
+
+
+ org.junit.jupiter
+ junit-jupiter
+ test
+
+
+ org.mockito
+ mockito-core
+ test
+
+
+ org.mockito
+ mockito-junit-jupiter
+ ${mockito.version}
+ test
+
+
+ org.assertj
+ assertj-core
+ test
+
+
+
+
+
+
+ org.sonatype.plugins
+ nexus-staging-maven-plugin
+
+
+
+
diff --git a/dapr-spring/dapr-spring-boot-4-autoconfigure/src/main/java/io/dapr/spring/boot4/autoconfigure/client/ObservationDaprClient.java b/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/ObservationDaprClient.java
similarity index 94%
rename from dapr-spring/dapr-spring-boot-4-autoconfigure/src/main/java/io/dapr/spring/boot4/autoconfigure/client/ObservationDaprClient.java
rename to dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/ObservationDaprClient.java
index 8b76f10479..575a1887fc 100644
--- a/dapr-spring/dapr-spring-boot-4-autoconfigure/src/main/java/io/dapr/spring/boot4/autoconfigure/client/ObservationDaprClient.java
+++ b/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/ObservationDaprClient.java
@@ -11,7 +11,7 @@
limitations under the License.
*/
-package io.dapr.spring.boot4.autoconfigure.client;
+package io.dapr.spring.observation.client;
import io.dapr.client.DaprClient;
import io.dapr.client.domain.BulkPublishRequest;
@@ -46,8 +46,11 @@
import io.grpc.stub.AbstractStub;
import io.micrometer.observation.Observation;
import io.micrometer.observation.ObservationRegistry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.util.context.Context;
import java.util.List;
import java.util.Map;
@@ -86,21 +89,53 @@ public ObservationDaprClient(DaprClient delegate, ObservationRegistry observatio
private Mono observe(Observation obs, Supplier> monoSupplier) {
return Mono.defer(() -> {
obs.start();
+ // Open a scope so the Micrometer-OTel bridge makes the new span current in the
+ // OTel thread-local; capture its W3C traceparent immediately, then close the scope.
+ // The captured context is written into the Reactor context so the downstream gRPC
+ // call (inside DaprClientImpl.deferContextual) uses this span as its parent.
+ SpanContext spanCtx;
+ try (Observation.Scope ignored = obs.openScope()) {
+ spanCtx = Span.current().getSpanContext();
+ }
return monoSupplier.get()
.doOnError(obs::error)
- .doFinally(signal -> obs.stop());
+ .doFinally(signal -> obs.stop())
+ .contextWrite(ctx -> enrichWithSpanContext(ctx, spanCtx));
});
}
private Flux observeFlux(Observation obs, Supplier> fluxSupplier) {
return Flux.defer(() -> {
obs.start();
+ SpanContext spanCtx;
+ try (Observation.Scope ignored = obs.openScope()) {
+ spanCtx = Span.current().getSpanContext();
+ }
return fluxSupplier.get()
.doOnError(obs::error)
- .doFinally(signal -> obs.stop());
+ .doFinally(signal -> obs.stop())
+ .contextWrite(ctx -> enrichWithSpanContext(ctx, spanCtx));
});
}
+ /**
+ * Enriches the Reactor {@link Context} with W3C {@code traceparent} (and optionally
+ * {@code tracestate}) extracted from the given OTel {@link SpanContext}.
+ * This bridges the Micrometer Observation span into the string-keyed Reactor context that
+ * {@link io.dapr.client.DaprClientImpl} reads to populate gRPC metadata headers.
+ */
+ private static Context enrichWithSpanContext(Context ctx, SpanContext spanCtx) {
+ if (spanCtx == null || !spanCtx.isValid()) {
+ return ctx;
+ }
+ ctx = ctx.put("traceparent", TraceContextFormat.formatW3cTraceparent(spanCtx));
+ String traceState = TraceContextFormat.formatTraceState(spanCtx);
+ if (!traceState.isEmpty()) {
+ ctx = ctx.put("tracestate", traceState);
+ }
+ return ctx;
+ }
+
private static String safe(String value) {
return value != null ? value : "";
}
diff --git a/dapr-spring/dapr-spring-boot-4-autoconfigure/src/main/java/io/dapr/spring/boot4/autoconfigure/client/ObservationDaprWorkflowClient.java b/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/ObservationDaprWorkflowClient.java
similarity index 66%
rename from dapr-spring/dapr-spring-boot-4-autoconfigure/src/main/java/io/dapr/spring/boot4/autoconfigure/client/ObservationDaprWorkflowClient.java
rename to dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/ObservationDaprWorkflowClient.java
index b1f3ec611c..4366857e75 100644
--- a/dapr-spring/dapr-spring-boot-4-autoconfigure/src/main/java/io/dapr/spring/boot4/autoconfigure/client/ObservationDaprWorkflowClient.java
+++ b/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/ObservationDaprWorkflowClient.java
@@ -11,15 +11,26 @@
limitations under the License.
*/
-package io.dapr.spring.boot4.autoconfigure.client;
+package io.dapr.spring.observation.client;
import io.dapr.config.Properties;
+import io.dapr.internal.opencensus.GrpcHelper;
import io.dapr.workflows.Workflow;
import io.dapr.workflows.client.DaprWorkflowClient;
import io.dapr.workflows.client.NewWorkflowOptions;
import io.dapr.workflows.client.WorkflowState;
+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.micrometer.observation.Observation;
import io.micrometer.observation.ObservationRegistry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import reactor.util.context.Context;
import javax.annotation.Nullable;
import java.time.Duration;
@@ -34,12 +45,22 @@
* {@code DaprWorkflowClient} without any code changes. Deprecated methods fall through to the
* parent implementation without any observation.
*
- * Constructor note: calling {@code super(properties)} eagerly creates a gRPC
+ *
Trace propagation: an {@link OtelTracingClientInterceptor} is registered on the gRPC
+ * channel. For each synchronous workflow RPC, the observation opens an OTel scope (via
+ * {@link Observation#openScope()}) before calling {@code super.*}, making the observation span the
+ * current OTel span in thread-local. The interceptor then reads {@link Span#current()} and injects
+ * its W3C {@code traceparent} (and {@code grpc-trace-bin}) into the gRPC request headers so the
+ * Dapr sidecar receives the full trace context.
+ *
+ *
Constructor note: calling {@code super(properties, interceptor)} eagerly creates a gRPC
* {@code ManagedChannel}, but the actual TCP connection is established lazily on the first RPC call,
* so construction succeeds even when the Dapr sidecar is not yet available.
*/
public class ObservationDaprWorkflowClient extends DaprWorkflowClient {
+ private static final OtelTracingClientInterceptor TRACING_INTERCEPTOR =
+ new OtelTracingClientInterceptor();
+
private final ObservationRegistry observationRegistry;
/**
@@ -50,7 +71,7 @@ public class ObservationDaprWorkflowClient extends DaprWorkflowClient {
*/
public ObservationDaprWorkflowClient(Properties properties,
ObservationRegistry observationRegistry) {
- super(properties);
+ super(properties, TRACING_INTERCEPTOR);
this.observationRegistry = Objects.requireNonNull(observationRegistry,
"observationRegistry must not be null");
}
@@ -75,7 +96,9 @@ public String scheduleNewWorkflow(String name) {
.highCardinalityKeyValue("dapr.workflow.name", name)
.start();
try {
- return super.scheduleNewWorkflow(name);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.scheduleNewWorkflow(name);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -90,7 +113,9 @@ public String scheduleNewWorkflow(String name, Object input
.highCardinalityKeyValue("dapr.workflow.name", name)
.start();
try {
- return super.scheduleNewWorkflow(name, input);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.scheduleNewWorkflow(name, input);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -108,7 +133,9 @@ public String scheduleNewWorkflow(String name, Object input
instanceId != null ? instanceId : "")
.start();
try {
- return super.scheduleNewWorkflow(name, input, instanceId);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.scheduleNewWorkflow(name, input, instanceId);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -127,7 +154,9 @@ public String scheduleNewWorkflow(String name,
.highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
.start();
try {
- return super.scheduleNewWorkflow(name, options);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.scheduleNewWorkflow(name, options);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -146,7 +175,9 @@ public void suspendWorkflow(String workflowInstanceId, @Nullable String reason)
.highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
.start();
try {
- super.suspendWorkflow(workflowInstanceId, reason);
+ try (Observation.Scope ignored = obs.openScope()) {
+ super.suspendWorkflow(workflowInstanceId, reason);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -161,7 +192,9 @@ public void resumeWorkflow(String workflowInstanceId, @Nullable String reason) {
.highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
.start();
try {
- super.resumeWorkflow(workflowInstanceId, reason);
+ try (Observation.Scope ignored = obs.openScope()) {
+ super.resumeWorkflow(workflowInstanceId, reason);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -176,7 +209,9 @@ public void terminateWorkflow(String workflowInstanceId, @Nullable Object output
.highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
.start();
try {
- super.terminateWorkflow(workflowInstanceId, output);
+ try (Observation.Scope ignored = obs.openScope()) {
+ super.terminateWorkflow(workflowInstanceId, output);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -196,7 +231,9 @@ public WorkflowState getWorkflowState(String instanceId, boolean getInputsAndOut
.highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
.start();
try {
- return super.getWorkflowState(instanceId, getInputsAndOutputs);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.getWorkflowState(instanceId, getInputsAndOutputs);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -217,7 +254,9 @@ public WorkflowState waitForWorkflowStart(String instanceId, Duration timeout,
.highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
.start();
try {
- return super.waitForWorkflowStart(instanceId, timeout, getInputsAndOutputs);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.waitForWorkflowStart(instanceId, timeout, getInputsAndOutputs);
+ }
} catch (TimeoutException e) {
obs.error(e);
throw e;
@@ -238,7 +277,9 @@ public WorkflowState waitForWorkflowCompletion(String instanceId, Duration timeo
.highCardinalityKeyValue("dapr.workflow.instance_id", instanceId)
.start();
try {
- return super.waitForWorkflowCompletion(instanceId, timeout, getInputsAndOutputs);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.waitForWorkflowCompletion(instanceId, timeout, getInputsAndOutputs);
+ }
} catch (TimeoutException e) {
obs.error(e);
throw e;
@@ -261,7 +302,9 @@ public void raiseEvent(String workflowInstanceId, String eventName, Object event
.highCardinalityKeyValue("dapr.workflow.event_name", eventName)
.start();
try {
- super.raiseEvent(workflowInstanceId, eventName, eventPayload);
+ try (Observation.Scope ignored = obs.openScope()) {
+ super.raiseEvent(workflowInstanceId, eventName, eventPayload);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -280,7 +323,9 @@ public boolean purgeWorkflow(String workflowInstanceId) {
.highCardinalityKeyValue("dapr.workflow.instance_id", workflowInstanceId)
.start();
try {
- return super.purgeWorkflow(workflowInstanceId);
+ try (Observation.Scope ignored = obs.openScope()) {
+ return super.purgeWorkflow(workflowInstanceId);
+ }
} catch (RuntimeException e) {
obs.error(e);
throw e;
@@ -292,4 +337,43 @@ public boolean purgeWorkflow(String workflowInstanceId) {
// Deprecated methods (getInstanceState, waitForInstanceStart, waitForInstanceCompletion,
// purgeInstance) are intentionally not overridden — they fall through to the parent
// implementation without any observation.
+
+ // -------------------------------------------------------------------------
+ // gRPC interceptor: injects the current OTel span's traceparent into headers
+ // -------------------------------------------------------------------------
+
+ /**
+ * A gRPC {@link ClientInterceptor} that reads the current OTel span from the thread-local
+ * context (set by {@link Observation#openScope()}) and injects its W3C {@code traceparent},
+ * {@code tracestate}, and {@code grpc-trace-bin} headers into every outbound RPC call.
+ *
+ * The interceptor is stateless: it reads {@link Span#current()} lazily at call time, so the
+ * same instance can be shared across all calls on the channel.
+ */
+ private static final class OtelTracingClientInterceptor implements ClientInterceptor {
+
+ @Override
+ public ClientCall interceptCall(
+ MethodDescriptor method, CallOptions options, Channel channel) {
+ return new ForwardingClientCall.SimpleForwardingClientCall<>(channel.newCall(method, options)) {
+ @Override
+ public void start(Listener responseListener, Metadata headers) {
+ SpanContext spanCtx = Span.current().getSpanContext();
+ if (spanCtx.isValid()) {
+ // Build a Reactor Context with the OTel span's values and delegate to GrpcHelper,
+ // which writes traceparent, tracestate AND grpc-trace-bin (the binary format that
+ // older Dapr sidecar versions require for gRPC trace propagation).
+ Context reactorCtx = Context.of("traceparent",
+ TraceContextFormat.formatW3cTraceparent(spanCtx));
+ String traceState = TraceContextFormat.formatTraceState(spanCtx);
+ if (!traceState.isEmpty()) {
+ reactorCtx = reactorCtx.put("tracestate", traceState);
+ }
+ GrpcHelper.populateMetadata(reactorCtx, headers);
+ }
+ super.start(responseListener, headers);
+ }
+ };
+ }
+ }
}
diff --git a/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/TraceContextFormat.java b/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/TraceContextFormat.java
new file mode 100644
index 0000000000..d5e3470907
--- /dev/null
+++ b/dapr-spring/dapr-spring-boot-observation/src/main/java/io/dapr/spring/observation/client/TraceContextFormat.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2025 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.spring.observation.client;
+
+import io.opentelemetry.api.trace.SpanContext;
+
+/**
+ * Formats an OpenTelemetry {@link SpanContext} as the W3C Trace Context header values
+ * ({@code traceparent} and {@code tracestate}).
+ *
+ * Package-private: shared between {@link ObservationDaprClient} (Reactor context for async
+ * gRPC calls on {@code DaprClient}) and {@link ObservationDaprWorkflowClient} (gRPC
+ * {@code ClientInterceptor} for synchronous workflow calls). This keeps a single source of truth
+ * for the on-wire trace format.
+ */
+final class TraceContextFormat {
+
+ private TraceContextFormat() {
+ }
+
+ /**
+ * Format a {@link SpanContext} as the value of the W3C {@code traceparent} header (version 00).
+ */
+ static String formatW3cTraceparent(SpanContext spanCtx) {
+ return "00-" + spanCtx.getTraceId() + "-" + spanCtx.getSpanId()
+ + "-" + spanCtx.getTraceFlags().asHex();
+ }
+
+ /**
+ * Format a {@link SpanContext}'s trace state as the value of the W3C {@code tracestate} header.
+ * Returns an empty string if the trace state is empty.
+ */
+ static String formatTraceState(SpanContext spanCtx) {
+ if (spanCtx.getTraceState().isEmpty()) {
+ return "";
+ }
+ StringBuilder sb = new StringBuilder();
+ spanCtx.getTraceState().forEach((k, v) -> {
+ if (sb.length() > 0) {
+ sb.append(',');
+ }
+ sb.append(k).append('=').append(v);
+ });
+ return sb.toString();
+ }
+}
diff --git a/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprClientTest.java b/dapr-spring/dapr-spring-boot-observation/src/test/java/io/dapr/spring/observation/client/ObservationDaprClientTest.java
similarity index 99%
rename from dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprClientTest.java
rename to dapr-spring/dapr-spring-boot-observation/src/test/java/io/dapr/spring/observation/client/ObservationDaprClientTest.java
index b7bb909149..e37a332df2 100644
--- a/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprClientTest.java
+++ b/dapr-spring/dapr-spring-boot-observation/src/test/java/io/dapr/spring/observation/client/ObservationDaprClientTest.java
@@ -11,7 +11,7 @@
limitations under the License.
*/
-package io.dapr.spring.boot.autoconfigure.client;
+package io.dapr.spring.observation.client;
import io.dapr.client.DaprClient;
import io.dapr.client.domain.DeleteStateRequest;
diff --git a/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprWorkflowClientTest.java b/dapr-spring/dapr-spring-boot-observation/src/test/java/io/dapr/spring/observation/client/ObservationDaprWorkflowClientTest.java
similarity index 99%
rename from dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprWorkflowClientTest.java
rename to dapr-spring/dapr-spring-boot-observation/src/test/java/io/dapr/spring/observation/client/ObservationDaprWorkflowClientTest.java
index f60dd2cf5a..ede8f49ce8 100644
--- a/dapr-spring/dapr-spring-boot-autoconfigure/src/test/java/io/dapr/spring/boot/autoconfigure/client/ObservationDaprWorkflowClientTest.java
+++ b/dapr-spring/dapr-spring-boot-observation/src/test/java/io/dapr/spring/observation/client/ObservationDaprWorkflowClientTest.java
@@ -11,7 +11,7 @@
limitations under the License.
*/
-package io.dapr.spring.boot.autoconfigure.client;
+package io.dapr.spring.observation.client;
import io.dapr.config.Properties;
import io.dapr.workflows.client.DaprWorkflowClient;
diff --git a/dapr-spring/pom.xml b/dapr-spring/pom.xml
index d411ceb51f..d2f35baaab 100644
--- a/dapr-spring/pom.xml
+++ b/dapr-spring/pom.xml
@@ -24,6 +24,7 @@
dapr-spring-messaging
dapr-spring-workflows
dapr-spring-boot-properties
+ dapr-spring-boot-observation
dapr-spring-boot-autoconfigure
dapr-spring-boot-4-autoconfigure
dapr-spring-boot-tests
@@ -76,6 +77,11 @@
dapr-spring-boot-properties
${project.version}
+