用法

请参阅模块 oslo_upgradecheck.__main__,了解如何使用此项目的一个示例。

每个使用该项目的应该创建一个继承自 oslo_upgradecheck.upgradecheck.UpgradeCommands 的类,并在其上实现检查方法。然后,这些检查方法应该被添加到 _upgrade_checks 元组中,以便在调用 oslo_upgradecheck.upgradecheck.UpgradeCommands.check() 方法时运行它们。例如

from oslo_upgradecheck import upgradecheck

class ProjectSpecificUpgradeCommands(upgradecheck.UpgradeCommands):
    def an_upgrade_check(self):
        if everything_is_awesome():
            return upgradecheck.Result(
                upgradecheck.Code.SUCCESS, 'Success details')
        else:
            return upgradecheck.Result(
                upgradecheck.Code.FAILURE, 'Failure details')

    _upgrade_checks = (('Awesome upgrade check', an_upgrade_check))

oslo.upgradecheck 还包含命令行参数处理的基本实现,可用于提供实现 $SERVICE-status upgrade check 命令所需的最低处理。要使用它,请编写一个创建上述类实例的方法,然后将该类的 check 函数传递给 oslo_upgradecheck.upgradecheck.main()。项目 ConfigOpts 实例也必须传递。在大多数项目中,这将只是 cfg.CONF。例如

from oslo_config import cfg

def main():
    return upgradecheck.main(
        conf=cfg.CONF,
        project='myprojectname',
        upgrade_command=ProjectSpecificUpgradeCommands(),
    )

然后,$SERVICE-status 命令的入口点应指向此函数。

或者,如果项目有自己的 CLI 代码,并且希望重用它,则只需确保在将 upgrade check 参数传递给 $SERVICE-status 命令时调用 inst.check 方法即可。

示例

以下是实现检查命令的完整功能示例

# Copyright 2018 Red Hat Inc.
#
# 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
#
#      https://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.

"""Example CLI command for running upgrade checks"""

import sys

from oslo_config import cfg
from oslo_upgradecheck import upgradecheck


class Checks(upgradecheck.UpgradeCommands):
    def success(self):
        return upgradecheck.Result(upgradecheck.Code.SUCCESS,
                                   'Always succeeds')

    def failure(self):
        return upgradecheck.Result(upgradecheck.Code.FAILURE, 'Always fails')

    _upgrade_checks = (('always succeeds', success),
                       ('always fails', failure),
                       )


def main():
    return upgradecheck.main(
        conf=cfg.CONF,
        project='myprojectname',
        upgrade_command=Checks(),
    )


if __name__ == '__main__':
    sys.exit(main())