firehose-template.yaml

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: New Relic AWS Unified Firehose Log Forwarder
Mappings:
  NewRelicDatacenterMap:
    Datacenter:
      US: 'https://aws-api.newrelic.com/firehose/v1'
      EU: 'https://aws-api.eu.newrelic.com/firehose/v1' # TODO update fedramp and govcloud urls.
Metadata:
  AWS::ServerlessRepo::Application:
    Name: NewRelic-Firehose-Log-Forwarding
    Description: Send log data from AWS to New Relic.
    Author: New Relic
    SpdxLicenseId: Apache-2.0
    LicenseUrl: LICENSE.txt
    ReadmeUrl: README.md
    HomePageUrl: https://github.com/newrelic/aws-unified-firehose-logging # TODO update this when the repo is public.
    SemanticVersion: 0.0.17
    SourceCodeUrl: https://github.com/newrelic/aws-unified-firehose-logging # TODO update this when the repo is public.
  
  AWS::CloudFormation::Interface:
    ParameterLabels:
      NewRelicRegion:
        default: 'New Relic Datacenter Region'
      LicenseKey:
        default: 'New Relic Ingest License Key'
      LogGroupConfig:
        default: 'Log Groups name & filter (if applicable)'        
      LoggingFirehoseStreamName:
        default: 'Name of Logging Firehose Delivery Stream'
      LoggingS3BackupBucketName:
        default: 'S3 Bucket Destination for failed Logging events'
      EnableCloudWatchLoggingForFirehose:
        default: 'Enables CloudWatch logging for Logging Firehose stream'
      CommonAttributes:
        default: 'Common Attributes to be added to the log events'
      StoreNRLicenseKeyInSecretManager:
        default: 'Store New Relic License Key in AWS Secrets Manager'

Parameters:
  LicenseKey:
    Type: String
    Description: New relic license key, create or get and existing key in https://one.newrelic.com/launcher/api-keys-ui.api-keys-launcher
    NoEcho: true
    MaxLength: 40
    AllowedPattern: "^[^\\s]+$"
    ConstraintDescription: "The parameter value cannot be empty, contain spaces, and must be alphanumeric and can contain symbols."
  NewRelicRegion:
    Type: String
    Description: Datacenter where the data will be sent (US/EU), DO NOT TOUCH
    Default: "US"
    AllowedValues:
      - "US"
      - "EU"
  NewRelicAccountId: 
    Type: String
    Description: Id of the account in New relic
    AllowedPattern: "[0-9]+"
    ConstraintDescription: must only contain numbers
  LogGroupConfig:
    Description: "JSON array of objects representing your LogGroup and Filters (if applicable). For example: [{\"LogGroupName\":\"logGroup1\",\"FilterPattern\":\"filter1\"}]"
    Type: String
  LoggingFirehoseStreamName:
    Type: String
    Description: Name of new Data Firehose Delivery Stream (must be unique per AWS account in the same AWS Region)
    Default: NewRelic-Logging-Delivery-Stream
    MaxLength: 64
    AllowedPattern: "[a-zA-Z0-9_.-]+"
    ConstraintDescription: must only contain letters (uppercase and lowercase), numbers, and characters '.', '_', and '-' with max length of 64 total characters
  LoggingS3BackupBucketName:
    Type: String
    Description: S3 Bucket Destination for failed events (must be globally unique across all AWS accounts in all AWS Regions within a partition)
    Default: firehose-logging-backup
    MinLength: 3
    MaxLength: 63
    AllowedPattern: (?!(^xn--|-s3alias$))^[a-z0-9.][a-z0-9-.]{1,61}[a-z0-9.]$
    ConstraintDescription: must adhere to the S3 bucket naming rules - https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
  EnableCloudWatchLoggingForFirehose:
    Type: String
    Description: Enables CloudWatch logging for the Firehose stream.
    Default: "false"
    AllowedValues:
      - "true"
      - "false"
  CommonAttributes:
    Type: String
    Description: "JSON array of objects representing your custom attributes. For example: [{\"AttributeName\":\"key1\",\"AttributeValue\":\"value1\"}]"
    Default: ""
  StoreNRLicenseKeyInSecretManager:
    Type: String
    Description: Should we store the New Relic license key in AWS Secrets Manager. Defaults to true.
    Default: "true"
    AllowedValues:
      - "true"
      - "false" 


Conditions:
  AddCloudwatchTrigger: !Not [ !Equals [!Ref LogGroupConfig , ""]]
  ShouldEnableCloudWatchLogging: !Equals [!Ref EnableCloudWatchLoggingForFirehose, "true"]
  ShouldCreateSecret: !Equals [ !Ref StoreNRLicenseKeyInSecretManager, "true" ]

