nova-manage db archive_deleted_rows is not multi-cell aware

The archive_deleted_rows cmd depend on DB connection config from config
file, and when applying super-conductor mode, there are several config
files for different cells. If so, the command can only archive rows in
cell0 DB as it only reads the nova.conf

This patch added code that provides --all-cells parameter to the
command and read info for all cells from the api_db and then archive
rows across all cells.

The --all-cells parameter is passed on to the purge command when
archive_deleted_rows is called with both --all-cells and --purge.

Co-Authored-By: melanie witt <[email protected]>

Change-Id: Id16c3d91d9ce5db9ffd125b59fffbfedf4a6843d
Closes-Bug: #1719487

Author: Kevin_Zheng

doc/source/cli/nova-manage.rst

@@ -62,7 +62,7 @@ Nova Database
     Returns exit code 0 if the database schema was synced successfully, or 1 if
     cell0 cannot be accessed.
 
-``nova-manage db archive_deleted_rows [--max_rows <number>] [--verbose] [--until-complete] [--before <date>] [--purge]``
+``nova-manage db archive_deleted_rows [--max_rows <number>] [--verbose] [--until-complete] [--before <date>] [--purge] [--all-cells]``
     Move deleted rows from production tables to shadow tables. Note that the
     corresponding rows in the ``instance_mappings``, ``request_specs`` and
     ``instance_group_member`` tables of the API database are purged when
@@ -78,7 +78,8 @@ Nova Database
     tables related to those instances. Specifying ``--purge`` will cause a
     *full* DB purge to be completed after archival. If a date range is desired
     for the purge, then run ``nova-manage db purge --before <date>`` manually
-    after archiving is complete.
+    after archiving is complete. Specifying ``--all-cells`` will
+    cause the process to run against all cell databases.
 
     **Return Codes**
 

nova/cmd/manage.py

@@ -414,17 +414,19 @@ def __init__(self):
         pass
 
     @staticmethod
-    def _print_dict(dct, dict_property="Property", dict_value='Value'):
+    def _print_dict(dct, dict_property="Property", dict_value='Value',
+                    sort_key=None):
         """Print a `dict` as a table of two columns.
 
         :param dct: `dict` to print
         :param dict_property: name of the first column
         :param wrap: wrapping for the second column
         :param dict_value: header label for the value (second) column
+        :param sort_key: key used for sorting the dict
         """
         pt = prettytable.PrettyTable([dict_property, dict_value])
         pt.align = 'l'
-        for k, v in sorted(dct.items()):
+        for k, v in sorted(dct.items(), key=sort_key):
             # convert dict to str to check length
             if isinstance(v, dict):
                 v = six.text_type(v)
@@ -495,9 +497,11 @@ def version(self):
                 'max_rows as a batch size for each iteration.'))
     @args('--purge', action='store_true', dest='purge', default=False,
           help='Purge all data from shadow tables after archive completes')
