openstack/oslo.upgradecheck
Code Issues Proposed changes
Files
c5d372945ba27603f13b5c2e9f1e0cfc4ca373c0
oslo.upgradecheck/oslo_upgradecheck/upgradecheck.py
T
Stephen Finucane a6293c406f mypy: Disable import-untyped error code 
This is nicer than a whole load of type: ignore comments.

Change-Id: I6cfc2907a3ce264aadd11c48a42fb5ea0d86c20b
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
2026-03-06 15:37:47 +00:00

267 lines
9.2 KiB
Python
Raw Blame History

# Copyright 2018 Red Hat Inc.
# Copyright 2016 IBM Corp.
#
# 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 collections.abc import Callable, Iterable
import json
import sys
import textwrap
import traceback
from typing import Any, TypedDict
import enum
from oslo_config import cfg
import prettytable
from oslo_upgradecheck._i18n import _
CONF: cfg.ConfigOpts | None = None
class Code(enum.IntEnum):
"""Status codes for the upgrade check command"""
# All upgrade readiness checks passed successfully and there is
# nothing to do.
SUCCESS = 0
# At least one check encountered an issue and requires further
# investigation. This is considered a warning but the upgrade may be OK.
WARNING = 1
# There was an upgrade status check failure that needs to be
# investigated. This should be considered something that stops an upgrade.
FAILURE = 2
UPGRADE_CHECK_MSG_MAP = {
Code.SUCCESS: _('Success'),
Code.WARNING: _('Warning'),
Code.FAILURE: _('Failure'),
}
class Result:
"""Class used for 'nova-status upgrade check' results.
The 'code' attribute is a Code enum.
The 'details' attribute is a translated message generally only used for
checks that result in a warning or failure code. The details should provide
information on what issue was discovered along with any remediation.
"""
def __init__(self, code: Code, details: str | None = None) -> None:
super().__init__()
self.code = code
self.details = details
class _OutputCheck(TypedDict):
check: str
result: Code
details: str | None
class _Output(TypedDict):
name: str
checks: list[_OutputCheck]
class UpgradeCommands:
"""Base class for upgrade checks
This class should be inherited by a class in each project that provides
the actual checks. Those checks should be added to the _upgrade_checks
class member so that they are run when the ``check`` method is called.
The subcommands here must not rely on the service object model since they
should be able to run on n-1 data. Any queries to the database should be
done through the sqlalchemy query language directly like the database
schema migrations.
"""
display_title = _('Upgrade Check Results')
_upgrade_checks: tuple[
tuple[
str,
Callable[..., Result]
| tuple[Callable[..., Result], dict[str, Any]],
],
...,
] = ()
def _get_details(self, upgrade_check_result: Result) -> str | None:
if upgrade_check_result.details is None:
return None
# wrap the text on the details to 60 characters
return '\n'.join(
textwrap.wrap(
upgrade_check_result.details, 60, subsequent_indent=' '
)
)
def check(self) -> Code:
"""Performs checks to see if the deployment is ready for upgrade.
These checks are expected to be run BEFORE services are restarted with
new code.
:returns: Code
"""
return_code = Code.SUCCESS
# This is a list if 2-item tuples for the check name and it's results.
check_results = []
for name, func in self._upgrade_checks:
if isinstance(func, tuple):
func_name, kwargs = func
result = func_name(self, **kwargs)
else:
result = func(self)
# store the result of the check for the summary table
check_results.append((name, result))
# we want to end up with the highest level code of all checks
if result.code > return_code:
return_code = result.code
# We're going to build a summary table that looks like:
# +----------------------------------------------------+
# | Upgrade Check Results |
# +----------------------------------------------------+
# | Check: Cells v2 |
# | Result: Success |
# | Details: None |
# +----------------------------------------------------+
# | Check: Placement API |
# | Result: Failure |
# | Details: There is no placement-api endpoint in the |
# | service catalog. |
# +----------------------------------------------------+
# Since registering opts can be overridden by consuming code, we can't
# assume that our locally defined option exists.
if (
CONF is not None
and hasattr(CONF, 'command')
and hasattr(CONF.command, 'json')
and CONF.command.json
):
# NOTE(bnemec): We use str on the translated string to
# force immediate translation if lazy translation is in use.
# See lp1801761 for details.
output: _Output = {'name': str(self.display_title), 'checks': []}
for name, result in check_results:
output['checks'].append(
{
'check': name,
'result': result.code,
'details': result.details,
}
)
print(json.dumps(output))
else:
# NOTE(bnemec): We use str on the translated string to
# force immediate translation if lazy translation is in use.
# See lp1801761 for details.
t = prettytable.PrettyTable(
[str(self.display_title)], hrules=prettytable.ALL
)
t.align = 'l'
for name, result in check_results:
cell = _(
'Check: %(name)s\nResult: %(result)s\nDetails: %(details)s'
) % {
'name': name,
'result': UPGRADE_CHECK_MSG_MAP[result.code],
'details': self._get_details(result),
}
t.add_row([cell])
print(t)
return return_code
def register_cli_options(
conf: cfg.ConfigOpts, upgrade_command: UpgradeCommands
) -> None:
"""Set up the command line options.
Adds a subcommand to support 'upgrade check' on the command line.
:param conf: An oslo.confg ConfigOpts instance on which to register the
upgrade check arguments.
:param upgrade_command: The UpgradeCommands instance.
"""
def add_parsers(subparsers: Any) -> None:
upgrade_action = subparsers.add_parser('upgrade')
upgrade_action.add_argument('check')
upgrade_action.set_defaults(action_fn=upgrade_command.check)
upgrade_action.add_argument(
'--json',
action='store_true',
help='Output the results in JSON format. Default is to print '
'results in human readable table format.',
)
opt = cfg.SubCommandOpt('command', handler=add_parsers)
conf.register_cli_opt(opt)
def run(conf: cfg.ConfigOpts) -> int:
"""Run the requested command.
:param conf: An oslo.confg ConfigOpts instance on which the upgrade
commands have been previously registered.
"""
try:
return conf.command.action_fn() # type: ignore
except Exception:
print(_('Error:\n%s') % traceback.format_exc())
# This is 255 so it's not confused with the upgrade check exit codes.
return 255
def main(
conf: cfg.ConfigOpts,
project: str,
upgrade_command: UpgradeCommands,
argv: list[str] = sys.argv[1:],
default_config_files: Iterable[str] | None = None,
) -> int:
"""Simple implementation of main for upgrade checks
This can be used in upgrade check commands to provide the minimum
necessary parameter handling and logic.
:param conf: An oslo.confg ConfigOpts instance on which to register the
upgrade check arguments.
:param project: The name of the project, to be used as an argument
to the oslo_config.ConfigOpts instance to find
configuration files.
:param upgrade_command: The UpgradeCommands instance.
:param argv: The command line arguments to parse. Defaults to sys.argv[1:].
:param default_config_files: The configuration files to load. For projects
that use non-standard default locations for
the configuration files, use this to override
the search behavior in oslo.config.
"""
global CONF
register_cli_options(conf, upgrade_command)
conf(args=argv, project=project, default_config_files=default_config_files)
CONF = conf
return run(conf)
View Git Blame Copy Permalink