寻求适配Python3.8/3.9的AWS ALB监听器规则优先级自动设置代码

以下是兼容Python 3.8/3.9的CloudFormation自定义资源Lambda代码，用于自动分配ALB ListenerRule的优先级：

import boto3
import json
import logging
from botocore.exceptions import ClientError

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

def lambda_handler(event, context):
    alb_client = boto3.client('elbv2')
    response_data = {}
    
    try:
        # 解析CloudFormation事件参数
        listener_arn = event['ResourceProperties']['ListenerArn']
        request_type = event['RequestType']
        
        # 获取当前Listener下的所有规则优先级
        existing_priorities = set()
        paginator = alb_client.get_paginator('describe_rules')
        for page in paginator.paginate(ListenerArn=listener_arn):
            for rule in page['Rules']:
                if rule['Priority'].isdigit():
                    existing_priorities.add(int(rule['Priority']))
        
        # 确定新的优先级（从1开始找最小未使用的）
        new_priority = 1
        while new_priority in existing_priorities:
            new_priority += 1
        
        if request_type == 'Create':
            # 创建ListenerRule
            rule_response = alb_client.create_rule(
                ListenerArn=listener_arn,
                Priority=new_priority,
                Conditions=event['ResourceProperties']['Conditions'],
                Actions=event['ResourceProperties']['Actions']
            )
            response_data['RuleArn'] = rule_response['Rules'][0]['RuleArn']
            response_data['Priority'] = new_priority
            send_success_response(event, context, response_data)
        
        elif request_type == 'Update':
            # 更新时保持原优先级，如需重新分配可修改逻辑
            rule_arn = event['PhysicalResourceId']
            rule_desc = alb_client.describe_rules(RuleArns=[rule_arn])['Rules'][0]
            current_priority = int(rule_desc['Priority'])
            
            alb_client.modify_rule(
                RuleArn=rule_arn,
                Conditions=event['ResourceProperties']['Conditions'],
                Actions=event['ResourceProperties']['Actions']
            )
            response_data['RuleArn'] = rule_arn
            response_data['Priority'] = current_priority
            send_success_response(event, context, response_data)
        
        elif request_type == 'Delete':
            # 删除ListenerRule
            rule_arn = event['PhysicalResourceId']
            alb_client.delete_rule(RuleArn=rule_arn)
            send_success_response(event, context, response_data)
    
    except ClientError as e:
        logger.error(f"Error handling request: {e}")
        send_failure_response(event, context, str(e))
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        send_failure_response(event, context, str(e))

def send_success_response(event, context, response_data):
    send_response(event, context, "SUCCESS", response_data)

def send_failure_response(event, context, error_message):
    send_response(event, context, "FAILED", {'Error': error_message})

def send_response(event, context, response_status, response_data):
    response_body = json.dumps({
        'Status': response_status,
        'Reason': f"See details in CloudWatch Log Stream: {context.log_stream_name}",
        'PhysicalResourceId': context.log_stream_name if response_status == 'FAILED' else (response_data.get('RuleArn') or context.log_stream_name),
        'StackId': event['StackId'],
        'RequestId': event['RequestId'],
        'LogicalResourceId': event['LogicalResourceId'],
        'Data': response_data
    })
    
    import urllib3
    http = urllib3.PoolManager()
    try:
        response = http.request(
            'PUT',
            event['ResponseURL'],
            body=response_body,
            headers={'Content-Type': ''},
            retries=False
        )
        logger.info(f"Response sent successfully: {response.status}")
    except Exception as e:
        logger.error(f"Failed to send response: {e}")

CloudFormation模板使用示例 #

在你的CloudFormation模板中，添加以下自定义资源调用上述Lambda：

Resources:
  AutoPriorityRuleLambda:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.9
      Handler: index.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn
      Code:
        ZipFile: |
          # 粘贴上面的Lambda代码
      Timeout: 30

  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: ALBRuleManagement
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - elbv2:DescribeRules
                  - elbv2:CreateRule
                  - elbv2:ModifyRule
                  - elbv2:DeleteRule
                Resource: "*"

  MyAutoPriorityListenerRule:
    Type: Custom::AutoPriorityListenerRule
    Properties:
      ServiceToken: !GetAtt AutoPriorityRuleLambda.Arn
      ListenerArn: !Ref MyALBListener
      Conditions:
        - Field: path-pattern
          Values: ['/api/*']
      Actions:
        - Type: forward
          TargetGroupArn: !Ref MyTargetGroup

关键适配说明 #

• 使用boto3分页器处理大量规则场景，避免单次调用数据截断
• 替换第三方requests模块为Lambda内置的urllib3，无需额外打包依赖
• 优化错误捕获与反馈逻辑，确保CloudFormation能及时获取资源状态
• 兼容Python 3.8/3.9的语法规范与Lambda运行时环境要求

内容的提问来源于stack exchange，提问作者Gladwin Ram