Merge "[OVN] Set NB/SB "connection" inactivity probe" into stable/train

networking_ovn/common/utils.py

@@ -540,3 +540,22 @@ def get_chassis_availability_zones(chassis):
         azs = [az.strip() for az in values.split(':') if az.strip()]
         break
     return azs
+
+
+def connection_config_to_target_string(connection_config):
+    """Converts the Neutron NB/SB connection parameter to the OVN target string
+    :param connection_config: Neutron OVN config parameter for the OVN NB or SB
+                              database. See "ovn_sb_connection" or
+                              "ovn_nb_connection" params.
+    :returns: (String) OVN NB/SB ``connection.target`` column value.
+    """
+    regex = re.compile(r'^(?P<proto>\w+)\:((?P<ip>.+)\:(?P<port>\d+)|'
+                       r'(?P<file>[\w\/\.]+))')
+    m = regex.match(connection_config)
+    if m:
+        _dict = m.groupdict()
+        if _dict['ip'] and _dict['port']:
+            return ('p' + _dict['proto'] + ':' + _dict['port'] + ':' +
+                    _dict['ip'])
+        elif _dict['file']:
+            return 'p' + _dict['proto'] + ':' + _dict['file']

networking_ovn/ml2/mech_driver.py

@@ -58,6 +58,7 @@ from networking_ovn.ml2 import qos_driver
 from networking_ovn.ml2 import trunk_driver
 from networking_ovn import ovn_db_sync
 from networking_ovn.ovsdb import impl_idl_ovn
+from networking_ovn.ovsdb import ovsdb_monitor
 from networking_ovn.ovsdb import worker
 import neutron.wsgi
 
@@ -221,6 +222,28 @@ class OVNMechanismDriver(api.MechanismDriver):
         """Pre-initialize the ML2/OVN driver."""
         atexit.register(self._clean_hash_ring)
         signal.signal(signal.SIGTERM, self._clean_hash_ring)
+        self._set_inactivity_probe()
+
+    def _set_inactivity_probe(self):
+        """Set 'connection.inactivity_probe' in NB and SB databases"""
+        inactivity_probe = config.get_ovn_ovsdb_probe_interval()
+        dbs = [(config.get_ovn_nb_connection(), 'OVN_Northbound',
+                impl_idl_ovn.OvsdbNbOvnIdl),
+               (config.get_ovn_sb_connection(), 'OVN_Southbound',
+                impl_idl_ovn.OvsdbSbOvnIdl)]
+        for connection, schema, klass in dbs:
+            target = utils.connection_config_to_target_string(connection)
+            if not target:
+                continue
+            idl = ovsdb_monitor.BaseOvnIdl.from_server(connection, schema)
+            with ovsdb_monitor.short_living_ovsdb_api(klass, idl) as idl_api:
+                conn = idlutils.row_by_value(idl_api, 'Connection', 'target',
+                                             target, None)
+                if conn:
+                    idl_api.db_set(
+                        'Connection', target,
+                        ('inactivity_probe', int(inactivity_probe))).execute(
+                        check_error=True)
 
     @staticmethod
     def should_post_fork_initialize(worker_class):

networking_ovn/ovsdb/ovsdb_monitor.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 import abc
+import contextlib
 import datetime
 
 from neutron_lib.plugins import constants
@@ -537,6 +538,7 @@ class OvnSbIdl(OvnIdlDistributedLock):
         helper.register_table('Encap')
         helper.register_table('Port_Binding')
         helper.register_table('Datapath_Binding')
+        helper.register_table('Connection')
         return cls(driver, connection_string, helper)
 
     def post_connect(self):
@@ -572,3 +574,20 @@ def _check_and_set_ssl_files(schema_name):
 
     if ca_cert_file:
         Stream.ssl_set_ca_cert_file(ca_cert_file)
+
+
+@contextlib.contextmanager
+def short_living_ovsdb_api(api_class, idl):
+    """Context manager for short living connections to the database.
+
+    :param api_class: Class implementing the database calls
+                      (e.g. from the impl_idl module)
+    :param idl: An instance of IDL class (e.g. instance of OvnNbIdl)
+    """
+    conn = connection.Connection(
+        idl, timeout=ovn_config.get_ovn_ovsdb_timeout())
+    api = api_class(conn)
+    try:
+        yield api
+    finally:
+        api.ovsdb_connection.stop()

