feat: capture message attributes and text for flow logs (#79)

* feat: capture message attributes and text for flow logs

Closes #78 

---------

Co-authored-by: Manik Magar <[email protected]>
Co-authored-by: Anil Konakalla <[email protected]>

Author: 3 people

.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/FlowLogConfig.java

@@ -0,0 +1,51 @@
+package com.avioconsulting.mule.logger.api.processor;
+
+import org.mule.runtime.extension.api.annotation.Alias;
+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.ParameterGroup;
+import org.mule.runtime.extension.api.annotation.param.display.Summary;
+
+@Alias("flow-log-config")
+public class FlowLogConfig {
+
+  @Parameter
+  @Summary("Name of the flow to associate given expression as attributes")
+  private String flowName;
+
+  @Parameter
+  @Optional
+  @Summary("A valid dataweave expression that resolves to a Map object with key-value pairs")
+  private String attributesExpressionText;
+
+  @Parameter
+  @Optional
+  @Summary("A valid dataweave expression that results in a String to append to default flow start message")
+  private String messageExpressionText;
+
+  public String getAttributesExpressionText() {
+    return attributesExpressionText;
+  }
+
+  public void setAttributesExpressionText(String attributesExpressionText) {
+    this.attributesExpressionText = attributesExpressionText;
+  }
+
+  public String getMessageExpressionText() {
+    return messageExpressionText;
+  }
+
+  public void setMessageExpressionText(String messageExpressionText) {
+    this.messageExpressionText = messageExpressionText;
+  }
+
+  public String getFlowName() {
+    return flowName;
+  }
+
+  public FlowLogConfig setFlowName(String flowName) {
+    this.flowName = flowName;
+    return this;
+  }
+
+}

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.FlowLogConfig;
 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,21 @@
 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.function.Function;
+import java.util.stream.Collectors;
+
 /**
  * This class represents an extension configuration, values set in this class
  * are commonly used across multiple
@@ -83,6 +91,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<FlowLogConfig> flowLogConfigs;
+
   @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 +154,10 @@ public class CustomLoggerConfiguration implements Startable, Initialisable {
   @Inject
   ExtensionsClient extensionsClient;
 
+  @Inject
+  ExpressionManager expressionManager;
+  private Map<String, FlowLogConfig> flowLogConfigMap;
+
   /**
    * Default constructor for auto-initialization
    */
@@ -175,6 +195,15 @@ public CustomLoggerConfiguration(CustomLoggerRegistrationService customLoggerReg
 
   private static boolean isNotificationListenerRegistered = false;
 
+  public Map<String, FlowLogConfig> getFlowLogConfigMap() {
+    return flowLogConfigMap;
+  }
+
+  public CustomLoggerConfiguration setFlowLogConfigs(List<FlowLogConfig> flowLogConfigs) {
+    this.flowLogConfigs = flowLogConfigs;
+    return this;
+  }
+
   public String getApplicationName() {
     return applicationName;
   }
@@ -291,6 +320,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 +345,8 @@ public void start() throws MuleException {
     customLoggerRegistrationService.setConfig(this);
     if (isEnableFlowLogs()) {
       classLogger.info("Flow logs enabled");
+      flowLogConfigMap = flowLogConfigs.stream().collect(
+          Collectors.toMap(FlowLogConfig::getFlowName, Function.identity()));
       synchronized (CustomLoggerConfiguration.class) {
         if (!isNotificationListenerRegistered) {
           classLogger.info("Creating and registering notification listener");
@@ -345,5 +380,17 @@ public void initialise() throws InitialisationException {
       throw new InitialisationException(createStaticMessage(
           "Encryption Algorithm must be provided if encryption password is being supplied"), this);
     }
+    flowLogConfigs.forEach(flowLogConfig -> {
+      if (flowLogConfig.getMessageExpressionText() == null
+          && flowLogConfig.getAttributesExpressionText() == null) {
+        try {
+          throw new InitialisationException(createStaticMessage(
+              "Both 'attributesExpressionText' and 'messageExpressionText' cannot be empty, at least one or both must be specified."),
+              this);
+        } catch (InitialisationException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    });
   }
 }

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

@@ -1,16 +1,22 @@
 package com.avioconsulting.mule.logger.internal.listeners;
 
-import com.avioconsulting.mule.logger.api.processor.AdditionalProperties;
-import com.avioconsulting.mule.logger.api.processor.ExceptionProperties;
-import com.avioconsulting.mule.logger.api.processor.LogProperties;
-import com.avioconsulting.mule.logger.api.processor.MessageAttributes;
+import com.avioconsulting.mule.logger.api.processor.*;
 import com.avioconsulting.mule.logger.internal.CustomLogger;
 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.Optional;
+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 +25,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 +34,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 +48,56 @@ 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;
+    FlowLogConfig flowLogConfig;
+    /**
+     * 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
+     **/
+    Optional<Map.Entry<String, FlowLogConfig>> matchedEntry = config.getFlowLogConfigMap().entrySet().stream()
+        .filter(entry -> matchWildcard(entry.getKey(), notification.getResourceIdentifier()))
+        .findFirst();
+    if (matchedEntry.isPresent()) {
+      flowLogConfig = matchedEntry.get().getValue();
+      TypedValue<Map<String, String>> evaluate = (TypedValue<Map<String, String>>) config
+          .getExpressionManager()
+          .evaluate("#[" + flowLogConfig.getAttributesExpressionText() + "]",
+              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();
+
+    // Exact match if no wildcards
+    if (searchString.equalsIgnoreCase(wildcardKey)) {
+      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);
+    }
+
+    // If wildcard key is just '*', match everything
+    return cleanWildcardKey.equals("*");
+  }
+
+}

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

@@ -1,11 +1,18 @@
 package com.avioconsulting.mule.logger.internal.listeners;
 
+import com.avioconsulting.mule.logger.api.processor.FlowLogConfig;
 import com.avioconsulting.mule.logger.internal.config.CustomLoggerConfiguration;
+import org.mule.runtime.api.metadata.TypedValue;
 import org.mule.runtime.api.notification.PipelineMessageNotification;
 import org.mule.runtime.api.notification.PipelineMessageNotificationListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
 /*
  * Listener for Mule notifications on flow start, end and completion.
  */
@@ -35,10 +42,24 @@ public void onNotification(PipelineMessageNotification notification) {
             + "]");
     if (config != null) {
       try {
+        String msgToAppend = "";
+        Optional<Map.Entry<String, FlowLogConfig>> matchedEntry = config.getFlowLogConfigMap().entrySet()
+            .stream()
+            .filter(entry -> matchWildcard(entry.getKey(), notification.getResourceIdentifier()))
+            .findFirst();
+        if (matchedEntry.isPresent()) {
+          FlowLogConfig flowLogConfig = matchedEntry.get().getValue();
+          TypedValue<String> evaluate = (TypedValue<String>) config
+              .getExpressionManager()
+              .evaluate("#[" + flowLogConfig.getMessageExpressionText() + "]",
+                  notification.getEvent().asBindingContext());
+          msgToAppend = evaluate.getValue();
+        }
         String message = "Event not processed yet, this should never be shown";
         switch (Integer.parseInt(notification.getAction().getIdentifier())) {
           case PipelineMessageNotification.PROCESS_START:
-            message = "Flow [" + notification.getResourceIdentifier() + "]" + " start";
+            message = "Flow [" + notification.getResourceIdentifier() + "]" + " start "
+                + (msgToAppend != null ? msgToAppend : "");
             break;
           case PipelineMessageNotification.PROCESS_COMPLETE:
             message = "Flow [" + notification.getResourceIdentifier() + "]" + " end";
@@ -49,9 +70,12 @@ 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 (ClassCastException castException) {
+        classLogger.error("Message expression text in flow-log-config needs to be a String", castException);
       } 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);
   }