[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.

For models smaller than the current limit of 20MB nothing changes.

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

@@ -31,6 +31,9 @@ Improve and use periodic boundary condition for seasonal component modeling ({pu
 Improve robustness w.r.t. outliers of detection and initialisation of seasonal components ({pull}90[#90])
 Explicit change point detection and modelling ({pull}92[#92])
 
+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
 
 Age seasonal components in proportion to the fraction of values with which they're updated ({pull}88[#88])

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>;
 
@@ -189,6 +205,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:
@@ -202,6 +221,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>
 
@@ -38,7 +39,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
@@ -59,18 +60,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

@@ -346,6 +346,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>;