Repository: ansible/awx
File: awx/main/dispatch/pool.py (Python)

Code before:
        logger.error("could not write payload to any queue, attempted order: {}".format(write_attempt_order))
        return None

    def stop(self, signum):
        try:
            for worker in self.workers:
                os.kill(worker.pid, signum)
        except Exception:
            logger.exception('could not kill {}'.format(worker.pid))


class AutoscalePool(WorkerPool):
    """
    An extended pool implementation that automatically scales workers up and
    down based on demand
    """

    pool_cls = StatefulPoolWorker

    def __init__(self, *args, **kwargs):
        self.max_workers = kwargs.pop('max_workers', None)
        super(AutoscalePool, self).__init__(*args, **kwargs)

        if self.max_workers is None:
            # Get memory from ansible-runner, consistent with Instance.local_health_check, important to be consistent
            total_memory_gb = get_mem_in_bytes()

            # This may replace memory calculation with a user override
            corrected_memory = get_corrected_memory(total_memory_gb)

            # Get same number as max forks based on memory, this function takes memory as bytes
            self.max_workers = get_mem_effective_capacity(corrected_memory, is_control_node=True)

            # add magic prime number of extra workers to ensure
            # we have a few extra workers to run the heartbeat
            self.max_workers += 7

        # max workers can't be less than min_workers
        self.max_workers = max(self.min_workers, self.max_workers)

        # the task manager enforces settings.TASK_MANAGER_TIMEOUT on its own
        # but if the task takes longer than the time defined here, we will force it to stop here
        self.task_manager_timeout = settings.TASK_MANAGER_TIMEOUT + settings.TASK_MANAGER_TIMEOUT_GRACE_PERIOD

        # initialize some things for subsystem metrics periodic gathering
        # the AutoscalePool class does not save these to redis directly, but reports via produce_subsystem_metrics
        self.scale_up_ct = 0
        self.worker_count_max = 0

        # last time we wrote current tasks, to avoid too much log spam
        self.last_task_list_log = time.monotonic()

Review comment (suggestion):
to give some history, the use of `psutil` here is completely ancient as of the original introduction https://github.com/ansible/awx/pull/2266

Then https://github.com/ansible/awx/pull/10954 changed to use the ansible-runner memory util. The issue of max_workers parity may not have really been on the radar at that point.

@kdelee did some work to fix basically this same problem in https://github.com/ansible/awx/pull/11800, but didn't touch the fact that psutil was used in a now one-off way.

Code after:
        logger.error("could not write payload to any queue, attempted order: {}".format(write_attempt_order))
        return None

    def stop(self, signum):
        try:
            for worker in self.workers:
                os.kill(worker.pid, signum)
        except Exception:
            logger.exception('could not kill {}'.format(worker.pid))


def get_auto_max_workers():
    """Method we normally rely on to get max_workers

    Uses almost same logic as Instance.local_health_check
    The important thing is to be MORE than Instance.capacity
    so that the task-manager does not over-schedule this node

    Ideally we would just use the capacity from the database plus reserve workers,
    but this poses some bootstrap problems where OCP task containers
    register themselves after startup
    """
    # Get memory from ansible-runner
    total_memory_gb = get_mem_in_bytes()

    # This may replace memory calculation with a user override
    corrected_memory = get_corrected_memory(total_memory_gb)

    # Get same number as max forks based on memory, this function takes memory as bytes
    mem_capacity = get_mem_effective_capacity(corrected_memory, is_control_node=True)

    # Follow same process for CPU capacity constraint
    cpu_count = get_cpu_count()
    corrected_cpu = get_corrected_cpu(cpu_count)
    cpu_capacity = get_cpu_effective_capacity(corrected_cpu, is_control_node=True)

    # Here is what is different from health checks,
    auto_max = max(mem_capacity, cpu_capacity)

    # add magic number of extra workers to ensure
    # we have a few extra workers to run the heartbeat
    auto_max += 7

    return auto_max


class AutoscalePool(WorkerPool):
    """
    An extended pool implementation that automatically scales workers up and
    down based on demand
    """