From 3a2d96cb5d8c7aecca5e50b24e88e5bc44a9f6d3 Mon Sep 17 00:00:00 2001 From: Seth Foster Date: Fri, 13 May 2022 17:08:46 -0400 Subject: [PATCH] Add subsystem metrics for task manager --- awx/main/analytics/subsystem_metrics.py | 80 ++++++++++++++++--------- awx/main/queue.py | 2 - awx/main/scheduler/task_manager.py | 50 +++++++++++++++- awx/main/tasks/system.py | 3 +- awx/settings/defaults.py | 3 + 5 files changed, 107 insertions(+), 31 deletions(-) diff --git a/awx/main/analytics/subsystem_metrics.py b/awx/main/analytics/subsystem_metrics.py index c1ad08a0cbbe..05bb6e8bc07e 100644 --- a/awx/main/analytics/subsystem_metrics.py +++ b/awx/main/analytics/subsystem_metrics.py @@ -8,7 +8,7 @@ from awx.main.consumers import emit_channel_notification root_key = 'awx_metrics' -logger = logging.getLogger('awx.main.wsbroadcast') +logger = logging.getLogger('awx.main.analytics') class BaseM: @@ -16,16 +16,22 @@ def __init__(self, field, help_text): self.field = field self.help_text = help_text self.current_value = 0 + self.metric_has_changed = False - def clear_value(self, conn): + def reset_value(self, conn): conn.hset(root_key, self.field, 0) self.current_value = 0 def inc(self, value): self.current_value += value + self.metric_has_changed = True def set(self, value): self.current_value = value + self.metric_has_changed = True + + def get(self): + return self.current_value def decode(self, conn): value = conn.hget(root_key, self.field) @@ -34,7 +40,7 @@ def decode(self, conn): def to_prometheus(self, instance_data): output_text = f"# HELP {self.field} {self.help_text}\n# TYPE {self.field} gauge\n" for instance in instance_data: - output_text += f'{self.field}{{node="{instance}"}} {instance_data[instance].get(self.field, -1)}\n' # TODO: fix because this -1 is neccessary when dealing with old instances (ex. you didn't clean up your database) + output_text += f'{self.field}{{node="{instance}"}} {instance_data[instance][self.field]}\n' return output_text @@ -46,8 +52,10 @@ def decode_value(self, value): return 0.0 def store_value(self, conn): - conn.hincrbyfloat(root_key, self.field, self.current_value) - self.current_value = 0 + if self.metric_has_changed: + conn.hincrbyfloat(root_key, self.field, self.current_value) + self.current_value = 0 + self.metric_has_changed = False class IntM(BaseM): @@ -58,8 +66,10 @@ def decode_value(self, value): return 0 def store_value(self, conn): - conn.hincrby(root_key, self.field, self.current_value) - self.current_value = 0 + if self.metric_has_changed: + conn.hincrby(root_key, self.field, self.current_value) + self.current_value = 0 + self.metric_has_changed = False class SetIntM(BaseM): @@ -70,10 +80,9 @@ def decode_value(self, value): return 0 def store_value(self, conn): - # do not set value if it has not changed since last time this was called - if self.current_value is not None: + if self.metric_has_changed: conn.hset(root_key, self.field, self.current_value) - self.current_value = None + self.metric_has_changed = False class SetFloatM(SetIntM): @@ -94,13 +103,13 @@ def __init__(self, field, help_text, buckets): self.sum = IntM(field + '_sum', '') super(HistogramM, self).__init__(field, help_text) - def clear_value(self, conn): + def reset_value(self, conn): conn.hset(root_key, self.field, 0) - self.inf.clear_value(conn) - self.sum.clear_value(conn) + self.inf.reset_value(conn) + self.sum.reset_value(conn) for b in self.buckets_to_keys.values(): - b.clear_value(conn) - super(HistogramM, self).clear_value(conn) + b.reset_value(conn) + super(HistogramM, self).reset_value(conn) def observe(self, value): for b in self.buckets: @@ -136,7 +145,7 @@ def to_prometheus(self, instance_data): class Metrics: - def __init__(self, auto_pipe_execute=True, instance_name=None): + def __init__(self, auto_pipe_execute=False, instance_name=None): self.pipe = redis.Redis.from_url(settings.BROKER_URL).pipeline() self.conn = redis.Redis.from_url(settings.BROKER_URL) self.last_pipe_execute = time.time() @@ -152,6 +161,8 @@ def __init__(self, auto_pipe_execute=True, instance_name=None): Instance = apps.get_model('main', 'Instance') if instance_name: self.instance_name = instance_name + elif settings.IS_TESTING(): + self.instance_name = "awx_testing" else: self.instance_name = Instance.objects.me().hostname @@ -167,10 +178,22 @@ def __init__(self, auto_pipe_execute=True, instance_name=None): HistogramM( 'callback_receiver_batch_events_insert_db', 'Number of events batch inserted into database', settings.SUBSYSTEM_METRICS_BATCH_INSERT_BUCKETS ), + SetFloatM('callback_receiver_event_processing_avg_seconds', 'Average processing time per event per callback receiver batch'), FloatM('subsystem_metrics_pipe_execute_seconds', 'Time spent saving metrics to redis'), IntM('subsystem_metrics_pipe_execute_calls', 'Number of calls to pipe_execute'), FloatM('subsystem_metrics_send_metrics_seconds', 'Time spent sending metrics to other nodes'), - SetFloatM('callback_receiver_event_processing_avg_seconds', 'Average processing time per event per callback receiver batch'), + SetFloatM('task_manager_get_tasks_seconds', 'Time spent in loading all tasks from db'), + SetFloatM('task_manager_start_task_seconds', 'Time spent starting task'), + SetFloatM('task_manager_process_running_tasks_seconds', 'Time spent processing running tasks'), + SetFloatM('task_manager_process_pending_tasks_seconds', 'Time spent processing pending tasks'), + SetFloatM('task_manager_generate_dependencies_seconds', 'Time spent generating dependencies for pending tasks'), + SetFloatM('task_manager_spawn_workflow_graph_jobs_seconds', 'Time spent spawning workflow jobs'), + SetFloatM('task_manager__schedule_seconds', 'Time spent in running the entire _schedule'), + IntM('task_manager_schedule_calls', 'Number of calls to task manager schedule'), + SetIntM('task_manager_tasks_started', 'Number of tasks started'), + SetIntM('task_manager_running_processed', 'Number of running tasks processed'), + SetIntM('task_manager_pending_processed', 'Number of pending tasks processed'), + SetIntM('task_manager_tasks_blocked', 'Number of tasks blocked from running'), ] # turn metric list into dictionary with the metric name as a key self.METRICS = {} @@ -180,11 +203,11 @@ def __init__(self, auto_pipe_execute=True, instance_name=None): # track last time metrics were sent to other nodes self.previous_send_metrics = SetFloatM('send_metrics_time', 'Timestamp of previous send_metrics call') - def clear_values(self, fields=None): - if not fields: - fields = self.METRICS.keys() - for m in fields: - self.METRICS[m].clear_value(self.conn) + def reset_values(self): + # intended to be called once on app startup to reset all metric + # values to 0 + for m in self.METRICS.values(): + m.reset_value(self.conn) self.metrics_have_changed = True self.conn.delete(root_key + "_lock") @@ -192,19 +215,22 @@ def inc(self, field, value): if value != 0: self.METRICS[field].inc(value) self.metrics_have_changed = True - if self.auto_pipe_execute is True and self.should_pipe_execute() is True: + if self.auto_pipe_execute is True: self.pipe_execute() def set(self, field, value): self.METRICS[field].set(value) self.metrics_have_changed = True - if self.auto_pipe_execute is True and self.should_pipe_execute() is True: + if self.auto_pipe_execute is True: self.pipe_execute() + def get(self, field): + return self.METRICS[field].get() + def observe(self, field, value): self.METRICS[field].observe(value) self.metrics_have_changed = True - if self.auto_pipe_execute is True and self.should_pipe_execute() is True: + if self.auto_pipe_execute is True: self.pipe_execute() def serialize_local_metrics(self): @@ -252,8 +278,8 @@ def pipe_execute(self): def send_metrics(self): # more than one thread could be calling this at the same time, so should - # get acquire redis lock before sending metrics - lock = self.conn.lock(root_key + '_lock', thread_local=False) + # acquire redis lock before sending metrics + lock = self.conn.lock(root_key + '_lock') if not lock.acquire(blocking=False): return try: diff --git a/awx/main/queue.py b/awx/main/queue.py index ebac0622e490..26d23a5cbb8a 100644 --- a/awx/main/queue.py +++ b/awx/main/queue.py @@ -8,7 +8,6 @@ # Django from django.conf import settings -import awx.main.analytics.subsystem_metrics as s_metrics __all__ = ['CallbackQueueDispatcher'] @@ -28,7 +27,6 @@ def __init__(self): self.queue = getattr(settings, 'CALLBACK_QUEUE', '') self.logger = logging.getLogger('awx.main.queue.CallbackQueueDispatcher') self.connection = redis.Redis.from_url(settings.BROKER_URL) - self.subsystem_metrics = s_metrics.Metrics() def dispatch(self, obj): self.connection.rpush(self.queue, json.dumps(obj, cls=AnsibleJSONEncoder)) diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py index 6fa200fcd68a..0d0284a2feb7 100644 --- a/awx/main/scheduler/task_manager.py +++ b/awx/main/scheduler/task_manager.py @@ -6,6 +6,9 @@ import logging import uuid import json +import time +import sys +import signal # Django from django.db import transaction, connection @@ -38,12 +41,24 @@ from awx.main.scheduler.dependency_graph import DependencyGraph from awx.main.scheduler.task_manager_models import TaskManagerInstances from awx.main.scheduler.task_manager_models import TaskManagerInstanceGroups +import awx.main.analytics.subsystem_metrics as s_metrics from awx.main.utils import decrypt_field logger = logging.getLogger('awx.main.scheduler') +def timeit(func): + def inner(*args, **kwargs): + t_now = time.perf_counter() + result = func(*args, **kwargs) + dur = time.perf_counter() - t_now + args[0].subsystem_metrics.inc("task_manager_" + func.__name__ + "_seconds", dur) + return result + + return inner + + class TaskManager: def __init__(self): """ @@ -62,6 +77,7 @@ def __init__(self): # will no longer be started and will be started on the next task manager cycle. self.start_task_limit = settings.START_TASK_LIMIT self.time_delta_job_explanation = timedelta(seconds=30) + self.subsystem_metrics = s_metrics.Metrics(auto_pipe_execute=False) def after_lock_init(self, all_sorted_tasks): """ @@ -100,6 +116,7 @@ def job_blocked_by(self, task): return None + @timeit def get_tasks(self, status_list=('pending', 'waiting', 'running')): jobs = [j for j in Job.objects.filter(status__in=status_list).prefetch_related('instance_group')] inventory_updates_qs = ( @@ -125,6 +142,7 @@ def get_inventory_source_tasks(self, all_sorted_tasks): inventory_ids.add(task.inventory_id) return [invsrc for invsrc in InventorySource.objects.filter(inventory_id__in=inventory_ids, update_on_launch=True)] + @timeit def spawn_workflow_graph_jobs(self, workflow_jobs): for workflow_job in workflow_jobs: if workflow_job.cancel_flag: @@ -231,7 +249,9 @@ def process_finished_workflow_jobs(self, workflow_jobs): schedule_task_manager() return result + @timeit def start_task(self, task, instance_group, dependent_tasks=None, instance=None): + self.subsystem_metrics.inc("task_manager_tasks_started", 1) self.start_task_limit -= 1 if self.start_task_limit == 0: # schedule another run immediately after this task manager @@ -291,6 +311,7 @@ def post_commit(): task.websocket_emit_status(task.status) # adds to on_commit connection.on_commit(post_commit) + @timeit def process_running_tasks(self, running_tasks): for task in running_tasks: self.dependency_graph.add_job(task) @@ -439,6 +460,7 @@ def gen_dep_for_inventory_update(self, inventory_task): latest_src_project_update.scm_inventory_updates.add(inventory_task) return created_dependencies + @timeit def generate_dependencies(self, undeped_tasks): created_dependencies = [] for task in undeped_tasks: @@ -453,6 +475,7 @@ def generate_dependencies(self, undeped_tasks): return created_dependencies + @timeit def process_pending_tasks(self, pending_tasks): running_workflow_templates = {wf.unified_job_template_id for wf in self.get_running_workflow_jobs()} tasks_to_update_job_explanation = [] @@ -461,6 +484,7 @@ def process_pending_tasks(self, pending_tasks): break blocked_by = self.job_blocked_by(task) if blocked_by: + self.subsystem_metrics.inc("task_manager_tasks_blocked", 1) task.log_lifecycle("blocked", blocked_by=blocked_by) job_explanation = gettext_noop(f"waiting for {blocked_by._meta.model_name}-{blocked_by.id} to finish") if task.job_explanation != job_explanation: @@ -602,17 +626,22 @@ def reap_jobs_from_orphaned_instances(self): def process_tasks(self, all_sorted_tasks): running_tasks = [t for t in all_sorted_tasks if t.status in ['waiting', 'running']] - self.process_running_tasks(running_tasks) + self.subsystem_metrics.inc("task_manager_running_processed", len(running_tasks)) pending_tasks = [t for t in all_sorted_tasks if t.status == 'pending'] + undeped_tasks = [t for t in pending_tasks if not t.dependencies_processed] dependencies = self.generate_dependencies(undeped_tasks) deps_of_deps = self.generate_dependencies(dependencies) dependencies += deps_of_deps self.process_pending_tasks(dependencies) + self.subsystem_metrics.inc("task_manager_pending_processed", len(dependencies)) + self.process_pending_tasks(pending_tasks) + self.subsystem_metrics.inc("task_manager_pending_processed", len(pending_tasks)) + @timeit def _schedule(self): finished_wfjs = [] all_sorted_tasks = self.get_tasks() @@ -648,6 +677,22 @@ def _schedule(self): self.process_tasks(all_sorted_tasks) return finished_wfjs + def record_aggregate_metrics(self, *args): + if not settings.IS_TESTING(): + # increment task_manager_schedule_calls regardless if the other + # metrics are recorded + s_metrics.Metrics(auto_pipe_execute=True).inc("task_manager_schedule_calls", 1) + _schedule_dur = self.subsystem_metrics.get("task_manager__schedule_seconds") + # to prevent overriding short-duration task manager calls, only + # record metrics if the total time to run task manager is greater + # than the user-defined setting + if _schedule_dur > settings.SUBSYSTEM_METRICS_TASK_MANAGER_EXECUTION_TIME: + self.subsystem_metrics.pipe_execute() + + def record_aggregate_metrics_and_exit(self, *args): + self.record_aggregate_metrics() + sys.exit(1) + def schedule(self): # Lock with advisory_lock('task_manager_lock', wait=False) as acquired: @@ -657,5 +702,8 @@ def schedule(self): return logger.debug("Starting Scheduler") with task_manager_bulk_reschedule(): + # if sigterm due to timeout, still record metrics + signal.signal(signal.SIGTERM, self.record_aggregate_metrics_and_exit) self._schedule() + self.record_aggregate_metrics() logger.debug("Finishing Scheduler") diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py index 8c698609a5cd..541415f2b854 100644 --- a/awx/main/tasks/system.py +++ b/awx/main/tasks/system.py @@ -103,7 +103,8 @@ def dispatch_startup(): # apply_cluster_membership_policies() cluster_node_heartbeat() - Metrics().clear_values() + m = Metrics() + m.reset_values() # Update Tower's rsyslog.conf file based on loggins settings in the db reconfigure_rsyslog() diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py index 24b4ca79ffcf..e3af5b71f2de 100644 --- a/awx/settings/defaults.py +++ b/awx/settings/defaults.py @@ -241,6 +241,9 @@ def IS_TESTING(argv=None): # Interval in seconds for saving local metrics to redis SUBSYSTEM_METRICS_INTERVAL_SAVE_TO_REDIS = 2 +# Only record stats from task manager cycles that are >= this execution time (seconds) +SUBSYSTEM_METRICS_TASK_MANAGER_EXECUTION_TIME = 15 + # The maximum allowed jobs to start on a given task manager cycle START_TASK_LIMIT = 100