[ML] Format memory instrumentation output

docs/CHANGELOG.asciidoc

@@ -56,6 +56,8 @@ necessary. This will improve the allocation of data frame analyses to cluster no
 (See {ml-pull}1003[#1003].)
 * Upgrade the compiler used on Linux from gcc 7.3 to gcc 7.5, and the binutils used in
 the build from version 2.20 to 2.34.  (See {ml-pull}1013[#1013].)
+* Add instrumentation of the peak memory consumption for data frame analytics jobs.
+(See {ml-pull}1022[#1022].)
 * Remove all memory overheads for computing tree SHAP values. (See {ml-pull}1023[#1023].)
 
 === Bug Fixes

include/api/CDataFrameAnalysisInstrumentation.h

@@ -30,7 +30,7 @@ class API_EXPORT CDataFrameAnalysisInstrumentation
     : public maths::CDataFrameAnalysisInstrumentationInterface {
 
 public:
-    CDataFrameAnalysisInstrumentation();
+    explicit CDataFrameAnalysisInstrumentation(const std::string& jobId);
 
     //! Adds \p delta to the memory usage statistics.
     void updateMemoryUsage(std::int64_t delta) override;
@@ -73,25 +73,36 @@ class API_EXPORT CDataFrameAnalysisInstrumentation
 
 private:
     void writeProgress(std::uint32_t step);
-    void writeMemory(std::uint32_t step);
+    void writeMemory(std::int64_t timestamp);
     void writeState(std::uint32_t step);
 
 private:
+    std::string m_JobId;
     std::atomic_bool m_Finished;
     std::atomic_size_t m_FractionalProgress;
     std::atomic<std::int64_t> m_Memory;
     core::CRapidJsonConcurrentLineWriter* m_Writer;
 };
 
+//! \brief Outlier instrumentation.
 class API_EXPORT CDataFrameOutliersInstrumentation final
     : public CDataFrameAnalysisInstrumentation {
-protected:
+public:
+    explicit CDataFrameOutliersInstrumentation(const std::string& jobId)
+        : CDataFrameAnalysisInstrumentation(jobId) {}
+
+private:
     counter_t::ECounterTypes memoryCounterType() override;
 };
 
+//! \brief Predictive model training instrumentation.
 class API_EXPORT CDataFrameTrainBoostedTreeInstrumentation final
     : public CDataFrameAnalysisInstrumentation {
-protected:
+public:
+    explicit CDataFrameTrainBoostedTreeInstrumentation(const std::string& jobId)
+        : CDataFrameAnalysisInstrumentation(jobId) {}
+
+private:
     counter_t::ECounterTypes memoryCounterType() override;
 };
 }

lib/api/CDataFrameAnalysisInstrumentation.cc

@@ -3,16 +3,21 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-
 #include <api/CDataFrameAnalysisInstrumentation.h>
 
+#include <core/CTimeUtils.h>
+
 namespace ml {
 namespace api {
 
 namespace {
 const std::string STEP_TAG{"step"};
 const std::string PROGRESS_TAG{"progress"};
-const std::string PEAK_MEMORY_USAGE_TAG{"peak_memory_usage"};
+const std::string PEAK_MEMORY_USAGE_TAG{"peak_usage_bytes"};
+const std::string TYPE_TAG{"type"};
+const std::string JOB_ID_TAG{"job_id"};
+const std::string TIMESTAMP_TAG{"timestamp"};
+const std::string MEMORY_TYPE{"analytics_memory_usage"};
 
 const std::size_t MAXIMUM_FRACTIONAL_PROGRESS{std::size_t{1}
                                               << ((sizeof(std::size_t) - 2) * 8)};
@@ -51,8 +56,8 @@ double CDataFrameAnalysisInstrumentation::progress() const {
                      static_cast<double>(MAXIMUM_FRACTIONAL_PROGRESS);
 }
 
-CDataFrameAnalysisInstrumentation::CDataFrameAnalysisInstrumentation()
-    : m_Finished{false}, m_FractionalProgress{0}, m_Memory{0}, m_Writer{nullptr} {
+CDataFrameAnalysisInstrumentation::CDataFrameAnalysisInstrumentation(const std::string& jobId)
+    : m_JobId{jobId}, m_Finished{false}, m_FractionalProgress{0}, m_Memory{0}, m_Writer{nullptr} {
 }
 
 void CDataFrameAnalysisInstrumentation::resetProgress() {
@@ -64,14 +69,14 @@ void CDataFrameAnalysisInstrumentation::writer(core::CRapidJsonConcurrentLineWri
     m_Writer = writer;
 }
 
-void CDataFrameAnalysisInstrumentation::nextStep(std::uint32_t /*step*/) {
-    // TODO reactivate state writing, once the Java backend can accept it
-    //    this->writeState(step);
+void CDataFrameAnalysisInstrumentation::nextStep(std::uint32_t step) {
+    this->writeState(step);
 }
 
 void CDataFrameAnalysisInstrumentation::writeState(std::uint32_t step) {
-    this->writeProgress(step);
-    this->writeMemory(step);
+    //this->writeProgress(step);
+    std::int64_t timestamp{core::CTimeUtils::toEpochMs(core::CTimeUtils::now())};
+    this->writeMemory(timestamp);
 }
 
 std::int64_t CDataFrameAnalysisInstrumentation::memory() const {
@@ -89,14 +94,19 @@ void CDataFrameAnalysisInstrumentation::writeProgress(std::uint32_t step) {
     }
 }
 
-void CDataFrameAnalysisInstrumentation::writeMemory(std::uint32_t step) {
+void CDataFrameAnalysisInstrumentation::writeMemory(std::int64_t timestamp) {
     if (m_Writer != nullptr) {
         m_Writer->StartObject();
-        m_Writer->Key(STEP_TAG);
-        m_Writer->Uint(step);
+        m_Writer->Key(MEMORY_TYPE);
+        m_Writer->StartObject();
+        m_Writer->Key(JOB_ID_TAG);
+        m_Writer->String(m_JobId);
+        m_Writer->Key(TIMESTAMP_TAG);
+        m_Writer->Int64(timestamp);
         m_Writer->Key(PEAK_MEMORY_USAGE_TAG);
         m_Writer->Int64(m_Memory.load());
         m_Writer->EndObject();
+        m_Writer->EndObject();
     }
 }
 

lib/api/CDataFrameOutliersRunner.cc

@@ -73,7 +73,8 @@ CDataFrameOutliersRunner::CDataFrameOutliersRunner(const CDataFrameAnalysisSpeci
 
 CDataFrameOutliersRunner::CDataFrameOutliersRunner(const CDataFrameAnalysisSpecification& spec)
     : CDataFrameAnalysisRunner{spec}, m_Method{static_cast<std::size_t>(
-                                          maths::COutliers::E_Ensemble)} {
+                                          maths::COutliers::E_Ensemble)},
+      m_Instrumentation{spec.jobId()} {
 }
 
 std::size_t CDataFrameOutliersRunner::numberExtraColumns() const {

lib/api/CDataFrameTrainBoostedTreeRunner.cc

@@ -69,7 +69,7 @@ CDataFrameTrainBoostedTreeRunner::CDataFrameTrainBoostedTreeRunner(
     const CDataFrameAnalysisSpecification& spec,
     const CDataFrameAnalysisParameters& parameters,
     TLossFunctionUPtr loss)
-    : CDataFrameAnalysisRunner{spec} {
+    : CDataFrameAnalysisRunner{spec}, m_Instrumentation{spec.jobId()} {
 
     m_DependentVariableFieldName = parameters[DEPENDENT_VARIABLE_NAME].as<std::string>();
 

lib/api/unittest/CDataFrameAnalysisInstrumentationTest.cc

@@ -0,0 +1,52 @@
+/*
+ * 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.
+ */
+
+#include <core/CTimeUtils.h>
+
+#include <api/CDataFrameAnalysisInstrumentation.h>
+
+#include <boost/test/unit_test.hpp>
+
+#include <string>
+
+BOOST_AUTO_TEST_SUITE(CDataFrameAnalysisInstrumentationTest)
+
+using namespace ml;
+
+BOOST_AUTO_TEST_CASE(testMemoryState) {
+    std::string jobId{"JOB123"};
+    std::int64_t memoryUsage{1000};
+    std::int64_t timeBefore{core::CTimeUtils::toEpochMs(core::CTimeUtils::now())};
+    std::stringstream outputStream;
+    {
+        core::CJsonOutputStreamWrapper streamWrapper(outputStream);
+        core::CRapidJsonConcurrentLineWriter writer(streamWrapper);
+        api::CDataFrameTrainBoostedTreeInstrumentation instrumentation(jobId);
+        instrumentation.updateMemoryUsage(memoryUsage);
+        instrumentation.writer(&writer);
+        instrumentation.nextStep(0);
+        outputStream.flush();
+    }
+    std::int64_t timeAfter{core::CTimeUtils::toEpochMs(core::CTimeUtils::now())};
+
+    rapidjson::Document results;
+    rapidjson::ParseResult ok(results.Parse(outputStream.str()));
+
+    BOOST_TEST_REQUIRE(static_cast<bool>(ok) == true);
+    BOOST_TEST_REQUIRE(results.IsArray() == true);
+    for (auto i = results.Begin(); i != results.End(); ++i) {
+        if (i->HasMember("analytics_memory_usage")) {
+            BOOST_TEST_REQUIRE((*i)["analytics_memory_usage"].IsObject() == true);
+            BOOST_TEST_REQUIRE((*i)["analytics_memory_usage"]["job_id"].GetString() == jobId);
+            BOOST_TEST_REQUIRE(
+                (*i)["analytics_memory_usage"]["peak_usage_bytes"].GetInt64() == memoryUsage);
+            BOOST_TEST_REQUIRE((*i)["analytics_memory_usage"]["timestamp"].GetInt64() >= timeBefore);
+            BOOST_TEST_REQUIRE((*i)["analytics_memory_usage"]["timestamp"].GetInt64() <= timeAfter);
+        }
+    }
+}
+
+BOOST_AUTO_TEST_SUITE_END()

lib/api/unittest/CDataFrameAnalyzerTrainingTest.cc

@@ -265,7 +265,7 @@ void addPredictionTestData(EPredictionType type,
         treeFactory.featureBagFraction(featureBagFraction);
     }
 
-    ml::api::CDataFrameTrainBoostedTreeInstrumentation instrumentation;
+    ml::api::CDataFrameTrainBoostedTreeInstrumentation instrumentation("testJob");
     treeFactory.analysisInstrumentation(instrumentation);
 
     auto tree = treeFactory.buildFor(*frame, weights.size());

lib/api/unittest/CDataFrameMockAnalysisRunner.cc

@@ -10,7 +10,7 @@
 #include <core/CLoopProgress.h>
 
 CDataFrameMockAnalysisRunner::CDataFrameMockAnalysisRunner(const ml::api::CDataFrameAnalysisSpecification& spec)
-    : ml::api::CDataFrameAnalysisRunner{spec} {
+    : ml::api::CDataFrameAnalysisRunner{spec}, m_Instrumentation{spec.jobId()} {
 }
 
 std::size_t CDataFrameMockAnalysisRunner::numberExtraColumns() const {

lib/api/unittest/CDataFrameMockAnalysisRunner.h

@@ -16,6 +16,10 @@
 #include <functional>
 
 class CDataFrameMockAnalysisState final : public ml::api::CDataFrameAnalysisInstrumentation {
+public:
+    CDataFrameMockAnalysisState(const std::string& jobId)
+        : ml::api::CDataFrameAnalysisInstrumentation(jobId) {}
+
 protected:
     ml::counter_t::ECounterTypes memoryCounterType() override;
 };

lib/api/unittest/Makefile

@@ -26,6 +26,7 @@ SRCS=\
 	CConfigUpdaterTest.cc \
 	CCsvInputParserTest.cc \
 	CCsvOutputWriterTest.cc \
+	CDataFrameAnalysisInstrumentationTest.cc \
 	CDataFrameAnalysisRunnerTest.cc \
 	CDataFrameAnalysisSpecificationTest.cc \
 	CDataFrameAnalyzerFeatureImportanceTest.cc \