CT-278 Support wildcards and matchAll for custom app event mapping (#47)

* Short-circuit `sendEvents` if nothing to send
* Log if no event mappers match the records
* Add Regex support for CustomAppEventMapping
* Implement `matchAll` for CustomAppEventMapping

Author: Oliver Hsu

config/connect-scalyr-sink-custom-app.json

@@ -8,6 +8,6 @@
     "topics": "logs",
     "api_key": "<log write api key from https://app.scalyr.com/keys>",
     "event_enrichment": "key1=value1,key2=value2",
-    "custom_app_event_mapping":"[{\"matcher\": {\"attribute\": \"app.name\", \"value\": \"myapp\"}, \"eventMapping\": {\"message\": \"message\", \"logfile\": \"log.path\", \"serverHost\": \"host.hostname\", \"parser\": \"fields.parser\", \"version\": \"app.version\", \"appField1\":\"appField1\", \"appField2\":\"nested.appField2\"}}]"
+    "custom_app_event_mapping":"[{\"matcher\": {\"attribute\": \"app.name\", \"value\": \"myApp.*\"}, \"eventMapping\": {\"message\": \"message\", \"logfile\": \"log.path\", \"serverHost\": \"host.hostname\", \"parser\": \"fields.parser\", \"version\": \"app.version\", \"appField1\":\"appField1\", \"appField2\":\"nested.appField2\"}}, {\"matcher\": {\"matchAll\": \"true\"}, \"eventMapping\": {\"message\": \"message\", \"logfile\": \"log.path\", \"serverHost\": \"host.hostname\", \"parser\": \"fields.parser\"}}]"
   }
 }

config/connect-scalyr-sink.properties

@@ -16,8 +16,13 @@ api_key=<log write api key from https://app.scalyr.com/keys>
 # Optional enrichment attributes in key=value format which are added to the Scalyr event attributes
 #event_enrichment=key1=value1,key2=value2
 
-# Custom application event mapping - only needed for custom applications
-#custom_app_event_mapping=[{"matcher": {"attribute": "app.name", "value": "myapp"}, "eventMapping": {"message": "message", "logfile": "log.path", "serverHost": "host.hostname", "parser": "fields.parser", "version": "app.version", "appField1", "appField1", "appField2", "nested.appField2"}}]
+# Custom application event mapping - only needed for custom applications.  Regex is supported for the `matcher.value`.
+#custom_app_event_mapping=[{"matcher": {"attribute": "app.name", "value": "myapp.*"}, "eventMapping": {"message": "message", "logfile": "log.path", "serverHost": "host.hostname", "parser": "fields.parser", "version": "app.version", "appField1", "appField1", "appField2", "nested.appField2"}}]
+
+# Custom application event mapping with match all and send_entire_record=true
+# logfile, serverHost, and parser event mappings are still recommended, since these are special fields.
+#send_entire_record=true
+#custom_app_event_mapping=[{"matcher": {"matchAll": true}, "eventMapping": {"logfile": "log.path", "serverHost": "host.hostname", "parser": "fields.parser"}}]
 
 # Advanced configuration options - these should not be modified in most cases
 # Compression type to use for sending log events.  Valid values are: `deflate`, `none`.

src/main/java/com/scalyr/integrations/kafka/ScalyrSinkConnectorConfig.java

