能否在CloudFormation模板中自动化测试AWS DMS端点？

CloudFormation本身并没有原生的资源类型来直接测试DMS连接，但你可以通过自定义资源（Custom Resource）+ Lambda函数的组合来完成这项自动化任务。下面是具体的实现思路和步骤：

核心实现逻辑 #

1. 编写Lambda测试函数
   用Python（或其他支持AWS SDK的语言）写一个Lambda函数，调用AWS DMS的test_connection API来验证复制实例和端点的连通性。函数需要做这些事：

   • 接收CloudFormation传递的参数（比如复制实例ARN、端点ARN）
   • 调用DMS发起连接测试请求
   • 轮询测试结果，直到状态明确或超时
   • 向CloudFormation返回测试成功/失败的响应，决定栈的创建状态

   这里给个Lambda核心代码片段：

   import boto3
   import time
   import cfnresponse
   
   dms = boto3.client('dms')
   
   def lambda_handler(event, context):
       try:
           if event['RequestType'] in ['Create', 'Update']:
               # 从CloudFormation参数获取资源ARN
               rep_instance_arn = event['ResourceProperties']['ReplicationInstanceArn']
               endpoint_arn = event['ResourceProperties']['EndpointArn']
               
               # 发起连接测试
               dms.test_connection(
                   ReplicationInstanceArn=rep_instance_arn,
                   EndpointArn=endpoint_arn
               )
               
               # 轮询测试结果，最多等待30秒
               timeout = time.time() + 30
               while time.time() < timeout:
                   conn_status = dms.describe_connections(
                       Filters=[
                           {'Name': 'replication-instance-arn', 'Values': [rep_instance_arn]},
                           {'Name': 'endpoint-arn', 'Values': [endpoint_arn]}
                       ]
                   )['Connections'][0]
                   
                   if conn_status['Status'] == 'successful':
                       cfnresponse.send(event, context, cfnresponse.SUCCESS, {'Result': 'Connection test passed'})
                       return
                   elif conn_status['Status'] == 'failed':
                       raise Exception(f"Test failed: {conn_status['Message']}")
                   time.sleep(2)
               
               raise Exception("Connection test timed out after 30 seconds")
           
           elif event['RequestType'] == 'Delete':
               # 删除栈时无需测试，直接返回成功
               cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
       except Exception as e:
           cfnresponse.send(event, context, cfnresponse.FAILED, {'Error': str(e)})
   

2. 在CloudFormation中定义自定义资源
   在你的模板里添加自定义资源，依赖于已创建的复制实例和端点，确保这些资源就绪后再执行测试：

   Resources:
     # 假设你已经定义了复制实例和端点：
     # MyReplicationInstance:
     #   Type: AWS::DMS::ReplicationInstance
     #   ...
     # MySourceEndpoint:
     #   Type: AWS::DMS::Endpoint
     #   ...
   
     DmsConnectionTestLambda:
       Type: AWS::Lambda::Function
       Properties:
         Runtime: python3.11
         Handler: index.lambda_handler
         Role: !GetAtt DmsTestLambdaRole.Arn
         Code:
           ZipFile: |
             # 上面的Lambda代码复制到这里
         Timeout: 60 # 设置足够的超时时间
   
     DmsTestLambdaRole:
       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: DmsTestPermissions
             PolicyDocument:
               Version: '2012-10-17'
               Statement:
                 - Effect: Allow
                   Action:
                     - dms:TestConnection
                     - dms:DescribeConnections
                   Resource:
                     - !Ref MyReplicationInstance
                     - !Ref MySourceEndpoint
   
     TestDmsConnection:
       Type: Custom::DmsConnectionTest
       Properties:
         ServiceToken: !GetAtt DmsConnectionTestLambda.Arn
         ReplicationInstanceArn: !Ref MyReplicationInstance
         EndpointArn: !Ref MySourceEndpoint
       DependsOn:
         - MyReplicationInstance
         - MySourceEndpoint
   

关键注意事项 #

• Lambda超时设置：DMS连接测试可能需要几秒到十几秒，建议把Lambda的超时时间设置为30-60秒，避免提前终止。
• 权限控制：确保Lambda角色拥有dms:TestConnection和dms:DescribeConnections权限，并且资源范围限制在你的目标实例和端点上，遵循最小权限原则。
• 失败处理：如果测试失败，CloudFormation会将整个栈标记为创建失败，你可以通过CloudFormation事件或Lambda日志查看具体错误原因。
• 更新场景：如果后续修改了端点或复制实例配置，自定义资源会在更新栈时重新执行测试，确保新配置的连通性。

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