OpenR Initialization Improvements

Summary:
This set of changes implements fix for Issue 1 and 2 as described in OpenR initialization improvements here: https://fb.workplace.com/groups/open.routing/permalink/1977709525965711/

Fix for 1st issue is addressed by (1) and (2) described below. Fix for 2nd issue is addressed by (3) described below.

1  Advertise local adjacencies from link-monitor to kvstore only after KVSTORE_SYNCED. We still keep adj hold timer to advertise adjacencies in case of KVSTORE_SYNCED signal not received by this hold timer. Changes to support this function are,
  - Create a dispatcher reader to pass kvstore updates from kvstore to link-monitor
  - Link-monitor module listens to this queue, and specifically reacts to KVSTORE_SYNCED signal
  - On KVSTORE_SYNCED signal, link-monitor cancels adjHoldTimer_ and advertise adjacencies to kvstore
  - Added a configuration knob to allow user disable this behavior
  - Add supporting test-cases in LinkMonitorTest.cpp

2  Decision to compute RIB only after both KVSTORE_SYNCED and ADJACENCY_DB_SYNCED signals received. Changes to support this function are,
  - In Decision module, now wait for both signals to trigger/unblock initial route build
  - Added a configuration knob to allow user disable this behavior
  - Add supporting test-cases in DecisionTest.cpp

3  adjOnlyUsedByOtherNode flag settings
  - Do not set adjOnlyUsedByOtherNode flag when local node is still in initialization
  - Add more tests coverage around use of GR cases

4  Reset counters during SetUp of prefix-manager UT tests to fix flakiness of a particular counter test

The Open/R initialization sequence improvements are fixed and enabled by default. Configuration knobs are provided in case of a need to go back to the old behavior.

Reviewed By: xiangxu1121

Differential Revision: D62284524

fbshipit-source-id: 737e43211f73129b56fee886daf80416eab670d5

openr/Main.cpp

@@ -196,10 +196,6 @@ main(int argc, char** argv) {
 
   // KvStore -> Subscribers
   ReplicateQueue<KvStorePublication> kvStoreUpdatesQueue;
-  auto decisionKvStoreUpdatesQueueReader =
-      kvStoreUpdatesQueue.getReader("decision");
-  auto prefixMgrKvStoreUpdatesReader =
-      kvStoreUpdatesQueue.getReader("prefixManager");
 
   // LinkMonitor -> KvStore/Decision
   ReplicateQueue<PeerEvent> peerUpdatesQueue;
@@ -319,13 +315,16 @@ main(int argc, char** argv) {
           *kvStorePublicationsDispatcherQueue));
 
   // make Decision/Prefix Manager subscribers of Dispatcher
-  decisionKvStoreUpdatesQueueReader = dispatcher->getReader(
+  auto decisionKvStoreUpdatesQueueReader = dispatcher->getReader(
       {Constants::kAdjDbMarker.toString(),
        Constants::kPrefixDbMarker.toString()});
 
-  prefixMgrKvStoreUpdatesReader =
+  auto prefixMgrKvStoreUpdatesReader =
       dispatcher->getReader({Constants::kPrefixDbMarker.toString()});
 
+  auto linkMonitorKvStoreUpdatesReader =
+      dispatcher->getReader({"link-monitor"});
+
   watchdog->addQueue(
       *kvStorePublicationsDispatcherQueue, "kvStorePublicationsQueue");
 
@@ -388,6 +387,7 @@ main(int argc, char** argv) {
           peerUpdatesQueue,
           logSampleQueue,
           kvRequestQueue,
+          std::move(linkMonitorKvStoreUpdatesReader),
           std::move(linkMonitorNeighborUpdatesQueueReader),
           std::move(linkMonitorNetlinkEventsQueueReader)));
   watchdog->addQueue(interfaceUpdatesQueue, "interfaceUpdatesQueue");

openr/ctrl-server/tests/OpenrCtrlHandlerTest.cpp

@@ -118,6 +118,7 @@ class OpenrCtrlFixture : public ::testing::Test {
         peerUpdatesQueue_,
         logSampleQueue_,
         kvRequestQueue_,
+        dispatcher_->getReader({"link-monitor"}),
         neighborUpdatesQueue_.getReader(),
         nlSock_->getReader());
     linkMonitorThread_ = std::thread([&]() { linkMonitor->run(); });

