在中调用管道中的 AWS Lambda 函数 CodePipeline

按照中的教程：创建一个简单的管道（S3 存储桶）前三个步骤创建 Amazon S3 存储桶、 CodeDeploy资源和两阶段管道。为您的实例类型选择 Amazon Linux 选项。你可以为管道使用任何你想要的名称，但本主题中的步骤使用 MyLambdaTestPipeline。

在管道的状态页面的 CodeDeploy 操作中，选择详细信息。在部署组的部署详细信息页中，从列表中选择一个实例 ID。

在 Amazon EC2 控制台中，在实例的详细信息选项卡上，复制公共IPv4地址中的 IP 地址（例如，192.0.2.4）。您应将此地址用作 AWS Lambda中的函数的目标。

登录 AWS Management Console 并打开IAM控制台，网址为https://console.aws.amazon.com/iam/。

选择 Policies，然后选择 Create Policy。选择JSON选项卡，然后将以下策略粘贴到该字段中。

{
  "Version": "2012-10-17", 
  "Statement": [
    {
      "Action": [ 
        "logs:*"
      ],
      "Effect": "Allow", 
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Action": [
        "codepipeline:PutJobSuccessResult",
        "codepipeline:PutJobFailureResult"
        ],
        "Effect": "Allow",
        "Resource": "*"
     }
  ]
} 

选择查看策略。

在 Review policy (查看策略) 页面上，在 Name (名称) 中，键入策略的名称（例如，CodePipelineLambdaExecPolicy）。在 Description (描述) 中，输入 Enables Lambda to execute code。

请选择创建策略。

在策略控制面板页上，选择 Roles (角色)，然后选择 Create role (创建角色)。

在创建角色页面上，选择 AWS 服务。选择 Lambda，然后选择 Next: Permissions (下一步: 权限)。

在附加权限策略页面上，选中旁边的复选框 CodePipelineLambdaExecPolicy，然后选择下一步：标签。选择下一步：审核。

在 Review (审核) 页面上的 Role name (角色名称) 中，输入名称，然后选择 Create role (创建角色)。

登录 AWS Management Console 并打开 AWS Lambda 控制台，网址为https://console.aws.amazon.com/lambda/。

在 Functions (LAM 函数) 页面上，选择 Create function (创建函数)。

在创建函数页面上，选择从头开始创作。在函数名称中，输入 Lambda 函数的名称（例如 MyLambdaFunctionForAWSCodePipeline）。在 Runtime 中，选择 Node.js 20.x。

在 Role (角色) 下，选择 Choose an existing role (选择现有角色)。在 Existing role (现有角色) 中，选择角色，然后选择 Create function (创建函数)。

此时将打开供您创建函数的详细信息页面。

在 Function code (函数代码) 框中，复制贴以下代码：

import { CodePipelineClient, PutJobSuccessResultCommand, PutJobFailureResultCommand } from "@aws-sdk/client-codepipeline";
import http from 'http';
import assert from 'assert';
 
