Merge branch 'develop' into ckelly_develop

Conflicts:
	scripts/launch/run_services.sh
	src/chimbuko.cpp

Author: giltirn

app/Makefile.am

@@ -32,17 +32,14 @@ bpfile_replay_LDADD = $(LDADD)
 
 
 if ENABLE_PROVDB
-bin_PROGRAMS += provdb_admin provdb_query provdb_shutdown
+bin_PROGRAMS += provdb_admin provdb_query
 
 provdb_admin_SOURCES = provdb_admin.cpp
 provdb_admin_LDADD = $(LDADD)
 
 provdb_query_SOURCES = provdb_query.cpp
 provdb_query_LDADD = $(LDADD)
 
-provdb_shutdown_SOURCES = provdb_shutdown.cpp
-provdb_shutdown_LDADD = $(LDADD)
-
 else
 	echo "Provenance DB not being built"
 endif

app/driver.cpp

@@ -1,8 +1,8 @@
+// #include "chimbuko/AD.hpp"
 #include "chimbuko/chimbuko.hpp"
 #include "chimbuko/verbose.hpp"
 #include "chimbuko/util/string.hpp"
 #include "chimbuko/util/commandLineParser.hpp"
-#include "chimbuko/util/error.hpp"
 #include <chrono>
 #include <cstdlib>
 
@@ -60,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.");
@@ -74,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)");
@@ -98,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"
@@ -116,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
@@ -141,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
@@ -158,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
@@ -170,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"))
@@ -203,7 +208,7 @@ int main(int argc, char ** argv){
 
   bool error = false;
 
-  try 
+  try
     {
       //Instantiate Chimbuko
       Chimbuko driver(params);
@@ -215,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);
@@ -267,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"
@@ -280,17 +285,14 @@ int main(int argc, char ** argv){
     }
   catch (const std::invalid_argument &e){
     std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : caught invalid argument: " << e.what() << std::endl;
-    if(params.err_outputpath.size()) recoverable_error(std::string("Driver : caught invalid argument: ") + e.what()); //ensure errors also written to error logs
     error = true;
   }
   catch (const std::ios_base::failure &e){
     std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : I/O base exception caught: " << e.what() << std::endl;
-    if(params.err_outputpath.size()) recoverable_error(std::string("Driver : I/O base exception caught: ") + e.what());
     error = true;
   }
   catch (const std::exception &e){
     std::cout << '[' << getDateTime() << ", rank " << params.rank << "] Driver : Exception caught: " << e.what() << std::endl;
-    if(params.err_outputpath.size()) recoverable_error(std::string("Driver : Exception caught: ") + e.what());
     error = true;
   }
 

app/provdb_admin.cpp

@@ -66,17 +66,6 @@ void pserver_goodbye(const tl::request& req) {
   progressStream << "ProvDB Admin: Pserver has said goodbye" << std::endl;
 }
 
-bool cmd_shutdown = false; //true if a client has requested that the server shut down
-
-void client_stop_rpc(const tl::request& req) {
-  std::lock_guard<tl::mutex> lock(*mtx);
-  cmd_shutdown = true;
-  progressStream << "ProvDB Admin: Received shutdown request from client" << std::endl;
-}
-
-
-
-
 
 struct ProvdbArgs{
   std::string ip;
@@ -154,7 +143,6 @@ int main(int argc, char** argv) {
     engine.define("client_goodbye",client_goodbye).disable_response();
     engine.define("pserver_hello",pserver_hello).disable_response();
     engine.define("pserver_goodbye",pserver_goodbye).disable_response();
-    engine.define("stop_server",client_stop_rpc).disable_response();
 
     std::string addr = (std::string)engine.self();  //ip and port of admin
 
@@ -231,10 +219,8 @@ int main(int argc, char** argv) {
 
 	    //If at least one client has previously connected but none are now connected, shutdown the server
 	    //If all clients disconnected we must also wait for the pserver to disconnect (if it is connected)
-
-	    //If args.autoshutdown is disabled we can force shutdown via a "stop_server" RPC
 	    if(
-	       (args.autoshutdown || cmd_shutdown)  && 
+	       args.autoshutdown && 
 	       ( a_client_has_connected && connected.size() == 0 ) &&
 	       ( !pserver_has_connected || (pserver_has_connected && !pserver_connected) )
 	       ){

app/provdb_shutdown.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;
   }