# octavia.cmd.status

# Source code for octavia.cmd.status

    # Copyright (c) 2018 NEC, 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 oslo_config import cfg
    from oslo_upgradecheck import common_checks
    from oslo_upgradecheck import upgradecheck
    from stevedore import driver as stevedore_driver

    # Need to import to load config
    from octavia.common import config  # noqa: F401 pylint: disable=unused-import
    from octavia.common import constants
    from octavia.controller.worker.v2 import taskflow_jobboard_driver as tsk_driver
    from octavia.i18n import _

    CONF = cfg.CONF

    class Checks(upgradecheck.UpgradeCommands):

        """Contains upgrade checks

        Various upgrade checks should be added as separate methods in this class
        and added to _upgrade_checks tuple.
        """

        def _check_persistence(self):
            try:
                pers_driver = tsk_driver.MysqlPersistenceDriver()
                with pers_driver.get_persistence() as pers:
                    if pers.engine.dialect.name == 'sqlite':
                        return upgradecheck.Result(
                            upgradecheck.Code.WARNING,
                            _('Persistence database is using sqlite backend. '
                              'Verification required if persistence_connecton URL '
                              'has been set properly.'))
                    return pers
            except Exception:
                return upgradecheck.Result(upgradecheck.Code.FAILURE,
                                           _('Failed to connect to persistence '
                                             'backend for AmphoraV2 provider.'))

        def _check_jobboard(self, persistence):
            try:
                jobboard_driver = stevedore_driver.DriverManager(
                    namespace='octavia.worker.jobboard_driver',
                    name=CONF.task_flow.jobboard_backend_driver,
                    invoke_args=(persistence,),
                    invoke_on_load=True).driver
                with jobboard_driver.job_board(persistence) as jb:
                    if jb.connected:
                        return upgradecheck.Result(
                            upgradecheck.Code.SUCCESS,
                            _('Persistence database and Jobboard backend for '
                              'AmphoraV2 provider configured.'))
            except Exception:
                # Return FAILURE later
                pass

            return upgradecheck.Result(
                upgradecheck.Code.FAILURE,
                _('Failed to connect to jobboard backend for AmphoraV2 provider. '
                  'Check jobboard configuration options in task_flow config '
                  'section.'))

        def _check_amphorav2(self):
            enabled_provider_drivers = CONF.api_settings.enabled_provider_drivers
            if (constants.AMPHORA in enabled_provider_drivers or
                    constants.AMPHORAV2 in enabled_provider_drivers):
                persistence = self._check_persistence()
                if isinstance(persistence, upgradecheck.Result):
                    return persistence
                return self._check_jobboard(persistence)
            return upgradecheck.Result(upgradecheck.Code.SUCCESS,
                                       _('AmphoraV2 provider is not enabled.'))

        _upgrade_checks = (
            (_('AmphoraV2 Check'), _check_amphorav2),
            (_('Policy File JSON to YAML Migration'),
             (common_checks.check_policy_json, {'conf': CONF})),
        )

    def main():
        return upgradecheck.main(
            CONF, project='octavia', upgrade_command=Checks())

this page last updated: 2026-04-13 15:26:39

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug