AD and pserver now collect / aggregate anomaly metrics, pserver sends to viz

giltirn · giltirn · commit c9946b3db710 · 2021-09-22T17:17:42.000-04:00
Simulator now simulates collection / aggregation of anomaly metrics
Added GlobalAnomalyMetrics to ADcombinedPSdata + unit test
In benchmark_suite/func_multimodal, the anomalous function now executes a varying number of times a function that performa sleep for a fixed time such that it is the parent that appears anomalous, not the sleep call
Fixed GlobalAnomalyMetrics unit test failing due to addition of _id field causing json outputs to differ
diff --git a/app/pserver.cpp b/app/pserver.cpp
@@ -125,6 +125,7 @@ int main (int argc, char ** argv){
   }
   GlobalAnomalyStats global_func_stats; //global anomaly statistics
   GlobalCounterStats global_counter_stats; //global counter statistics
+  GlobalAnomalyMetrics global_anom_metrics; //global anomaly metrics
   PSglobalFunctionIndexMap global_func_index_map; //mapping of function name to global index
 
   //Optionally load previously-computed AD algorithm statistics
@@ -187,13 +188,14 @@ int main (int argc, char ** argv){
 
     net.add_payload(new NetPayloadUpdateParams(param, args.freeze_params));
     net.add_payload(new NetPayloadGetParams(param));
-    net.add_payload(new NetPayloadRecvCombinedADdata(&global_func_stats, &global_counter_stats));
+    net.add_payload(new NetPayloadRecvCombinedADdata(&global_func_stats, &global_counter_stats, &global_anom_metrics));
     net.add_payload(new NetPayloadGlobalFunctionIndexMapBatched(&global_func_index_map));
     net.init(nullptr, nullptr, args.nt);
 
     //Start sending anomaly statistics to viz
     stat_sender.add_payload(new PSstatSenderGlobalAnomalyStatsPayload(&global_func_stats));
     stat_sender.add_payload(new PSstatSenderGlobalCounterStatsPayload(&global_counter_stats));
+    stat_sender.add_payload(new PSstatSenderGlobalAnomalyMetricsPayload(&global_anom_metrics));
     stat_sender.run_stat_sender(args.ws_addr, args.stat_outputdir);
 
     //Register a signal handler that prevents the application from exiting on SIGTERM; instead this signal will be handled by ZeroMQ and will cause the pserver to shutdown gracefully
diff --git a/benchmark_suite/func_multimodal/main.C b/benchmark_suite/func_multimodal/main.C
@@ -14,9 +14,19 @@ T strToAny(const std::string &s){
   return out;
 }
 
+int round_to_nearest_100(int val){
+  int nhundred = ( val + 50 ) / 100;
+  return nhundred * 100;
+}
+void sleep_100ms(){
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+}
 
+//In order to make this function exhibit the anomaly and not just sleep_for, call sleep in units of 100ms
 void the_function(const int sleep_ms){
-  std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));
+  int nhundred = sleep_ms / 100;
+  for(int i=0;i<nhundred;i++)
+    sleep_100ms();
 }
 
 int main(int argc, char **argv){
@@ -29,19 +39,22 @@ int main(int argc, char **argv){
   std::stringstream ss;
   ss << argv[2];
 
-  std::cout << "Mode times: ";
+  std::cout << "Mode times (rounded to nearest 100ms): ";
   std::vector<int> mode_times;
   while(ss.good()) {
     std::string substr;
     getline(ss, substr, ','); //get first string delimited by comma
-    mode_times.push_back(strToAny<int>(substr));
+    int val = strToAny<int>(substr);
+    mode_times.push_back(round_to_nearest_100(val));
     std::cout << mode_times.back() << " ";
   }
   std::cout << std::endl;
 
-  int anom_time = strToAny<int>(argv[3]);
+  int anom_time = round_to_nearest_100(strToAny<int>(argv[3]));
   int anom_freq = strToAny<int>(argv[4]);
 
+  std::cout << "Anomaly time (rounded to nearest 100ms): " << anom_time << std::endl;
+
   std::mt19937 gen(1234);
   std::uniform_int_distribution<> dist(0, mode_times.size()-1);
 
diff --git a/include/chimbuko/AD.hpp b/include/chimbuko/AD.hpp
@@ -10,6 +10,7 @@
 #include "chimbuko/ad/ADNetClient.hpp"
 #include "chimbuko/ad/ADLocalFuncStatistics.hpp"
 #include "chimbuko/ad/ADLocalCounterStatistics.hpp"
+#include "chimbuko/ad/ADLocalAnomalyMetrics.hpp"
 #include "chimbuko/ad/ADcombinedPSdata.hpp"
 #include "chimbuko/ad/ADProvenanceDBclient.hpp"
 #include "chimbuko/ad/ADMetadataParser.hpp"
diff --git a/include/chimbuko/ad/ADcombinedPSdata.hpp b/include/chimbuko/ad/ADcombinedPSdata.hpp
@@ -2,35 +2,44 @@
 #include <chimbuko_config.h>
 #include <chimbuko/ad/ADLocalFuncStatistics.hpp>
 #include <chimbuko/ad/ADLocalCounterStatistics.hpp>
+#include <chimbuko/ad/ADLocalAnomalyMetrics.hpp>
 
 namespace chimbuko{
   /** 
    * @brief A class that combines all data that is sent to the PS into a single send transmission
    */
   class ADcombinedPSdata{
   public:
-    ADcombinedPSdata(ADLocalFuncStatistics &func_stats, ADLocalCounterStatistics &counter_stats, PerfStats* perf=nullptr): 
-      m_func_stats(func_stats), m_counter_stats(counter_stats), m_perf(perf){}
+    ADcombinedPSdata(ADLocalFuncStatistics &func_stats, 
+		     ADLocalCounterStatistics &counter_stats, 
+		     ADLocalAnomalyMetrics &anom_metrics,
+		     PerfStats* perf=nullptr): 
+      m_func_stats(func_stats), m_counter_stats(counter_stats), m_anom_metrics(anom_metrics), m_perf(perf){}
     
     ADLocalCounterStatistics &get_counter_stats(){ return m_counter_stats; }
     const ADLocalCounterStatistics &get_counter_stats() const{ return m_counter_stats; }
 
     ADLocalFuncStatistics &get_func_stats(){ return m_func_stats; }
     const ADLocalFuncStatistics &get_func_stats() const{ return m_func_stats; }
 
+    ADLocalAnomalyMetrics &get_anom_metrics(){ return m_anom_metrics; }
+    const ADLocalAnomalyMetrics &get_anom_metrics() const{ return m_anom_metrics; }
+
+
     /**
      * @brief The State object is what is serialized for the communication
      */
     struct State{
       ADLocalFuncStatistics::State func_stats_state;
       ADLocalCounterStatistics::State counter_stats_state;
+      ADLocalAnomalyMetrics::State anom_metrics_state;
 
       /**
        * @brief Serialize using cereal
        */
       template<class Archive>
       void serialize(Archive & archive){
-	archive(func_stats_state , counter_stats_state);
+	archive(func_stats_state , counter_stats_state, anom_metrics_state);
       }
 
       /**
@@ -80,6 +89,7 @@ namespace chimbuko{
   private:
     ADLocalFuncStatistics &m_func_stats;
     ADLocalCounterStatistics &m_counter_stats;
+    ADLocalAnomalyMetrics &m_anom_metrics;
     PerfStats* m_perf;
   };
 
diff --git a/include/chimbuko/chimbuko.hpp b/include/chimbuko/chimbuko.hpp
@@ -198,7 +198,9 @@ namespace chimbuko {
      * @brief Gather and send the required data to the pserver
      */
     void gatherAndSendPSdata(const Anomalies &anomalies,
-			     const int step) const;
+			     const int step,
+			     const unsigned long first_event_ts,
+			     const unsigned long last_event_ts) const;
 
     /**
      * @brief Send new metadata entries collected during current fram to provenance DB
diff --git a/include/chimbuko/pserver/NetPayloadRecvCombinedADdata.hpp b/include/chimbuko/pserver/NetPayloadRecvCombinedADdata.hpp
@@ -2,6 +2,7 @@
 #include <chimbuko_config.h>
 #include <chimbuko/pserver/GlobalAnomalyStats.hpp>
 #include <chimbuko/pserver/GlobalCounterStats.hpp>
+#include <chimbuko/pserver/GlobalAnomalyMetrics.hpp>
 
 namespace chimbuko{
   
@@ -11,9 +12,12 @@ namespace chimbuko{
   class NetPayloadRecvCombinedADdata: public NetPayloadBase{
     GlobalAnomalyStats * m_global_anom_stats;
     GlobalCounterStats * m_global_counter_stats;
+    GlobalAnomalyMetrics * m_global_anom_metrics;
   public:
-    NetPayloadRecvCombinedADdata(GlobalAnomalyStats * global_anom_stats, GlobalCounterStats * global_counter_stats): m_global_anom_stats(global_anom_stats),
-														     m_global_counter_stats(global_counter_stats){}
+    NetPayloadRecvCombinedADdata(GlobalAnomalyStats * global_anom_stats, GlobalCounterStats * global_counter_stats,
+				 GlobalAnomalyMetrics * global_anom_metrics): m_global_anom_stats(global_anom_stats),
+									      m_global_counter_stats(global_counter_stats),
+									      m_global_anom_metrics(global_anom_metrics){}
     MessageKind kind() const override{ return MessageKind::AD_PS_COMBINED_STATS; }
     MessageType type() const override{ return MessageType::REQ_ADD; }
     void action(Message &response, const Message &message) override;
diff --git a/sim/include/sim/pserver.hpp b/sim/include/sim/pserver.hpp
@@ -3,16 +3,21 @@
 #include<chimbuko/param.hpp>
 #include<chimbuko/pserver/GlobalAnomalyStats.hpp>
 #include<chimbuko/pserver/GlobalCounterStats.hpp>
+#include<chimbuko/pserver/GlobalAnomalyMetrics.hpp>
 
 namespace chimbuko_sim{
   using namespace chimbuko;
-
-  //An object that represents the pserver for the purposes of aggregating data over AD instances and writing streaming output to disk
+  
+  /**
+   * @brief An object that represents the pserver for the purposes of aggregating data over AD instances and writing streaming output to disk
+   */
   class pserverSim{
-    GlobalAnomalyStats global_func_stats; //global anomaly statistics
-    GlobalCounterStats global_counter_stats; //global counter statistics
+    GlobalAnomalyStats global_func_stats; /**< global anomaly statistics */
+    GlobalCounterStats global_counter_stats; /**< global counter statistics */
+    GlobalAnomalyMetrics global_anomaly_metrics; /**< global anomaly metrics */
     PSstatSenderGlobalAnomalyStatsPayload anomaly_stats_payload;
     PSstatSenderGlobalCounterStatsPayload counter_stats_payload;
+    PSstatSenderGlobalAnomalyMetricsPayload anomaly_metrics_payload;
     LocalNet m_net; /**< The net server */
     ParamInterface* m_ad_params; /**< If using an AD algorithm to analyze the trace, these are the global parameters*/
   public:
@@ -37,6 +42,13 @@ namespace chimbuko_sim{
     void addCounterData(const ADLocalCounterStatistics &loc){
       global_counter_stats.add_counter_data(loc);
     }
+
+    /**
+     * @brief Stand-in for the anomaly metrics comms from AD -> pserver
+     */
+    void addAnomalyMetrics(const ADLocalAnomalyMetrics &loc){
+      global_anomaly_metrics.add(loc);
+    }
     
     /**
      * @brief Mirror write functionality of PSstatSender
diff --git a/sim/src/ad.cpp b/sim/src/ad.cpp
@@ -292,6 +292,9 @@ void ADsim::step(const unsigned long step){
     prof_stats.gatherStatistics(&this_step_func_execs); //takes map of fid -> vector of ExecData iterators
     prof_stats.gatherAnomalies(anom);
     getPserver().addAnomalyData(prof_stats);
+
+    ADLocalAnomalyMetrics metrics(m_pid, m_rid, step, step_start_time, step_end_time, anom);
+    getPserver().addAnomalyMetrics(metrics);
   }
 }
 
diff --git a/sim/src/pserver.cpp b/sim/src/pserver.cpp
@@ -8,7 +8,8 @@ void pserverSim::writeStreamingOutput() const{
   nlohmann::json json_packet;
   anomaly_stats_payload.add_json(json_packet);
   counter_stats_payload.add_json(json_packet);
-    
+  anomaly_metrics_payload.add_json(json_packet);
+  
   static int iter = 0;
 
   std::ostringstream fname; fname << "pserver_output_stats_" << iter << ".json";
@@ -19,7 +20,8 @@ void pserverSim::writeStreamingOutput() const{
   }
 }
 
-pserverSim::pserverSim(): anomaly_stats_payload(&global_func_stats), counter_stats_payload(&global_counter_stats), m_ad_params(nullptr){
+pserverSim::pserverSim(): anomaly_stats_payload(&global_func_stats), counter_stats_payload(&global_counter_stats), 
+			  anomaly_metrics_payload(&global_anomaly_metrics), m_ad_params(nullptr){
   if(adAlgorithmParams().algorithm != "none"){
     m_ad_params = ParamInterface::set_AdParam(adAlgorithmParams().algorithm);
     m_net.add_payload(new NetPayloadUpdateParams(m_ad_params, false));
diff --git a/src/ad/ADcombinedPSdata.cpp b/src/ad/ADcombinedPSdata.cpp
@@ -16,12 +16,14 @@ ADcombinedPSdata::State ADcombinedPSdata::get_state() const{
   State state;
   state.func_stats_state = m_func_stats.get_state();
   state.counter_stats_state = m_counter_stats.get_state();
+  state.anom_metrics_state = m_anom_metrics.get_state();
   return state;
 }
 
 void ADcombinedPSdata::set_state(const ADcombinedPSdata::State &s){
   m_func_stats.set_state(s.func_stats_state);
   m_counter_stats.set_state(s.counter_stats_state);
+  m_anom_metrics.set_state(s.anom_metrics_state);
 }
 
 
@@ -43,7 +45,8 @@ std::pair<size_t, size_t> ADcombinedPSdata::send(ADNetClient &net_client) const{
     return std::make_pair(0, 0);
 
   int step = m_func_stats.getAnomalyData().get_step();
-  if(m_counter_stats.getIOstep() != step) fatal_error("Step value mismatch between members");
+  if(m_counter_stats.getIOstep() != step) fatal_error("Step value mismatch between counter stats and func anomaly stats");
+  if(m_anom_metrics.get_step() != step) fatal_error("Step value mismatch between anomaly metrics and func anomaly stats");
 
   Message msg;
   msg.set_info(net_client.get_client_rank(), net_client.get_server_rank(), MessageType::REQ_ADD, MessageKind::AD_PS_COMBINED_STATS, step);
diff --git a/src/chimbuko.cpp b/src/chimbuko.cpp
@@ -461,7 +461,9 @@ void Chimbuko::sendNewMetadataToProvDB(int step) const{
 }
 
 void Chimbuko::gatherAndSendPSdata(const Anomalies &anomalies,
-				   const int step) const{
+				   const int step,
+				   const unsigned long first_event_ts,
+				   const unsigned long last_event_ts) const{
   if(m_net_client && m_net_client->use_ps()){
     PerfTimer timer;
   
@@ -477,10 +479,15 @@ void Chimbuko::gatherAndSendPSdata(const Anomalies &anomalies,
     ADLocalCounterStatistics count_stats(m_params.program_idx, step, nullptr, &m_perf); //currently collect all counters
     count_stats.gatherStatistics(m_counter->getCountersByIndex());
     m_perf.add("ad_gather_send_ps_data_gather_counter_stats_time_ms", timer.elapsed_ms());
+    
+    //Gather anomaly metrics
+    timer.start();
+    ADLocalAnomalyMetrics metrics(m_params.program_idx, m_params.rank, step, first_event_ts, last_event_ts, anomalies);
+    m_perf.add("ad_gather_send_ps_data_gather_metrics_time_ms", timer.elapsed_ms());
 
     //Send the data in a single communication
     timer.start();
-    ADcombinedPSdata comb_stats(prof_stats, count_stats);
+    ADcombinedPSdata comb_stats(prof_stats, count_stats, metrics);
     comb_stats.send(*m_net_client);
     m_perf.add("ad_gather_send_ps_data_gather_send_all_stats_to_ps_time_ms", timer.elapsed_ms());
   }
@@ -569,7 +576,7 @@ void Chimbuko::run(unsigned long long& n_func_events,
 
     //Gather and send statistics and data to the pserver
     timer.start();
-    gatherAndSendPSdata(anomalies, step);
+    gatherAndSendPSdata(anomalies, step, first_event_ts, last_event_ts);
     m_perf.add("ad_run_gather_send_ps_data_time_ms", timer.elapsed_ms());
 
     //Trim the call list
diff --git a/src/pserver/GlobalAnomalyMetrics.cpp b/src/pserver/GlobalAnomalyMetrics.cpp
@@ -45,6 +45,8 @@ nlohmann::json GlobalAnomalyMetrics::get_json(){
   
   nlohmann::json out = nlohmann::json::array();
 
+  static size_t _id = 0;
+
   for(auto const &pelem : m_anomaly_metrics_recent){
     for(auto const &relem : pelem.second){
       for(auto const &felem : relem.second){
@@ -66,6 +68,7 @@ nlohmann::json GlobalAnomalyMetrics::get_json(){
 	elem["fname"] = felem.second.get_func();
 	elem["new_data"] = felem.second.get_json();
 	elem["all_data"] = full_fit->second.get_json();
+	elem["_id"] = _id++;
 
 	out.push_back(std::move(elem));
       }
diff --git a/src/pserver/NetPayloadRecvCombinedADdata.cpp b/src/pserver/NetPayloadRecvCombinedADdata.cpp
@@ -7,15 +7,18 @@ void NetPayloadRecvCombinedADdata::action(Message &response, const Message &mess
   check(message);
   if(m_global_anom_stats == nullptr) throw std::runtime_error("Cannot update global anomaly statistics as stats object has not been linked");
   if(m_global_counter_stats == nullptr) throw std::runtime_error("Cannot update global counter statistics as stats object has not been linked");
+  if(m_global_anom_metrics == nullptr) throw std::runtime_error("Cannot update global anomaly metrics as stats object has not been linked");
   
   ADLocalFuncStatistics locf;
   ADLocalCounterStatistics locc(0,0,nullptr);
+  ADLocalAnomalyMetrics locm;
 
-  ADcombinedPSdata comb(locf,locc);
+  ADcombinedPSdata comb(locf,locc,locm);
   comb.net_deserialize(message.buf());
 
   m_global_anom_stats->add_anomaly_data(locf);
   m_global_counter_stats->add_counter_data(locc);
+  m_global_anom_metrics->add(locm);
 
   response.set_msg("", false);
 }
diff --git a/src/pserver/PSstatSender.cpp b/src/pserver/PSstatSender.cpp
@@ -91,7 +91,7 @@ static void send_stat(std::string url, std::string stat_save_dir,
 
       //Send if object has content
       if(json_packet.size() != 0){
-	std::string packet = json_packet.dump();
+	std::string packet = json_packet.dump(4);
             
 	if(write_disk){ //Send data to disk
 	  std::stringstream fname; fname << stat_save_dir << "/pserver_output_stats_" << iter << ".json";
diff --git a/test/unit_tests/ad/ADcombinedPSdata.cpp b/test/unit_tests/ad/ADcombinedPSdata.cpp
diff --git a/test/unit_tests/pserver/GlobalAnomalyMetrics.cpp b/test/unit_tests/pserver/GlobalAnomalyMetrics.cpp

Original file line number	Diff line number	Diff line change
`@@ -292,6 +292,9 @@ void ADsim::step(const unsigned long step){`
`292`	`292`	`prof_stats.gatherStatistics(&this_step_func_execs); //takes map of fid -> vector of ExecData iterators`
`293`	`293`	`prof_stats.gatherAnomalies(anom);`
`294`	`294`	`getPserver().addAnomalyData(prof_stats);`
	`295`	`+`
	`296`	`+ ADLocalAnomalyMetrics metrics(m_pid, m_rid, step, step_start_time, step_end_time, anom);`
	`297`	`+ getPserver().addAnomalyMetrics(metrics);`
`295`	`298`	`}`
`296`	`299`	`}`
`297`	`300`