Merge pull request #942 from lsst/tickets/DM-51795

Changed FQDN calls to blocking or using stored value.

Author: jgates108

src/cconfig/CzarConfig.h

@@ -125,7 +125,7 @@ class CzarConfig {
     int getMaxTableSizeMB() const { return _maxTableSizeMB->getVal(); }
     int getMaxSqlConnectionAttempts() const { return _maxSqlConnectionAttempts->getVal(); }
 
-    /// The size of the TCP connection pool witin the client API that is used
+    /// The size of the TCP connection pool within the client API that is used
     /// by the merger to pool result files from workers via the HTTP protocol.
     int getResultMaxHttpConnections() const { return _resultMaxHttpConnections->getVal(); }
 

src/czar/Czar.cc

@@ -163,7 +163,8 @@ Czar::Czar(string const& configFilePath, string const& czarName)
           _uqFactory(),
           _clientToQuery(),
           _monitorSleepTime(_czarConfig->getMonitorSleepTimeMilliSec()),
-          _activeWorkerMap(new ActiveWorkerMap(_czarConfig)) {
+          _activeWorkerMap(new ActiveWorkerMap(_czarConfig)),
+          _fqdn(util::getCurrentHostFqdnBlocking()) {
     // set id counter to milliseconds since the epoch, mod 1 year.
     struct timeval tv;
     gettimeofday(&tv, nullptr);

src/czar/Czar.h

@@ -161,6 +161,8 @@ class Czar {
 
     std::shared_ptr<http::ClientConnPool> getCommandHttpPool() const { return _commandHttpPool; }
 
+    std::string const& getFqdn() const { return _fqdn; }
+
     /// Startup time of czar, sent to workers so they can detect that the czar was
     /// was restarted when this value changes.
     static uint64_t const czarStartupTime;
@@ -254,6 +256,9 @@ class Czar {
     /// Pool of http client connections for sending commands (UberJobs
     /// and worker status requests).
     std::shared_ptr<http::ClientConnPool> _commandHttpPool;
+
+    /// FQDN for this czar.
+    std::string const _fqdn;
 };
 
 }  // namespace lsst::qserv::czar

src/czar/CzarRegistry.cc

@@ -80,13 +80,14 @@ void CzarRegistry::_registryUpdateLoop() {
     string const url = "http://" + _czarConfig->replicationRegistryHost() + ":" +
                        to_string(_czarConfig->replicationRegistryPort()) + "/czar";
     vector<string> const headers = {"Content-Type: application/json"};
+    string const fqdn = util::getCurrentHostFqdnBlocking();
     json const request = json::object({{"instance_id", _czarConfig->replicationInstanceId()},
                                        {"auth_key", _czarConfig->replicationAuthKey()},
                                        {"czar",
                                         {{"name", _czarConfig->name()},
                                          {"id", _czarConfig->id()},
                                          {"management-port", _czarConfig->replicationHttpPort()},
-                                         {"management-host-name", util::get_current_host_fqdn()}}}});
+                                         {"management-host-name", fqdn}}}});
     string const requestContext = "Czar: '" + http::method2string(method) + "' request to '" + url + "'";
     LOGS(_log, LOG_LVL_TRACE,
          __func__ << " czarPost url=" << url << " request=" << request.dump() << " headers=" << headers[0]);
