template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Description: |
  Send logs to Coralogix from AWS (S3, Cloudtrail, Cloudwatch, msk, SNS, SQS, Kinesis and more)
  Please report any issues to: github.com/coralogix/coralogix-aws-shipper/issues
Metadata:
  AWS::ServerlessRepo::Application:
    Name: coralogix-aws-shipper
    Description: Send logs to Coralogix from AWS (S3, Cloudtrail, Cloudwatch, msk, SNS, SQS, Kinesis and more)
    Author: Coralogix
    SpdxLicenseId: Apache-2.0
    LicenseUrl: LICENSE
    ReadmeUrl: README.md
    # NOTE: there is a 10 Label limit for metadata in the AWS Serverless Repo metadata
    Labels:
      - coralogix
      - logs
      - sns
      - s3
      - cloudwatch
      - cloudtrail
      - vpc
      - sqs
      - kinesis
      - cloudfront
    HomePageUrl: https://coralogix.com
    SemanticVersion: 1.0.7
    SourceCodeUrl: https://github.com/coralogix/coralogix-aws-shipper

  AWS::CloudFormation::Interface:
    ParameterGroups:
      - Label:
          default: Coralogix configuration
        Parameters:
          - IntegrationType
          - CoralogixRegion
          - CustomDomain
          - ApplicationName
          - SubsystemName
          - ApiKey
          - StoreAPIKeyInSecretsManager
      - Label:
          default: S3/CloudTrail/VpcFlow/S3Csv configuration
        Parameters:
          - S3BucketName
          - S3KeyPrefix
          - S3KeySuffix
          - NewlinePattern
          - SNSTopicArn
          - SQSTopicArn
          - CSVDelimiter
      - Label:
          default: Kafka & MSK configuration
        Parameters:
          - KafkaTopic
          - KafkaBrokers
          - KafkaSubnets
          - KafkaSecurityGroups
          - MSKClusterArn
          - KafkaBatchSize
      - Label:
          default: Cloudwatch configuration
        Parameters:
          - CloudWatchLogGroupName
          - CloudWatchLogGroupPrefix
      - Label:
          default: SNS configuration
        Parameters:
          - SNSIntegrationTopicArn
      - Label:
          default: SQS configuration
        Parameters:
          - SQSIntegrationTopicArn
      - Label:
          default: Kinesis configuration
        Parameters:
          - KinesisStreamArn
      - Label:
          default: Generic Config
        Parameters:
          - NotificationEmail
          - BlockingPattern
          - SamplingRate
          - AddMetadata
          - CustomMetadata
      - Label:
          default: Lambda configuration
        Parameters:
          - FunctionMemorySize
          - FunctionTimeout
          - LogLevel
          - LambdaLogRetention
      - Label:
          default: VPC configuration Optional
        Parameters:
          - LambdaSubnetID
          - LambdaSecurityGroupID
          - UsePrivateLink