+    @args('--all-cells', action='store_true', dest='all_cells',
+          default=False, help='Run command across all cells.')
     def archive_deleted_rows(self, max_rows=1000, verbose=False,
                              until_complete=False, purge=False,
-                             before=None):
+                             before=None, all_cells=False):
         """Move deleted rows from production tables to shadow tables.
 
         Returns 0 if nothing was archived, 1 if some number of rows were
@@ -520,7 +524,7 @@ def archive_deleted_rows(self, max_rows=1000, verbose=False,
             # NOTE(tssurya): This check has been added to validate if the API
             # DB is reachable or not as this is essential for purging the
             # related API database records of the deleted instances.
-            objects.CellMappingList.get_all(ctxt)
+            cell_mappings = objects.CellMappingList.get_all(ctxt)
         except db_exc.CantStartEngineError:
             print(_('Failed to connect to API DB so aborting this archival '
                     'attempt. Please check your config file to make sure that '
@@ -538,59 +542,146 @@ def archive_deleted_rows(self, max_rows=1000, verbose=False,
             before_date = None
 
         table_to_rows_archived = {}
-        deleted_instance_uuids = []
         if until_complete and verbose:
             sys.stdout.write(_('Archiving') + '..')  # noqa
-        while True:
-            try:
-                run, deleted_instance_uuids = db.archive_deleted_rows(
-                                                max_rows, before=before_date)
-            except KeyboardInterrupt:
-                run = {}
-                if until_complete and verbose:
-                    print('.' + _('stopped'))  # noqa
-                    break
-            for k, v in run.items():
-                table_to_rows_archived.setdefault(k, 0)
-                table_to_rows_archived[k] += v
-            if deleted_instance_uuids:
-                table_to_rows_archived.setdefault('instance_mappings', 0)
-                table_to_rows_archived.setdefault('request_specs', 0)
-                table_to_rows_archived.setdefault('instance_group_member', 0)
-                deleted_mappings = objects.InstanceMappingList.destroy_bulk(
-                                            ctxt, deleted_instance_uuids)
-                table_to_rows_archived['instance_mappings'] += deleted_mappings
-                deleted_specs = objects.RequestSpec.destroy_bulk(
-                                            ctxt, deleted_instance_uuids)
-                table_to_rows_archived['request_specs'] += deleted_specs
-                deleted_group_members = (
-                    objects.InstanceGroup.destroy_members_bulk(
-                        ctxt, deleted_instance_uuids))
-                table_to_rows_archived['instance_group_member'] += (
-                    deleted_group_members)
-            if not until_complete:
-                break
-            elif not run:
-                if verbose:
-                    print('.' + _('complete'))  # noqa
+
+        interrupt = False
+
+        if all_cells:
+            # Sort first by cell name, then by table:
+            # +--------------------------------+-------------------------+
+            # | Table                          | Number of Rows Archived |
+            # +--------------------------------+-------------------------+
+            # | cell0.block_device_mapping     | 1                       |
+            # | cell1.block_device_mapping     | 1                       |
+            # | cell1.instance_actions         | 2                       |
+            # | cell1.instance_actions_events  | 2                       |
+            # | cell2.block_device_mapping     | 1                       |
+            # | cell2.instance_actions         | 2                       |
+            # | cell2.instance_actions_events  | 2                       |
+            # ...
+            def sort_func(item):
+                cell_name, table = item[0].split('.')
+                return cell_name, table
+            print_sort_func = sort_func
+        else:
+            cell_mappings = [None]
+            print_sort_func = None
+        total_rows_archived = 0
+        for cell_mapping in cell_mappings:
+            # NOTE(Kevin_Zheng): No need to calculate limit for each
+            # cell if until_complete=True.
+            # We need not adjust max rows to avoid exceeding a specified total
+            # limit because with until_complete=True, we have no total limit.
+            if until_complete:
+                max_rows_to_archive = max_rows
+            elif max_rows > total_rows_archived:
+                # We reduce the max rows to archive based on what we've
+                # archived so far to avoid potentially exceeding the specified
+                # total limit.
+                max_rows_to_archive = max_rows - total_rows_archived
+            else:
                 break
-            if verbose:
-                sys.stdout.write('.')
+            # If all_cells=False, cell_mapping is None
+            with context.target_cell(ctxt, cell_mapping) as cctxt:
+                cell_name = cell_mapping.name if cell_mapping else None
+                try:
+                    rows_archived = self._do_archive(
+                        table_to_rows_archived,
+                        cctxt,
+                        max_rows_to_archive,
+                        until_complete,
+                        verbose,
+                        before_date,
+                        cell_name)
+                except KeyboardInterrupt:
+                    interrupt = True
+                    break
+                # TODO(melwitt): Handle skip/warn for unreachable cells. Note
+                # that cell_mappings = [None] if not --all-cells
+                total_rows_archived += rows_archived
+
+        if until_complete and verbose:
+            if interrupt:
+                print('.' + _('stopped'))  # noqa
+            else:
+                print('.' + _('complete'))  # noqa
+
         if verbose:
             if table_to_rows_archived:
                 self._print_dict(table_to_rows_archived, _('Table'),
-                                 dict_value=_('Number of Rows Archived'))
+                                 dict_value=_('Number of Rows Archived'),
+                                 sort_key=print_sort_func)
             else:
                 print(_('Nothing was archived.'))
 
         if table_to_rows_archived and purge:
             if verbose:
                 print(_('Rows were archived, running purge...'))
-            self.purge(purge_all=True, verbose=verbose)
+            self.purge(purge_all=True, verbose=verbose, all_cells=all_cells)
 
         # NOTE(danms): Return nonzero if we archived something
         return int(bool(table_to_rows_archived))
 
+    def _do_archive(self, table_to_rows_archived, cctxt, max_rows,
+                    until_complete, verbose, before_date, cell_name):
+        """Helper function for archiving deleted rows for a cell.
+
+        This will archive deleted rows for a cell database and remove the
+        associated API database records for deleted instances.
+
+        :param table_to_rows_archived: Dict tracking the number of rows
+            archived by <cell_name>.<table name>. Example:
+            {'cell0.instances': 2,
+             'cell1.instances': 5}
+        :param cctxt: Cell-targeted nova.context.RequestContext if archiving
+            across all cells
+        :param max_rows: Maximum number of deleted rows to archive
+        :param until_complete: Whether to run continuously until all deleted
+            rows are archived
+        :param verbose: Whether to print how many rows were archived per table
+        :param before_date: Archive rows that were deleted before this date
+        :param cell_name: Name of the cell or None if not archiving across all
+            cells
+        """
+        ctxt = context.get_admin_context()
+        while True:
+            run, deleted_instance_uuids, total_rows_archived = \
+                db.archive_deleted_rows(cctxt, max_rows, before=before_date)
+            for table_name, rows_archived in run.items():
+                if cell_name:
+                    table_name = cell_name + '.' + table_name
+                table_to_rows_archived.setdefault(table_name, 0)
+                table_to_rows_archived[table_name] += rows_archived
+            if deleted_instance_uuids:
+                table_to_rows_archived.setdefault(
+                    'API_DB.instance_mappings', 0)
+                table_to_rows_archived.setdefault(
+                    'API_DB.request_specs', 0)
+                table_to_rows_archived.setdefault(
+                    'API_DB.instance_group_member', 0)
+                deleted_mappings = objects.InstanceMappingList.destroy_bulk(
+                            ctxt, deleted_instance_uuids)
+                table_to_rows_archived[
+                    'API_DB.instance_mappings'] += deleted_mappings
+                deleted_specs = objects.RequestSpec.destroy_bulk(
+                    ctxt, deleted_instance_uuids)
+                table_to_rows_archived[
+                    'API_DB.request_specs'] += deleted_specs
+                deleted_group_members = (
+                    objects.InstanceGroup.destroy_members_bulk(
+                        ctxt, deleted_instance_uuids))
+                table_to_rows_archived[
+                    'API_DB.instance_group_member'] += deleted_group_members
+            # If we're not archiving until there is nothing more to archive, we
+            # have reached max_rows in this cell DB or there was nothing to
+            # archive.
+            if not until_complete or not run:
+                break
+            if verbose:
+                sys.stdout.write('.')
+        return total_rows_archived
+
     @args('--before', metavar='<before>', dest='before',
           help='If specified, purge rows from shadow tables that are older '
                'than this. Accepts date strings in the default format output '

nova/db/api.py

@@ -1772,14 +1772,15 @@ def task_log_get(context, task_name, period_beginning,
 ####################
 
 
-def archive_deleted_rows(max_rows=None, before=None):
+def archive_deleted_rows(context=None, max_rows=None, before=None):
     """Move up to max_rows rows from production tables to the corresponding
     shadow tables.
 