@@ -65,9 +65,12 @@ public class ScalyrSinkConnectorConfig extends AbstractConfig {
     + BATCH_SEND_SIZE_BYTES_CONFIG + " may not be reached for longer periods of time.";
   public static final String CUSTOM_APP_EVENT_MAPPING_CONFIG = "custom_app_event_mapping";
   private static final String CUSTOM_APP_EVENT_MAPPING_DOC = "JSON config describing how to map custom application nested Kafka messages to Scalyr events." +
-    "  Multiple custom application event mappings can be specified in a JSON list.  Example config JSON:\n"
-    + "[{\"matcher\": { \"attribute\": \"app.name\", \"value\": \"customApp\"},\n" +
-    " \"eventMapping\": { \"message\": \"message\", \"logfile\": \"log.path\", \"serverHost\": \"host.hostname\", \"parser\": \"fields.parser\", \"version\": \"app.version\"} }]";
+    "  Multiple custom application event mappings can be specified in a JSON list and are evaluated in the order specified in the list." +
+    "  The matcher.value supports regex to match the entire field value.  Example config JSON:\n" +
+    "[{\"matcher\": { \"attribute\": \"app.name\", \"value\": \"mpApp.*\"},\n" +
+    " \"eventMapping\": { \"message\": \"message\", \"logfile\": \"log.path\", \"serverHost\": \"host.hostname\", \"parser\": \"fields.parser\", \"version\": \"app.version\"} },\n" +
+    "{\"matcher\": { \"matchAll\": true},\n" +
+    " \"eventMapping\": { \"message\": \"message\", \"logfile\": \"log.path\", \"serverHost\": \"host.hostname\", \"parser\": \"fields.parser\"} }]";
   public static final String SEND_ENTIRE_RECORD = "send_entire_record";
   private static final String SEND_ENTIRE_RECORD_DOC = "If true, send the entire Kafka Connect record value serialized to JSON as the message field.";
 
@@ -138,10 +141,23 @@ public static ConfigDef configDef() {
       if (customAppEventMappings.isEmpty()) {
         throw new ConfigException("No custom event mappings are defined");
       }
+      int numMatchAll = 0;
       for (CustomAppEventMapping mapping : customAppEventMappings) {
-        if (mapping.getMatcherFields().isEmpty() || Strings.isNullOrEmpty(mapping.getMatcherValue())) {
+        if (!mapping.isMatchAll() && (mapping.getMatcherFields().isEmpty() || Strings.isNullOrEmpty(mapping.getMatcherValue()))) {
           throw new ConfigException("Custom event application mapping matcher not defined");
         }
+        if (mapping.isMatchAll()) {
+          numMatchAll++;
+        }
+      }
+
+      // Only one matchAll is allowed to avoid ambiguity
+      if (numMatchAll > 1) {
+        throw new ConfigException("More than one match all custom event mapping matcher is defined");
+      }
+      // Custom event mappings are evaluated in order.  Match all should be last so that other custom event mappings are evaluated.
+      if (numMatchAll == 1 && !customAppEventMappings.get(customAppEventMappings.size() - 1).isMatchAll()) {
+        throw new ConfigException("Custom event mapper with matchAll=true must be listed last. Otherwise, some mappers would be ignored.");
       }
     } catch (IOException | IllegalArgumentException e) {
       throw new ConfigException("Invalid custom application event mapping JSON", e);

src/main/java/com/scalyr/integrations/kafka/ScalyrSinkTask.java

@@ -17,6 +17,7 @@
 package com.scalyr.integrations.kafka;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.RateLimiter;
 import com.scalyr.api.internal.ScalyrUtil;
 import com.scalyr.integrations.kafka.mapping.CustomAppEventMapping;
 import com.scalyr.integrations.kafka.mapping.EventMapper;
@@ -39,6 +40,7 @@
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.LongConsumer;
 import java.util.stream.Collectors;
 
@@ -74,6 +76,8 @@ public class ScalyrSinkTask extends SinkTask {
 
   private long lastBatchSendTimeMs;
 
+  private static final RateLimiter noRecordLogRateLimiter = RateLimiter.create(1.0/30);  // 1 permit every 30 seconds to not log
+
   /**
    * Default constructor called by Kafka Connect.
    */
@@ -146,16 +150,23 @@ public void put(Collection<SinkRecord> records) {
       throw lastError;
     }
 
+    AtomicInteger recordCount = new AtomicInteger();
     records.stream()
       .map(eventMapper::createEvent)
       .filter(Objects::nonNull)
       .forEach(event -> {
+        recordCount.incrementAndGet();
         if (eventBuffer.estimatedSerializedBytes() + event.estimatedSerializedBytes() >= batchSendSizeBytes) {
           sendEvents();
         }
         eventBuffer.addEvent(event);
       });
 
+    // No early return when recordCount == 0 to send events when batchSendWaitMs exceeded
+    if (recordCount.get() == 0 && noRecordLogRateLimiter.tryAcquire()) {
+      log.warn("No records matched an event mapper.  Records not sent to Scalyr.  Check the custom_app_event_mapping matcher configuration.");
+    }
+
     // Send events when batchSendWaitMs exceeded
     if (ScalyrUtil.currentTimeMillis() - lastBatchSendTimeMs >= batchSendWaitMs) {
       sendEvents();
@@ -166,6 +177,9 @@ public void put(Collection<SinkRecord> records) {
    * Call addEvents with EventBuffer
    */
   private void sendEvents() {
+    if (eventBuffer.length() == 0) {
+      return;
+    }
     PerfStats perfStats = new PerfStats(log);
     perfStats.recordEvents(eventBuffer);
     pendingAddEvents = addEventsClient.log(eventBuffer.getEvents(), pendingAddEvents).whenComplete(this::processResponse);
@@ -188,9 +202,7 @@ public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
     log.debug("Flushing data to Scalyr with the following offsets: {}", currentOffsets);
 
     // Send pending events in buffer
-    if (eventBuffer.length() > 0) {
-      sendEvents();
-    }
+    sendEvents();
 
     // Wait for in-flight addEvents requests to complete
     waitForRequestsToComplete();

src/main/java/com/scalyr/integrations/kafka/mapping/CustomAppEventMapping.java

@@ -54,6 +54,15 @@
 
  `matcher` defines an attribute to determine whether the event mapping applies to the message.
  The event mapping is only applied to messages where the `matcher.attribute` value matches the `matcher.value`.
+ `matcher.value` can be a regex.
+
+ `matcher` can also be defined to match all events:
+ ```
+ "matcher": {
+   "matchAll": true
+ }
+ ```
+ When `matchAll=true`, `attribute` and `value` fields are not required and are ignored.
 
  `eventMapping` defines the message fields that are mapped to Scalyr event attributes.
  The attribute is the Scalyr event key.  The attribute value specifies the
@@ -78,23 +87,27 @@ public class CustomAppEventMapping {
   public static final String LOG_FILE = "logfile";
   public static final String PARSER = "parser";
   public static final String MESSAGE = "message";
+  private static final String MATCHER_NOT_DEFINED_ERROR = "matcher not defined in custom application event mapping";
 
   private static final Set<String> standardAttrs = ImmutableSet.of(SERVER_HOST, LOG_FILE, PARSER, MESSAGE);
 
-  public void setMatcher(Matcher matcher) {
+  public CustomAppEventMapping setMatcher(Matcher matcher) {
     this.matcher = matcher;
+    return this;
   }
 
-  public void setEventMapping(Map<String, String> eventMapping) {
+  public CustomAppEventMapping setEventMapping(Map<String, String> eventMapping) {
     this.eventMapping = eventMapping;
+    return this;
   }
 
-  public void setDelimiter(String fieldDelimiter) {
+  public CustomAppEventMapping setDelimiter(String fieldDelimiter) {
     this.delimiter = fieldDelimiter;
+    return this;
   }
 
   public String getMatcherValue() {
-    Preconditions.checkArgument(matcher != null, "matcher not defined in custom application event mapping");
+    Preconditions.checkArgument(matcher != null, MATCHER_NOT_DEFINED_ERROR);
     return matcher.value;
   }
 
@@ -103,10 +116,15 @@ public String getMatcherValue() {
    * using the delimiter as the field separator.
    */
   public List<String> getMatcherFields() {
-    Preconditions.checkArgument(matcher != null, "matcher not defined in custom application event mapping");
+    Preconditions.checkArgument(matcher != null, MATCHER_NOT_DEFINED_ERROR);
     return splitAttrFields(matcher.attribute);
   }
 
+  public boolean isMatchAll() {
+    Preconditions.checkArgument(matcher != null, MATCHER_NOT_DEFINED_ERROR);
+    return matcher.matchAll;
+  }
+
   public List<String> getServerHostFields() {
     return getAttribute(SERVER_HOST);
   }
@@ -147,15 +165,23 @@ private List<String> splitAttrFields(String attrFields) {
    * Defines the field and value to determine whether the message matches the custom app event mapping.
    */
   public static class Matcher {
+    private boolean matchAll;
     private String attribute;
     private String value;
 
-    public void setAttribute(String attribute) {
+    public Matcher setMatchAll(Boolean matchAll) {
+      this.matchAll = matchAll;
+      return this;
+    }
+
+    public Matcher setAttribute(String attribute) {
       this.attribute = attribute;
+      return this;
     }
 
-    public void setValue(String value) {
+    public Matcher setValue(String value) {
       this.value = value;
+      return this;
     }
 
     @Override

src/main/java/com/scalyr/integrations/kafka/mapping/CustomAppMessageMapper.java

@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.regex.Pattern;
 
 /**
  * Maps custom app messages to Scalyr events using {@link CustomAppEventMapping} event mapping definition.
@@ -32,7 +33,8 @@ public class CustomAppMessageMapper implements MessageMapper {
   private final List<String> parserFields;
   private final List<String> matcherFields;
   private final Map<String, List<String>> additionalAttrsFields;
-  private final String matcherValue;
+  private final Pattern matcherRegex;
+  private final boolean matchAll;
 
   /**
    * CustomApplicationDefinition fields are memoized to instance variables
@@ -44,8 +46,9 @@ public CustomAppMessageMapper(CustomAppEventMapping customAppEventMapping) {
     serverHostFields = customAppEventMapping.getServerHostFields();
     parserFields = customAppEventMapping.getParserFields();
     matcherFields = customAppEventMapping.getMatcherFields();
+    matchAll = customAppEventMapping.isMatchAll();
+    matcherRegex = matchAll ? null : Pattern.compile(customAppEventMapping.getMatcherValue());
     additionalAttrsFields = customAppEventMapping.getAdditionalAttrFields();
-    matcherValue = customAppEventMapping.getMatcherValue();
   }
 
   @Override
@@ -78,7 +81,11 @@ public Map<String, Object> getAdditionalAttrs(SinkRecord record) {
 
   @Override
   public boolean matches(SinkRecord record) {
+    if (matchAll) {
+      return true;
+    }
+
     Object fieldValue = FieldExtractor.getField(record.value(), matcherFields);
-    return matcherValue.equals(fieldValue);
+    return fieldValue == null ? false : matcherRegex.matcher(fieldValue.toString()).matches();
   }
 }

src/main/java/com/scalyr/integrations/kafka/mapping/EventMapper.java

@@ -34,7 +34,9 @@
 /**
  * Converts a SinkRecord to a Scalyr Event.
  * Determines which {@link MessageMapper} to use and maps the SinkRecord to an Event using the
- * MessageMapper.
+ * MessageMapper.  MessageMappers are evaluated in the following order:
+ * 1. Scalyr supported message sources
+ * 2. Custom app event mappers in the order they are defined in connector config.
  */
 public class EventMapper {
   private static final Logger log = LoggerFactory.getLogger(EventMapper.class);

src/test/java/com/scalyr/integrations/kafka/ScalyrSinkConnectorConfigTest.java

@@ -17,6 +17,7 @@
 package com.scalyr.integrations.kafka;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
@@ -211,22 +212,41 @@ public void testCustomAppEventMappingValidator() throws IOException {
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
     config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
 
+    // match all - parsed matchAll ok even though no matcher fields are defined
+    config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, TestValues.CUSTOM_APP_MATCH_ALL_EVENT_MAPPING_JSON);
+    new ScalyrSinkConnectorConfig(config);
+    config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
+
     // No matcher field
-    Map<String, Object> customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    Map<String, Object> customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.remove("matcher");
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
     config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
 
+    // multiple match all
+    config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(ImmutableList.of(
+      TestValues.createCustomAppEventMapping(".", true),
+      TestValues.createCustomAppEventMapping(".", true))));
+    fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
+    config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
+
+    // match all not last custom app event mapping
+    config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(ImmutableList.of(
+      TestValues.createCustomAppEventMapping(".", true),
+      TestValues.createCustomAppEventMapping(".", false))));
+    fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
+    config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
+
     // No event mapping when SEND_ENTIRE_RECORD=false
-    customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.remove("eventMapping");
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
     config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
 
     // No event mapping when SEND_ENTIRE_RECORD=true
-    customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.remove("eventMapping");
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     config.put(SEND_ENTIRE_RECORD, "true");
@@ -235,28 +255,28 @@ public void testCustomAppEventMappingValidator() throws IOException {
     config.remove(SEND_ENTIRE_RECORD);
 
     // Empty matcher
-    customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.put("matcher", Collections.EMPTY_MAP);
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
     config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
 
     // Empty event mapping
-    customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.put("eventMapping", Collections.EMPTY_MAP);
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
     config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
 
     // No message or application attributes defined
-    customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.put("eventMapping", TestUtils.makeMap("logfile", "/var/log/syslog", "serverHost", "testHost", "parser", "systemLog"));
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);
     config.remove(CUSTOM_APP_EVENT_MAPPING_CONFIG);
 
     // Invalid fields
-    customAppEventMapping = TestValues.createCustomAppEventMapping(".");
+    customAppEventMapping = TestValues.createCustomAppEventMapping(".", false);
     customAppEventMapping.put("extraField", "bad");
     config.put(CUSTOM_APP_EVENT_MAPPING_CONFIG, objectMapper.writeValueAsString(Collections.singletonList(customAppEventMapping)));
     fails(() -> new ScalyrSinkConnectorConfig(config), ConfigException.class);