Reworked the way that normal events are recorded: The decision as to whether to record a normal event is now made by the Anomalies class rather than by each ADOutlier derived class. Now all outlier algorithms will always have the normal event with the lowest score recorded, rather than just for HBOS

Added dedicated unit tests for Anomalies class
Anomalies class checks inserted events have the expected label (had to fix a lot of unit tests!)
The number of normal events in the Chimbuko log output now reflects the total number of normal events and not just the selection that were recorded
Disabled the ADMonitoring check that counters are on thread 0 because a recent TAU change has made this no longer true!

Author: giltirn

include/chimbuko/util/Anomalies.hpp

@@ -12,37 +12,52 @@ namespace chimbuko{
   public:
     enum class EventType { Outlier, Normal };
 
+    Anomalies(): m_n_events_total(0){}
+
+    /**
+     * @brief Record an anomalous event
+     */
+    void recordAnomaly(CallListIterator_t event);
+
     /**
-     * @brief Insert a detected outlier/normal execution
+     * @brief Record a normal event if it has the lowest score for its function
+     * @return true if the event was recorded, false otherwise
      */
-    void insert(CallListIterator_t event, EventType type);
+    bool recordNormalEventConditional(CallListIterator_t event);
+   
+    /**
+     * @brief Get all recorded outliers/normal events
+     */
+    const std::vector<CallListIterator_t> & allEventsRecorded(EventType type) const{ return type == EventType::Outlier ? m_all_outliers : m_all_normal_execs; }
 
     /**
-     * @brief Get the outlier/normal events associated with a given function
+     * @brief Get number of outliers/normal events recorded
+     *
+     * Note: This is not all of the normal events, only the selection of normal events that we keep for comparison purposes
      */
-    const std::vector<CallListIterator_t> & funcEvents(const unsigned long func_id, EventType type) const;
+    size_t nEventsRecorded(EventType type) const { return allEventsRecorded(type).size(); }
 
     /**
-     * @brief Get all outliers/normal events
+     * @brief  Get the total number of events analyzed (both recorded and unrecorded)
      */
-    const std::vector<CallListIterator_t> & allEvents(EventType type) const{ return type == EventType::Outlier ? m_all_outliers : m_all_normal_execs; }
+    size_t nEvents() const{ return m_n_events_total; }
 
     /**
-     * @brief Get number of outliers/normal events associated with a given function
+     * @brief Count the number of events recorded of a given type for a particular function
      */
-    size_t nFuncEvents(const unsigned long func_id, EventType type) const{ return funcEvents(func_id, type).size();  }
+    size_t nFuncEventsRecorded(unsigned long fid, EventType type) const;
 
     /**
-     * @brief Get number of outliers/normal events
+     * @brief Return an array of iterators to events of the given type associated with a specific function
      */
-    size_t nEvents(EventType type) const { return allEvents(type).size(); }
+    std::vector<CallListIterator_t> funcEventsRecorded(unsigned long fid, EventType type) const;
 
   private:
     std::vector<CallListIterator_t> m_all_outliers; /**< Array of outliers */
-    std::unordered_map<unsigned long, std::vector<CallListIterator_t> > m_func_outliers; /**< Map of function index to associated outliers */
-
     std::vector<CallListIterator_t> m_all_normal_execs; /**< Array of normal executions (the algorithm will capture a limited number of these for comparison with outliers)*/
-    std::unordered_map<unsigned long, std::vector<CallListIterator_t> > m_func_normal_execs; /**< Map of function index to associated normal executions */
+    std::unordered_map<unsigned long, size_t> m_func_normal_exec_idx; /**< Map of function index to the index of the array entry containing the normal execution recorded for that function*/
+
+    size_t m_n_events_total; /**< Total number of events analyzed (both recorded and unrecorded)*/
   };
 
 };

sim/src/ad.cpp

