def check_log_groups()

in MWAA/verify_env/verify_env.py [0:0]
50 lines of code
3 McCabe index (conditional complexity)

def check_log_groups(input_env, env_name, logs_client, cloudtrail_client):
    '''check if cloudwatch log groups exists, if not check cloudtrail to see why they weren't created'''
    loggroups = logs_client.describe_log_groups(
        logGroupNamePrefix='airflow-'+env_name
    )['logGroups']
    num_of_enabled_log_groups = sum(
        input_env['LoggingConfiguration'][logConfig]['Enabled'] is True
        for logConfig in input_env['LoggingConfiguration']
    )
    num_of_found_log_groups = len(loggroups)
    print('### Checking if log groups were created successfully...\n')
    if num_of_found_log_groups < num_of_enabled_log_groups:
        print('The number of log groups is less than the number of enabled suggesting an error creating', "🚫")
        print('checking cloudtrail for CreateLogGroup/DeleteLogGroup requests...\n')
        events = cloudtrail_client.lookup_events(
            LookupAttributes=[
                {
                    'AttributeKey': 'EventName',
                    'AttributeValue': 'CreateLogGroup'
                },
            ],
            StartTime=input_env['CreatedAt'] - timedelta(minutes=15),
            EndTime=input_env['CreatedAt']
        )['Events']
        events = events + cloudtrail_client.lookup_events(
            LookupAttributes=[
                {
                    'AttributeKey': 'EventName',
                    'AttributeValue': 'DeleteLogGroup'
                },
            ],
            StartTime=input_env['CreatedAt'] - timedelta(minutes=15),
            EndTime=input_env['CreatedAt']
        )['Events']
        events = events + cloudtrail_client.lookup_events(
            LookupAttributes=[
                {
                    'AttributeKey': 'EventName',
                    'AttributeValue': 'DeleteLogGroup'
                },
            ],
            StartTime=datetime.now() - timedelta(minutes=30),
            EndTime=datetime.now()
        )['Events']
        for event in events:
            print('Found CloudTrail event: ', event)
        print('if events are failing, try creating the log groups manually\n')
    else:
        print("number of log groups match suggesting they've been created successfully", "✅")
    return loggroups