export const handler = (event, context) => {
 
    const codepipeline = new CodePipelineClient();
 
    // Retrieve the Job ID from the Lambda action
    const jobId = event["CodePipeline.job"].id;
 
    // Retrieve the value of UserParameters from the Lambda action configuration in CodePipeline, in this case a URL which will be
    // health checked by this function.
    const url = event["CodePipeline.job"].data.actionConfiguration.configuration.UserParameters;
 
    // Notify CodePipeline of a successful job
    const putJobSuccess = async function(message) {
        const command = new PutJobSuccessResultCommand({
            jobId: jobId
        });
         try {
             await codepipeline.send(command);
             context.succeed(message);
         } catch (err) {
             context.fail(err); 
         }
    };
 
    // Notify CodePipeline of a failed job
    const putJobFailure = async function(message) {
        const command = new PutJobFailureResultCommand({
            jobId: jobId,
            failureDetails: {
                message: JSON.stringify(message),
                type: 'JobFailed',
                externalExecutionId: context.awsRequestId
            }
        });
        await codepipeline.send(command);
        context.fail(message);
    };
 
    // Validate the URL passed in UserParameters
    if(!url || url.indexOf('http://') === -1) {
        putJobFailure('The UserParameters field must contain a valid URL address to test, including http:// or https://');  
        return;
    }
 
    // Helper function to make a HTTP GET request to the page.
    // The helper will test the response and succeed or fail the job accordingly
    const getPage = function(url, callback) {
        var pageObject = {
            body: '',
            statusCode: 0,
            contains: function(search) {
                return this.body.indexOf(search) > -1;
            }
        };
        http.get(url, function(response) {
            pageObject.body = '';
            pageObject.statusCode = response.statusCode;
 
            response.on('data', function (chunk) {
                pageObject.body += chunk;
            });
 
            response.on('end', function () {
                callback(pageObject);
            });
 
            response.resume();
        }).on('error', function(error) {
            // Fail the job if our request failed
            putJobFailure(error);
        });
    };
 
    getPage(url, function(returnedPage) {
        try {
            // Check if the HTTP response has a 200 status
            assert(returnedPage.statusCode === 200);
            // Check if the page contains the text "Congratulations"
            // You can change this to check for different text, or add other tests as required
            assert(returnedPage.contains('Congratulations'));
 
            // Succeed the job
            putJobSuccess("Tests passed.");
        } catch (ex) {
            // If any of the assertions failed then fail the job
            putJobFailure(ex);
        }
    });
};

保留 Handler (处理程序) 和 Role (角色) 的默认值 CodePipelineLambdaExecRole。

在 Basic settings (基本设置) 中，对于 Timeout (超时)，输入 20 秒。

选择保存。

登录 AWS Management Console 并打开 CodePipeline 控制台，网址为 http://console.aws.amazon.com/codesuite/codepipeline/home。

在 Welcome (欢迎使用) 页面上，选择您创建的管道。

在管道视图页中，选择 Edit。

在 “编辑” 页面上，选择 + 添加阶段，在部署阶段之后添加一个带有 CodeDeploy 操作的阶段。输入阶段的名称（例如，LambdaStage），然后选择 Add stage (添加阶段)。

选择 + 添加操作组。在编辑操作的操作名称中，键入 Lambda 操作的名称（例如，MyLambdaAction）。在 Provider (提供方) 中，选择 AWS Lambda。在函数名称中，选择或输入 Lambda 函数的名称（例如，MyLambdaFunctionForAWSCodePipeline）。在用户参数中，指定您之前复制的 Amazon EC2 实例（例如http://192.0.2.4）的 IP 地址，然后选择完成。

在 Edit action (编辑操作) 页面上，选择 Save (保存)。

在管道详细信息页中，选择发布更改。这会通过管道运行在源操作中指定的每个源位置中提供的最新修订。

Lambda 操作完成后，选择详细信息链接以在 Amazon 中查看该函数的日志流 CloudWatch，包括事件的计费时长。如果函数失败，则 CloudWatch 日志会提供有关原因的信息。

打开 Lambda 控制台，选择创建函数。

在创建函数页面上，选择从头开始创作。在函数名称中，为您的 Lambda 函数输入一个名称。

在 Runtime (运行时) 中，选择 Python 2.7。

在选择或创建执行角色下，选择使用现有角色。在 Existing role (现有角色) 中，选择角色，然后选择 Create function (创建函数)。

此时将打开供您创建函数的详细信息页面。

在 Function code (函数代码) 框中，复制贴以下代码：

from __future__ import print_function
from boto3.session import Session

import json
import urllib
import boto3
import zipfile
import tempfile
import botocore
import traceback

print('Loading function')

cf = boto3.client('cloudformation')
code_pipeline = boto3.client('codepipeline')

def find_artifact(artifacts, name):
    """Finds the artifact 'name' among the 'artifacts'
    
    Args:
        artifacts: The list of artifacts available to the function
        name: The artifact we wish to use
    Returns:
        The artifact dictionary found
    Raises:
        Exception: If no matching artifact is found
    
    """
    for artifact in artifacts:
        if artifact['name'] == name:
            return artifact
            
    raise Exception('Input artifact named "{0}" not found in event'.format(name))

def get_template(s3, artifact, file_in_zip):
    """Gets the template artifact
    
    Downloads the artifact from the S3 artifact store to a temporary file
    then extracts the zip and returns the file containing the CloudFormation
    template.
    
    Args:
        artifact: The artifact to download
        file_in_zip: The path to the file within the zip containing the template
        
    Returns:
        The CloudFormation template as a string
        
    Raises:
        Exception: Any exception thrown while downloading the artifact or unzipping it
    
    """
    tmp_file = tempfile.NamedTemporaryFile()
    bucket = artifact['location']['s3Location']['bucketName']
    key = artifact['location']['s3Location']['objectKey']
    
    with tempfile.NamedTemporaryFile() as tmp_file:
        s3.download_file(bucket, key, tmp_file.name)
        with zipfile.ZipFile(tmp_file.name, 'r') as zip:
            return zip.read(file_in_zip)   
   
def update_stack(stack, template):
    """Start a CloudFormation stack update
    
    Args:
        stack: The stack to update
        template: The template to apply
        
    Returns:
        True if an update was started, false if there were no changes
        to the template since the last update.
        
    Raises:
        Exception: Any exception besides "No updates are to be performed."
    
    """
    try:
        cf.update_stack(StackName=stack, TemplateBody=template)
        return True
        
    except botocore.exceptions.ClientError as e:
        if e.response['Error']['Message'] == 'No updates are to be performed.':
            return False
        else:
            raise Exception('Error updating CloudFormation stack "{0}"'.format(stack), e)

def stack_exists(stack):
    """Check if a stack exists or not
    
    Args:
        stack: The stack to check
        
    Returns:
        True or False depending on whether the stack exists
        
    Raises:
        Any exceptions raised .describe_stacks() besides that
        the stack doesn't exist.
        
    """
    try:
        cf.describe_stacks(StackName=stack)
        return True
    except botocore.exceptions.ClientError as e:
        if "does not exist" in e.response['Error']['Message']:
            return False
        else:
            raise e

def create_stack(stack, template):
    """Starts a new CloudFormation stack creation
    
    Args:
        stack: The stack to be created
        template: The template for the stack to be created with
        
    Throws:
        Exception: Any exception thrown by .create_stack()
    """
    cf.create_stack(StackName=stack, TemplateBody=template)
 
def get_stack_status(stack):
    """Get the status of an existing CloudFormation stack
    
    Args:
        stack: The name of the stack to check
        
    Returns:
        The CloudFormation status string of the stack such as CREATE_COMPLETE
        
    Raises:
        Exception: Any exception thrown by .describe_stacks()
        
    """
    stack_description = cf.describe_stacks(StackName=stack)
    return stack_description['Stacks'][0]['StackStatus']
  
def put_job_success(job, message):
    """Notify CodePipeline of a successful job
    
    Args:
        job: The CodePipeline job ID
        message: A message to be logged relating to the job status
        
    Raises:
        Exception: Any exception thrown by .put_job_success_result()
    
    """
    print('Putting job success')
    print(message)
    code_pipeline.put_job_success_result(jobId=job)
  
def put_job_failure(job, message):
    """Notify CodePipeline of a failed job
    
    Args:
        job: The CodePipeline job ID
        message: A message to be logged relating to the job status
        
    Raises:
        Exception: Any exception thrown by .put_job_failure_result()
    
    """
    print('Putting job failure')
    print(message)
    code_pipeline.put_job_failure_result(jobId=job, failureDetails={'message': message, 'type': 'JobFailed'})
 
