Add trace filter support, custom headers and improve opentelemetry

modules/core/src/main/java/org/apache/synapse/SynapseConstants.java

@@ -740,4 +740,6 @@ public enum ENDPOINT_TIMEOUT_TYPE { ENDPOINT_TIMEOUT, GLOBAL_TIMEOUT, HTTP_CONNE
     public static final String CONNECTOR_ARTIFACT = "CONNECTOR_ARTIFACT";
     public static final String APPEND_ARTIFACT_IDENTIFIER = "APPEND_ARTIFACT_IDENTIFIER";
     public static final String EXPOSE_VERSIONED_SERVICES = "expose.versioned.services";
+
+    public static final String SKIP_MAIN_SEQUENCE = "SKIP_MAIN_SEQUENCE";
 }

modules/core/src/main/java/org/apache/synapse/aspects/AspectConfiguration.java

@@ -47,7 +47,9 @@ public class AspectConfiguration implements StatisticsConfigurable, Identifiable
 
 	private Integer hashCode = null;
 
-	public AspectConfiguration(String id) {
+    private boolean traceFilterEnable = false;
+
+    public AspectConfiguration(String id) {
 		this.id = id;
 	}
 
@@ -120,4 +122,15 @@ public Integer getHashCode() {
 		return hashCode;
 	}
 
+    public void enableTraceFilter() {
+        this.traceFilterEnable = true;
+    }
+
+    public void disableTraceFilter() {
+        this.traceFilterEnable = false;
+    }
+
+    public boolean isTraceFilterEnable() {
+        return traceFilterEnable;
+    }
 }

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/collectors/OpenEventCollector.java

