firehose-cloudwatch-trigger-stack.yaml

AWSTemplateFormatVersion: '2010-09-09'

Parameters:
  LogGroupConfig:
    Description: "JSON array of objects representing your LogGroup and Filters (if applicable). For example: [{\"LogGroupName\":\"logGroup1\",\"FilterPattern\":\"filter1\"}]"
    Type: String
  LogGroupArns:
    Description: "Comma-separated list of CloudWatch Log Group ARNs to create subscription to Data Firehose"
    Type: CommaDelimitedList
  InvalidLogGroups:
    Description: "Comma-separated list of CloudWatch Log Groups provided in use input which are invalid and should be skipped."
    Type: CommaDelimitedList
  LoggingFirehoseStreamArn:
    Type: String
    Description: "Data Firehose arn to create cloudwatch log group subscription"

Conditions:
  HasValidLogGroups: !Not [!Equals [!Select [0, !Ref LogGroupArns], ""]]

Resources:
  NewRelicLogsCloudWatchFirehoseLogGroupTriggers:
    Type: 'Custom::CloudWatchNotifications'
    Condition: HasValidLogGroups
    Properties:
      ServiceToken: !GetAtt NewRelicLogsCloudWatchFirehoseEventLambda.Arn
      LoggingFirehoseStreamArn: !Ref LoggingFirehoseStreamArn
      LogGroupConfig: !Ref LogGroupConfig
      InvalidLogGroups: !Ref InvalidLogGroups

  NewRelicLogsCloudWatchFirehoseLambdaIAMRole:
    Type: "AWS::IAM::Role"
    Condition: HasValidLogGroups
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - "sts:AssumeRole"
      Policies:
        - PolicyName: "LambdaExecutionPolicy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - logs:PutSubscriptionFilter
                  - logs:DeleteSubscriptionFilter
                Resource: !Ref LogGroupArns
              - Effect: Allow
                Action:
                  - 'logs:CreateLogGroup'
                  - 'logs:CreateLogStream'
                  - 'logs:PutLogEvents'
                Resource: 'arn:aws:logs:*:*:*'
              - Effect: Allow
                Action:
                  - 'iam:PassRole'
                Resource: !GetAtt NewRelicLogsCloudWatchFirehoseIAMRole.Arn

  NewRelicLogsCloudWatchFirehoseIAMRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - logs.amazonaws.com
            Action:
              - "sts:AssumeRole"
      Policies:
        - PolicyName: "FirehoseWritePolicy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - firehose:PutRecord
                  - firehose:PutRecordBatch
                Resource: !Ref LoggingFirehoseStreamArn
             
  NewRelicLogsCloudWatchFirehoseEventLambda:
    Type: 'AWS::Lambda::Function'
    Condition: HasValidLogGroups
    Properties:
      Environment:
        Variables:
          CloudWatchFirehoseRoleArn: !GetAtt NewRelicLogsCloudWatchFirehoseIAMRole.Arn
      Code:
        ZipFile: |
          import json
          import boto3
          import cfnresponse
          import logging
          import os
          import hashlib

          logger = logging.getLogger()
          logger.setLevel(logging.INFO)

          log_client = boto3.client('logs')
          
          def lambda_handler(event, context):
              response = {}
              try:
                  event_data = event['ResourceProperties']
                  request_type = event['RequestType']
                  firehose_arn = event_data.get('LoggingFirehoseStreamArn', '')
                  log_group_config_str = event_data.get('LogGroupConfig', [])
                  invalid_log_groups = set(event_data.get('InvalidLogGroups', []))

                  # Parsing LogGroupConfig JSON array
                  log_group_config = json.loads(log_group_config_str)

                  # Unique filter name for this stack using firehose ARN
                  filter_name = f'NewRelicLogsFirehoseSubscription_{hashlib.sha256(firehose_arn.encode()).hexdigest()[:20]}'

                  if request_type in ['Create', 'Update']:                                                   
                      for log_group in log_group_config:                      
                          log_group_name = log_group['LogGroupName']
                          if log_group_name in invalid_log_groups:
                              logger.info(f'Log group {log_group_name} is invalid. Skipping...')
                              continue
                          filter_pattern = log_group['FilterPattern'] if 'FilterPattern' in log_group else ''
                          cloudwatch_firehose_role_arn = os.environ['CloudWatchFirehoseRoleArn']

                          response = log_client.put_subscription_filter(
                            logGroupName=log_group_name,
                            roleArn=cloudwatch_firehose_role_arn,
                            filterName=filter_name,
                            filterPattern=filter_pattern,
                            destinationArn=firehose_arn
                          )

                  elif request_type == 'Delete':
                      try:
                          for log_group in log_group_config:
                              log_group_name = log_group['LogGroupName']
                              if log_group_name in invalid_log_groups:
                                  logger.info(f'Log group {log_group_name} is invalid. Skipping...')
                                  continue
                              
                              response = log_client.delete_subscription_filter(
                                logGroupName=log_group_name,
                                filterName=filter_name
                              )
                      except Exception as e:
                          logger.error(f'Delete failed for the log group subscription filters with error: {str(e)}')
                          cfnresponse.send(event, context, cfnresponse.SUCCESS, response)
          
                  cfnresponse.send(event, context, cfnresponse.SUCCESS, response)
              except Exception as e:
                  logger.error(f'Error: {str(e)}')
                  cfnresponse.send(event, context, cfnresponse.FAILED, {}, reason=f'{str(e)}')
      Handler: index.lambda_handler
      Role: !GetAtt NewRelicLogsCloudWatchFirehoseLambdaIAMRole.Arn
      Runtime: python3.12
      Timeout: 120