def continue_job_later(job, message):
    """Notify CodePipeline of a continuing job
    
    This will cause CodePipeline to invoke the function again with the
    supplied continuation token.
    
    Args:
        job: The JobID
        message: A message to be logged relating to the job status
        continuation_token: The continuation token
        
    Raises:
        Exception: Any exception thrown by .put_job_success_result()
    
    """
    
    # Use the continuation token to keep track of any job execution state
    # This data will be available when a new job is scheduled to continue the current execution
    continuation_token = json.dumps({'previous_job_id': job})
    
    print('Putting job continuation')
    print(message)
    code_pipeline.put_job_success_result(jobId=job, continuationToken=continuation_token)

def start_update_or_create(job_id, stack, template):
    """Starts the stack update or create process
    
    If the stack exists then update, otherwise create.
    
    Args:
        job_id: The ID of the CodePipeline job
        stack: The stack to create or update
        template: The template to create/update the stack with
    
    """
    if stack_exists(stack):
        status = get_stack_status(stack)
        if status not in ['CREATE_COMPLETE', 'ROLLBACK_COMPLETE', 'UPDATE_COMPLETE']:
            # If the CloudFormation stack is not in a state where
            # it can be updated again then fail the job right away.
            put_job_failure(job_id, 'Stack cannot be updated when status is: ' + status)
            return
        
        were_updates = update_stack(stack, template)
        
        if were_updates:
            # If there were updates then continue the job so it can monitor
            # the progress of the update.
            continue_job_later(job_id, 'Stack update started')  
            
        else:
            # If there were no updates then succeed the job immediately 
            put_job_success(job_id, 'There were no stack updates')    
    else:
        # If the stack doesn't already exist then create it instead
        # of updating it.
        create_stack(stack, template)
        # Continue the job so the pipeline will wait for the CloudFormation
        # stack to be created.
        continue_job_later(job_id, 'Stack create started') 

def check_stack_update_status(job_id, stack):
    """Monitor an already-running CloudFormation update/create
    
    Succeeds, fails or continues the job depending on the stack status.
    
    Args:
        job_id: The CodePipeline job ID
        stack: The stack to monitor
    
    """
    status = get_stack_status(stack)
    if status in ['UPDATE_COMPLETE', 'CREATE_COMPLETE']:
        # If the update/create finished successfully then
        # succeed the job and don't continue.
        put_job_success(job_id, 'Stack update complete')
        
    elif status in ['UPDATE_IN_PROGRESS', 'UPDATE_ROLLBACK_IN_PROGRESS', 
    'UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS', 'CREATE_IN_PROGRESS', 
    'ROLLBACK_IN_PROGRESS', 'UPDATE_COMPLETE_CLEANUP_IN_PROGRESS']:
        # If the job isn't finished yet then continue it
        continue_job_later(job_id, 'Stack update still in progress') 
       
    else:
        # If the Stack is a state which isn't "in progress" or "complete"
        # then the stack update/create has failed so end the job with
        # a failed result.
        put_job_failure(job_id, 'Update failed: ' + status)

def get_user_params(job_data):
    """Decodes the JSON user parameters and validates the required properties.
    
    Args:
        job_data: The job data structure containing the UserParameters string which should be a valid JSON structure
        
    Returns:
        The JSON parameters decoded as a dictionary.
        
    Raises:
        Exception: The JSON can't be decoded or a property is missing.
        
    """
    try:
        # Get the user parameters which contain the stack, artifact and file settings
        user_parameters = job_data['actionConfiguration']['configuration']['UserParameters']
        decoded_parameters = json.loads(user_parameters)
            
    except Exception as e:
        # We're expecting the user parameters to be encoded as JSON
        # so we can pass multiple values. If the JSON can't be decoded
        # then fail the job with a helpful message.
        raise Exception('UserParameters could not be decoded as JSON')
    
    if 'stack' not in decoded_parameters:
        # Validate that the stack is provided, otherwise fail the job
        # with a helpful message.
        raise Exception('Your UserParameters JSON must include the stack name')
    
    if 'artifact' not in decoded_parameters:
        # Validate that the artifact name is provided, otherwise fail the job
        # with a helpful message.
        raise Exception('Your UserParameters JSON must include the artifact name')
    
    if 'file' not in decoded_parameters:
        # Validate that the template file is provided, otherwise fail the job
        # with a helpful message.
        raise Exception('Your UserParameters JSON must include the template file name')
    
    return decoded_parameters
    
