Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save azakordonets/0ed63d56e4954fd4df429527fd9f2d56 to your computer and use it in GitHub Desktop.
Save azakordonets/0ed63d56e4954fd4df429527fd9f2d56 to your computer and use it in GitHub Desktop.
import boto3
import json
# Create CloudWatchLogs client
cloudwatch_logs = boto3.client('logs')
paginator = cloudwatch_logs.get_paginator('describe_log_groups')
subscription_details = {}
for response in paginator.paginate():
for logGroup in response['logGroups']:
log_group_name = logGroup['logGroupName']
subscription_filters = cloudwatch_logs.describe_subscription_filters(logGroupName = log_group_name)
if len(subscription_filters['subscriptionFilters']) > 0:
subs_filter_destination_arn = subscription_filters['subscriptionFilters'][0]['destinationArn']
filter_name = subs_filter_destination_arn.split('function:')[1]
print(f"Log group: ${log_group_name} -> ${filter_name}")
if filter_name in subscription_details:
subscription_details[filter_name]['count'] = subscription_details[filter_name]['count'] + 1
subscription_details[filter_name]['log_groups'].append(log_group_name)
else:
subscription_details[filter_name] = {'count': 1, 'log_groups': [log_group_name]}
else:
print(f"Log group: ${log_group_name} -> None")
if 'None' in subscription_details:
subscription_details['None']['count'] = subscription_details['None']['count'] + 1
subscription_details['None']['log_groups'].append(log_group_name)
else:
subscription_details['None'] = {'count': 1, 'log_groups': [log_group_name]}
print("I have finished calculating")
print(json.dumps(subscription_details))
@helenp1983
Copy link

helenp1983 commented Jul 27, 2022

Error when running in lambda

[ERROR] IndexError: list index out of range
Traceback (most recent call last):
File "/var/lang/lib/python3.7/imp.py", line 234, in load_module
return load_source(name, filename, file)
File "/var/lang/lib/python3.7/imp.py", line 171, in load_source
module = _load(spec)
File "", line 696, in _load
File "", line 677, in _load_unlocked
File "", line 728, in exec_module
File "", line 219, in _call_with_frames_removed
File "/var/task/lambda_function.py", line 13, in
filter_name = subs_filter_destination_arn.split('function:')[1]

some changes to get working in a lambda function.

import boto3
import json
def lambda_handler(event, context): 
    # Create CloudWatchLogs client
    cloudwatch_logs = boto3.client('logs')
    paginator = cloudwatch_logs.get_paginator('describe_log_groups')
    subscription_details = {}
    for response in paginator.paginate():
        for logGroup in response['logGroups']:
            log_group_name = logGroup['logGroupName']
            subscription_filters = cloudwatch_logs.describe_subscription_filters(logGroupName = log_group_name)
            if 'subscriptionFilters' in subscription_filters and len(subscription_filters['subscriptionFilters']) > 0:
                subs_filter_destination_arn = subscription_filters['subscriptionFilters'][0]['destinationArn']
                #print (subs_filter_destination_arn) 
                filter_name = subs_filter_destination_arn.split('destination:')[1]
                print(f"Log group: ${log_group_name} -> ${filter_name}")
                if filter_name in subscription_details:
                    subscription_details[filter_name]['count'] = subscription_details[filter_name]['count'] + 1
                    subscription_details[filter_name]['log_groups'].append(log_group_name)
                else:
                    subscription_details[filter_name] = {'count': 1, 'log_groups': [log_group_name]}
            else:
                print(f"Log group: ${log_group_name} -> None")
                if 'None' in subscription_details:
                    subscription_details['None']['count'] = subscription_details['None']['count'] + 1
                    subscription_details['None']['log_groups'].append(log_group_name)
                else:
                    subscription_details['None'] = {'count': 1, 'log_groups': [log_group_name]}
    print("I have finished calculating")
    print(json.dumps(subscription_details))

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment