Added unit tests for ADAnomalyProvenance::getProvenanceEntries

To ADAnomalyProvenance, added the ability to set a minimum runtime for anomalies recorded to the provenance database + unit test
Integrated the above into main Chimbuko class
Added cmdline options to set the anomaly minimum runtime for provenance output to driver and driver_multirank

Author: giltirn

app/driver.cpp

@@ -113,6 +113,8 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, step_report_freq, "Set the steps between Chimbuko reporting IO step progress. Use 0 to deactivate this logging entirely (default 1)");
     addOptionalCommandLineArg(p, prov_record_startstep, "If != -1, the IO step on which to start recording provenance information for anomalies (for testing, default -1)");
     addOptionalCommandLineArg(p, prov_record_stopstep, "If != -1, the IO step on which to stop recording provenance information for anomalies (for testing, default -1)");
+    addOptionalCommandLineArg(p, prov_min_anom_time, "Set the minimum exclusive runtime (in microseconds) for anomalies to recorded in the provenance output (default 0)");
+
     addOptionalCommandLineArg(p, analysis_step_freq, "Set the frequency in IO steps between analyzing the data. Data will be accumulated over intermediate steps. (default 1)");
     addOptionalCommandLineArg(p, monitoring_watchlist_file, "Provide a filename containing the counter watchlist for the integration with the monitoring plugin. Empty string (default) uses the default subset. File format is JSON: \"[ [<COUNTER NAME>, <FIELD NAME>], ... ]\" where COUNTER NAME is the name of the counter in the input data stream and FIELD NAME the name of the counter in the provenance output.");
     addOptionalCommandLineArg(p, monitoring_counter_prefix, "Provide an optional prefix marking a set of monitoring plugin counters to be captured, on top of or superseding the watchlist. Empty string (default) is ignored.");

app/driver_multirank.cpp

@@ -84,6 +84,7 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, step_report_freq, "Set the steps between Chimbuko reporting IO step progress. Use 0 to deactivate this logging entirely (default 1)");
     addOptionalCommandLineArg(p, prov_record_startstep, "If != -1, the IO step on which to start recording provenance information for anomalies (for testing, default -1)");
     addOptionalCommandLineArg(p, prov_record_stopstep, "If != -1, the IO step on which to stop recording provenance information for anomalies (for testing, default -1)");
+    addOptionalCommandLineArg(p, prov_min_anom_time, "Set the minimum exclusive runtime (in microseconds) for anomalies to recorded in the provenance output (default 0)");
     addOptionalCommandLineArg(p, analysis_step_freq, "Set the frequency in IO steps between analyzing the data. Data will be accumulated over intermediate steps. (default 1)");      
     addOptionalCommandLineArg(p, read_ignored_corrid_funcs, "Set path to a file containing functions (one per line) for which the correlation ID counter should be ignored. If an empty string (default) no IDs will be ignored");
     addOptionalCommandLineArg(p, monitoring_watchlist_file, "Provide a filename containing the counter watchlist for the integration with the monitoring plugin. Empty string (default) uses the default subset. File format is JSON: \"[ [<COUNTER NAME>, <FIELD NAME>], ... ]\" where COUNTER NAME is the name of the counter in the input data stream and FIELD NAME the name of the counter in the provenance output.");

include/chimbuko/ad/ADAnomalyProvenance.hpp

