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 <n...@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()); + } }