- SelectFilter added. Use a regular expression to specify what to keep in each event

- GapDetection will now print gaps each time the statistics is printed. Events that appear out of order will now be reflected without need to shut down the receiver.

Author: anders-wartoft

README.md

@@ -16,8 +16,9 @@ java -jar target/LogGenerator-with-dependencies.jar -i kafka -icn test2 -itn tes
 When running the last command, press Ctrl-C to see the gaps in the received data. Since we started the counter on 100, there should at least be one gap: 1-99.
 
 ### Latest Release notes
-#### 1.04
-- Bug fixe in output item factory. Missing TCP and File outputs.
+#### 1.05
+- SelectFilter added. Use a regular expression to specify what to keep in each event
+- GapDetection will now print gaps each time the statistics is printed. Events that appear out of order will now be reflected without need to shut down the receiver.
 
 ### Input modules:
 There are input module for the following tasks:
@@ -285,6 +286,22 @@ Parameters: `-he {header text}`
 
 Example: `-he {syslog-header}`, `-he "My custom header with date: {date:yyyyMMdd}: "`
 
+### Select by regex
+If you want to filter out everything but a specific part of an event, you can use the SelectFilter.
+Specify what to keep with a group in a regular expression.
+
+Parameters: `-se {regex with a capture group to keep}`
+
+Example: `-se '-(\d+}$'`
+
+If you have events that are formatted like this:
+
+`[Sat Dec 03 00:35:57.399 Usb Host Notification Apple80211Set: seqNum 5460 Total 1 chg 0 en0]`
+
+and for example want to extract the first word after the timestamp, you can use the following parameters:
+
+`-se '\.\d{3} (\S+) '`
+
 #### Replace by regex
 A use case is if you have a lot of nice logs, but the date is not possible to use. You can load the file and add a regex to find the date, then replace the date with a {date:...} variable or a static string.
 There must be a capture group in the regex. The text matched by the capture group will be replaced by the value.
@@ -337,6 +354,35 @@ Example: `-gd "<(\d+)> -dd true -gdjr true"`
 A good use of gap detection is to send events over unreliable media and check if all events were delivered.
 To assure that the gap detection is on, start the counter on the sending side with a number that is larger than 1. In that case, the gap detector will produce at least one gap (1-your start number).
 
+A special use case is to continuously monitor for missed events and also to be able to react to events that come out of order and not report them as missing.
+For example, log events are generated with:
+
+`java -jar LogGenerator.jar -i counter -string test- -o udp -oh localhost -op 9999 -e 10 -s true`
+
+`-e 100` will limit the generated events to 100 every second.
+
+You can now start a server and monitor for missing events. 
+
+`java -jar LogGenerator.jar -i udp -ip 9999 -gd '-(\d+)$' -cgd true -o null -s true`
+
+If you start the server after the client you will have some missing events (after 30 seconds), for example:
+``` 
+Transaction:  transferred 439 lines in 43939 milliseconds, 0,010 kEPS 0,001 MBPS
+Gaps found: 1.
+1-442
+Number of unique received numbers: 439
+Next expected number: 882
+```
+If you now restart the client and wait another 30 seconds, you should se a smaller gap, or none at all.
+```
+Transaction transfer: TRANSACTION transferred 285 lines in 28391 milliseconds, 0,010 kEPS 0,001 MBPS
+Gaps found: 1.
+286-442
+Number of unique received numbers: 724
+Next expected number: 882
+```
+So, the gap detection can detect events that should have been delivered earlier, and remove them from the gaps.
+
 ### Variables
 #### Date
 A Date variable will take the current time and format according to a Java 

pom.xml

@@ -4,7 +4,7 @@
 
   <groupId>nu.sitia.LogGenerator</groupId>
   <artifactId>LogGenerator</artifactId>
-  <version>1.04-SNAPSHOT</version>
+  <version>1.05-SNAPSHOT</version>
   <packaging>jar</packaging>
 
   <name>LogGenerator</name>

src/main/java/nu/sitia/loggenerator/Configuration.java