Resources:

  NewRelicLogsLicenseKeySecret:
    Type: 'AWS::SecretsManager::Secret'
    Condition: ShouldCreateSecret
    Properties:
      Description: The New Relic license key, for sending telemetry
      Name : !Join ['-', ['nr-license-key', !Select [4, !Split ['-', !Select [2, !Split ['/', !Ref AWS::StackId]]]]]]
      SecretString: !Sub '{ "api_key": "${LicenseKey}"}'

  NewRelicLogsS3FirehoseEventsBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Join ['-', [!Ref LoggingS3BackupBucketName, !Select [0, !Split ['-', !Select [2, !Split ['/', !Ref 'AWS::StackId' ]]]]]]
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true

  NewRelicLogsFirehoseRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - firehose.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: /service-role/
      Policies:
        - PolicyName: Firehose-S3Access # https://docs.aws.amazon.com/firehose/latest/dev/controlling-access.html#using-iam-s3
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 's3:AbortMultipartUpload'
                  - 's3:GetBucketLocation'
                  - 's3:GetObject'
                  - 's3:ListBucket'
                  - 's3:ListBucketMultipartUploads'
                  - 's3:PutObject'
                Resource:
                  - !GetAtt NewRelicLogsS3FirehoseEventsBucket.Arn
                  - !Join ['', [!GetAtt NewRelicLogsS3FirehoseEventsBucket.Arn, '/*']]
        - PolicyName: Firehose-CloudWatchAccess
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - logs:PutLogEvents
                Resource: 'arn:aws:logs:*:*:*'
        - !If
          - ShouldCreateSecret
          - PolicyName: Firehose-SecretsManagerAccess
            PolicyDocument:
              Version: 2012-10-17
              Statement:
                - Effect: Allow
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: !Ref NewRelicLogsLicenseKeySecret
          - !Ref "AWS::NoValue"
      Description: Role to allow firehose stream to access resources including putting events into S3 backup bucket
      RoleName: !Join ['-', ['NewRelicLogsFirehoseRole', !Select [0, !Split ['-', !Select [2, !Split ['/', !Ref 'AWS::StackId' ]]]]]]
  
  NewRelicLogsFirehoseErrorLogGroup:
    Type: AWS::Logs::LogGroup
    Condition: ShouldEnableCloudWatchLogging
    Properties:
      LogGroupName: !Join ['/', ['/aws/datafirehose-newrelic/', !Ref 'AWS::StackName', 'FirehoseError']]
  
  NewRelicLogsFirehoseLogStream:
    Type: AWS::Logs::LogStream
    Condition: ShouldEnableCloudWatchLogging
    Properties:
      LogGroupName: !Ref NewRelicLogsFirehoseErrorLogGroup
      LogStreamName: "FirehoseErrorLogStream"
  
  
  NewRelicLogsUserInputParserLambdaIAMRole:
    Type: "AWS::IAM::Role"
    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:DescribeLogGroups'
                  - 'logs:CreateLogGroup'
                  - 'logs:CreateLogStream'
                  - 'logs:PutLogEvents'
                Resource: 'arn:aws:logs:*:*:*'

  NewRelicLogsUserInputParserLambda:
    Type: 'AWS::Serverless::Function'
    Properties: 
      InlineCode: |
          import json
          import cfnresponse
          import logging
          import boto3
          logger = logging.getLogger()
          logger.setLevel(logging.INFO)
          logs_client = boto3.client('logs')
          
          def handler(event, context):
              response = {}
              try:
                  if event['RequestType'] == 'Delete':
                      cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
                      return {}
                  else:
                      event_data = event['ResourceProperties']
                      region = event_data['AWS_REGION']
                      account_id = event_data['AWS_ACCOUNT_ID']
                      awsRealm = event_data['AWS_REALM']
                      common_attributes_str = event_data['CommonAttributes']
                      isAttributesValid = True
                      isLogGroupValid = True

                      try:
                          if common_attributes_str.strip(): 
                              attributes = json.loads(common_attributes_str)
                              if not isinstance(attributes, list):
                                  raise ValueError('CommonAttributes must be a JSON array')
                              for attribute in attributes:
                                  if not isinstance(attribute, dict):
                                      raise ValueError("Each element in CommonAttributes should be a JSON object.")
                                  if 'AttributeName' not in attribute or 'AttributeValue' not in attribute:
                                      raise ValueError("Each element in CommonAttributes should have 'AttributeName' and 'AttributeValue' keys.")
                                  if not attribute['AttributeName'] or not attribute['AttributeValue']:
                                      raise ValueError("Each element in CommonAttributes should have non-empty 'AttributeName' and 'AttributeValue' values.")

                          response['UserInputCommonAttributesErrorMessages'] = 'No Errors Found in User Input for setting up custom attributes.' 
                      except Exception as e:
                          logger.error(f'CommonAttributes provided {common_attributes_str} is not a valid JSON, the error is:  {str(e)}')
                          isAttributesValid = False
                          response['UserInputCommonAttributesErrorMessages'] = (
                              'Validation Failed for Input Provided. The CommonAttributes provided : {} is not a valid JSON. '
                              'Please provide a valid JSON for CommonAttributes.'.format(common_attributes_str)
                          )

                      try:
                          log_group_config = event_data['LogGroupConfig']
                          if log_group_config.strip():
                              log_group_config_json = json.loads(log_group_config)
                              if not isinstance(log_group_config_json, list):
                                  raise ValueError('LogGroupConfig must be a JSON array')
                              for log_group in log_group_config_json:
                                  if not isinstance(log_group, dict):
                                      raise ValueError("Each element in LogGroupConfig should be a JSON object.")
                                  if 'LogGroupName' not in log_group:
                                      raise ValueError("Each element in LogGroupConfig should have 'LogGroupName' key.")
                                  if not log_group['LogGroupName']:
                                      raise ValueError("Each element in LogGroupConfig should have non-empty 'LogGroupName' value.")
                                      
                          response['LogGroupErrorMessages'] = 'No Errors Found in User Input for Log Group'
                      except Exception as e:
                          logger.error(f'LogGroup provided {log_group_config} is not a valid JSON, the error is:  {str(e)}')
                          isLogGroupValid = False
                          response['LogGroupErrorMessages'] = (
                              'Validation Failed for Input Provided. The LogGroup provided : {} is not a valid JSON. '
                              'Please provide a valid JSON for LogGroup.'.format(log_group_config)
                          )
                      
                      if(not isAttributesValid or not isLogGroupValid):
                          response['LogGroupArns'] = ''
                          response['InvalidLogGroups'] = ''
                          response['CommonAttributes']= []
                          cfnresponse.send(event, context, cfnresponse.SUCCESS, response)
                          return

                      # these parameter are needed for entity synthesis
                      additional_attributes = [
                         {"AttributeName": "aws.accountId", "AttributeValue": account_id},
                         {"AttributeName": "aws.region", "AttributeValue": region},
                         {"AttributeName": "instrumentation.provider", "AttributeValue": "aws"},
                         {"AttributeName": "instrumentation.name", "AttributeValue": "firehose"},
                         {"AttributeName": "instrumentation.version", "AttributeValue": "1.0.0"},
                         {"AttributeName": "aws.realm", "AttributeValue": awsRealm}
                      ]

                      
                      # Convert the json to the correct json format
                      if common_attributes_str.strip():
                          common_attributes = json.loads(common_attributes_str)
                      else:
                          common_attributes = []

                      # Filter out AttributeNames that are needed for entity synthesis from common_attributes
                      existing_attribute_names = {attr['AttributeName'] for attr in additional_attributes}
                      common_attributes = [
                          attr for attr in common_attributes if attr['AttributeName'] not in existing_attribute_names
                      ]
                      
                      common_attributes.extend(additional_attributes)

                      log_group_config_json = json.loads(event_data.get('LogGroupConfig', '[]').strip() or '[]')

                      # Create the log group arns from the log group json if given log group exists
                      log_group_arns = []
                      invalid_log_groups = []
                      for log_group in log_group_config_json:
                          log_group_name = log_group['LogGroupName']
                          logger.info(f'Describing log group {log_group_name}')
                          # Matches based on prefix, results in a list of log groups. Processes the list further to find the exact match.
                          responseLogGroup = logs_client.describe_log_groups(logGroupNamePrefix=log_group_name) 
                          doesLogGroupExist = False
                          # Check if the log group with the exact name exists in the list of log groups
                          for logGroup in responseLogGroup.get('logGroups', []):
                            if logGroup['logGroupName'] == log_group_name:
                              doesLogGroupExist = True
                              log_group_arn = logGroup['arn']
                              log_group_arns.append(log_group_arn)
                              break
                          if not doesLogGroupExist:
                            logger.error(f'No log group exists with the name: {log_group_name}')
                            invalid_log_groups.append(log_group_name)
                      response['CommonAttributes'] = common_attributes
                      response['LogGroupArns'] = ','.join(log_group_arns)
                      response['InvalidLogGroups'] = ','.join(invalid_log_groups)
                      response['LogGroupErrorMessages'] = "No Errors Found in User Input for Log Group"
                      if invalid_log_groups:
                            response['LogGroupErrorMessages'] = (
                              'Validation Failed for Input Provided. These Log Groups: [{}] do not exist in your account.'
                              'Please setup Cloudwatch to Data Firehose subscription manually for additional log groups including these failed ones to stream with the resource Logical ID: "LoggingFirehoseStreamToNewRelic".'
                              'While setting up the subscription manuually you can use the IAM role with resource Logical ID: "CloudWatchFirehoseRole" created by this deployment.'.format(','.join(invalid_log_groups))
                            )

                      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.handler
      Runtime: python3.12
      Role: !GetAtt NewRelicLogsUserInputParserLambdaIAMRole.Arn
      Timeout: 120
      MemorySize: 128

  NewRelicLogsResourceForUserInputParsing:
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt NewRelicLogsUserInputParserLambda.Arn
      LogGroupConfig: !Ref LogGroupConfig
      CommonAttributes: !Ref CommonAttributes
      AWS_ACCOUNT_ID: !Ref AWS::AccountId
      AWS_REGION: !Ref AWS::Region
      AWS_REALM: !Ref AWS::Partition

  NewRelicLogsFirehoseStreamToNewRelic:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: !Join ['-', [!Ref LoggingFirehoseStreamName, !Select [0, !Split ['-', !Select [2, !Split ['/', !Ref 'AWS::StackId' ]]]]]]
      DeliveryStreamType: DirectPut
      HttpEndpointDestinationConfiguration:
        RequestConfiguration:
          ContentEncoding: GZIP
          # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-kinesisfirehose-deliverystream-httpendpointrequestconfiguration.html
          CommonAttributes: !GetAtt NewRelicLogsResourceForUserInputParsing.CommonAttributes
        EndpointConfiguration:
          Name: New Relic
          Url: !FindInMap [NewRelicDatacenterMap, Datacenter, !Ref NewRelicRegion]
          AccessKey: !If [ShouldCreateSecret, !Ref "AWS::NoValue",!Ref LicenseKey]
        SecretsManagerConfiguration:
          Enabled: !If [ShouldCreateSecret, true, false] 
          SecretARN: !If [ShouldCreateSecret, !Ref NewRelicLogsLicenseKeySecret, !Ref "AWS::NoValue"]
        BufferingHints:
          IntervalInSeconds: 60
          SizeInMBs: 1
        RetryOptions:
          DurationInSeconds: 60
        S3Configuration:
          CompressionFormat: GZIP
          BucketARN: !GetAtt NewRelicLogsS3FirehoseEventsBucket.Arn
          RoleARN: !GetAtt NewRelicLogsFirehoseRole.Arn
        CloudWatchLoggingOptions: 
          !If 
            - ShouldEnableCloudWatchLogging
            - 
              Enabled: true
              LogGroupName: !Ref NewRelicLogsFirehoseErrorLogGroup
              LogStreamName: !Ref NewRelicLogsFirehoseLogStream
            - 
              Ref: "AWS::NoValue"
        RoleARN: !GetAtt NewRelicLogsFirehoseRole.Arn
  
  NewRelicLogsCloudWatchFirehoseTrigger:
    Type: 'AWS::CloudFormation::Stack'
    Condition: AddCloudwatchTrigger
    Properties:
      TemplateURL: firehose-cloudwatch-trigger-stack.yaml
      Parameters:
        LogGroupConfig: !Ref LogGroupConfig
        LogGroupArns: !GetAtt NewRelicLogsResourceForUserInputParsing.LogGroupArns
        InvalidLogGroups: !GetAtt NewRelicLogsResourceForUserInputParsing.InvalidLogGroups
        LoggingFirehoseStreamArn: !GetAtt NewRelicLogsFirehoseStreamToNewRelic.Arn

Outputs:
  NewRelicLogsLoggingFirehoseStreamArn:
    Description: The ARN of the Logging DataFirehose Stream.
    Value: !GetAtt NewRelicLogsFirehoseStreamToNewRelic.Arn
  NewRelicLogsUserInputLogGroupErrors:
    Description: Contains Details about Errors in User Input for LogGroup.
    Value: !GetAtt NewRelicLogsResourceForUserInputParsing.LogGroupErrorMessages
  NewRelicLogsUserInputCommonAttributeErrors:
    Description: Contains Details about Errors in User Input for setting up Common Attributes in lambda.
    Value: !GetAtt NewRelicLogsResourceForUserInputParsing.UserInputCommonAttributesErrorMessages