基于AWS架构的用户Profile更新功能POC技术实现咨询

Great question! Let’s break down how to implement this end-to-end POC step by step, focusing on simplicity, reliability, and alignment with your existing tech stack.

High-Level Flow Overview #

Here’s the core workflow we’ll build:

1. Web frontend sends a PUT /updateProfile request to AWS API Gateway
2. API Gateway routes the request to an orchestrator Lambda function
3. The orchestrator Lambda:
   • Calls your EC2-hosted REST API to update the Postgres Profile table
   • Puts an audit event (user ID, timestamp, changes, etc.) into a Kinesis Data Stream
4. A second Lambda processes the Kinesis stream and writes the audit record to your Postgres Audit table
5. The orchestrator Lambda returns a success/error response back to the web frontend

Step-by-Step Implementation Guide #

1. Set Up AWS API Gateway #

• Create a new REST API in the AWS Console.
• Add a PUT method to a resource (e.g., /profile) named updateProfile.
• Configure the integration type as Lambda Proxy, pointing to your upcoming ProfileOrchestratorLambda.
• Enable CORS if your web app is hosted on a separate domain.
• Deploy the API to a stage (e.g., dev) to get your public endpoint URL.

2. Create the Profile Orchestrator Lambda #

This Lambda handles the dual tasks of calling your EC2 API and triggering the Kinesis stream. We’ll use Python for example code, but you can adapt it to Java/C# if preferred.

Key Setup:

• Assign an IAM role with permissions:
  • kinesis:PutRecord for your Kinesis stream
  • VPC access (if your EC2 instance is in a private VPC) or outbound internet access (if EC2 is public)
• Set environment variables: EC2_API_URL (your EC2 REST endpoint, e.g., http://ec2-xx-xx-xx-xx.compute-1.amazonaws.com/api/profile) and KINESIS_STREAM_NAME.

Sample Code:

import json
import requests
import boto3
import os

kinesis_client = boto3.client('kinesis')
EC2_API_URL = os.environ['EC2_API_URL']
KINESIS_STREAM_NAME = os.environ['KINESIS_STREAM_NAME']

def lambda_handler(event, context):
    # Parse incoming request from API Gateway
    request_body = json.loads(event['body'])
    user_id = request_body.get('userId')
    updated_fields = request_body.get('updatedFields')

    try:
        # 1. Call EC2 API to update Profile table
        ec2_response = requests.put(
            f"{EC2_API_URL}/{user_id}",
            json=updated_fields,
            timeout=10
        )
        ec2_response.raise_for_status()

        # 2. Prepare and send audit event to Kinesis
        audit_event = {
            'userId': user_id,
            'timestamp': context.get_remaining_time_in_millis(),
            'action': 'PROFILE_UPDATE',
            'changedFields': updated_fields,
            'sourceIp': event['requestContext']['identity']['sourceIp']
        }

        kinesis_client.put_record(
            StreamName=KINESIS_STREAM_NAME,
            Data=json.dumps(audit_event),
            PartitionKey=user_id  # Group events by user for consistency
        )

        return {
            'statusCode': 200,
            'body': json.dumps({'message': 'Profile updated successfully'})
        }
    except requests.exceptions.RequestException as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': f'Failed to update profile: {str(e)}'})
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': f'Unexpected error: {str(e)}'})
        }

3. Configure Kinesis Data Stream #

• Create a new Kinesis Data Stream (use On-Demand mode for POC to avoid capacity planning).
• Add a trigger from this stream to your upcoming AuditProcessorLambda:
  • Set batch size to 10 (adjust as needed)
  • Choose starting position as Latest

4. Create the Audit Processor Lambda #

This Lambda reads from Kinesis and writes audit records to Postgres. Again, Python example below:

Key Setup:

• Assign an IAM role with permissions:
  • kinesis:GetRecords and kinesis:GetShardIterator for your stream
  • VPC access to your Postgres instance (or use RDS Proxy for better security)
• Store DB credentials in AWS Secrets Manager (instead of plain env vars) for production, but use env vars for POC: DB_HOST, DB_NAME, DB_USER, DB_PASSWORD.

Sample Code:

import json
import psycopg2
import os

DB_HOST = os.environ['DB_HOST']
DB_NAME = os.environ['DB_NAME']
DB_USER = os.environ['DB_USER']
DB_PASSWORD = os.environ['DB_PASSWORD']

def lambda_handler(event, context):
    conn = None
    try:
        # Connect to Postgres
        conn = psycopg2.connect(
            host=DB_HOST,
            database=DB_NAME,
            user=DB_USER,
            password=DB_PASSWORD
        )
        cur = conn.cursor()

        # Process each Kinesis record
        for record in event['Records']:
            audit_event = json.loads(record['kinesis']['data'])
            
            cur.execute("""
                INSERT INTO audit (user_id, action, changed_fields, timestamp, source_ip)
                VALUES (%s, %s, %s, TO_TIMESTAMP(%s/1000), %s)
            """, (
                audit_event['userId'],
                audit_event['action'],
                json.dumps(audit_event['changedFields']),
                audit_event['timestamp'],
                audit_event['sourceIp']
            ))
        
        conn.commit()
        cur.close()
        return {'statusCode': 200, 'body': 'Audit records processed successfully'}
    except Exception as e:
        if conn:
            conn.rollback()
        return {'statusCode': 500, 'body': f'Error processing audit records: {str(e)}'}
    finally:
        if conn:
            conn.close()

5. EC2 REST API Setup #

• Ensure your EC2 instance is accessible to the orchestrator Lambda:
  • If Lambda is in a VPC, place EC2 in the same VPC and allow inbound traffic from Lambda’s security group.
  • If Lambda is public, assign EC2 a public IP and allow inbound traffic from Lambda’s public IP range.
• Your Java/C# API should expose a PUT /profile/{userId} endpoint that updates the Postgres Profile table, returning appropriate HTTP status codes (200 for success, 404 if user not found).

Key Considerations for POC & Production #

• Reliability: For POC, accept eventual consistency (if Kinesis fails, retry in Lambda). For production, use AWS Step Functions to orchestrate tasks with retries and dead-letter queues.
• Security: Add Cognito authentication to API Gateway to secure user profile updates. Use VPC endpoints for all AWS services to keep traffic within AWS’s network.
• Monitoring: Enable CloudWatch Logs for all Lambdas and API Gateway. Set up alarms for failed invocations or Kinesis stream backlogs.
• Scalability: Kinesis and Lambda auto-scale, but for production, use an EC2 Auto Scaling Group for your REST API.

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