Changeset 950cd4

Timestamp:

Aug 11, 2025, 5:44:42 PM (2 months ago)

Author:

Frederik Heber <frederik.heber@…>

Branches:

Candidate_v1.7.0, stable

Children:

74d798

Parents:

e5f9e7

git-author:

Frederik Heber <frederik.heber@…> (07/04/25 20:57:33)

git-committer:

Frederik Heber <frederik.heber@…> (08/11/25 17:44:42)

Message:

Added HealthProbeListener to PoolWorker.

• this allows a simple tcp liveness probe where kubernetes only checks for a connection on a port can be established.

Location:

ThirdParty/JobMarket/src/JobMarket

Files:

• Pool/PoolWorker.cpp (modified) (4 diffs)
• Pool/PoolWorker.hpp (modified) (3 diffs)
• WorkerOptions.cpp (modified) (1 diff)
• WorkerOptions.hpp (modified) (2 diffs)
• poolworker_main.cpp (modified) (3 diffs)

ThirdParty/JobMarket/src/JobMarket/Pool/PoolWorker.cpp

@@ -56 +56 @@
     const std::string& _service,
     const std::string& listenhost,
-    const std::string& listenservice) :
+    const std::string& listenservice,
+    const std::string& healthport) :
   io_service(_io_service),
   PoolListener(_io_service, boost::lexical_cast<unsigned short>(listenservice), *this),
+  HealthListener(_io_service, boost::lexical_cast<unsigned short>(healthport), PoolListener),
   MyAddress(listenhost, listenservice),
   ServerAddress(_host, _service),
@@ -70 +72 @@
   if (PoolListener.getExitflag() == Listener::OkFlag) {
     // always enroll and make listenining initiation depend on its success
-    const boost::function<void ()> initiateme =
-        boost::bind(&PoolListener_t::initiateSocket, boost::ref(PoolListener));
+    const boost::function<void ()> initiateme = 
+        boost::bind(&PoolWorker::initiateSockets, this);
     const boost::function<void ()> shutdown =
         boost::bind(&PoolWorker::finish, this); // no need to remove from pool
@@ -82 +84 @@
   }
 }
+
+void PoolWorker::initiateSockets() 
+{
+  PoolListener.initiateSocket();
+  HealthListener.initiateSocket();
+}
+
+/// Handle completion of a accept server operation.
+void PoolWorker::HealthProbeListener_t::handle_Accept(const boost::system::error_code& e, connection_ptr conn)
+{
+  DEBUG_FUNCTION_ENTRYEXIT
+  if (!e)
+  {
+    if (PoolListener.getExitflag() == Listener::OkFlag) {
+      LOG(2, "DEBUG: Ping from health check");
+      // and listen for following connections
+      initiateSocket();
+    } else {
+      LOG(2, "INFO: Exitflag is " <<  PoolListener.getExitflag() << ", unhealthy ...");
+      closeSocket();
+    }
+  }
+  else
+  {
+    // An error occurred. Log it and return. Since we are not starting a new
+    // accept operation the io_service will run out of work to do and the
+    // server will exit.
+    setExitflag( ErrorFlag );
+    ELOG(0, e.message());
+  }
+}
+
 
 /// Handle completion of a accept server operation.
@@ -324 +358 @@
 {
   // somehow stop listener
+  HealthListener.closeSocket();
+
+  // somehow stop listener
   PoolListener.closeSocket();
 

ThirdParty/JobMarket/src/JobMarket/Pool/PoolWorker.hpp

@@ -45 +45 @@
       const std::string& service,
       const std::string& listenhost,
-      const std::string& listenservice);
+      const std::string& listenservice,
+      const std::string& healthport);
 
   /** Returns the flag of the handled operation.
@@ -122 +123 @@
   };
 
+    /**
+   * Handles TCP liveness probes.
+   */
+  class HealthProbeListener_t : public Listener
+  {
+  public:
+    HealthProbeListener_t(
+        boost::asio::io_service& io_service,
+        unsigned short port,
+        const PoolListener_t& _PoolListener) :
+      Listener(io_service, port),
+      PoolListener(_PoolListener)
+    {}
+
+    virtual ~HealthProbeListener_t() {}
+
+  protected:
+    /// Handle completion of a accept controller operation.
+    void handle_Accept(const boost::system::error_code& e, connection_ptr conn);
+
+  private:
+    const PoolListener_t& PoolListener;
+  };
+
+private:
+    void initiateSockets();
+
 private:
   //!> reference to io_service which we use for connections
@@ -128 +156 @@
   //!> The listener for the WorkerPool
   PoolListener_t PoolListener;
+
+  //!> The listener for the WorkerPool
+  HealthProbeListener_t HealthListener;
 
   //!> address of this worker

ThirdParty/JobMarket/src/JobMarket/WorkerOptions.cpp

@@ -85 +85 @@
   return 0;
 }
+
+int WorkerOptions::parseHealthProbePort(boost::program_options::variables_map &vm) {
+  if (vm.count("healthport")) {
+    try {
+      healthprobeport = vm["healthport"].as< std::string >();
+    } catch (boost::bad_lexical_cast) {
+      ELOG(1, "Could not read " << vm["healthport"].as< std::string >() << " as digits.");
+      return 255;
+    }
+    LOG(1, "INFO: Using port " << healthprobeport << " to listen fdor health probe connects.");
+  } else {
+    healthprobeport = "";
+  }
+  return 0;
+}

ThirdParty/JobMarket/src/JobMarket/WorkerOptions.hpp

@@ -29 +29 @@
   int parseLocalhost(boost::program_options::variables_map &vm);
   int parseListenPort(boost::program_options::variables_map &vm);
+  int parseHealthProbePort(boost::program_options::variables_map &vm);
 
   std::string server;
@@ -34 +35 @@
   std::string hostname;
   std::string listenport;
+  std::string healthprobeport;
 };
 

ThirdParty/JobMarket/src/JobMarket/poolworker_main.cpp

@@ -58 +58 @@
       ("server", boost::program_options::value< std::string>(), "connect to server at this address (host:port)")
       ("listen", boost::program_options::value< std::string >(), "listen on this port")
+      ("healthport", boost::program_options::value< std::string >(), "listen on this port for health probes")
       ("hostname", boost::program_options::value< std::string>(), "name of host on which this codes runs and which server can resolve")
   ;
@@ -76 +77 @@
   status = WorkerOpts.parseLocalhost(vm);
   if (status) return status;
+  status = WorkerOpts.parseHealthProbePort(vm);
+  if (status) return status;
   status = WorkerOpts.parseSignals(vm);
   if (status) return status;
@@ -83 +86 @@
   {
     boost::asio::io_service io_service;
-    PoolWorker client(io_service, WorkerOpts.server, WorkerOpts.serverport, WorkerOpts.hostname, WorkerOpts.listenport);
+    PoolWorker client(io_service, WorkerOpts.server, WorkerOpts.serverport, WorkerOpts.hostname, WorkerOpts.listenport, "8092");
 
     // catch ctrl-c and shutdown worker properly