Add a method to retrieve and register "rpc_workers" config knob

This new method retrieves the config option "rpc_workers" from the
configuration. If this option is not loaded, the method registers
the ``neutron.conf.service.SERVICE_OPTS`` options before reading
the knob again.

Closes-Bug: #2004656
Related-Bug: #1889737

Change-Id: I1f99cb32f33cc91141136cb4e3fbd33715530c59

neutron/cmd/upgrade_checks/checks.py

@@ -27,6 +27,7 @@ from sqlalchemy import or_
 
 from neutron._i18n import _
 from neutron.cmd.upgrade_checks import base
+from neutron.conf import service as conf_service
 from neutron.db.extra_dhcp_opt import models as extra_dhcp_opt_models
 from neutron.db.models import agent as agent_model
 from neutron.db.models import external_net
@@ -183,7 +184,7 @@ class CoreChecks(base.BaseChecks):
     @staticmethod
     def worker_count_check(checker):
 
-        if cfg.CONF.api_workers and cfg.CONF.rpc_workers:
+        if cfg.CONF.api_workers and conf_service.get_rpc_workers():
             return upgradecheck.Result(
                 upgradecheck.Code.SUCCESS, _("Number of workers already "
                                              "defined in config"))

neutron/common/utils.py

@@ -60,6 +60,8 @@ from sqlalchemy.sql.expression import func as sql_func
 import neutron
 from neutron._i18n import _
 from neutron.api import api_common
+from neutron.conf import service as conf_service
+
 
 TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
 LOG = logging.getLogger(__name__)
@@ -1041,7 +1043,8 @@ def disable_notifications(function):
 
     @functools.wraps(function)
     def wrapper(*args, **kwargs):
-        if cfg.CONF.rpc_workers is None or cfg.CONF.rpc_workers >= 1:
+        rpc_workers = conf_service.get_rpc_workers()
+        if rpc_workers is None or rpc_workers >= 1:
             return function(*args, **kwargs)
     return wrapper
 

neutron/conf/service.py

@@ -53,3 +53,12 @@ RPC_EXTRA_OPTS = [
 
 def register_service_opts(opts, conf=cfg.CONF):
     conf.register_opts(opts)
+
+
+def get_rpc_workers(conf=cfg.CONF):
+    """Retrieve the conf knob rpc_workers, register option first if needed"""
+    try:
+        return conf.rpc_workers
+    except cfg.NoSuchOptError:
+        register_service_opts(SERVICE_OPTS, conf=conf)
+        return conf.rpc_workers

neutron/plugins/ml2/plugin.py

@@ -109,6 +109,7 @@ from neutron.api.rpc.handlers import metadata_rpc
 from neutron.api.rpc.handlers import resources_rpc
 from neutron.api.rpc.handlers import securitygroups_rpc
 from neutron.common import utils
+from neutron.conf import service as conf_service
 from neutron.db import address_group_db
 from neutron.db import address_scope_db
 from neutron.db import agents_db
@@ -401,7 +402,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
     def _start_rpc_notifiers(self):
         """Initialize RPC notifiers for agents."""
         self.ovo_notifier = None
-        if cfg.CONF.rpc_workers is None or cfg.CONF.rpc_workers >= 1:
+        rpc_workers = conf_service.get_rpc_workers()
+        if rpc_workers is None or rpc_workers >= 1:
             self.ovo_notifier = ovo_rpc.OVOServerRpcInterface()
         self.notifier = rpc.AgentNotifierApi(topics.AGENT)
         if cfg.CONF.enable_traditional_dhcp:

neutron/tests/unit/conf/test_service.py

@@ -0,0 +1,33 @@
+# Copyright (c) 2023 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 oslo_config import cfg
+
+from neutron.conf import service
+from neutron.tests import base as tests_base
+
+
+class GetRpcWorkers(tests_base.BaseTestCase):
+
+    def test_no_previous_registration(self):
+        self.assertIsNone(service.get_rpc_workers(conf=cfg.CONF))
+        cfg.CONF.set_override('rpc_workers', 150)
+        self.assertEqual(150, service.get_rpc_workers(conf=cfg.CONF))
+
+    def test_previous_registration(self):
+        service.register_service_opts(service.SERVICE_OPTS, conf=cfg.CONF)
+        self.assertIsNone(service.get_rpc_workers(conf=cfg.CONF))
+        cfg.CONF.set_override('rpc_workers', 200)
+        self.assertEqual(200, service.get_rpc_workers(conf=cfg.CONF))