feat: capture message attributes for flow logs

.gitignore

@@ -3,4 +3,7 @@
 target/
 *.iml
 anypoint-pom.xml
-unit-test.log
+unit-test.log
+
+docs/
+.flattened-pom.xml

pom.xml

@@ -10,7 +10,7 @@
   </parent>
 
   <artifactId>mule-custom-logger</artifactId>
-  <version>3.0.1-SNAPSHOT</version>
+  <version>3.1.0-SNAPSHOT</version>
   <packaging>mule-extension</packaging>
   <name>Mule Custom Logger</name>
   <description>Mule Custom Logger module that provides standard structured logging</description>

src/main/java/com/avioconsulting/mule/logger/api/processor/FlowLogAttributesExpression.java

@@ -0,0 +1,60 @@
+package com.avioconsulting.mule.logger.api.processor;
+
+import org.mule.runtime.extension.api.annotation.Alias;
+import org.mule.runtime.extension.api.annotation.param.Parameter;
+import org.mule.runtime.extension.api.annotation.param.display.Summary;
+
+import java.util.Objects;
+
+@Alias("flow-attributes-expression")
+public class FlowLogAttributesExpression {
+
+  @Parameter
+  @Summary("Name of the flow to associate given expression as attributes")
+  private String flowName;
+  @Parameter
+  @Summary("A valid dataweave expression that resolves to a Map object with key-value pairs")
+  private String expressionText;
+
+  public String getFlowName() {
+    return flowName;
+  }
+
+  public FlowLogAttributesExpression setFlowName(String flowName) {
+    this.flowName = flowName;
+    return this;
+  }
+
+  public String getExpressionText() {
+    return expressionText;
+  }
+
+  public FlowLogAttributesExpression setExpressionText(String expressionText) {
+    this.expressionText = expressionText;
+    return this;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o)
+      return true;
+    if (o == null || getClass() != o.getClass())
+      return false;
+    FlowLogAttributesExpression that = (FlowLogAttributesExpression) o;
+    return Objects.equals(getFlowName(), that.getFlowName())
+        && Objects.equals(getExpressionText(), that.getExpressionText());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(getFlowName(), getExpressionText());
+  }
+
+  @Override
+  public String toString() {
+    return "FlowLogAttributesExpression{" +
+        "flowName='" + flowName + '\'' +
+        ", attributeExpression='" + expressionText + '\'' +
+        '}';
+  }
+}

src/main/java/com/avioconsulting/mule/logger/api/processor/MessageAttributes.java

@@ -41,6 +41,10 @@ public List<MessageAttribute> getAttributeList() {
     return this.messageAttributes;
   }
 