@@ -113,7 +114,7 @@ void CzarRegistry::_registryWorkerInfoLoop() {
     string const replicationInstanceId = _czarConfig->replicationInstanceId();
     string const replicationAuthKey = _czarConfig->replicationAuthKey();
     uint64_t const czarStartTime = Czar::czarStartupTime;
-
+    string const fqdn = util::getCurrentHostFqdnBlocking();
     vector<string> const headers;
     auto const method = http::Method::GET;
     string const url = "http://" + _czarConfig->replicationRegistryHost() + ":" +
@@ -133,9 +134,9 @@ void CzarRegistry::_registryWorkerInfoLoop() {
                 protojson::WorkerContactInfo::WCMapPtr wMap = _buildMapFromJson(response);
                 // Update the values in the map
                 {
-                    auto czInfo = protojson::CzarContactInfo::create(
-                            _czarConfig->name(), _czarConfig->id(), _czarConfig->replicationHttpPort(),
-                            util::get_current_host_fqdn(), czarStartTime);
+                    auto czInfo = protojson::CzarContactInfo::create(_czarConfig->name(), _czarConfig->id(),
+                                                                     _czarConfig->replicationHttpPort(), fqdn,
+                                                                     czarStartTime);
                     lock_guard lck(_cmapMtx);
                     if (wMap != nullptr) {
                         _contactMap = wMap;

src/qdisp/UberJob.cc

@@ -127,7 +127,7 @@ void UberJob::runUberJob() {
     uint64_t maxTableSizeMB = czarConfig->getMaxTableSizeMB();
     auto czInfo = protojson::CzarContactInfo::create(
             czarConfig->name(), czarConfig->id(), czarConfig->replicationHttpPort(),
-            util::get_current_host_fqdn(), czar::Czar::czarStartupTime);
+            czar::Czar::getCzar()->getFqdn(), czar::Czar::czarStartupTime);
     auto scanInfoPtr = exec->getScanInfo();
     bool scanInteractive = exec->getScanInteractive();
 

src/wmain/WorkerMain.cc

@@ -95,46 +95,6 @@ std::shared_ptr<wpublish::ChunkInventory> makeChunkInventory(string const& worke
     return inventory;
 }
 
-/**
- * This function will keep periodically updating worker's info in the Replication
- * System's Registry.
- * @param id The unique identifier of a worker to be registered.
- * @note The thread will terminate the process if the registraton request to the Registry
- * was explicitly denied by the service. This means the application may be misconfigured.
- * Transient communication errors when attempting to connect or send requests to
- * the Registry will be posted into the log stream and ignored.
- */
-void registryUpdateLoop(string const& id) {
-    auto const workerConfig = wconfig::WorkerConfig::instance();
-    auto const method = http::Method::POST;
-    string const url = "http://" + workerConfig->replicationRegistryHost() + ":" +
-                       to_string(workerConfig->replicationRegistryPort()) + "/qserv-worker";
-    vector<string> const headers = {"Content-Type: application/json"};
-    json const request = json::object({{"version", http::MetaModule::version},
-                                       {"instance_id", workerConfig->replicationInstanceId()},
-                                       {"auth_key", workerConfig->replicationAuthKey()},
-                                       {"worker",
-                                        {{"name", id},
-                                         {"management-port", workerConfig->replicationHttpPort()},
-                                         {"management-host-name", util::get_current_host_fqdn()}}}});
-    string const requestContext =
-            "WorkerMain: '" + http::method2string(method) + "' request to '" + url + "'";
-    http::Client client(method, url, request.dump(), headers);
-    while (true) {
-        try {
-            json const response = client.readAsJson();
-            if (0 == response.at("success").get<int>()) {
-                string const error = response.at("error").get<string>();
-                LOGS(_log, LOG_LVL_ERROR, requestContext + " was denied, error: '" + error + "'.");
-                abort();
-            }
-        } catch (exception const& ex) {
-            LOGS(_log, LOG_LVL_WARN, requestContext + " failed, ex: " + ex.what());
-        }
-        this_thread::sleep_for(chrono::seconds(max(1U, workerConfig->replicationRegistryHearbeatIvalSec())));
-    }
-}
-
 }  // namespace
 
 namespace lsst::qserv::wmain {
@@ -265,24 +225,70 @@ WorkerMain::WorkerMain() {
 
     // Begin periodically updating worker's status in the Replication System's registry
     // in the detached thread. This will continue before the application gets terminated.
-    thread registryUpdateThread(::registryUpdateLoop, _foreman->chunkInventory()->id());
-    registryUpdateThread.detach();
+    thread registryUpdateThread(&WorkerMain::_registryUpdateLoop, this);
+    _registryUpdateThread = move(registryUpdateThread);
 }
 
 void WorkerMain::waitForTerminate() {
     unique_lock uniq(_terminateMtx);
-    _terminateCv.wait(uniq, [this]() { return _terminate; });
+    _terminateCv.wait(uniq, [this]() -> bool { return _terminate; });
 }
 
 void WorkerMain::terminate() {
-    lock_guard lck(_terminateMtx);
-    _terminate = true;
+    {
+        lock_guard lck(_terminateMtx);
+        if (_terminate.exchange(true)) return;
+        ;
+    }
     _terminateCv.notify_all();
+    _controlHttpSvc->stop();
 }
 
 WorkerMain::~WorkerMain() {
     LOGS(_log, LOG_LVL_INFO, "WorkerMain shutdown.");
-    _controlHttpSvc->stop();
+    terminate();
+    _registryUpdateThread.join();
+}
+
+/**
+ * This function will keep periodically updating worker's info in the Replication
+ * System's Registry.
+ * @param id The unique identifier of a worker to be registered.
+ * @note The thread will terminate the process if the registraton request to the Registry
+ * was explicitly denied by the service. This means the application may be misconfigured.
+ * Transient communication errors when attempting to connect or send requests to
+ * the Registry will be posted into the log stream and ignored.
+ */
+void WorkerMain::_registryUpdateLoop() {
+    string const id = _foreman->chunkInventory()->id();
+    auto const workerConfig = wconfig::WorkerConfig::instance();
+    auto const method = http::Method::POST;
+    string const url = "http://" + workerConfig->replicationRegistryHost() + ":" +
+                       to_string(workerConfig->replicationRegistryPort()) + "/qserv-worker";
+    vector<string> const headers = {"Content-Type: application/json"};
+    json const request = json::object({{"version", http::MetaModule::version},
+                                       {"instance_id", workerConfig->replicationInstanceId()},
+                                       {"auth_key", workerConfig->replicationAuthKey()},
+                                       {"worker",
+                                        {{"name", id},
+                                         {"management-port", workerConfig->replicationHttpPort()},
+                                         {"management-host-name", _foreman->getFqdn()}}}});
+    string const requestContext =
+            "WorkerMain: '" + http::method2string(method) + "' request to '" + url + "'";
+    http::Client client(method, url, request.dump(), headers);
+    while (!_terminate) {
+        try {
+            json const response = client.readAsJson();
+            if (0 == response.at("success").get<int>()) {
+                string const error = response.at("error").get<string>();
+                LOGS(_log, LOG_LVL_ERROR, requestContext + " was denied, error: '" + error + "'.");
+                abort();
+            }
+        } catch (exception const& ex) {
+            LOGS(_log, LOG_LVL_WARN, requestContext + " failed, ex: " + ex.what());
+        }
+        this_thread::sleep_for(chrono::seconds(max(1U, workerConfig->replicationRegistryHearbeatIvalSec())));
+    }
 }
 
 }  // namespace lsst::qserv::wmain

src/wmain/WorkerMain.h

@@ -59,12 +59,16 @@ class WorkerMain {
 
     std::string getName() const { return _name; }
 
+    /// End WorkerMain, calling this multiple times is harmless.
     void terminate();
     void waitForTerminate();
 
 private:
     WorkerMain();
 
+    void _registryUpdateLoop();
+    std::thread _registryUpdateThread;
+
     /// Weak pointer to allow global access without complicating lifetime issues.
     static std::weak_ptr<WorkerMain> _globalWorkerMain;
 
@@ -85,7 +89,7 @@ class WorkerMain {
     std::shared_ptr<wcomms::HttpSvc> _controlHttpSvc;
 
     /// Set to true when the program should terminate.
-    bool _terminate = false;
+    std::atomic<bool> _terminate{false};
     std::mutex _terminateMtx;
     std::condition_variable _terminateCv;
 };