Fix unbounded thread pool in OVN MaintenanceThread

Change-Id Ie01fd776 (commit d98d028) switched
MaintenanceThread's PeriodicWorker from the default
SynchronousExecutor to futurist.ThreadPoolExecutor, passing the
class as a bare reference without specifying max_workers.

When max_workers is unset, futurist.ThreadPoolExecutor defaults to
os.cpu_count() * 5 (480 on a 96-CPU node). The executor's
_maybe_spin_up() creates a new idle worker thread on every submit()
call, and futurist.ThreadPoolExecutor never shrinks its pool.
Threads accumulate indefinitely until RuntimeError: can't start new
thread.

Since PeriodicWorker runs periodic tasks sequentially (one at a
time), only a single worker thread is needed. Bound max_workers=1
to prevent thread accumulation.

Closes-Bug: #2149830
Related-Bug: #2097257
Generated-By: claude-opus-4-6 (OpenCode)
Signed-off-by: Adam Harwell <flux.adam@gmail.com>
Change-Id: Id0b71902a7dc842f2190b310a222f49f39db91ba
(cherry picked from commit 153db8a)

Author: rm-you

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

@@ -104,7 +104,8 @@ def start(self):
         if self._thread is None:
             self._worker = periodics.PeriodicWorker(
                 self._callables,
-                executor_factory=futurist.ThreadPoolExecutor)
+                executor_factory=lambda: futurist.ThreadPoolExecutor(
+                    max_workers=1))
             self._thread = threading.Thread(target=self._worker.start)
             self._thread.daemon = True
             self._thread.start()

neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_maintenance.py

@@ -13,6 +13,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import threading
+import time
 from unittest import mock
 
 from futurist import periodics
@@ -35,6 +37,49 @@
 from neutron.tests.unit import testlib_api
 
 
+class TestMaintenanceThread(base.BaseTestCase):
+
+    def test_start_bounds_thread_pool_executor(self):
+        """Verify MaintenanceThread runs periodic tasks with one worker."""
+        num_tasks = 5
+        run_counter = [0] * num_tasks
+        active_threads = []
+        target_runs = 5
+
+        class _PeriodicTask:
+            def __init__(self, idx):
+                self.idx = idx
+
+            @periodics.periodic(spacing=0.05)
+            def trivial_task(self):
+                run_counter[self.idx] += 1
+                active_threads.append(threading.active_count())
+                time.sleep(0)
+                if run_counter[self.idx] >= target_runs:
+                    raise periodics.NeverAgain
+
+        mt = maintenance.MaintenanceThread()
+        for idx in range(num_tasks):
+            mt.add_periodics(_PeriodicTask(idx))
+
+        threads_before = threading.active_count()
+        mt.start()
+        threads_after_start = threading.active_count()
+        self.assertGreater(threads_after_start, threads_before)
+
+        executor = mt._worker._executor_factory()
+        # Let some tasks be executed.
+        time.sleep(2)
+        threads_after_wait = threading.active_count()
+        self.assertEqual(threads_after_wait, threads_after_start + 1)
+
+        executor.shutdown(wait=False)
+        mt._thread.join(timeout=3)
+        threads_end = threading.active_count()
+        self.assertEqual(threads_end, threads_after_wait)
+        self.assertTrue(all(target_runs == _c for _c in run_counter))
+
+
 class TestHasLockPeriodicDecorator(base.BaseTestCase):
 
     def test_decorator_no_limit_have_lock(self):