Parameters:
  CoralogixRegion:
    Type: String
    Description: |
      The Coralogix location region, possible options are [EU1, EU2, AP1, AP2, US1, US2, Custom]
      If this value is set to Custom you must specify the Custom Domain to use via the CustomDomain parameter.
    AllowedValues:
      - EU1
      - EU2
      - AP1
      - AP2
      - US1
      - US2
      - Custom
    Default: Custom
  
  CustomDomain:
    Type: String
    Description: The Custom Domain. If set, will be the domain used to send telemetry (e.g. cx123.coralogix.com)
    Default: ''

  ApiKey:
    Type: String
    Description: |
      Your Coralogix Send Your Data - API Key which is used to validate your authenticity (https://coralogix.com/docs/send-your-data-api-key/)
      This value can be a Coralogix API Key or an AWS Secret Manager ARN that holds the API Key
    MinLength: 1
    NoEcho: true

  ApplicationName:
    Type: String
    Description: The name of your application (https://coralogix.com/docs/application-and-subsystem-names/)
    MinLength: 1

  SubsystemName:
    Type: String
    Description: The subsystem name of your application (https://coralogix.com/docs/application-and-subsystem-names/)
    Default: ''

  NewlinePattern:
    Type: String
    Description: Regular expression to detect a new log line for multiline logs from S3 source, e.g., use expression \n(?=\d{2}\-\d{2}\s\d{2}\:\d{2}\:\d{2}\.\d{3})
    Default: ''
  
  AddMetadata:
    Type: String
    Description: |
      Add metadata to the log message. Expects comma separated values. Options are bucket_name,key_name,stream_name
    Default: ''
  CustomMetadata:
    Type: String
    Description: |
      Add custom metadata to the log message. Expects comma separated values. Options are key1=value1,key2=value2
    Default: ''
  BlockingPattern:
    Type: String
    Description: Regular expression to detect lines that should be excluded from sent to Coralogix
    Default: ''

  SamplingRate:
    Type: Number
    Description: Send messages with specific rate (1 out of N) e.g., put the value 10 if you want to send every 10th log 
    MinValue: 1
    Default: 1

  LogLevel:
    Type: String
    Description: 'Log level for the Lambda function. Can be one of: INFO, WARN, ERROR, DEBUG'
    AllowedValues:
      - "INFO"
      - "WARN"
      - "ERROR"
      - "DEBUG"
    Default: "WARN"

  S3BucketName:
    Type: String
    Description: |
      The name of the AWS S3 bucket to watch
    AllowedPattern: '^[0-9A-Za-z\.\-_]*(?<!\.)$'
    Default: ""
    MaxLength: 63

  CSVDelimiter:
    Type: String
    Description: |
      The delimiter used in the CSV file to process
      This value is applied when the S3Csv integration type is selected
    MaxLength: 1
    Default: ','

  S3KeyPrefix:
    Type: String
    Description: |
      The AWS S3 path prefix to watch. This value is ignored
      when the SNSTopicArn / SQSTopicArn parameter is provided.
    MaxLength: 1024
    Default: ''

  S3KeySuffix:
    Type: String
    Description: |
      The AWS S3 path suffix to watch. This value is ignored
      when the SNSTopicArn parameter is provided.
    MaxLength: 1024
    Default: ''

  FunctionMemorySize:
    Type: Number
    Description: Memory size for lambda function
    MinValue: 128
    MaxValue: 10240
    Default: 1024

  FunctionTimeout:
    Type: Number
    Description: Timeout for the lambda function
    MinValue: 30
    MaxValue: 900
    Default: 300
  
  SNSTopicArn:
    Type: String
    Description: The ARN for the SNS topic that contains the SNS subscription responsible for retrieving logs from Amazon S3
    Default: ''
  SQSTopicArn:
    Type: String
    Description: The ARN for the SQS topic that contains the SQS subscription responsible for retrieving logs from Amazon S3
    Default: ''
  SQSIntegrationTopicArn:
    Type: String
    Description: The ARN of SQS topic to subscribe to retrieving messages
    Default: ''
  SNSIntegrationTopicArn:
    Type: String
    Description: The ARN of SNS topic to subscribe to retrieving messages
    Default: ''
  KinesisStreamArn:
    Type: String
    Description: The ARN of Kinesis stream to subscribe to retrieving messages
    Default: ''
  IntegrationType:
    Type: String
    Description: 'The integration type. Can be one of: S3, CloudTrail, VpcFlow, CloudWatch, S3Csv, Sns, Sqs, Kinesis, CloudFront, Kafka, MSK, EcrScan'
    AllowedValues:
      - S3
      - CloudTrail
      - CloudWatch
      - VpcFlow
      - S3Csv
      - Sns
      - Sqs
      - Kinesis
      - CloudFront
      - Kafka
      - MSK
      - EcrScan
    Default: S3

  CloudWatchLogGroupName:
    Type: String
    Description: 'A comma separated list of CloudWatch log groups names to watch  e.g, (log-group1,log-group2,log-group3)'
    Default: ''

  CloudWatchLogGroupPrefix:
    Type: String
    Description: 'Prefix of the CloudWatch log groups that will trigger the lambda, in case that your log groups are log-group1, log-group2, log-group3 then you can set the value to `log-group`. When using this variable you will not be able to see the log groups as trigger for the lambda. The parameter dose not replace CloudWatchLogGroupName parameter'
    Default: ''

  LambdaLogRetention:
    Type: Number
    Description: CloudWatch log retention days for logs generated by the Lambda function
    MinValue: 1
    Default: 5

  NotificationEmail:
    Type: String
    Description: Failure notification email address 
    MaxLength: 320
    Default: ''

  StoreAPIKeyInSecretsManager:
    Type: String
    Description: |
      Store the API key in AWS Secrets Manager. ApiKeys are stored in secret manager
      by default. If this option is set to false, the ApiKey will apeear in plain text as an 
      environment variable in the lambda function console.
    AllowedValues:
      - true
      - false
    Default: true

  LambdaSubnetID:
    Type: String
    Description: ID of Subnet into which to deploy the integration
    Default: ''

  LambdaSecurityGroupID:
    Type: String
    Description: ID of the SecurityGroup into which to deploy the integration
    Default: ''

  UsePrivateLink:
    Type: String
    Description: Will you be using our PrivateLink?
    AllowedValues:
      - true
      - false
    Default: false
  
  MSKClusterArn:
    Type: String
    Description: The ARN of the MSK cluster to subscribe to retrieving messages
    Default: ''

  KafkaTopic:
    Type: String
    Description: The name of the Kafka topic to subscribe to retrieving messages
    Default: ''

  KafkaBatchSize:
    Type: Number
    Description: The maximum number of records to retrieve per batch from Kafka
    MinValue: 1
    MaxValue: 10000
    Default: 100

  KafkaBrokers:
    Type: CommaDelimitedList
    Description: |
      Comma-delimited list of host and port pair addresses of your Kafka brokers
      [Not Required when using MSK]
    Default: ''
  
  KafkaSubnets:
    Type: CommaDelimitedList
    Description: |
      The subnets associated with your VPC for each Kafka broker
      [Not Required when using MSK]
    Default: ""

  KafkaSecurityGroups:
    Type: CommaDelimitedList
    Description: |
      The security groups associated with your VPC for each Kafka broker
      [Not Required when using MSK]
    Default: ""

  EnableDLQ:
    Type: String
    Description: Enable Dead Letter Queue for the Lambda function
    AllowedValues:
      - true
      - false
    Default: false

  DLQRetryLimit:
    Type: Number
    Description: The maximum number of times to retry the function execution in case of failure
    MinValue: 0
    MaxValue: 5
    Default: 3

  DLQRetryDelay:
    Type: Number
    Description: The delay in seconds between retries
    MinValue: 0
    MaxValue: 900
    Default: 900
  
  DLQS3Bucket:
    Type: String
    Description: The S3 bucket to store the DLQ failed messages after retry limit is reached
    Default: ''

Mappings:
  CoralogixRegionMap:
    EU1:
      Domain: coralogix.com
    EU2:
      Domain: eu2.coralogix.com
    AP1:
      Domain: coralogix.in
    AP2:
      Domain: coralogixsg.com
    US1:
      Domain: coralogix.us
    US2:
      Domain: cx498.coralogix.com
    Custom:
      Domain: ""

Conditions:
  # DLQ conditions
  DLQEnabled: !Equals [ !Ref EnableDLQ, 'true' ]
  DLQEnabledAndUseDefault: !And
    - !Condition DLQEnabled
    - !Condition UseDefault
  
  DLQEnabledAndUseSNSTopicARN: !And
    - !Condition DLQEnabled
    - !Condition UseSNSTopicARN

  DLQEnabledAndUseSQSTopicARN: !And
    - !Condition DLQEnabled
    - !Condition UseSQSTopicARN

  DLQEnabledAndUseKinesisStreamARN: !And
    - !Condition DLQEnabled
    - !Condition UseKinesisStreamARN

  DLQEnabledAndUseKafkaTopicARN: !And
    - !Condition DLQEnabled
    - !Condition UseKafka
  
  DLQEnabledAndUseMSK: !And
    - !Condition DLQEnabled
    - !Condition UseMSK

  DLQEnabledAndUseECRScan: !And
    - !Condition DLQEnabled
    - !Condition UseECRScan

  DLQEnabledAndUseCloudwatchLogs: !And
    - !Condition DLQEnabled
    - !Condition UseCloudwatchLogs

  IsKafkaIntegration: !Equals [ !Ref IntegrationType, 'Kafka' ]
  BlockPatternNotSet: !Equals [ !Ref BlockingPattern, '' ]
  CSVDelimiterUse: !Equals [!Ref IntegrationType, 'S3Csv']
  NewlinePatternNotSet: !Equals [ !Ref NewlinePattern, '' ]
  AddMetadataNotSet: !Equals [ !Ref AddMetadata, '']
  CustomMetadataNotSet: !Equals [ !Ref CustomMetadata, '']
  IsSNSIntegration: !Equals [ !Ref IntegrationType, 'Sns' ]
  UseECRScan: !Equals [ !Ref IntegrationType, 'EcrScan' ]
  IsSQSIntegration: !Equals [ !Ref IntegrationType, 'Sqs' ]
  IsKinesisIntegration: !Equals [ !Ref IntegrationType, 'Kinesis' ]
  IsNotificationEnabled: !Not [ !Equals [ !Ref NotificationEmail, '' ] ]
  IsCustomDomain: !Equals [ !Ref CoralogixRegion, Custom ]
  S3KeyPrefixIsSet: !Not [ !Equals [ !Ref S3KeyPrefix, '' ] ]
  S3SuffixIsSet: !Not [ !Equals [ !Ref S3KeySuffix, '' ] ]
  IsApiKeyNotArn: !Equals [!Ref ApiKey , !Select [0,!Split [":" , !Ref ApiKey]]]
  ApiKeyIsArn: !Not [!Condition IsApiKeyNotArn]
  UseCloudwatchLogsWithSecretPolicy: !And
    - !Condition UseCloudwatchLogs
    - !Or
      - !Condition StoreAPIKeyInSecretsManager
      - !Condition ApiKeyIsArn
  UseAWSDefaultPrefix: !Or 
    - !Equals [ !Ref IntegrationType, 'VpcFlow' ]
    - !Equals [ !Ref IntegrationType, 'CloudTrail' ]
  UseAWSDefaultVpcFlowSuffix: !Equals [!Ref IntegrationType, 'VpcFlow']
  UseAWSDefaultCloudTrailSuffix: !Equals [!Ref IntegrationType, 'CloudTrail']
  StoreAPIKeyInSecretsManager: !And
    - !Equals [ !Ref StoreAPIKeyInSecretsManager, 'true' ]
    - !Condition IsApiKeyNotArn
  IsPrivateLink: !Equals [ !Ref UsePrivateLink, 'true' ]
  UseSQSTopicARN: 
    Fn::Or: 
    - !And 
      - !Not [ !Equals [ !Ref SQSTopicArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
    - !And 
      - !Not [ !Equals [ !Ref SQSIntegrationTopicArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
  UseKinesisStreamARN: !And 
      - !Not [ !Equals [ !Ref KinesisStreamArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
  UseSNSTopicARN: 
    Fn::Or: 
    - !And 
      - !Not [ !Equals [ !Ref SNSTopicArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
    - !And 
      - !Not [ !Equals [ !Ref SNSIntegrationTopicArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
  UseVpcConfig: !And 
    - !Not [ !Equals [ !Ref LambdaSubnetID, '' ] ]
    - !Not [ !Equals [ !Ref LambdaSecurityGroupID, '' ] ]
  UseCloudwatchLogs: !And
    - !Not [ !Equals [ !Ref CloudWatchLogGroupName, '' ] ]
    - !Not [ !Condition UseSNSTopicARN ] 
  UseDefault: !And 
    - !Not [ !Condition UseCloudwatchLogs ]
    - !Not [ !Condition UseSNSTopicARN  ]
    - !Not [ !Condition UseSQSTopicARN  ]
    - !Not [ !Condition UseKinesisStreamARN  ]
    - !Not [ !Condition UseMSK ]
    - !Not [ !Condition UseKafka ]
    - !Not [ !Condition UseECRScan]

  UseDefaultWithNotification: !And
    - !Condition UseDefault
    - !Condition IsNotificationEnabled
  UseCloudwatchLogsWithNotification: !And
    - !Condition UseCloudwatchLogs
    - !Condition IsNotificationEnabled
  UseKinesisTopicARNWithNotification: !And
    - !Condition UseKinesisStreamARN
    - !Condition IsNotificationEnabled
  UseSQSTopicARNWithNotification: !And
    - !Condition UseSQSTopicARN
    - !Condition IsNotificationEnabled
  UseSNSTopicARNWithNotification: !And
    - !Condition UseSNSTopicARN
    - !Condition IsNotificationEnabled
  UseMSK: !And
    - !Not [ !Equals [ !Ref MSKClusterArn, '' ] ]
    - !Not [ !Equals [ !Ref KafkaTopic, '' ] ]
    - !Equals [ !Ref IntegrationType, 'MSK' ]
  UseKafka: !And
    - !Not [ !Equals [ !Ref KafkaTopic, '' ] ]
    - !Equals [ !Ref MSKClusterArn, '' ]
    - !Equals [ !Ref IntegrationType, 'Kafka' ]

  UseMSKWithNotification: !And
    - !Condition UseMSK
    - !Condition IsNotificationEnabled

  UseKatkaTopicARNWithNotification: !And
    - !Condition IsKafkaIntegration
    - !Condition IsNotificationEnabled
    - !Condition UseKafka

  UseECRScanWithNotification: !And
    - !Condition UseECRScan
    - !Condition IsNotificationEnabled

Rules:
  ValidateDLQ:
    RuleCondition: !Equals [ !Ref EnableDLQ, 'true' ]
    Assertions:
      - Assert: !Not [ !Equals [ !Ref DLQS3Bucket, '' ] ]
        AssertDescription: DLQS3Bucket must be set when EnableDLQ is set to true
  ValidateCloudWatchLogs:
    RuleCondition: !And 
      - !Not [ !Equals [ !Ref CloudWatchLogGroupName, '' ] ]
      - !Equals [ !Ref SNSTopicArn, '' ]
      - !Equals [ !Ref SQSTopicArn, '' ]
      - !Equals [ !Ref KinesisStreamArn, '' ]
    Assertions:
      - Assert: !Not [ !Equals [ !Ref CloudWatchLogGroupName, '' ] ]
        AssertDescription: CloudWatchLogGroupName must be set when IntegrationType is set to cloudwatch
      - Assert: !Equals [ !Ref S3BucketName, '' ]
        AssertDescription: S3Bucket parameter must be empty when IntegrationType is set to cloudwatch 

  ValidateCustomDomain:
    RuleCondition: !Equals [ !Ref CoralogixRegion, Custom ] 
    Assertions:
      - Assert: !Not [ !Equals [ !Ref CustomDomain, '' ] ]
        AssertDescription: CustomDomain must be set when CoralogixRegion is set to Custom
  ValidateSQSTopicArnParam:
    RuleCondition: !And
      - !Not [ !Equals [ !Ref SQSTopicArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
    Assertions:
      - Assert: !And 
          - !Equals [ !Ref S3KeyPrefix, '' ]
          - !Equals [ !Ref S3KeySuffix, '' ]
        AssertDescription: S3Prefix and S3Suffix must be empty when SQSTopicArn is set

  ValidateSNSTopicArnParam:
    RuleCondition: !And
      - !Not [ !Equals [ !Ref SNSTopicArn, '' ] ]
      - !Equals [ !Ref CloudWatchLogGroupName, '' ]
    Assertions:
      - Assert: !And 
          - !Equals [ !Ref S3KeyPrefix, '' ]
          - !Equals [ !Ref S3KeySuffix, '' ]
        AssertDescription: S3Prefix and S3Suffix must be empty when SNSTopicArn is set

  ValidatePrivateLinkConfig:
    RuleCondition: !Equals [ !Ref UsePrivateLink, 'true']
    Assertions:
      - Assert: !And
        - !Not [ !Equals [ !Ref LambdaSubnetID, '' ] ]
        - !Not [ !Equals [ !Ref LambdaSecurityGroupID, '' ] ]
        AssertDescription: Subnet ID and Security Group ID must be set when UsePrivateLink is set to true

  ValidateKafkaIntegrationParams:
    RuleCondition: !Equals [ !Ref IntegrationType, 'Kafka' ]
    Assertions:
      - Assert: !Not [ !Equals [ !Ref KafkaTopic, '' ] ]
        AssertDescription: KafkaTopic must be set when IntegrationType is set to Kafka

  ValidateMSKIntegrationParams:
    RuleCondition: !Equals [ !Ref IntegrationType, 'MSK' ]
    Assertions:
      - Assert: !Not [ !Equals [ !Ref MSKClusterArn, '' ] ]
        AssertDescription: MSKClusterArn must be set when IntegrationType is set to MSK
      - Assert: !Not [ !Equals [ !Ref KafkaTopic, '' ] ]
        AssertDescription: KafkaTopic must be set when IntegrationType is set to MSK

Globals:
  Function:
    Description: Send logs to Coralogix.
    Handler: bootstrap
    Runtime: provided.al2023
    Architectures: 
      - arm64
    MemorySize: !Ref FunctionMemorySize
    Timeout: !Ref FunctionTimeout
    # EventInvokeConfig: !If
    #   - IsNotificationEnabled
    #   - !Ref AWS::NoValue
    #   - DestinationConfig:
    #       OnFailure:
    #         Type: SNS

    VpcConfig: !If
      - UseVpcConfig
      - SecurityGroupIds:
          - !Ref LambdaSecurityGroupID
        SubnetIds:
          - !Ref LambdaSubnetID
      - !Ref AWS::NoValue

    Environment:
      Variables:
        CORALOGIX_ENDPOINT: !If
          - IsCustomDomain
          - !If
            - IsPrivateLink
            - !Sub
              - https://ingress.private.${domain}
              - domain: !Ref CustomDomain
            - !Sub
              - https://ingress.${domain}
              - domain: !Ref CustomDomain
          - !If
            - IsPrivateLink
            - !Sub
              - https://ingress.private.${domain}
              - domain: !FindInMap [CoralogixRegionMap, !Ref CoralogixRegion , Domain]
            - !Sub
              - https://ingress.${domain}
              - domain: !FindInMap [CoralogixRegionMap, !Ref CoralogixRegion , Domain]
        CORALOGIX_API_KEY: !If
          - StoreAPIKeyInSecretsManager
          - !Ref Secret
          - !Ref ApiKey
        APP_NAME: !Ref ApplicationName
        SUB_NAME: !Ref SubsystemName
        SAMPLING: !Ref SamplingRate
        INTEGRATION_TYPE: !Ref IntegrationType
        RUST_LOG: !Ref LogLevel
        ADD_METADATA: !If
          - AddMetadataNotSet
          - !Ref AWS::NoValue
          - !Ref AddMetadata
        CUSTOM_METADATA: !If
          - CustomMetadataNotSet
          - !Ref AWS::NoValue
          - !Ref CustomMetadata
        NEWLINE_PATTERN: !If
          - NewlinePatternNotSet
          - !Ref AWS::NoValue
          - !Ref NewlinePattern
        BLOCKING_PATTERN: !If
          - BlockPatternNotSet
          - !Ref AWS::NoValue
          - !Ref BlockingPattern
        CSV_DELIMITER: !If
          - CSVDelimiterUse
          - !Ref CSVDelimiter
          - !Ref AWS::NoValue

        DLQ_RETRY_LIMIT: !Ref DLQRetryLimit
        DLQ_ARN: !If
          - DLQEnabled
          - !GetAtt DeadLetterQueue.Arn
          - !Ref AWS::NoValue
        DLQ_URL:
          Fn::If:
            - DLQEnabled
            - !Ref DeadLetterQueue
            - !Ref AWS::NoValue
        DLQ_S3_BUCKET: !If
          - DLQEnabled
          - !Ref DLQS3Bucket
          - !Ref AWS::NoValue
        

Resources:
  Secret:
    Type: 'AWS::SecretsManager::Secret'
    Condition: StoreAPIKeyInSecretsManager
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      Description: Coralogix API Key Secret
      SecretString: !Ref ApiKey
      # using StackId here as using the function name creates a circular dependency
      Name: !Sub 
        - 'lambda/coralogix/${AWS::Region}/coralogix-aws-shipper/${stack}'
        - stack: !Select
            - 1
            - !Split
                - "/"
                - !Select
                    - 5
                    - !Split
                        - ":"
                        - !Ref "AWS::StackId"          

  DeadLetterQueue:
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Condition: DLQEnabled
    Type: AWS::SQS::Queue
    Properties: 
      QueueName: "coralogix-aws-shipper-dlq"
      DelaySeconds: !Ref DLQRetryDelay
      MessageRetentionPeriod: 1209600
      VisibilityTimeout: !Ref FunctionTimeout

  ################################
  # ----- With Defaults -------  #
  ################################

  LambdaFunctionDefault:
    Condition: UseDefault
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS
      Policies:
        - S3ReadPolicy:
            BucketName: !Ref S3BucketName
        - !If
          - ApiKeyIsArn
          - Statement:
              - Effect: Allow
                Action:
                  - 'secretsmanager:GetSecretValue'
                Resource: !Ref ApiKey
          - !If
            - StoreAPIKeyInSecretsManager
            - Statement:
                - Effect: Allow
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: !Ref Secret
            # Note, this is a hack to get around the fact that you can't have a condition on a policy
            # If the Apikey is not an ARn or we are not storing the key in secrets manager, then we don't need access
            # to secrets manager
            - Statement:
                - Effect: Deny
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: '*'

  LambdaLogGroupDefault:
    Condition: UseDefault
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/lambda/${LambdaFunctionDefault}'
      RetentionInDays: !Ref LambdaLogRetention

  LambdaFunctionInvokePermissionDefault:
    Type: AWS::Lambda::Permission
    Condition: UseDefault
    Properties:
      FunctionName: !GetAtt [ LambdaFunctionDefault, Arn ]
      Action: lambda:InvokeFunction
      Principal: s3.amazonaws.com
      SourceAccount: !Ref AWS::AccountId
      SourceArn: !Sub "arn:aws:s3:::${S3BucketName}"

  LambdaTriggerDefault:
    Condition: UseDefault
    Type: Custom::LambdaTrigger
    DependsOn: LambdaFunctionInvokePermissionDefault
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ LambdaFunctionDefault, Arn ]
      Command: configure_s3_trigger
      Bucket: !Ref S3BucketName
      Suffix: !If
        - S3SuffixIsSet
        - !Ref S3KeySuffix
        - !If
          - UseAWSDefaultCloudTrailSuffix
          - '.json.gz'
          - !If
            - UseAWSDefaultVpcFlowSuffix
            - '.log.gz'
            - ""
      Prefix: !If
        - S3KeyPrefixIsSet
        - !Ref S3KeyPrefix
        - !If
          - UseAWSDefaultPrefix
          - 'AWSLogs/'
          - ''

  LambdaFunctionNotificationSubscriptionDefault:
    Condition: UseDefaultWithNotification
    Type: AWS::SNS::Subscription
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref  LambdaFunctionDefault.DestinationTopic  

  LambdaFunctionDefaultDLQ:
    Condition: DLQEnabledAndUseDefault
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionDefault', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq


  ################################
  # ----- With SNS TopiC ------- #
  ################################

  LambdaFunctionWithSNSTopic:
    Condition: UseSNSTopicARN
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - !If 
          - IsSNSIntegration
          - !Ref AWS::NoValue
          - S3ReadPolicy: 
              BucketName: !Ref S3BucketName
        - !If
          - ApiKeyIsArn
          - Statement:
              - Effect: Allow
                Action:
                  - 'secretsmanager:GetSecretValue'
                Resource: !Ref ApiKey
          - !If
            - StoreAPIKeyInSecretsManager
            - Statement:
                - Effect: Allow
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: !Ref Secret
            # Note, this is a hack to get around the fact that you can't have a condition on a policy
            # If the Apikey is not an ARn or we are not storing the key in secrets manager, then we don't need access
            # to secrets manager
            - Statement:
                - Effect: Deny
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: '*'
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS
      Events:
        SNSEvent:
          Type: SNS
          Properties:
            Topic: !If
              - IsSNSIntegration
              - !Ref SNSIntegrationTopicArn
              - !Ref SNSTopicArn

  LambdaLogGroupWithSNSTopic:
    Condition: UseSNSTopicARN
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/lambda/${LambdaFunctionWithSNSTopic}'
      RetentionInDays: !Ref LambdaLogRetention

  LambdaFunctionInvokePermissionWithSNSTopic:
    Type: AWS::Lambda::Permission
    Condition: UseSNSTopicARN
    Properties:
      FunctionName: !GetAtt [ LambdaFunctionWithSNSTopic, Arn ]
      Action: lambda:InvokeFunction
      Principal: s3.amazonaws.com
      SourceAccount: !Ref AWS::AccountId
      SourceArn: !Sub "arn:aws:s3:::${S3BucketName}"

  LambdaFunctionNotificationSubscriptionWithSNSTopic:
    Type: AWS::SNS::Subscription
    Condition: UseSNSTopicARNWithNotification
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref LambdaFunctionWithSNSTopic.DestinationTopic

  LambdaFunctionWithSNSTopicDLQ:
    Condition: DLQEnabledAndUseSNSTopicARN
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionWithSNSTopic', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq

  ################################
  # ----- With Cloudwatch -----  #
  ################################

  LambdaFunctionCloudwatchLogs:
    Condition: UseCloudwatchLogs
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - !If
          - ApiKeyIsArn
          - Statement:
              - Effect: Allow
                Action:
                  - 'secretsmanager:GetSecretValue'
                Resource: !Ref ApiKey
          - !If
            - StoreAPIKeyInSecretsManager
            - Statement:
                - Effect: Allow
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: !Ref Secret
            # Note, this is a hack to get around the fact that you can't have a condition on a policy
            # If the Apikey is not an ARn or we are not storing the key in secrets manager, then we don't need access
            # to secrets manager
            - Statement:
                - Effect: Deny
                  Action:
                    - 'secretsmanager:GetSecretValue'
                  Resource: '*'

        # - !If
        #   - ApiKeyIsArn
        #   - Statement:
        #       - Effect: Allow
        #         Action:
        #           - 'secretsmanager:GetSecretValue'
        #         Resource: !Ref ApiKey
        #   - !If
        #     - StoreAPIKeyInSecretsManager
        #     - Statement:
        #         - Effect: Allow
        #           Action:
        #             - 'secretsmanager:GetSecretValue'
        #           Resource: !Ref Secret
        #     - !Ref AWS::NoValue
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS

  LambdaFunctionCloudwatchLogsPolicies:
    Condition: UseCloudwatchLogsWithSecretPolicy
    Type: "AWS::IAM::ManagedPolicy"
    Properties:
      Path: "/"
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - "secretsmanager:GetSecretValue"
            Resource: !If
              - ApiKeyIsArn
              - !Ref ApiKey
              - !If
                - StoreAPIKeyInSecretsManager
                - !Ref Secret
                - !Ref AWS::NoValue

  LambdaLogGroupCloudwatchLogs:
    Condition: UseCloudwatchLogs
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/lambda/${LambdaFunctionCloudwatchLogs}'
      RetentionInDays: !Ref LambdaLogRetention

  LambdaFunctionNotificationSubscriptionCloudwatchLogs:
    Type: AWS::SNS::Subscription
    Condition: UseCloudwatchLogsWithNotification
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Sub LambdaFunctionCloudwatchLogs.DestinationTopic

  LambdaTriggerCloudwatchLogs:
    Condition: UseCloudwatchLogs
    Type: Custom::LambdaTrigger
    # DependsOn: LambdaFunctionInvokePermissionCloudwatchLogs
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionCloudwatchLogs', Arn ]
      CloudwatchGroup: !Ref CloudWatchLogGroupName
      CloudwatchPrefix: !Ref CloudWatchLogGroupPrefix
      Command: configure_cloudwatch_trigger

  # LambdaFunctionInvokePermissionCloudwatchLogs:
  #   Type: AWS::Lambda::Permission
  #   Condition: UseCloudwatchLogs
  #   Properties:
  #     FunctionName: !GetAtt [ 'LambdaFunctionCloudwatchLogs', Arn ]
  #     Action: lambda:InvokeFunction
  #     Principal: logs.amazonaws.com
  #     SourceAccount: !Ref AWS::AccountId
  #     SourceArn: !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:*:*"
  
  LambdaFunctionCloudwatchLogsDLQ:
    Condition: DLQEnabledAndUseCloudwatchLogs
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionCloudwatchLogs', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq

  ################################
  # ----- With SQS TopiC ------- #
  ################################

  LambdaFunctionWithSQSTopic:
    Condition: UseSQSTopicARN
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - !If 
          - IsSQSIntegration
          - !Ref AWS::NoValue
          - S3ReadPolicy: 
              BucketName: !Ref S3BucketName
        - Statement:
            - Effect: Allow
              Action:
                - 'secretsmanager:GetSecretValue'
              Resource: !Sub 'arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:*'
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !If
              - IsSQSIntegration
              - !Ref SQSIntegrationTopicArn
              - !Ref SQSTopicArn

  LambdaLogGroupWithSQSTopic:
    Condition: UseSQSTopicARN
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/lambda/${LambdaFunctionWithSQSTopic}'
      RetentionInDays: !Ref LambdaLogRetention

  LambdaFunctionInvokePermissionWithSQSTopic:
    Type: AWS::Lambda::Permission
    Condition: UseSQSTopicARN
    Properties:
      FunctionName: !GetAtt [ LambdaFunctionWithSQSTopic, Arn ]
      Action: lambda:InvokeFunction
      Principal: s3.amazonaws.com
      SourceAccount: !Ref AWS::AccountId
      SourceArn: !Sub "arn:aws:s3:::${S3BucketName}"

  LambdaFunctionNotificationSubscriptionWithSQSTopic:
    Type: AWS::SNS::Subscription
    Condition: UseSQSTopicARNWithNotification
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref LambdaFunctionWithSQSTopic.DestinationTopic

  LambdaFunctionWithSQSTopicDLQ:
    Condition: DLQEnabledAndUseSQSTopicARN
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionWithSQSTopic', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq

  ################################
  # ----- With Kinesis TopiC ------- #
  ################################

  LambdaFunctionWithKinesisTopic:
    Condition: UseKinesisStreamARN
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - !If 
          - IsKinesisIntegration
          - !Ref AWS::NoValue
          - S3ReadPolicy: 
              BucketName: !Ref S3BucketName
        - Statement:
            - Effect: Allow
              Action:
                - 'secretsmanager:GetSecretValue'
              Resource: !Sub 'arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:*'
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS
      Events:
        KinesisEvent:
          Type: Kinesis
          Properties:
            StartingPosition: LATEST
            Stream: !Ref KinesisStreamArn

  LambdaLogGroupWithKinesisTopic:
    Condition: UseKinesisStreamARN
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/lambda/${LambdaFunctionWithKinesisTopic}'
      RetentionInDays: !Ref LambdaLogRetention

  LambdaFunctionInvokePermissionWithKinesisTopic:
    Type: AWS::Lambda::Permission
    Condition: UseKinesisStreamARN
    Properties:
      FunctionName: !GetAtt [ LambdaFunctionWithKinesisTopic, Arn ]
      Action: lambda:InvokeFunction
      Principal: s3.amazonaws.com
      SourceAccount: !Ref AWS::AccountId
      SourceArn: !Sub "arn:aws:s3:::${S3BucketName}"

  LambdaFunctionNotificationSubscriptionWithKinesisTopic:
    Type: AWS::SNS::Subscription
    Condition: UseKinesisTopicARNWithNotification
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref LambdaFunctionWithKinesisTopic.DestinationTopic

  LambdaFunctionWithKinesisTopicDLQ:
    Condition: DLQEnabledAndUseKinesisStreamARN
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionWithKinesisTopic', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq

  ################################
  # ----- With MSK TopiC ------- #
  ################################

  LambdaFunctionWithMSK:
    Condition: UseMSK
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - AWSLambdaMSKExecutionRole
        - Statement:
            - Effect: Allow
              Action:
                - 'secretsmanager:GetSecretValue'
              Resource: !Sub 'arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:*'
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS
      # Events:
      #   MSKEvent:
      #     Type: MSK
      #     Properties:
      #       Stream: !Ref MSKClusterArn
      #       StartingPosition: LATEST
            
      #       Topics: 
      #         - !Ref KafkaTopic
      
  LambdaTriggerMskTopic:
    Condition: UseMSK
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaName: !GetAtt [ 'LambdaFunctionWithMSK', Arn ]
      MSKClusterArn: !Ref MSKClusterArn
      TopicsList: !Ref KafkaTopic

  LambdaLogGroupMSK:
    Condition: UseMSK
    DeletionPolicy: Retain
    UpdateReplacePolicy: Retain
    Type: 'AWS::Logs::LogGroup'
    Properties:
      LogGroupName: !Sub '/aws/lambda/${LambdaFunctionWithMSK}'
      RetentionInDays: !Ref LambdaLogRetention

  LambdaFunctionNotificationSubscriptionMSK:
    Condition: UseMSKWithNotification
    Type: AWS::SNS::Subscription
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref  LambdaFunctionWithMSK.DestinationTopic  

  LambdaFunctionWithMSKDLQ:
    Condition: DLQEnabledAndUseMSK
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionWithMSK', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq

  ##################################
  # ----- With Kafka TopiC ------- #
  ##################################

  LambdaFunctionWithKafka:
    Condition: UseKafka
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - AWSLambdaMSKExecutionRole
        - Statement:
            - Effect: Allow
              Action:
                - 'secretsmanager:GetSecretValue'
              Resource: !Sub 'arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:*'
        - Statement:
            - Sid: VPCAccessPolicy
              Effect: Allow
              Action:
                - ec2:CreateNetworkInterface
                - ec2:DescribeNetworkInterfaces
                - ec2:DeleteNetworkInterface
                - ec2:DescribeVpcs
                - ec2:DescribeSubnets
                - ec2:DescribeSecurityGroups
              Resource: "*"
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS

  LambdaTrigger:
    Type: Custom::KafkaTrigger
    Condition: UseKafka
    Properties:
      ServiceToken:
        Fn::GetAtt:
          - CustomResourceLambdaTriggerFunction
          - Arn
      Brokers: !Ref KafkaBrokers
      Topic: !Ref KafkaTopic
      Function: !GetAtt LambdaFunctionWithKafka.Arn
      BatchSize: !Ref KafkaBatchSize
      StartingPosition: LATEST
      SecurityGroupIds: !Ref KafkaSecurityGroups
      SubnetIds: !Ref KafkaSubnets
      Command: configure_kafka_trigger

  LambdaFunctionNotificationSubscriptionWithKafkaTopic:
    Type: AWS::SNS::Subscription
    Condition: UseKatkaTopicARNWithNotification
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref LambdaFunctionWithKafka.DestinationTopic

  LambdaFunctionWithKafkaDLQ:
    Condition: DLQEnabledAndUseKafkaTopicARN
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionWithKafka', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq

##################################
# ----- With ECR Scan Image -----#
##################################

  LambdaFunctionWithECRScan:
    Condition: UseECRScan
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: .
      Policies:
        - AWSLambdaBasicExecutionRole
        - Statement:
            - Effect: Allow
              Action:
                - 'ecr:DescribeImageScanFindings'
              Resource: !Sub 'arn:aws:ecr:${AWS::Region}:${AWS::AccountId}:repository/*'
            - Effect: Allow
              Action:
                - 'secretsmanager:GetSecretValue'
              Resource: !Sub 'arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:*'
        - Statement:
            - Effect: Allow
              Action:
                - events:PutEvents
              Resource: "*"
      EventInvokeConfig: 
        DestinationConfig:
          OnFailure:
            Type: SNS

  LambdaFunctionEcrInvokePermission:
    Condition: UseECRScan
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !GetAtt LambdaFunctionWithECRScan.Arn
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceAccount: !Ref AWS::AccountId
      SourceArn: !GetAtt ECRScanTrigger.Arn

  ECRScanTrigger:
    Condition: UseECRScan
    Type: AWS::Events::Rule
    Properties:
      Description: "Trigger for ECR image scan completion"
      State: ENABLED
      EventPattern:
        source:
          - "aws.ecr"
        detail-type:
          - "ECR Image Scan"
        detail:
          scan-status:
            - "COMPLETE"
      Targets:
        - Arn: !GetAtt LambdaFunctionWithECRScan.Arn
          Id: "TargetFunctionV1"
  
  LambdaFunctionNotificationSubscriptionWithECRScanTopic:
    Type: AWS::SNS::Subscription
    Condition: UseECRScanWithNotification
    Properties:
      Protocol: email
      Endpoint: !Ref NotificationEmail
      TopicArn: !Ref LambdaFunctionWithECRScan.DestinationTopic

  LambdaFunctionWithECRScanDLQ:
    Condition: DLQEnabledAndUseECRScan
    Type: Custom::LambdaTrigger
    Properties:
      ServiceToken: !GetAtt [ CustomResourceLambdaTriggerFunction, Arn ]
      LambdaArn: !GetAtt [ 'LambdaFunctionWithECRScan', Arn ]
      DLQArn: !GetAtt [ 'DeadLetterQueue', Arn ]
      Command: configure_dlq
  
  ################################ 
  # ----- Custom Resource -----  #
  ################################

  # define a custom resource to create the trigger.
  CustomResourceLambdaTriggerFunction:
    Type: AWS::Serverless::Function
    Properties:
      VpcConfig:
        Fn::If:
          - UseVpcConfig
          - SecurityGroupIds:
              - !Ref LambdaSecurityGroupID
            SubnetIds:
              - !Ref LambdaSubnetID
          - Ref: AWS::NoValue
      Handler: index.lambda_handler
      Runtime: python3.12
      Timeout: 50
      Policies:
        - Statement:
          - Sid: IAMAccess
            Effect: Allow
            Action:
              - iam:*
            Resource: "*"
        - Statement:
            - Sid: SQSAccess
              Effect: Allow
              Action:
                - sqs:*
              Resource: "*"
        - Statement:
            - Sid: LambdaAccess
              Effect: Allow
              Action:
                - lambda:*
              Resource: "*"
        - Statement:
            - Sid: S3NotificationPolicy
              Effect: Allow
              Action:
                - s3:GetBucketNotification
                - s3:PutBucketNotification
              Resource: !Sub "arn:aws:s3:::${S3BucketName}"

            - Sid: PutSubscriptionFilter
              Effect: Allow
              Action:
                - "logs:PutSubscriptionFilter"
                - "logs:DeleteSubscriptionFilter"
                - "logs:DescribeSubscriptionFilters"
              Resource: !Sub  "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:*:*"

      InlineCode: |
        import json
        import boto3
        # import cfnresponse
        import json, time, boto3, time
        from urllib import request, parse, error

        def add_inline_policy(lambda_arn: str, policy_name: str, policy_document: dict):
            try:
                iam = boto3.client('iam')
                aws_lambda = boto3.client('lambda')
                
                resp = aws_lambda.get_function_configuration(FunctionName=lambda_arn)
                role_arn = resp['Role']
                role_name = role_arn.split('/')[-1]
                
                print(f'creating policy for {lambda_arn}...')
                policy_resp = iam.create_policy(
                    PolicyName=policy_name.replace('-', '_'),
                    PolicyDocument=json.dumps(policy_document),
                    Description=f'policy generated by CustomResource for {lambda_arn}',
                )
                
                print(f'attaching {policy_name} for {lambda_arn} to role {role_name}...')
                role_attach_response = iam.attach_role_policy(
                    RoleName=role_name,
                    PolicyArn=policy_resp['Policy']['Arn']
                )
                
                time.sleep(30) # wait for policy to attach
                print(f'policy atttachment result: {role_attach_response}')
            except Exception as e:
                print(f'add_inline_policy function failed: {e}')
                raise e


        class CFNResponse:
            '''
            CFNResponse class, temporary class used to handle cloudformation responses
            as currently lambda has an issue importing cfnresponse module.
            '''
            SUCCESS = "SUCCESS"
            FAILED = "FAILED"

            def __init__(self, event, context):
                self.event = event
                self.context = context
                self.response_url = event['ResponseURL']

            def send(self, status, response_data={}, physical_resource_id=None, no_echo=False, reason=None):
                response_body = {
                    'Status': status,
                    'Reason': reason or "See the details in CloudWatch Log Stream: {}".format(self.context.log_stream_name),
                    'PhysicalResourceId': physical_resource_id or self.context.log_stream_name,
                    'StackId': self.event['StackId'],
                    'RequestId': self.event['RequestId'],
                    'LogicalResourceId': self.event['LogicalResourceId'],
                    'NoEcho': no_echo,
                    'Data': response_data
                }
                json_response_body = json.dumps(response_body).encode('utf-8')
                headers = {
                    'content-type': '',
                    'content-length': str(len(json_response_body))
                }
                try:
                    req = request.Request(self.response_url, data=json_response_body, headers=headers, method='PUT')
                    with request.urlopen(req) as response:
                        print("Status code:", response.getcode())
                        print("Response:", response.read().decode('utf-8'))
                except error.HTTPError as e:
                    print("HTTPError:", e.reason)
                    print("Status code:", e.code)
                except error.URLError as e:
                    print("URLError:", e.reason)


        class ConfigureS3Trigger:
            '''
            ConfigureS3Trigger class used to configure S3 Trigger implementation
            '''
            
            def __init__(self, event, context, cfn):
                self.context = context
                self.event = event
                self.cfn = cfn
                self.cfn.SUCCESS 
                self.s3 = boto3.client('s3')
                
            def handle(self):
                responseStatus = self.cfn.SUCCESS
                try:
                    print("Request Type:", self.event['RequestType'])
                    bucket = self.event['ResourceProperties']['Bucket']
                    BucketNotificationConfiguration = self.s3.get_bucket_notification_configuration(
                        Bucket=bucket
                    )
                    BucketNotificationConfiguration.pop('ResponseMetadata')
                    BucketNotificationConfiguration.setdefault('LambdaFunctionConfigurations', [])

                    if self.event['RequestType'] in ['Update', 'Delete']:
                        BucketNotificationConfiguration['LambdaFunctionConfigurations'] = list(
                            filter(
                                lambda configuration: configuration.get('Id') != self.event['PhysicalResourceId'],
                                BucketNotificationConfiguration['LambdaFunctionConfigurations']
                            )
                        )
                    if self.event['RequestType'] in ['Create', 'Update']:
                        BucketNotificationConfiguration['LambdaFunctionConfigurations'].append({
                            'Id': self.event.get('PhysicalResourceId', self.context.aws_request_id),
                            'LambdaFunctionArn': self.event['ResourceProperties']['LambdaArn'],
                            'Filter': {
                                'Key': {
                                    'FilterRules': [
                                        {
                                            'Name': 'prefix',
                                            'Value': self.event['ResourceProperties'].get('Prefix', '')
                                        },
                                        {
                                            'Name': 'suffix',
                                            'Value': self.event['ResourceProperties'].get('Suffix', '')
                                        },
                                    ]
                                }
                            },
                            'Events': [
                                's3:ObjectCreated:*'
                            ]
                        })
                    if len(BucketNotificationConfiguration['LambdaFunctionConfigurations']) == 0:
                        BucketNotificationConfiguration.pop('LambdaFunctionConfigurations')
                    print(f'nofication configuration: {BucketNotificationConfiguration}')
                    self.s3.put_bucket_notification_configuration(
                        Bucket=bucket,
                        NotificationConfiguration=BucketNotificationConfiguration
                    )
                    responseStatus = self.cfn.SUCCESS
                    print(self.event['RequestType'], "request completed....")
                except Exception as e:
                    print("Failed to process:", e)
                    responseStatus = self.cfn.FAILED
                finally:
                    print("Sending response to custom resource")
                    self.cfn.send(
                        responseStatus,
                        physical_resource_id=self.event.get('PhysicalResourceId', self.context.aws_request_id)  
                    )
                    
                
        class ConfigureKafkaTrigger:
            '''
            ConfigureKafkaTrigger 
            ''' 
            
            def __init__(self, event, context, cfn):
                self.cfn = cfn
                self.event = event
                self.context = context
                self.aws_lambda = boto3.client("lambda")
          
            def delete_event_source_mappings(self, function_name):
                mappings = self.aws_lambda.list_event_source_mappings(
                    FunctionName=function_name,
                )["EventSourceMappings"]
                for mapping in mappings:
                    # disable mapping
                    if mapping["State"] == "Enabled":
                        self.aws_lambda.update_event_source_mapping(
                            UUID=mapping["UUID"],
                            FunctionName=function_name,
                            Enabled=False
                        )
                    time.sleep(10) # wait for mapping to be disabled
                    self.aws_lambda.delete_event_source_mapping(UUID=mapping["UUID"])
            
            def handle(self):
                if self.event['ResourceProperties']['Command'] == 'configure_msk_trigger':
                    self.handle_msk()
                    return
                self.handle_kafka()
            
            def handle_msk(self):
                try:        
                    lambda_name = self.event['ResourceProperties']['LambdaName']
                    msk_cluster_arn = self.event['ResourceProperties']['MSKClusterArn']
                    if self.event['RequestType'] in ['Create', 'Update']:
                        TopicsListString = self.event['ResourceProperties']['TopicsList']
                        TopicsList = TopicsListString.split(',')
                        for topic in TopicsList:
                            response = self.aws_lambda.create_event_source_mapping(
                                EventSourceArn=msk_cluster_arn,
                                FunctionName=lambda_name,
                                Topics=[
                                    topic
                                ],
                                StartingPosition='LATEST',
                                BatchSize=100,
                            )
                    responseStatus = self.cfn.SUCCESS
                    print(self.event['RequestType'], "request completed....")
                except Exception as exc:
                    print("Failed to process:", exc)
                    responseStatus = self.cfn.FAILED
                finally:
                    self.cfn.send(
                        responseStatus,
                        physical_resource_id=self.event.get('PhysicalResourceId', self.context.aws_request_id)
                    )
          
            def handle_kafka(self):
                responseStatus = self.cfn.SUCCESS
                physicalResourceId = self.event.get("PhysicalResourceId")
                function_name = self.event["ResourceProperties"]["Function"]
                try:
                    print("Request Type:", self.event["RequestType"])
                    if self.event["RequestType"] in ["Create", "Update"]:
                        if self.event["RequestType"] == "Update":
                            print('Update event detected, deleting previous mapping(s)')
                            self.delete_event_source_mappings(function_name)
                        response = self.aws_lambda.create_event_source_mapping(
                            FunctionName=self.event["ResourceProperties"]["Function"],
                            BatchSize=int(self.event["ResourceProperties"]["BatchSize"]),
                            StartingPosition=self.event["ResourceProperties"]["StartingPosition"],
                            Topics=[
                                self.event["ResourceProperties"]["Topic"]
                            ],
                            SelfManagedEventSource={
                                "Endpoints": {
                                    "KAFKA_BOOTSTRAP_SERVERS": self.event["ResourceProperties"]["Brokers"]
                                }
                            },
                            SourceAccessConfigurations=list([
                                {
                                    "Type": "VPC_SUBNET",
                                    "URI": "subnet:" + subnetId
                                } for subnetId in self.event["ResourceProperties"]["SubnetIds"]
                            ]) + list([
                                {
                                    "Type": "VPC_SECURITY_GROUP",
                                    "URI": "security_group:" + securityGroupId
                                } for securityGroupId in self.event["ResourceProperties"]["SecurityGroupIds"]
                            ])
                        )
                        physicalResourceId = response["UUID"]
                        print(f"EventSourceMapping successfully created: {physicalResourceId}")
                    elif self.event["RequestType"] == "Delete":
                        self.delete_event_source_mappings(function_name)
                        print("EventSourceMapping successfully deleted")
                except Exception as exc:
                    print("Failed to process:", exc)
                    responseStatus = self.cfn.FAILED
                finally:
                    self.cfn.send(responseStatus, physical_resource_id=physicalResourceId)      


        class ConfigureCloudwatchTrigger:
            '''
            CloudwatchTrigger
            '''
            
            def __init__(self, event, context, cfn):
                self.event = event
                self.context = context
                self.cfn = cfn
                self.aws_lambda = boto3.client("lambda")
                self.cloudwatch_logs = boto3.client('logs')
                self.lambda_client = boto3.client('lambda')

            def handle(self):
                try:        
                    lambda_arn = self.event['ResourceProperties']['LambdaArn']
                    region = self.context.invoked_function_arn.split(":")[3]
                    account_id = self.context.invoked_function_arn.split(":")[4]
                    StringlogGroupName = self.event['ResourceProperties']['CloudwatchGroup']
                    logGroupName = StringlogGroupName.split(',')
                    LambdaPremissionPreFix = self.event['ResourceProperties']['CloudwatchPrefix'].split(',')
                    if self.event['RequestType'] in ['Update', 'Delete']:
                        for log_group in logGroupName:
                            response = self.cloudwatch_logs.describe_subscription_filters(logGroupName=log_group)
                            for filter in response['subscriptionFilters']:
                                if filter['filterName'] == f'coralogix-aws-shipper-cloudwatch-trigger-{lambda_arn[-4:]}':
                                    self.cloudwatch_logs.delete_subscription_filter(
                                        filterName=f'coralogix-aws-shipper-cloudwatch-trigger-{lambda_arn[-4:]}',
                                        logGroupName=log_group
                                    )
                                if not LambdaPremissionPreFix:
                                  response = self.aws_lambda.remove_permission(
                                      FunctionName=lambda_arn,
                                      StatementId=f'allow-trigger-from-{log_group.replace("/", "-")}'
                                  )
                    if self.event['RequestType'] in ['Create', 'Update']:
                      if LambdaPremissionPreFix and LambdaPremissionPreFix != [""]:
                        for prefix in LambdaPremissionPreFix:
                          try:
                            self.lambda_client.add_permission(
                              FunctionName=lambda_arn,
                              StatementId=f'allow-trigger-from-{prefix.replace("/", "-")}-log-groups',
                              Action='lambda:InvokeFunction',
                              Principal='logs.amazonaws.com',
                              SourceArn=f'arn:aws:logs:{region}:{account_id}:log-group:{prefix}*:*',
                            )
                          except Exception as e:
                            print("permission already exists: ", str(e))

                      for log_group in logGroupName:
                          response = self.cloudwatch_logs.describe_subscription_filters(
                              logGroupName=log_group,
                              filterNamePrefix=f'coralogix-aws-shipper-cloudwatch-trigger-{lambda_arn[-4:]}'
                          )
                          if not LambdaPremissionPreFix or LambdaPremissionPreFix == [""]:
                            if not response.get("subscriptionFilters") or response.get("subscriptionFilters")[0].get("destinationArn") != lambda_arn:
                                response = self.aws_lambda.add_permission(
                                    FunctionName=lambda_arn,
                                    StatementId=f'allow-trigger-from-{log_group.replace("/", "-")}',
                                    Action='lambda:InvokeFunction',
                                    Principal='logs.amazonaws.com',
                                    SourceArn=f'arn:aws:logs:{region}:{account_id}:log-group:{log_group}:*',
                                )
                            time.sleep(1)
                          self.cloudwatch_logs.put_subscription_filter(
                              destinationArn=self.event['ResourceProperties']['LambdaArn'],
                              filterName=f'coralogix-aws-shipper-cloudwatch-trigger-{lambda_arn[-4:]}',
                              filterPattern='',
                              logGroupName=log_group
                          )

                    responseStatus = self.cfn.SUCCESS
                    print(self.event['RequestType'], "request completed....")
                except Exception as e:
                    print("Failed to process:", e)
                    responseStatus = self.cfn.FAILED
                finally:
                    print("Sending response to custom resource")
                    self.cfn.send(
                        responseStatus,
                        physical_resource_id=self.event.get('PhysicalResourceId', self.context.aws_request_id)
                    )
        
        class ConfigureDLQ:
            '''
            ConfigureDLQ        
            '''
            
            def __init__(self, event, context, cfn):
                self.event = event
                self.context = context
                self.cfn = cfn
                self.aws_lambda = boto3.client("lambda")

            def handle(self):
                try:
                    lambda_arn = self.event['ResourceProperties']['LambdaArn']
                    if self.event['RequestType'] in ['Create', 'Update']:
                        # Add inline policy to allow lambda to send to sqs
                        lambda_name = lambda_arn.split(':')[-1]
                        policy_name = f'{lambda_name}_sqs_send_policy'
                        policy_document = {
                            "Version": "2012-10-17",
                            "Statement": [
                                {
                                    "Effect": "Allow",
                                    "Action": [
                                        "sqs:SendMessage",
                                        "sqs:ReceiveMessage",
                                        "sqs:DeleteMessage",
                                        "sqs:GetQueueAttributes"
                                    ],
                                    "Resource": self.event['ResourceProperties']['DLQArn']
                                },
                                {
                                    "Effect": "Allow",
                                    "Action": [
                                        "s3:PutObject",
                                        "s3:PutObjectAcl",
                                        "s3:AbortMultipartUpload",
                                        "s3:DeleteObject",
                                        "s3:PutObjectTagging",
                                        "s3:PutObjectVersionTagging"
                                    ],
                                    "Resource": "*"
                                }
                            ]
                        }
                        
                        add_inline_policy(lambda_arn, policy_name, policy_document)
                        
                        self.aws_lambda.update_function_configuration(
                            FunctionName=lambda_arn,
                            DeadLetterConfig={
                                'TargetArn': self.event['ResourceProperties']['DLQArn']
                            },
                        )
                        
                        # Create event source mapping
                        self.aws_lambda.create_event_source_mapping(
                            EventSourceArn=self.event['ResourceProperties']['DLQArn'],
                            FunctionName=lambda_arn,
                            Enabled=True,
                            BatchSize=1,
                        )
                    responseStatus = self.cfn.SUCCESS
                    print(self.event['RequestType'], "request completed....")
                except Exception as e:
                    print("Failed to process:", e)
                    responseStatus = self.cfn.FAILED
                finally:
                    print("Sending response to custom resource")
                    self.cfn.send(
                        responseStatus,
                        physical_resource_id=self.event.get('PhysicalResourceId', self.context.aws_request_id)
                    )   

        def lambda_handler(event, context):
            print("Received event:", json.dumps(event))
            cfn = CFNResponse(event, context)
            # identify the command to run
            command = event['ResourceProperties']['Command']
            
            match command:
                case 'configure_s3_trigger':
                    ConfigureS3Trigger(event, context, cfn).handle()
                    
                case 'configure_kafka_trigger' | 'configure_msk_trigger':
                    ConfigureKafkaTrigger(event, context, cfn).handle()
                    
                case 'configure_cloudwatch_trigger':
                    ConfigureCloudwatchTrigger(event, context, cfn).handle()
                    
                case 'configure_dlq':
                    ConfigureDLQ(event, context, cfn).handle()
            
                case _:
                    raise ValueError(f"invalid or unsupported custom resource command: {command}")
            
            return