Merge pull request #18 from CODARcode/sm_release

HBOS functionality

Author: giltirn

app/driver.cpp

@@ -1,3 +1,4 @@
+// #include "chimbuko/AD.hpp"
 #include "chimbuko/chimbuko.hpp"
 #include "chimbuko/verbose.hpp"
 #include "chimbuko/util/string.hpp"
@@ -59,12 +60,15 @@ struct setLoggingHeadRankArg: public optionalCommandLineArgBase<ChimbukoParams>{
   }
 };
 
-  
+
 
 optionalArgsParser & getOptionalArgsParser(){
   static bool initialized = false;
   static optionalArgsParser p;
   if(!initialized){
+    addOptionalCommandLineArg(p, ad_algorithm, "Set an AD algorithm to use: hbos or sstd.");
+    addOptionalCommandLineArg(p, hbos_threshold, "Set Threshold for HBOS anomaly detection filter.");
+    addOptionalCommandLineArg(p, hbos_use_global_threshold, "Set true to use a global threshold in HBOS algorithm. Dafault is true.");
     addOptionalCommandLineArg(p, program_idx, "Set the index associated with the instrumented program. Use to label components of a workflow. (default 0)");
     addOptionalCommandLineArg(p, outlier_sigma, "Set the number of standard deviations that defines an anomalous event (default 6)");
     addOptionalCommandLineArg(p, pserver_addr, "Set the address of the parameter server. If empty (default) the pserver will not be used.");
@@ -73,9 +77,9 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, anom_win_size, "When anomaly data are recorded a window of this size (in units of function execution events) around the anomalous event are also recorded (default 10)");
     addOptionalCommandLineArg(p, prov_outputpath, "Output provenance data to this directory. Can be used in place of or in conjunction with the provenance database. An empty string \"\" (default) disables this output");
 #ifdef ENABLE_PROVDB
-    addOptionalCommandLineArg(p, provdb_addr, "Address of the provenance database. If empty (default) the provenance DB will not be used.\nHas format \"ofi+tcp;ofi_rxm://${IP_ADDR}:${PORT}\". Should also accept \"tcp://${IP_ADDR}:${PORT}\"");    
-    addOptionalCommandLineArg(p, nprovdb_shards, "Number of provenance database shards. Clients connect to shards round-robin by rank (default 1)");    
-#endif    
+    addOptionalCommandLineArg(p, provdb_addr, "Address of the provenance database. If empty (default) the provenance DB will not be used.\nHas format \"ofi+tcp;ofi_rxm://${IP_ADDR}:${PORT}\". Should also accept \"tcp://${IP_ADDR}:${PORT}\"");
+    addOptionalCommandLineArg(p, nprovdb_shards, "Number of provenance database shards. Clients connect to shards round-robin by rank (default 1)");
+#endif
 #ifdef _PERF_METRIC
     addOptionalCommandLineArg(p, perf_outputpath, "Output path for AD performance monitoring data. If an empty string (default) no output is written.");
     addOptionalCommandLineArg(p, perf_step, "How frequently (in IO steps) the performance data is dumped (default 10)");
@@ -97,7 +101,7 @@ optionalArgsParser & getOptionalArgsParser(){
 };
 
 void printHelp(){
-  std::cout << "Usage: driver <Trace engine type> <Trace directory> <Trace file prefix> <Options>\n" 
+  std::cout << "Usage: driver <Trace engine type> <Trace directory> <Trace file prefix> <Options>\n"
 	    << "Where <Trace engine type> : BPFile or SST\n"
 	    << "      <Trace directory>   : The directory in which the BPFile or SST file is located\n"
 	    << "      <Trace file prefix> : The prefix of the file (the trace file name without extension e.g. \"tau-metrics-mybinary\" for \"tau-metrics-mybinary.bp\")\n"
@@ -115,14 +119,16 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
   // Parse command line arguments (cf chimbuko.hpp for detailed description of parameters)
   // -----------------------------------------------------------------------
   ChimbukoParams params;
-      
+
   //Parameters for the connection to the instrumented binary trace output
   params.trace_engineType = argv[1]; // BPFile or SST
   params.trace_data_dir = argv[2]; // *.bp location
   std::string bp_prefix = argv[3]; // bp file prefix (e.g. tau-metrics-[nwchem])
 
   //The remainder are optional arguments. Enable using the appropriate command line switch
-
+  params.ad_algorithm = "hbos";
+  params.hbos_threshold = 0.99;
+  params.hbos_use_global_threshold = true;
   params.pserver_addr = "";  //don't use pserver by default
   params.hpserver_nthr = 1;
   params.outlier_sigma = 6.0;     // anomaly detection algorithm parameter
@@ -140,7 +146,7 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
   params.rank = -1234; //assign an invalid value as default for use below
   params.outlier_statistic = "exclusive_runtime";
   params.step_report_freq = 1;
-  
+
   getOptionalArgsParser().parse(params, argc-4, (const char**)(argv+4));
 
   //By default assign the rank index of the trace data as the MPI rank of the AD process
@@ -157,7 +163,7 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
   //Thus we need to obtain the input data rank also from the command line and modify the filename accordingly
   if(params.override_rank)
     params.trace_inputFile = bp_prefix + "-" + std::to_string(overrideRankArg::input_data_rank()) + ".bp";
-  
+
   //If neither the provenance database or the provenance output path are set, default to outputting to pwd
   if(params.prov_outputpath.size() == 0
 #ifdef ENABLE_PROVDB
@@ -169,21 +175,21 @@ ChimbukoParams getParamsFromCommandLine(int argc, char** argv, const int mpi_wor
 
   return params;
 }
-  
- 
+
+
 
 
 int main(int argc, char ** argv){
   if(argc == 1 || (argc == 2 && std::string(argv[1]) == "-help") ){
     printHelp();
     return 0;
   }
-      
+
   assert( MPI_Init(&argc, &argv) == MPI_SUCCESS );
 
   int mpi_world_rank, mpi_world_size;
   MPI_Comm_rank(MPI_COMM_WORLD, &mpi_world_rank);
-  MPI_Comm_size(MPI_COMM_WORLD, &mpi_world_size);    
+  MPI_Comm_size(MPI_COMM_WORLD, &mpi_world_size);
 
   //Parse environment variables
   if(const char* env_p = std::getenv("CHIMBUKO_VERBOSE"))
@@ -202,7 +208,7 @@ int main(int argc, char ** argv){
 
   bool error = false;
 
-  try 
+  try
     {
       //Instantiate Chimbuko
       Chimbuko driver(params);
@@ -214,24 +220,24 @@ int main(int argc, char ** argv){
       unsigned long total_n_outliers = 0, n_outliers = 0;
       unsigned long total_processing_time = 0, processing_time = 0;
       unsigned long long n_func_events = 0, n_comm_events = 0, n_counter_events = 0;
-      unsigned long long total_n_func_events = 0, total_n_comm_events = 0, total_n_counter_events = 0; 
+      unsigned long long total_n_func_events = 0, total_n_comm_events = 0, total_n_counter_events = 0;
       high_resolution_clock::time_point t1, t2;
 
       // -----------------------------------------------------------------------
       // Start analysis
       // -----------------------------------------------------------------------
-      headProgressStream(params.rank) << "Driver rank " << params.rank 
-				  << ": analysis start " << (driver.use_ps() ? "with": "without") 
+      headProgressStream(params.rank) << "Driver rank " << params.rank
+				  << ": analysis start " << (driver.use_ps() ? "with": "without")
 				  << " pserver" << std::endl;
 
       t1 = high_resolution_clock::now();
-      driver.run(n_func_events, 
+      driver.run(n_func_events,
 		 n_comm_events,
 		 n_counter_events,
 		 n_outliers,
 		 frames);
       t2 = high_resolution_clock::now();
-        
+
       if (params.rank == 0) {
 	headProgressStream(params.rank) << "Driver rank " << params.rank << ": analysis done!\n";
 	driver.show_status(true);
@@ -266,7 +272,7 @@ int main(int argc, char ** argv){
 	total_n_comm_events = global_measures[1];
 	total_n_counter_events = global_measures[2];
       }
-        
+
       headProgressStream(params.rank) << "Driver rank " << params.rank << ": Final report\n"
 				  << "Avg. num. frames over MPI ranks : " << (double)total_frames/(double)mpi_world_size << "\n"
 				  << "Avg. processing time over MPI ranks : " << (double)total_processing_time/(double)mpi_world_size << " msec\n"

app/pserver.cpp

@@ -16,10 +16,12 @@
 
 #include <chimbuko/param/sstd_param.hpp>
 #include <chimbuko/util/commandLineParser.hpp>
+#include <chimbuko/util/error.hpp>
 #include <fstream>
 #include "chimbuko/verbose.hpp"
 
 
+
 using namespace chimbuko;
 
 struct pserverArgs{
@@ -40,6 +42,7 @@ struct pserverArgs{
   int stat_send_freq;
 
   std::string stat_outputdir;
+  std::string ad;
 
 #ifdef _USE_ZMQNET
   int max_pollcyc_msg;
@@ -50,8 +53,8 @@ struct pserverArgs{
 #ifdef ENABLE_PROVDB
   std::string provdb_addr;
 #endif
- 
-  pserverArgs(): nt(-1), logdir("."), ws_addr(""), load_params_set(false), save_params_set(false), freeze_params(false), stat_send_freq(1000), stat_outputdir(""), port(5559)
+
+  pserverArgs(): ad("hbos"), nt(-1), logdir("."), ws_addr(""), load_params_set(false), save_params_set(false), freeze_params(false), stat_send_freq(1000), stat_outputdir(""), port(5559)
 #ifdef _USE_ZMQNET
 	       , max_pollcyc_msg(10), zmq_io_thr(1), autoshutdown(true)
 #endif
@@ -64,6 +67,7 @@ struct pserverArgs{
     static bool init = false;
     static commandLineParser<pserverArgs> p;
     if(!init){
+      addOptionalCommandLineArg(p, ad, "Set AD algorithm to use.");
       addOptionalCommandLineArg(p, nt, "Set the number of RPC handler threads (max-2 by default)");
       addOptionalCommandLineArg(p, logdir, "Set the output log directory (default: job directory)");
       addOptionalCommandLineArg(p, port, "Set the pserver port (default: 5559)");
@@ -79,7 +83,7 @@ struct pserverArgs{
       addOptionalCommandLineArg(p, autoshutdown, "If enabled the pserver will automatically shutdown when all clients have disconnected (default: true)");
 #endif
 #ifdef ENABLE_PROVDB
-      addOptionalCommandLineArg(p, provdb_addr, "Address of the provenance database. If empty (default) the global function and counter statistics will not be send to the provenance DB.\nHas format \"ofi+tcp;ofi_rxm://${IP_ADDR}:${PORT}\". Should also accept \"tcp://${IP_ADDR}:${PORT}\"");    
+      addOptionalCommandLineArg(p, provdb_addr, "Address of the provenance database. If empty (default) the global function and counter statistics will not be send to the provenance DB.\nHas format \"ofi+tcp;ofi_rxm://${IP_ADDR}:${PORT}\". Should also accept \"tcp://${IP_ADDR}:${PORT}\"");
 #endif
 
       init = true;
@@ -107,13 +111,18 @@ int main (int argc, char ** argv){
   if(const char* env_p = std::getenv("CHIMBUKO_VERBOSE")){
     progressStream << "Pserver: Enabling verbose debug output" << std::endl;
     enableVerboseLogging() = true;
-  }       
+  }
 
-  SstdParam param; //global collection of parameters used to identify anomalies
+  ParamInterface * param = ParamInterface::set_AdParam(args.ad); //"hbos"); //sstd"); //HbosParam param; //global collection of parameters used to identify anomalies
+  if (param == nullptr) {
+    fatal_error("INCORRECT algorithm for AdParam: Not Found. Choose sstd or hbos.");
+    // verboseStream << "INCORRECT algorithm for AdParam: Not Found. Choose sstd or hbos." << std::endl;
+    // exit(EXIT_FAILURE);
+  }
   GlobalAnomalyStats global_func_stats; //global anomaly statistics
   GlobalCounterStats global_counter_stats; //global counter statistics
   PSglobalFunctionIndexMap global_func_index_map; //mapping of function name to global index
-  
+
   //Optionally load previously-computed AD algorithm statistics
   if(args.load_params_set){
     progressStream << "Pserver: Loading parameters from input file " << args.load_params << std::endl;
@@ -122,7 +131,7 @@ int main (int argc, char ** argv){
     nlohmann::json in_p;
     in >> in_p;
     global_func_index_map.deserialize(in_p["func_index_map"]);
-    param.assign(in_p["alg_params"].dump());
+    param->assign(in_p["alg_params"].dump()); //param.assign(in_p["alg_params"].dump());
   }
 
 #ifdef _USE_MPINET
@@ -172,8 +181,8 @@ int main (int argc, char ** argv){
       if(args.stat_outputdir.size()) std::cout << "(dir @ " << args.stat_outputdir << ")";
     }
 
-    net.add_payload(new NetPayloadUpdateParams(&param, args.freeze_params));
-    net.add_payload(new NetPayloadGetParams(&param));
+    net.add_payload(new NetPayloadUpdateParams(param, args.freeze_params)); //new NetPayloadUpdateParams(&param, args.freeze_params));
+    net.add_payload(new NetPayloadGetParams(param)); //new NetPayloadGetParams(&param));
     net.add_payload(new NetPayloadUpdateAnomalyStats(&global_func_stats));
     net.add_payload(new NetPayloadUpdateCounterStats(&global_counter_stats));
     net.add_payload(new NetPayloadGlobalFunctionIndexMapBatched(&global_func_index_map));
@@ -206,7 +215,7 @@ int main (int argc, char ** argv){
       progressStream << "Pserver: sending final statistics to provDB" << std::endl;
       provdb_client.sendMultipleData(global_func_stats.collect_func_data(), GlobalProvenanceDataType::FunctionStats);
       provdb_client.sendMultipleData(global_counter_stats.get_json_state(), GlobalProvenanceDataType::CounterStats);
-      progressStream << "Pserver: disconnecting from provDB" << std::endl;      
+      progressStream << "Pserver: disconnecting from provDB" << std::endl;
       provdb_client.disconnect();
     }
 #endif
@@ -218,7 +227,7 @@ int main (int argc, char ** argv){
     o.open(args.logdir + "/parameters.txt");
     if (o.is_open())
       {
-	param.show(o);
+	param->show(o); //param.show(o);
 	o.close();
       }
   }
@@ -242,12 +251,12 @@ int main (int argc, char ** argv){
 
   //Optionally save the final AD algorithm parameters
   if(args.save_params_set){
-    progressStream << "PServer: Saving parameters to output file " << args.save_params << std::endl;   
+    progressStream << "PServer: Saving parameters to output file " << args.save_params << std::endl;
     std::ofstream out(args.save_params);
     if(!out.good()) throw std::runtime_error("Could not write anomaly algorithm parameters to the file provided");
     nlohmann::json out_p;
     out_p["func_index_map"] = global_func_index_map.serialize();
-    out_p["alg_params"] = nlohmann::json::parse(param.serialize());
+    out_p["alg_params"] = nlohmann::json::parse(param->serialize()); //param.serialize());
     out << out_p;
   }
 

configure.ac

@@ -94,7 +94,7 @@ if test "x$with_adios2" != "x"; then
 	   AC_MSG_NOTICE([Detected ADIOS 2.6+])
    fi
 else
-	AC_MSG_FAILURE(["ADIOS2 install path must be specified with --with-adios2"])	
+	AC_MSG_FAILURE(["ADIOS2 install path must be specified with --with-adios2"])
 fi
 
 #Choose network

include/Makefile.am

@@ -1 +1 @@
-nobase_include_HEADERS = chimbuko/chimbuko.hpp chimbuko/ad/ADOutlier.hpp chimbuko/ad/ADNetClient.hpp chimbuko/ad/ADAnomalyProvenance.hpp chimbuko/ad/ADParser.hpp chimbuko/ad/ADProvenanceDBengine.hpp chimbuko/ad/ADLocalFuncStatistics.hpp chimbuko/ad/ADMetadataParser.hpp chimbuko/ad/ADDefine.hpp chimbuko/ad/ADglobalFunctionIndexMap.hpp chimbuko/ad/ADProvenanceDBclient.hpp chimbuko/ad/ADEvent.hpp chimbuko/ad/ADCounter.hpp chimbuko/ad/ADLocalCounterStatistics.hpp chimbuko/ad/ExecData.hpp chimbuko/ad/AnomalyData.hpp chimbuko/ad/utils.hpp chimbuko/ad/ADio.hpp chimbuko/verbose.hpp chimbuko/pserver/global_anomaly_stats.hpp chimbuko/pserver/PSglobalFunctionIndexMap.hpp chimbuko/pserver/PSstatSender.hpp chimbuko/pserver/global_counter_stats.hpp chimbuko/param/sstd_param.hpp chimbuko/AD.hpp chimbuko/message.hpp chimbuko/net.hpp chimbuko/pserver.hpp chimbuko/param.hpp chimbuko/net/mpi_net.hpp chimbuko/net/zmq_net.hpp chimbuko/net/zmqme_net.hpp chimbuko/util/RunStats.hpp chimbuko/util/ADIOS2parseUtils.hpp chimbuko/util/map.hpp chimbuko/util/string.hpp chimbuko/util/DispatchQueue.hpp chimbuko/util/PerfStats.hpp chimbuko/util/RunMetric.hpp chimbuko/util/Anomalies.hpp chimbuko/util/barrier.hpp chimbuko/util/threadPool.hpp chimbuko/util/hash.hpp chimbuko/util/mtQueue.hpp chimbuko/util/commandLineParser.hpp 
+nobase_include_HEADERS = chimbuko/chimbuko.hpp chimbuko/ad/ADOutlier.hpp chimbuko/ad/ADNetClient.hpp chimbuko/ad/ADAnomalyProvenance.hpp chimbuko/ad/ADParser.hpp chimbuko/ad/ADProvenanceDBengine.hpp chimbuko/ad/ADLocalFuncStatistics.hpp chimbuko/ad/ADMetadataParser.hpp chimbuko/ad/ADDefine.hpp chimbuko/ad/ADglobalFunctionIndexMap.hpp chimbuko/ad/ADProvenanceDBclient.hpp chimbuko/ad/ADEvent.hpp chimbuko/ad/ADCounter.hpp chimbuko/ad/ADLocalCounterStatistics.hpp chimbuko/ad/ExecData.hpp chimbuko/ad/AnomalyData.hpp chimbuko/ad/utils.hpp chimbuko/ad/ADio.hpp chimbuko/verbose.hpp chimbuko/pserver/global_anomaly_stats.hpp chimbuko/pserver/PSglobalFunctionIndexMap.hpp chimbuko/pserver/PSstatSender.hpp chimbuko/pserver/global_counter_stats.hpp chimbuko/param/sstd_param.hpp chimbuko/param/hbos_param.hpp chimbuko/AD.hpp chimbuko/message.hpp chimbuko/net.hpp chimbuko/pserver.hpp chimbuko/param.hpp chimbuko/net/mpi_net.hpp chimbuko/net/zmq_net.hpp chimbuko/net/zmqme_net.hpp chimbuko/util/RunStats.hpp chimbuko/util/ADIOS2parseUtils.hpp chimbuko/util/map.hpp chimbuko/util/string.hpp chimbuko/util/DispatchQueue.hpp chimbuko/util/PerfStats.hpp chimbuko/util/RunMetric.hpp chimbuko/util/Anomalies.hpp chimbuko/util/barrier.hpp chimbuko/util/threadPool.hpp chimbuko/util/hash.hpp chimbuko/util/mtQueue.hpp chimbuko/util/commandLineParser.hpp

include/chimbuko/ad/ADAnomalyProvenance.hpp

@@ -26,21 +26,21 @@ namespace chimbuko{
      * @param io_step Index of io step
      * @param io_step_tstart Timestamp of beginning of io frame
      * @param io_step_tend Timestamp of end of io frame
-     */     
+     */
     ADAnomalyProvenance(const ExecData_t &call,
 			const ADEvent &event_man, //for stack trace
 			const ParamInterface &algo_params, //algorithm parameters
 			const ADCounter &counters, //for counters
 			const ADMetadataParser &metadata, //for env information including GPU context/device/stream
 			const int window_size,
-			const int io_step, 
+			const int io_step,
 			const unsigned long io_step_tstart, const unsigned long io_step_tend);
 
     /**
      * @brief Serialize anomaly data into JSON construct
      */
     nlohmann::json get_json() const;
-    
+
     /**
      * @brief Extract the json provDB entries for the anomalies and normal events from an Anomalies collection
      *
@@ -69,7 +69,7 @@ namespace chimbuko{
 				     const unsigned int anom_win_size,
 				     const ParamInterface &algo_params,
 				     const ADEvent &event_man,
-				     const ADCounter &counters, 
+				     const ADCounter &counters,
 				     const ADMetadataParser &metadata);
 
     /**
@@ -98,20 +98,20 @@ namespace chimbuko{
 				     const unsigned int anom_win_size,
 				     const ParamInterface &algo_params,
 				     const ADEvent &event_man,
-				     const ADCounter &counters, 
+				     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
      */
     void getStackInformation(const ExecData_t &call, const ADEvent &event_man);
-    
+
     /**
      * @brief Get counters in execution window
      */