def verify_vcpus_limit()

in source/soca/cluster_manager/add_nodes.py [0:0]

• 61 lines of code
• 21 McCabe index (conditional complexity)

def verify_vcpus_limit(instance_type, desired_capacity, quota_info):
    cpus_count_pattern = re.search(r'[.](\d+)', instance_type)
    instance_family = instance_type[0].upper()
    if cpus_count_pattern:
        vcpus_per_instance = int(cpus_count_pattern.group(1)) * 4
    else:
        if 'xlarge' in instance_type:
            vcpus_per_instance = 4
        else:
            vcpus_per_instance = 2

    total_vcpus_requested = vcpus_per_instance * int(desired_capacity)
    running_vcpus = 0

    max_vcpus_allowed = False
    quota_name = False

    if not quota_info or instance_type not in quota_info.keys():
        # Get quota
        token = True
        next_token = False
        while token is True:
            if next_token is False:
                response = servicequotas.list_service_quotas(
                    ServiceCode="ec2",
                    MaxResults=100)

            else:
                response = servicequotas.list_service_quotas(
                    ServiceCode="ec2",
                    MaxResults=100,
                    NextToken=next_token)
            try:
                next_token = response['NextToken']
            except KeyError:
                token = False

            for quota in response["Quotas"]:
                # Remove 'Running' to prevent error when using R instance family
                if "running on-demand" in quota["QuotaName"].lower() and instance_family in quota["QuotaName"].replace("Running",""):
                    max_vcpus_allowed = quota["Value"]
                    quota_name = quota["QuotaName"]
                    #print("Instance Type {}. Quota {}, Max CPUs {}".format(instance_type, quota_name, max_vcpus_allowed))
    else:
        max_vcpus_allowed = quota_info[instance_type]["max_vcpus_allowed"]
        quota_name = quota_info[instance_type]["quota_name"]

    if max_vcpus_allowed is False:
        return {"message": "Unable to find ServiceQuota for {}".format(instance_type), "quota_info": quota_info}

    # list all ec2 instances
    if "standard" in quota_name.lower():
        instances_family_allowed_in_quota = re.search(r"running on-demand standard \((.*)\) instances", quota_name.lower()).group(1).split(',')
    else:
        instances_family_allowed_in_quota = list(re.search(r"running on-demand (.*) instances", quota_name.lower()).group(1))

    if not quota_info or instance_type not in quota_info.keys():
        all_instances_available = ec2._service_model.shape_for('InstanceType').enum
        all_instances_for_quota = [instance_family for x in instances_family_allowed_in_quota for instance_family in all_instances_available if instance_family.startswith(x.rstrip().lstrip())]
        required_api_calls = ceil(len(all_instances_for_quota) / 190)
        for i in range(0, required_api_calls):
            # DescribeInstances has a limit of 200 attributes per filter
            instances_to_check = all_instances_for_quota[i * 190:(i + 1) * 190]
            if instances_to_check:
                running_vcpus += find_running_cpus_per_instance(instances_to_check)

    else:
        running_vcpus = quota_info[instance_type]["vcpus_provisioned"]

    quota_info[instance_type] = {"max_vcpus_allowed": max_vcpus_allowed,
                                 "vcpus_provisioned": running_vcpus + total_vcpus_requested,
                                 "quota_name": quota_name}

    if max_vcpus_allowed >= (running_vcpus + total_vcpus_requested):
        return {"message": True, "quota_info": quota_info}
    else:
        return {"message": "Job cannot start due to AWS Service limit. Max Vcpus allowed {}. Detected running Vcpus {}. Requested Vcpus for this job {}. Quota Name {}".format(max_vcpus_allowed, running_vcpus, total_vcpus_requested, quota_name), "quota_info": quota_info}