@@ -41,6 +41,8 @@ namespace chimbuko{
      * @param step The io step
      * @param first_event_ts The timestamp of the first event in the io step
      * @param last_event_ts The timestamp of the last event in the io step
+     *
+     * Note the minimum anomaly time for recorded data does not apply to this call
      */
     nlohmann::json getEventProvenance(const ExecData_t &call,
 				      const int step,
@@ -52,6 +54,13 @@ namespace chimbuko{
      */
     void setWindowSize(const int sz){ m_window_size = sz; }
 
+    /**
+     * @brief Set the minimum exclusive runtime (in microseconds) for recorded anomalies (default 0) 
+     *
+     * Anomalies with exclusive runtime less than this will not have their data recorded
+     */
+    void setMinimumAnomalyTime(const unsigned long to){ m_min_anom_time = to; }
+
     /**
      * @brief If linked, performance information will be gathered
      */
@@ -135,6 +144,7 @@ namespace chimbuko{
     int m_window_size; /**< The number of events either side of the anomaly to capture in the window*/
     ADMonitoring const *m_monitoring; /**< Node state information from TAU's monitoring plugin*/
     ParamInterface const *m_algo_params; /**< The algorithm parameters*/
+    unsigned long m_min_anom_time; /**< Anomalies with exclusive runtime less than this will not have their data recorded*/
 
     ADNormalEventProvenance m_normalevents; /**< Maintain information on a selection of normal events*/
   };

include/chimbuko/chimbuko.hpp

@@ -48,6 +48,7 @@ namespace chimbuko {
 #endif
     int prov_record_startstep; /**< If != -1, the IO step on which to start recording provenance information for anomalies */
     int prov_record_stopstep; /**< If != -1, the IO step on which to stop recording provenance information for anomalies */
+    unsigned long prov_min_anom_time; /**< The minimum exclusive runtime (in microseconds) for anomalies recorded in the provenance output (default 0) */
 
     unsigned int anom_win_size; /**< When anomaly data are recorded, a window of this size (in units of events) around the anomalous event are also recorded (used both for viz and provDB)*/
 

src/ad/ADAnomalyProvenance.cpp

@@ -5,7 +5,7 @@
 
 using namespace chimbuko;
 
-ADAnomalyProvenance::ADAnomalyProvenance(const ADEventIDmap &event_man): m_perf(nullptr), m_event_man(&event_man), m_monitoring(nullptr), m_metadata(nullptr), m_algo_params(nullptr), m_window_size(5){}
+ADAnomalyProvenance::ADAnomalyProvenance(const ADEventIDmap &event_man): m_perf(nullptr), m_event_man(&event_man), m_monitoring(nullptr), m_metadata(nullptr), m_algo_params(nullptr), m_window_size(5), m_min_anom_time(0){}
 
 
 inline nlohmann::json getCallStackEntry(const ExecData_t &call){
@@ -225,23 +225,24 @@ void ADAnomalyProvenance::getProvenanceEntries(std::vector<nlohmann::json> &anom
 
   //Gather provenance of anomalies and for each one try to obtain a normal execution
   timer.start();
-  anom_event_entries.resize(anomalies.nEvents(Anomalies::EventType::Outlier));
-  size_t i=0;
   std::unordered_set<unsigned long> normal_event_fids;
 
   for(auto anom_it : anomalies.allEvents(Anomalies::EventType::Outlier)){
     timer2.start();
-    anom_event_entries[i++] = getEventProvenance(*anom_it, step, first_event_ts, last_event_ts);
+    if(anom_it->get_exclusive() < m_min_anom_time) continue; //skip executions with too short runtimes to avoid filling the database with irrelevant anomalies
+
+    anom_event_entries.push_back(getEventProvenance(*anom_it, step, first_event_ts, last_event_ts));
     if(m_perf) m_perf->add("ad_extract_send_prov_anom_data_generation_per_anom_ms", timer2.elapsed_ms());
 
     //Get the associated normal event if one has not been recorded for this function on this step
     if(!normal_event_fids.count(anom_it->get_fid())){
+      timer2.start();
       //if normal event not available put into the list of outstanding requests and it will be recorded next time a normal event for this function is obtained
       //if normal event is available, delete internal copy within m_normalevent_prov so the normal event isn't added more than once
-      timer2.start();
       auto nev = m_normalevents.getNormalEvent(anom_it->get_pid(), anom_it->get_rid(), anom_it->get_tid(), anom_it->get_fid(), add_outstanding, do_delete);
       if(nev.second) normal_event_entries.push_back(std::move(nev.first));
-      normal_event_fids.insert(anom_it->get_fid());
+      
+      normal_event_fids.insert(anom_it->get_fid()); //make sure we don't record more than one normal event for this fid
       if(m_perf) m_perf->add("ad_extract_send_prov_normalevent_gather_per_anom_ms", timer2.elapsed_ms());
     }
 

src/chimbuko.cpp

@@ -37,7 +37,8 @@ ChimbukoParams::ChimbukoParams(): rank(-1234),  //not set!
                                   func_threshold_file(""),
                                   ignored_func_file(""),
                                   monitoring_watchlist_file(""),
-                                  monitoring_counter_prefix("")
+                                  monitoring_counter_prefix(""),
+                                  prov_min_anom_time(0)
 {}
 
 void ChimbukoParams::print() const{
@@ -316,6 +317,7 @@ void Chimbuko::init_provenance_gatherer(){
   m_anomaly_provenance->linkMonitoring(m_monitoring);
   m_anomaly_provenance->linkMetadata(m_metadata_parser);
   m_anomaly_provenance->setWindowSize(m_params.anom_win_size);
+  m_anomaly_provenance->setMinimumAnomalyTime(m_params.prov_min_anom_time);
   m_ptr_registry.registerPointer(m_anomaly_provenance);
 }
 

test/unit_tests/ad/ADAnomalyProvenance.cpp

@@ -499,3 +499,105 @@ TEST(TestADAnomalyProvenance, extractsHostname){
   nlohmann::json output = prov.getEventProvenance(exec1,8,800,1200);  
   EXPECT_EQ(output["hostname"], "TheHost");
 }
+
+
+template<typename T>
+T & listEntry(std::list<T> &l, int i){
+  auto it = std::next(l.begin(),i);
+  return *it;
+}
+
+TEST(TestADAnomalyProvenance, getProvenanceEntries){
+  std::vector<ExecData_t> events = {
+    createFuncExecData_t(1,2,3, 55, "theparent", 800, 200),
+    createFuncExecData_t(1,2,3, 33, "thefunc", 900, 100),
+    createFuncExecData_t(1,2,4, 11, "theotherparent", 1100, 200),
+    createFuncExecData_t(1,2,4, 22, "theotherfunc", 1150, 50)
+  };
+  bindParentChild(events[0],events[1]);
+  bindParentChild(events[2],events[3]);
+  events[1].set_label(-1);
+  events[3].set_label(-1);
+  
+  ADEvent event_man;
+  std::vector<CallListIterator_t> event_its;
+  for(auto &e :  events) event_its.push_back(event_man.addCall(e));
+
+  Anomalies anoms;
+  anoms.insert(event_its[1],Anomalies::EventType::Outlier);
+  anoms.insert(event_its[3],Anomalies::EventType::Outlier);
+  
+  std::vector<nlohmann::json> anom_entries, normal_entries;
+  {
+    ADAnomalyProvenance prov(event_man);
+    prov.getProvenanceEntries(anom_entries, normal_entries, anoms, 0, 800, 1200);
+  }
+
+  ASSERT_EQ(anom_entries.size(),2);
+  ASSERT_EQ(normal_entries.size(),0); //didn't put in normal events
+
+  std::string got, expect;
+  
+  got = anom_entries[0]["event_id"]; expect = event_its[1]->get_id().toString();
+  EXPECT_EQ(got,expect);  
+  got = anom_entries[1]["event_id"]; expect = event_its[3]->get_id().toString();
+  EXPECT_EQ(got,expect);
+
+  //Repeat but add normal events for both
+  events.push_back( createFuncExecData_t(1,2,3, 33, "thefunc", 700, 50) );
+  events.push_back( createFuncExecData_t(1,2,4, 22, "theotherfunc", 400, 25) );
+  event_its.push_back(event_man.addCall(events[4]));
+  event_its.push_back(event_man.addCall(events[5]));
+  anoms.insert(event_its[4],Anomalies::EventType::Normal);
+  anoms.insert(event_its[5],Anomalies::EventType::Normal);
+
+  anom_entries.clear();
+  normal_entries.clear();
+
+  { //create prov anew to ensure normal event logic works as expected
+    ADAnomalyProvenance prov(event_man);
+    prov.getProvenanceEntries(anom_entries, normal_entries, anoms, 0, 800, 1200);
+  }
+  ASSERT_EQ(anom_entries.size(),2);
+  ASSERT_EQ(normal_entries.size(),2);
+
+  got = normal_entries[0]["event_id"]; expect = event_its[4]->get_id().toString();
+  EXPECT_EQ(got,expect);
+
+  got = normal_entries[1]["event_id"]; expect = event_its[5]->get_id().toString();
+  EXPECT_EQ(got,expect);
+
+  //Test the minimum runtime
+  anom_entries.clear();
+  normal_entries.clear();
+
+  { //create prov anew to ensure normal event logic works as expected
+    ADAnomalyProvenance prov(event_man);
+    prov.setMinimumAnomalyTime(60); //exclude second anomaly
+    prov.getProvenanceEntries(anom_entries, normal_entries, anoms, 0, 800, 1200);
+  }
+
+  ASSERT_EQ(anom_entries.size(),1);
+  ASSERT_EQ(normal_entries.size(),1);
+  
+  got = anom_entries[0]["event_id"]; expect = event_its[1]->get_id().toString();
+  EXPECT_EQ(got,expect);  
+  got = normal_entries[0]["event_id"]; expect = event_its[4]->get_id().toString();
+  EXPECT_EQ(got,expect);  
+
+
+  //Check that if we have multiple anomalies for the same function we only get one normal event
+  events.push_back( createFuncExecData_t(1,2,3, 33, "thefunc", 700, 75) );
+  event_its.push_back(event_man.addCall(events.back()));
+  anoms.insert(event_its.back(),Anomalies::EventType::Outlier);
+
+  anom_entries.clear();
+  normal_entries.clear();
+  
+  {
+    ADAnomalyProvenance prov(event_man);
+    prov.getProvenanceEntries(anom_entries, normal_entries, anoms, 0, 800, 1200);
+  }
+  ASSERT_EQ(anom_entries.size(),3);
+  ASSERT_EQ(normal_entries.size(),2);
+}