Merge "Split Hash Ring probing from the maintenance task" into stable/ussuri

1 files changed, 17 insertions, 3 deletions

diff --git a/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py b/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py
index e766f3d56f..9ea84198e7 100644
--- a/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py
+++ b/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py
@@ -16,6 +16,7 @@ import atexit
 import copy
 import datetime
 import functools
+import multiprocessing
 import operator
 import signal
 import threading
@@ -33,6 +34,7 @@ from neutron_lib import exceptions as n_exc
 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
@@ -111,6 +113,8 @@ class OVNMechanismDriver(api.MechanismDriver):
         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()
@@ -302,6 +306,17 @@ class OVNMechanismDriver(api.MechanismDriver):
                                 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)
@@ -316,6 +331,7 @@ class OVNMechanismDriver(api.MechanismDriver):
             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)
@@ -359,10 +375,8 @@ class OVNMechanismDriver(api.MechanismDriver):
             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 _wait_for_pg_drop_event(self):
         """Wait for event that occurs when neutron_pg_drop Port Group exists.