-
Notifications
You must be signed in to change notification settings - Fork 867
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Pulsar batch receive instrumentation (#8173)
- Loading branch information
Showing
13 changed files
with
440 additions
and
67 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
27 changes: 27 additions & 0 deletions
27
...a/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/BasePulsarRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry; | ||
|
|
||
| import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.UrlData; | ||
|
|
||
| public class BasePulsarRequest { | ||
|
|
||
| private final String destination; | ||
| private final UrlData urlData; | ||
|
|
||
| protected BasePulsarRequest(String destination, UrlData urlData) { | ||
| this.destination = destination; | ||
| this.urlData = urlData; | ||
| } | ||
|
|
||
| public String getDestination() { | ||
| return destination; | ||
| } | ||
|
|
||
| public UrlData getUrlData() { | ||
| return urlData; | ||
| } | ||
| } |
75 changes: 75 additions & 0 deletions
75
...javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarBatchMessagingAttributesGetter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry; | ||
|
|
||
| import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesGetter; | ||
| import io.opentelemetry.semconv.trace.attributes.SemanticAttributes; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.StreamSupport; | ||
| import javax.annotation.Nullable; | ||
|
|
||
| enum PulsarBatchMessagingAttributesGetter | ||
| implements MessagingAttributesGetter<PulsarBatchRequest, Void> { | ||
| INSTANCE; | ||
|
|
||
| @Override | ||
| public String getSystem(PulsarBatchRequest request) { | ||
| return "pulsar"; | ||
| } | ||
|
|
||
| @Override | ||
| public String getDestinationKind(PulsarBatchRequest request) { | ||
| return SemanticAttributes.MessagingDestinationKindValues.TOPIC; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public String getDestination(PulsarBatchRequest request) { | ||
| return request.getDestination(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isTemporaryDestination(PulsarBatchRequest request) { | ||
| return false; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public String getConversationId(PulsarBatchRequest message) { | ||
| return null; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public Long getMessagePayloadSize(PulsarBatchRequest request) { | ||
| return StreamSupport.stream(request.getMessages().spliterator(), false) | ||
| .map(message -> (long) message.size()) | ||
| .reduce(Long::sum) | ||
| .orElse(null); | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public Long getMessagePayloadCompressedSize(PulsarBatchRequest request) { | ||
| return null; | ||
| } | ||
|
|
||
| @Nullable | ||
| @Override | ||
| public String getMessageId(PulsarBatchRequest request, @Nullable Void response) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public List<String> getMessageHeader(PulsarBatchRequest request, String name) { | ||
| return StreamSupport.stream(request.getMessages().spliterator(), false) | ||
| .map(message -> message.getProperty(name)) | ||
| .filter(Objects::nonNull) | ||
| .collect(Collectors.toList()); | ||
| } | ||
| } |
42 changes: 42 additions & 0 deletions
42
.../io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarBatchRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry; | ||
|
|
||
| import static io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.parseUrl; | ||
|
|
||
| import io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.UrlParser.UrlData; | ||
| import org.apache.pulsar.client.api.Message; | ||
| import org.apache.pulsar.client.api.Messages; | ||
|
|
||
| public final class PulsarBatchRequest extends BasePulsarRequest { | ||
| private final Messages<?> messages; | ||
|
|
||
| private PulsarBatchRequest(Messages<?> messages, String destination, UrlData urlData) { | ||
| super(destination, urlData); | ||
| this.messages = messages; | ||
| } | ||
|
|
||
| public static PulsarBatchRequest create(Messages<?> messages, String url) { | ||
| return new PulsarBatchRequest(messages, getTopicName(messages), parseUrl(url)); | ||
| } | ||
|
|
||
| private static String getTopicName(Messages<?> messages) { | ||
| String topicName = null; | ||
| for (Message<?> message : messages) { | ||
| String name = message.getTopicName(); | ||
| if (topicName == null) { | ||
| topicName = name; | ||
| } else if (!topicName.equals(name)) { | ||
| return null; | ||
| } | ||
| } | ||
| return topicName; | ||
| } | ||
|
|
||
| public Messages<?> getMessages() { | ||
| return messages; | ||
| } | ||
| } |
32 changes: 32 additions & 0 deletions
32
...javaagent/instrumentation/pulsar/v2_8/telemetry/PulsarBatchRequestSpanLinksExtractor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.pulsar.v2_8.telemetry; | ||
|
|
||
| import io.opentelemetry.context.Context; | ||
| import io.opentelemetry.context.propagation.TextMapPropagator; | ||
| import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksBuilder; | ||
| import io.opentelemetry.instrumentation.api.instrumenter.SpanLinksExtractor; | ||
| import io.opentelemetry.instrumentation.api.internal.PropagatorBasedSpanLinksExtractor; | ||
| import org.apache.pulsar.client.api.Message; | ||
|
|
||
| final class PulsarBatchRequestSpanLinksExtractor implements SpanLinksExtractor<PulsarBatchRequest> { | ||
| private final SpanLinksExtractor<PulsarRequest> singleRecordLinkExtractor; | ||
|
|
||
| PulsarBatchRequestSpanLinksExtractor(TextMapPropagator propagator) { | ||
| this.singleRecordLinkExtractor = | ||
| new PropagatorBasedSpanLinksExtractor<>(propagator, MessageTextMapGetter.INSTANCE); | ||
| } | ||
|
|
||
| @Override | ||
| public void extract( | ||
| SpanLinksBuilder spanLinks, Context parentContext, PulsarBatchRequest request) { | ||
|
|
||
| for (Message<?> message : request.getMessages()) { | ||
| singleRecordLinkExtractor.extract( | ||
| spanLinks, Context.root(), PulsarRequest.create(message, request.getUrlData())); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.