Initialize the network segment ranges only in first WSGI worker

The implementation done in [1] does not fully work across WSGi workers.
The method ``NetworkSegmentRange.new_default`` tries to first check if
the default segment range for a specific driver (VLAN, tunnelled) is
present. However, as seen in production environments, this method is
not multiprocess safe.

Instead, this patch is limiting the execution of the network segment
ranges initialization to the first WSGI worker (there must be at least
one worker).

This patch also wraps the VLAN and tunnelled drivers initialization
inside a database transaction context. All the operations executed in
this method (register clean-up, new default registers creation and
ranges sync) are done in one single database transaction, that ensures
its isolation and integrity.

NOTE: The same initialization method, when called, removes the
duplicated registers created by [1] in first place. A Neutron API update
and restart will fix the database ``network_segment_ranges`` registers.

NOTE: The ranges class variable (``_TunnelTypeDriverBase.tunnel_ranges``
or ``VlanTypeDriver.network_vlan_ranges``) stores initially the
configured segment ranges (static configuration file). If the network
segment range plugin is loaded, it will store the segment ranges from
the database. But this variable should not be public; instead of this,
the method ``get_network_segment_ranges`` provides the needed class API
to retrieve this information.

[1]https://review.opendev.org/c/openstack/neutron/+/938319

Closes-Bug: #2106463
Change-Id: Ibc42f900214e1f7631e266bccd083a2ef4111585
(cherry picked from commit 39d95a1)

Author: ralonsoh

neutron/common/wsgi_utils.py

@@ -20,6 +20,9 @@
 from neutron.common import utils
 
 
