[ML] Improve autodetect logic for persistence

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;