networking_ovn/tests/functional/test_mech_driver.py

@@ -16,11 +16,13 @@ import mock
 import re
 
 import netaddr
+from neutron.agent.linux import utils as linux_utils
 from neutron.tests import base as tests_base
 from neutron_lib.api.definitions import portbindings
 from neutron_lib import constants
 from oslo_config import cfg
 from oslo_utils import uuidutils
+from ovsdbapp.backend.ovs_idl import event
 
 from networking_ovn.common import config
 from networking_ovn.common import constants as ovn_const
@@ -776,3 +778,50 @@ class TestMetadataPorts(base.TestOVNFunctionalBase):
         self.assertIsNone(self._check_subnet_dhcp_options(subnet['id'],
                                                           '2001:db8::/64'))
         self._check_metadata_port(self.n1_id, [])
+
+
+class ConnectionInactivityProbeSetEvent(event.WaitEvent):
+    """Wait for a Connection (NB/SB) to have the inactivity probe set"""
+    ONETIME = False
+    def __init__(self, target, inactivity_probe):
+        table = 'Connection'
+        events = (self.ROW_UPDATE,)
+        super(ConnectionInactivityProbeSetEvent, self).__init__(events, table,
+                                                                None)
+        self.event_name = "ConnectionEvent"
+        self.target = target
+        self.inactivity_probe = inactivity_probe
+    def match_fn(self, event, row, old):
+        return row.target in self.target
+    def run(self, event, row, old):
+        if (row.inactivity_probe and
+                row.inactivity_probe[0] == self.inactivity_probe):
+            self.event.set()
+
+
+class TestSetInactivityProbe(base.TestOVNFunctionalBase):
+
+    def setUp(self, *args):
+        super(TestSetInactivityProbe, self).setUp(*args)
+        self.dbs = [(config.get_ovn_nb_connection(), 'ptcp:1000:1.2.3.4'),
+                    (config.get_ovn_sb_connection(), 'ptcp:1001:1.2.3.4')]
+        linux_utils.execute(
+            ['ovn-nbctl', '--db=%s' % self.dbs[0][0],
+             'set-connection', self.dbs[0][1]], run_as_root=True)
+        linux_utils.execute(
+            ['ovn-sbctl', '--db=%s' % self.dbs[1][0],
+             'set-connection', self.dbs[1][1]], run_as_root=True)
+
+    def test_1(self):
+        mock.patch.object(config, 'get_ovn_ovsdb_probe_interval',
+                          return_value='2500').start()
+        nb_connection = ConnectionInactivityProbeSetEvent(self.dbs[0][1], 2500)
+        sb_connection = ConnectionInactivityProbeSetEvent(self.dbs[1][1], 2500)
+        self.nb_api.idl.notify_handler.watch_event(nb_connection)
+        self.sb_api.idl.notify_handler.watch_event(sb_connection)
+        with mock.patch.object(utils, 'connection_config_to_target_string') \
+                as mock_target:
+            mock_target.side_effect = [self.dbs[0][1], self.dbs[1][1]]
+            self.mech_driver._set_inactivity_probe()
+        self.assertTrue(nb_connection.wait())
+        self.assertTrue(sb_connection.wait())

networking_ovn/tests/unit/common/test_utils.py

@@ -259,3 +259,17 @@ class TestDHCPUtils(base.TestCase):
                             'ntp_server': '10.0.2.1',
                             'bootfile_name': 'homer_simpson.bin'}
         self.assertEqual(expected_options, options)
+
+
+class TestConnectionConfigToTargetString(base.TestCase):
+    def test_strings(self):
+        config_target = (
+            ('ssl:1.2.3.4:5678', 'pssl:5678:1.2.3.4'),
+            ('tcp:1.2.3.4:5678', 'ptcp:5678:1.2.3.4'),
+            ('ssl:[::1]:5678', 'pssl:5678:[::1]'),
+            ('tcp:[::1]:5678', 'ptcp:5678:[::1]'),
+            ('unix:/var/run/ovs/db.sock', 'punix:/var/run/ovs/db.sock'),
+            ('wrong_value', None))
+        for config, target in config_target:
+            output = utils.connection_config_to_target_string(config)
+            self.assertEqual(target, output)