+    :param context: nova.context.RequestContext for database access
     :param max_rows: Maximum number of rows to archive (required)
     :param before: optional datetime which when specified filters the records
         to only archive those records deleted before the given date
-    :returns: 2-item tuple:
+    :returns: 3-item tuple:
 
         - dict that maps table name to number of rows archived from that table,
           for example::
@@ -1790,8 +1791,10 @@ def archive_deleted_rows(max_rows=None, before=None):
                 'pci_devices': 2,
             }
         - list of UUIDs of instances that were archived
+        - total number of rows that were archived
     """
-    return IMPL.archive_deleted_rows(max_rows=max_rows, before=before)
+    return IMPL.archive_deleted_rows(context=context, max_rows=max_rows,
+                                     before=before)
 
 
 def pcidevice_online_data_migration(context, max_count):

nova/db/sqlalchemy/api.py

@@ -5569,14 +5569,15 @@ def _archive_deleted_rows_for_table(metadata, tablename, max_rows, before):
     return rows_archived, deleted_instance_uuids
 
 
-def archive_deleted_rows(max_rows=None, before=None):
+def archive_deleted_rows(context=None, max_rows=None, before=None):
     """Move up to max_rows rows from production tables to the corresponding
     shadow tables.
 
+    :param context: nova.context.RequestContext for database access
     :param max_rows: Maximum number of rows to archive (required)
     :param before: optional datetime which when specified filters the records
         to only archive those records deleted before the given date
-    :returns: 2-item tuple:
+    :returns: 3-item tuple:
 
         - dict that maps table name to number of rows archived from that table,
           for example::
@@ -5587,11 +5588,12 @@ def archive_deleted_rows(max_rows=None, before=None):
                 'pci_devices': 2,
             }
         - list of UUIDs of instances that were archived
