Split Hash Ring probing from the maintenance task

This patch split out the Hash Ring probing out of the maitenance task
into it's own thread. The idea is to speed up the start of probing by
doing it right after adding a node to the Hash Ring.

By doing that, we avoid the problem of delaying probing in case the
connection with OVSDB takes longer than expected to connect and the hash
ring nodes are considered dead as they weren't probed in time.

The patch re-uses the same classes as before to start this new thread
(instead of reusing the maintenance task thread). It adds a layer of
synchronization with a lock to make sure that only one new Hash Ring
probing thread is started.

(cherry picked from commit 240f2c6)

Closes-Bug: #1991655
Change-Id: Ic04493f20eb9aecda563942c51f343dc4202523a
Signed-off-by: Lucas Alvares Gomes <[email protected]>

Author: umago

neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py

@@ -16,6 +16,7 @@
 import copy
 import datetime
 import functools
+import multiprocessing
 import operator
 import signal
 import threading
@@ -34,6 +35,7 @@
 from neutron_lib.plugins import directory
 from neutron_lib.plugins.ml2 import api
 from neutron_lib.utils import helpers
+from oslo_concurrency import lockutils
 from oslo_concurrency import processutils
 from oslo_config import cfg
 from oslo_db import exception as os_db_exc
@@ -110,6 +112,8 @@ def initialize(self):
         self._plugin_property = None
         self._ovn_client_inst = None
         self._maintenance_thread = None
+        self._hash_ring_thread = None
+        self._hash_ring_probe_event = multiprocessing.Event()
         self.node_uuid = None
         self.hash_ring_group = ovn_const.HASH_RING_ML2_GROUP
         self.sg_enabled = ovn_acl.is_sg_enabled()
@@ -326,6 +330,17 @@ def should_post_fork_initialize(worker_class):
                                 worker.MaintenanceWorker,
                                 service.RpcWorker)
 
+    @lockutils.synchronized('hash_ring_probe_lock', external=True)
+    def _start_hash_ring_probe(self):
+        if not self._hash_ring_probe_event.is_set():
+            self._hash_ring_thread = maintenance.MaintenanceThread()
+            self._hash_ring_thread.add_periodics(
+                maintenance.HashRingHealthCheckPeriodics(
+                    self.hash_ring_group))
+            self._hash_ring_thread.start()
+            LOG.info("Hash Ring probing thread has started")
+            self._hash_ring_probe_event.set()
+
     def post_fork_initialize(self, resource, event, trigger, payload=None):
         # Initialize API/Maintenance workers with OVN IDL connections
         worker_class = ovn_utils.get_method_class(trigger)
@@ -339,6 +354,7 @@ def post_fork_initialize(self, resource, event, trigger, payload=None):
             admin_context = n_context.get_admin_context()
             self.node_uuid = ovn_hash_ring_db.add_node(admin_context,
                                                        self.hash_ring_group)
+            self._start_hash_ring_probe()
 
         n_agent.AgentCache(self)  # Initialize singleton agent cache
         self.nb_ovn, self.sb_ovn = impl_idl_ovn.get_ovn_idls(self, trigger)
@@ -383,10 +399,8 @@ def post_fork_initialize(self, resource, event, trigger, payload=None):
             self._maintenance_thread = maintenance.MaintenanceThread()
             self._maintenance_thread.add_periodics(
                 maintenance.DBInconsistenciesPeriodics(self._ovn_client))
-            self._maintenance_thread.add_periodics(
-                maintenance.HashRingHealthCheckPeriodics(
-                    self.hash_ring_group))
             self._maintenance_thread.start()
+            LOG.info("Maintenance task thread has started")
 
     def _create_security_group_precommit(self, resource, event, trigger,
                                          payload):