The logic of extracting the provenance data for normal and anomalous events is now contained in a static member function of ADAnomalyProvenance rather than being part of the AD main code

Author: giltirn

include/chimbuko/ad/ADAnomalyProvenance.hpp

@@ -4,6 +4,9 @@
 #include <chimbuko/param.hpp>
 #include "chimbuko/ad/ADCounter.hpp"
 #include "chimbuko/ad/ADMetadataParser.hpp"
+#include "chimbuko/ad/ADNormalEventProvenance.hpp"
+#include "chimbuko/util/Anomalies.hpp"
+#include "chimbuko/util/PerfStats.hpp"
 
 namespace chimbuko{
 
@@ -38,6 +41,71 @@ namespace chimbuko{
      */
     nlohmann::json get_json() const;
 
+    /**
+     * @brief Extract the json provDB entries for the anomalies and normal events from an Anomalies collection
+     *
+     * @param anom_event_entries The provDB entries for anomalous events
+     * @param normal_event_entries The provDB entries for normal events
+     * @param normal_event_manager An instance of ADNormalEventProvenance that maintains normal events between calls
+     * @param perf Performance timing
+     * @param anomalies The Anomalies object containing anomalies and select normal events for this io step
+     * @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
+     * @param anom_win_size The window size of events to capture around each anomaly
+     * @param algo_params The outlier algorithm parameters
+     * @param event_man The event manager object
+     * @param counters The counter manager object
+     * @param metadata The metadata manager object
+     */
+    static void getProvenanceEntries(std::vector<nlohmann::json> &anom_event_entries,
+				     std::vector<nlohmann::json> &normal_event_entries,
+				     ADNormalEventProvenance &normal_event_manager,
+				     PerfStats &perf,
+				     const Anomalies &anomalies,
+				     const int step,
+				     const unsigned long first_event_ts,
+				     const unsigned long last_event_ts,
+				     const unsigned int anom_win_size,
+				     const ParamInterface &algo_params,
+				     const ADEvent &event_man,
+				     const ADCounter &counters, 
+				     const ADMetadataParser &metadata);
+
+    /**
+     * @brief Extract the json provDB entries for the anomalies and normal events from an Anomalies collection
+     *
+     * @param anom_event_entries The provDB entries for anomalous events
+     * @param normal_event_entries The provDB entries for normal events
+     * @param normal_event_manager An instance of ADNormalEventProvenance that maintains normal events between calls
+     * @param anomalies The Anomalies object containing anomalies and select normal events for this io step
+     * @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
+     * @param anom_win_size The window size of events to capture around each anomaly
+     * @param algo_params The outlier algorithm parameters
+     * @param event_man The event manager object
+     * @param counters The counter manager object
+     * @param metadata The metadata manager object
+     */
+    static void getProvenanceEntries(std::vector<nlohmann::json> &anom_event_entries,
+				     std::vector<nlohmann::json> &normal_event_entries,
+				     ADNormalEventProvenance &normal_event_manager,
+				     const Anomalies &anomalies,
+				     const int step,
+				     const unsigned long first_event_ts,
+				     const unsigned long last_event_ts,
+				     const unsigned int anom_win_size,
+				     const ParamInterface &algo_params,
+				     const ADEvent &event_man,
+				     const ADCounter &counters, 
+				     const ADMetadataParser &metadata){
+      PerfStats perf;
+      getProvenanceEntries(anom_event_entries, normal_event_entries,
+			   normal_event_manager, perf, anomalies, step, first_event_ts, last_event_ts,
+			   anom_win_size, algo_params, event_man, counters, metadata);
+    }		     
+
   private:
     /**
      * @brief Get the call stack

src/ad/ADAnomalyProvenance.cpp

@@ -168,3 +168,68 @@ nlohmann::json ADAnomalyProvenance::get_json() const{
 
   };
 }
+
+
+
+
+
+void ADAnomalyProvenance::getProvenanceEntries(std::vector<nlohmann::json> &anom_event_entries,
+					       std::vector<nlohmann::json> &normal_event_entries,
+					       ADNormalEventProvenance &normal_event_manager,
+					       PerfStats &perf,
+					       const Anomalies &anomalies,
+					       const int step,
+					       const unsigned long first_event_ts,
+					       const unsigned long last_event_ts,
+					       const unsigned int anom_win_size,
+					       const ParamInterface &algo_params,
+					       const ADEvent &event_man,
+					       const ADCounter &counters, 
+					       const ADMetadataParser &metadata){
+
+  constexpr bool do_delete = true;
+  constexpr bool add_outstanding = true;
+
+  PerfTimer timer,timer2;
+
+  //Put new normal event provenance into m_normalevent_prov
+  timer.start();
+  for(auto norm_it : anomalies.allEvents(Anomalies::EventType::Normal)){
+    timer2.start();
+    ADAnomalyProvenance extract_prov(*norm_it, event_man, algo_params, 
+				     counters, metadata, anom_win_size,
+				     step, first_event_ts, last_event_ts);
+    normal_event_manager.addNormalEvent(norm_it->get_pid(), norm_it->get_rid(), norm_it->get_tid(), norm_it->get_fid(), extract_prov.get_json());
+    perf.add("ad_extract_send_prov_normalevent_update_per_event_ms", timer2.elapsed_ms());
+  }
+  perf.add("ad_extract_send_prov_normalevent_update_total_ms", timer.elapsed_ms());
+
+  //Get any outstanding normal events from previous timesteps that we couldn't previously provide
+  timer.start();
+  
+  normal_event_entries = normal_event_manager.getOutstandingRequests(do_delete); //allow deletion of internal copy of events that are returned
+
+  perf.add("ad_extract_send_prov_normalevent_get_outstanding_ms", timer.elapsed_ms());
+
+  //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;
+  for(auto anom_it : anomalies.allEvents(Anomalies::EventType::Outlier)){
+    timer2.start();
+    ADAnomalyProvenance extract_prov(*anom_it, event_man, algo_params,
+				     counters, metadata, anom_win_size,
+				     step, first_event_ts, last_event_ts);
+    anom_event_entries[i++] = extract_prov.get_json();
+    perf.add("ad_extract_send_prov_anom_data_generation_per_anom_ms", timer2.elapsed_ms());
+      
+    //Get the associated normal event
+    //if normal event not available put into the list of outstanding requests
+    //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 = normal_event_manager.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));
+    perf.add("ad_extract_send_prov_normalevent_gather_per_anom_ms", timer2.elapsed_ms());
+  }
+
+}

src/chimbuko.cpp

@@ -377,56 +377,22 @@ void Chimbuko::extractAndSendProvenance(const Anomalies &anomalies,
 					const int step,
 					const unsigned long first_event_ts,
 					const unsigned long last_event_ts) const{
-  constexpr bool do_delete = true;
-  constexpr bool add_outstanding = true;
-
   //Gather provenance data on anomalies and send to provenance database
   if(m_params.prov_outputpath.length() > 0
 #ifdef ENABLE_PROVDB
      || m_provdb_client->isConnected()
 #endif
      ){
-    PerfTimer timer,timer2;
-
-    //Put new normal event provenance into m_normalevent_prov
-    timer.start();
-    for(auto norm_it : anomalies.allEvents(Anomalies::EventType::Normal)){
-      timer2.start();
-      ADAnomalyProvenance extract_prov(*norm_it, *m_event, *m_outlier->get_global_parameters(), 
-				       *m_counter, *m_metadata_parser, m_params.anom_win_size,
-				       step, first_event_ts, last_event_ts);
-      m_normalevent_prov->addNormalEvent(norm_it->get_pid(), norm_it->get_rid(), norm_it->get_tid(), norm_it->get_fid(), extract_prov.get_json());
-      m_perf.add("ad_extract_send_prov_normalevent_update_per_event_ms", timer2.elapsed_ms());
-    }
-    m_perf.add("ad_extract_send_prov_normalevent_update_total_ms", timer.elapsed_ms());
-
-    //Get any outstanding normal events from previous timesteps that we couldn't previously provide
-    timer.start();
-    std::vector<nlohmann::json> normalevent_prov = m_normalevent_prov->getOutstandingRequests(do_delete); //allow deletion of internal copy of events that are returned
-    m_perf.add("ad_extract_send_prov_normalevent_get_outstanding_ms", timer.elapsed_ms());
-
-    //Gather provenance of anomalies and for each one try to obtain a normal execution
-    timer.start();
-    std::vector<nlohmann::json> anomaly_prov(anomalies.nEvents(Anomalies::EventType::Outlier));
-    size_t i=0;
-    for(auto anom_it : anomalies.allEvents(Anomalies::EventType::Outlier)){
-      timer2.start();
-      ADAnomalyProvenance extract_prov(*anom_it, *m_event, *m_outlier->get_global_parameters(), 
-				       *m_counter, *m_metadata_parser, m_params.anom_win_size,
-				       step, first_event_ts, last_event_ts);
-      anomaly_prov[i++] = extract_prov.get_json();
-      m_perf.add("ad_extract_send_prov_anom_data_generation_per_anom_ms", timer2.elapsed_ms());
-      
-      //Get the associated normal event
-      //if normal event not available put into the list of outstanding requests
-      //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_normalevent_prov->getNormalEvent(anom_it->get_pid(), anom_it->get_rid(), anom_it->get_tid(), anom_it->get_fid(), add_outstanding, do_delete); 
-      if(nev.second) normalevent_prov.push_back(std::move(nev.first));
-      m_perf.add("ad_extract_send_prov_normalevent_gather_per_anom_ms", timer2.elapsed_ms());
-    }
+    //Get the provenance data
+    PerfTimer timer;
+    std::vector<nlohmann::json> anomaly_prov, normalevent_prov;
+    ADAnomalyProvenance::getProvenanceEntries(anomaly_prov, normalevent_prov, *m_normalevent_prov, m_perf,
+					      anomalies, step, first_event_ts, last_event_ts, m_params.anom_win_size,
+					      *m_outlier->get_global_parameters(), *m_event, *m_counter, *m_metadata_parser);
     m_perf.add("ad_extract_send_prov_provenance_data_generation_total_ms", timer.elapsed_ms());
 
+
+    //Write and send provenance dataX
     if(anomaly_prov.size() > 0){
       timer.start();
       m_io->writeJSON(anomaly_prov, step, "anomalies");