In ProvDBpruneCore/ProvDBprune, algorithm initialization now performed in constructor (changes to factory function to support)

Fixed PSshardProvenanceDBclient not defining handshakes (use empty function body as not needed)
Added a unit test for ProvDBprune that tests pruning and stats rebuild for multiple shards

Author: giltirn

include/chimbuko/core/provdb/ProvDBpruneCore.hpp

@@ -10,7 +10,9 @@ namespace chimbuko {
 
   class ProvDBpruneCore{
   public:
-    void prune(const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &params_ser, sonata::Database &db);
+    ProvDBpruneCore(const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &model_ser);
+
+    void prune(sonata::Database &db);
 
     /**
      * @brief Module implementation of pruning given the database shard. This same instance of ProvDBpruneCore will be called in turn for all shards
@@ -23,6 +25,8 @@ namespace chimbuko {
      */
     virtual void finalize(sonata::Database &global_db){}
 
+  public:
+    std::unique_ptr<ADOutlier> m_outlier;
   };  
 
 }

include/chimbuko/core/pserver/PSshardProvenanceDBclient.hpp

@@ -22,6 +22,17 @@ namespace chimbuko{
      * @param addr_file_dir The directory containing the address files created by the provDB
      */
     void connectShard(const std::string &addr_file_dir, int shard, int nshards, int ninstances);
+
+    /**
+     * @brief No handshake is needed
+     */
+    void handshakeHello(thallium::engine &eng, thallium::endpoint &server) override{};
+
+    /**
+     * @brief No handshake is needed.
+     */
+    void handshakeGoodbye(thallium::engine &eng, thallium::endpoint &server) override{};
+
   };
 
 }

include/chimbuko/modules/factory.hpp

@@ -19,8 +19,12 @@ namespace chimbuko{
 
     /**
      *@brief A factory function for ProvDBpruneCore instances
+     *@param algorithm The AD algorithm
+     *@param algo_params Parameters for the algoritm
+     *@param model_ser The serialized model
      */   
-    std::unique_ptr<ProvDBpruneCore> factoryInstantiateProvDBprune(const std::string &module); 
+    std::unique_ptr<ProvDBpruneCore> factoryInstantiateProvDBprune(const std::string &module, 
+								   const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &model_ser); 
 
     /**
      *@brief A factory function for PSmoduleDataManager instances

include/chimbuko/modules/performance_analysis/provdb/ProvDBprune.hpp

@@ -14,6 +14,8 @@ namespace chimbuko {
 
       class ProvDBprune: public ProvDBpruneCore{
       public:  
+	ProvDBprune(const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &model_ser): ProvDBpruneCore(algorithm,algo_params,model_ser){}
+
 	/**
 	 * @brief Prune the database shard. Both the anomalies and normalexecs will be updated
 	 */

src/core/provdb/ProvDBpruneCore.cpp

@@ -4,10 +4,11 @@
 
 using namespace chimbuko;
 
-void ProvDBpruneCore::prune(const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &params_ser, sonata::Database &db){
-  std::unique_ptr<ADOutlier> ad(ADOutlier::set_algorithm(0,algorithm,algo_params));
-  ad->setGlobalParameters(params_ser); //input model
-  ad->setGlobalModelSyncFrequency(0); //fix model
-
-  this->pruneImplementation(*ad, db);
+ProvDBpruneCore::ProvDBpruneCore(const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &model_ser): m_outlier(ADOutlier::set_algorithm(0,algorithm,algo_params)){
+  m_outlier->setGlobalParameters(model_ser); //input model
+  m_outlier->setGlobalModelSyncFrequency(0); //fix model
+}
+  
+void ProvDBpruneCore::prune(sonata::Database &db){
+  this->pruneImplementation(*m_outlier, db);
 }

src/modules/factory.cpp

@@ -29,9 +29,10 @@ std::unique_ptr<chimbuko::PSmoduleDataManagerCore> chimbuko::modules::factoryIns
   }
 }
 
