# octavia.common.base_taskflow

# Source code for octavia.common.base_taskflow

    # Copyright 2014-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.
    #

    import concurrent.futures
    import datetime
    import functools
    import time

    from oslo_config import cfg
    from oslo_log import log
    from oslo_utils import uuidutils
    from taskflow.conductors.backends import impl_blocking
    from taskflow import engines
    from taskflow import exceptions as taskflow_exc
    from taskflow.jobs.base import Job
    from taskflow.listeners import base
    from taskflow.listeners import logging
    from taskflow.persistence import models
    from taskflow import states

    from octavia.amphorae.driver_exceptions import exceptions as drv_exceptions
    from octavia.common import exceptions

    LOG = log.getLogger(__name__)

    CONF = cfg.CONF

    # We do not need to log retry exception information. Warning "Could not connect
    #  to instance" will be logged as usual.

    def retryMaskFilter(record):
        if record.exc_info is not None and isinstance(
                record.exc_info[1], (
                    drv_exceptions.AmpConnectionRetry,
                    exceptions.ComputeWaitTimeoutException)):
            return False
        return True

    LOG.logger.addFilter(retryMaskFilter)

    def _details_filter(obj):
        if isinstance(obj, dict):
            ret = {}
            for key in obj:
                if (key in ('certificate', 'private_key', 'passphrase') and
                        isinstance(obj[key], str)):
                    ret[key] = '***'
                elif key == 'intermediates' and isinstance(obj[key], list):
                    ret[key] = ['***'] * len(obj[key])
                else:
                    ret[key] = _details_filter(obj[key])
            return ret
        if isinstance(obj, list):
            return [_details_filter(e) for e in obj]
        return obj

    class FilteredJob(Job):
        def __str__(self):
            # Override the default __str__ method from taskflow.job.base.Job,
            # filter out private information from details
            cls_name = type(self).__name__
            details = _details_filter(self.details)
            return "{}: {} (priority={}, uuid={}, details={})".format(
                cls_name, self.name, self.priority,
                self.uuid, details)

    class JobDetailsFilter(log.logging.Filter):

        def filter(self, record):
            # If the first arg is a Job, convert it now to a string with our custom
            # method
            if isinstance(record.args[0], Job):
                arg0 = record.args[0]
                record.args = (FilteredJob.__str__(arg0),) + record.args[1:]
            return True

    class BaseTaskFlowEngine:
        """This is the task flow engine

        Use this engine to start/load flows in the
        code
        """

        def __init__(self):
            # work around for https://bugs.python.org/issue7980
            datetime.datetime.strptime('2014-06-19 22:47:16', '%Y-%m-%d %H:%M:%S')
            self.executor = concurrent.futures.ThreadPoolExecutor(
                max_workers=CONF.task_flow.max_workers)

        def taskflow_load(self, flow, **kwargs):
            eng = engines.load(
                flow,
                engine=CONF.task_flow.engine,
                executor=self.executor,
                never_resolve=CONF.task_flow.disable_revert,
                **kwargs)
            eng.compile()
            eng.prepare()

            return eng

    class ExtendExpiryListener(base.Listener):

        def __init__(self, engine, job):
            super().__init__(engine)
            self.job = job

        def _task_receiver(self, state, details):
            self.job.extend_expiry(cfg.CONF.task_flow.jobboard_expiration_time)

        def _flow_receiver(self, state, details):
            self.job.extend_expiry(cfg.CONF.task_flow.jobboard_expiration_time)

        def _retry_receiver(self, state, details):
            self.job.extend_expiry(cfg.CONF.task_flow.jobboard_expiration_time)

    class DynamicLoggingConductor(impl_blocking.BlockingConductor):

        def _listeners_from_job(self, job, engine):
            listeners = super()._listeners_from_job(
                job, engine)
            listeners.append(logging.DynamicLoggingListener(engine, log=LOG))

            return listeners

        def _on_job_done(self, job, fut):
            super()._on_job_done(job, fut)
            # Double check that job is complete.
            if (not CONF.task_flow.jobboard_save_logbook and
                    job.state == states.COMPLETE):
                LOG.debug("Job %s is complete. Cleaning up job logbook.", job.name)
                if job.book:
                    try:
                        self._persistence.get_connection().destroy_logbook(
                            job.book.uuid)
                    except taskflow_exc.NotFound:
                        LOG.debug("Logbook for job %s has been already cleaned up",
                                  job.name)

    class ExtendExpiryDynamicLoggingConductor(DynamicLoggingConductor):

        def _listeners_from_job(self, job, engine):
            listeners = super()._listeners_from_job(job, engine)
            listeners.append(ExtendExpiryListener(engine, job))
            return listeners

    class TaskFlowServiceController:

        def __init__(self, driver):
            self.driver = driver

            # Install filter for taskflow executor logger
            taskflow_logger = log.logging.getLogger(
                "taskflow.conductors.backends.impl_executor")
            taskflow_logger.addFilter(JobDetailsFilter())

        def run_poster(self, flow_factory, *args, **kwargs):
            with self.driver.persistence_driver.get_persistence() as persistence:
                with self.driver.job_board(persistence) as job_board:
                    job_id = uuidutils.generate_uuid()
                    job_name = '-'.join([flow_factory.__name__, job_id])
                    job_logbook = models.LogBook(job_name)
                    flow_detail = models.FlowDetail(
                        job_name, job_id)
                    job_details = {
                        'store': kwargs.pop('store')
                    }
                    job_logbook.add(flow_detail)
                    persistence.get_connection().save_logbook(job_logbook)
                    engines.save_factory_details(flow_detail, flow_factory,
                                                 args, kwargs,
                                                 backend=persistence)

                    job_board.post(job_name, book=job_logbook,
                                   details=job_details)
                    self._wait_for_job(job_board)

                    return job_id

        def _wait_for_job(self, job_board):
            # Wait for job to its complete state
            for job in job_board.iterjobs():
                LOG.debug("Waiting for job %s to finish", job.name)
                job.wait()

        def run_conductor(self, name):
            with self.driver.persistence_driver.get_persistence() as persistence:
                with self.driver.job_board(persistence) as board:
                    # Redis and etcd do not expire jobs by default, so jobs won't
                    # be resumed with restart of controller. Add expiry for board
                    # and use special listener.
                    if (CONF.task_flow.jobboard_backend_driver in (
                            'etcd_taskflow_driver',
                            'redis_taskflow_driver')):
                        conductor = ExtendExpiryDynamicLoggingConductor(
                            name, board, persistence=persistence,
                            engine=CONF.task_flow.engine,
                            engine_options={
                                'max_workers': CONF.task_flow.max_workers
                            })
                        if (CONF.task_flow.jobboard_backend_driver ==
                                'redis_taskflow_driver'):
                            # Hack for redis only:
                            # The TTL of the jobs of the Redis Jobboard driver can
                            # be only overriden by using the 'expiry' parameter of
                            # the 'claim' function
                            # For the Etcd driver, the default TTL for all the
                            # locks can be configured while creating the backend
                            board.claim = functools.partial(
                                board.claim,
                                expiry=CONF.task_flow.jobboard_expiration_time)
                    else:
                        conductor = DynamicLoggingConductor(
                            name, board, persistence=persistence,
                            engine=CONF.task_flow.engine)

                    waiter_th = concurrent.futures.ThreadPoolExecutor(
                        max_workers=1)
                    waiter_th.submit(self._waiter, conductor)

                    conductor.run()

        def _extend_jobs(self, conductor, expiration_time):
            conductor_name = conductor._name

            with self.driver.persistence_driver.get_persistence() as persistence:
                with self.driver.job_board(persistence) as board:
                    for job in board.iterjobs():
                        try:
                            owner = board.find_owner(job)
                        except TypeError:
                            # taskflow throws an exception if a job is not owned
                            # (probably a bug in taskflow)
                            continue
                        # Only extend expiry for jobs that are owner by our
                        # conductor (from the same process)
                        if owner == conductor_name:
                            if job.expires_in() < expiration_time / 2:
                                LOG.debug("Extend expiry for job %s", job.name)
                                job.extend_expiry(expiration_time)

        def _waiter(self, conductor):
            expiration_time = CONF.task_flow.jobboard_expiration_time

            while True:
                self._extend_jobs(conductor, expiration_time)

                time.sleep(expiration_time / 4)

this page last updated: 2025-10-06 11:22:08

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