@@ -265,14 +265,14 @@ void ADsim::step(const unsigned long step){
     m_outlier->linkExecDataMap(&this_step_func_execs);
     anom = m_outlier->run(step);
   }else{
-    //Collect outliers and 1 normal event/func into Anomalies object
+    //Collect outliers and 1 normal event/func into Anomalies object (Anomalies object decides which to keep)
     for(const auto &exec_it : this_step_execs){
-      if(exec_it->get_label() == -1) anom.insert(exec_it, Anomalies::EventType::Outlier);
-      else if(anom.nFuncEvents(exec_it->get_fid(), Anomalies::EventType::Normal) == 0) anom.insert(exec_it, Anomalies::EventType::Normal);
+      if(exec_it->get_label() == -1) anom.recordAnomaly(exec_it);
+      else anom.recordNormalEventConditional(exec_it);
     }
   }
-  int nanom = anom.nEvents(Anomalies::EventType::Outlier);
-  int nnorm = anom.nEvents(Anomalies::EventType::Normal);
+  int nanom = anom.nEventsRecorded(Anomalies::EventType::Outlier);
+  int nnorm = anom.nEventsRecorded(Anomalies::EventType::Normal);
   std::cout << "Step " << step << " rank " << m_rid << " Anomalies object contains : " << nanom << " anomalies and " << nnorm << " normal events (max 1)" << std::endl;
 
   //Extract provenance data and send to the provDB

src/ad/ADAnomalyProvenance.cpp

