def scale_asg_if_needed()

in lambdas/scale_out_runner/app.py [0:0]

• 26 lines of code
• 8 McCabe index (conditional complexity)

def scale_asg_if_needed(num_queued_jobs: int) -> dict:
    asg = boto3.client('autoscaling', region_name=ASG_REGION_NAME)

    resp = asg.describe_auto_scaling_groups(
        AutoScalingGroupNames=[ASG_GROUP_NAME],
    )

    asg_info = resp['AutoScalingGroups'][0]

    current = asg_info['DesiredCapacity']
    max_size = asg_info['MaxSize']

    busy = 0
    for instance in asg_info['Instances']:
        if instance['LifecycleState'] == 'InService' and instance['ProtectedFromScaleIn']:
            busy += 1
    app.log.info("Busy instances: %d, num_queued_jobs: %d, current_size: %d", busy, num_queued_jobs, current)

    new_size = num_queued_jobs + busy
    if new_size > current:
        if new_size <= max_size or current < max_size:
            try:
                new_size = min(new_size, max_size)
                asg.set_desired_capacity(AutoScalingGroupName=ASG_GROUP_NAME, DesiredCapacity=new_size)
                return {'new_capcity': new_size}
            except asg.exceptions.ScalingActivityInProgressFault as e:
                return {'error': str(e)}
        else:
            return {'capacity_at_max': True}
    else:
        return {'idle_instances': True}