ADProvenanceDBengine now initializes the Thallium server with a json configuration rather than the basic constructor

Mercury authorization key can now optionally be passed to ADProvenanceDBengine
Added cmdline options to pass Mercury authorization key to pserver, driver and provdb_commit
run_services.sh and chimbuko_config.sh now allow passing extra arguments to provdb_commit

Author: giltirn

app/driver.cpp

@@ -88,6 +88,7 @@ optionalArgsParser & getOptionalArgsParser(){
     addOptionalCommandLineArg(p, provdb_addr_dir, "Directory in which the provenance database outputs its address files. If empty (default) the provenance DB will not be used.");
     addOptionalCommandLineArg(p, nprovdb_shards, "Number of provenance database shards. Clients connect to shards round-robin by rank (default 1)");
     addOptionalCommandLineArg(p, nprovdb_instances, "Number of provenance database instances. Shards are divided uniformly over instances. (default 1)");
+    addOptionalCommandLineArg(p, provdb_mercury_auth_key, "Set the Mercury authorization key for connection to the provDB (default \"\")");
 #endif
 #ifdef _PERF_METRIC
     addOptionalCommandLineArg(p, perf_outputpath, "Output path for AD performance monitoring data. If an empty string (default) no output is written.");

app/provdb_commit.cpp

@@ -49,8 +49,9 @@ struct Args{
   int ninstances;
   int nshards;
   int freq_ms;
-
-  Args(): addr_file_dir("."), instance(0), ninstances(1), nshards(1), freq_ms(30000){}
+  std::string provdb_mercury_auth_key; //An authorization key for initializing Mercury (optional, default "")
+  
+  Args(): addr_file_dir("."), instance(0), ninstances(1), nshards(1), freq_ms(30000), provdb_mercury_auth_key(""){}
 };
 
 
@@ -63,7 +64,8 @@ int main(int argc, char** argv){
   addOptionalCommandLineArg(parser, ninstances, "Specify the number of server instances (default 1)");
   addOptionalCommandLineArg(parser, nshards, "Specify the total number of database shards (default 1)");
   addOptionalCommandLineArg(parser, freq_ms, "Specify the frequency in ms at which commit is called (default 30000)");
-
+  addOptionalCommandLineArg(parser, provdb_mercury_auth_key, "Set the Mercury authorization key for connection to the provDB (default \"\")");
+  
   if(argc-1 < parser.nMandatoryArgs() || (argc == 2 && std::string(argv[1]) == "-help")){
     parser.help(std::cout);
     return 0;    
@@ -78,6 +80,11 @@ int main(int argc, char** argv){
     return 0;
   } 
 
+  if(args.provdb_mercury_auth_key != ""){
+    CprogressStream << "setting Mercury authorization key to \"" << args.provdb_mercury_auth_key << "\"" << std::endl;
+    ADProvenanceDBengine::setMercuryAuthorizationKey(args.provdb_mercury_auth_key);
+  }
+  
   ProvDBsetup setup(args.nshards, args.ninstances);
   std::string addr = setup.getInstanceAddress(args.instance, args.addr_file_dir);
   int instance_shard_offset = setup.getShardOffsetInstance(args.instance);

app/pserver.cpp

@@ -54,6 +54,7 @@ struct pserverArgs{
 
 #ifdef ENABLE_PROVDB
   std::string provdb_addr_dir;
+  std::string provdb_mercury_auth_key; //An authorization key for initializing Mercury (optional, default "")
 #endif
 
   std::string prov_outputpath;
@@ -63,7 +64,7 @@ struct pserverArgs{
 	       , max_pollcyc_msg(10), zmq_io_thr(1), autoshutdown(true)
 #endif
 #ifdef ENABLE_PROVDB
-	       , provdb_addr_dir("")
+	       , provdb_addr_dir(""), provdb_mercury_auth_key("")
 #endif
   {}
 
@@ -88,6 +89,7 @@ struct pserverArgs{
 #endif
 #ifdef ENABLE_PROVDB
       addOptionalCommandLineArg(p, provdb_addr_dir, "The directory containing the address file written out by the provDB server. An empty string will disable the connection to the global DB.  (default empty, disabled)");
+      addOptionalCommandLineArg(p, provdb_mercury_auth_key, "Set the Mercury authorization key for connection to the provDB (default \"\")");
 #endif
       addOptionalCommandLineArg(p, prov_outputpath, "Output global 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");
       addOptionalCommandLineArg(p, model_update_freq, "The frequency in ms at which the global AD model is updated (default 1000ms)");
@@ -158,6 +160,11 @@ int main (int argc, char ** argv){
   PSstatSender stat_sender(args.stat_send_freq);
 
 #ifdef ENABLE_PROVDB
+  if(args.provdb_mercury_auth_key != ""){
+    progressStream << "Pserver: setting Mercury authorization key to \"" << args.provdb_mercury_auth_key << "\"" << std::endl;
+    ADProvenanceDBengine::setMercuryAuthorizationKey(args.provdb_mercury_auth_key);
+  }
+  
   PSProvenanceDBclient provdb_client;
 #endif
 

include/chimbuko/ad/ADProvenanceDBengine.hpp

@@ -19,8 +19,10 @@ namespace chimbuko{
     struct data_v{
       thallium::engine* m_eng;
       std::pair<std::string, int> m_protocol; /**< The protocol and mode (client/server)*/
+      std::string m_mercury_auth_key; /**< Mercury authorization key (optional)*/
+      
       bool m_is_initialized;
-      inline data_v(): m_eng(nullptr), m_protocol({"ofi+tcp;ofi_rxm", THALLIUM_CLIENT_MODE}), m_is_initialized(false){}
+      inline data_v(): m_eng(nullptr), m_protocol({"ofi+tcp;ofi_rxm", THALLIUM_CLIENT_MODE}), m_mercury_auth_key(""), m_is_initialized(false){}
 
       /**
        * @brief Initialize the engine (if not already initialized)
@@ -52,6 +54,13 @@ namespace chimbuko{
       data().m_protocol = {transport, mode};
     }
 
+    /**
+     * @brief Set the Mercury authorization key. Must be done before the engine is created
+     */
+    static inline void setMercuryAuthorizationKey(const std::string &to){
+      data().m_mercury_auth_key = to;
+    }
+    
     /**
      * @brief Get the protocol used to create the engine.
      */

include/chimbuko/chimbuko.hpp

@@ -45,6 +45,7 @@ namespace chimbuko {
     std::string provdb_addr_dir; /**< Directory in which the provenance database writes its address files. If an empty string the provDB will not be used*/
     int nprovdb_shards; /**< Number of database shards*/
     int nprovdb_instances; /**< Number of instances of the provenance database server*/
+    std::string provdb_mercury_auth_key; /**< An authorization key for initializing Mercury (optional, default "")*/
 #endif
     int prov_record_startstep; /**< If != -1, the IO step on which to start recording provenance information for anomalies */
     int prov_record_stopstep; /**< If != -1, the IO step on which to stop recording provenance information for anomalies */

scripts/launch/chimbuko_config.sh

@@ -36,6 +36,8 @@ provdb_auto_interface=true #if true, the server will automatically choose its in
 #With "verbs" provider (used for infiniband, iWarp, etc) we need to also specify the domain, which can be found by running fi_info (on a compute node)
 provdb_domain= #only needed for verbs provider. If left blank it will be chosen automatically. <------------ ***SET ME (if using verbs)***
 
+commit_extra_args="" #extra arguments for the committer
+
 export FI_UNIVERSE_SIZE=1600  # Defines the expected number of provenance DB clients per instance <------------- *** SET ME (should be larger than the number of clients/instance)
 export FI_MR_CACHE_MAX_COUNT=0   # disable MR cache in libfabric; still problematic as of libfabric 1.10.1
 export FI_OFI_RXM_USE_SRX=1 # use shared recv context in RXM; should improve scalability

scripts/launch/run_services.sh

@@ -159,8 +159,8 @@ if (( ${use_provdb} == 1 )); then
     echo "Instantiating committer"
     echo "==========================================="
     for((i=0;i<provdb_ninstances;i++)); do
-	echo "Chimbuko services launching provDB committer ${i} of ${provdb_ninstances}"
-	provdb_commit "${provdb_addr_dir}" -instance ${i} -ninstances ${provdb_ninstances} -nshards ${provdb_nshards} -freq_ms ${provdb_commit_freq} > ${log_dir}/committer_${i}.log 2>&1 &
+	echo "Chimbuko services launching provDB committer ${i} of ${provdb_ninstances} using extra args \"${commit_extra_args}\""
+	provdb_commit "${provdb_addr_dir}" -instance ${i} -ninstances ${provdb_ninstances} -nshards ${provdb_nshards} -freq_ms ${provdb_commit_freq} ${commit_extra_args} > ${log_dir}/committer_${i}.log 2>&1 &
 	sleep 1
     done
     sleep 3

src/ad/ADProvenanceDBengine.cpp

@@ -1,5 +1,6 @@
 #include<chimbuko/ad/ADProvenanceDBclient.hpp>
 #include<chimbuko/verbose.hpp>
+#include <nlohmann/json.hpp>
 
 #ifdef ENABLE_PROVDB
 
@@ -9,7 +10,14 @@ using namespace chimbuko;
 void ADProvenanceDBengine::data_v::initialize(){
   if(!m_is_initialized){
     verboseStream << "ADProvenanceDBengine initializing Thallium engine" << std::endl;
-    m_eng = new thallium::engine(m_protocol.first, m_protocol.second, true, -1); //3rd arg: use dedicated progress thread, 4th arg: use this thread for RPCs
+
+    nlohmann::json cfgj;
+    cfgj["use_progress_thread"] = true;
+    cfgj["rpc_thread_count"] = -1;  //use this thread for RPCs
+    if(m_mercury_auth_key != "") cfgj["mercury"]["auth_key"] = m_mercury_auth_key;
+    std::string config = cfgj.dump();
+    
+    m_eng = new thallium::engine(m_protocol.first, m_protocol.second, config.c_str());
     m_is_initialized = true;
     verboseStream << "ADProvenanceDBengine initialized Thallium engine" << std::endl;
   }

src/chimbuko.cpp

@@ -21,7 +21,7 @@ ChimbukoParams::ChimbukoParams(): rank(-1234),  //not set!
 				  hbos_use_global_threshold(true),
 				  hbos_max_bins(200),
 #ifdef ENABLE_PROVDB
-				  provdb_addr_dir(""), nprovdb_shards(1), nprovdb_instances(1),
+				  provdb_addr_dir(""), nprovdb_shards(1), nprovdb_instances(1), provdb_mercury_auth_key(""),
 #endif
 				  prov_outputpath(""),
                                   prov_record_startstep(-1),
@@ -281,6 +281,11 @@ void Chimbuko::init_counter(){
 
 #ifdef ENABLE_PROVDB
 void Chimbuko::init_provdb(){
+  if(m_params.provdb_mercury_auth_key != ""){
+    headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " setting Mercury authorization key to \"" << m_params.provdb_mercury_auth_key << "\"" << std::endl;
+    ADProvenanceDBengine::setMercuryAuthorizationKey(m_params.provdb_mercury_auth_key);
+  }
+  
   m_provdb_client = new ADProvenanceDBclient(m_params.rank);
   if(m_params.provdb_addr_dir.length() > 0){
     headProgressStream(m_params.rank) << "driver rank " << m_params.rank << " connecting to provenance database" << std::endl;