def update_cluster()

in images/controlplane/upgrade/app/cluster_upgrade.py [0:0]

• 34 lines of code
• 5 McCabe index (conditional complexity)

def update_cluster(config_dict):
    cluster_name = config_dict[ConfigKey.cluster_name_key]
    cluster_version = config_dict[ConfigKey.cluster_version_key]

    logger.info(f'Cluster found, named: {config_dict[ConfigKey.cluster_name_key]}')

    # write config
    configure_cli = f'aws eks update-kubeconfig --name {cluster_name}'
    output = subprocess.run(f'{configure_cli}', encoding='utf-8', capture_output=True, shell=True)
    if output.returncode != 0:
        raise RuntimeError(f'Falied to create kube config file {output.stderr}.')
    logger.info('Create kube config file.')

    # updates existing config file
    output = subprocess.run(['/usr/local/bin/eksctl', 'utils', 'write-kubeconfig', '--cluster', cluster_name],
                            encoding='utf-8', capture_output=True)
    logger.info(output)

    # update .kube/config for aws2 version of cli
    subprocess.run(['sed', '-i', '-e', 's/command: aws/command: \/usr\/local\/bin\/aws/g', kube_config_path],
                   encoding='utf-8', capture_output=True)

    # enable cluster logs
    eks = boto3.client('eks')
    response = eks.describe_cluster(name=config_dict[ConfigKey.cluster_name_key])

    update_status = response['cluster']['status']
    if update_status == 'ACTIVE':
        logger.info('run eksctl enable logs command')
        output = subprocess.run(
            f'/usr/local/bin/eksctl utils update-cluster-logging --cluster {cluster_name} --enable-types=controllerManager --approve',
            encoding='utf-8', capture_output=True, shell=True)
        command_output = get_stdout(output)
        logger.info(f'enable cluster control plane logs: {command_output}')

        # upgrade eksctl cluster
        logger.info('run eksctl upgrade control plane command')
        output = subprocess.run(
            f'/usr/local/bin/eksctl upgrade cluster --name={cluster_name} --version={cluster_version} --approve',
            encoding='utf-8', capture_output=True, shell=True)
        command_output = get_stdout(output)
        logger.info(f'update cluster control plane: {command_output}')
    elif update_status == 'UPDATING':
        logger.info('Cluster is already updating')
    else:
        logger.info(f'Cluster not in upgrading status - : {update_status}')