Make peer keepalive timeout configurable (#390)

Reduce default to 2 mins from 8 mins Signed-off-by: Tom Flynn <tom.flynn@gmail.com>
noironetworks · Apr 28, 2021 · f9ff4ed · f9ff4ed
1 parent 5deb5d4
commit f9ff4ed
Show file tree

Hide file tree

Showing 15 changed files with 72 additions and 18 deletions.
diff --git a/agent-ovs/lib/Agent.cpp b/agent-ovs/lib/Agent.cpp
@@ -179,6 +179,7 @@ void Agent::setProperties(const boost::property_tree::ptree& properties) {
     static const std::string OPFLEX_STATS_SYSTEM_INTERVAL("opflex.statistics.system.interval");
     static const std::string OPFLEX_PRR_INTERVAL("opflex.timers.prr");
     static const std::string OPFLEX_HANDSHAKE("opflex.timers.handshake-timeout");
+    static const std::string OPFLEX_KEEPALIVE("opflex.timers.keepalive-timeout");
     static const std::string DISABLED_FEATURES("feature.disabled");
     static const std::string BEHAVIOR_L34FLOWS_WITHOUT_SUBNET("behavior.l34flows-without-subnet");
 
@@ -455,6 +456,12 @@ void Agent::setProperties(const boost::property_tree::ptree& properties) {
         LOG(INFO) << "peer handshake timeout set to " << peerHandshakeTimeout << " ms";
     }
 
+    optional<uint32_t> keepaliveOpt = properties.get_optional<uint32_t>(OPFLEX_KEEPALIVE);
+    if (keepaliveOpt) {
+        keepaliveTimeout = keepaliveOpt.get();
+        LOG(INFO) << "keepalive timeout set to " << keepaliveTimeout << " ms";
+    }
+
     LOG(INFO) << "Agent mode set to " <<
        ((this->rendererFwdMode == opflex::ofcore::OFConstants::TRANSPORT_MODE)?
         "transport-mode" : "stitched-mode");
@@ -520,6 +527,7 @@ void Agent::applyProperties() {
 
     framework.setPrrTimerDuration(prr_timer);
     framework.setHandshakeTimeout(peerHandshakeTimeout);
+    framework.setKeepaliveTimeout(keepaliveTimeout);
 }
 
 void Agent::start() {

diff --git a/agent-ovs/lib/include/opflexagent/Agent.h b/agent-ovs/lib/include/opflexagent/Agent.h
@@ -330,6 +330,8 @@ typedef opflex::ofcore::OFConstants::OpflexElementMode opflex_elem_t;
     boost::uint_t<64>::fast prr_timer = 7200;  /* seconds */
     /* handshake timeout */
     uint32_t peerHandshakeTimeout = 45000;
+    /* keepalive timeout */
+    uint32_t keepaliveTimeout = 120000;
 
     std::set<std::string> endpointSourceFSPaths;
     std::set<std::string> disabledFeaturesSet;

diff --git a/agent-ovs/opflex-agent-ovs.conf.in b/agent-ovs/opflex-agent-ovs.conf.in
@@ -99,7 +99,11 @@
            //
            // How long to wait for the initial peer
            // handshake to complete (in ms)
-           // "handshake-timeout" : 45000
+           // "handshake-timeout" : 45000,
+           //
+           // How long to wait (in ms) for keepalive echo to
+           // be ack'd before timing out connection
+           // "keepalive-timeout" : 120000
        },
        // Statistics. Counters for various artifacts.
        // mode: can be either

diff --git a/agent-ovs/ovs/OvsdbConnection.cpp b/agent-ovs/ovs/OvsdbConnection.cpp
@@ -77,7 +77,7 @@ void OvsdbConnection::on_state_change(yajr::Peer * p, void * data,
     switch (stateChange) {
         case yajr::StateChange::CONNECT: {
             conn->setConnected(true);
-            p->startKeepAlive(0, 5000, 60000);
+            p->startKeepAlive(0, 5000, 180000);
             // OVSDB monitor call
             conn->sendMonitorRequests();
         }

diff --git a/libopflex/comms/CommunicationPeer.cpp b/libopflex/comms/CommunicationPeer.cpp
@@ -77,13 +77,13 @@ namespace yajr {
 void CommunicationPeer::startKeepAlive(
         uint64_t begin,
         uint64_t repeat,
-        uint64_t interval) {
-    LOG(DEBUG) << this << " interval=" << interval << " begin=" << begin << " repeat=" << repeat;
+        uint64_t timeoutAfter) {
+    LOG(DEBUG) << this << " timeoutAfter=" << timeoutAfter << " begin=" << begin << " repeat=" << repeat;
 
     sendEchoReq();
     bumpLastHeard();
 
-    keepAliveInterval_ = interval;
+    keepAliveInterval_ = timeoutAfter;
     uv_timer_start(&keepAliveTimer_, on_timeout, begin, repeat);
 }
 
@@ -339,11 +339,7 @@ void CommunicationPeer::timeout() {
         return;
     }
 
-    if (rtt <= (keepAliveInterval_ >> 3u) ) {
-        return;
-    }
-
-    if (rtt > (keepAliveInterval_ << 3u) ) {
+    if (rtt > keepAliveInterval_) {
         LOG(WARNING) << this << " tearing down the connection upon timeout";
         /* close the connection and hope for the best */
         this->onDisconnect();

diff --git a/libopflex/engine/OpflexClientConnection.cpp b/libopflex/engine/OpflexClientConnection.cpp
@@ -151,7 +151,7 @@ void OpflexClientConnection::on_state_change(Peer * p, void * data,
 
         if (conn->pool->clientCtx.get())
             ZeroCopyOpenSSL::attachTransport(p, conn->pool->clientCtx.get());
-        p->startKeepAlive(10000, 15000, 60000);
+        p->startKeepAlive(10000, 15000, conn->getKeepaliveTimeout());
 
         conn->pool->updatePeerStatus(conn->hostname, conn->port,
                                      PeerStatusListener::CONNECTED);

diff --git a/libopflex/engine/OpflexPEHandler.cpp b/libopflex/engine/OpflexPEHandler.cpp
@@ -125,6 +125,7 @@ class SendIdentityReq : public OpflexMessage {
 OpflexPEHandler::OpflexPEHandler(OpflexConnection* conn, Processor* processor_)
     : OpflexHandler(conn), processor(processor_) {
     conn->setHandshakeTimeout(processor_->getHandshakeTimeout());
+    conn->setKeepaliveTimeout(processor_->getKeepaliveTimeout());
 }
 
 void OpflexPEHandler::connected() {

diff --git a/libopflex/engine/OpflexServerConnection.cpp b/libopflex/engine/OpflexServerConnection.cpp
@@ -118,7 +118,7 @@ void OpflexServerConnection::on_state_change(yajr::Peer * p, void * data,
             ZeroCopyOpenSSL::Ctx* serverCtx = conn->listener->serverCtx.get();
             if (serverCtx)
                 ZeroCopyOpenSSL::attachTransport(p, serverCtx);
-            p->startKeepAlive(10000, 15000, 45000);
+            p->startKeepAlive(10000, 15000, 120000);
 
             conn->handler->connected();
         }

diff --git a/libopflex/engine/include/opflex/engine/Processor.h b/libopflex/engine/include/opflex/engine/Processor.h
@@ -205,6 +205,20 @@ class Processor : public internal::AbstractObjectListener,
         peerHandshakeTimeout = timeout;
     }
 
+    /**
+     * Get the keepalive timeout
+     */
+    uint32_t getKeepaliveTimeout() const {
+        return keepaliveTimeout;
+    }
+
+    /**
+     * Set the keepalive timeout
+     */
+    void setKeepaliveTimeout(const uint32_t timeout) {
+        keepaliveTimeout = timeout;
+    }
+\
     /**
      * Set the prr timer duration in secs
      */
@@ -524,6 +538,7 @@ class Processor : public internal::AbstractObjectListener,
     uint64_t prrTimerDuration = DEFAULT_PRR_TIMER_DURATION;
 
     uint32_t peerHandshakeTimeout = 45000;
+    uint32_t keepaliveTimeout = 120000;
 
     /**
      *  policy refresh timer duration in msecs

diff --git a/libopflex/engine/include/opflex/engine/internal/OpflexConnection.h b/libopflex/engine/include/opflex/engine/internal/OpflexConnection.h
@@ -125,6 +125,22 @@ class OpflexConnection : public opflex::jsonrpc::RpcConnection {
         handshakeTimeout = timeout;
     }
 
+    /**
+     * Get the keepalive timeout (in ms)
+     * @return timeout
+     */
+    uint32_t getKeepaliveTimeout() const {
+        return keepaliveTimeout;
+    }
+
+    /**
+     * Set the keepalive timeout (in ms)
+     * @param timeout timeout
+     */
+    void setKeepaliveTimeout(uint32_t timeout) {
+        keepaliveTimeout = timeout;
+    }
+
 protected:
     /**
      * The handler for the connection
@@ -133,6 +149,7 @@ class OpflexConnection : public opflex::jsonrpc::RpcConnection {
 
 private:
     uint32_t handshakeTimeout;
+    uint32_t keepaliveTimeout;
 
     virtual void notifyReady();
     virtual void notifyFailed() {}

diff --git a/libopflex/include/opflex/ofcore/OFFramework.h b/libopflex/include/opflex/ofcore/OFFramework.h
@@ -741,6 +741,12 @@ class OFFramework : private boost::noncopyable {
      */
      void setHandshakeTimeout(const uint32_t timeout);
 
+    /**
+     * Set the keepalive timeout
+     * @param timeout keepalive timeout in milliseconds
+     */
+    void setKeepaliveTimeout(const uint32_t timeout);
+
     /**
      * Start the framework.  This will start all the framework threads
      * and attempt to connect to configured OpFlex peers.

diff --git a/libopflex/include/opflex/yajr/internal/comms.hpp b/libopflex/include/opflex/yajr/internal/comms.hpp
@@ -571,12 +571,12 @@ class CommunicationPeer : public Peer, virtual public ::yajr::Peer {
      * Start TCP keepalive to peer
      * @param begin delay before starting
      * @param repeat repeat
-     * @param interval interval
+     * @param timeoutAfter timeout after
      */
     virtual void startKeepAlive(
-            uint64_t begin    =  100,
-            uint64_t repeat   = 1250,
-            uint64_t interval = 9000);
+            uint64_t begin        =  100,
+            uint64_t repeat       = 1250,
+            uint64_t timeoutAfter = 9000);
 
     /**
      * Stop TCP keepalive

diff --git a/libopflex/include/opflex/yajr/yajr.hpp b/libopflex/include/opflex/yajr/yajr.hpp
@@ -222,12 +222,12 @@ class Peer {
      *
      * @param begin delay before starting
      * @param repeat repeat
-     * @param interval interval
+     * @param timeoutAfter timeoutAfter
      */
     virtual void startKeepAlive(
             uint64_t                begin             = 100,
             uint64_t                repeat            = 1250,
-            uint64_t                interval          = 9000
+            uint64_t                timeoutAfter      = 9000
     ) = 0;
 
     /**

diff --git a/libopflex/ofcore/OFFramework.cpp b/libopflex/ofcore/OFFramework.cpp
@@ -115,6 +115,10 @@ void OFFramework::setHandshakeTimeout(const uint32_t timeout) {
     pimpl->processor.setHandshakeTimeout(timeout);
 }
 
+void OFFramework::setKeepaliveTimeout(const uint32_t timeout) {
+    pimpl->processor.setKeepaliveTimeout(timeout);
+}
+
 void OFFramework::start() {
     LOG(DEBUG) << "Starting OpFlex Framework";
     pimpl->started = true;

diff --git a/libopflex/ofcore/test/OFFramework_test.cpp b/libopflex/ofcore/test/OFFramework_test.cpp
@@ -64,6 +64,7 @@ BOOST_AUTO_TEST_CASE( test_misc ) {
     BOOST_CHECK_EQUAL(opflex::ofcore::OFConstants::TRANSPORT_MODE, fw.getElementMode());
     fw.setPrrTimerDuration(12345);
     fw.setHandshakeTimeout(54321);
+    fw.setKeepaliveTimeout(123456);
     boost::asio::ip::address_v4 proxy;
     fw.getV4Proxy(proxy);
     fw.getV6Proxy(proxy);