+  public void addAttributes(Map<String, String> attributes) {
+    attributes.forEach((key, value) -> messageAttributes.add(new MessageAttribute(key, value)));
+  }
+
   public Map<String, String> getAttributes() {
     Map<String, String> attributes = new LinkedHashMap<>();
     if (messageAttributes != null) {

src/main/java/com/avioconsulting/mule/logger/internal/config/CustomLoggerConfiguration.java

@@ -4,6 +4,7 @@
 
 import com.avioconsulting.mule.logger.api.processor.Compressor;
 import com.avioconsulting.mule.logger.api.processor.EncryptionAlgorithm;
+import com.avioconsulting.mule.logger.api.processor.FlowLogAttributesExpression;
 import com.avioconsulting.mule.logger.api.processor.LogProperties;
 import com.avioconsulting.mule.logger.internal.CustomLogger;
 import com.avioconsulting.mule.logger.internal.CustomLoggerOperation;
@@ -18,14 +19,20 @@
 import org.mule.runtime.api.lifecycle.Startable;
 import org.mule.runtime.api.meta.ExpressionSupport;
 import org.mule.runtime.api.notification.NotificationListenerRegistry;
+import org.mule.runtime.core.api.el.ExpressionManager;
 import org.mule.runtime.extension.api.annotation.Expression;
 import org.mule.runtime.extension.api.annotation.Operations;
+import org.mule.runtime.extension.api.annotation.param.NullSafe;
 import org.mule.runtime.extension.api.annotation.param.Optional;
 import org.mule.runtime.extension.api.annotation.param.Parameter;
 import org.mule.runtime.extension.api.annotation.param.display.*;
 import org.mule.runtime.extension.api.client.ExtensionsClient;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
 /**
  * This class represents an extension configuration, values set in this class
  * are commonly used across multiple
@@ -83,6 +90,14 @@ public class CustomLoggerConfiguration implements Startable, Initialisable {
   @Expression(ExpressionSupport.NOT_SUPPORTED)
   private LogProperties.LogLevel flowLogLevel;
 
+  @Parameter
+  @DisplayName("Flow Log Attributes")
+  @Summary("The level flow logs will be logged at if enabled")
+  @NullSafe
+  @Optional
+  @Expression(ExpressionSupport.NOT_SUPPORTED)
+  private List<FlowLogAttributesExpression> flowLogAttributes;
+
   @Parameter
   @DisplayName("Flow Log Category Suffix")
   @Summary("This category will be appended to the default logger category and used for all flow logs")
@@ -138,6 +153,10 @@ public class CustomLoggerConfiguration implements Startable, Initialisable {
   @Inject
   ExtensionsClient extensionsClient;
 
+  @Inject
+  ExpressionManager expressionManager;
+  private Map<String, String> flowLogAttributesMap;
+
   /**
    * Default constructor for auto-initialization
    */
@@ -175,6 +194,15 @@ public CustomLoggerConfiguration(CustomLoggerRegistrationService customLoggerReg
 
   private static boolean isNotificationListenerRegistered = false;
 
+  public Map<String, String> getFlowLogAttributesMap() {
+    return flowLogAttributesMap;
+  }
+
+  public CustomLoggerConfiguration setFlowLogAttributes(List<FlowLogAttributesExpression> flowLogAttributes) {
+    this.flowLogAttributes = flowLogAttributes;
+    return this;
+  }
+
   public String getApplicationName() {
     return applicationName;
   }
@@ -291,6 +319,10 @@ public ExtensionsClient getExtensionsClient() {
     return extensionsClient;
   }
 
+  public ExpressionManager getExpressionManager() {
+    return expressionManager;
+  }
+
   /**
    * This method is invoked by the MuleSoft application when the AVIO Custom
    * Logger is invoked to create the connection.
@@ -312,6 +344,8 @@ public void start() throws MuleException {
     customLoggerRegistrationService.setConfig(this);
     if (isEnableFlowLogs()) {
       classLogger.info("Flow logs enabled");
+      flowLogAttributesMap = flowLogAttributes.stream().collect(Collectors
+          .toMap(FlowLogAttributesExpression::getFlowName, FlowLogAttributesExpression::getExpressionText));
       synchronized (CustomLoggerConfiguration.class) {
         if (!isNotificationListenerRegistered) {
           classLogger.info("Creating and registering notification listener");

src/main/java/com/avioconsulting/mule/logger/internal/listeners/CustomLoggerAbstractNotificationListener.java

@@ -8,9 +8,17 @@
 import com.avioconsulting.mule.logger.internal.config.CustomLoggerConfiguration;
 import org.mule.runtime.api.component.location.ComponentLocation;
 import org.mule.runtime.api.event.Event;
+import org.mule.runtime.api.metadata.TypedValue;
+import org.mule.runtime.api.notification.EnrichedServerNotification;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 public abstract class CustomLoggerAbstractNotificationListener {
   protected final CustomLoggerConfiguration config;
+  private Map<String, String> emptyAttributes = Collections.emptyMap();
 
   public CustomLoggerAbstractNotificationListener(CustomLoggerConfiguration config) {
     this.config = config;
@@ -19,7 +27,7 @@ public CustomLoggerAbstractNotificationListener(CustomLoggerConfiguration config
   protected abstract org.slf4j.Logger getClassLogger();
 
   protected void logMessage(ComponentLocation location, Event event, String logMessage, String categoryPrefix,
-      LogProperties.LogLevel level) {
+      LogProperties.LogLevel level, Map<String, String> additionalAttributes) {
     CustomLogger logger = config.getLogger();
     LogProperties logProperties = new LogProperties();
     MessageAttributes messageAttributes = new MessageAttributes();
@@ -28,6 +36,7 @@ protected void logMessage(ComponentLocation location, Event event, String logMes
           .getValue();
       messageAttributes.setOTelContextObject(oTelContextObject);
     }
+    messageAttributes.addAttributes(additionalAttributes);
     ExceptionProperties exceptionProperties = new ExceptionProperties();
     AdditionalProperties additionalProperties = new AdditionalProperties();
     additionalProperties.setIncludeLocationInfo(true);
@@ -41,4 +50,66 @@ protected void logMessage(ComponentLocation location, Event event, String logMes
     logger.log(logProperties, messageAttributes, exceptionProperties, additionalProperties, config,
         location, correlationId);
   }
+
+  protected Map<String, String> getFlowLogAttributes(EnrichedServerNotification notification) {
+    Map<String, String> value = emptyAttributes;
+    String expression = config.getFlowLogAttributesMap().get(notification.getResourceIdentifier());
+    if (expression != null) {
+      TypedValue<Map<String, String>> evaluate = (TypedValue<Map<String, String>>) config.getExpressionManager()
+          .evaluate("#[" + expression + "]",
+              notification.getEvent().asBindingContext());
+      value = evaluate.getValue();
+      if (value == null)
+        value = emptyAttributes;
+    }
+    /**
+     * Flow name can contain wildcard (*)
+     * We only look for wildcard either starting of the string or ending of the
+     * string
+     * ex: mq-listener-* will look for all the flows that starts with mq-listener
+     * ex: *-mq-flow will look for all the flows that ends with -mq-flow
+     **/
+    else {
+      List<Map.Entry<String, String>> matchedEntries = config.getFlowLogAttributesMap().entrySet().stream()
+          .filter(entry -> matchWildcard(entry.getKey(), notification.getResourceIdentifier()))
+          .collect(Collectors.toList());
+      if (!matchedEntries.isEmpty()) {
+        expression = matchedEntries.get(0).getValue();
+        TypedValue<Map<String, String>> evaluate = (TypedValue<Map<String, String>>) config
+            .getExpressionManager()
+            .evaluate("#[" + expression + "]",
+                notification.getEvent().asBindingContext());
+        value = evaluate.getValue();
+        if (value == null)
+          value = emptyAttributes;
+      }
+    }
+    return value;
+  }
+
+  public boolean matchWildcard(String wildcardKey, String searchString) {
+    // Trim the wildcard key
+    String cleanWildcardKey = wildcardKey.trim();
+
+    // If wildcard key is just '*', match everything
+    if (cleanWildcardKey.equals("*")) {
+      return true;
+    }
+
+    // Handle start wildcard
+    if (cleanWildcardKey.startsWith("*")) {
+      String suffix = cleanWildcardKey.substring(1);
+      return searchString.endsWith(suffix);
+    }
+
+    // Handle end wildcard
+    if (cleanWildcardKey.endsWith("*")) {
+      String prefix = cleanWildcardKey.substring(0, cleanWildcardKey.length() - 1);
+      return searchString.startsWith(prefix);
+    }
+
+    // Exact match if no wildcards
+    return searchString.equals(wildcardKey);
+  }
+
 }

src/main/java/com/avioconsulting/mule/logger/internal/listeners/CustomLoggerFlowRefNotificationListener.java

@@ -59,8 +59,9 @@ public void onNotification(MessageProcessorNotification notification) {
             return;
         }
         classLogger.info(message);
+        Map<String, String> flowLogAttributes = getFlowLogAttributes(notification);
         logMessage(location, notification.getEvent(), message, FLOW_REF_CATEGORY_SUFFIX,
-            config.getFlowLogLevel());
+            config.getFlowLogLevel(), flowLogAttributes);
       } catch (Exception e) {
         classLogger.error("Error processing flow notification", e);
       }