@@ -102,7 +102,9 @@ public static Integer reportEntryEvent(MessageContext messageContext, String com
 			StatisticsOpenEvent openEvent = new StatisticsOpenEvent(statisticDataUnit);
             addEventAndIncrementCount(messageContext, openEvent);
 
-            if (isOpenTelemetryEnabled()) {
+            boolean isFiltered = isSpanFilteredMediator(componentName) ||
+                    (aspectConfiguration != null && aspectConfiguration.isTraceFilterEnable());
+            if (!isFiltered && isOpenTelemetryEnabled()) {
 				OpenTelemetryManagerHolder.getOpenTelemetryManager().getHandler()
 						.handleOpenEntryEvent(statisticDataUnit, messageContext);
 			}
@@ -134,7 +136,9 @@ public static Integer reportChildEntryEvent(MessageContext messageContext, Strin
 			reportMediatorStatistics(messageContext, componentName, componentType, isContentAltering, statisticDataUnit,
 			                         aspectConfiguration);
 
-			if (isOpenTelemetryEnabled()) {
+            boolean isFiltered = isSpanFilteredMediator(componentName) ||
+                    (aspectConfiguration != null && aspectConfiguration.isTraceFilterEnable());
+            if (!isFiltered && isOpenTelemetryEnabled()) {
 				OpenTelemetryManagerHolder.getOpenTelemetryManager().getHandler()
 						.handleOpenChildEntryEvent(statisticDataUnit, messageContext);
 			}
@@ -168,7 +172,9 @@ public static Integer reportFlowContinuableEvent(MessageContext messageContext,
 			reportMediatorStatistics(messageContext, componentName, componentType, isContentAltering, statisticDataUnit,
 			                         aspectConfiguration);
 
-			if (isOpenTelemetryEnabled()) {
+            boolean isFiltered = isSpanFilteredMediator(componentName) ||
+                    (aspectConfiguration != null && aspectConfiguration.isTraceFilterEnable());
+            if (!isFiltered && isOpenTelemetryEnabled()) {
 				OpenTelemetryManagerHolder.getOpenTelemetryManager().getHandler()
 						.handleOpenFlowContinuableEvent(statisticDataUnit, messageContext);
 			}
@@ -203,7 +209,9 @@ public static Integer reportFlowSplittingEvent(MessageContext messageContext, St
 			reportMediatorStatistics(messageContext, componentName, componentType, isContentAltering, statisticDataUnit,
 			                         aspectConfiguration);
 
-			if (isOpenTelemetryEnabled()) {
+            boolean isFiltered = isSpanFilteredMediator(componentName) ||
+                    (aspectConfiguration != null && aspectConfiguration.isTraceFilterEnable());
+            if (!isFiltered && isOpenTelemetryEnabled()) {
 				OpenTelemetryManagerHolder.getOpenTelemetryManager().getHandler()
 						.handleOpenFlowSplittingEvent(statisticDataUnit, messageContext);
 			}
@@ -237,7 +245,9 @@ public static Integer reportFlowAggregateEvent(MessageContext messageContext, St
 			reportMediatorStatistics(messageContext, componentName, componentType, isContentAltering, statisticDataUnit,
 			                         aspectConfiguration);
 
-			if (isOpenTelemetryEnabled()) {
+            boolean isFiltered = isSpanFilteredMediator(componentName) ||
+                    (aspectConfiguration != null && aspectConfiguration.isTraceFilterEnable());
+            if (!isFiltered && isOpenTelemetryEnabled()) {
 				OpenTelemetryManagerHolder.getOpenTelemetryManager().getHandler()
 						.handleOpenFlowAggregateEvent(statisticDataUnit, messageContext);
 			}

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/collectors/RuntimeStatisticCollector.java

@@ -18,6 +18,9 @@
 
 package org.apache.synapse.aspects.flow.statistics.collectors;
 
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.synapse.MessageContext;
@@ -71,6 +74,16 @@ public abstract class RuntimeStatisticCollector {
 
     public static long eventExpireTime;
 
+    /**
+     * Is mediator span filter enabled in otlp tracing.
+     */
+    private static boolean isMediatorSpanFilterEnabled;
+
+    /**
+     * Mediator list to filter spans in otlp tracing.
+     */
+    private static Set<String> mediatorSpanFilterSet;
+
     /**
      * Initialize statistics collection when ESB starts.
      */
@@ -129,6 +142,21 @@ public static void init() {
                 }
                 OpenTelemetryManagerHolder.loadTracerConfigurations();
                 OpenTelemetryManagerHolder.setCollectingFlags(isCollectingPayloads, isCollectingProperties, isCollectingVariables);
+                String mediatorSpanFilterListStr = SynapsePropertiesLoader.getPropertyValue(
+                        TelemetryConstants.OLTP_FILTERED_MEDIATOR_NAMES, null);
+                isMediatorSpanFilterEnabled = mediatorSpanFilterListStr != null &&
+                        !mediatorSpanFilterListStr.isEmpty();
+                if (isMediatorSpanFilterEnabled) {
+                    mediatorSpanFilterSet = Stream.of(mediatorSpanFilterListStr.split(","))
+                            .map(String::trim)        // remove leading/trailing spaces
+                            .map(String::toLowerCase) // convert to lowercase
+                            .collect(Collectors.toSet());
+                    if (log.isDebugEnabled()) {
+                        log.debug(
+                                "Mediator span filter is enabled. Mediators with the following names " +
+                                        "will be not traced: " + mediatorSpanFilterListStr);
+                    }
+                }
             }
         } else {
             if (log.isDebugEnabled()) {
@@ -150,6 +178,20 @@ protected static void setStatisticsTraceId(MessageContext msgCtx) {
         }
     }
 
+    /**
+     * Set message Id of the message context as statistic trace Id at the beginning of the statistic flow.
+     *
+     * @param msgCtx Axis2 message context.
+     */
+    protected static void setStatisticsTraceId(org.apache.axis2.context.MessageContext msgCtx) {
+        if (msgCtx.getProperty(StatisticsConstants.FLOW_STATISTICS_ID) == null && msgCtx.getMessageID() != null) {
+            msgCtx.setProperty(StatisticsConstants.FLOW_STATISTICS_ID,
+                    msgCtx.getMessageID().replace(':', '_'));
+        } else if (msgCtx.getMessageID() == null) {
+            log.error("Message ID is null");
+        }
+    }
+
     /**
      * Returns true if statistics is collected in this message flow path.
      *
@@ -241,6 +283,19 @@ public static void setCollectingAllStatistics(boolean state) {
         log.info("Collecting statistics for all artifacts state changed to: " + state);
     }
 
+    /**
+     * Check whether the given mediator is filtered out from tracing when using OpenTelemetry.
+     *
+     * @param componentName Name of the mediator
+     * @return true if the mediator is filtered out
+     */
+    public static boolean isSpanFilteredMediator(String componentName) {
+        if (isMediatorSpanFilterEnabled && componentName != null) {
+            return mediatorSpanFilterSet.contains(componentName.split(":")[0].trim().toLowerCase());
+        }
+        return false;
+    }
+
     /**
      * Helper method to add event and increment stat count so that it denotes, open event is added.
      *

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/data/raw/BasicStatisticDataUnit.java

@@ -18,6 +18,7 @@
 
 package org.apache.synapse.aspects.flow.statistics.data.raw;
 
+import java.util.Map;
 import org.apache.synapse.MessageContext;
 import org.apache.synapse.aspects.flow.statistics.elasticsearch.ElasticMetadata;
 import org.apache.synapse.core.SynapseEnvironment;
@@ -67,7 +68,12 @@ public class BasicStatisticDataUnit {
 	 */
 	private MessageContext messageContext;
 
-	public String getStatisticId() {
+    /**
+     * Custom properties to be added to the statistic data unit.
+     */
+    private Map<String, Object> customProperties;
+
+    public String getStatisticId() {
 		return statisticId;
 	}
 
@@ -130,4 +136,12 @@ public void setMessageContext(MessageContext messageContext) {
 	public MessageContext getMessageContext() {
 		return messageContext;
 	}
+
+    public Map<String, Object> getCustomProperties() {
+        return customProperties;
+    }
+
+    public void setCustomProperties(Map<String, Object> customProperties) {
+        this.customProperties = customProperties;
+    }
 }

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/data/raw/StatisticDataUnit.java

@@ -79,7 +79,12 @@ public class StatisticDataUnit extends BasicStatisticDataUnit {
 	 */
 	private ComponentType componentType;
 
-	/**
+    /**
+     * Component Type String of the reporting component.
+     */
+    private String componentTypeString;
+
+    /**
 	 * Unique Id of the reporting component.
 	 */
 	private String componentId;
@@ -126,7 +131,19 @@ public class StatisticDataUnit extends BasicStatisticDataUnit {
 	 */
 	private String statusCode;
 
-	/**
+    /**
+     * Error code if the statistic event is reported due to an error
+     */
+    private String errorCode;
+
+    /**
+     * Error message if the statistic event is reported due to an error
+     */
+    private String errorMessage;
+
+    private boolean isOuterLayerSpan = false;
+
+    /**
 	 * Status description of the response. Optional
 	 */
 	private String statusDescription;
@@ -299,4 +316,36 @@ public String getPropertyValue() {
 	public void setPropertyValue(String propertyValue) {
 		this.propertyValue = propertyValue;
 	}
+
+    public String getComponentTypeString() {
+        return componentTypeString;
+    }
+
+    public void setComponentTypeString(String componentTypeString) {
+        this.componentTypeString = componentTypeString;
+    }
+
+    public String getErrorCode() {
+        return errorCode;
+    }
+
+    public void setErrorCode(String errorCode) {
+        this.errorCode = errorCode;
+    }
+
+    public String getErrorMessage() {
+        return errorMessage;
+    }
+
+    public void setErrorMessage(String errorMessage) {
+        this.errorMessage = errorMessage;
+    }
+
+    public boolean isOuterLayerSpan() {
+        return isOuterLayerSpan;
+    }
+
+    public void setOuterLayerSpan(boolean isOuterLayerSpan) {
+        this.isOuterLayerSpan = isOuterLayerSpan;
+    }
 }

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/data/raw/StatisticsLog.java

@@ -170,7 +170,9 @@ public class StatisticsLog {
 	 */
 	private String statusDescription;
 
-	/**
+    private Map<String, Object> customProperties;
+
+    /**
 	 * Elastic analytics metadata holder.
 	 */
 	private ElasticMetadata elasticMetadata;
@@ -201,6 +203,7 @@ public StatisticsLog(StatisticDataUnit statisticDataUnit) {
 		this.transportHeaderMap = statisticDataUnit.getTransportHeaderMap();
 		this.statusCode = statisticDataUnit.getStatusCode();
 		this.statusDescription = statisticDataUnit.getStatusDescription();
+        this.customProperties = statisticDataUnit.getCustomProperties();
 	}
 
 	public StatisticsLog(ComponentType componentType, String componentName, int parentIndex) {
@@ -428,4 +431,8 @@ public ElasticMetadata getElasticMetadata() {
 	public void setElasticMetadata(ElasticMetadata elasticMetadata) {
 		this.elasticMetadata = elasticMetadata;
 	}
+
+    public Map<String, Object> getCustomProperties() {
+        return customProperties;
+    }
 }

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/tracing/opentelemetry/management/TelemetryConstants.java

@@ -82,6 +82,13 @@ public class TelemetryConstants {
     public static final String ENDPOINT_ATTRIBUTE_KEY = "Endpoint";
     public static final String CORRELATION_ID_ATTRIBUTE_KEY = "CorrelationId";
 
+    public static final String ERROR_CODE_ATTRIBUTE_KEY = "error.code";
+    public static final String ERROR_MESSAGE_ATTRIBUTE_KEY = "error.message";
+
     public static final String OTEL_RESOURCE_ATTRIBUTE_KEY = "opentelemetry.properties.resource_attributes";
     public static final String OTEL_RESOURCE_ATTRIBUTES_ENVIRONMENT_VARIABLE_NAME = "OTEL_RESOURCE_ATTRIBUTES";
+
+    public static final String OLTP_CUSTOM_SPAN_TAGS = "oltp.custom.span.header.tags";
+
+    public static final String OLTP_FILTERED_MEDIATOR_NAMES = "oltp.filtered.mediator.names";
 }

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/tracing/opentelemetry/management/handling/event/CloseEventHandler.java

@@ -32,13 +32,31 @@ public interface CloseEventHandler {
      */
     void handleCloseEntryEvent(BasicStatisticDataUnit basicStatisticDataUnit, MessageContext synCtx);
 
+    /**
+     * Handles a close entry event.
+     *
+     * @param basicStatisticDataUnit Basic statistic data unit object.
+     * @param msgCtx                 Axis2 Message context.
+     */
+    void handleCloseEntryEvent(BasicStatisticDataUnit basicStatisticDataUnit,
+                               org.apache.axis2.context.MessageContext msgCtx);
+
     /**
      * Handles a forceful close event.
      * @param basicStatisticDataUnit    Basic statistic data unit object.
      * @param synCtx                    Message context.
      */
     void handleCloseFlowForcefully(BasicStatisticDataUnit basicStatisticDataUnit, MessageContext synCtx);
 
+    /**
+     * Handles a forceful close event.
+     *
+     * @param basicStatisticDataUnit Basic statistic data unit object.
+     * @param msgCtx                 Axis2 Message context.
+     */
+    void handleCloseFlowForcefully(BasicStatisticDataUnit basicStatisticDataUnit,
+                                   org.apache.axis2.context.MessageContext msgCtx);
+
     /**
      * Handles a try end flow event.
      * @param basicStatisticDataUnit    Basic statistic data unit object.
@@ -59,4 +77,13 @@ public interface CloseEventHandler {
      * @param synCtx                    Message context.
      */
     void handleCloseEntryWithErrorEvent(BasicStatisticDataUnit basicStatisticDataUnit, MessageContext synCtx);
+
+    /**
+     * Handles a close entry with error event.
+     *
+     * @param basicStatisticDataUnit Basic statistic data unit object.
+     * @param msgCtx                 Message context.
+     */
+    void handleCloseEntryWithErrorEvent(BasicStatisticDataUnit basicStatisticDataUnit,
+                                        org.apache.axis2.context.MessageContext msgCtx);
 }

modules/core/src/main/java/org/apache/synapse/aspects/flow/statistics/tracing/opentelemetry/management/handling/event/OpenEventHandler.java

@@ -34,6 +34,14 @@ public interface OpenEventHandler {
      */
     void handleOpenEntryEvent(StatisticDataUnit statisticDataUnit, MessageContext synCtx);
 
+    /**
+     * Handles an open entry event.
+     *
+     * @param statisticDataUnit Statistic data unit object.
+     * @param msgCtx            Axis2 Message context.
+     */
+    void handleOpenEntryEvent(StatisticDataUnit statisticDataUnit, org.apache.axis2.context.MessageContext msgCtx);
+
     /**
      * Handles an open child entry event.
      * @param statisticDataUnit Statistic data unit object.