[ML] Add internal flag to flush api determining whether to refresh indices or not. (#2532)

When datafeeds send flush requests they don't require the indices to be refreshed.

Round trip the "should_refresh" flag from ES to ml-cpp code and back again.

Author: edsavage

include/api/CAnomalyJob.h

@@ -504,6 +504,9 @@ class API_EXPORT CAnomalyJob : public CDataProcessor {
     //! Flag indicating whether or not time has been advanced.
     bool m_TimeAdvanced{false};
 
+    //! Flag indicating whether or not a flush control message should trigger a refresh of the datafeed
+    bool m_RefreshRequired{true};
+
     //! Introduced in version 8.6
     //! The initial value of the end time of the last bucket
     //! out of latency window we've seen, i.e. this member records

include/api/CJsonOutputWriter.h

@@ -244,8 +244,10 @@ class API_EXPORT CJsonOutputWriter {
                                const model::SCategorizerStats& categorizerStats,
                                const TOptionalTime& timestamp);
 
-    //! Acknowledge a flush request by echoing back the flush ID
-    void acknowledgeFlush(const std::string& flushId, core_t::TTime lastFinalizedBucketEnd);
+    //! Acknowledge a flush request by echoing back the flush ID and the "refreshRequired" flag
+    void acknowledgeFlush(const std::string& flushId,
+                          core_t::TTime lastFinalizedBucketEnd,
+                          bool refreshRequired = true);
 
     //! Write a category definition
     void writeCategoryDefinition(const std::string& partitionFieldName,

lib/api/CAnomalyJob.cc

@@ -323,6 +323,7 @@ bool CAnomalyJob::handleControlMessage(const std::string& controlMessage) {
         return false;
     }
 
+    bool refreshRequired{true};
     switch (controlMessage[0]) {
     case ' ':
         // Spaces are just used to fill the buffers and force prior messages
@@ -358,6 +359,17 @@ bool CAnomalyJob::handleControlMessage(const std::string& controlMessage) {
         break;
     case 'w':
         this->processPersistControlMessage(controlMessage.substr(1));
+        break;
+    case 'z':
+        LOG_TRACE(<< "Received control message '" << controlMessage << "'");
+        // "refreshRequired" parameter comes after the initial z.
+        if (core::CStringUtils::stringToType(controlMessage.substr(1), refreshRequired) == false) {
+            LOG_ERROR(<< "Received request to flush with invalid control message '"
+                      << controlMessage << "'");
+        } else {
+            m_RefreshRequired = refreshRequired;
+        }
+
         break;
     default:
         LOG_WARN(<< "Ignoring unknown control message of length "
@@ -454,7 +466,7 @@ void CAnomalyJob::acknowledgeFlush(const std::string& flushId) {
     } else {
         LOG_TRACE(<< "Received flush control message with ID " << flushId);
     }
-    m_JsonOutputWriter.acknowledgeFlush(flushId, m_LastFinalisedBucketEndTime);
+    m_JsonOutputWriter.acknowledgeFlush(flushId, m_LastFinalisedBucketEndTime, m_RefreshRequired);
 }
 
 void CAnomalyJob::updateConfig(const std::string& config) {

lib/api/CJsonOutputWriter.cc

@@ -67,6 +67,7 @@ const std::string INFLUENCERS("influencers");
 const std::string FLUSH("flush");
 const std::string ID("id");
 const std::string LAST_FINALIZED_BUCKET_END("last_finalized_bucket_end");
+const std::string REFRESH_REQUIRED("refresh_required");
 const std::string CATEGORY_ID("category_id");
 const std::string CATEGORY_DEFINITION("category_definition");
 const std::string TERMS("terms");
@@ -874,7 +875,8 @@ void CJsonOutputWriter::writeCategorizerStats(const std::string& partitionFieldN
 }
 
 void CJsonOutputWriter::acknowledgeFlush(const std::string& flushId,
-                                         core_t::TTime lastFinalizedBucketEnd) {
+                                         core_t::TTime lastFinalizedBucketEnd,
+                                         bool refreshRequired) {
     m_Writer.StartObject();
     m_Writer.Key(FLUSH);
     m_Writer.StartObject();
@@ -883,6 +885,8 @@ void CJsonOutputWriter::acknowledgeFlush(const std::string& flushId,
     m_Writer.String(flushId);
     m_Writer.Key(LAST_FINALIZED_BUCKET_END);
     m_Writer.Time(lastFinalizedBucketEnd);
+    m_Writer.Key(REFRESH_REQUIRED);
+    m_Writer.Bool(refreshRequired);
 
     m_Writer.EndObject();
     m_Writer.EndObject();

lib/api/unittest/CFieldDataCategorizerTest.cc

@@ -573,7 +573,7 @@ BOOST_AUTO_TEST_CASE(testHandleControlMessages) {
 
     const std::string& output{outputStrm.str()};
     LOG_DEBUG(<< "Output is: " << output);
-    BOOST_REQUIRE_EQUAL(0, output.find("[{\"flush\":{\"id\":\"7\",\"last_finalized_bucket_end\":0}}"));
+    BOOST_REQUIRE_EQUAL(0, output.find("[{\"flush\":{\"id\":\"7\",\"last_finalized_bucket_end\":0,\"refresh_required\":true}}"));
 }
 
 BOOST_AUTO_TEST_CASE(testRestoreStateFailsWithEmptyState) {