@@ -69,4 +70,5 @@ public void onNotification(MessageProcessorNotification notification) {
           "Configuration hasn't been supplied to notification listener yet, flow logs won't be generated.");
     }
   }
+
 }

src/main/java/com/avioconsulting/mule/logger/internal/listeners/CustomLoggerPipelineNotificationListener.java

@@ -6,6 +6,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 /*
  * Listener for Mule notifications on flow start, end and completion.
  */
@@ -49,9 +51,10 @@ public void onNotification(PipelineMessageNotification notification) {
             return;
         }
         classLogger.debug(message);
+        Map<String, String> flowLogAttributes = getFlowLogAttributes(notification);
         logMessage(notification.getComponent().getLocation(), notification.getEvent(), message,
             config.getFlowCategorySuffix(),
-            config.getFlowLogLevel());
+            config.getFlowLogLevel(), flowLogAttributes);
       } catch (Exception e) {
         classLogger.error("Error processing flow notification", e);
       }

src/test/java/com/avioconsulting/mule/logger/CustomLoggerArtifactTest.java

@@ -13,7 +13,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.time.Duration;
+import java.util.Collections;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -44,7 +44,8 @@ public void cleanup() throws Exception {
   @Test
   public void testLoggerConfigForCorrelationId() throws Exception {
     // TODO: Intercept logs and validate entries
-    CoreEvent coreEvent = flowRunner("custom-logger-configFlow").run();
+    CoreEvent coreEvent = flowRunner("custom-logger-configFlow")
+        .withAttributes(Collections.singletonMap("some", "value")).run();
     Assert.assertNotNull(coreEvent);
   }
 

src/test/resources/custom-logger-config.xml

@@ -5,11 +5,20 @@
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
 http://www.mulesoft.org/schema/mule/avio-logger http://www.mulesoft.org/schema/mule/avio-logger/current/mule-avio-logger.xsd">
 
-    <avio-logger:config name="AVIO_Logger_Config" doc:name="AVIO Logger Config" doc:id="84bd2712-015b-46b8-93e9-94536774b8ad"
-                      applicationVersion="1" applicationName="munit" environment="test" compressor="GZIP" encryptionAlgorithm="PBEWithHmacSHA512AndAES_128" encryptionPassword="example" enableFlowLogs="true"/>
 
     <avio-logger:config name="AVIO_Logger_Config_Json" doc:name="AVIO Logger Config" doc:id="84bd2712-015b-46b8-93e9-94536774b8ad"
-                        applicationVersion="1" applicationName="munit" environment="test" compressor="GZIP" encryptionAlgorithm="PBEWithHmacSHA512AndAES_128" encryptionPassword="example" enableFlowLogs="true" formatAsJson="true"/>
+                        applicationVersion="1" applicationName="munit" environment="test" compressor="GZIP" encryptionAlgorithm="PBEWithHmacSHA512AndAES_128" encryptionPassword="example" enableFlowLogs="true" formatAsJson="true" flowLogLevel="INFO">
+        <avio-logger:flow-log-attributes >
+            <avio-logger:flow-attributes-expression flowName="custom-logger-configFlow" expressionText="attributes" />
+        </avio-logger:flow-log-attributes>
+    </avio-logger:config>
+
+    <avio-logger:config name="AVIO_Logger_Config" doc:name="AVIO Logger Config" doc:id="84bd2712-015b-46b8-93e9-94536774b8ad"
+                      applicationVersion="1" applicationName="munit" environment="test" compressor="GZIP" encryptionAlgorithm="PBEWithHmacSHA512AndAES_128" encryptionPassword="example" enableFlowLogs="true" flowLogLevel="INFO">
+        <avio-logger:flow-log-attributes >
+            <avio-logger:flow-attributes-expression flowName="custom-logger-configFlow" expressionText="attributes" />
+        </avio-logger:flow-log-attributes>
+    </avio-logger:config>
 
     <flow name="custom-logger-configFlow" doc:id="0068e855-24dc-41a4-8e13-41ca38de99ba">
         <set-variable value="#[{'traceId': '76d5bcae3d49ff2e1b5ace9f0dcbee42','spanId': 'fa6fbe46daf007b9','spanIdLong': '18045851443427018681','traceparent': '00-76d5bcae3d49ff2e1b5ace9f0dcbee42-fa6fbe46daf007b9-01','TRACE_TRANSACTION_ID': 'bfacf7c0-d583-11ee-adfa-bcd074a0357f','traceIdLongLowPart': '1971114969454603842'}]" doc:name="Set Variable" doc:id="d7a7dade-4453-40cb-97bc-3ca7034372c6" variableName="OTEL_TRACE_CONTEXT"/>