Function identify_orphaned_workflows

st2common/st2common/services/workflows.py:1485–1563

()

def identify_orphaned_workflows():
    orphaned = []

    # Identify expiry datetime.
    gc_max_idle = cfg.CONF.workflow_engine.gc_max_idle_sec
    utc_now_dt = date_utils.get_datetime_utc_now()
    expiry_dt = utc_now_dt - datetime.timedelta(seconds=gc_max_idle)

    # Identify action executions that are still running. The action execution start timestamp
    # does not necessary means it is the max idle time. The use of workflow_executions_idled_ttl
    # to filter is to reduce the number of action executions that need to be evaluated.
    query_filters = {
        "runner__name": "orquesta",
        "status": ac_const.LIVEACTION_STATUS_RUNNING,
        "start_timestamp__lte": expiry_dt,
    }
    ac_ex_dbs = ex_db_access.ActionExecution.query(**query_filters)

    for ac_ex_db in ac_ex_dbs:
        # Figure out the runtime for the action execution.
        status_change_logs = sorted(
            [
                log
                for log in ac_ex_db.log
                if log["status"] == ac_const.LIVEACTION_STATUS_RUNNING
            ],
            key=lambda x: x["timestamp"],
            reverse=True,
        )

        if len(status_change_logs) <= 0:
            continue

        runtime = (utc_now_dt - status_change_logs[0]["timestamp"]).total_seconds()

        # Fetch the task executions for the workflow execution.
        # Ensure that the root action execution is not being selected.
        wf_ex_id = ac_ex_db.context["workflow_execution"]
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id)
        query_filters = {"workflow_execution": wf_ex_id, "id__ne": ac_ex_db.id}
        tk_ac_ex_dbs = ex_db_access.ActionExecution.query(**query_filters)

        # The workflow execution is orphaned if there are
        # no task executions and runtime passed expiry.
        if len(tk_ac_ex_dbs) <= 0 and runtime > gc_max_idle:
            msg = "The action execution is orphaned and will be canceled by the garbage collector."
            update_progress(wf_ex_db, msg)
            orphaned.append(ac_ex_db)
            continue

        # The workflow execution is orphaned if there are no active task execution and
        # the end_timestamp of the most recent task execution passed expiry.
        has_active_tasks = len([t for t in tk_ac_ex_dbs if t.end_timestamp is None]) > 0

        completed_tasks = [
            t
            for t in tk_ac_ex_dbs
            if t.end_timestamp is not None and t.end_timestamp <= expiry_dt