Skip to content

Commit 06ff4ad

Browse files
thjaecklethjaeckle
authored
Merge pull request #2060 from beyonnex-io/bugfix/fix-traces
fixed tracing and moving traces to their correct parent in many places
2 parents a9ebd6e + 61ffa98 commit 06ff4ad

File tree

30 files changed

+399
-195
lines changed

30 files changed

+399
-195
lines changed

‎connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/persistence/ConnectionPersistenceActor.java

+31-24
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,21 @@
3737
import javax.annotation.Nullable;
3838
import javax.annotation.concurrent.Immutable;
3939

40+
import org.apache.pekko.actor.ActorRef;
41+
import org.apache.pekko.actor.ActorSystem;
42+
import org.apache.pekko.actor.Props;
43+
import org.apache.pekko.actor.Status;
44+
import org.apache.pekko.actor.SupervisorStrategy;
45+
import org.apache.pekko.cluster.Cluster;
46+
import org.apache.pekko.cluster.routing.ClusterRouterPool;
47+
import org.apache.pekko.cluster.routing.ClusterRouterPoolSettings;
48+
import org.apache.pekko.japi.pf.ReceiveBuilder;
49+
import org.apache.pekko.pattern.Patterns;
50+
import org.apache.pekko.persistence.RecoveryCompleted;
51+
import org.apache.pekko.routing.Broadcast;
52+
import org.apache.pekko.routing.ConsistentHashingPool;
53+
import org.apache.pekko.routing.ConsistentHashingRouter;
54+
import org.apache.pekko.routing.Pool;
4055
import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
4156
import org.eclipse.ditto.base.model.exceptions.DittoRuntimeExceptionBuilder;
4257
import org.eclipse.ditto.base.model.headers.DittoHeaders;
@@ -108,14 +123,14 @@
108123
import org.eclipse.ditto.connectivity.service.messaging.validation.DittoConnectivityCommandValidator;
109124
import org.eclipse.ditto.connectivity.service.util.ConnectionPubSub;
110125
import org.eclipse.ditto.connectivity.service.util.ConnectivityMdcEntryKey;
111-
import org.eclipse.ditto.internal.utils.pekko.PingCommand;
112-
import org.eclipse.ditto.internal.utils.pekko.logging.CommonMdcEntryKey;
113-
import org.eclipse.ditto.internal.utils.pekko.logging.DittoDiagnosticLoggingAdapter;
114-
import org.eclipse.ditto.internal.utils.pekko.logging.DittoLoggerFactory;
115126
import org.eclipse.ditto.internal.utils.cluster.DistPubSubAccess;
116127
import org.eclipse.ditto.internal.utils.config.InstanceIdentifierSupplier;
117128
import org.eclipse.ditto.internal.utils.config.ScopedConfig;
118129
import org.eclipse.ditto.internal.utils.metrics.DittoMetrics;
130+
import org.eclipse.ditto.internal.utils.pekko.PingCommand;
131+
import org.eclipse.ditto.internal.utils.pekko.logging.CommonMdcEntryKey;
132+
import org.eclipse.ditto.internal.utils.pekko.logging.DittoDiagnosticLoggingAdapter;
133+
import org.eclipse.ditto.internal.utils.pekko.logging.DittoLoggerFactory;
119134
import org.eclipse.ditto.internal.utils.persistence.mongo.config.ActivityCheckConfig;
120135
import org.eclipse.ditto.internal.utils.persistence.mongo.config.SnapshotConfig;
121136
import org.eclipse.ditto.internal.utils.persistence.mongo.streaming.MongoReadJournal;
@@ -124,27 +139,12 @@
124139
import org.eclipse.ditto.internal.utils.persistentactors.commands.CommandStrategy;
125140
import org.eclipse.ditto.internal.utils.persistentactors.commands.DefaultContext;
126141
import org.eclipse.ditto.internal.utils.persistentactors.events.EventStrategy;
142+
import org.eclipse.ditto.internal.utils.tracing.span.StartedSpan;
127143
import org.eclipse.ditto.json.JsonValue;
128144
import org.eclipse.ditto.thingsearch.model.signals.commands.subscription.CreateSubscription;
129145

130146
import com.typesafe.config.Config;
131147