@@ -210,7 +210,7 @@ void ADAnomalyProvenance::getProvenanceEntries(std::vector<nlohmann::json> &anom
 
   //Put new normal event provenance into m_normalevent_prov
   timer.start();
-  for(auto norm_it : anomalies.allEvents(Anomalies::EventType::Normal)){
+  for(auto norm_it : anomalies.allEventsRecorded(Anomalies::EventType::Normal)){
     timer2.start();
     m_normalevents.addNormalEvent(norm_it->get_pid(), norm_it->get_rid(), norm_it->get_tid(), norm_it->get_fid(), getEventProvenance(*norm_it, step, first_event_ts, last_event_ts));
     if(m_perf) m_perf->add("ad_extract_send_prov_normalevent_update_per_event_ms", timer2.elapsed_ms());
@@ -228,7 +228,7 @@ void ADAnomalyProvenance::getProvenanceEntries(std::vector<nlohmann::json> &anom
   timer.start();
   std::unordered_set<unsigned long> normal_event_fids;
 
-  for(auto anom_it : anomalies.allEvents(Anomalies::EventType::Outlier)){
+  for(auto anom_it : anomalies.allEventsRecorded(Anomalies::EventType::Outlier)){
     timer2.start();
     if(anom_it->get_exclusive() < m_min_anom_time) continue; //skip executions with too short runtimes to avoid filling the database with irrelevant anomalies
 

src/ad/ADLocalAnomalyMetrics.cpp

@@ -15,7 +15,7 @@ ADLocalAnomalyMetrics::State::State(const ADLocalAnomalyMetrics &parent){
 
 ADLocalAnomalyMetrics::ADLocalAnomalyMetrics(int app, int rank, int step, unsigned long first_event_ts, unsigned long last_event_ts, const Anomalies &anom): 
   m_app(app), m_rank(rank), m_step(step), m_first_event_ts(first_event_ts), m_last_event_ts(last_event_ts), m_perf(nullptr){
-  const std::vector<CallListIterator_t> & outliers = anom.allEvents(Anomalies::EventType::Outlier);
+  const std::vector<CallListIterator_t> & outliers = anom.allEventsRecorded(Anomalies::EventType::Outlier);
   for(auto const &cit: outliers){
     int fid = cit->get_fid();
     auto fit = m_func_anom_metrics.find(fid);

src/ad/ADLocalFuncStatistics.cpp

@@ -59,18 +59,15 @@ void ADLocalFuncStatistics::gatherStatistics(const ExecDataMap_t* exec_data){
 }
 
 void ADLocalFuncStatistics::gatherAnomalies(const Anomalies &anom){
-  //Loop over functions and get the number of anomalies
-  for(auto &fstats: m_funcstats){
-    unsigned long func_id = fstats.second.id;
-    fstats.second.n_anomaly += anom.nFuncEvents(func_id, Anomalies::EventType::Outlier);
-  }
-
   //Gather information on the number of anomalies and stats on their scores
-  const std::vector<CallListIterator_t> &anomalies = anom.allEvents(Anomalies::EventType::Outlier);
+  const std::vector<CallListIterator_t> &anomalies = anom.allEventsRecorded(Anomalies::EventType::Outlier);
   m_anom_data.incr_n_anomalies(anomalies.size());
 
-  for(auto const &it : anomalies)
+  for(auto const &it : anomalies){
     m_anom_data.add_outlier_score(it->get_outlier_score());
+    ++m_funcstats[it->get_fid()].n_anomaly; //increment func anomalies count
+  }
+
 }
 
 ADLocalFuncStatistics::State ADLocalFuncStatistics::get_state() const{

src/ad/ADMonitoring.cpp

@@ -57,7 +57,7 @@ void ADMonitoring::extractCounters(const CountersByIndex_t &counters){
 
       CounterDataListIterator_t cdi = *vit->second.rbegin();
       const CounterData_t &cd = *cdi;
-      if(cd.get_tid() != 0) recoverable_error("Expected counter to be on thread 0!");
+      //if(cd.get_tid() != 0) recoverable_error("Expected counter to be on thread 0! Info: "+ cd.get_json().dump());    //This doesn't seem to be true any more because Kevin made changes to the monitoring plugin
       entry.value = cd.get_value();
       entry.assigned = true;
       m_timestamp = cd.get_ts(); //overwrite timestamp of update (all monitoring data in a dump should arrive at the same time)

src/ad/ADOutlier.cpp

@@ -242,16 +242,11 @@ unsigned long ADOutlierSSTD::compute_outliers(Anomalies &outliers,
 	verboseStream << "!!!!!!!Detected outlier on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid()
 		      << " runtime " << runtime << " mean " << mean << " std " << std << std::endl;
 	n_outliers += 1;
-	outliers.insert(itt, Anomalies::EventType::Outlier); //insert into data structure containing captured anomalies
-      }else{
-	//Capture maximum of one normal execution per io step
-	if(outliers.nFuncEvents(func_id, Anomalies::EventType::Normal) == 0){
-	  verboseStream << "Detected normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid()
-			<< " runtime " << runtime << " mean " << mean << " std " << std << std::endl;
-
-	  outliers.insert(itt, Anomalies::EventType::Normal);
-	}
-      }
+	outliers.recordAnomaly(itt); //insert into data structure containing captured anomalies
+      }else if(outliers.recordNormalEventConditional(itt)){  
+	verboseStream << "Recorded normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid()
+		      << " runtime " << runtime << " mean " << mean << " std " << std << std::endl;
+      }	  
     }
   }
 
@@ -460,11 +455,6 @@ unsigned long ADOutlierHBOS::compute_outliers(Anomalies &outliers,
   const double bin_width = hist.bin_edges().at(1) - hist.bin_edges().at(0);
   verboseStream << "Bin width: " << bin_width << std::endl;
 
-  //Maintain the normal execution with the lowest score (most likely)
-  bool lowest_score_set = false;
-  double lowest_score_val = std::numeric_limits<double>::max();
-  CallListIterator_t lowest_score_it;
-
   int top_out = 0;
   for (auto itt : data) {
     if (itt->get_label() == 0) {
@@ -508,31 +498,18 @@ unsigned long ADOutlierHBOS::compute_outliers(Anomalies &outliers,
       if (ad_score >= l_threshold) {
 	itt->set_label(-1);
 	verboseStream << "!!!!!!!Detected outlier on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << runtime_i << " score " << ad_score << " (threshold " << l_threshold << ")" << std::endl;
-	outliers.insert(itt, Anomalies::EventType::Outlier); //insert into data structure containing captured anomalies
+	outliers.recordAnomaly(itt); //insert into data structure containing captured anomalies
 	n_outliers += 1;
       }else {
-        //Capture maximum of one normal execution per io step
         itt->set_label(1);
-
-	//Record the normal event with the lowest score (most likely)
-	if(ad_score < lowest_score_val){
-	  lowest_score_val = ad_score;
-	  lowest_score_it = itt;
-	  lowest_score_set = true;
+	if(outliers.recordNormalEventConditional(itt)){  
+	  verboseStream << "Recorded normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << this->getStatisticValue(*itt) << " score " << itt->get_outlier_score() << " (threshold " << l_threshold << ")" << std::endl;
 	}
-
       }
 
     }//if unlabeled point
   } //loop over data points
 
-  //Record only the normal event with the lowest score
-  if(lowest_score_set){
-    auto itt = lowest_score_it;
-    verboseStream << "Recorded normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << this->getStatisticValue(*itt) << " score " << itt->get_outlier_score() << " (threshold " << l_threshold << ")" << std::endl;
-    outliers.insert(itt, Anomalies::EventType::Normal);
-  }
-
   return n_outliers;
 }
 
@@ -745,23 +722,19 @@ unsigned long ADOutlierCOPOD::compute_outliers(Anomalies &outliers,
       itt->set_outlier_score(ad_score);
       verboseStream << "runtime: " << runtime_i << " ad_score: " << ad_score << ", l_threshold: " << l_threshold << std::endl;
 
-      //Compare the ad_score with the threshold
       if (ad_score >= l_threshold) {
 	itt->set_label(-1);
-	verboseStream << "!!!!!!!Detected outlier on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << runtime_i << std::endl;
-	outliers.insert(itt, Anomalies::EventType::Outlier); //insert into data structure containing captured anomalies
-	++n_outliers;
-      }
-      else {
-        //Capture maximum of one normal execution per io step
+	verboseStream << "!!!!!!!Detected outlier on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << runtime_i << " score " << ad_score << " (threshold " << l_threshold << ")" << std::endl;
+	outliers.recordAnomaly(itt); //insert into data structure containing captured anomalies
+	n_outliers += 1;
+      }else {
         itt->set_label(1);
-        if(outliers.nFuncEvents(func_id, Anomalies::EventType::Normal) == 0) {
-      	   verboseStream << "Detected normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << runtime_i << std::endl;
-      	   outliers.insert(itt, Anomalies::EventType::Normal);
-        }
+	if(outliers.recordNormalEventConditional(itt)){  
+	  verboseStream << "Recorded normal event on func id " << func_id << " (" << itt->get_funcname() << ") on thread " << itt->get_tid() << " runtime " << this->getStatisticValue(*itt) << " score " << itt->get_outlier_score() << " (threshold " << l_threshold << ")" << std::endl;
+	}
       }
-    }
-  }
+    }//unlabeled point
+  }//data loop
 
   return n_outliers;
 }

src/chimbuko.cpp

@@ -663,10 +663,11 @@ bool Chimbuko::runFrame(unsigned long long& n_func_events,
     timer.start();
     Anomalies anomalies = m_outlier->run(step);
     m_perf.add("ad_run_anom_detection_time_ms", timer.elapsed_ms());
-    m_perf.add("ad_run_anomaly_count", anomalies.nEvents(Anomalies::EventType::Outlier));
+    m_perf.add("ad_run_anomaly_count", anomalies.nEventsRecorded(Anomalies::EventType::Outlier));
+    m_perf.add("ad_run_n_exec_analyzed", anomalies.nEvents());
 
-    int nout = anomalies.nEvents(Anomalies::EventType::Outlier);
-    int nnormal = anomalies.nEvents(Anomalies::EventType::Normal);
+    int nout = anomalies.nEventsRecorded(Anomalies::EventType::Outlier);
+    int nnormal = anomalies.nEvents() - nout; //this is the total number of normal events, not just of those that were recorded
     n_outliers += nout;
     m_n_outliers_accum_prd += nout;
 
@@ -698,7 +699,7 @@ bool Chimbuko::runFrame(unsigned long long& n_func_events,
     //Enable update of analysis time window bound on next step
     m_execdata_first_event_ts_set = false;
 
-    if(do_step_report){ headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " event analysis complete: total=" << nout + nnormal << " normal=" << nnormal << " anomalous=" << nout << std::endl; }
+    if(do_step_report){ headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " function execution analysis complete: total=" << nout + nnormal << " normal=" << nnormal << " anomalous=" << nout << std::endl; }
   }//if(do_run_analysis)
 
   m_perf.add("ad_run_total_step_time_excl_parse_ms", step_timer.elapsed_ms());

src/util/Anomalies.cpp

@@ -1,26 +1,49 @@
 #include "chimbuko/util/Anomalies.hpp"
+#include "chimbuko/util/error.hpp"
 
 using namespace chimbuko;
 
-void Anomalies::insert(CallListIterator_t event, EventType type){
-  switch(type){
-  case EventType::Outlier:
-    m_all_outliers.push_back(event);
-    m_func_outliers[event->get_fid()].push_back(event);
-    break;
-  case EventType::Normal:
+void Anomalies::recordAnomaly(CallListIterator_t event){
+  if(event->get_label() != -1){ fatal_error("Event is not an anomaly!"); }
+
+  m_all_outliers.push_back(event);
+  ++m_n_events_total; //increment counter of all events
+}
+
+bool Anomalies::recordNormalEventConditional(CallListIterator_t event){
+  if(event->get_label() != 1){ fatal_error("Event is not a normal event!"); }
+  ++m_n_events_total; //increment counter of all events
+
+  auto eit = m_func_normal_exec_idx.find(event->get_fid());
+  if(eit == m_func_normal_exec_idx.end()){ //Add the event to the array and mark its location
     m_all_normal_execs.push_back(event);
-    m_func_normal_execs[event->get_fid()].push_back(event);
-    break;
-  default:
-    throw std::runtime_error("Invalid type");
+    m_func_normal_exec_idx[event->get_fid()] = m_all_normal_execs.size()-1;
+    return true;
+  }else if(event->get_outlier_score() < m_all_normal_execs[eit->second]->get_outlier_score()){  //Replace the existing normal execution if the current has a lower score
+    m_all_normal_execs[eit->second] = event;
+    return true;
   }
+  return false;
 }
 
-const std::vector<CallListIterator_t> & Anomalies::funcEvents(const unsigned long func_id, EventType type) const{
-  static std::vector<CallListIterator_t> empty;
-  auto const &mp = type == EventType::Outlier ? m_func_outliers : m_func_normal_execs;
-  auto it = mp.find(func_id);
-  if(it != mp.end()) return it->second;
-  else return empty;
+size_t Anomalies::nFuncEventsRecorded(unsigned long fid, EventType type) const{
+  size_t out = 0;
+  if(type == EventType::Outlier){
+    for(auto it : m_all_outliers)
+      if(it->get_fid() == fid) ++out;
+  }else if(m_func_normal_exec_idx.count(fid)){
+    out = 1;
+  }
+  return out;
+}
+
+std::vector<CallListIterator_t> Anomalies::funcEventsRecorded(unsigned long fid, EventType type) const{
+  std::vector<CallListIterator_t> out;
+  if(type == EventType::Outlier){
+    for(auto it : m_all_outliers)
+      if(it->get_fid() == fid) out.push_back(it);
+  }else if(m_func_normal_exec_idx.count(fid)){
+    out.push_back( m_all_normal_execs[m_func_normal_exec_idx.find(fid)->second] );
+  }
+  return out;
 }

test/test_stat_sender.cpp

@@ -110,9 +110,11 @@ TEST(PSstatSenderTest, StatSenderGlobalAnomalyStatsBounce)
   CallList_t call_list;
   auto it1 = call_list.insert(call_list.end(), createFuncExecData_t(pid,rid,tid,func_id,func_name, 100, 200) );
   auto it2 = call_list.insert(call_list.end(), createFuncExecData_t(pid,rid,tid,func_id,func_name, 300, 400) );
+  it1->set_label(-1);
+  it2->set_label(-1);
   Anomalies anom;
-  anom.insert(it1, Anomalies::EventType::Outlier);
-  anom.insert(it2, Anomalies::EventType::Outlier);
+  anom.recordAnomaly(it1);
+  anom.recordAnomaly(it2);
 
   ExecDataMap_t dmap;
   dmap[func_id].push_back(it1);