Treat big changes in searchCount as significant and persist the document after such changes (#44413)

Author: przemekwitek

x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/DatafeedJobsIT.java

@@ -39,8 +39,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.apache.lucene.util.LuceneTestCase.AwaitsFix;
-
 import static org.elasticsearch.xpack.ml.support.BaseMlIntegTestCase.createDatafeed;
 import static org.elasticsearch.xpack.ml.support.BaseMlIntegTestCase.createDatafeedBuilder;
 import static org.elasticsearch.xpack.ml.support.BaseMlIntegTestCase.createScheduledJob;
@@ -102,7 +100,6 @@ public void testLookbackOnly() throws Exception {
         waitUntilJobIsClosed(job.getId());
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/44335")
     public void testDatafeedTimingStats_DatafeedRecreated() throws Exception {
         client().admin().indices().prepareCreate("data")
             .addMapping("type", "time", "type=date")

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedTimingStatsReporter.java

@@ -77,10 +77,20 @@ private void flush() {
      * Returns true if given stats objects differ from each other by more than 10% for at least one of the statistics.
      */
     public static boolean differSignificantly(DatafeedTimingStats stats1, DatafeedTimingStats stats2) {
-        return differSignificantly(stats1.getTotalSearchTimeMs(), stats2.getTotalSearchTimeMs())
+        return countsDifferSignificantly(stats1.getSearchCount(), stats2.getSearchCount())
+            || differSignificantly(stats1.getTotalSearchTimeMs(), stats2.getTotalSearchTimeMs())
             || differSignificantly(stats1.getAvgSearchTimePerBucketMs(), stats2.getAvgSearchTimePerBucketMs());
     }
 
+    /**
+     * Returns {@code true} if one of the ratios { value1 / value2, value2 / value1 } is smaller than MIN_VALID_RATIO.
+     * This can be interpreted as values { value1, value2 } differing significantly from each other.
+     */
+    private static boolean countsDifferSignificantly(long value1, long value2) {
+        return (((double) value2) / value1 < MIN_VALID_RATIO)
+            || (((double) value1) / value2 < MIN_VALID_RATIO);
+    }
+
     /**
      * Returns {@code true} if one of the ratios { value1 / value2, value2 / value1 } is smaller than MIN_VALID_RATIO or
      * the absolute difference |value1 - value2| is greater than MAX_VALID_ABS_DIFFERENCE_MS.

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/datafeed/DatafeedTimingStatsReporterTests.java

@@ -18,6 +18,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.verifyZeroInteractions;
 
@@ -44,28 +45,40 @@ public void testReportSearchDuration_Null() {
         verifyZeroInteractions(jobResultsPersister);
     }
 
+    public void testReportSearchDuration_Zero() {
+        DatafeedTimingStatsReporter timingStatsReporter =
+            new DatafeedTimingStatsReporter(new DatafeedTimingStats(JOB_ID), jobResultsPersister);
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 0, 0, 0.0)));
+
+        timingStatsReporter.reportSearchDuration(TimeValue.ZERO);
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 1, 0, 0.0)));
+
+        verify(jobResultsPersister).persistDatafeedTimingStats(new DatafeedTimingStats(JOB_ID, 1, 0, 0.0), RefreshPolicy.IMMEDIATE);
+        verifyNoMoreInteractions(jobResultsPersister);
+    }
+
     public void testReportSearchDuration() {
         DatafeedTimingStatsReporter timingStatsReporter =
-            new DatafeedTimingStatsReporter(new DatafeedTimingStats(JOB_ID, 3, 10, 10000.0), jobResultsPersister);
-        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 3, 10, 10000.0)));
+            new DatafeedTimingStatsReporter(new DatafeedTimingStats(JOB_ID, 13, 10, 10000.0), jobResultsPersister);
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 13, 10, 10000.0)));
 
         timingStatsReporter.reportSearchDuration(ONE_SECOND);
-        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 4, 10, 11000.0)));
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 14, 10, 11000.0)));
 
         timingStatsReporter.reportSearchDuration(ONE_SECOND);
-        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 5, 10, 12000.0)));
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 15, 10, 12000.0)));
 
         timingStatsReporter.reportSearchDuration(ONE_SECOND);
-        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 6, 10, 13000.0)));
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 16, 10, 13000.0)));
 
         timingStatsReporter.reportSearchDuration(ONE_SECOND);
-        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 7, 10, 14000.0)));
+        assertThat(timingStatsReporter.getCurrentTimingStats(), equalTo(new DatafeedTimingStats(JOB_ID, 17, 10, 14000.0)));
 
         InOrder inOrder = inOrder(jobResultsPersister);
         inOrder.verify(jobResultsPersister).persistDatafeedTimingStats(
-            new DatafeedTimingStats(JOB_ID, 5, 10, 12000.0), RefreshPolicy.IMMEDIATE);
+            new DatafeedTimingStats(JOB_ID, 15, 10, 12000.0), RefreshPolicy.IMMEDIATE);
         inOrder.verify(jobResultsPersister).persistDatafeedTimingStats(
-            new DatafeedTimingStats(JOB_ID, 7, 10, 14000.0), RefreshPolicy.IMMEDIATE);
+            new DatafeedTimingStats(JOB_ID, 17, 10, 14000.0), RefreshPolicy.IMMEDIATE);
         verifyNoMoreInteractions(jobResultsPersister);
     }
 
@@ -134,5 +147,9 @@ public void testTimingStatsDifferSignificantly() {
             DatafeedTimingStatsReporter.differSignificantly(
                 new DatafeedTimingStats(JOB_ID, 5, 10, 100000.0), new DatafeedTimingStats(JOB_ID, 5, 10, 110001.0)),
             is(true));
+        assertThat(
+            DatafeedTimingStatsReporter.differSignificantly(
+                new DatafeedTimingStats(JOB_ID, 5, 10, 100000.0), new DatafeedTimingStats(JOB_ID, 50, 10, 100000.0)),
+            is(true));
     }
 }