[ML] Refactor model snapshot writing to use concurrent writer (#12)

This refactors the writing of the model snapshot document and
model size stats document out of CJsonOutputWriter.

It achieves the following improvements:

- model snapshot documents do not have to be queued up. They
can be written directly leveraging the safe concurrent writer
that was implemented for forecasts. This minimizes the delay
of a model snapshot being written which in failure cases may
result into having a more recent snapshot available.
- better separation of concerns and a smaller CJsonOutputWriter class.
- simplifies adding new fields in the model snapshot document as
needed for writing the minimum compatible version.

Author: dimitris-athanasiou

bin/autodetect/Main.cc

@@ -45,6 +45,7 @@
 #include <api/CIoManager.h>
 #include <api/CJsonOutputWriter.h>
 #include <api/CLengthEncodedInputParser.h>
+#include <api/CModelSnapshotJsonWriter.h>
 #include <api/COutputChainer.h>
 #include <api/CSingleStreamDataAdder.h>
 #include <api/CSingleStreamSearcher.h>
@@ -271,8 +272,7 @@ int main(int argc, char **argv)
 
     ml::core::CJsonOutputStreamWrapper wrappedOutputStream(ioMgr.outputStream());
 
-    // output writer for CFieldDataTyper and persistence callback
-    ml::api::CJsonOutputWriter outputWriter(jobId, wrappedOutputStream);
+    ml::api::CModelSnapshotJsonWriter modelSnapshotWriter(jobId, wrappedOutputStream);
     if (fieldConfig.initFromCmdLine(fieldConfigFile,
                                     clauseTokens) == false)
     {
@@ -286,8 +286,7 @@ int main(int argc, char **argv)
                              fieldConfig,
                              modelConfig,
                              wrappedOutputStream,
-                             boost::bind(&ml::api::CJsonOutputWriter::reportPersistComplete,
-                                         &outputWriter, _1, _2, _3, _4, _5, _6, _7, _8),
+                             boost::bind(&ml::api::CModelSnapshotJsonWriter::write, &modelSnapshotWriter, _1),
                              periodicPersister.get(),
                              maxQuantileInterval,
                              timeField,
@@ -312,8 +311,10 @@ int main(int argc, char **argv)
     // Chain the categorizer's output to the anomaly detector's input
     ml::api::COutputChainer outputChainer(job);
 
+    ml::api::CJsonOutputWriter fieldDataTyperOutputWriter(jobId, wrappedOutputStream);
+
     // The typer knows how to assign categories to records
-    ml::api::CFieldDataTyper typer(jobId, fieldConfig, limits, outputChainer, outputWriter);
+    ml::api::CFieldDataTyper typer(jobId, fieldConfig, limits, outputChainer, fieldDataTyperOutputWriter);
 
     if (fieldConfig.fieldNameSuperset().count(ml::api::CFieldDataTyper::MLCATEGORY_NAME) > 0)
     {
@@ -339,7 +340,7 @@ int main(int argc, char **argv)
     // as it must be finalised before the skeleton is destroyed, and C++
     // destruction order means the skeleton will be destroyed before the output
     // writer as it was constructed last.
-    outputWriter.finalise();
+    fieldDataTyperOutputWriter.finalise();
 
     if (!ioLoopSucceeded)
     {

include/api/CAnomalyJob.h

@@ -32,6 +32,7 @@
 #include <api/CDataProcessor.h>
 #include <api/CForecastRunner.h>
 #include <api/CJsonOutputWriter.h>
+#include <api/CModelSnapshotJsonWriter.h>
 #include <api/ImportExport.h>
 
 #include <boost/shared_ptr.hpp>
@@ -120,15 +121,7 @@ class API_EXPORT CAnomalyJob : public CDataProcessor
 
 
     public:
-        typedef std::function<void(
-                                 core_t::TTime,
-                                 const std::string &,
-                                 const std::string &,
-                                 size_t,
-                                 const model::CResourceMonitor::SResults &,
-                                 const std::string &,
-                                 core_t::TTime,
-                                 core_t::TTime)> TPersistCompleteFunc;
+        typedef std::function<void(const CModelSnapshotJsonWriter::SModelSnapshotReport &)> TPersistCompleteFunc;
         typedef model::CAnomalyDetector::TAnomalyDetectorPtr TAnomalyDetectorPtr;
         typedef std::vector<TAnomalyDetectorPtr> TAnomalyDetectorPtrVec;
         typedef std::vector<TAnomalyDetectorPtr>::iterator TAnomalyDetectorPtrVecItr;

include/api/CJsonOutputWriter.h

@@ -16,7 +16,6 @@
 #define INCLUDED_ml_api_CJsonOutputWriter_h
 
 #include <core/CJsonOutputStreamWrapper.h>
-#include <core/CMutex.h>
 #include <core/CoreTypes.h>
 #include <core/CRapidJsonConcurrentLineWriter.h>
 #include <core/CSmallVector.h>
@@ -36,7 +35,6 @@
 
 #include <iosfwd>
 #include <map>
-#include <queue>
 #include <sstream>
 #include <string>
 #include <utility>
@@ -190,98 +188,10 @@ class API_EXPORT CJsonOutputWriter : public COutputHandler
         typedef TTimeBucketDataMap::iterator           TTimeBucketDataMapItr;
         typedef TTimeBucketDataMap::const_iterator     TTimeBucketDataMapCItr;
 
-
-        static const std::string JOB_ID;
-        static const std::string TIMESTAMP;
-        static const std::string BUCKET;
-        static const std::string LOG_TIME;
-        static const std::string DETECTOR_INDEX;
-        static const std::string RECORDS;
-        static const std::string EVENT_COUNT;
-        static const std::string IS_INTERIM;
-        static const std::string PROBABILITY;
-        static const std::string RAW_ANOMALY_SCORE;
-        static const std::string ANOMALY_SCORE;
-        static const std::string RECORD_SCORE;
-        static const std::string INITIAL_RECORD_SCORE;
-        static const std::string INFLUENCER_SCORE;
-        static const std::string INITIAL_INFLUENCER_SCORE;
-        static const std::string FIELD_NAME;
-        static const std::string BY_FIELD_NAME;
-        static const std::string BY_FIELD_VALUE;
-        static const std::string CORRELATED_BY_FIELD_VALUE;
-        static const std::string TYPICAL;
-        static const std::string ACTUAL;
-        static const std::string CAUSES;
-        static const std::string FUNCTION;
-        static const std::string FUNCTION_DESCRIPTION;
-        static const std::string OVER_FIELD_NAME;
-        static const std::string OVER_FIELD_VALUE;
-        static const std::string PARTITION_FIELD_NAME;
-        static const std::string PARTITION_FIELD_VALUE;
-        static const std::string INITIAL_SCORE;
-        static const std::string BUCKET_INFLUENCERS;
-        static const std::string INFLUENCERS;
-        static const std::string INFLUENCER_FIELD_NAME;
-        static const std::string INFLUENCER_FIELD_VALUE;
-        static const std::string INFLUENCER_FIELD_VALUES;
-        static const std::string FLUSH;
-        static const std::string ID;
-        static const std::string LAST_FINALIZED_BUCKET_END;
-        static const std::string QUANTILE_STATE;
-        static const std::string QUANTILES;
-        static const std::string MODEL_SIZE_STATS;
-        static const std::string MODEL_BYTES;
-        static const std::string TOTAL_BY_FIELD_COUNT;
-        static const std::string TOTAL_OVER_FIELD_COUNT;
-        static const std::string TOTAL_PARTITION_FIELD_COUNT;
-        static const std::string BUCKET_ALLOCATION_FAILURES_COUNT;
-        static const std::string MEMORY_STATUS;
-        static const std::string CATEGORY_DEFINITION;
-        static const std::string CATEGORY_ID;
-        static const std::string TERMS;
-        static const std::string REGEX;
-        static const std::string MAX_MATCHING_LENGTH;
-        static const std::string EXAMPLES;
-        static const std::string MODEL_SNAPSHOT;
-        static const std::string SNAPSHOT_ID;
-        static const std::string SNAPSHOT_DOC_COUNT;
-        static const std::string DESCRIPTION;
-        static const std::string LATEST_RECORD_TIME;
-        static const std::string BUCKET_SPAN;
-        static const std::string LATEST_RESULT_TIME;
-        static const std::string PROCESSING_TIME;
-        static const std::string TIME_INFLUENCER;
-        static const std::string PARTITION_SCORES;
-        static const std::string SCHEDULED_EVENTS;
-
     private:
         typedef CCategoryExamplesCollector::TStrSet TStrSet;
         typedef TStrSet::const_iterator TStrSetCItr;
 
-        struct SModelSnapshotReport
-        {
-            SModelSnapshotReport(core_t::TTime snapshotTimestamp,
-                                 const std::string &description,
-                                 const std::string &snapshotId,
-                                 size_t numDocs,
-                                 const model::CResourceMonitor::SResults &modelSizeStats,
-                                 const std::string &normalizerState,
-                                 core_t::TTime latestRecordTime,
-                                 core_t::TTime latestFinalResultTime);
-
-            core_t::TTime                     s_SnapshotTimestamp;
-            std::string                       s_Description;
-            std::string                       s_SnapshotId;
-            size_t                            s_NumDocs;
-            model::CResourceMonitor::SResults s_ModelSizeStats;
-            std::string                       s_NormalizerState;
-            core_t::TTime                     s_LatestRecordTime;
-            core_t::TTime                     s_LatestFinalResultTime;
-        };
-
-        typedef std::queue<SModelSnapshotReport> TModelSnapshotReportQueue;
-
     public:
         //! Constructor that causes output to be written to the specified wrapped stream
         CJsonOutputWriter(const std::string &jobId,
@@ -352,19 +262,6 @@ class API_EXPORT CJsonOutputWriter : public COutputHandler
         //! from the CResourceMonitor via a callback
         void reportMemoryUsage(const model::CResourceMonitor::SResults &results);
 
-        //! Report information about completion of model persistence.
-        //! This method can be called in a thread other than the one
-        //! receiving the majority of results, so reporting is done
-        //! asynchronously.
-        void reportPersistComplete(core_t::TTime snapshotTimestamp,
-                                   const std::string &description,
-                                   const std::string &snapshotId,
-                                   size_t numDocs,
-                                   const model::CResourceMonitor::SResults &modelSizeStats,
-                                   const std::string &normalizerState,
-                                   core_t::TTime latestRecordTime,
-                                   core_t::TTime latestFinalResultTime);
-
         //! Acknowledge a flush request by echoing back the flush ID
         void acknowledgeFlush(const std::string &flushId, core_t::TTime lastFinalizedBucketEnd);
 
@@ -427,49 +324,33 @@ class API_EXPORT CJsonOutputWriter : public COutputHandler
         void addPartitionScores(const CHierarchicalResultsWriter::TResults &results,
                                 TDocumentWeakPtr weakDoc);
 
-        //! Write any model snapshot reports that are queuing up.
-        void writeModelSnapshotReports(void);
-
-        //! Write the JSON object showing current levels of resource usage, as
-        //! given to us from the CResourceMonitor via a callback
-        void writeMemoryUsageObject(const model::CResourceMonitor::SResults &results);
-
-        //! Write the quantile's state
-        void writeQuantileState(const std::string &state, core_t::TTime timestamp);
-
     private:
         //! The job ID
-        std::string                       m_JobId;
+        std::string                          m_JobId;
 
         //! JSON line writer
-        core::CRapidJsonConcurrentLineWriter        m_Writer;
+        core::CRapidJsonConcurrentLineWriter m_Writer;
 
         //! Time of last non-interim bucket written to output
-        core_t::TTime                     m_LastNonInterimBucketTime;
+        core_t::TTime                        m_LastNonInterimBucketTime;
 
         //! Has the output been finalised?
-        bool                              m_Finalised;
+        bool                                 m_Finalised;
 
         //! Max number of records to write for each bucket/detector
-        size_t                            m_RecordOutputLimit;
+        size_t                               m_RecordOutputLimit;
 
         //! Vector for building up documents representing nested sub-results.
         //! The documents in this vector will reference memory owned by
         //! m_JsonPoolAllocator.  (Hence this is declared after the memory pool
         //! so that it's destroyed first when the destructor runs.)
-        TDocumentWeakPtrVec               m_NestedDocs;
+        TDocumentWeakPtrVec                  m_NestedDocs;
 
         //! Bucket data waiting to be written.  The map is keyed on bucket time.
         //! The documents in this map will reference memory owned by
         //! m_JsonPoolAllocator.  (Hence this is declared after the memory pool
         //! so that it's destroyed first when the destructor runs.)
-        TTimeBucketDataMap                m_BucketDataByTime;
-
-        //! Protects the m_ModelSnapshotReports from concurrent access.
-        core::CMutex                      m_ModelSnapshotReportsQueueMutex;
-
-        //! Queue of model snapshot reports waiting to be output.
-        TModelSnapshotReportQueue         m_ModelSnapshotReports;
+        TTimeBucketDataMap                   m_BucketDataByTime;
 };
 
 

include/api/CModelSizeStatsJsonWriter.h

@@ -0,0 +1,46 @@
+/*
+ * ELASTICSEARCH CONFIDENTIAL
+ *
+ * Copyright (c) 2018 Elasticsearch BV. All Rights Reserved.
+ *
+ * Notice: this software, and all information contained
+ * therein, is the exclusive property of Elasticsearch BV
+ * and its licensors, if any, and is protected under applicable
+ * domestic and foreign law, and international treaties.
+ *
+ * Reproduction, republication or distribution without the
+ * express written consent of Elasticsearch BV is
+ * strictly prohibited.
+ */
+#ifndef INCLUDED_ml_api_CModelSizeStatsJsonWriter_h
+#define INCLUDED_ml_api_CModelSizeStatsJsonWriter_h
+
+#include <core/CNonInstantiatable.h>
+#include <core/CRapidJsonConcurrentLineWriter.h>
+
+#include <model/CResourceMonitor.h>
+
+#include <api/ImportExport.h>
+
+#include <string>
+
+namespace ml
+{
+namespace api
+{
+
+//! \brief
+//! A static utility for writing the model_size_stats document in JSON.
+class API_EXPORT CModelSizeStatsJsonWriter : private core::CNonInstantiatable
+{
+    public:
+        //! Writes the model size stats in the \p results in JSON format.
+        static void write(const std::string &jobId,
+                          const model::CResourceMonitor::SResults &results,
+                          core::CRapidJsonConcurrentLineWriter &writer);
+};
+
+}
+}
+
+#endif // INCLUDED_ml_api_CModelSizeStatsJsonWriter_h

include/api/CModelSnapshotJsonWriter.h

@@ -0,0 +1,80 @@
+/*
+ * ELASTICSEARCH CONFIDENTIAL
+ *
+ * Copyright (c) 2018 Elasticsearch BV. All Rights Reserved.
+ *
+ * Notice: this software, and all information contained
+ * therein, is the exclusive property of Elasticsearch BV
+ * and its licensors, if any, and is protected under applicable
+ * domestic and foreign law, and international treaties.
+ *
+ * Reproduction, republication or distribution without the
+ * express written consent of Elasticsearch BV is
+ * strictly prohibited.
+ */
+#ifndef INCLUDED_ml_api_CModelSnapshotJsonWriter_h
+#define INCLUDED_ml_api_CModelSnapshotJsonWriter_h
+
+#include <core/CJsonOutputStreamWrapper.h>
+#include <core/CRapidJsonConcurrentLineWriter.h>
+
+#include <model/CResourceMonitor.h>
+
+#include <api/ImportExport.h>
+
+#include <string>
+
+namespace ml
+{
+namespace api
+{
+
+//! \brief
+//! Write model snapshots in JSON format
+//!
+//! DESCRIPTION:\n
+//! Outputs the model snapshot documents that accompany each state persist.
+//!
+class API_EXPORT CModelSnapshotJsonWriter
+{
+    public:
+        //! Structure to store the model snapshot metadata
+        struct SModelSnapshotReport
+        {
+            core_t::TTime                     s_SnapshotTimestamp;
+            std::string                       s_Description;
+            std::string                       s_SnapshotId;
+            size_t                            s_NumDocs;
+            model::CResourceMonitor::SResults s_ModelSizeStats;
+            std::string                       s_NormalizerState;
+            core_t::TTime                     s_LatestRecordTime;
+            core_t::TTime                     s_LatestFinalResultTime;
+        };
+
+    public:
+        //! Constructor that causes output to be written to the specified wrapped stream
+        CModelSnapshotJsonWriter(const std::string &jobId,
+                                 core::CJsonOutputStreamWrapper &strmOut);
+
+        //! Writes the given model snapshot in JSON format.
+        void write(const SModelSnapshotReport &report);
+
+        //! Write the quantile's state
+        static void writeQuantileState(const std::string &jobId,
+                                       const std::string &state,
+                                       core_t::TTime timestamp,
+                                       core::CRapidJsonConcurrentLineWriter &writer);
+
+    private:
+        //! The job ID
+        std::string                          m_JobId;
+
+        //! JSON line writer
+        core::CRapidJsonConcurrentLineWriter m_Writer;
+};
+
+
+}
+}
+
+#endif // INCLUDED_ml_api_CModelSnapshotJsonWriter_h