MinimumCircuitBreakerStack.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform:
  - AWS::LanguageExtensions
  - AWS::Serverless-2016-10-31
Description: SAM Template for CbEventSourceManager

Parameters:
  CircuitBreakerAlarmArn:
    Description: Alarm that is base for the circuit breaker
    Type: String
  ManagedFunctionArn:
    Description: SQS Consumer Function ARN
    Type: String
  ManagedQueueArn:
    Description: Consumed SQS ARN
    Type: String
  ConsumerClosedConcurrency:
    Description: Closed concurrency for the consumer
    Type: Number

Resources:
  FunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub
        - '/aws/lambda/${LambdaName}'
        - { LambdaName: !Ref InlineCbEventSourceManagerFunction }
      RetentionInDays: 14

  InlineCbEventSourceManagerFunction:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: nodejs20.x
      AutoPublishAlias: live
      AutoPublishAliasAllProperties: true
      Architectures:
        - arm64
      Handler: index.handler
      InlineCode: |
        const { LambdaClient, ListEventSourceMappingsCommand, UpdateEventSourceMappingCommand } = require("@aws-sdk/client-lambda");
        const lambda = new LambdaClient({ region: process.env.AWS_REGION });

        exports.handler = async (event) => {
            console.debug(`Got event: ${JSON.stringify(event)}`);

            const mappings = await getEventSourceMappings();
            if (!mappings) {
                console.error("No event source mappings found");
                return;
            }
            if (mappings.EventSourceMappings.length > 1) {
                console.error("Multiple event source mappings found");
                return;
            }
            
            if (event.detail.state.value === "ALARM") {
                await openCircuit(mappings.EventSourceMappings[0]);
            } else if (event.detail.state.value === "OK") {
                await closeCircuit(mappings.EventSourceMappings[0]);
            } else if (event.detail.state.value === "INSUFFICIENT_DATA" && event.detail.previousState.value === "ALARM") {
                await halfOpen(mappings.EventSourceMappings[0]);
            }
        };

        async function getEventSourceMappings() {
            const params = {
                EventSourceArn: process.env.QUEUE_ARN,
                FunctionName: process.env.FUNCTION_ARN
            };

            try {
                return await lambda.send(new ListEventSourceMappingsCommand(params));
            } catch (err) {
                console.error(`Error getting event source mappings: ${err}`);
            }
        }

        async function openCircuit(mapping) {
            const params = {
                UUID: mapping.UUID,
                Enabled: false
            };

            try {
                const data = await lambda.send(new UpdateEventSourceMappingCommand(params));
                console.debug(`Updated mapping: ${JSON.stringify(data)}`);
            } catch (err) {
                console.error(`Error updating event source mapping: ${err}`);
            }
        }

        async function closeCircuit(mapping) {
            const closedConcurrency = parseInt(process.env.CLOSED_CONCURRENCY);
            const params = {
                UUID: mapping.UUID,
                Enabled: true,
                ScalingConfig: {
                    MaximumConcurrency: closedConcurrency
                }
            };

            try {
                const data = await lambda.send(new UpdateEventSourceMappingCommand(params));
                console.debug(`Updated mapping: ${JSON.stringify(data)}`);
            } catch (err) {
                console.error(`Error updating event source mapping: ${err}`);
            }
        }

        async function halfOpen(mapping) {
            const params = {
                UUID: mapping.UUID,
                Enabled: true,
                ScalingConfig: {
                    MaximumConcurrency: 2
                }
            };

            try {
                const data = await lambda.send(new UpdateEventSourceMappingCommand(params));
                console.debug(`Updated mapping: ${JSON.stringify(data)}`);
            } catch (err) {
                console.error(`Error updating event source mapping: ${err}`);
            }
        }
      Tracing: Active
      Timeout: 10
      MemorySize: 128
      Environment:
        Variables:
          FUNCTION_ARN: !Ref ManagedFunctionArn
          QUEUE_ARN: !Ref ManagedQueueArn
          CLOSED_CONCURRENCY: !Ref ConsumerClosedConcurrency
      Events:
        OpenCircuitRule:
          Type: EventBridgeRule
          Properties:
            Pattern:
              source:
                - "aws.cloudwatch"
              detail-type:
                - "CloudWatch Alarm State Change"
              resources:
                - !Ref CircuitBreakerAlarmArn
      Policies:
        - Statement:
            - Sid: ListEventSourceMappings
              Effect: Allow
              Action:
                - lambda:listEventSourceMappings
              Resource: "*"
            - Sid: UpdateEventSourceMapping
              Effect: Allow
              Action:
                - lambda:UpdateEventSourceMapping
              Resource: !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:event-source-mapping:*

  FunctionErrorsAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmDescription: Lambda errors
      Namespace: AWS/Lambda
      MetricName: Errors
      Dimensions:
        - Name: FunctionName
          Value: !Ref InlineCbEventSourceManagerFunction
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 2
      Threshold: 5
      ComparisonOperator: GreaterThanOrEqualToThreshold
      TreatMissingData: notBreaching
      AlarmActions:
        - <YOUR_SNS_TOPIC>
      OKActions:
        - <YOUR_SNS_TOPIC>