metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nickal...@apache.org
Subject [metron] branch master updated: METRON-1778 Out-of-order timestamps may delay flush in Storm Profiler (nickwallen) closes apache/metron#1197
Date Tue, 26 Feb 2019 19:34:58 GMT
This is an automated email from the ASF dual-hosted git repository.

nickallen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/metron.git


The following commit(s) were added to refs/heads/master by this push:
     new fec36c7  METRON-1778 Out-of-order timestamps may delay flush in Storm Profiler (nickwallen)
closes apache/metron#1197
fec36c7 is described below

commit fec36c7ed0b2d29f4dc48574adf02c7fcb13e5fb
Author: nickwallen <nick@nickallen.org>
AuthorDate: Tue Feb 26 14:34:39 2019 -0500

    METRON-1778 Out-of-order timestamps may delay flush in Storm Profiler (nickwallen) closes
apache/metron#1197
---
 .../profiler/storm/FixedFrequencyFlushSignal.java  |  82 ++++++-----
 .../storm/FixedFrequencyFlushSignalTest.java       | 151 +++++++++++++++++++--
 2 files changed, 179 insertions(+), 54 deletions(-)

diff --git a/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignal.java
b/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignal.java
index 02503c2..db4f99b 100644
--- a/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignal.java
+++ b/metron-analytics/metron-profiler-storm/src/main/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignal.java
@@ -32,14 +32,14 @@ public class FixedFrequencyFlushSignal implements FlushSignal {
   protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
   /**
-   * The latest known timestamp.
+   * Tracks the min timestamp.
    */
-  private long currentTime;
+  private long minTime;
 
   /**
-   * The time when the next flush should occur.
+   * Tracks the max timestamp.
    */
-  private long flushTime;
+  private long maxTime;
 
   /**
    * The amount of time between flushes in milliseconds.
@@ -47,7 +47,6 @@ public class FixedFrequencyFlushSignal implements FlushSignal {
   private long flushFrequency;
 
   public FixedFrequencyFlushSignal(long flushFrequencyMillis) {
-
     if(flushFrequencyMillis < 0) {
       throw new IllegalArgumentException("flush frequency must be >= 0");
     }
@@ -61,9 +60,8 @@ public class FixedFrequencyFlushSignal implements FlushSignal {
    */
   @Override
   public void reset() {
-    flushTime = 0;
-    currentTime = 0;
-
+    minTime = Long.MAX_VALUE;
+    maxTime = Long.MIN_VALUE;
     LOG.debug("Flush counters reset");
   }
 
@@ -74,31 +72,34 @@ public class FixedFrequencyFlushSignal implements FlushSignal {
    */
   @Override
   public void update(long timestamp) {
+    if(LOG.isWarnEnabled()) {
+      checkIfOutOfOrder(timestamp);
+    }
 
-    if(timestamp > currentTime) {
-
-      // need to update current time
-      LOG.debug("Updating current time; last={}, new={}", currentTime, timestamp);
-      currentTime = timestamp;
-
-    } else if ((currentTime - timestamp) > flushFrequency) {
-
-      // significantly out-of-order timestamps
-      LOG.warn("Timestamps out-of-order by '{}' ms. This may indicate a problem in the data.
last={}, current={}",
-              (currentTime - timestamp),
-              timestamp,
-              currentTime);
+    if(timestamp < minTime) {
+      minTime = timestamp;
     }
 
-    if(flushTime == 0) {
+    if(timestamp > maxTime) {
+      maxTime = timestamp;
+    }
+  }
 
-      // set the next time to flush
-      flushTime = currentTime + flushFrequency;
-      LOG.debug("Setting flush time; '{}' ms until flush; flushTime={}, currentTime={}, flushFreq={}",
-              timeToNextFlush(),
-              flushTime,
-              currentTime,
-              flushFrequency);
+  /**
+   * Checks if the timestamp is significantly out-of-order.
+   *
+   * @param timestamp The last timestamp.
+   */
+  private void checkIfOutOfOrder(long timestamp) {
+    // do not warn if this is the first timestamp we've seen, which will always be 'out-of-order'
+    if (maxTime > Long.MIN_VALUE) {
+
+      long outOfOrderBy = maxTime - timestamp;
+      if (Math.abs(outOfOrderBy) > flushFrequency) {
+        LOG.warn("Timestamp out-of-order by {} ms. This may indicate a problem in the data.
" +
+                        "timestamp={}, maxKnown={}, flushFreq={} ms",
+                outOfOrderBy, timestamp, maxTime, flushFrequency);
+      }
     }
   }
 
@@ -109,27 +110,20 @@ public class FixedFrequencyFlushSignal implements FlushSignal {
    */
   @Override
   public boolean isTimeToFlush() {
+    boolean flush = false;
 
-    boolean flush = currentTime > flushTime;
-    LOG.debug("Flush={}, '{}' ms until flush; currentTime={}, flushTime={}",
-            flush,
-            timeToNextFlush(),
-            currentTime,
-            flushTime);
+    long flushTime = minTime + flushFrequency;
+    if(maxTime >= flushTime) {
+      flush = true;
+    }
 
+    LOG.debug("'{}' ms until flush; flush?={}, minTime={}, maxTime={}, flushTime={}",
+            Math.max(0, flushTime - maxTime), flush, minTime, maxTime, flushTime);
     return flush;
   }
 
   @Override
   public long currentTimeMillis() {
-    return currentTime;
-  }
-
-  /**
-   * Returns the number of milliseconds to the next flush.
-   * @return The time left until the next flush.
-   */
-  private long timeToNextFlush() {
-    return Math.max(0, flushTime - currentTime);
+    return maxTime;
   }
 }
diff --git a/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignalTest.java
b/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignalTest.java
index 8b8813b..a753dc4 100644
--- a/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignalTest.java
+++ b/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/FixedFrequencyFlushSignalTest.java
@@ -31,41 +31,172 @@ public class FixedFrequencyFlushSignalTest {
 
   @Test
   public void testSignalFlush() {
+    int flushFreq = 1000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
 
-    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(1000);
+    // not ready to flush; we have not seen any messages yet
+    assertFalse(signal.isTimeToFlush());
 
-    // not time to flush yet
+    // not ready to flush; flushTime = min + flushFreq = 5000 + 1000 = 6000; flush if anything
>= flushTime
+    signal.update(5000);
     assertFalse(signal.isTimeToFlush());
 
-    // advance time
+    // ready to flush; flushTime = min + flushFreq = 5000 + 1000 = 6000; max(5000,7000) >=
6000
+    signal.update(7000);
+    assertTrue(signal.isTimeToFlush());
+  }
+
+  @Test
+  public void testOutOfOrderTimestamps() {
+    int flushFreq = 5000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
+
+    // not ready to flush; flushTime = min + flushFreq = 5000 + 5000 = 10000; flush if anything
>= flushTime
     signal.update(5000);
+    assertFalse(signal.isTimeToFlush());
 
-    // not time to flush yet
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 5000 = 6000
+    signal.update(1000);
     assertFalse(signal.isTimeToFlush());
 
-    // advance time
+    // ready to flush; flushTime = min + flushFreq = 1000 + 5000 = 6000; max(5000,1000,7000)
>= 6000
     signal.update(7000);
+    assertTrue(signal.isTimeToFlush());
+
+    // ready to flush, still
+    signal.update(3000);
+    assertTrue(signal.isTimeToFlush());
 
-    // time to flush
+    // ready to flush, still
     assertTrue(signal.isTimeToFlush());
   }
 
   @Test
-  public void testOutOfOrderTimestamps() {
-    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(1000);
+  public void testOutOfOrderTimestampsNoFlush() {
+    int flushFreq = 7000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
 
-    // advance time, out-of-order
+    // not ready to flush; flushTime = min + flushFreq = 5000 + 7000 = 12000; flush if anything
>= flushTime
     signal.update(5000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 7000 = 8000
     signal.update(1000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 7000 = 8000
     signal.update(7000);
+    assertFalse(signal.isTimeToFlush());
+
+    // ready to flush; flushTime = min + flushFreq = 1000 + 7000 = 8000; max(5000,1000,7000,3000)
>= 8000
+    signal.update(3000);
+    assertFalse(signal.isTimeToFlush());
+  }
+  
+  @Test
+  public void testTimestampsDescending() {
+    int flushFreq = 3000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
+
+    // not ready to flush; flushTime = min + flushFreq = 4100 + 3000 = 7100; flush if anything
>= flushTime
+    signal.update(4100);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 3000 + 3000 = 6000
     signal.update(3000);
+    assertFalse(signal.isTimeToFlush());
 
-    // need to flush @ 5000 + 1000 = 6000. if anything > 6000 (even out-of-order), then
it should signal a flush
+    // not ready to flush; flushTime = min + flushFreq = 2000 + 3000 = 5000
+    signal.update(2000);
+    assertFalse(signal.isTimeToFlush());
+
+    // ready to flush; flushTime = min + flushFreq = 1000 + 3000 = 4000; max(4100,3000,2000,1000)
>= 4000
+    signal.update(1000);
     assertTrue(signal.isTimeToFlush());
   }
 
+  @Test
+  public void testTimestampsDescendingNoFlush() {
+    int flushFreq = 4000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
+
+    // not ready to flush; flushTime = min + flushFreq = 4000 + 4000 = 8000; flush if anything
>= flushTime
+    signal.update(4000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 3000 + 4000 = 7000
+    signal.update(3000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 2000 + 4000 = 6000
+    signal.update(2000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 4000 = 5000
+    signal.update(1000);
+    assertFalse(signal.isTimeToFlush());
+  }
+
+  @Test
+  public void testTimestampsAscending() {
+    int flushFreq = 3000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 3000 = 4000; flush if anything
>= flushTime
+    signal.update(1000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 3000 = 4000
+    signal.update(2000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 3000 = 4000
+    signal.update(3000);
+    assertFalse(signal.isTimeToFlush());
+
+    // ready to flush; flushTime = min + flushFreq = 1000 + 3000 = 4000; max(1000,2000,3000,4000)
>= 4000
+    signal.update(4000);
+  }
+
+  @Test
+  public void testTimestampsAscendingNoFlush() {
+    int flushFreq = 4000;
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(flushFreq);
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 4000 = 5000; flush if anything
>= flushTime
+    signal.update(1000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 4000 = 5000
+    signal.update(2000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 4000 = 5000
+    signal.update(3000);
+    assertFalse(signal.isTimeToFlush());
+
+    // not ready to flush; flushTime = min + flushFreq = 1000 + 4000 = 5000
+    signal.update(4000);
+    assertFalse(signal.isTimeToFlush());
+  }
+
   @Test(expected = IllegalArgumentException.class)
   public void testNegativeFrequency() {
+    // a negative flush frequency makes no sense
     new FixedFrequencyFlushSignal(-1000);
   }
+
+  @Test
+  public void testReset() {
+    FixedFrequencyFlushSignal signal = new FixedFrequencyFlushSignal(4000);
+    signal.update(1000);
+    signal.update(6000);
+
+    // ready to flush; flushTime = min + flushFreq = 1000 + 4000 = 5000; max(1000,6000) >=
5000
+    assertTrue(signal.isTimeToFlush());
+
+    // reset should turn off the flush signal
+    signal.reset();
+    assertFalse(signal.isTimeToFlush());
+  }
 }


Mime
View raw message