-std::unique_ptr<chimbuko::ProvDBpruneCore> chimbuko::modules::factoryInstantiateProvDBprune(const std::string &module){
+std::unique_ptr<chimbuko::ProvDBpruneCore> chimbuko::modules::factoryInstantiateProvDBprune(const std::string &module, 
+											    const std::string &algorithm, const ADOutlier::AlgoParams &algo_params, const std::string &model_ser){
   if(module == "performance_analysis"){
-    return std::unique_ptr<ProvDBpruneCore>(new performance_analysis::ProvDBprune);
+    return std::unique_ptr<ProvDBpruneCore>(new performance_analysis::ProvDBprune(algorithm,algo_params,model_ser) );
   }else{
     fatal_error("Unknown module");
   }

test/unit_tests/modules/performance_analysis/provdb/Makefile.am

@@ -2,7 +2,10 @@ AM_CPPFLAGS = -I$(top_srcdir)/include -I$(top_srcdir)/3rdparty @PS_FLAGS@
 LDADD = $(top_builddir)/src/libchimbuko.la -lgtest -lstdc++fs
 
 testdir = $(prefix)/test/unit_tests/modules/performance_analysis/provdb
-test_PROGRAMS = ProvDBpruneInterface
+test_PROGRAMS = ProvDBpruneInterface ProvDBprune
 
 ProvDBpruneInterface_SOURCES = ProvDBpruneInterface.cpp ../../../unit_test_main_mpi.cpp ProvDBtester.cpp
 ProvDBpruneInterface_LDADD = $(LDADD)
+
+ProvDBprune_SOURCES = ProvDBprune.cpp ../../../unit_test_main_mpi.cpp ProvDBtester.cpp
+ProvDBprune_LDADD = $(LDADD)

test/unit_tests/modules/performance_analysis/provdb/ProvDBprune.cpp

@@ -0,0 +1,125 @@
+#include<chimbuko/modules/performance_analysis/provdb/ProvDBprune.hpp>
+#include<chimbuko/modules/performance_analysis/provdb/ProvDBmoduleSetup.hpp>
+#include<chimbuko/core/pserver/PSshardProvenanceDBclient.hpp>
+#include<chimbuko/core/pserver/PSglobalProvenanceDBclient.hpp>
+
+#include "gtest/gtest.h"
+#include "../../../unit_test_common.hpp"
+#include "ProvDBtester.hpp"
+
+using namespace chimbuko;
+using namespace chimbuko::modules::performance_analysis;
+
+TEST(TestProvDBprune, works){  
+  int nshards = 2;
+  ProvDBmoduleSetup setup;
+  ProvDBtester pdb(nshards, setup);
+
+  {
+    std::ofstream of("/tmp/provider.address.0");
+    of << pdb.getAddr();
+  }
+  std::vector<std::unique_ptr<PSshardProvenanceDBclient> > shard_clients(nshards);
+  for(int i=0;i<nshards;i++){
+    shard_clients[i].reset(new PSshardProvenanceDBclient(setup.getMainDBcollections()));
+    shard_clients[i]->connectShard("/tmp",i,nshards,1);
+  }
+  
+  PSglobalProvenanceDBclient glob_client(setup.getGlobalDBcollections());
+  glob_client.connectServer(pdb.getAddr());
+
+  //Put some anomaly data on the shards. Use both shards to check aggregation 
+  nlohmann::json anom, norm;
+
+  //event that should be removed
+  norm["runtime_exclusive"] = 100;
+  norm["fid"] = 1234;
+  norm["blah"] = "norm";
+
+  shard_clients[0]->sendData(norm, "anomalies");
+
+  //event that should be kept
+  anom["runtime_exclusive"] = 1000;
+  anom["fid"] = 1234;
+  anom["blah"] = "real_anom1";
+  anom["entry"] = 33; //need this info to gather anomaly metrics on kept anomalies
+  anom["exit"] = 1033;
+  anom["io_step"] = 13;
+  anom["outlier_severity"] = 1000;
+  anom["rid"] = 88;
+
+  shard_clients[0]->sendData(anom, "anomalies");
+
+  //event that should be removed
+  norm["runtime_exclusive"] = 103;
+  norm["fid"] = 1234;
+  norm["blah"] = "norm2";
+
+  shard_clients[1]->sendData(norm, "anomalies");
+
+  //event that should be kept
+  anom["runtime_exclusive"] = 1200;
+  anom["fid"] = 1234;
+  anom["blah"] = "real_anom2";
+  anom["entry"] = 44;
+  anom["exit"] = 1244;
+  anom["io_step"] = 17;
+  anom["outlier_severity"] = 1200;
+  anom["rid"] = 88;
+
+  shard_clients[1]->sendData(anom, "anomalies");
+
+  //populate the global database
+  nlohmann::json fstats;
+  fstats["fid"] = 1234;
+  glob_client.sendData(fstats, "func_stats");
+
+  double mean = 100;
+  double stddev = 10;
+  int count = 1000;
+  
+  SstdParam param;
+  param[1234].set_eta(mean);
+  param[1234].set_rho(pow(stddev,2) * (count-1) );
+  param[1234].set_count(count);
+  ADOutlier::AlgoParams ap; ap.sstd_sigma = 5;
+
+  //Do the business
+  ProvDBprune pruner("sstd", ap, param.serialize());
+  for(int i=0;i<nshards;i++) pruner.prune(shard_clients[i]->getDatabase());
+  pruner.finalize(glob_client.getDatabase());
+
+  //check the shards
+  {
+    auto sdata = shard_clients[0]->retrieveAllData("anomalies");
+    EXPECT_EQ(sdata.size(),1);
+    auto s = nlohmann::json::parse(sdata[0]);
+    EXPECT_NEAR(s["outlier_score"].template get<double>(), 90, 1e-5);
+  }
+  {
+    auto sdata = shard_clients[1]->retrieveAllData("anomalies");
+    EXPECT_EQ(sdata.size(),1);
+    auto s = nlohmann::json::parse(sdata[0]);
+    EXPECT_NEAR(s["outlier_score"].template get<double>(), 110, 1e-5);
+  }
+  
+  //check the global database
+  auto glob_data = glob_client.retrieveAllData("func_stats");
+  EXPECT_EQ(glob_data.size(), 1);
+  for(auto const &e : glob_data){
+    nlohmann::json je = nlohmann::json::parse(e);    
+    std::cout << je.dump(4);
+    
+    EXPECT_EQ(je["anomaly_metrics"]["first_io_step"].template get<int>(), 13);
+    EXPECT_EQ(je["anomaly_metrics"]["last_io_step"].template get<int>(), 17);
+    EXPECT_EQ(je["anomaly_metrics"]["min_timestamp"].template get<unsigned long>(), 33);
+    EXPECT_EQ(je["anomaly_metrics"]["max_timestamp"].template get<unsigned long>(), 1244);
+    EXPECT_EQ(je["anomaly_metrics"]["score"]["count"].template get<int>(), 2);
+    EXPECT_EQ(je["anomaly_metrics"]["severity"]["count"].template get<int>(), 2);
+    EXPECT_EQ(je["anomaly_metrics"]["severity"]["accumulate"].template get<unsigned long>(), 2200);
+    EXPECT_EQ(je["anomaly_metrics"]["anomaly_count"]["count"].template get<int>(), 2);  //2 different timesteps
+    EXPECT_EQ(je["anomaly_metrics"]["anomaly_count"]["accumulate"].template get<unsigned long>(), 2);
+  }
+}
+
+

test/unit_tests/run_all.sh

@@ -46,6 +46,7 @@ set -o pipefail
 ./modules/performance_analysis/ad/COPODOutlier
 ./modules/performance_analysis/ad/FuncAnomalyMetrics
 ./modules/performance_analysis/provdb/ProvDBpruneInterface
+./modules/performance_analysis/provdb/ProvDBprune
 ./modules/performance_analysis/pserver/GlobalCounterStats
 ./modules/performance_analysis/pserver/GlobalAnomalyStats
 ./modules/performance_analysis/pserver/GlobalAnomalyMetrics