def setup_s3_client(job_data):
    """Creates an S3 client
    
    Uses the credentials passed in the event by CodePipeline. These
    credentials can be used to access the artifact bucket.
    
    Args:
        job_data: The job data structure
        
    Returns:
        An S3 client with the appropriate credentials
        
    """
    key_id = job_data['artifactCredentials']['accessKeyId']
    key_secret = job_data['artifactCredentials']['secretAccessKey']
    session_token = job_data['artifactCredentials']['sessionToken']
    
    session = Session(aws_access_key_id=key_id,
        aws_secret_access_key=key_secret,
        aws_session_token=session_token)
    return session.client('s3', config=botocore.client.Config(signature_version='s3v4'))

def lambda_handler(event, context):
    """The Lambda function handler
    
    If a continuing job then checks the CloudFormation stack status
    and updates the job accordingly.
    
    If a new job then kick of an update or creation of the target
    CloudFormation stack.
    
    Args:
        event: The event passed by Lambda
        context: The context passed by Lambda
        
    """
    try:
        # Extract the Job ID
        job_id = event['CodePipeline.job']['id']
        
        # Extract the Job Data 
        job_data = event['CodePipeline.job']['data']
        
        # Extract the params
        params = get_user_params(job_data)
        
        # Get the list of artifacts passed to the function
        artifacts = job_data['inputArtifacts']
        
        stack = params['stack']
        artifact = params['artifact']
        template_file = params['file']
        
        if 'continuationToken' in job_data:
            # If we're continuing then the create/update has already been triggered
            # we just need to check if it has finished.
            check_stack_update_status(job_id, stack)
        else:
            # Get the artifact details
            artifact_data = find_artifact(artifacts, artifact)
            # Get S3 client to access artifact with
            s3 = setup_s3_client(job_data)
            # Get the JSON template file out of the artifact
            template = get_template(s3, artifact_data, template_file)
            # Kick off a stack update or create
            start_update_or_create(job_id, stack, template)  

    except Exception as e:
        # If any other exceptions which we didn't expect are raised
        # then fail the job and log the exception message.
        print('Function failed due to exception.') 
        print(e)
        traceback.print_exc()
        put_job_failure(job_id, 'Function exception: ' + str(e))
      
    print('Function complete.')   
    return "Complete."

将处理程序保留为默认值，将角色保留为您之前选择或创建的名称 CodePipelineLambdaExecRole。

在 Basic settings (基本设置) 中，对于 Timeout (超时)，将默认值 3 秒替换为 20 秒。

选择保存。

在 CodePipeline 控制台中，编辑管道，将该函数添加为管道中某个阶段的操作。对要更改的管道选择编辑，然后选择添加操作组。在编辑操作页面的操作名称中，为您的操作输入一个名称。在操作提供程序中，选择 Lambda。

在输入构件中，选择 MyTemplate。在中 UserParameters，必须提供一个JSON包含三个参数的字符串：

使用大括号 ({})，并使用逗号分隔各个参数。例如，创建一个名为的堆栈 MyTestStack，用于带有输入工件的管道 MyTemplateUserParameters，在中输入：{“stack”:”MyTestStack“，“文件”：“template-package/template.json”，“artifact”：”MyTemplate"}.

保存您对管道所做的更改，然后手动发布更改，以便测试操作和 Lambda 函数。