openr/decision/Decision.cpp

@@ -149,6 +149,14 @@ Decision::Decision(
     unreceivedRouteTypes_.emplace(thrift::PrefixType::CONFIG);
   }
 
+  if (auto enableInitOptimization =
+          config_->getConfig().enable_init_optimization()) {
+    enableInitOptimization_ = *enableInitOptimization;
+  }
+  if (enableInitOptimization_) {
+    XLOG(INFO) << "[Initialization] Init Optimization enabled";
+  }
+
   // Schedule periodic timer for counter submission
   counterUpdateTimer_ = folly::AsyncTimeout::make(*getEvb(), [this]() noexcept {
     updateGlobalCounters();
@@ -212,24 +220,28 @@ Decision::Decision(
 
               if (event == thrift::InitializationEvent::KVSTORE_SYNCED) {
                 // Received all initial KvStore publications.
-                XLOG(INFO) << "[Initialization] All initial publications are "
-                              "received from KvStore.";
-                initialKvStoreSynced_ = true;
-                triggerInitialBuildRoutes();
                 auto timeout = *config_->getConfig()
                                     .decision_config()
                                     ->unblock_initial_routes_ms();
-                XLOG(DBG1) << fmt::format(
+                XLOG(INFO) << fmt::format(
                     "Initial kv store synced. Waiting {}ms for initial routes to be computed.",
                     timeout);
                 unblockInitialRoutesTimeout_->scheduleTimeout(
                     std::chrono::milliseconds(timeout));
+                initialKvStoreSynced_ = true;
               } else {
                 // Received all locally originated adjacency keys
                 XLOG(INFO)
                     << "[Initialization] Received all locally originated adjacency keys";
                 initialSelfAdjSynced_ = true;
               }
+
+              if (initialKvStoreSynced_ && unblockSelfAdjSync()) {
+                if (!unblockInitialRoutes_) {
+                  XLOG(INFO) << "Trigger initial routes build";
+                  triggerInitialBuildRoutes();
+                }
+              }
             });
       } catch (const std::exception& e) {
 #ifndef NO_FOLLY_EXCEPTION_TRACER
@@ -959,6 +971,24 @@ Decision::rebuildRoutes(std::string const& event) {
   routeUpdatesQueue_.push(std::move(update));
 }
 
+/**
+ * @brief  Tells if wait for adjacency-db sync is required and if required
+ *         if adjacency-db sync signal is received
+ *
+ * @params None
+ *
+ * @return Returns true when both conditions are met
+ *         returns false otherwise
+ */
+bool
+Decision::unblockSelfAdjSync() {
+  if (!enableInitOptimization_) {
+    return true;
+  }
+
+  return initialSelfAdjSynced_;
+}
+
 bool
 Decision::unblockInitialRoutesBuild() {
   if (unblockInitialRoutes_) {
@@ -978,6 +1008,11 @@ Decision::unblockInitialRoutesBuild() {
                << "Have not yet received initial KvStore publication.";
     return false;
   }
+  if (!unblockSelfAdjSync()) {
+    XLOG(DBG1) << blockedLogPrefix
+               << "Have not yet received initial self adjacency publication.";
+    return false;
+  }
   if (!initialRibPolicyRead_) {
     XLOG(DBG1) << blockedLogPrefix << "Have not read persisted Rib policy.";
     return false;

openr/decision/Decision.h

@@ -257,6 +257,11 @@ class Decision : public OpenrEventBase {
    */
   void rebuildRoutes(std::string const& event);
 
+  /*
+   * Return true if unblocked for self adjacency signal
+   */
+  bool unblockSelfAdjSync();
+
   /*
    * Return true if all conditions of initial routes build are fulfilled.
    */
@@ -362,6 +367,7 @@ class Decision : public OpenrEventBase {
    * received in OpenR intialization procedure.
    */
   bool initialSelfAdjSynced_{false};
+  bool enableInitOptimization_{false};
 
   /**
    * Boolean flag indicating whether rib policy is read from persisted file in