Skip to content

Commit be05a47

Browse files
artur-ciocanuArtur Ciocanu
artur-ciocanu
and
Artur Ciocanu
authored
Add Micrometer Observation support to Spring Dapr Messaging (#1150)
* Add Micrometer Observation support to Spring Dapr Messaging Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> * Remove direct Micrometer deps it is part of Spring Boot Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> * Remove another explicit dependency Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> * Hide default observation convention implementation Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> * Fix typo in default message builder Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> * Ensure trace is properly sent using OTEL Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> --------- Signed-off-by: Artur Ciocanu <ciocanu@adobe.com> Co-authored-by: Artur Ciocanu <ciocanu@adobe.com>
1 parent 0b7a051 commit be05a47

File tree

11 files changed

+448
-16
lines changed

11 files changed

+448
-16
lines changed

‎dapr-spring/dapr-spring-boot-autoconfigure/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,8 @@
2828
<optional>true</optional>
2929
</dependency>
3030
<dependency>
31-
<groupId>org.springframework.boot</groupId>
32-
<artifactId>spring-boot-starter</artifactId>
31+
<groupId>org.springframework.boot</groupId>
32+
<artifactId>spring-boot-starter</artifactId>
3333
</dependency>
3434
<dependency>
3535
<groupId>org.springframework.boot</groupId>

‎dapr-spring/dapr-spring-boot-autoconfigure/src/main/java/io/dapr/spring/boot/autoconfigure/pubsub/DaprPubSubProperties.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ public class DaprPubSubProperties {
2525
* Name of the PubSub Dapr component.
2626
*/
2727
private String name;
28+
private boolean observationEnabled;
2829

2930
public String getName() {
3031
return name;
@@ -34,4 +35,11 @@ public void setName(String name) {
3435
this.name = name;
3536
}
3637

38+
public boolean isObservationEnabled() {
39+
return observationEnabled;
40+
}
41+
42+
public void setObservationEnabled(boolean observationEnabled) {
43+
this.observationEnabled = observationEnabled;
44+
}
3745
}

‎dapr-spring/dapr-spring-messaging/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,6 @@
1212
<artifactId>dapr-spring-messaging</artifactId>
1313
<name>dapr-spring-messaging</name>
1414
<description>Dapr Spring Messaging</description>
15-
<packaging>jar</packaging>
15+
<packaging>jar</packaging>
1616

1717
</project>

‎dapr-spring/dapr-spring-messaging/src/main/java/io/dapr/spring/messaging/DaprMessagingTemplate.java‎

Lines changed: 149 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -15,20 +15,104 @@
1515

1616
import io.dapr.client.DaprClient;
1717
import io.dapr.client.domain.Metadata;
18+
import io.dapr.spring.messaging.observation.DaprMessagingObservationConvention;
19+
import io.dapr.spring.messaging.observation.DaprMessagingObservationDocumentation;
20+
import io.dapr.spring.messaging.observation.DaprMessagingSenderContext;
21+
import io.micrometer.observation.Observation;
22+
import io.micrometer.observation.ObservationRegistry;
23+
import io.opentelemetry.api.OpenTelemetry;
24+
import io.opentelemetry.context.propagation.TextMapSetter;
25+
import org.slf4j.Logger;
26+
import org.slf4j.LoggerFactory;
27+
import org.springframework.beans.factory.BeanNameAware;
28+
import org.springframework.beans.factory.SmartInitializingSingleton;
29+
import org.springframework.context.ApplicationContext;
30+
import org.springframework.context.ApplicationContextAware;
1831
import reactor.core.publisher.Mono;
32+
import reactor.util.context.Context;
1933

34+
import javax.annotation.Nullable;
35+
36+
import java.util.HashMap;
2037
import java.util.Map;
2138

22-
public class DaprMessagingTemplate<T> implements DaprMessagingOperations<T> {
39+
/**
40+
* Create a new DaprMessagingTemplate.
41+
* @param <T> templated message type
42+
*/
43+
public class DaprMessagingTemplate<T> implements DaprMessagingOperations<T>, ApplicationContextAware, BeanNameAware,
44+
SmartInitializingSingleton {
2345

46+
private static final Logger LOGGER = LoggerFactory.getLogger(DaprMessagingTemplate.class);
2447
private static final String MESSAGE_TTL_IN_SECONDS = "10";
48+
private static final DaprMessagingObservationConvention DEFAULT_OBSERVATION_CONVENTION =
49+
DaprMessagingObservationConvention.getDefault();
2550

2651
private final DaprClient daprClient;
2752
private final String pubsubName;
53+
private final Map<String, String> metadata;
54+
private final boolean observationEnabled;
55+
56+
@Nullable
57+
private ApplicationContext applicationContext;
58+
59+
@Nullable
60+
private String beanName;
61+
62+
@Nullable
63+
private OpenTelemetry openTelemetry;
64+
65+
@Nullable
66+
private ObservationRegistry observationRegistry;
2867

29-
public DaprMessagingTemplate(DaprClient daprClient, String pubsubName) {
68+
@Nullable
69+
private DaprMessagingObservationConvention observationConvention;
70+
71+
/**
72+
* Constructs a new DaprMessagingTemplate.
73+
* @param daprClient Dapr client
74+
* @param pubsubName pubsub name
75+
* @param observationEnabled whether to enable observations
76+
*/
77+
public DaprMessagingTemplate(DaprClient daprClient, String pubsubName, boolean observationEnabled) {
3078
this.daprClient = daprClient;
3179
this.pubsubName = pubsubName;
80+
this.metadata = Map.of(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS);
81+
this.observationEnabled = observationEnabled;
82+
}
83+
84+
@Override
85+
public void setApplicationContext(ApplicationContext applicationContext) {
86+
this.applicationContext = applicationContext;
87+
}
88+
89+
@Override
90+
public void setBeanName(String beanName) {
91+
this.beanName = beanName;
92+
}
93+
94+
/**
95+
* If observations are enabled, attempt to obtain the Observation registry and
96+
* convention.
97+
*/
98+
@Override
99+
public void afterSingletonsInstantiated() {
100+
if (!observationEnabled) {
101+
LOGGER.debug("Observations are not enabled - not recording");
102+
return;
103+
}
104+
105+
if (applicationContext == null) {
106+
LOGGER.warn("Observations enabled but application context null - not recording");
107+
return;
108+
}
109+
110+
observationRegistry = applicationContext.getBeanProvider(ObservationRegistry.class)
111+
.getIfUnique(() -> observationRegistry);
112+
this.openTelemetry = this.applicationContext.getBeanProvider(OpenTelemetry.class)
113+
.getIfUnique(() -> this.openTelemetry);
114+
observationConvention = applicationContext.getBeanProvider(DaprMessagingObservationConvention.class)
115+
.getIfUnique(() -> observationConvention);
32116
}
33117

34118
@Override
@@ -38,29 +122,83 @@ public void send(String topic, T message) {
38122

39123
@Override
40124
public SendMessageBuilder<T> newMessage(T message) {
41-
return new SendMessageBuilderImpl<>(this, message);
125+
return new DefaultSendMessageBuilder<>(this, message);
42126
}
43127

44128
private void doSend(String topic, T message) {
45129
doSendAsync(topic, message).block();
46130
}
47131

48132
private Mono<Void> doSendAsync(String topic, T message) {
49-
return daprClient.publishEvent(pubsubName,
50-
topic,
51-
message,
52-
Map.of(Metadata.TTL_IN_SECONDS, MESSAGE_TTL_IN_SECONDS));
133+
LOGGER.trace("Sending message to '{}' topic", topic);
134+
135+
if (canUseObservation()) {
136+
return publishEventWithObservation(pubsubName, topic, message);
137+
}
138+
139+
return publishEvent(pubsubName, topic, message);
140+
}
141+
142+
private boolean canUseObservation() {
143+
return observationEnabled
144+
&& observationRegistry != null
145+
&& openTelemetry != null
146+
&& beanName != null;
147+
}
148+
149+
private Mono<Void> publishEvent(String pubsubName, String topic, T message) {
150+
return daprClient.publishEvent(pubsubName, topic, message, metadata);
151+
}
152+
153+
private Mono<Void> publishEventWithObservation(String pubsubName, String topic, T message) {
154+
DaprMessagingSenderContext senderContext = DaprMessagingSenderContext.newContext(topic, this.beanName);
155+
Observation observation = createObservation(senderContext);
156+
157+
return observation.observe(() ->
158+
publishEvent(pubsubName, topic, message)
159+
.contextWrite(getReactorContext())
160+
.doOnError(err -> {
161+
LOGGER.error("Failed to send msg to '{}' topic", topic, err);
162+
163+
observation.error(err);
164+
observation.stop();
165+
})
166+
.doOnSuccess(ignore -> {
167+
LOGGER.trace("Sent msg to '{}' topic", topic);
168+
169+
observation.stop();
170+
})
171+
);
172+
}
173+
174+
private Context getReactorContext() {
175+
Map<String, String> map = new HashMap<>();
176+
TextMapSetter<Map<String, String>> setter = (carrier, key, value) -> map.put(key, value);
177+
io.opentelemetry.context.Context otelContext = io.opentelemetry.context.Context.current();
178+
179+
openTelemetry.getPropagators().getTextMapPropagator().inject(otelContext, map, setter);
180+
181+
return Context.of(map);
182+
}
183+
184+
private Observation createObservation(DaprMessagingSenderContext senderContext) {
185+
return DaprMessagingObservationDocumentation.TEMPLATE_OBSERVATION.observation(
186+
observationConvention,
187+
DEFAULT_OBSERVATION_CONVENTION,
188+
() -> senderContext,
189+
observationRegistry
190+
);
53191
}
54192

55-
private static class SendMessageBuilderImpl<T> implements SendMessageBuilder<T> {
193+
private static class DefaultSendMessageBuilder<T> implements SendMessageBuilder<T> {
56194

57195
private final DaprMessagingTemplate<T> template;
58196

59197
private final T message;
60198

61199
private String topic;
62200

63-
SendMessageBuilderImpl(DaprMessagingTemplate<T> template, T message) {
201+
DefaultSendMessageBuilder(DaprMessagingTemplate<T> template, T message) {
64202
this.template = template;
65203
this.message = message;
66204
}
@@ -74,12 +212,12 @@ public SendMessageBuilder<T> withTopic(String topic) {
74212

75213
@Override
76214
public void send() {
77-
this.template.doSend(this.topic, this.message);
215+
template.doSend(topic, message);
78216
}
79217

80218
@Override
81219
public Mono<Void> sendAsync() {
82-
return this.template.doSendAsync(this.topic, this.message);
220+
return template.doSendAsync(topic, message);
83221
}
84222

85223
}

‎dapr-spring/dapr-spring-messaging/src/main/java/io/dapr/spring/messaging/observation/DaprMessagingObservationConvention.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
/*
2+
* Copyright 2024 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.spring.messaging.observation;
15+
16+
import io.micrometer.observation.Observation.Context;
17+
import io.micrometer.observation.ObservationConvention;
18+
19+
/**
20+
* {@link ObservationConvention} for Dapr Messaging.
21+
*
22+
*/
23+
public interface DaprMessagingObservationConvention extends ObservationConvention<DaprMessagingSenderContext> {
24+
25+
@Override
26+
default boolean supportsContext(Context context) {
27+
return context instanceof DaprMessagingSenderContext;
28+
}
29+
30+
@Override
31+
default String getName() {
32+
return "spring.dapr.messaging.template";
33+
}
34+
35+
static DaprMessagingObservationConvention getDefault() {
36+
return DefaultDaprMessagingObservationConvention.INSTANCE;
37+
}
38+
39+
}

‎dapr-spring/dapr-spring-messaging/src/main/java/io/dapr/spring/messaging/observation/DaprMessagingObservationDocumentation.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
/*
2+
* Copyright 2024 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.spring.messaging.observation;
15+
16+
import io.micrometer.common.docs.KeyName;
17+
import io.micrometer.observation.Observation;
18+
import io.micrometer.observation.Observation.Context;
19+
import io.micrometer.observation.ObservationConvention;
20+
import io.micrometer.observation.docs.ObservationDocumentation;
21+
22+
/**
23+
* An {@link Observation} for {@link io.dapr.spring.messaging.DaprMessagingTemplate}.
24+
*
25+
*/
26+
public enum DaprMessagingObservationDocumentation implements ObservationDocumentation {
27+
28+
/**
29+
* Observation created when a Dapr template sends a message.
30+
*/
31+
TEMPLATE_OBSERVATION {
32+
33+
@Override
34+
public Class<? extends ObservationConvention<? extends Context>> getDefaultConvention() {
35+
return DefaultDaprMessagingObservationConvention.class;
36+
}
37+
38+
@Override
39+
public String getPrefix() {
40+
return "spring.dapr.messaging.template";
41+
}
42+
43+
@Override
44+
public KeyName[] getLowCardinalityKeyNames() {
45+
return TemplateLowCardinalityTags.values();
46+
}
47+
};
48+
49+
/**
50+
* Low cardinality tags.
51+
*/
52+
public enum TemplateLowCardinalityTags implements KeyName {
53+
/**
54+
* Bean name of the template that sent the message.
55+
*/
56+
BEAN_NAME {
57+
58+
@Override
59+
public String asString() {
60+
return "spring.dapr.messaging.template.name";
61+
}
62+
}
63+
}
64+
}

0 commit comments

Comments
 (0)