+FIRST_WORKER_ID = 1
+
+
 def get_start_time(default=None, current_time=False):
     """Return the 'start-time=%t' config varible in the WSGI config
 

neutron/plugins/ml2/drivers/helpers.py

@@ -15,7 +15,6 @@
 
 import functools
 
-from neutron_lib import context
 from neutron_lib.db import api as db_api
 from neutron_lib import exceptions
 from neutron_lib.plugins import constants as plugin_constants
@@ -165,8 +164,7 @@ def allocate_partially_specified_segment(self, context, **filters):
                     context.elevated()):
                 LOG.debug('    - %s', srange)
 
-    @db_api.retry_db_errors
-    def _delete_expired_default_network_segment_ranges(self, start_time):
-        ns_range.NetworkSegmentRange.\
-            delete_expired_default_network_segment_ranges(
-                context.get_admin_context(), self.get_type(), start_time)
+    def _delete_expired_default_network_segment_ranges(self, ctx, start_time):
+        (ns_range.NetworkSegmentRange.
+        delete_expired_default_network_segment_ranges(
+            ctx, self.get_type(), start_time))

neutron/plugins/ml2/drivers/type_tunnel.py

@@ -127,7 +127,7 @@ def _initialize(self, raw_tunnel_ranges):
             # allocation during driver initialization, instead of using the
             # directory.get_plugin() method - the normal way used elsewhere to
             # check if a plugin is loaded.
-            self.sync_allocations()
+            self._sync_allocations()
 
     def _parse_tunnel_ranges(self, tunnel_ranges, current_range):
         for entry in tunnel_ranges:
@@ -145,17 +145,15 @@ def _parse_tunnel_ranges(self, tunnel_ranges, current_range):
                  {'type': self.get_type(), 'range': current_range})
 
     @db_api.retry_db_errors
-    def _populate_new_default_network_segment_ranges(self, start_time):
-        ctx = context.get_admin_context()
-        with db_api.CONTEXT_WRITER.using(ctx):
-            for tun_min, tun_max in self.tunnel_ranges:
-                range_obj.NetworkSegmentRange.new_default(
-                    ctx, self.get_type(), None, tun_min, tun_max, start_time)
+    def _populate_new_default_network_segment_ranges(self, ctx, start_time):
+        for tun_min, tun_max in self.tunnel_ranges:
+            range_obj.NetworkSegmentRange.new_default(
+                ctx, self.get_type(), None, tun_min, tun_max, start_time)
 
     @db_api.retry_db_errors
-    def _get_network_segment_ranges_from_db(self):
+    def _get_network_segment_ranges_from_db(self, ctx=None):
         ranges = []
-        ctx = context.get_admin_context()
+        ctx = ctx or context.get_admin_context()
         with db_api.CONTEXT_READER.using(ctx):
             range_objs = (range_obj.NetworkSegmentRange.get_objects(
                 ctx, network_type=self.get_type()))
@@ -164,21 +162,27 @@ def _get_network_segment_ranges_from_db(self):
 
         return ranges
 
+    @db_api.retry_db_errors
     def initialize_network_segment_range_support(self, start_time):
-        self._delete_expired_default_network_segment_ranges(start_time)
-        self._populate_new_default_network_segment_ranges(start_time)
-        # Override self.tunnel_ranges with the network segment range
-        # information from DB and then do a sync_allocations since the
-        # segment range service plugin has not yet been loaded at this
-        # initialization time.
-        self.tunnel_ranges = self._get_network_segment_ranges_from_db()
-        self.sync_allocations()
+        admin_context = context.get_admin_context()
+        with db_api.CONTEXT_WRITER.using(admin_context):
+            self._delete_expired_default_network_segment_ranges(
+                admin_context, start_time)
+            self._populate_new_default_network_segment_ranges(
+                admin_context, start_time)
+            # Override self.tunnel_ranges with the network segment range
+            # information from DB and then do a sync_allocations since the
+            # segment range service plugin has not yet been loaded at this
+            # initialization time.
+            self.tunnel_ranges = self._get_network_segment_ranges_from_db(
+                ctx=admin_context)
+            self._sync_allocations(ctx=admin_context)
 
     def update_network_segment_range_allocations(self):
-        self.sync_allocations()
+        self._sync_allocations()
 
     @db_api.retry_db_errors
-    def sync_allocations(self):
+    def _sync_allocations(self, ctx=None):
         # determine current configured allocatable tunnel ids
         tunnel_ids = set()
         ranges = self.get_network_segment_ranges()
@@ -187,7 +191,7 @@ def sync_allocations(self):
 
         tunnel_id_getter = operator.attrgetter(self.segmentation_key)
         tunnel_col = getattr(self.model, self.segmentation_key)
-        ctx = context.get_admin_context()
+        ctx = ctx or context.get_admin_context()
         with db_api.CONTEXT_WRITER.using(ctx):
             # Check if the allocations are updated: if the total number of
             # allocations for this tunnel type matches the allocations of the

neutron/plugins/ml2/drivers/type_vlan.py

@@ -56,16 +56,13 @@ def __init__(self):
         self.model_segmentation_id = vlan_alloc_model.VlanAllocation.vlan_id
         self._parse_network_vlan_ranges()
 
-    @db_api.retry_db_errors
-    def _populate_new_default_network_segment_ranges(self, start_time):
-        ctx = context.get_admin_context()
-        with db_api.CONTEXT_WRITER.using(ctx):
-            for (physical_network, vlan_ranges) in (
-                    self.network_vlan_ranges.items()):
-                for vlan_min, vlan_max in vlan_ranges:
-                    range_obj.NetworkSegmentRange.new_default(
-                        ctx, self.get_type(), physical_network, vlan_min,
-                        vlan_max, start_time)
+    def _populate_new_default_network_segment_ranges(self, ctx, start_time):
+        for (physical_network, vlan_ranges) in (
+                self.network_vlan_ranges.items()):
+            for vlan_min, vlan_max in vlan_ranges:
+                range_obj.NetworkSegmentRange.new_default(
+                    ctx, self.get_type(), physical_network, vlan_min,
+                    vlan_max, start_time)
 
     def _parse_network_vlan_ranges(self):
         try:
@@ -78,8 +75,8 @@ def _parse_network_vlan_ranges(self):
         LOG.info("Network VLAN ranges: %s", self.network_vlan_ranges)
 
     @db_api.retry_db_errors
-    def _sync_vlan_allocations(self):
-        ctx = context.get_admin_context()
+    def _sync_vlan_allocations(self, ctx=None):
+        ctx = ctx or context.get_admin_context()
         with db_api.CONTEXT_WRITER.using(ctx):
             # VLAN ranges per physical network:
             #   {phy1: [(1, 10), (30, 50)], ...}
@@ -142,9 +139,9 @@ def _sync_vlan_allocations(self):
                                                      vlan_ids)
 
     @db_api.retry_db_errors
-    def _get_network_segment_ranges_from_db(self):
+    def _get_network_segment_ranges_from_db(self, ctx=None):
         ranges = {}
-        ctx = context.get_admin_context()
+        ctx = ctx or context.get_admin_context()
         with db_api.CONTEXT_READER.using(ctx):
             range_objs = (range_obj.NetworkSegmentRange.get_objects(
                 ctx, network_type=self.get_type()))
@@ -171,15 +168,21 @@ def initialize(self):
             self._sync_vlan_allocations()
         LOG.info("VlanTypeDriver initialization complete")
 
+    @db_api.retry_db_errors
     def initialize_network_segment_range_support(self, start_time):
-        self._delete_expired_default_network_segment_ranges(start_time)
-        self._populate_new_default_network_segment_ranges(start_time)
-        # Override self.network_vlan_ranges with the network segment range
-        # information from DB and then do a sync_allocations since the
-        # segment range service plugin has not yet been loaded at this
-        # initialization time.
-        self.network_vlan_ranges = self._get_network_segment_ranges_from_db()
-        self._sync_vlan_allocations()
+        admin_context = context.get_admin_context()
+        with db_api.CONTEXT_WRITER.using(admin_context):
+            self._delete_expired_default_network_segment_ranges(
+                admin_context, start_time)
+            self._populate_new_default_network_segment_ranges(
+                admin_context, start_time)
+            # Override self.network_vlan_ranges with the network segment range
+            # information from DB and then do a sync_allocations since the
+            # segment range service plugin has not yet been loaded at this
+            # initialization time.
+            self.network_vlan_ranges = (
+                self._get_network_segment_ranges_from_db(ctx=admin_context))
+            self._sync_vlan_allocations(ctx=admin_context)
 
     def update_network_segment_range_allocations(self):
         self._sync_vlan_allocations()

neutron/plugins/ml2/managers.py

@@ -32,6 +32,7 @@
 import stevedore
 
 from neutron._i18n import _
+from neutron.common import wsgi_utils
 from neutron.conf.plugins.ml2 import config
 from neutron.db import segments_db
 from neutron.objects import ports
@@ -205,6 +206,9 @@ def initialize(self):
             driver.obj.initialize()
 
     def initialize_network_segment_range_support(self, start_time):
+        if wsgi_utils.get_api_worker_id() != wsgi_utils.FIRST_WORKER_ID:
+            return
+
         for network_type, driver in self.drivers.items():
             if network_type in constants.NETWORK_SEGMENT_RANGE_TYPES:
                 LOG.info("Initializing driver network segment range support "

neutron/tests/functional/plugins/ml2/drivers/test_type_tunnel.py

@@ -0,0 +1,92 @@
+# Copyright 2025 Red Hat Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from concurrent import futures
+import time
+
+from neutron_lib import constants
+from neutron_lib import context
+from neutron_lib.db import api as db_api
+from oslo_config import cfg
+
+from neutron.conf import common as common_config
+from neutron.conf.plugins.ml2 import config as ml2_config
+from neutron.conf.plugins.ml2.drivers import driver_type as driver_type_config
+from neutron.objects import network_segment_range as range_obj
+from neutron.plugins.ml2.drivers import type_geneve
+from neutron.tests.unit import testlib_api
+
+
+def _initialize_network_segment_range_support(type_driver, start_time):
+    # This method is similar to
+    # ``_TunnelTypeDriverBase.initialize_network_segment_range_support``.
+    # The method first deletes the existing default network ranges and then
+    # creates the new ones. It also adds an extra second before closing the
+    # DB transaction.
+    admin_context = context.get_admin_context()
+    with db_api.CONTEXT_WRITER.using(admin_context):
+        type_driver._delete_expired_default_network_segment_ranges(
+            admin_context, start_time)
+        type_driver._populate_new_default_network_segment_ranges(
+            admin_context, start_time)
+        time.sleep(1)
+
+
+class TunnelTypeDriverBaseTestCase(testlib_api.SqlTestCase):
+    def setUp(self):
+        super().setUp()
+        cfg.CONF.register_opts(common_config.core_opts)
+        ml2_config.register_ml2_plugin_opts()
+        driver_type_config.register_ml2_drivers_geneve_opts()
+        ml2_config.cfg.CONF.set_override(
+            'service_plugins', 'network_segment_range')
+        self.min = 1001
+        self.max = 1020
+        self.net_type = constants.TYPE_GENEVE
+        ml2_config.cfg.CONF.set_override(
+            'vni_ranges', f'{self.min}:{self.max}', group='ml2_type_geneve')
+        self.admin_ctx = context.get_admin_context()
+        self.type_driver = type_geneve.GeneveTypeDriver()
+        self.type_driver.initialize()
+
+    def test_initialize_network_segment_range_support(self):
+        # Execute the initialization several times with different start times.
+        for start_time in range(3):
+            self.type_driver.initialize_network_segment_range_support(
+                start_time)
+            sranges = range_obj.NetworkSegmentRange.get_objects(self.admin_ctx)
+            self.assertEqual(1, len(sranges))
+            self.assertEqual(self.net_type, sranges[0].network_type)
+            self.assertEqual(self.min, sranges[0].minimum)
+            self.assertEqual(self.max, sranges[0].maximum)
+            self.assertEqual([(self.min, self.max)],
+                             self.type_driver.tunnel_ranges)
+
+    def test_initialize_network_segment_range_support_parallel_execution(self):
+        max_workers = 3
+        _futures = []
+        with futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+            for idx in range(max_workers):
+                _futures.append(executor.submit(
+                    _initialize_network_segment_range_support,
+                    self.type_driver, idx))
+            for _future in _futures:
+                _future.result()
+
+        sranges = range_obj.NetworkSegmentRange.get_objects(self.admin_ctx)
+        self.assertEqual(1, len(sranges))
+        self.assertEqual(self.net_type, sranges[0].network_type)
+        self.assertEqual(self.min, sranges[0].minimum)
+        self.assertEqual(self.max, sranges[0].maximum)

neutron/tests/unit/plugins/ml2/drivers/base_type_tunnel.py

@@ -49,7 +49,7 @@ def setUp(self):
         super().setUp()
         self.driver = self.DRIVER_CLASS()
         self.driver.tunnel_ranges = TUNNEL_RANGES
-        self.driver.sync_allocations()
+        self.driver._sync_allocations()
         self.context = context.Context()
 
     def test_tunnel_type(self):
@@ -84,7 +84,7 @@ def test_sync_tunnel_allocations(self):
             self.driver.get_allocation(self.context, (TUN_MAX + 1)))
 
         self.driver.tunnel_ranges = UPDATED_TUNNEL_RANGES
-        self.driver.sync_allocations()
+        self.driver._sync_allocations()
 
         self.assertIsNone(
             self.driver.get_allocation(self.context, (TUN_MIN + 5 - 1)))
@@ -108,7 +108,7 @@ def _test_sync_allocations_and_allocated(self, tunnel_id):
         self.driver.reserve_provider_segment(self.context, segment)
 
         self.driver.tunnel_ranges = UPDATED_TUNNEL_RANGES
-        self.driver.sync_allocations()
+        self.driver._sync_allocations()
 
         self.assertTrue(
             self.driver.get_allocation(self.context, tunnel_id).allocated)
@@ -127,7 +127,7 @@ def verify_no_chunk(iterable, chunk_size):
             return []
         with mock.patch.object(
                 type_tunnel, 'chunks', side_effect=verify_no_chunk) as chunks:
-            self.driver.sync_allocations()
+            self.driver._sync_allocations()
             # No writing operation is done, fast exit: current allocations
             # already present.
             self.assertEqual(0, len(chunks.mock_calls))
@@ -295,7 +295,7 @@ def setUp(self):
         super().setUp()
         self.driver = self.DRIVER_CLASS()
         self.driver.tunnel_ranges = self.TUNNEL_MULTI_RANGES
-        self.driver.sync_allocations()
+        self.driver._sync_allocations()
         self.context = context.Context()
 
     def test_release_segment(self):
@@ -486,7 +486,7 @@ def test__populate_new_default_network_segment_ranges(self):
         # one of the `service_plugins`
         self.driver._initialize(RAW_TUNNEL_RANGES)
         self.driver.initialize_network_segment_range_support(self.start_time)
-        self.driver.sync_allocations()
+        self.driver._sync_allocations()
         ret = obj_network_segment_range.NetworkSegmentRange.get_objects(
             self.context)
         self.assertEqual(1, len(ret))
@@ -502,9 +502,9 @@ def test__populate_new_default_network_segment_ranges(self):
 
     def test__delete_expired_default_network_segment_ranges(self):
         self.driver.tunnel_ranges = TUNNEL_RANGES
-        self.driver.sync_allocations()
+        self.driver._sync_allocations()
         self.driver._delete_expired_default_network_segment_ranges(
-            self.start_time)
+            self.context, self.start_time)
         ret = obj_network_segment_range.NetworkSegmentRange.get_objects(
             self.context, network_type=self.driver.get_type())
         self.assertEqual(0, len(ret))

neutron/tests/unit/plugins/ml2/drivers/test_type_vlan.py

@@ -25,11 +25,14 @@
 from oslo_config import cfg
 from testtools import matchers
 
+from neutron.common import wsgi_utils
+from neutron.conf.plugins.ml2 import config as ml2_config
 from neutron.objects import network_segment_range as obj_network_segment_range
 from neutron.objects.plugins.ml2 import vlanallocation as vlan_alloc_obj
 from neutron.plugins.ml2.drivers import type_vlan
 from neutron.tests.unit import testlib_api
 
+
 PROVIDER_NET = 'phys_net1'
 TENANT_NET = 'phys_net2'
 UNCONFIGURED_NET = 'no_net'
@@ -361,6 +364,9 @@ def test_allocate_tenant_segment_in_order_of_config(self):
 class VlanTypeTestWithNetworkSegmentRange(testlib_api.SqlTestCase):
 
     def setUp(self):
+        ml2_config.register_ml2_plugin_opts()
+        mock.patch.object(wsgi_utils, 'get_api_worker_id',
+                          return_value=wsgi_utils.FIRST_WORKER_ID).start()
         super().setUp()
         cfg.CONF.set_override('network_vlan_ranges',
                               NETWORK_VLAN_RANGES,
@@ -400,7 +406,7 @@ def test__populate_new_default_network_segment_ranges(self):
 
     def test__delete_expired_default_network_segment_ranges(self):
         self.driver._delete_expired_default_network_segment_ranges(
-            self.start_time)
+            self.context, self.start_time)
         ret = obj_network_segment_range.NetworkSegmentRange.get_objects(
             self.context, network_type=self.driver.get_type())
         self.assertEqual(0, len(ret))