[OVN] Add revision number maintenance methods

The OVNRevisionNumbers maintenance methods are used to check
inconsistences between the Neutron and the OVN databases. Two
methods are implemented:
- get_inconsistent_resources: to retrieve those registers with
  different revision number
- get_deleted_resources: to retrieve the list of OVNRevisionNumbers
  orphan registers (the resource register is deleted).

Previous paths in networking-ovn tree:
./networking_ovn/db/maintenance -> ./neutron/db/ovn_revision_numbers_db

Co-Authored-By: Lucas Alvares Gomes <lucasagomes@gmail.com>

Change-Id: I948af3d3b93731dbf0773615759b49bf6b4a8f5f
Partially-Implements: blueprint neutron-ovn-merge

neutron/db/ovn_revision_numbers_db.py

@@ -13,10 +13,14 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import datetime
+
 from neutron_lib.db import api as db_api
 from neutron_lib import exceptions as n_exc
 from oslo_config import cfg
 from oslo_log import log
+from oslo_utils import timeutils
+import sqlalchemy as sa
 from sqlalchemy.orm import exc
 
 from neutron.db.models import l3  # noqa
@@ -39,11 +43,33 @@ TYPE_ROUTER_PORTS = 'router_ports'
 TYPE_SECURITY_GROUPS = 'security_groups'
 TYPE_FLOATINGIPS = 'floatingips'
 TYPE_SUBNETS = 'subnets'
-TYPES_OVN = (TYPE_NETWORKS, TYPE_PORTS, TYPE_SECURITY_GROUP_RULES,
-             TYPE_ROUTERS, TYPE_ROUTER_PORTS, TYPE_SECURITY_GROUPS,
-             TYPE_FLOATINGIPS, TYPE_SUBNETS)
+
+_TYPES_PRIORITY_ORDER = (
+    TYPE_NETWORKS,
+    TYPE_SECURITY_GROUPS,
+    TYPE_SUBNETS,
+    TYPE_ROUTERS,
+    TYPE_PORTS,
+    TYPE_ROUTER_PORTS,
+    TYPE_FLOATINGIPS,
+    TYPE_SECURITY_GROUP_RULES)
+
+# The order in which the resources should be created or updated by the
+# maintenance task: Root ones first and leafs at the end.
+MAINTENANCE_CREATE_UPDATE_TYPE_ORDER = {
+    t: n for n, t in enumerate(_TYPES_PRIORITY_ORDER, 1)}
+
+# The order in which the resources should be deleted by the maintenance
+# task: Leaf ones first and roots at the end.
+MAINTENANCE_DELETE_TYPE_ORDER = {
+    t: n for n, t in enumerate(reversed(_TYPES_PRIORITY_ORDER), 1)}
+
 INITIAL_REV_NUM = -1
 
+# Time (in seconds) used to identify if an entry is new before considering
+# it an inconsistency
+INCONSISTENCIES_OLDER_THAN = 60
+
 
 # 1:2 mapping for OVN, neutron router ports are simple ports, but
 # for OVN we handle LSP & LRP objects
@@ -63,7 +89,7 @@ class UnknownResourceType(n_exc.NeutronException):
 
 def get_revision_number(resource, resource_type):
     """Get the resource's revision number based on its type."""
-    if resource_type in TYPES_OVN:
+    if resource_type in _TYPES_PRIORITY_ORDER:
         return resource['revision_number']
     raise UnknownResourceType(resource_type=resource_type)
 
@@ -164,3 +190,47 @@ def bump_revision(context, resource, resource_type):
              '%(res_uuid)s (type: %(res_type)s) to %(rev_num)d',
              {'res_uuid': resource['id'], 'res_type': resource_type,
               'rev_num': revision_number})
+
+
+def get_inconsistent_resources(context):
+    """Get a list of inconsistent resources.
+
+    :returns: A list of objects which the revision number from the
+              ovn_revision_number and standardattributes tables differs.
+    """
+    sort_order = sa.case(value=ovn_models.OVNRevisionNumbers.resource_type,
+                         whens=MAINTENANCE_CREATE_UPDATE_TYPE_ORDER)
+    time_ = (timeutils.utcnow() -
+             datetime.timedelta(seconds=INCONSISTENCIES_OLDER_THAN))
+    with db_api.CONTEXT_READER.using(context):
+        query = context.session.query(ovn_models.OVNRevisionNumbers).join(
+            standard_attr.StandardAttribute,
+            ovn_models.OVNRevisionNumbers.standard_attr_id ==
+            standard_attr.StandardAttribute.id)
+        # Filter out new entries
+        query = query.filter(
+            standard_attr.StandardAttribute.created_at < time_)
+        # Filter for resources which revision_number differs
+        query = query.filter(
+            ovn_models.OVNRevisionNumbers.revision_number !=
+            standard_attr.StandardAttribute.revision_number)
+        return query.order_by(sort_order).all()
+
+
+def get_deleted_resources(context):
+    """Get a list of resources that failed to be deleted in OVN.
+
+    Get a list of resources that have been deleted from neutron but not
+    in OVN. Once a resource is deleted in Neutron the ``standard_attr_id``
+    foreign key in the ovn_revision_numbers table will be set to NULL.
+
+    Upon successfully deleting the resource in OVN the entry in the
+    ovn_revision_number should also be deleted but if something fails
+    the entry will be kept and returned in this list so the maintenance
+    thread can later fix it.
+    """
+    sort_order = sa.case(value=ovn_models.OVNRevisionNumbers.resource_type,
+                         whens=MAINTENANCE_DELETE_TYPE_ORDER)
+    with db_api.CONTEXT_READER.using(context):
+        return context.session.query(ovn_models.OVNRevisionNumbers).filter_by(
+            standard_attr_id=None).order_by(sort_order).all()

