[ML] Improve autodetect logic for persistence (#437)

Changed the logic surrounding persistence of both state and quantiles on
graceful shutdown so that persistence only occurs if and only if at
least one input record has been processed or time has been advanced.

closes #393

Author: edsavage

docs/CHANGELOG.asciidoc

@@ -44,6 +44,8 @@ to the model. (See {pull}214[#214].)
 
 * Handle NaNs when detrending seasonal components. {ml-pull}408[#408]
 
+* Improve autodetect logic for persistence. {ml-pull}437[#437]
+
 == {es} version 7.0.0-alpha1
 
 == {es} version 6.7.0

include/api/CAnomalyJob.h

@@ -182,6 +182,9 @@ class API_EXPORT CAnomalyJob : public CDataProcessor {
     //! How many records did we handle?
     virtual uint64_t numRecordsHandled() const;
 
+    //! Is persistence needed?
+    virtual bool isPersistenceNeeded(const std::string& description) const;
+
     //! Log a list of the detectors and keys
     void description() const;
 
@@ -454,6 +457,9 @@ class API_EXPORT CAnomalyJob : public CDataProcessor {
     //! The hierarchical results normalizer.
     model::CHierarchicalResultsNormalizer m_Normalizer;
 
+    //! Flag indicating whether or not time has been advanced.
+    bool m_TimeAdvanced{false};
+
     friend class ::CBackgroundPersisterTest;
     friend class ::CAnomalyJobTest;
 };

include/api/CDataProcessor.h

@@ -82,6 +82,9 @@ class API_EXPORT CDataProcessor : private core::CNonCopyable {
     //! Access the output handler
     virtual COutputHandler& outputHandler() = 0;
 
+    //! Is persistence needed?
+    virtual bool isPersistenceNeeded(const std::string& description) const = 0;
+
     //! Create debug for a record.  This is expensive so should NOT be
     //! called for every record as a matter of course.
     static std::string debugPrintRecord(const TStrStrUMap& dataRowFields);

include/api/CFieldDataTyper.h

@@ -98,6 +98,9 @@ class API_EXPORT CFieldDataTyper : public CDataProcessor {
     virtual bool restoreState(core::CDataSearcher& restoreSearcher,
                               core_t::TTime& completeToTime);
 
+    //! Is persistence needed?
+    virtual bool isPersistenceNeeded(const std::string& description) const;
+
     //! Persist current state
     virtual bool persistState(core::CDataAdder& persister);
 

include/api/COutputChainer.h

@@ -78,6 +78,9 @@ class API_EXPORT COutputChainer : public COutputHandler {
     //! Persist current state due to the periodic persistence being triggered.
     virtual bool periodicPersistState(CBackgroundPersister& persister);
 
+    //! Is persistence needed?
+    virtual bool isPersistenceNeeded(const std::string& description) const;
+
     //! The chainer does consume control messages, because it passes them on
     //! to whatever processor it's chained to.
     virtual bool consumesControlMessages();

include/api/COutputHandler.h

@@ -91,6 +91,9 @@ class API_EXPORT COutputHandler : private core::CNonCopyable {
     //! Persist current state due to the periodic persistence being triggered.
     virtual bool periodicPersistState(CBackgroundPersister& persister);
 
+    //! Is persistence needed?
+    virtual bool isPersistenceNeeded(const std::string& description) const;
+
     //! Does this handler deal with control messages?
     virtual bool consumesControlMessages();
 

include/config/CAutoconfigurer.h

@@ -46,6 +46,9 @@ class CONFIG_EXPORT CAutoconfigurer : public api::CDataProcessor {
     //! Generate the report.
     virtual void finalise();
 
+    //! Is persistence needed?
+    virtual bool isPersistenceNeeded(const std::string& description) const;
+
     //! No-op.
     virtual bool restoreState(core::CDataSearcher& restoreSearcher,
                               core_t::TTime& completeToTime);

lib/api/CAnomalyJob.cc

@@ -220,8 +220,10 @@ bool CAnomalyJob::handleRecord(const TStrStrUMap& dataRowFields) {
 }
 
 void CAnomalyJob::finalise() {
-    // Persist final state of normalizer
-    m_JsonOutputWriter.persistNormalizer(m_Normalizer, m_LastNormalizerPersistTime);
+    // Persist final state of normalizer iff an input record has been handled or time has been advanced.
+    if (this->isPersistenceNeeded("quantiles state")) {
+        m_JsonOutputWriter.persistNormalizer(m_Normalizer, m_LastNormalizerPersistTime);
+    }
 
     // Prune the models so that the final persisted state is as neat as possible
     this->pruneAllModels();
@@ -396,11 +398,23 @@ void CAnomalyJob::advanceTime(const std::string& time_) {
         LOG_TRACE(<< "Received request to advance time to " << time);
     }
 
+    m_TimeAdvanced = true;
+
     this->outputBucketResultsUntil(time);
 
     this->timeNow(time);
 }
 
+bool CAnomalyJob::isPersistenceNeeded(const std::string& description) const {
+    if ((m_NumRecordsHandled == 0) && (m_TimeAdvanced == false)) {
+        LOG_DEBUG(<< "Will not attempt to persist " << description
+                  << ". Zero records were handled and time has not been advanced.");
+        return false;
+    }
+
+    return true;
+}
+
 void CAnomalyJob::outputBucketResultsUntil(core_t::TTime time) {
     // If the bucket time has increased, output results for all field names
     core_t::TTime bucketLength = m_ModelConfig.bucketLength();

lib/api/CCmdSkeleton.cc

@@ -56,8 +56,7 @@ bool CCmdSkeleton::persistState() {
         return true;
     }
 
-    if (m_Processor.numRecordsHandled() == 0) {
-        LOG_DEBUG(<< "Zero records were handled - will not attempt to persist state");
+    if (m_Processor.isPersistenceNeeded("state") == false) {
         return true;
     }
 

lib/api/CFieldDataTyper.cc

@@ -335,6 +335,20 @@ bool CFieldDataTyper::persistState(core::CDataAdder& persister) {
     return this->doPersistState(m_DataTyper->makePersistFunc(), m_ExamplesCollector, persister);
 }
 
+bool CFieldDataTyper::isPersistenceNeeded(const std::string& description) const {
+    // Pass on the request in case we're chained
+    if (m_OutputHandler.isPersistenceNeeded(description)) {
+        return true;
+    }
+
+    if (m_NumRecordsHandled == 0) {
+        LOG_DEBUG(<< "Zero records were handled - will not attempt to persist "
+                  << description << ".");
+        return false;
+    }
+    return true;
+}
+
 bool CFieldDataTyper::doPersistState(const CDataTyper::TPersistFunc& dataTyperPersistFunc,
                                      const CCategoryExamplesCollector& examplesCollector,
                                      core::CDataAdder& persister) {

lib/api/COutputChainer.cc

@@ -113,6 +113,10 @@ bool COutputChainer::periodicPersistState(CBackgroundPersister& persister) {
     return m_DataProcessor.periodicPersistState(persister);
 }
 
+bool COutputChainer::isPersistenceNeeded(const std::string& description) const {
+    return m_DataProcessor.isPersistenceNeeded(description);
+}
+
 bool COutputChainer::consumesControlMessages() {
     return true;
 }

lib/api/COutputHandler.cc

@@ -46,6 +46,11 @@ bool COutputHandler::periodicPersistState(CBackgroundPersister& /* persister */)
     return true;
 }
 
+bool COutputHandler::isPersistenceNeeded(const std::string& /*description*/) const {
+    // NOOP unless overridden
+    return false;
+}
+
 COutputHandler::CPreComputedHash::CPreComputedHash(size_t hash) : m_Hash(hash) {
 }
 

lib/api/unittest/CAnomalyJobTest.cc

@@ -483,6 +483,109 @@ void CAnomalyJobTest::testSkipTimeControlMessage() {
     CPPUNIT_ASSERT_EQUAL(std::size_t(11), countBuckets("bucket", outputStrm.str() + "]"));
 }
 
+void CAnomalyJobTest::testIsPersistenceNeeded() {
+
+    model::CLimits limits;
+    api::CFieldConfig fieldConfig;
+    api::CFieldConfig::TStrVec clauses;
+    clauses.push_back("count");
+    fieldConfig.initFromClause(clauses);
+    model::CAnomalyDetectorModelConfig modelConfig =
+        model::CAnomalyDetectorModelConfig::defaultConfig(BUCKET_SIZE);
+
+    {
+        // check that persistence is not needed if no input records have been handled
+        // and the time has not been advanced
+
+        std::stringstream outputStrm;
+        core::CJsonOutputStreamWrapper wrappedOutputStream(outputStrm);
+
+        api::CAnomalyJob job("job", limits, fieldConfig, modelConfig, wrappedOutputStream);
+
+        CPPUNIT_ASSERT_EQUAL(false, job.isPersistenceNeeded("test state"));
+
+        job.finalise();
+        wrappedOutputStream.syncFlush();
+
+        std::string output = outputStrm.str();
+        LOG_DEBUG(<< "Output has yielded: " << output);
+
+        // check that no quantile state was persisted
+        core::CRegex regex;
+        regex.init("\n");
+        core::CRegex::TStrVec lines;
+        regex.split(output, lines);
+        CPPUNIT_ASSERT_EQUAL(false, findLine("\"quantiles\":{\"job_id\":\"job\",\"quantile_state\".*",
+                                             lines));
+    }
+
+    core_t::TTime time = 3600;
+    {
+        // check that persistence is needed if an input record has been handled
+
+        std::stringstream outputStrm;
+        core::CJsonOutputStreamWrapper wrappedOutputStream(outputStrm);
+
+        api::CAnomalyJob job("job", limits, fieldConfig, modelConfig, wrappedOutputStream);
+
+        api::CAnomalyJob::TStrStrUMap dataRows;
+
+        std::ostringstream ss;
+        ss << time;
+        dataRows["time"] = ss.str();
+        CPPUNIT_ASSERT(job.handleRecord(dataRows));
+
+        CPPUNIT_ASSERT_EQUAL(true, job.isPersistenceNeeded("test state"));
+
+        job.finalise();
+        wrappedOutputStream.syncFlush();
+
+        std::string output = outputStrm.str();
+        LOG_DEBUG(<< "Output has yielded: " << output);
+
+        // check that the quantile state has actually been persisted
+        core::CRegex regex;
+        regex.init("\n");
+        core::CRegex::TStrVec lines;
+        regex.split(output, lines);
+        CPPUNIT_ASSERT_EQUAL(true, findLine("\"quantiles\":{\"job_id\":\"job\",\"quantile_state\".*",
+                                            lines));
+    }
+
+    {
+        // check that persistence is needed if time has been advanced (via a control message)
+        // even if no input data has been handled
+
+        std::stringstream outputStrm;
+        core::CJsonOutputStreamWrapper wrappedOutputStream(outputStrm);
+
+        api::CAnomalyJob job("job", limits, fieldConfig, modelConfig, wrappedOutputStream);
+
+        api::CAnomalyJob::TStrStrUMap dataRows;
+
+        time = 39600;
+        dataRows["."] = "t39600";
+        CPPUNIT_ASSERT(job.handleRecord(dataRows));
+        CPPUNIT_ASSERT(job.isPersistenceNeeded("test state"));
+
+        CPPUNIT_ASSERT_EQUAL(true, job.isPersistenceNeeded("test state"));
+
+        job.finalise();
+        wrappedOutputStream.syncFlush();
+
+        std::string output = outputStrm.str();
+        LOG_DEBUG(<< "Output has yielded: " << output);
+
+        // check that the quantile state has actually been persisted
+        core::CRegex regex;
+        regex.init("\n");
+        core::CRegex::TStrVec lines;
+        regex.split(output, lines);
+        CPPUNIT_ASSERT_EQUAL(true, findLine("\"quantiles\":{\"job_id\":\"job\",\"quantile_state\".*",
+                                            lines));
+    }
+}
+
 void CAnomalyJobTest::testModelPlot() {
     core_t::TTime bucketSize = 10000;
     model::CLimits limits;
@@ -651,6 +754,8 @@ CppUnit::Test* CAnomalyJobTest::suite() {
     suiteOfTests->addTest(new CppUnit::TestCaller<CAnomalyJobTest>(
         "CAnomalyJobTest::testSkipTimeControlMessage",
         &CAnomalyJobTest::testSkipTimeControlMessage));
+    suiteOfTests->addTest(new CppUnit::TestCaller<CAnomalyJobTest>(
+        "CAnomalyJobTest::testIsPersistenceNeeded", &CAnomalyJobTest::testIsPersistenceNeeded));
     suiteOfTests->addTest(new CppUnit::TestCaller<CAnomalyJobTest>(
         "CAnomalyJobTest::testModelPlot", &CAnomalyJobTest::testModelPlot));
     suiteOfTests->addTest(new CppUnit::TestCaller<CAnomalyJobTest>(

lib/api/unittest/CAnomalyJobTest.h

@@ -17,6 +17,7 @@ class CAnomalyJobTest : public CppUnit::TestFixture {
     void testOutOfSequence();
     void testControlMessages();
     void testSkipTimeControlMessage();
+    void testIsPersistenceNeeded();
     void testModelPlot();
     void testInterimResultEdgeCases();
     void testRestoreFailsWithEmptyStream();

lib/api/unittest/CMockDataProcessor.cc

@@ -47,6 +47,15 @@ bool CMockDataProcessor::handleRecord(const TStrStrUMap& dataRowFields) {
 void CMockDataProcessor::finalise() {
 }
 
+bool CMockDataProcessor::isPersistenceNeeded(const std::string& description) const {
+    if (m_NumRecordsHandled == 0) {
+        LOG_DEBUG(<< "Zero records were handled - will not attempt to persist "
+                  << description << ".");
+        return false;
+    }
+    return true;
+}
+
 bool CMockDataProcessor::restoreState(ml::core::CDataSearcher& restoreSearcher,
                                       ml::core_t::TTime& completeToTime) {
     // Pass on the request in case we're chained

lib/api/unittest/CMockDataProcessor.h

@@ -40,6 +40,8 @@ class CMockDataProcessor : public ml::api::CDataProcessor {
 
     virtual void finalise();
 
+    virtual bool isPersistenceNeeded(const std::string& description) const;
+
     //! Restore previously saved state
     virtual bool restoreState(ml::core::CDataSearcher& restoreSearcher,
                               ml::core_t::TTime& completeToTime);

lib/config/CAutoconfigurer.cc

@@ -165,6 +165,10 @@ void CAutoconfigurer::finalise() {
     m_Impl->finalise();
 }
 
+bool CAutoconfigurer::isPersistenceNeeded(const std::string& /*description*/) const {
+    return false;
+}
+
 bool CAutoconfigurer::restoreState(core::CDataSearcher& /*restoreSearcher*/,
                                    core_t::TTime& /*completeToTime*/) {
     return true;