[ML] Feature/forecast scale (#89)

This implements the C++ side of forecast persistence. An additional parameter allows the forecast runner to persist models on disk for temporary purposes. Models are loaded back into memory one by one.

Author: Hendrik Muhs

bin/autodetect/Makefile

@@ -12,6 +12,7 @@ INSTALL_DIR=$(CPP_PLATFORM_HOME)/bin
 ML_LIBS=$(LIB_ML_CORE) $(LIB_ML_MATHS) $(LIB_ML_MODEL) $(LIB_ML_API)
 
 USE_BOOST=1
+USE_BOOST_FILESYSTEM_LIBS=1
 USE_BOOST_PROGRAMOPTIONS_LIBS=1
 USE_RAPIDJSON=1
 USE_EIGEN=1

docs/CHANGELOG.asciidoc

@@ -17,6 +17,9 @@
 
 Improve and use periodic boundary condition for seasonal component modeling ({pull}84[#84])
 
+Forecasting of Machine Learning job time series is now supported for large jobs by temporarily storing
+model state on disk ({pull}89[#89])
+
 === Bug Fixes
 
 === Regressions

include/api/CForecastRunner.h

@@ -21,6 +21,7 @@
 #include <model/CForecastDataSink.h>
 #include <model/CResourceMonitor.h>
 
+#include <boost/filesystem.hpp>
 #include <boost/unordered_set.hpp>
 
 #include <condition_variable>
@@ -68,10 +69,22 @@ class API_EXPORT CForecastRunner final : private core::CNonCopyable {
     static const size_t DEFAULT_EXPIRY_TIME = 14 * core::constants::DAY;
 
     //! max memory allowed to use for forecast models
-    static const size_t MAX_FORECAST_MODEL_MEMORY = 20971520; // 20MB
+    static const size_t MAX_FORECAST_MODEL_MEMORY = 20971520ull; // 20MB
+
+    //! Note: This value measures the size in memory, not the size of the persistence,
+    //! which is likely higher and would be hard to calculate upfront
+    //! max memory allowed to use for forecast models persisting to disk
+    static const size_t MAX_FORECAST_MODEL_PERSISTANCE_MEMORY = 524288000ull; // 500MB
+
+    //! Note: This value is lower than on X-pack side to prevent side-effects,
+    //! if you change this value also change the limit on X-pack side.
+    //! The purpose of this value is to guard the rest of the system regarding
+    //! an out of disk space
+    //! minimum disk space required for disk persistence
+    static const size_t MIN_FORECAST_AVAILABLE_DISK_SPACE = 4294967296ull; // 4GB
 
     //! minimum time between stat updates to prevent to many updates in a short time
-    static const uint64_t MINIMUM_TIME_ELAPSED_FOR_STATS_UPDATE = 3000; // 3s
+    static const uint64_t MINIMUM_TIME_ELAPSED_FOR_STATS_UPDATE = 3000ul; // 3s
 
 private:
     static const std::string ERROR_FORECAST_REQUEST_FAILED_TO_PARSE;
@@ -82,6 +95,8 @@ class API_EXPORT CForecastRunner final : private core::CNonCopyable {
     static const std::string ERROR_NO_CREATE_TIME;
     static const std::string ERROR_BAD_MEMORY_STATUS;
     static const std::string ERROR_MEMORY_LIMIT;
+    static const std::string ERROR_MEMORY_LIMIT_DISK;
+    static const std::string ERROR_MEMORY_LIMIT_DISKSPACE;
     static const std::string ERROR_NOT_SUPPORTED_FOR_POPULATION_MODELS;
     static const std::string ERROR_NO_SUPPORTED_FUNCTIONS;
     static const std::string WARNING_DURATION_LIMIT;
@@ -100,6 +115,7 @@ class API_EXPORT CForecastRunner final : private core::CNonCopyable {
     using TForecastModelWrapper = model::CForecastDataSink::SForecastModelWrapper;
     using TForecastResultSeries = model::CForecastDataSink::SForecastResultSeries;
     using TForecastResultSeriesVec = std::vector<TForecastResultSeries>;
+    using TMathsModelPtr = std::shared_ptr<maths::CModel>;
 
     using TStrUSet = boost::unordered_set<std::string>;
 
@@ -186,6 +202,9 @@ class API_EXPORT CForecastRunner final : private core::CNonCopyable {
 
         //! A collection storing important messages from forecasting
         TStrUSet s_Messages;
+
+        //! A directory to persist models on disk
+        std::string s_TemporaryFolder;
     };
 
 private:
@@ -199,6 +218,9 @@ class API_EXPORT CForecastRunner final : private core::CNonCopyable {
     //! Check for new jobs, blocks while waiting
     bool tryGetJob(SForecast& forecastJob);
 
+    //! check for sufficient disk space
+    bool sufficientAvailableDiskSpace(const boost::filesystem::path& path);
+
     //! pushes new jobs into the internal 'queue' (thread boundary)
     bool push(SForecast& forecastJob);
 

include/core/RestoreMacros.h

@@ -39,6 +39,23 @@ namespace core {
         continue;                                                                  \
     }
 
+#define RESTORE_ENUM(tag, target, enumtype)                                        \
+    if (name == tag) {                                                             \
+        int value;                                                                 \
+        if (core::CStringUtils::stringToType(traverser.value(), value) == false) { \
+            LOG_ERROR(<< "Failed to restore " #tag ", got " << traverser.value()); \
+            return false;                                                          \
+        }                                                                          \
+        target = enumtype(value);                                                  \
+        continue;                                                                  \
+    }
+
+#define RESTORE_ENUM_CHECKED(tag, target, enumtype, restoreSuccess)            \
+    if (name == tag) {                                                         \
+        restoreSuccess = true;                                                 \
+        RESTORE_ENUM(tag, target, enumtype)                                    \
+    }
+
 #define RESTORE_SETUP_TEARDOWN(tag, setup, restore, teardown)                      \
     if (name == tag) {                                                             \
         setup;                                                                     \

include/model/CAnomalyDetector.h

@@ -234,7 +234,9 @@ class MODEL_EXPORT CAnomalyDetector : private core::CNonCopyable {
     CForecastDataSink::SForecastModelPrerequisites getForecastPrerequisites() const;
 
     //! Generate maths models for forecasting
-    CForecastDataSink::SForecastResultSeries getForecastModels() const;
+    CForecastDataSink::SForecastResultSeries
+    getForecastModels(bool persistOnDisk = false,
+                      const std::string& persistenceFolder = EMPTY_STRING) const;
 
     //! Remove dead models, i.e. those models that have more-or-less
     //! reverted back to their non-informative state.  BE CAREFUL WHEN

include/model/CCountingModelFactory.h

@@ -121,6 +121,9 @@ class MODEL_EXPORT CCountingModelFactory : public CModelFactory {
     virtual void bucketResultsDelay(std::size_t bucketResultsDelay);
     //@}
 
+    //! Get the minimum seasonal variance scale
+    virtual double minimumSeasonalVarianceScale() const;
+
 private:
     //! Get the field values which partition the data for modeling.
     virtual TStrCRefVec partitioningFields() const;

include/model/CEventRateModelFactory.h

@@ -131,6 +131,9 @@ class MODEL_EXPORT CEventRateModelFactory : public CModelFactory {
     virtual void bucketResultsDelay(std::size_t bucketResultsDelay);
     //@}
 
+    //! Get the minimum seasonal variance scale
+    virtual double minimumSeasonalVarianceScale() const;
+
 private:
     //! Get the field values which partition the data for modeling.
     virtual TStrCRefVec partitioningFields() const;

include/model/CEventRatePopulationModelFactory.h

@@ -133,6 +133,9 @@ class MODEL_EXPORT CEventRatePopulationModelFactory : public CModelFactory {
     virtual void bucketResultsDelay(std::size_t bucketResultsDelay);
     //@}
 
+    //! Get the minimum seasonal variance scale
+    virtual double minimumSeasonalVarianceScale() const;
+
 private:
     //! Get the field values which partition the data for modeling.
     virtual TStrCRefVec partitioningFields() const;

include/model/CForecastDataSink.h

@@ -14,6 +14,7 @@
 
 #include <maths/CModel.h>
 
+#include <model/CModelParams.h>
 #include <model/ImportExport.h>
 #include <model/ModelTypes.h>
 
@@ -37,7 +38,7 @@ namespace model {
 //! to change (e.g. the json writing should not happen in this class).
 class MODEL_EXPORT CForecastDataSink final : private core::CNonCopyable {
 public:
-    using TMathsModelPtr = std::unique_ptr<maths::CModel>;
+    using TMathsModelPtr = std::shared_ptr<maths::CModel>;
     using TStrUMap = boost::unordered_set<std::string>;
 
     //! Wrapper for 1 timeseries model, its feature and by Field
@@ -58,18 +59,21 @@ class MODEL_EXPORT CForecastDataSink final : private core::CNonCopyable {
 
     //! Everything that defines 1 series of forecasts
     struct MODEL_EXPORT SForecastResultSeries {
-        SForecastResultSeries();
+        SForecastResultSeries(const SModelParams& modelParams);
 
         SForecastResultSeries(SForecastResultSeries&& other);
 
         SForecastResultSeries(const SForecastResultSeries& that) = delete;
         SForecastResultSeries& operator=(const SForecastResultSeries&) = delete;
 
+        SModelParams s_ModelParams;
         int s_DetectorIndex;
         std::vector<SForecastModelWrapper> s_ToForecast;
+        std::string s_ToForecastPersisted;
         std::string s_PartitionFieldName;
         std::string s_PartitionFieldValue;
         std::string s_ByFieldName;
+        double s_MinimumSeasonalVarianceScale;
     };
 
     //! \brief Data describing prerequisites prior predictions

include/model/CForecastModelPersist.h

@@ -0,0 +1,108 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+#ifndef INCLUDED_ml_model_CForecastModelPersist_h
+#define INCLUDED_ml_model_CForecastModelPersist_h
+
+#include <core/CJsonStatePersistInserter.h>
+#include <core/CJsonStateRestoreTraverser.h>
+
+#include <maths/CModel.h>
+
+#include <model/CModelParams.h>
+#include <model/ImportExport.h>
+#include <model/ModelTypes.h>
+
+#include <boost/filesystem.hpp>
+
+#include <fstream>
+#include <memory>
+
+namespace ml {
+namespace model {
+
+//! \brief Persist/Restore CModel sub-classes to/from text representations for
+//!  the purpose of forecasting.
+//!
+//! DESCRIPTION:\n
+//! Persists/Restores models to disk for the purpose of restoring and forecasting
+//! on them.
+//!
+//! IMPLEMENTATION DECISIONS:\n
+//! Only as complete as required for forecasting.
+//!
+//! Persist and Restore are only done to avoid heap memory usage using temporary disk space.
+//! No need for backwards compatibility and version'ing as code will only be used
+//! locally never leaving process/io boundaries.
+class MODEL_EXPORT CForecastModelPersist final {
+public:
+    using TMathsModelPtr = std::shared_ptr<maths::CModel>;
+
+public:
+    class MODEL_EXPORT CPersist final {
+    public:
+        explicit CPersist(const std::string& temporaryPath);
+
+        //! add a model to the persistence
+        void addModel(const maths::CModel* model,
+                      const model_t::EFeature feature,
+                      const std::string& byFieldValue);
+
+        //! close the outputStream
+        const std::string& finalizePersistAndGetFile();
+
+    private:
+        static void persistOneModel(core::CStatePersistInserter& inserter,
+                                    const maths::CModel* model,
+                                    const model_t::EFeature feature,
+                                    const std::string& byFieldValue);
+
+    private:
+        //! the filename where to persist to
+        boost::filesystem::path m_FileName;
+
+        //! the actual file where it models are persisted to
+        std::ofstream m_OutStream;
+
+        //! number of models persisted
+        size_t m_ModelCount;
+    };
+
+    class MODEL_EXPORT CRestore final {
+    public:
+        explicit CRestore(const SModelParams& modelParams,
+                          double minimumSeasonalVarianceScale,
+                          const std::string& fileName);
+
+        //! add a model to the persistence
+        bool nextModel(TMathsModelPtr& model, model_t::EFeature& feature, std::string& byFieldValue);
+
+    private:
+        static bool restoreOneModel(core::CStateRestoreTraverser& traverser,
+                                    SModelParams modelParams,
+                                    double inimumSeasonalVarianceScale,
+                                    TMathsModelPtr& model,
+                                    model_t::EFeature& feature,
+                                    std::string& byFieldValue);
+
+    private:
+        //! model parameters required in order to restore the model
+        SModelParams m_ModelParams;
+
+        //! minimum seasonal variance scale specific to the model
+        double m_MinimumSeasonalVarianceScale;
+
+        //! the actual file where it models are persisted to
+        std::ifstream m_InStream;
+
+        //! the persist inserter
+        core::CJsonStateRestoreTraverser m_RestoreTraverser;
+    }; // class CRestore
+};     // class CForecastModelPersist
+}
+}
+
+#endif // INCLUDED_ml_model_CForecastModelPersist_h

include/model/CMetricModelFactory.h

@@ -134,6 +134,9 @@ class MODEL_EXPORT CMetricModelFactory : public CModelFactory {
     virtual void bucketResultsDelay(std::size_t bucketResultsDelay);
     //@}
 
+    //! Get the minimum seasonal variance scale
+    virtual double minimumSeasonalVarianceScale() const;
+
 private:
     //! Get the field values which partition the data for modeling.
     virtual TStrCRefVec partitioningFields() const;

include/model/CMetricPopulationModelFactory.h

@@ -133,6 +133,9 @@ class MODEL_EXPORT CMetricPopulationModelFactory : public CModelFactory {
     virtual void bucketResultsDelay(std::size_t bucketResultsDelay);
     //@}
 
+    //! Get the minimum seasonal variance scale
+    virtual double minimumSeasonalVarianceScale() const;
+
 private:
     //! Get the field values which partition the data for modeling.
     virtual TStrCRefVec partitioningFields() const;

include/model/CModelFactory.h

@@ -351,6 +351,9 @@ class MODEL_EXPORT CModelFactory {
     //! component.
     std::size_t componentSize() const;
 
+    //! Get the minimum seasonal variance scale, specific to the model
+    virtual double minimumSeasonalVarianceScale() const = 0;
+
 protected:
     using TMultivariatePriorPtrVec = std::vector<TMultivariatePriorPtr>;
     using TOptionalSearchKey = boost::optional<CSearchKey>;