def check_connectivity_to_dep_services()

in MWAA/verify_env/verify_env.py [0:0]

• 51 lines of code
• 2 McCabe index (conditional complexity)

def check_connectivity_to_dep_services(input_env, input_subnets, ec2_client, ssm_client, mwaa_utilized_services):
    '''
    uses ssm document AWSSupport-ConnectivityTroubleshooter to check connectivity between MWAA's enis
    and a list of services. More information on this document can be found here
    https://docs.aws.amazon.com/systems-manager/latest/userguide/automation-awssupport-connectivitytroubleshooter.html
    '''
    print("### Testing connectivity to the following service endpoints from MWAA enis...")
    vpc = subnets[0]['VpcId']
    security_groups = input_env['NetworkConfiguration']['SecurityGroupIds']
    for service in mwaa_utilized_services:
        # retry 5 times for just one of the enis the service uses
        for i in range(0, 5):
            try:
                # get ENIs used by MWAA
                enis = get_enis(subnet_ids, vpc, security_groups)
                if not enis:
                    print("no enis found for MWAA, exiting test for ", service['service'])
                    print("please try accessing the airflow UI and then try running this script again")
                    break
                eni = list(enis.values())[0]
                interface_ip = ec2_client.describe_network_interfaces(
                    NetworkInterfaceIds=[eni]
                )['NetworkInterfaces'][0]['PrivateIpAddress']
                ssm_execution_id = ''
                ssm_execution_id = ssm_client.start_automation_execution(
                    DocumentName='AWSSupport-ConnectivityTroubleshooter',
                    DocumentVersion='$DEFAULT',
                    Parameters={
                        'SourceIP': [interface_ip],
                        'DestinationIP': [get_ip_address(service['service'], input_subnets[0]['VpcId'])],
                        'DestinationPort': [service['port']],
                        'SourceVpc': [vpc],
                        'DestinationVpc': [vpc],
                        'SourcePortRange': ["0-65535"]
                    }
                )['AutomationExecutionId']
                wait_for_ssm_step_one_to_finish(ssm_execution_id, ssm_client)
                execution = ssm_client.get_automation_execution(
                    AutomationExecutionId=ssm_execution_id
                )['AutomationExecution']
                # check if the failure is due to not finding the eni. If it is, retry testing the service again
                if execution['StepExecutions'][0]['StepStatus'] != 'Failed':
                    print('Testing connectivity between eni', eni, "with private ip of",
                          interface_ip, "and", service['service'], "on port", service['port'])
                    print("Please follow this link to view the results of the test:")
                    print("https://console.aws.amazon.com/systems-manager/automation/execution/" + ssm_execution_id +
                          "?REGION=" + REGION + "\n")
                    break
            except ClientError as client_error:
                print('Attempt', i, 'Encountered error', client_error.response['Error']['Message'], ' retrying...')
    print("")