# octavia.controller.healthmanager.health_manager

# Source code for octavia.controller.healthmanager.health_manager

    # Copyright 2015 Hewlett-Packard Development Company, L.P.
    #
    # 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 concurrent import futures
    import functools
    import time

    from oslo_config import cfg
    from oslo_db import exception as db_exc
    from oslo_log import log as logging
    from oslo_utils import excutils

    from octavia.common import constants
    from octavia.controller.worker.v2 import controller_worker as cw2
    from octavia.db import api as db_api
    from octavia.db import repositories as repo

    CONF = cfg.CONF
    LOG = logging.getLogger(__name__)

    def wait_done_or_dead(futs, dead, check_timeout=1):
        while True:
            _done, not_done = futures.wait(futs, timeout=check_timeout)
            if not not_done:
                break
            if dead.is_set():
                for fut in not_done:
                    # This may not actually be able to cancel, but try to
                    # if we can.
                    fut.cancel()

    def update_stats_on_done(stats, fut):
        # This utilizes the fact that python, non-primitive types are
        # passed by reference (not by value)...
        stats['failover_attempted'] += 1
        try:
            fut.result()
        except futures.CancelledError:
            stats['failover_cancelled'] += 1
        except Exception:
            stats['failover_failed'] += 1

    class HealthManager:
        def __init__(self, exit_event):
            self.cw = cw2.ControllerWorker()
            self.threads = CONF.health_manager.failover_threads
            self.executor = futures.ThreadPoolExecutor(max_workers=self.threads)
            self.amp_repo = repo.AmphoraRepository()
            self.amp_health_repo = repo.AmphoraHealthRepository()
            self.lb_repo = repo.LoadBalancerRepository()
            self.dead = exit_event

        def _test_and_set_failover_prov_status(self, lock_session, lb_id):
            if self.lb_repo.set_status_for_failover(lock_session, lb_id,
                                                    constants.PENDING_UPDATE):
                return True
            db_lb = self.lb_repo.get(lock_session, id=lb_id)
            prov_status = db_lb.provisioning_status
            LOG.warning("Load balancer %(id)s is in immutable state "
                        "%(state)s. Skipping failover.",
                        {"state": prov_status, "id": db_lb.id})
            return False

        def health_check(self):
            """Check for stale amphorae and process them

            ... until either no more stale amphora were found or all executor
            threads are busy.
            """
            stats = {
                'failover_attempted': 0,
                'failover_failed': 0,
                'failover_cancelled': 0,
            }
            futs = []
            while not self.dead.is_set():
                amp_health = None
                lock_session = None
                try:
                    lock_session = db_api.get_session()
                    lock_session.begin()
                    amp_health = self.amp_health_repo.get_stale_amphora(
                        lock_session)
                    if amp_health:
                        amp = self.amp_repo.get(lock_session,
                                                id=amp_health.amphora_id)
                        # If there is an associated LB, attempt to set it to
                        # PENDING_UPDATE. If it is already immutable, skip the
                        # amphora on this cycle
                        if amp and amp.load_balancer_id:
                            if not self._test_and_set_failover_prov_status(
                                    lock_session, amp.load_balancer_id):
                                lock_session.rollback()
                                break
                    lock_session.commit()
                except db_exc.DBDeadlock:
                    LOG.debug('Database reports deadlock. Skipping.')
                    lock_session.rollback()
                    amp_health = None
                except db_exc.RetryRequest:
                    LOG.debug('Database is requesting a retry. Skipping.')
                    lock_session.rollback()
                    amp_health = None
                except db_exc.DBConnectionError:
                    db_api.wait_for_connection(self.dead)
                    lock_session.rollback()
                    amp_health = None
                    if not self.dead.is_set():
                        # amphora heartbeat timestamps should also be outdated
                        # while DB is unavailable and soon after DB comes back
                        # online. Sleeping off the full "heartbeat_timeout"
                        # interval to give the amps a chance to check in before
                        # we start failovers.
                        time.sleep(CONF.health_manager.heartbeat_timeout)
                except Exception:
                    with excutils.save_and_reraise_exception():
                        if lock_session:
                            lock_session.rollback()

                # No more stale amps found
                if amp_health is None:
                    break

                LOG.info("Stale amphora's id is: %s", amp_health.amphora_id)
                fut = self.executor.submit(
                    self.cw.failover_amphora, amp_health.amphora_id, reraise=True)
                fut.add_done_callback(
                    functools.partial(update_stats_on_done, stats)
                )
                futs.append(fut)
                # All threads are/were busy
                if len(futs) == self.threads:
                    break
            if futs:
                LOG.info("Waiting for %s failovers to finish",
                         len(futs))
                wait_done_or_dead(futs, self.dead)
            if stats['failover_attempted'] > 0:
                LOG.info("Attempted %s failovers of amphora",
                         stats['failover_attempted'])
                LOG.info("Failed at %s failovers of amphora",
                         stats['failover_failed'])
                LOG.info("Cancelled %s failovers of amphora",
                         stats['failover_cancelled'])
                happy_failovers = stats['failover_attempted']
                happy_failovers -= stats['failover_cancelled']
                happy_failovers -= stats['failover_failed']
                LOG.info("Successfully completed %s failovers of amphora",
                         happy_failovers)

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