Repository: ansible/awx
File: awx/main/scheduler/task_manager.py (Python)

Code before:
            subsystem_metrics = s_metrics.Metrics(auto_pipe_execute=False)
            subsystem_metrics.inc("task_manager_total_runs", 1)
            if (
                "_schedule_seconds" not in self.aggregate_metrics
                or self.aggregate_metrics["_schedule_seconds"] > settings.SUBSYSTEM_METRICS_TASK_MANAGER_EXECUTION_TIME
            ):
                # clear all task manager related values to wipe out data
                # from previous task manager run
                task_manager_metrics = [m for m in subsystem_metrics.METRICS.keys() if m.startswith("task_manager") and not m.endswith("total_runs")]
                subsystem_metrics.clear_values(fields=task_manager_metrics)
                for a, v in self.aggregate_metrics.items():
                    subsystem_metrics.set("task_manager_" + a, v)
            subsystem_metrics.pipe_execute()
        sys.exit(1)

    def schedule(self):
        # Lock
        with advisory_lock('task_manager_lock', wait=False) as acquired:
            with transaction.atomic():
                if acquired is False:
                    logger.debug("Not running scheduler, another task holds lock")
                    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)
                    self._schedule()
                    self.record_aggregate_metrics()
                logger.debug("Finishing Scheduler")


Review comment (suggestion):
@kdelee as discussed, this will ensure we record the metrics in the event the dispatcher kills the task manager due to the 5 minute timeout.

it will also record a `task_manager_total_runs` metric that increments each time the task manager runs

Code after:
            # Only record metrics if the last time recording was more
            # than SUBSYSTEM_METRICS_TASK_MANAGER_RECORD_INTERVAL ago.
            # Prevents a short-duration task manager that runs directly after a
            # long task manager to override useful metrics.
            current_time = time.time()
            time_last_recorded = current_time - self.subsystem_metrics.decode("task_manager_recorded_timestamp")
            if time_last_recorded > settings.SUBSYSTEM_METRICS_TASK_MANAGER_RECORD_INTERVAL:
                logger.debug(f"recording metrics, last recorded {time_last_recorded} seconds ago")
                self.subsystem_metrics.set("task_manager_recorded_timestamp", current_time)
                self.subsystem_metrics.pipe_execute()
            else:
                logger.debug(f"skipping recording metrics, last recorded {time_last_recorded} seconds ago")

    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:
            with transaction.atomic():
                if acquired is False:
                    logger.debug("Not running scheduler, another task holds lock")
                    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")