+        - total number of rows that were archived
     """
     table_to_rows_archived = {}
     deleted_instance_uuids = []
     total_rows_archived = 0
-    meta = MetaData(get_engine(use_slave=True))
+    meta = MetaData(get_engine(use_slave=True, context=context))
     meta.reflect()
     # Reverse sort the tables so we get the leaf nodes first for processing.
     for table in reversed(meta.sorted_tables):
@@ -5615,7 +5617,7 @@ def archive_deleted_rows(max_rows=None, before=None):
             table_to_rows_archived[tablename] = rows_archived
         if total_rows_archived >= max_rows:
             break
-    return table_to_rows_archived, deleted_instance_uuids
+    return table_to_rows_archived, deleted_instance_uuids, total_rows_archived
 
 
 def _purgeable_tables(metadata):

nova/tests/functional/db/test_archive.py

@@ -94,7 +94,8 @@ def test_archive_deleted_rows(self):
         self.assertTrue(len(instance.system_metadata),
                         'No system_metadata for instance: %s' % server_id)
         # Now try and archive the soft deleted records.
-        results, deleted_instance_uuids = db.archive_deleted_rows(max_rows=100)
+        results, deleted_instance_uuids, archived = \
+            db.archive_deleted_rows(max_rows=100)
         # verify system_metadata was dropped
         self.assertIn('instance_system_metadata', results)
         self.assertEqual(len(instance.system_metadata),
@@ -105,6 +106,7 @@ def test_archive_deleted_rows(self):
         # by the archive
         self.assertIn('instance_actions', results)
         self.assertIn('instance_actions_events', results)
+        self.assertEqual(sum(results.values()), archived)
 
     def test_archive_deleted_rows_with_undeleted_residue(self):
         # Boots a server, deletes it, and then tries to archive it.
@@ -136,7 +138,8 @@ def test_archive_deleted_rows_with_undeleted_residue(self):
         self.assertTrue(len(instance.system_metadata),
                         'No system_metadata for instance: %s' % server_id)
         # Now try and archive the soft deleted records.
-        results, deleted_instance_uuids = db.archive_deleted_rows(max_rows=100)
+        results, deleted_instance_uuids, archived = \
+            db.archive_deleted_rows(max_rows=100)
         # verify system_metadata was dropped
         self.assertIn('instance_system_metadata', results)
         self.assertEqual(len(instance.system_metadata),
@@ -147,6 +150,7 @@ def test_archive_deleted_rows_with_undeleted_residue(self):
         # by the archive
         self.assertIn('instance_actions', results)
         self.assertIn('instance_actions_events', results)
+        self.assertEqual(sum(results.values()), archived)
 
     def _get_table_counts(self):
         engine = sqlalchemy_api.get_engine()
@@ -165,7 +169,7 @@ def test_archive_then_purge_all(self):
         server = self._create_server()
         server_id = server['id']
         self._delete_server(server_id)
-        results, deleted_ids = db.archive_deleted_rows(max_rows=1000)
+        results, deleted_ids, archived = db.archive_deleted_rows(max_rows=1000)
         self.assertEqual([server_id], deleted_ids)
 
         lines = []
@@ -178,6 +182,7 @@ def status(msg):
                                                      None, status_fn=status)
         self.assertNotEqual(0, deleted)
         self.assertNotEqual(0, len(lines))
+        self.assertEqual(sum(results.values()), archived)
         for line in lines:
             self.assertIsNotNone(re.match(r'Deleted [1-9][0-9]* rows from .*',
                                           line))
@@ -190,8 +195,9 @@ def test_archive_then_purge_by_date(self):
         server = self._create_server()
         server_id = server['id']
         self._delete_server(server_id)
-        results, deleted_ids = db.archive_deleted_rows(max_rows=1000)
+        results, deleted_ids, archived = db.archive_deleted_rows(max_rows=1000)
         self.assertEqual([server_id], deleted_ids)
+        self.assertEqual(sum(results.values()), archived)
 
         pre_purge_results = self._get_table_counts()
 
@@ -224,9 +230,10 @@ def test_purge_with_real_date(self):
         server = self._create_server()
         server_id = server['id']
         self._delete_server(server_id)
-        results, deleted_ids = db.archive_deleted_rows(max_rows=1000)
+        results, deleted_ids, archived = db.archive_deleted_rows(max_rows=1000)
         self.assertEqual([server_id], deleted_ids)
         date = dateutil_parser.parse('oct 21 2015', fuzzy=True)
         admin_context = context.get_admin_context()
         deleted = sqlalchemy_api.purge_shadow_tables(admin_context, date)
         self.assertEqual(0, deleted)
+        self.assertEqual(sum(results.values()), archived)