Fail start up if state is missing (#4)

* Fail start up if state is missing

* Fail categorizer restore if state is missing

Author: davidkyle

lib/api/CAnomalyJob.cc

@@ -795,7 +795,7 @@ void CAnomalyJob::writeOutResults(bool interim, model::CHierarchicalResults &res
         typedef ml::core::CScopedRapidJsonPoolAllocator<CJsonOutputWriter> TScopedAllocator;
         static const std::string ALLOCATOR_ID("CAnomalyJob::writeOutResults");
         TScopedAllocator scopedAllocator(ALLOCATOR_ID, m_JsonOutputWriter);
-        
+
         api::CHierarchicalResultsWriter writer(m_Limits, m_ModelConfig,
                                                 boost::bind(&CJsonOutputWriter::acceptResult,
                                                             &m_JsonOutputWriter,
@@ -888,8 +888,9 @@ bool CAnomalyJob::restoreState(core::CDataSearcher &restoreSearcher,
 
         if (strm->fail())
         {
-            // This is not fatal - we just didn't find the given document number
-            return true;
+            // This is fatal. If the stream exists and has failed then state is missing
+            LOG_ERROR("State restoration search returned failed stream");
+            return false;
         }
 
         // We're dealing with streaming JSON state
@@ -960,9 +961,8 @@ bool CAnomalyJob::restoreState(core::CStateRestoreTraverser &traverser,
     if (traverser.isEof())
     {
         m_RestoredStateDetail.s_RestoredStateStatus = E_NoDetectorsRecovered;
-        LOG_DEBUG("No data store results - assuming no state exists");
-        // This is not an error if no data has been persisted
-        return true;
+        LOG_ERROR("Expected persisted state but no state exists");
+        return false;
     }
 
     core_t::TTime lastBucketEndTime(0);
@@ -996,7 +996,7 @@ bool CAnomalyJob::restoreState(core::CStateRestoreTraverser &traverser,
     if (stateVersion != model::CAnomalyDetector::STATE_VERSION)
     {
         m_RestoredStateDetail.s_RestoredStateStatus = E_IncorrectVersion;
-        LOG_INFO("Restored anomaly detector state version is " << stateVersion <<
+        LOG_ERROR("Restored anomaly detector state version is " << stateVersion <<
                  " - ignoring it as current state version is " <<
                  model::CAnomalyDetector::STATE_VERSION);
 

lib/api/CFieldDataTyper.cc

@@ -273,14 +273,15 @@ bool CFieldDataTyper::restoreState(core::CDataSearcher &restoreSearcher,
 
         if (strm->bad())
         {
-            LOG_ERROR("State restoration search returned bad stream");
+            LOG_ERROR("Categorizer state restoration returned a bad stream");
             return false;
         }
 
         if (strm->fail())
         {
-            // This is not fatal - we just didn't find the given document number
-            return true;
+            // This is fatal. If the stream exists and has failed then state is missing
+            LOG_ERROR("Categorizer state restoration returned a failed stream");
+            return false;
         }
 
         // We're dealing with streaming JSON state
@@ -308,7 +309,14 @@ bool CFieldDataTyper::restoreState(core::CDataSearcher &restoreSearcher,
 
 bool CFieldDataTyper::acceptRestoreTraverser(core::CStateRestoreTraverser &traverser)
 {
-    if (traverser.name() == VERSION_TAG)
+    const std::string &firstFieldName = traverser.name();
+    if (traverser.isEof())
+    {
+        LOG_ERROR("Expected categorizer persisted state but no state exists");
+        return false;
+    }
+
+    if (firstFieldName == VERSION_TAG)
     {
         std::string version;
         if (core::CStringUtils::stringToType(traverser.value(), version) == false)

lib/api/unittest/CAnomalyJobTest.cc

@@ -39,6 +39,22 @@
 namespace
 {
 
+//! \brief
+//! Mock object for state restore unit tests.
+//!
+//! DESCRIPTION:\n
+//! CDataSearcher that returns an empty stream.
+//!
+class CEmptySearcher : public ml::core::CDataSearcher
+{
+    public:
+        //! Do a search that results in an empty input stream.
+        virtual TIStreamP search(size_t /*currentDocNum*/, size_t /*limit*/)
+        {
+            return TIStreamP(new std::istringstream());
+        }
+};
+
 //! \brief
 //! Mock object for unit tests
 //!
@@ -1878,6 +1894,27 @@ void CAnomalyJobTest::testInterimResultEdgeCases(void)
     std::remove(logFile);
 }
 
+void CAnomalyJobTest::testRestoreFailsWithEmptyStream(void)
+{
+    model::CLimits limits;
+    api::CFieldConfig fieldConfig;
+    api::CFieldConfig::TStrVec clauses;
+    clauses.push_back("value");
+    clauses.push_back("partitionfield=greenhouse");
+    fieldConfig.initFromClause(clauses);
+    model::CAnomalyDetectorModelConfig modelConfig =
+        model::CAnomalyDetectorModelConfig::defaultConfig(BUCKET_SIZE);
+    std::ostringstream outputStrm;
+    core::CJsonOutputStreamWrapper wrappedOutputStream(outputStrm);
+
+    api::CAnomalyJob job("job", limits, fieldConfig, modelConfig,
+            wrappedOutputStream);
+
+    core_t::TTime completeToTime(0);
+    CEmptySearcher restoreSearcher;
+    CPPUNIT_ASSERT(job.restoreState(restoreSearcher, completeToTime) == false);
+}
+
 CppUnit::Test* CAnomalyJobTest::suite(void)
 {
     CppUnit::TestSuite *suiteOfTests = new CppUnit::TestSuite("CAnomalyJobTest");
@@ -1906,5 +1943,8 @@ CppUnit::Test* CAnomalyJobTest::suite(void)
     suiteOfTests->addTest( new CppUnit::TestCaller<CAnomalyJobTest>(
                                    "CAnomalyJobTest::testInterimResultEdgeCases",
                                    &CAnomalyJobTest::testInterimResultEdgeCases) );
+    suiteOfTests->addTest( new CppUnit::TestCaller<CAnomalyJobTest>(
+                                   "CAnomalyJobTest::testRestoreFailsWithEmptyStream",
+                                   &CAnomalyJobTest::testRestoreFailsWithEmptyStream) );
     return suiteOfTests;
 }

lib/api/unittest/CAnomalyJobTest.h

@@ -31,6 +31,7 @@ class CAnomalyJobTest : public CppUnit::TestFixture
         void testBucketSelection(void);
         void testModelPlot(void);
         void testInterimResultEdgeCases(void);
+        void testRestoreFailsWithEmptyStream(void);
 
         static CppUnit::Test *suite(void);
 

lib/api/unittest/CFieldDataTyperTest.cc

@@ -38,6 +38,22 @@ using namespace api;
 namespace
 {
 
+//! \brief
+//! Mock object for state restore unit tests.
+//!
+//! DESCRIPTION:\n
+//! CDataSearcher that returns an empty stream.
+//!
+class CEmptySearcher : public ml::core::CDataSearcher
+{
+    public:
+        //! Do a search that results in an empty input stream.
+        virtual TIStreamP search(size_t /*currentDocNum*/, size_t /*limit*/)
+        {
+            return TIStreamP(new std::istringstream());
+        }
+};
+
 class CTestOutputHandler : public COutputHandler
 {
     public:
@@ -325,6 +341,23 @@ void CFieldDataTyperTest::testHandleControlMessages(void)
                          output.find("[{\"flush\":{\"id\":\"7\",\"last_finalized_bucket_end\":0}}"));
 }
 
+void CFieldDataTyperTest::testRestoreStateFailsWithEmptyState(void)
+{
+    model::CLimits limits;
+    CFieldConfig config;
+    CPPUNIT_ASSERT(config.initFromFile("testfiles/new_persist_categorization.conf"));
+
+    std::ostringstream outputStrm;
+    CNullOutput nullOutput;
+    core::CJsonOutputStreamWrapper wrappedOutputStream(outputStrm);
+    CJsonOutputWriter writer("job", wrappedOutputStream);
+    CFieldDataTyper typer("job", config, limits, nullOutput, writer, nullptr);
+
+    core_t::TTime completeToTime(0);
+    CEmptySearcher restoreSearcher;
+    CPPUNIT_ASSERT(typer.restoreState(restoreSearcher, completeToTime) == false);
+}
+
 CppUnit::Test* CFieldDataTyperTest::suite()
 {
     CppUnit::TestSuite *suiteOfTests = new CppUnit::TestSuite("CFieldDataTyperTest");
@@ -341,6 +374,9 @@ CppUnit::Test* CFieldDataTyperTest::suite()
     suiteOfTests->addTest( new CppUnit::TestCaller<CFieldDataTyperTest>(
                                    "CFieldDataTyperTest::testHandleControlMessages",
                                    &CFieldDataTyperTest::testHandleControlMessages) );
+    suiteOfTests->addTest( new CppUnit::TestCaller<CFieldDataTyperTest>(
+                                   "CFieldDataTyperTest::testRestoreStateFailsWithEmptyState",
+                                   &CFieldDataTyperTest::testRestoreStateFailsWithEmptyState) );
     return suiteOfTests;
 }
 

lib/api/unittest/CFieldDataTyperTest.h

@@ -26,6 +26,7 @@ class CFieldDataTyperTest : public CppUnit::TestFixture
         void testNodeReverseSearch(void);
         void testPassOnControlMessages(void);
         void testHandleControlMessages(void);
+        void testRestoreStateFailsWithEmptyState(void);
 
         static CppUnit::Test *suite();
 

lib/api/unittest/CMockSearcher.cc

@@ -27,7 +27,7 @@ CMockSearcher::TIStreamP CMockSearcher::search(size_t currentDocNum, size_t /*li
 {
     if (currentDocNum == 0)
     {
-        LOG_ERROR("Current doc number cannot be 0 - KV store requires 1-based numbers");
+        LOG_ERROR("Current doc number cannot be 0 - data store requires 1-based numbers");
         return TIStreamP();
     }