ProvDBpruneInterface can now be provided with a prune_type. The associated collection is scanned and events *not* of this type are removed. Added unit test form Normal prune_type

ProvDBtester now cleans up its unqlite files on destruction, allowing multiple instances to be created in a single executable (not at the same time though!)

Author: giltirn

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

@@ -15,7 +15,11 @@ namespace chimbuko {
        */
       class ProvDBpruneInterface: public ADDataInterface{
       public:
-	ProvDBpruneInterface(const ADOutlier &ad, sonata::Database &db);
+	/**
+	 * @brief Constructor
+	 * @param prune_type The class/type of recorded event that is being pruned. Events *not* of this type will be removed from the collection after reevaluation
+	 */
+	ProvDBpruneInterface(const ADOutlier &ad, sonata::Database &db, ADDataInterface::EventType prune_type);
 
 	/**
 	 * @brief Get the values associated with each recorded anomaly
@@ -39,6 +43,7 @@ namespace chimbuko {
 	std::unique_ptr<sonata::Collection> m_collection;
 	std::unordered_map<unsigned long, std::vector<std::pair<uint64_t, double> > > m_data; //[fid] -> [  (record_id, value), ... ]
 	const ADOutlier &m_ad; //the outlier algorithm to allow access to the model data when updating records
+	ADDataInterface::EventType m_prune_type; //which type to prune
       };    
 
     }

src/modules/performance_analysis/provdb/ProvDBprune.cpp

@@ -7,7 +7,7 @@ using namespace chimbuko::modules::performance_analysis;
 
 void ProvDBprune::pruneImplementation(ADOutlier &ad, sonata::Database &db){
   //Prune the outliers and update scores / model on remaining
-  ProvDBpruneInterface pi(ad, db);
+  ProvDBpruneInterface pi(ad, db, ADDataInterface::EventType::Outlier);
   ad.run(pi);
 
   //Prune normal execs and update scores / model on remaining

src/modules/performance_analysis/provdb/ProvDBpruneInterface.cpp

@@ -6,12 +6,13 @@
 using namespace chimbuko;
 using namespace chimbuko::modules::performance_analysis;
 
-ProvDBpruneInterface::ProvDBpruneInterface(const ADOutlier &ad, sonata::Database &db): m_database(db), m_ad(ad), ADDataInterface(){
-  m_collection.reset(new sonata::Collection(db.open("anomalies")));
+ProvDBpruneInterface::ProvDBpruneInterface(const ADOutlier &ad, sonata::Database &db, ADDataInterface::EventType prune_type): m_database(db), m_ad(ad), m_prune_type(prune_type), ADDataInterface(){
+  std::string coll = m_prune_type == ADDataInterface::EventType::Outlier ? "anomalies" : "normalexecs";
+  m_collection.reset(new sonata::Collection(db.open(coll)));
   //To avoid loading all items into memory we must loop over the database using its jx9 interface
   std::string script = R"(
 $ret = [];
-while( ($rec = db_fetch('anomalies')) != NULL ){
+while( ($rec = db_fetch(')" + coll + R"(')) != NULL ){
   $r = [ $rec.__id, $rec.fid, $rec.runtime_exclusive ];
   array_push($ret, $r);
 }
@@ -52,8 +53,10 @@ void ProvDBpruneInterface::recordDataSetLabelsInternal(const std::vector<Elem> &
   std::vector<uint64_t> to_update;
   std::vector<double> update_scores;
 
+  ADDataInterface::EventType type_to_remove = m_prune_type == ADDataInterface::EventType::Outlier ? ADDataInterface::EventType::Normal : ADDataInterface::EventType::Outlier;
+
   for(auto const &e: data){ //we used the record id as the index, which is unique
-    if(e.label == ADDataInterface::EventType::Normal){
+    if(e.label == type_to_remove){
       verboseStream << "Pruning record " << e.index << " with value " << e.value << " and score " << e.score << std::endl;
       to_prune.push_back(e.index);
     }else{

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

@@ -9,7 +9,7 @@
 using namespace chimbuko;
 using namespace chimbuko::modules::performance_analysis;
 
-TEST(TestProvDBpruneOutlierInterface, basic){  
+TEST(TestProvDBpruneOutlierInterface, pruneNormalFromAnomalies){  
   int nshards = 1;
   ProvDBmoduleSetup setup;
   ProvDBtester pdb(nshards, setup);
@@ -52,17 +52,94 @@ TEST(TestProvDBpruneOutlierInterface, basic){
     std::unique_ptr<ADOutlier> ad(ADOutlier::set_algorithm(0,"sstd",ap));
     ad->setGlobalParameters(param.serialize()); //input model
     ad->setGlobalModelSyncFrequency(0); //fix model
-    ProvDBpruneInterface pi(*ad, pdb.getShard(0));
+    ProvDBpruneInterface pi(*ad, pdb.getShard(0), ADDataInterface::EventType::Outlier);
     ad->run(pi);
   }
 
   auto all_data = main_client.retrieveAllData("anomalies");
   EXPECT_EQ(all_data.size(), 2);
   for(auto const &e : all_data){
     nlohmann::json je = nlohmann::json::parse(e);
-    EXPECT_GE(je["runtime_exclusive"].template get<double>(), 1000);
     std::cout << je.dump(4) << std::endl;
+    double runtime = je["runtime_exclusive"].template get<double>();
+    EXPECT_GE(runtime, 1000);
+
+    //Check scores
+    double expect_score = (runtime - mean)/stddev; //sstd
+    EXPECT_NEAR(je["outlier_score"].template get<double>(), expect_score, 1e-5);
+    
+    //Check model
+    EXPECT_NEAR(je["algo_params"]["mean"].template get<double>(), mean, 1e-8);
+    EXPECT_NEAR(je["algo_params"]["stddev"].template get<double>(), stddev, 1e-8);
+  }
+
+}
+
+TEST(TestProvDBpruneOutlierInterface, pruneAnomaliesFromNormal){  
+  int nshards = 1;
+  ProvDBmoduleSetup setup;
+  ProvDBtester pdb(nshards, setup);
+
+  ADProvenanceDBclient main_client(setup.getMainDBcollections(), 0);
+  main_client.connectSingleServer(pdb.getAddr(),nshards);
+
+  nlohmann::json anom;
+  anom["runtime_exclusive"] = 100;
+  anom["fid"] = 1234;
+  anom["blah"] = "norm";
+
+  main_client.sendData(anom, "normalexecs");
+
+  anom["runtime_exclusive"] = 1000;
+  anom["fid"] = 1234;
+  anom["blah"] = "real_anom1";
+
+  main_client.sendData(anom, "normalexecs");
+
+  anom["runtime_exclusive"] = 1200;
+  anom["fid"] = 1234;
+  anom["blah"] = "real_anom2";
+
+  main_client.sendData(anom, "normalexecs");
+
+  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);
+  
+  EXPECT_EQ( main_client.retrieveAllData("normalexecs").size(), 3 );
+
+  ADOutlier::AlgoParams ap; ap.sstd_sigma = 5;
+  {
+    std::unique_ptr<ADOutlier> ad(ADOutlier::set_algorithm(0,"sstd",ap));
+    ad->setGlobalParameters(param.serialize()); //input model
+    ad->setGlobalModelSyncFrequency(0); //fix model
+    ProvDBpruneInterface pi(*ad, pdb.getShard(0), ADDataInterface::EventType::Normal);
+    ad->run(pi);
+  }
+
+  auto all_data = main_client.retrieveAllData("normalexecs");
+  EXPECT_EQ(all_data.size(), 1);
+  for(auto const &e : all_data){
+    nlohmann::json je = nlohmann::json::parse(e);
+    std::cout << je.dump(4) << std::endl;
+    double runtime = je["runtime_exclusive"].template get<double>();
+    EXPECT_EQ(runtime, 100);
+
+    //Check scores
+    double expect_score = (runtime - mean)/stddev; //sstd
+    EXPECT_NEAR(je["outlier_score"].template get<double>(), expect_score, 1e-5);
+    
+    //Check model
+    EXPECT_NEAR(je["algo_params"]["mean"].template get<double>(), mean, 1e-8);
+    EXPECT_NEAR(je["algo_params"]["stddev"].template get<double>(), stddev, 1e-8);
   }
 
 }
 
+
+

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

@@ -3,6 +3,7 @@
 #include<chimbuko/core/util/string.hpp>
 #include<unistd.h>
 #include<sys/types.h>
+#include<stdio.h>
 
 using namespace chimbuko;
 
@@ -17,15 +18,17 @@ ProvDBtester::ProvDBtester(int nshards, const ProvDBmoduleSetupCore &setup){
   pid_t pid = getpid();
 
   std::string glob_db_name = "provdb.global";
-  std::string glob_db_config = stringize("{ \"path\" : \"/tmp/%s.%lu.unqlite\" }", glob_db_name.c_str(),pid);
+  std::string fn = stringize("/tmp/%s.%lu.unqlite", glob_db_name.c_str(),pid); filenames.push_back(fn);
+  std::string glob_db_config = stringize("{ \"path\" : \"%s\" }", fn.c_str());
   admin->createDatabase(addr, 0, glob_db_name, "unqlite", glob_db_config);
 
   db_shard_names.resize(nshards);
   shard_providers.resize(nshards);
   for(int s=0;s<nshards;s++){
     shard_providers[s] = new sonata::Provider(engine, s+1); 
     std::string db_name = stringize("provdb.%d",s);
-    std::string config = stringize("{ \"path\" : \"/tmp/%s.%lu.unqlite\" }", db_name.c_str(),pid);
+    fn = stringize("/tmp/%s.%lu.unqlite", db_name.c_str(),pid); filenames.push_back(fn);
+    std::string config = stringize("{ \"path\" : \"%s\" }", fn.c_str());
     admin->createDatabase(addr, s+1, db_name, "unqlite", config);
     db_shard_names[s] = db_name;
   }
@@ -50,4 +53,5 @@ ProvDBtester::~ProvDBtester(){
   delete admin;
   delete glob_provider;
   for(int i=0;i<shard_providers.size();i++) delete shard_providers[i];
+  for(auto const &fn : filenames) assert(remove(fn.c_str()) == 0);
 }

test/unit_tests/modules/performance_analysis/provdb/ProvDBtester.hpp

@@ -16,6 +16,7 @@ namespace chimbuko{
     std::vector<std::string> db_shard_names;
     std::vector<sonata::Database> db;
     sonata::Database glob_db;
+    std::vector<std::string> filenames;
   public:
     ProvDBtester(int nshards, const ProvDBmoduleSetupCore &setup);