You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2026/05/07 11:22:02 UTC
(camel) branch main updated: CAMEL-23387: camel-telemetry - Add span decorators for AWS (#23038)
This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 54cd1a091bfd CAMEL-23387: camel-telemetry - Add span decorators for AWS (#23038)
54cd1a091bfd is described below
commit 54cd1a091bfd0b934fc320201342349d77a01ad7
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu May 7 13:21:56 2026 +0200
CAMEL-23387: camel-telemetry - Add span decorators for AWS (#23038)
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---
.../decorators/AwsKinesisSpanDecorator.java | 72 +++++++++++++++++
.../telemetry/decorators/AwsS3SpanDecorator.java | 80 +++++++++++++++++++
.../telemetry/decorators/AwsSnsSpanDecorator.java | 79 ++++++++++++++++++
.../telemetry/decorators/AwsSqsSpanDecorator.java | 93 ++++++++++++++++++++++
.../org.apache.camel.telemetry.SpanDecorator | 4 +
.../decorators/AwsKinesisSpanDecoratorTest.java | 72 +++++++++++++++++
.../decorators/AwsS3SpanDecoratorTest.java | 64 +++++++++++++++
.../decorators/AwsSnsSpanDecoratorTest.java | 74 +++++++++++++++++
.../decorators/AwsSqsSpanDecoratorTest.java | 80 +++++++++++++++++++
9 files changed, 618 insertions(+)
diff --git a/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsKinesisSpanDecorator.java b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsKinesisSpanDecorator.java
new file mode 100644
index 000000000000..0de906988e5e
--- /dev/null
+++ b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsKinesisSpanDecorator.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.telemetry.Span;
+
+public class AwsKinesisSpanDecorator extends AbstractMessagingSpanDecorator {
+
+ static final String KINESIS_PARTITION_KEY = "partitionKey";
+ static final String KINESIS_SHARD_ID = "shardId";
+ static final String KINESIS_APPROX_ARRIVAL_TIME = "approximateArrivalTimestamp";
+
+ /**
+ * Constants copied from {@link org.apache.camel.component.aws2.kinesis.Kinesis2Constants}
+ */
+ static final String SEQUENCE_NUMBER = "CamelAwsKinesisSequenceNumber";
+ static final String APPROX_ARRIVAL_TIME = "CamelAwsKinesisApproximateArrivalTimestamp";
+ static final String PARTITION_KEY = "CamelAwsKinesisPartitionKey";
+ static final String SHARD_ID = "CamelAwsKinesisShardId";
+
+ @Override
+ public String getComponent() {
+ return "aws2-kinesis";
+ }
+
+ @Override
+ public String getComponentClassName() {
+ return "org.apache.camel.component.aws2.kinesis.Kinesis2Component";
+ }
+
+ @Override
+ public void beforeTracingEvent(Span span, Exchange exchange, Endpoint endpoint) {
+ super.beforeTracingEvent(span, exchange, endpoint);
+
+ String partitionKey = exchange.getIn().getHeader(PARTITION_KEY, String.class);
+ if (partitionKey != null) {
+ span.setTag(KINESIS_PARTITION_KEY, partitionKey);
+ }
+
+ String shardId = exchange.getIn().getHeader(SHARD_ID, String.class);
+ if (shardId != null) {
+ span.setTag(KINESIS_SHARD_ID, shardId);
+ }
+
+ String approxArrivalTimestamp = exchange.getIn().getHeader(APPROX_ARRIVAL_TIME, String.class);
+ if (approxArrivalTimestamp != null) {
+ span.setTag(KINESIS_APPROX_ARRIVAL_TIME, approxArrivalTimestamp);
+ }
+ }
+
+ @Override
+ protected String getMessageId(Exchange exchange) {
+ return exchange.getIn().getHeader(SEQUENCE_NUMBER, String.class);
+ }
+
+}
diff --git a/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsS3SpanDecorator.java b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsS3SpanDecorator.java
new file mode 100644
index 000000000000..cff4ee500a4d
--- /dev/null
+++ b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsS3SpanDecorator.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.telemetry.Span;
+
+public class AwsS3SpanDecorator extends AbstractSpanDecorator {
+
+ static final String S3_BUCKET_NAME = "bucketName";
+ static final String S3_KEY = "key";
+ static final String S3_VERSION_ID = "versionId";
+ static final String S3_OPERATION = "operation";
+ static final String S3_STORAGE_CLASS = "storageClass";
+
+ /**
+ * Constants copied from {@link org.apache.camel.component.aws2.s3.AWS2S3Constants}
+ */
+ static final String BUCKET_NAME = "CamelAwsS3BucketName";
+ static final String KEY = "CamelAwsS3Key";
+ static final String VERSION_ID = "CamelAwsS3VersionId";
+ static final String OPERATION = "CamelAwsS3Operation";
+ static final String STORAGE_CLASS = "CamelAwsS3StorageClass";
+
+ @Override
+ public String getComponent() {
+ return "aws2-s3";
+ }
+
+ @Override
+ public String getComponentClassName() {
+ return "org.apache.camel.component.aws2.s3.AWS2S3Component";
+ }
+
+ @Override
+ public void beforeTracingEvent(Span span, Exchange exchange, Endpoint endpoint) {
+ super.beforeTracingEvent(span, exchange, endpoint);
+
+ String bucketName = exchange.getIn().getHeader(BUCKET_NAME, String.class);
+ if (bucketName != null) {
+ span.setTag(S3_BUCKET_NAME, bucketName);
+ }
+
+ String key = exchange.getIn().getHeader(KEY, String.class);
+ if (key != null) {
+ span.setTag(S3_KEY, key);
+ }
+
+ String versionId = exchange.getIn().getHeader(VERSION_ID, String.class);
+ if (versionId != null) {
+ span.setTag(S3_VERSION_ID, versionId);
+ }
+
+ String operation = exchange.getIn().getHeader(OPERATION, String.class);
+ if (operation != null) {
+ span.setTag(S3_OPERATION, operation);
+ }
+
+ String storageClass = exchange.getIn().getHeader(STORAGE_CLASS, String.class);
+ if (storageClass != null) {
+ span.setTag(S3_STORAGE_CLASS, storageClass);
+ }
+ }
+
+}
diff --git a/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsSnsSpanDecorator.java b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsSnsSpanDecorator.java
new file mode 100644
index 000000000000..cd54483335a5
--- /dev/null
+++ b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsSnsSpanDecorator.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.telemetry.Span;
+
+public class AwsSnsSpanDecorator extends AbstractMessagingSpanDecorator {
+
+ static final String SNS_SUBJECT = "subject";
+ static final String SNS_MESSAGE_STRUCTURE = "messageStructure";
+ static final String SNS_MESSAGE_GROUP_ID = "messageGroupId";
+ static final String SNS_SEQUENCE_NUMBER = "sequenceNumber";
+
+ /**
+ * Constants copied from {@link org.apache.camel.component.aws2.sns.Sns2Constants}
+ */
+ static final String MESSAGE_ID = "CamelAwsSnsMessageId";
+ static final String SUBJECT = "CamelAwsSnsSubject";
+ static final String MESSAGE_STRUCTURE = "CamelAwsSnsMessageStructure";
+ static final String MESSAGE_GROUP_ID = "CamelAwsSnsMessageGroupId";
+ static final String SEQUENCE_NUMBER = "CamelAwsSnsSequenceNumber";
+
+ @Override
+ public String getComponent() {
+ return "aws2-sns";
+ }
+
+ @Override
+ public String getComponentClassName() {
+ return "org.apache.camel.component.aws2.sns.Sns2Component";
+ }
+
+ @Override
+ public void beforeTracingEvent(Span span, Exchange exchange, Endpoint endpoint) {
+ super.beforeTracingEvent(span, exchange, endpoint);
+
+ String subject = exchange.getIn().getHeader(SUBJECT, String.class);
+ if (subject != null) {
+ span.setTag(SNS_SUBJECT, subject);
+ }
+
+ String messageStructure = exchange.getIn().getHeader(MESSAGE_STRUCTURE, String.class);
+ if (messageStructure != null) {
+ span.setTag(SNS_MESSAGE_STRUCTURE, messageStructure);
+ }
+
+ String messageGroupId = exchange.getIn().getHeader(MESSAGE_GROUP_ID, String.class);
+ if (messageGroupId != null) {
+ span.setTag(SNS_MESSAGE_GROUP_ID, messageGroupId);
+ }
+
+ String sequenceNumber = exchange.getIn().getHeader(SEQUENCE_NUMBER, String.class);
+ if (sequenceNumber != null) {
+ span.setTag(SNS_SEQUENCE_NUMBER, sequenceNumber);
+ }
+ }
+
+ @Override
+ protected String getMessageId(Exchange exchange) {
+ return exchange.getIn().getHeader(MESSAGE_ID, String.class);
+ }
+
+}
diff --git a/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsSqsSpanDecorator.java b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsSqsSpanDecorator.java
new file mode 100644
index 000000000000..75f6c70cbbea
--- /dev/null
+++ b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/AwsSqsSpanDecorator.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.telemetry.Span;
+
+public class AwsSqsSpanDecorator extends AbstractMessagingSpanDecorator {
+
+ static final String SQS_MD5_OF_BODY = "md5OfBody";
+ static final String SQS_RECEIPT_HANDLE = "receiptHandle";
+ static final String SQS_DELAY_SECONDS = "delaySeconds";
+ static final String SQS_OPERATION = "operation";
+ static final String SQS_MESSAGE_GROUP_ID = "messageGroupId";
+ static final String SQS_SEQUENCE_NUMBER = "sequenceNumber";
+
+ /**
+ * Constants copied from {@link org.apache.camel.component.aws2.sqs.Sqs2Constants}
+ */
+ static final String MESSAGE_ID = "CamelAwsSqsMessageId";
+ static final String MD5_OF_BODY = "CamelAwsSqsMD5OfBody";
+ static final String RECEIPT_HANDLE = "CamelAwsSqsReceiptHandle";
+ static final String DELAY_HEADER = "CamelAwsSqsDelaySeconds";
+ static final String OPERATION = "CamelAwsSqsOperation";
+ static final String MESSAGE_GROUP_ID = "CamelAwsMessageGroupId";
+ static final String SEQUENCE_NUMBER = "CamelAwsSqsSequenceNumber";
+
+ @Override
+ public String getComponent() {
+ return "aws2-sqs";
+ }
+
+ @Override
+ public String getComponentClassName() {
+ return "org.apache.camel.component.aws2.sqs.Sqs2Component";
+ }
+
+ @Override
+ public void beforeTracingEvent(Span span, Exchange exchange, Endpoint endpoint) {
+ super.beforeTracingEvent(span, exchange, endpoint);
+
+ String md5OfBody = exchange.getIn().getHeader(MD5_OF_BODY, String.class);
+ if (md5OfBody != null) {
+ span.setTag(SQS_MD5_OF_BODY, md5OfBody);
+ }
+
+ String receiptHandle = exchange.getIn().getHeader(RECEIPT_HANDLE, String.class);
+ if (receiptHandle != null) {
+ span.setTag(SQS_RECEIPT_HANDLE, receiptHandle);
+ }
+
+ Integer delaySeconds = exchange.getIn().getHeader(DELAY_HEADER, Integer.class);
+ if (delaySeconds != null) {
+ span.setTag(SQS_DELAY_SECONDS, delaySeconds.toString());
+ }
+
+ String operation = exchange.getIn().getHeader(OPERATION, String.class);
+ if (operation != null) {
+ span.setTag(SQS_OPERATION, operation);
+ }
+
+ String messageGroupId = exchange.getIn().getHeader(MESSAGE_GROUP_ID, String.class);
+ if (messageGroupId != null) {
+ span.setTag(SQS_MESSAGE_GROUP_ID, messageGroupId);
+ }
+
+ String sequenceNumber = exchange.getIn().getHeader(SEQUENCE_NUMBER, String.class);
+ if (sequenceNumber != null) {
+ span.setTag(SQS_SEQUENCE_NUMBER, sequenceNumber);
+ }
+ }
+
+ @Override
+ protected String getMessageId(Exchange exchange) {
+ return exchange.getIn().getHeader(MESSAGE_ID, String.class);
+ }
+
+}
diff --git a/components/camel-telemetry/src/main/resources/META-INF/services/org.apache.camel.telemetry.SpanDecorator b/components/camel-telemetry/src/main/resources/META-INF/services/org.apache.camel.telemetry.SpanDecorator
index e3eb8a1bbef0..1c7f8c2e793e 100644
--- a/components/camel-telemetry/src/main/resources/META-INF/services/org.apache.camel.telemetry.SpanDecorator
+++ b/components/camel-telemetry/src/main/resources/META-INF/services/org.apache.camel.telemetry.SpanDecorator
@@ -19,6 +19,10 @@ org.apache.camel.telemetry.decorators.ActiveMQ6SpanDecorator
org.apache.camel.telemetry.decorators.ActiveMQSpanDecorator
org.apache.camel.telemetry.decorators.AhcSpanDecorator
org.apache.camel.telemetry.decorators.AmqpSpanDecorator
+org.apache.camel.telemetry.decorators.AwsKinesisSpanDecorator
+org.apache.camel.telemetry.decorators.AwsS3SpanDecorator
+org.apache.camel.telemetry.decorators.AwsSnsSpanDecorator
+org.apache.camel.telemetry.decorators.AwsSqsSpanDecorator
org.apache.camel.telemetry.decorators.AzureServiceBusSpanDecorator
org.apache.camel.telemetry.decorators.CometdSpanDecorator
org.apache.camel.telemetry.decorators.CometdsSpanDecorator
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsKinesisSpanDecoratorTest.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsKinesisSpanDecoratorTest.java
new file mode 100644
index 000000000000..34c61f1e707a
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsKinesisSpanDecoratorTest.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.telemetry.mock.MockSpanAdapter;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AwsKinesisSpanDecoratorTest {
+
+ @Test
+ public void testGetMessageId() {
+ String sequenceNumber = "49545115243490985018280067714973144582180062593244200961";
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(message.getHeader(AwsKinesisSpanDecorator.SEQUENCE_NUMBER, String.class)).thenReturn(sequenceNumber);
+
+ AbstractMessagingSpanDecorator decorator = new AwsKinesisSpanDecorator();
+
+ assertEquals(sequenceNumber, decorator.getMessageId(exchange));
+ }
+
+ @Test
+ public void testPre() {
+ String partitionKey = "partition-1";
+ String shardId = "shardId-000000000000";
+ String approxArrivalTimestamp = "2026-04-30T12:00:00Z";
+
+ Endpoint endpoint = Mockito.mock(Endpoint.class);
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(endpoint.getEndpointUri()).thenReturn("aws2-kinesis:myStream");
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(message.getHeader(AwsKinesisSpanDecorator.PARTITION_KEY, String.class)).thenReturn(partitionKey);
+ Mockito.when(message.getHeader(AwsKinesisSpanDecorator.SHARD_ID, String.class)).thenReturn(shardId);
+ Mockito.when(message.getHeader(AwsKinesisSpanDecorator.APPROX_ARRIVAL_TIME, String.class))
+ .thenReturn(approxArrivalTimestamp);
+
+ AbstractMessagingSpanDecorator decorator = new AwsKinesisSpanDecorator();
+
+ MockSpanAdapter span = new MockSpanAdapter();
+
+ decorator.beforeTracingEvent(span, exchange, endpoint);
+
+ assertEquals(partitionKey, span.tags().get(AwsKinesisSpanDecorator.KINESIS_PARTITION_KEY));
+ assertEquals(shardId, span.tags().get(AwsKinesisSpanDecorator.KINESIS_SHARD_ID));
+ assertEquals(approxArrivalTimestamp, span.tags().get(AwsKinesisSpanDecorator.KINESIS_APPROX_ARRIVAL_TIME));
+ }
+
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsS3SpanDecoratorTest.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsS3SpanDecoratorTest.java
new file mode 100644
index 000000000000..30144f6c6f6c
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsS3SpanDecoratorTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.telemetry.mock.MockSpanAdapter;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AwsS3SpanDecoratorTest {
+
+ @Test
+ public void testPre() {
+ String bucketName = "my-bucket";
+ String key = "my/object/key.txt";
+ String versionId = "version-1234";
+ String operation = "putObject";
+ String storageClass = "STANDARD";
+
+ Endpoint endpoint = Mockito.mock(Endpoint.class);
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(endpoint.getEndpointUri()).thenReturn("aws2-s3:myBucket");
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(exchange.getExchangeId()).thenReturn("exchange-1");
+ Mockito.when(message.getHeader(AwsS3SpanDecorator.BUCKET_NAME, String.class)).thenReturn(bucketName);
+ Mockito.when(message.getHeader(AwsS3SpanDecorator.KEY, String.class)).thenReturn(key);
+ Mockito.when(message.getHeader(AwsS3SpanDecorator.VERSION_ID, String.class)).thenReturn(versionId);
+ Mockito.when(message.getHeader(AwsS3SpanDecorator.OPERATION, String.class)).thenReturn(operation);
+ Mockito.when(message.getHeader(AwsS3SpanDecorator.STORAGE_CLASS, String.class)).thenReturn(storageClass);
+
+ AbstractSpanDecorator decorator = new AwsS3SpanDecorator();
+
+ MockSpanAdapter span = new MockSpanAdapter();
+
+ decorator.beforeTracingEvent(span, exchange, endpoint);
+
+ assertEquals(bucketName, span.tags().get(AwsS3SpanDecorator.S3_BUCKET_NAME));
+ assertEquals(key, span.tags().get(AwsS3SpanDecorator.S3_KEY));
+ assertEquals(versionId, span.tags().get(AwsS3SpanDecorator.S3_VERSION_ID));
+ assertEquals(operation, span.tags().get(AwsS3SpanDecorator.S3_OPERATION));
+ assertEquals(storageClass, span.tags().get(AwsS3SpanDecorator.S3_STORAGE_CLASS));
+ }
+
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsSnsSpanDecoratorTest.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsSnsSpanDecoratorTest.java
new file mode 100644
index 000000000000..3a9ebb255360
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsSnsSpanDecoratorTest.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.telemetry.mock.MockSpanAdapter;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AwsSnsSpanDecoratorTest {
+
+ @Test
+ public void testGetMessageId() {
+ String messageId = "abcd";
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(message.getHeader(AwsSnsSpanDecorator.MESSAGE_ID, String.class)).thenReturn(messageId);
+
+ AbstractMessagingSpanDecorator decorator = new AwsSnsSpanDecorator();
+
+ assertEquals(messageId, decorator.getMessageId(exchange));
+ }
+
+ @Test
+ public void testPre() {
+ String subject = "my-subject";
+ String messageStructure = "json";
+ String messageGroupId = "myGroup";
+ String sequenceNumber = "1234567890";
+
+ Endpoint endpoint = Mockito.mock(Endpoint.class);
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(endpoint.getEndpointUri()).thenReturn("aws2-sns:myTopic");
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(message.getHeader(AwsSnsSpanDecorator.SUBJECT, String.class)).thenReturn(subject);
+ Mockito.when(message.getHeader(AwsSnsSpanDecorator.MESSAGE_STRUCTURE, String.class)).thenReturn(messageStructure);
+ Mockito.when(message.getHeader(AwsSnsSpanDecorator.MESSAGE_GROUP_ID, String.class)).thenReturn(messageGroupId);
+ Mockito.when(message.getHeader(AwsSnsSpanDecorator.SEQUENCE_NUMBER, String.class)).thenReturn(sequenceNumber);
+
+ AbstractMessagingSpanDecorator decorator = new AwsSnsSpanDecorator();
+
+ MockSpanAdapter span = new MockSpanAdapter();
+
+ decorator.beforeTracingEvent(span, exchange, endpoint);
+
+ assertEquals(subject, span.tags().get(AwsSnsSpanDecorator.SNS_SUBJECT));
+ assertEquals(messageStructure, span.tags().get(AwsSnsSpanDecorator.SNS_MESSAGE_STRUCTURE));
+ assertEquals(messageGroupId, span.tags().get(AwsSnsSpanDecorator.SNS_MESSAGE_GROUP_ID));
+ assertEquals(sequenceNumber, span.tags().get(AwsSnsSpanDecorator.SNS_SEQUENCE_NUMBER));
+ }
+
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsSqsSpanDecoratorTest.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsSqsSpanDecoratorTest.java
new file mode 100644
index 000000000000..644f5a447fd2
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/decorators/AwsSqsSpanDecoratorTest.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.telemetry.decorators;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.telemetry.mock.MockSpanAdapter;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AwsSqsSpanDecoratorTest {
+
+ @Test
+ public void testGetMessageId() {
+ String messageId = "abcd";
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.MESSAGE_ID, String.class)).thenReturn(messageId);
+
+ AbstractMessagingSpanDecorator decorator = new AwsSqsSpanDecorator();
+
+ assertEquals(messageId, decorator.getMessageId(exchange));
+ }
+
+ @Test
+ public void testPre() {
+ String md5OfBody = "098f6bcd4621d373cade4e832627b4f6";
+ String receiptHandle = "MbZj6wDWli+JvwwJaBV+3dcjk2YW2vA3";
+ Integer delaySeconds = 30;
+ String operation = "sendMessage";
+ String messageGroupId = "myGroup";
+ String sequenceNumber = "1234567890";
+
+ Endpoint endpoint = Mockito.mock(Endpoint.class);
+ Exchange exchange = Mockito.mock(Exchange.class);
+ Message message = Mockito.mock(Message.class);
+
+ Mockito.when(endpoint.getEndpointUri()).thenReturn("aws2-sqs:myQueue");
+ Mockito.when(exchange.getIn()).thenReturn(message);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.MD5_OF_BODY, String.class)).thenReturn(md5OfBody);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.RECEIPT_HANDLE, String.class)).thenReturn(receiptHandle);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.DELAY_HEADER, Integer.class)).thenReturn(delaySeconds);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.OPERATION, String.class)).thenReturn(operation);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.MESSAGE_GROUP_ID, String.class)).thenReturn(messageGroupId);
+ Mockito.when(message.getHeader(AwsSqsSpanDecorator.SEQUENCE_NUMBER, String.class)).thenReturn(sequenceNumber);
+
+ AbstractMessagingSpanDecorator decorator = new AwsSqsSpanDecorator();
+
+ MockSpanAdapter span = new MockSpanAdapter();
+
+ decorator.beforeTracingEvent(span, exchange, endpoint);
+
+ assertEquals(md5OfBody, span.tags().get(AwsSqsSpanDecorator.SQS_MD5_OF_BODY));
+ assertEquals(receiptHandle, span.tags().get(AwsSqsSpanDecorator.SQS_RECEIPT_HANDLE));
+ assertEquals(delaySeconds.toString(), span.tags().get(AwsSqsSpanDecorator.SQS_DELAY_SECONDS));
+ assertEquals(operation, span.tags().get(AwsSqsSpanDecorator.SQS_OPERATION));
+ assertEquals(messageGroupId, span.tags().get(AwsSqsSpanDecorator.SQS_MESSAGE_GROUP_ID));
+ assertEquals(sequenceNumber, span.tags().get(AwsSqsSpanDecorator.SQS_SEQUENCE_NUMBER));
+ }
+
+}