132-
import org.apache.pekko.actor.ActorRef;
133-
import org.apache.pekko.actor.ActorSystem;
134-
import org.apache.pekko.actor.Props;
135-
import org.apache.pekko.actor.Status;
136-
import org.apache.pekko.actor.SupervisorStrategy;
137-
import org.apache.pekko.cluster.Cluster;
138-
import org.apache.pekko.cluster.routing.ClusterRouterPool;
139-
import org.apache.pekko.cluster.routing.ClusterRouterPoolSettings;
140-
import org.apache.pekko.japi.pf.ReceiveBuilder;
141-
import org.apache.pekko.pattern.Patterns;
142-
import org.apache.pekko.persistence.RecoveryCompleted;
143-
import org.apache.pekko.routing.Broadcast;
144-
import org.apache.pekko.routing.ConsistentHashingPool;
145-
import org.apache.pekko.routing.ConsistentHashingRouter;
146-
import org.apache.pekko.routing.Pool;
147-
148148
/**
149149
* Handles {@code *Connection} commands and manages the persistence of connection. The actual connection handling to the
150150
* remote server is delegated to a child actor that uses a specific client (AMQP 1.0 or 0.9.1).
@@ -572,22 +572,29 @@ private void askSelfForRetrieveConnectionStatus(@Nullable final CharSequence cor
572572

573573
@Override
574574
public void onMutation(final Command<?> command, final ConnectivityEvent<?> event,
575-
final WithDittoHeaders response, final boolean becomeCreated, final boolean becomeDeleted) {
575+
final WithDittoHeaders response, final boolean becomeCreated, final boolean becomeDeleted,
576+
@Nullable final StartedSpan startedSpan) {
576577
if (command instanceof StagedCommand stagedCommand) {
577578
interpretStagedCommand(stagedCommand.withSenderUnlessDefined(getSender()));
579+
if (startedSpan != null) {
580+
startedSpan.finish();
581+
}
578582
} else {
579-
super.onMutation(command, event, response, becomeCreated, becomeDeleted);
583+
super.onMutation(command, event, response, becomeCreated, becomeDeleted, startedSpan);
580584
}
581585
}
582586

583587
@Override
584588
public void onStagedMutation(final Command<?> command, final CompletionStage<ConnectivityEvent<?>> event,
585589
final CompletionStage<WithDittoHeaders> response, final boolean becomeCreated,
586-
final boolean becomeDeleted) {
590+
final boolean becomeDeleted, @Nullable final StartedSpan startedSpan) {
587591
if (command instanceof StagedCommand stagedCommand) {
588592
interpretStagedCommand(stagedCommand.withSenderUnlessDefined(getSender()));
593+
if (startedSpan != null) {
594+
startedSpan.finish();
595+
}
589596
} else {
590-
super.onStagedMutation(command, event, response, becomeCreated, becomeDeleted);
597+
super.onStagedMutation(command, event, response, becomeCreated, becomeDeleted, startedSpan);
591598
}
592599
}
593600

‎edge/service/src/main/java/org/eclipse/ditto/edge/service/dispatching/ThingsAggregatorProxyActor.java

+35-40
Original file line numberDiff line numberDiff line change
@@ -27,19 +27,30 @@
2727

2828
import javax.annotation.Nullable;
2929

30+
import org.apache.pekko.NotUsed;
31+
import org.apache.pekko.actor.ActorRef;
32+
import org.apache.pekko.actor.Props;
33+
import org.apache.pekko.cluster.pubsub.DistributedPubSubMediator;
34+
import org.apache.pekko.japi.pf.PFBuilder;
35+
import org.apache.pekko.japi.pf.ReceiveBuilder;
36+
import org.apache.pekko.pattern.Patterns;
37+
import org.apache.pekko.stream.Materializer;
38+
import org.apache.pekko.stream.SourceRef;
39+
import org.apache.pekko.stream.javadsl.Sink;
40+
import org.apache.pekko.stream.javadsl.Source;
3041
import org.eclipse.ditto.base.model.entity.id.WithEntityId;
3142
import org.eclipse.ditto.base.model.exceptions.DittoInternalErrorException;
3243
import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
44+
import org.eclipse.ditto.base.model.headers.DittoHeaderDefinition;
3345
import org.eclipse.ditto.base.model.headers.DittoHeaders;
34-
import org.eclipse.ditto.base.model.headers.DittoHeadersSettable;
3546
import org.eclipse.ditto.base.model.json.Jsonifiable;
3647
import org.eclipse.ditto.base.model.signals.commands.Command;
3748
import org.eclipse.ditto.base.model.signals.commands.CommandResponse;
3849
import org.eclipse.ditto.base.model.signals.commands.WithEntity;
50+
import org.eclipse.ditto.internal.utils.cluster.DistPubSubAccess;
3951
import org.eclipse.ditto.internal.utils.pekko.actors.AbstractActorWithShutdownBehaviorAndRequestCounting;
4052
import org.eclipse.ditto.internal.utils.pekko.logging.DittoDiagnosticLoggingAdapter;
4153
import org.eclipse.ditto.internal.utils.pekko.logging.DittoLoggerFactory;
42-
import org.eclipse.ditto.internal.utils.cluster.DistPubSubAccess;
4354
import org.eclipse.ditto.internal.utils.tracing.DittoTracing;
4455
import org.eclipse.ditto.internal.utils.tracing.span.SpanOperationName;
4556
import org.eclipse.ditto.internal.utils.tracing.span.StartedSpan;
@@ -53,18 +64,6 @@
5364
import org.eclipse.ditto.things.model.signals.commands.query.RetrieveThings;
5465
import org.eclipse.ditto.things.model.signals.commands.query.RetrieveThingsResponse;
5566

56-
import org.apache.pekko.NotUsed;
57-
import org.apache.pekko.actor.ActorRef;
58-
import org.apache.pekko.actor.Props;
59-
import org.apache.pekko.cluster.pubsub.DistributedPubSubMediator;
60-
import org.apache.pekko.japi.pf.PFBuilder;
61-
import org.apache.pekko.japi.pf.ReceiveBuilder;
62-
import org.apache.pekko.pattern.Patterns;
63-
import org.apache.pekko.stream.Materializer;
64-
import org.apache.pekko.stream.SourceRef;
65-
import org.apache.pekko.stream.javadsl.Sink;
66-
import org.apache.pekko.stream.javadsl.Source;
67-
6867
/**
6968
* Acts as a client for {@code ThingsAggregatorActor} which responds
7069
* to a {@link RetrieveThings} command via a {@link SourceRef} which is a pointer in the cluster emitting the retrieved
@@ -108,8 +107,8 @@ public static Props props(final ActorRef pubSubMediator) {
108107
@Override
109108
public Receive handleMessage() {
110109
return ReceiveBuilder.create()
111-
.match(RetrieveThings.class, rt -> handleRetrieveThings(rt, rt))
112-
.match(SudoRetrieveThings.class, srt -> handleSudoRetrieveThings(srt, srt))
110+
.match(RetrieveThings.class, this::handleRetrieveThings)
111+
.match(SudoRetrieveThings.class, this::handleSudoRetrieveThings)
113112
.matchAny(m -> {
114113
log.warning("Got unknown message: {}", m);
115114
unhandled(m);
@@ -122,52 +121,47 @@ public void serviceUnbind(final Control serviceUnbind) {
122121
// nothing to do
123122
}
124123

125-
private void handleRetrieveThings(final RetrieveThings rt, final Object msgToAsk) {
124+
private void handleRetrieveThings(final RetrieveThings rt) {
126125
final List<ThingId> thingIds = rt.getEntityIds();
127126
log.withCorrelationId(rt)
128127
.info("Got '{}' message. Retrieving requested '{}' Things..",
129128
RetrieveThings.class.getSimpleName(), thingIds.size());
130-
131-
final ActorRef sender = getSender();
132-
askTargetActor(rt, thingIds, msgToAsk, sender);
129+
askTargetActor(rt, thingIds, getSender());
133130
}
134131

135-
private void handleSudoRetrieveThings(final SudoRetrieveThings srt, final Object msgToAsk) {
132+
private void handleSudoRetrieveThings(final SudoRetrieveThings srt) {
136133
final List<ThingId> thingIds = srt.getThingIds();
137134
log.withCorrelationId(srt)
138135
.info("Got '{}' message. Retrieving requested '{}' Things..",
139136
SudoRetrieveThings.class.getSimpleName(), thingIds.size());
140-
141-
final ActorRef sender = getSender();
142-
askTargetActor(srt, thingIds, msgToAsk, sender);
137+
askTargetActor(srt, thingIds, getSender());
143138
}
144139

145-
private void askTargetActor(final Command<?> command, final List<ThingId> thingIds,
146-
final Object msgToAsk, final ActorRef sender) {
147-
148-
final Object tracedMsgToAsk;
140+
private void askTargetActor(final Command<?> command, final List<ThingId> thingIds, final ActorRef sender)
141+
{
142+
final DittoHeaders dittoHeaders = command.getDittoHeaders();
149143
final var startedSpan = DittoTracing.newPreparedSpan(
150-
command.getDittoHeaders(),
144+
dittoHeaders,
151145
TRACE_AGGREGATOR_RETRIEVE_THINGS
152146
)
153147
.tag("size", Integer.toString(thingIds.size()))
154148
.start();
155-
if (msgToAsk instanceof DittoHeadersSettable<?> dittoHeadersSettable) {
156-
tracedMsgToAsk = dittoHeadersSettable.setDittoHeaders(
157-
DittoHeaders.of(startedSpan.propagateContext(dittoHeadersSettable.getDittoHeaders()))
158-
);
159-
} else {
160-
tracedMsgToAsk = msgToAsk;
161-
}
149+
final Command<?> tracedCommand = command.setDittoHeaders(
150+
DittoHeaders.of(startedSpan.propagateContext(
151+
dittoHeaders.toBuilder()
152+
.removeHeader(DittoHeaderDefinition.W3C_TRACEPARENT.getKey())
153+
.build()
154+
))
155+
);
162156

163157
final DistributedPubSubMediator.Publish pubSubMsg =
164-
DistPubSubAccess.publishViaGroup(command.getType(), tracedMsgToAsk);
158+
DistPubSubAccess.publishViaGroup(tracedCommand.getType(), tracedCommand);
165159

166160
withRequestCounting(
167161
Patterns.ask(pubSubMediator, pubSubMsg, Duration.ofSeconds(ASK_TIMEOUT))
168162
.thenAccept(response -> {
169163
if (response instanceof SourceRef) {
170-
handleSourceRef((SourceRef<?>) response, thingIds, command, sender, startedSpan);
164+
handleSourceRef((SourceRef<?>) response, thingIds, tracedCommand, sender, startedSpan);
171165
} else if (response instanceof DittoRuntimeException dre) {
172166
startedSpan.tagAsFailed(dre).finish();
173167
sender.tell(response, getSelf());
@@ -177,7 +171,7 @@ private void askTargetActor(final Command<?> command, final List<ThingId> thingI
177171
response.getClass().getSimpleName(), response);
178172
final DittoInternalErrorException responseEx =
179173
DittoInternalErrorException.newBuilder()
180-
.dittoHeaders(command.getDittoHeaders())
174+
.dittoHeaders(tracedCommand.getDittoHeaders())
181175
.build();
182176
startedSpan.tagAsFailed(responseEx).finish();
183177
sender.tell(responseEx, getSelf());
@@ -187,7 +181,8 @@ private void askTargetActor(final Command<?> command, final List<ThingId> thingI
187181
}
188182

189183
private void handleSourceRef(final SourceRef<?> sourceRef, final List<ThingId> thingIds,
190-
final Command<?> originatingCommand, final ActorRef originatingSender, final StartedSpan startedSpan) {
184+
final Command<?> originatingCommand, final ActorRef originatingSender, final StartedSpan startedSpan)
185+
{
191186
final Function<Jsonifiable<?>, PlainJson> thingPlainJsonSupplier;
192187
final Function<List<PlainJson>, CommandResponse<?>> overallResponseSupplier;
193188
final UnaryOperator<List<PlainJson>> plainJsonSorter = supplyPlainJsonSorter(thingIds);

‎gateway/service/src/main/java/org/eclipse/ditto/gateway/service/endpoints/directives/RequestTracingDirective.java

+15-13
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,13 @@
3030
import javax.annotation.Nullable;
3131
import javax.annotation.concurrent.Immutable;
3232

33+
import org.apache.pekko.http.javadsl.model.HttpHeader;
34+
import org.apache.pekko.http.javadsl.model.HttpRequest;
35+
import org.apache.pekko.http.javadsl.model.HttpResponse;
36+
import org.apache.pekko.http.javadsl.model.Uri;
37+
import org.apache.pekko.http.javadsl.server.Complete;
38+
import org.apache.pekko.http.javadsl.server.Route;
39+
import org.apache.pekko.http.javadsl.server.RouteResult;
3340
import org.eclipse.ditto.base.model.common.HttpStatus;
3441
import org.eclipse.ditto.base.model.common.HttpStatusCodeOutOfRangeException;
3542
import org.eclipse.ditto.base.model.headers.DittoHeaderDefinition;
@@ -43,14 +50,6 @@
4350
import org.eclipse.ditto.internal.utils.tracing.span.SpanTagKey;
4451
import org.eclipse.ditto.internal.utils.tracing.span.StartedSpan;
4552

46-
import org.apache.pekko.http.javadsl.model.HttpHeader;
47-
import org.apache.pekko.http.javadsl.model.HttpRequest;
48-
import org.apache.pekko.http.javadsl.model.HttpResponse;
49-
import org.apache.pekko.http.javadsl.model.Uri;
50-
import org.apache.pekko.http.javadsl.server.Complete;
51-
import org.apache.pekko.http.javadsl.server.Route;
52-
import org.apache.pekko.http.javadsl.server.RouteResult;
53-
5453
/**
5554
* Custom Pekko Http directive tracing the request.
5655
*/
@@ -116,7 +115,7 @@ private static SpanOperationName resolveSpanOperationName(final HttpRequest http
116115

117116
private static URI getTraceUri(final HttpRequest httpRequest) {
118117
final var traceInformationGenerator = TraceInformationGenerator.getInstance();
119-
final var traceInformation = traceInformationGenerator.apply(String.valueOf(getRelativeUri(httpRequest)));
118+
final var traceInformation = traceInformationGenerator.apply(getRelativeUriPath(httpRequest));
120119
return traceInformation.getTraceUri();
121120
}
122121

@@ -125,6 +124,12 @@ private static Uri getRelativeUri(final HttpRequest httpRequest) {
125124
return uri.toRelative();
126125
}
127126

127+
private static String getRelativeUriPath(final HttpRequest httpRequest) {
128+
final var uri = httpRequest.getUri();
129+
final var relativeUri = uri.toRelative();
130+
return relativeUri.path();
131+
}
132+
128133
private static String getRequestMethodName(final HttpRequest httpRequest) {
129134
final var httpMethod = httpRequest.method();
130135
return httpMethod.name();
@@ -246,10 +251,7 @@ private static void addRequestResponseTags(
246251
@Nullable final CharSequence correlationId
247252
) {
248253
startedSpan.tag(SpanTagKey.REQUEST_METHOD_NAME.getTagForValue(getRequestMethodName(httpRequest)));
249-
@Nullable final var relativeRequestUri = tryToGetRelativeRequestUri(httpRequest, correlationId);
250-
if (null != relativeRequestUri) {
251-
startedSpan.tag(SpanTagKey.REQUEST_URI.getTagForValue(relativeRequestUri));
252-
}
254+
startedSpan.tag(SpanTagKey.REQUEST_URI.getTagForValue(URI.create(getRelativeUri(httpRequest).toString())));
253255
@Nullable final var httpStatus = tryToGetResponseHttpStatus(httpResponse, correlationId);
254256
if (null != httpStatus) {
255257
startedSpan.tag(SpanTagKey.HTTP_STATUS.getTagForValue(httpStatus));

‎gateway/service/src/main/java/org/eclipse/ditto/gateway/service/endpoints/directives/auth/DittoGatewayAuthenticationDirectiveFactory.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
import java.util.Collection;
1717
import java.util.concurrent.Executor;
1818

19+
import javax.annotation.Nullable;
20+
1921
import org.apache.pekko.actor.ActorSystem;
2022
import org.eclipse.ditto.gateway.service.security.authentication.AuthenticationChain;
2123
import org.eclipse.ditto.gateway.service.security.authentication.AuthenticationFailureAggregator;
@@ -32,7 +34,6 @@
3234
import org.slf4j.Logger;
3335
import org.slf4j.LoggerFactory;
3436

35-
import com.mongodb.lang.Nullable;
3637
import com.typesafe.config.Config;
3738

3839
/**

‎internal/models/signalenrichment/pom.xml

+4
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,10 @@
4646
<groupId>org.eclipse.ditto</groupId>
4747
<artifactId>ditto-internal-models-signal</artifactId>
4848
</dependency>
49+
<dependency>
50+
<groupId>org.eclipse.ditto</groupId>
51+
<artifactId>ditto-internal-utils-tracing</artifactId>
52+
</dependency>
4953

5054
<!-- test-only -->
5155
<dependency>

0 commit comments

Comments
 (0)