@@ -82,6 +82,7 @@ public class Configuration {
                 keys.add(new Item("-ib", "--input-batch-size", "How many rows to read before sending to processing"));
                 keys.add(new Item("-ob", "--output-batch-size", "How many rows to write (new line separated) every time the send method is called"));
                 keys.add(new Item("-gd", "--gap-detection", "Regex to find the event serial number. In the regex, the first capture group must be the number, e.g. \"<(\\d+)>\""));
+                keys.add(new Item("-cgd", "--continuous-gap-detection", "Should Gap Detection be printed every time a statistics report is printed? Valid values are: false, true. Default is false"));
                 keys.add(new Item("-dd", "--duplicate-detection", "If -gd (--gap-detection) is enabled, use this flag to also get a report on all serial numbers that occurs mote than once. Valid values are: false, true"));
                 keys.add(new Item("-gdjr", "--gap-detection-json-report", "Should the gap detector report in JSON format instead of printable format? Valid values are: false, true"));
 
@@ -110,6 +111,7 @@ public class Configuration {
                 keys.add(new Item("-eiq", "--elastic-input-query", "A query string used to get the response from Elastic"));
                 keys.add(new Item("-jf", "--json-filter", "Get a specific node in the JSON input. Use . to traverse the JSON input. If the returned object is an array, the array items are sent in the event chain as new events"));
                 keys.add(new Item("-jfp", "--json-file-path", "Get a specific node in the JSON file input. Use . to traverse the JSON input. If the returned object is an array, the array items are sent in the event chain as new events"));
+                keys.add(new Item("-se", "--select", "Remove everything except what matches the next argument's first group. Example -se 'userid:(\\S+)'"));
         }
         static final Map<String, String> standardVariables = new HashMap<>();
         static {

src/main/java/nu/sitia/loggenerator/ItemProxy.java

@@ -17,6 +17,7 @@
 
 package nu.sitia.loggenerator;
 
+import nu.sitia.loggenerator.filter.GapDetectionFilter;
 import nu.sitia.loggenerator.filter.ProcessFilter;
 import nu.sitia.loggenerator.inputitems.InputItem;
 import nu.sitia.loggenerator.outputitems.OutputItem;
@@ -64,6 +65,11 @@ public class ItemProxy {
     /** If -t time:xxx, this is the start time + xxx */
     private long endTime = 0;
 
+    /** If we have a gapDetector and the flag -cgd is true, then show
+     * gaps every time the statistics has been printed.
+     */
+    private ProcessFilter gapDetector = null;
+
     /**
      * Default constructor
      * @param input Input to use
@@ -113,6 +119,15 @@ public ItemProxy(InputItem input, OutputItem output, List<ProcessFilter> filterL
             shutdownHandlers.add((ShutdownHandler) output);
         }
 
+        String isContinousGapDetectionString = config.getValue("-cgd");
+        if (isContinousGapDetectionString != null && isContinousGapDetectionString.equalsIgnoreCase("true")) {
+            gapDetector = getGapDetector(filterList);
+            if (null == gapDetector) {
+                throw new RuntimeException("The flag -cgd cannot be used without a GapDetector (-gd)");
+            }
+        }
+
+
         this.sentEvents = 0;
 
         // Ctrl-C
@@ -133,6 +148,18 @@ public ItemProxy(InputItem input, OutputItem output, List<ProcessFilter> filterL
 
     }
 
+
+    /**
+     * Search the list of filters and, if present, return a GapDetecttionFilter
+     * @param filterList a List<filter> to search
+     * @return GapDetectionFilter or null
+     */
+    private ProcessFilter getGapDetector(List<ProcessFilter>filterList) {
+        return filterList.stream().reduce(null, (_sofar, element) ->
+                element instanceof GapDetectionFilter ? element: null
+        );
+    }
+
     /**
      * Main loop. Pump messages from the input, modify and then
      * write to the output. Batching may be done in the input and
@@ -166,7 +193,11 @@ public void pump() {
             sentEvents += toSend.size();
 
             if (statistics != null) {
-                statistics.calculateStatistics(filtered);
+                boolean hasPrinted = statistics.calculateStatistics(filtered);
+                if (hasPrinted && null != gapDetector) {
+                    // Also, print the gapDetection periodically
+                    System.out.println(((GapDetectionFilter)gapDetector).getDetector().toString());
+                }
             }
             // Should we throttle the output to lower the eps?
             throttle(statistics);

src/main/java/nu/sitia/loggenerator/filter/FilterListFactory.java

@@ -42,6 +42,7 @@ public List<ProcessFilter> create(Configuration config) {
         String doubleDetection = config.getValue("-dd");
         String jsonFilter = config.getValue("-jf");
         String jsonReport = config.getValue("-gdjr");
+        String select = config.getValue("-se");
 
         boolean dd = "true".equalsIgnoreCase(doubleDetection);
 
@@ -79,7 +80,9 @@ public List<ProcessFilter> create(Configuration config) {
             filterList.add(new GapDetectionFilter(gapRegex, dd,
                     "true".equalsIgnoreCase(jsonReport)));
         }
-
+        if (select != null) {
+            filterList.add(new SelectFilter(select));
+        }
         return filterList;
     }
 }

src/main/java/nu/sitia/loggenerator/filter/GapDetectionFilter.java

@@ -101,6 +101,14 @@ public void shutdown() {
         }
     }
 
+    /**
+     * Get the detector for this filter
+     * @return the GapDetector
+     */
+    public GapDetector getDetector() {
+        return detector;
+    }
+
     /**
      * toString in json format
      * @return the internal state

src/main/java/nu/sitia/loggenerator/filter/JsonFilter.java

@@ -20,7 +20,6 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ArrayNode;
 import nu.sitia.loggenerator.util.JsonUtil;
 
 import java.util.ArrayList;

src/main/java/nu/sitia/loggenerator/filter/SelectFilter.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2022 sitia.nu https://github.com/anders-wartoft/LogGenerator
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit
+ * persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+ *  WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package nu.sitia.loggenerator.filter;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class SelectFilter implements ProcessFilter {
+
+    /** Cached regex pattern */
+    private final Pattern pattern;
+
+    /** for toString() */
+    private final String regex;
+
+    /**
+     * Create a RegexFilter and set all parameters
+     * @param regex The value to search for
+     */
+    public SelectFilter(String regex) {
+        this.regex = regex;
+        // What to look for
+        if (null == regex) {
+            throw new RuntimeException("regex is null");
+        }
+        pattern = Pattern.compile(regex);
+    }
+
+    /**
+     * Filter one string
+     *
+     * @param toFilter The string to change
+     * @return toFilter with a header added before the string.
+     */
+    private String filter(String toFilter) {
+        Matcher matcher = pattern.matcher(toFilter);
+        if (matcher.find()) {
+            return matcher.group(1);
+        }
+        return toFilter;
+    }
+
+    @Override
+    public List<String> filter(List<String> toFilter) {
+        List<String> filtered = new ArrayList<>();
+        toFilter.forEach(s ->
+                filtered.add(filter(s)));
+
+        return filtered;
+    }
+
+    /**
+     * The current configuration
+     * @return A printout of the current configuration
+     */
+    @Override
+    public String toString() {
+        return "SelectFilter" + System.lineSeparator() + regex + System.lineSeparator();
+    }
+}

src/main/java/nu/sitia/loggenerator/util/LogStatistics.java

@@ -64,6 +64,9 @@ public long getTransactionMessages() {
         return transactionMessages;
     }
 
+    /** Flag to be able to signal to the ItemProxy if statistics has been printed */
+    private boolean hasPrinted = false;
+
     /**
      * Constructor
      * @param config Used to get the printout every ms value
@@ -83,9 +86,12 @@ public LogStatistics(Configuration config) {
      * Inspect the data that is about to be sent for connection
      * messages.
      * @param filtered The list of strings to send
+     * @return true iff the statistics have been printed out
      */
-    public void calculateStatistics(List<String> filtered) {
+    public boolean calculateStatistics(List<String> filtered) {
+        hasPrinted = false;
         filtered.forEach(this::checkMessage);
+        return hasPrinted;
     }
 
     /**
@@ -149,6 +155,7 @@ private void checkLog(String log) {
      * @param bytes How many characters the compound logs have contained (close enough to bytes received)
      */
     private void printMetrics(String beginning, long ms, long nrMessages, String name, long bytes) {
+        hasPrinted = true;
         System.out.format("%s: %s transferred %d lines in %d milliseconds, %.3f kEPS %.3f MBPS%n",
                 beginning,
                 name,

src/main/java/nu/sitia/loggenerator/util/gapdetector/GapDetector.java

@@ -159,19 +159,19 @@ private Map<Long, Long> getSorted(Map<Long, Long>toSort) {
     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();
+        if (gaps.size() > 0) {
+            sb.append("Gaps found: ").append(gaps.size()).append(".").append(System.lineSeparator());
+            for (Gap gap : gaps) {
+                sb.append(gap.getFrom()).append("-").append(gap.getTo()).append(System.lineSeparator());
+            }
+        }
         if (duplicateDetection) {
             sb.append("Duplicate detection found: ").append(duplicates.size()).append(" duplicate (or more) values.").append(System.lineSeparator());
             Map<Long, Long> sortedMap = getSorted(duplicates);
             sortedMap.forEach((key, value) -> sb.append(key).append(" - ").append(value).append(System.lineSeparator()));
         }
         sb.append("Number of unique received numbers: ").append(nrReceived).append(System.lineSeparator());
         sb.append("Next expected number: ").append(expectedNumber).append(System.lineSeparator());
-        if (gaps.size() > 0) {
-            sb.append("Gaps found: ").append(gaps.size()).append(".").append(System.lineSeparator());
-            for (Gap gap : gaps) {
-                sb.append(gap.getFrom()).append("-").append(gap.getTo()).append(System.lineSeparator());
-            }
-        }
         return sb.toString();
     }