External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-07 17:37:37 -02:30
Code Issues Packages Projects Releases Wiki Activity

pid kill each of the 3 task managers on timeout

Browse Source
This commit is contained in:
Seth Foster
2022-07-06 00:44:30 -04:00
parent b26d2ab0e9
commit b3eb9e0193
2 changed files with 9 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
awx/main/dispatch/pool.py
View File

@@ -401,13 +401,15 @@ class AutoscalePool(WorkerPool):
# the task manager to never do more work # the task manager to never do more work
current_task = w.current_task current_task = w.current_task
if current_task and isinstance(current_task, dict): if current_task and isinstance(current_task, dict):
if current_task.get('task', '').endswith('tasks.run_task_manager'): endings = ['tasks.task_manager', 'tasks.dependency_manager', 'tasks.workflow_manager']
current_task_name = current_task.get('task', '')
if any([current_task_name.endswith(e) for e in endings]):
if 'started' not in current_task: if 'started' not in current_task:
w.managed_tasks[current_task['uuid']]['started'] = time.time() w.managed_tasks[current_task['uuid']]['started'] = time.time()
age = time.time() - current_task['started'] age = time.time() - current_task['started']
w.managed_tasks[current_task['uuid']]['age'] = age w.managed_tasks[current_task['uuid']]['age'] = age
if age > (settings.TASK_MANAGER_TIMEOUT + settings.TASK_MANAGER_TIMEOUT_GRACE_PERIOD): if age > (settings.TASK_MANAGER_TIMEOUT + settings.TASK_MANAGER_TIMEOUT_GRACE_PERIOD):
logger.error(f'run_task_manager has held the advisory lock for {age}, sending SIGTERM to {w.pid}') # noqa logger.error(f'{current_task_name} has held the advisory lock for {age}, sending SIGTERM to {w.pid}') # noqa
os.kill(w.pid, signal.SIGTERM) os.kill(w.pid, signal.SIGTERM)
for m in orphaned: for m in orphaned:

12
awx/main/scheduler/task_manager.py
View File

@@ -59,7 +59,8 @@ def timeit(func):
class TaskBase: class TaskBase:
def __init__(self): def __init__(self, prefix=""):
self.prefix = prefix
# initialize each metric to 0 and force metric_has_changed to true. This # initialize each metric to 0 and force metric_has_changed to true. This
# ensures each task manager metric will be overridden when pipe_execute # ensures each task manager metric will be overridden when pipe_execute
# is called later. # is called later.
@@ -127,8 +128,7 @@ class TaskBase:
class WorkflowManager(TaskBase): class WorkflowManager(TaskBase):
def __init__(self): def __init__(self):
self.prefix = "workflow_manager" super().__init__(prefix="workflow_manager")
super().__init__()
@timeit @timeit
def spawn_workflow_graph_jobs(self, workflow_jobs): def spawn_workflow_graph_jobs(self, workflow_jobs):
@@ -298,8 +298,7 @@ class WorkflowManager(TaskBase):
class DependencyManager(TaskBase): class DependencyManager(TaskBase):
def __init__(self): def __init__(self):
self.prefix = "dependency_manager" super().__init__(prefix="dependency_manager")
super().__init__()
def create_project_update(self, task, project_id=None): def create_project_update(self, task, project_id=None):
if project_id is None: if project_id is None:
@@ -497,8 +496,7 @@ class TaskManager(TaskBase):
# 5 minutes to start pending jobs. If this limit is reached, pending jobs # 5 minutes to start pending jobs. If this limit is reached, pending jobs
# will no longer be started and will be started on the next task manager cycle. # will no longer be started and will be started on the next task manager cycle.
self.time_delta_job_explanation = timedelta(seconds=30) self.time_delta_job_explanation = timedelta(seconds=30)
self.prefix = "task_manager" super().__init__(prefix="task_manager")
super().__init__()
def after_lock_init(self, all_sorted_tasks): def after_lock_init(self, all_sorted_tasks):
""" """