[OVN] Set NB/SB "connection" inactivity probe

For each NB/SB databases, set the inactivity probe timeout value on
the Neutron configured connection register.

Neutron connects to the NB/SB databases with the value defined in
"ovn_nb_connection"/"ovn_sb_connection" config parameters, with the
format:
- tcp:IP:PORT
- ssl:IP:PORT
- unix:socket_file

The connection register "target" column has the following format:
- ptcp:PORT:IP
- pssl:PORT:IP
- punix:socket_file

Conflicts:
    neutron/common/ovn/utils.py
    neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovsdb_monitor.py
    neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

Closes-Bug: #1935842
Change-Id: If82ff2c92b8bec18ae3e895ce19a65cfeed9c90d
(cherry picked from commit 28f3017a90)
(cherry picked from commit f8aefd600b)
(cherry picked from commit f6eda4f965)

neutron/common/ovn/utils.py

@@ -548,3 +548,23 @@ def get_chassis_availability_zones(chassis):
 
 def get_network_name_from_datapath(datapath):
     return datapath.external_ids['name'].replace('neutron-', '')
+
+
+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']

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

@@ -224,6 +224,7 @@ class OVNMechanismDriver(api.MechanismDriver):
         atexit.register(self._clean_hash_ring)
         signal.signal(signal.SIGTERM, self._clean_hash_ring)
         self._create_neutron_pg_drop()
+        self._set_inactivity_probe()
 
     def _create_neutron_pg_drop(self):
         """Create neutron_pg_drop Port Group.
@@ -260,6 +261,28 @@ class OVNMechanismDriver(api.MechanismDriver):
                 else:
                     raise
 
+    def _set_inactivity_probe(self):
+        """Set 'connection.inactivity_probe' in NB and SB databases"""
+        inactivity_probe = ovn_conf.get_ovn_ovsdb_probe_interval()
+        dbs = [(ovn_conf.get_ovn_nb_connection(), 'OVN_Northbound',
+                impl_idl_ovn.OvsdbNbOvnIdl),
+               (ovn_conf.get_ovn_sb_connection(), 'OVN_Southbound',
+                impl_idl_ovn.OvsdbSbOvnIdl)]
+        for connection, schema, klass in dbs:
+            target = ovn_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):
         return worker_class in (neutron.wsgi.WorkerService,

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

@@ -34,6 +34,7 @@ from neutron._i18n import _
 from neutron.common.ovn import constants as ovn_const
 from neutron.common.ovn import exceptions as ovn_exc
 from neutron.common.ovn import utils
+from neutron.common import utils as n_utils
 from neutron.conf.plugins.ml2.drivers.ovn import ovn_conf as cfg
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import commands as cmd
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovsdb_monitor
@@ -138,6 +139,10 @@ class Backend(ovs_idl.Backend):
 
     _tables = tables
 
+    @n_utils.classproperty
+    def connection_string(cls):
+        raise NotImplementedError()
+
     def is_table_present(self, table_name):
         return table_name in self._tables
 
@@ -198,6 +203,10 @@ class OvsdbNbOvnIdl(nb_impl_idl.OvnNbApiIdlImpl, Backend):
     def __init__(self, connection):
         super(OvsdbNbOvnIdl, self).__init__(connection)
 
+    @n_utils.classproperty
+    def connection_string(cls):
+        return cfg.get_ovn_nb_connection()
+
     @classmethod
     def from_worker(cls, worker_class, driver=None):
         args = (cfg.get_ovn_nb_connection(), 'OVN_Northbound')
@@ -768,6 +777,10 @@ class OvsdbSbOvnIdl(sb_impl_idl.OvnSbApiIdlImpl, Backend):
     def __init__(self, connection):
         super(OvsdbSbOvnIdl, self).__init__(connection)
 
+    @n_utils.classproperty
+    def connection_string(cls):
+        return cfg.get_ovn_sb_connection()
+
     @classmethod
     def from_worker(cls, worker_class, driver=None):
         args = (cfg.get_ovn_sb_connection(), 'OVN_Southbound')

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

@@ -544,6 +544,7 @@ class OvnSbIdl(OvnIdlDistributedLock):
         helper.register_table('Port_Binding')
         helper.register_table('Datapath_Binding')
         helper.register_table('MAC_Binding')
+        helper.register_table('Connection')
         return cls(driver, connection_string, helper)
 
     def post_connect(self):

neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

@@ -19,11 +19,14 @@ 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 ovsdbapp.tests.functional import base as ovs_base
 
+from neutron.agent.linux import utils as linux_utils
 from neutron.common.ovn import constants as ovn_const
 from neutron.common.ovn import utils
 from neutron.common import utils as n_utils
+from neutron.conf.plugins.ml2.drivers.ovn import ovn_conf
 from neutron.db import ovn_revision_numbers_db as db_rev
 from neutron.plugins.ml2.drivers.ovn.mech_driver import mech_driver
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import impl_idl_ovn
@@ -762,3 +765,54 @@ class TestCreateDefaultDropPortGroup(ovs_base.FunctionalTestCase,
 
     def test_without_ports(self):
         self._test_pg_with_ports(expected_ports=[])
+
+
+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().__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):
+        super().setUp()
+        self.dbs = [(ovn_conf.get_ovn_nb_connection(), 'ptcp:1000:1.2.3.4'),
+                    (ovn_conf.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(ovn_conf, '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())

neutron/tests/unit/common/ovn/test_utils.py

@@ -265,3 +265,19 @@ class TestDHCPUtils(base.BaseTestCase):
                             'ntp_server': '10.0.2.1',
                             'bootfile_name': 'homer_simpson.bin'}
         self.assertEqual(expected_options, options)
+
+
+class TestConnectionConfigToTargetString(base.BaseTestCase):
+
+    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)