neutron/tests/unit/db/test_ovn_revision_numbers_db.py

@@ -15,13 +15,25 @@
 
 import mock
 
+from neutron_lib import constants as n_const
 from neutron_lib import context
 from neutron_lib.db import api as db_api
 from oslo_db import exception as db_exc
 
+from neutron.api import extensions
+from neutron.common import config
 from neutron.db.models import ovn as ovn_models
 from neutron.db import ovn_revision_numbers_db as ovn_rn_db
+import neutron.extensions
+from neutron.services.revisions import revision_plugin
 from neutron.tests.unit.db import test_db_base_plugin_v2
+from neutron.tests.unit.extensions import test_l3
+from neutron.tests.unit.extensions import test_securitygroup
+
+
+EXTENSIONS_PATH = ':'.join(neutron.extensions.__path__)
+PLUGIN_CLASS = (
+    'neutron.tests.unit.db.test_ovn_revision_numbers_db.TestMaintenancePlugin')
 
 
 class TestRevisionNumber(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
@@ -93,3 +105,143 @@ class TestRevisionNumber(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
         except db_exc.DBDuplicateEntry:
             self.fail("create_initial_revision shouldn't raise "
                       "DBDuplicateEntry when may_exist is True")
+
+
+class TestMaintenancePlugin(test_securitygroup.SecurityGroupTestPlugin,
+                            test_l3.TestL3NatBasePlugin):
+
+    __native_pagination_support = True
+    __native_sorting_support = True
+
+    supported_extension_aliases = ['external-net', 'security-group']
+
+
+class TestRevisionNumberMaintenance(test_securitygroup.SecurityGroupsTestCase,
+                                    test_l3.L3NatTestCaseMixin):
+
+    def setUp(self):
+        service_plugins = {
+            'router':
+            'neutron.tests.unit.extensions.test_l3.TestL3NatServicePlugin'}
+        l3_plugin = test_l3.TestL3NatServicePlugin()
+        sec_plugin = test_securitygroup.SecurityGroupTestPlugin()
+        ext_mgr = extensions.PluginAwareExtensionManager(
+            EXTENSIONS_PATH, {'router': l3_plugin, 'sec': sec_plugin}
+        )
+        super(TestRevisionNumberMaintenance, self).setUp(
+            plugin=PLUGIN_CLASS, service_plugins=service_plugins)
+        app = config.load_paste_app('extensions_test_app')
+        self.ext_api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
+        self.session = db_api.get_writer_session()
+        revision_plugin.RevisionPlugin()
+        self.net = self._make_network(self.fmt, 'net1', True)['network']
+
+        # Mock the default value for INCONSISTENCIES_OLDER_THAN so
+        # tests won't need to wait for the timeout in order to validate
+        # the database inconsistencies
+        self.older_than_mock = mock.patch(
+            'neutron.db.ovn_revision_numbers_db.INCONSISTENCIES_OLDER_THAN',
+            -1)
+        self.older_than_mock.start()
+        self.addCleanup(self.older_than_mock.stop)
+        self.ctx = context.get_admin_context()
+
+    def _create_initial_revision(self, resource_uuid, resource_type,
+                                 revision_number=ovn_rn_db.INITIAL_REV_NUM,
+                                 may_exist=False):
+        with self.ctx.session.begin(subtransactions=True):
+            ovn_rn_db.create_initial_revision(
+                self.ctx, resource_uuid, resource_type,
+                revision_number=revision_number, may_exist=may_exist)
+
+    def test_get_inconsistent_resources(self):
+        # Set the intial revision to -1 to force it to be incosistent
+        self._create_initial_revision(
+            self.net['id'], ovn_rn_db.TYPE_NETWORKS, revision_number=-1)
+        res = ovn_rn_db.get_inconsistent_resources(self.ctx)
+        self.assertEqual(1, len(res))
+        self.assertEqual(self.net['id'], res[0].resource_uuid)
+
+    def test_get_inconsistent_resources_older_than(self):
+        # Stop the mock so the INCONSISTENCIES_OLDER_THAN will have
+        # it's default value
+        self.older_than_mock.stop()
+        self._create_initial_revision(
+            self.net['id'], ovn_rn_db.TYPE_NETWORKS, revision_number=-1)
+        res = ovn_rn_db.get_inconsistent_resources(self.ctx)
+
+        # Assert that nothing is returned because the entry is not old
+        # enough to be picked as an inconsistency
+        self.assertEqual(0, len(res))
+
+        # Start the mock again and make sure it nows shows up as an
+        # inconsistency
+        self.older_than_mock.start()
+        res = ovn_rn_db.get_inconsistent_resources(self.ctx)
+        self.assertEqual(1, len(res))
+        self.assertEqual(self.net['id'], res[0].resource_uuid)
+
+    def test_get_inconsistent_resources_consistent(self):
+        # Set the initial revision to 0 which is the initial revision_number
+        # for recently created resources
+        self._create_initial_revision(
+            self.net['id'], ovn_rn_db.TYPE_NETWORKS, revision_number=0)
+        res = ovn_rn_db.get_inconsistent_resources(self.ctx)
+        # Assert nothing is inconsistent
+        self.assertEqual([], res)
+
+    def test_get_deleted_resources(self):
+        self._create_initial_revision(
+            self.net['id'], ovn_rn_db.TYPE_NETWORKS, revision_number=0)
+        self._delete('networks', self.net['id'])
+        res = ovn_rn_db.get_deleted_resources(self.ctx)
+        self.assertEqual(1, len(res))
+        self.assertEqual(self.net['id'], res[0].resource_uuid)
+        self.assertIsNone(res[0].standard_attr_id)
+
+    def _prepare_resources_for_ordering_test(self, delete=False):
+        subnet = self._make_subnet(self.fmt, {'network': self.net}, '10.0.0.1',
+                                   '10.0.0.0/24')['subnet']
+        self._set_net_external(self.net['id'])
+        info = {'network_id': self.net['id']}
+        router = self._make_router(self.fmt, None,
+                                   external_gateway_info=info)['router']
+        fip = self._make_floatingip(self.fmt, self.net['id'])['floatingip']
+        port = self._make_port(self.fmt, self.net['id'])['port']
+        sg = self._make_security_group(self.fmt, 'sg1', '')['security_group']
+        rule = self._build_security_group_rule(
+            sg['id'], 'ingress', n_const.PROTO_NUM_TCP)
+        sg_rule = self._make_security_group_rule(
+            self.fmt, rule)['security_group_rule']
+
+        self._create_initial_revision(router['id'], ovn_rn_db.TYPE_ROUTERS)
+        self._create_initial_revision(subnet['id'], ovn_rn_db.TYPE_SUBNETS)
+        self._create_initial_revision(fip['id'], ovn_rn_db.TYPE_FLOATINGIPS)
+        self._create_initial_revision(port['id'], ovn_rn_db.TYPE_PORTS)
+        self._create_initial_revision(port['id'], ovn_rn_db.TYPE_ROUTER_PORTS)
+        self._create_initial_revision(sg['id'], ovn_rn_db.TYPE_SECURITY_GROUPS)
+        self._create_initial_revision(sg_rule['id'],
+                                      ovn_rn_db.TYPE_SECURITY_GROUP_RULES)
+        self._create_initial_revision(self.net['id'], ovn_rn_db.TYPE_NETWORKS)
+
+        if delete:
+            self._delete('security-group-rules', sg_rule['id'])
+            self._delete('floatingips', fip['id'])
+            self._delete('ports', port['id'])
+            self._delete('security-groups', sg['id'])
+            self._delete('routers', router['id'])
+            self._delete('subnets', subnet['id'])
+            self._delete('networks', self.net['id'])
+
+    def test_get_inconsistent_resources_order(self):
+        self._prepare_resources_for_ordering_test()
+        res = ovn_rn_db.get_inconsistent_resources(self.ctx)
+        actual_order = tuple(r.resource_type for r in res)
+        self.assertEqual(ovn_rn_db._TYPES_PRIORITY_ORDER, actual_order)
+
+    def test_get_deleted_resources_order(self):
+        self._prepare_resources_for_ordering_test(delete=True)
+        res = ovn_rn_db.get_deleted_resources(self.ctx)
+        actual_order = tuple(r.resource_type for r in res)
+        self.assertEqual(tuple(reversed(ovn_rn_db._TYPES_PRIORITY_ORDER)),
+                         actual_order)