阿里云全栈实战（八）：Serverless 与事件驱动

1
2
3
4
5
6
7
8
9

# Install the s CLI
npm install -g @serverless-devs/s

# Configure credentials
s config add \
  --AccountID <your-account-id> \
  --AccessKeyID <your-ak> \
  --AccessKeySecret <your-sk> \
  -a default

1
2
3
4
5
6

image-processor/
  ├── s.yaml              # Serverless Devs config
  ├── code/
  │   ├── index.py        # Function handler
  │   └── requirements.txt
  └── README.md

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# code/index.py
import json
import logging

logger = logging.getLogger()

def handler(event, context):
    """
    FC handler for HTTP trigger.
    
    Args:
        event: The HTTP request body (bytes).
        context: FC context object with request ID, credentials, 
                 function name, memory limit, etc.
    
    Returns:
        dict with statusCode, headers, and body for HTTP response.
    """
    logger.info(f"Request ID: {context.request_id}")
    logger.info(f"Function: {context.function.name}")
    logger.info(f"Memory limit: {context.function.memory_size} MiB")
    
    # Parse the event (HTTP trigger sends the request body)
    try:
        body = json.loads(event)
    except (json.JSONDecodeError, TypeError):
        body = {}
    
    name = body.get("name", "World")
    
    response = {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json"
        },
        "body": json.dumps({
            "message": f"Hello, {name}!",
            "request_id": context.request_id
        })
    }
    
    return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# s.yaml
edition: 3.0.0
name: image-processor
access: default

resources:
  hello:
    component: fc3
    props:
      region: cn-beijing
      functionName: 入门示例
      description: "Simple hello world function"
      runtime: python3.10
      handler: index.handler
      memorySize: 256
      timeout: 30
      code: ./code
      triggers:
        - triggerName: http-trigger
          triggerType: http
          triggerConfig:
            authType: anonymous
            methods:
              - GET
              - POST

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# Deploy the function
cd image-processor
s deploy

# Invoke locally (for development)
s local invoke -e '{"name": "Alibaba Cloud"}'

# Invoke remotely (the deployed function)
s invoke -e '{"name": "Alibaba Cloud"}'

# Get the HTTP trigger URL
s info
# Output includes: url: https://入门示例-xxxx.cn-beijing.fcapp.run

1
2
3
4
5
6
7
8

# Test the deployed function
curl -X POST \
  https://入门示例-xxxx.cn-beijing.fcapp.run \
  -H "Content-Type: application/json" \
  -d '{"name": "Serverless"}'

# Response:
# {"message": "Hello, Serverless!", "request_id": "1-6789abcd-..."}

1
2
3
4
5

# View recent function logs
s logs --tail

# View logs for a specific request
s logs --request-id "1-6789abcd-..."

1
2
3
4
5
6
7
8
9

# Create via CLI
aliyun fc CreateTrigger \
  --functionName my-api \
  --triggerName http-trigger \
  --triggerType http \
  --triggerConfig '{
    "authType": "anonymous",
    "methods": ["GET", "POST", "PUT", "DELETE"]
  }'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Trigger on file uploads to a specific prefix
aliyun fc CreateTrigger \
  --functionName image-processor \
  --triggerName oss-upload \
  --triggerType oss \
  --triggerConfig '{
    "bucketName": "my-upload-bucket",
    "events": ["oss:ObjectCreated:PutObject", "oss:ObjectCreated:PostObject"],
    "filter": {
      "key": {
        "prefix": "uploads/",
        "suffix": ".jpg"
      }
    }
  }'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

{
  "events": [
    {
      "eventName": "ObjectCreated:PutObject",
      "eventSource": "acs:oss",
      "eventTime": "2026-05-22T08:30:00.000Z",
      "region": "cn-beijing",
      "oss": {
        "bucket": {
          "name": "my-upload-bucket"
        },
        "object": {
          "key": "uploads/photo-001.jpg",
          "size": 2048576
        }
      }
    }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# Run every day at 02:00 UTC+8
aliyun fc CreateTrigger \
  --functionName daily-report \
  --triggerName daily-cron \
  --triggerType timer \
  --triggerConfig '{
    "cronExpression": "0 0 2 * * *",
    "enable": true,
    "payload": "{\"report_type\": \"daily_summary\"}"
  }'

# Run every 5 minutes
aliyun fc CreateTrigger \
  --functionName health-checker \
  --triggerName every-5min \
  --triggerType timer \
  --triggerConfig '{
    "cronExpression": "@every 5m",
    "enable": true
  }'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

aliyun fc CreateTrigger \
  --functionName log-analyzer \
  --triggerName sls-trigger \
  --triggerType log \
  --triggerConfig '{
    "sourceConfig": {
      "logstore": "access-log"
    },
    "jobConfig": {
      "maxRetryTime": 3,
      "triggerInterval": 60
    },
    "logConfig": {
      "project": "my-log-project",
      "logstore": "fc-error-log"
    },
    "functionParameter": {
      "alert_threshold": 100
    },
    "enable": true
  }'

1
2
3
4
5
6
7
8
9

aliyun fc CreateTrigger \
  --functionName order-processor \
  --triggerName mns-queue \
  --triggerType mns_topic \
  --triggerConfig '{
    "topicName": "order-events",
    "region": "cn-beijing",
    "filterTag": "new-order"
  }'

1
2
3
4
5
6
7
8
9

aliyun fc CreateTrigger \
  --functionName event-handler \
  --triggerName eb-trigger \
  --triggerType eventbridge \
  --triggerConfig '{
    "triggerEnable": true,
    "asyncInvocationType": false,
    "eventRuleFilterPattern": "{\"source\":[\"acs.oss\"],\"type\":[\"oss:BucketCreated\"]}"
  }'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# code/index.py
import time
import json

# Module-level code runs during cold start
INIT_TIME = time.time()
print(f"Cold start initialization at {INIT_TIME}")

# Expensive imports happen here
import numpy as np
from PIL import Image

INIT_DURATION = time.time() - INIT_TIME
print(f"Init took {INIT_DURATION:.3f}s")

def handler(event, context):
    start = time.time()
    
    # Your business logic here
    result = {"cold_start_init_ms": round(INIT_DURATION * 1000)}
    
    duration = time.time() - start
    result["handler_ms"] = round(duration * 1000)
    
    return {
        "statusCode": 200,
        "body": json.dumps(result)
    }

1
2
3
4
5

# Keep 5 instances always warm
aliyun fc PutProvisionConfig \
  --functionName image-processor \
  --qualifier LATEST \
  --target 5

1
2
3
4
5
6

# pre-warm function (called by timer trigger)
def handler(event, context):
    """
    Do nothing — the point is to create warm instances.
    """
    return {"statusCode": 200, "body": "warm"}

1
2
3
4
5
6
7

# Timer trigger: warm up at 8:55 AM every weekday
triggers:
  - triggerName: pre-warm
    triggerType: timer
    triggerConfig:
      cronExpression: "0 55 8 ? * MON-FRI"
      enable: true

1
2
3
4
5
6
7
8

# Bad: 150 MiB package with all of scipy/numpy/pandas
pip install -r requirements.txt -t ./code/

# Good: only install what you need
pip install Pillow requests -t ./code/ --no-cache-dir

# Better: use layers for shared dependencies
# (covered in the next section)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# Bad: connects to DB on every cold start, even if the request
# doesn't need it
import pymysql
conn = pymysql.connect(host="...", db="...")  # Runs on cold start

def handler(event, context):
    cursor = conn.cursor()
    ...

# Good: lazy initialization
_conn = None

def get_connection():
    global _conn
    if _conn is None or not _conn.open:
        _conn = pymysql.connect(host="...", db="...")
    return _conn

def handler(event, context):
    conn = get_connection()
    cursor = conn.cursor()
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# Create a directory matching the runtime's expected path
mkdir -p layer/python/lib/python3.10/site-packages

# Install dependencies into the layer
pip install Pillow numpy requests \
  -t layer/python/lib/python3.10/site-packages \
  --no-cache-dir

# Zip it
cd layer && zip -r ../image-processing-layer.zip . && cd ..

# Create the layer in FC
aliyun fc CreateLayerVersion \
  --layerName image-processing-deps \
  --code '{"zipFile":"'$(base64 -w0 image-processing-layer.zip)'"}'  \
  --compatibleRuntime '["python3.10"]' \
  --description "Pillow + numpy + requests for image processing"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# s.yaml
resources:
  image-processor:
    component: fc3
    props:
      region: cn-beijing
      functionName: image-processor
      runtime: python3.10
      handler: index.handler
      memorySize: 1024
      timeout: 120
      code: ./code
      layers:
        - acs:fc:cn-beijing:123456789:layers/image-processing-deps/versions/1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# Dockerfile for a Rust-based function
FROM rust:1.78-slim as builder
WORKDIR /app
COPY . .
RUN cargo build --release

FROM debian:bookworm-slim
COPY --from=builder /app/target/release/my-function /app/bootstrap
EXPOSE 9000
CMD ["/app/bootstrap"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# s.yaml for custom container
resources:
  rust-function:
    component: fc3
    props:
      region: cn-beijing
      functionName: rust-processor
      runtime: custom-container
      handler: not-used
      memorySize: 512
      timeout: 60
      customContainerConfig:
        image: registry.cn-beijing.aliyuncs.com/my-repo/rust-function:latest
        port: 9000

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

{
  "id": "a]b-1234-efgh-5678",
  "source": "acs.oss",
  "type": "oss:ObjectCreated:PutObject",
  "specversion": "1.0",
  "datacontenttype": "application/json",
  "time": "2026-05-22T08:30:00Z",
  "subject": "acs:oss:cn-beijing:123456789:my-bucket",
  "data": {
    "region": "cn-beijing",
    "bucket": {
      "name": "my-upload-bucket"
    },
    "object": {
      "key": "uploads/photo-001.jpg",
      "size": 2048576,
      "eTag": "abc123..."
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# Create a custom event bus
aliyun eventbridge CreateEventBus \
  --EventBusName my-app-bus \
  --Description "Custom events for my application"

# Create a rule: route OSS uploads to a Function Compute function
aliyun eventbridge CreateRule \
  --EventBusName default \
  --RuleName oss-upload-to-fc \
  --Description "Route OSS uploads to image processor" \
  --FilterPattern '{
    "source": ["acs.oss"],
    "type": ["oss:ObjectCreated:PutObject", "oss:ObjectCreated:PostObject"],
    "subject": ["acs:oss:cn-beijing:*:my-upload-bucket"]
  }' \
  --Targets '[{
    "Id": "fc-image-processor",
    "Type": "acs.fc.function",
    "Endpoint": "acs:fc:cn-beijing:123456789:functions/image-processor",
    "ParamList": [
      {
        "ResourceKey": "Body",
        "Form": "ORIGINAL"
      }
    ]
  }]'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
  "source": ["acs.oss"],
  "type": [{"prefix": "oss:ObjectCreated"}],
  "data": {
    "object": {
      "size": [{"numeric": [">", 1048576]}],
      "key": [{"prefix": "uploads/"}, {"suffix": ".jpg"}]
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# Create a rule with event transformation
aliyun eventbridge CreateRule \
  --EventBusName default \
  --RuleName transform-example \
  --FilterPattern '{"source": ["acs.oss"]}' \
  --Targets '[{
    "Id": "fc-target",
    "Type": "acs.fc.function",
    "Endpoint": "acs:fc:cn-beijing:123456789:functions/my-func",
    "ParamList": [
      {
        "ResourceKey": "Body",
        "Form": "TEMPLATE",
        "Template": "{\"bucket\":\"$.data.bucket.name\",\"key\":\"$.data.object.key\",\"size\":$.data.object.size}",
        "Value": "{\"bucket\":\"$.data.bucket.name\",\"key\":\"$.data.object.key\",\"size\":\"$.data.object.size\"}"
      }
    ]
  }]'

1
2
3
4
5

{
  "bucket": "my-upload-bucket",
  "key": "uploads/photo-001.jpg",
  "size": 2048576
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from alibabacloud_eventbridge.client import Client
from alibabacloud_eventbridge.models import CloudEvent

client = Client(
    access_key_id="<ak>",
    access_key_secret="<sk>",
    endpoint="cn-beijing.eventbridge.aliyuncs.com"
)

event = CloudEvent(
    source="my.application",
    type="order.created",
    data='{"order_id": "ORD-12345", "amount": 99.99, "currency": "CNY"}',
    subject="acs:my-app:cn-beijing:123456789:orders/ORD-12345"
)

client.put_events(event_bus_name="my-app-bus", event_list=[event])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# Step 1: Create an API Group
aliyun cloudapi CreateApiGroup \
  --RegionId cn-beijing \
  --GroupName "my-serverless-api" \
  --Description "Serverless API backed by FC"

# Step 2: Create an API definition
aliyun cloudapi CreateApi \
  --RegionId cn-beijing \
  --GroupId <group-id> \
  --ApiName "GetUser" \
  --Visibility PUBLIC \
  --AuthType APP \
  --RequestConfig '{
    "RequestProtocol": "HTTPS",
    "RequestHttpMethod": "GET",
    "RequestPath": "/users/[userId]",
    "RequestMode": "MAPPING"
  }' \
  --ServiceConfig '{
    "ServiceProtocol": "FunctionCompute",
    "FunctionComputeConfig": {
      "RegionId": "cn-beijing",
      "FunctionName": "get-user",
      "Qualifier": "LATEST",
      "RoleArn": "acs:ram::123456789:role/api-gateway-fc"
    }
  }' \
  --ResultType JSON \
  --ResultSample '{"user_id": "123", "name": "Test User"}'

1
2
3
4
5
6
7
8

# Bind a custom domain to your API Group
aliyun cloudapi SetDomain \
  --RegionId cn-beijing \
  --GroupId <group-id> \
  --DomainName api.example.com \
  --CertificateName my-cert \
  --CertificateBody "$(cat cert.pem)" \
  --CertificatePrivateKey "$(cat key.pem)"

1
2
3
4
5
6
7
8
9

# Create a throttling policy
aliyun cloudapi CreateTrafficControl \
  --RegionId cn-beijing \
  --TrafficControlName "default-limit" \
  --ApiDefault 1000 \
  --UserDefault 100 \
  --AppDefault 200 \
  --TrafficControlUnit MINUTE \
  --Description "1000 req/min total, 100 per user, 200 per app"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

def handler(event, context):
    # Handle CORS preflight
    http_method = event.get("httpMethod", "GET")
    
    cors_headers = {
        "Access-Control-Allow-Origin": "https://example.com",
        "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type, Authorization",
        "Access-Control-Max-Age": "86400"
    }
    
    if http_method == "OPTIONS":
        return {
            "statusCode": 204,
            "headers": cors_headers,
            "body": ""
        }
    
    # Normal request handling
    result = {"message": "Hello from API"}
    
    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
            **cors_headers
        },
        "body": json.dumps(result)
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

  User uploads image to OSS
           │
           ▼
    ┌──────────────┐
    │     OSS      │  (source bucket: upload-images)
    │  PutObject   │
    └──────┬───────┘
           │  CloudEvent
           ▼
    ┌──────────────┐
    │  EventBridge │  (filter: *.jpg, *.png, > 10KB)
    │  Event Rule  │
    └──────┬───────┘
           │  Invoke
           ▼
    ┌──────────────┐
    │   Function   │  (image-processor)
    │   Compute    │  - Resize to 1200px, 600px, 150px
    │              │  - Add watermark
    │              │  - Generate WebP variants
    └──────┬───────┘
           │  PutObject
           ▼
    ┌──────────────┐
    │     OSS      │  (target bucket: processed-images)
    │   /resized/  │
    │   /thumbs/   │
    │   /webp/     │
    └──────────────┘

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

# code/index.py
import json
import logging
import os
import io
from urllib.parse import unquote

import oss2
from PIL import Image, ImageDraw, ImageFont

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

# Configuration
SOURCE_BUCKET = os.environ.get("SOURCE_BUCKET", "upload-images")
TARGET_BUCKET = os.environ.get("TARGET_BUCKET", "processed-images")
OSS_ENDPOINT = os.environ.get("OSS_ENDPOINT", "https://oss-cn-beijing-internal.aliyuncs.com")
WATERMARK_TEXT = os.environ.get("WATERMARK_TEXT", "chenk.top")

# Resize configurations
SIZES = {
    "large": 1200,
    "medium": 600,
    "thumb": 150
}

def get_oss_client(context, bucket_name):
    """Create OSS client using FC's temporary STS credentials."""
    creds = context.credentials
    auth = oss2.StsAuth(
        creds.access_key_id,
        creds.access_key_secret,
        creds.security_token
    )
    return oss2.Bucket(auth, OSS_ENDPOINT, bucket_name)

def add_watermark(image, text):
    """Add a semi-transparent text watermark to the bottom-right corner."""
    draw = ImageDraw.Draw(image)
    
    # Use a size proportional to the image
    font_size = max(20, image.width // 30)
    try:
        font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", font_size)
    except (IOError, OSError):
        font = ImageFont.load_default()
    
    # Calculate position (bottom-right with padding)
    bbox = draw.textbbox((0, 0), text, font=font)
    text_width = bbox[2] - bbox[0]
    text_height = bbox[3] - bbox[1]
    padding = 20
    x = image.width - text_width - padding
    y = image.height - text_height - padding
    
    # Draw with semi-transparent white
    draw.text((x, y), text, fill=(255, 255, 255, 128), font=font)
    
    return image

def resize_image(image, max_width):
    """Resize image maintaining aspect ratio."""
    if image.width <= max_width:
        return image.copy()
    
    ratio = max_width / image.width
    new_height = int(image.height * ratio)
    return image.resize((max_width, new_height), Image.LANCZOS)

def process_image(source_client, target_client, object_key):
    """Download, process, and upload image variants."""
    logger.info(f"Processing: {object_key}")
    
    # Download the original image
    result = source_client.get_object(object_key)
    image_data = result.read()
    image = Image.open(io.BytesIO(image_data))
    
    # Convert to RGBA for watermark support
    if image.mode != "RGBA":
        image = image.convert("RGBA")
    
    # Extract the filename without extension
    base_name = os.path.splitext(os.path.basename(object_key))[0]
    results = []
    
    for size_name, max_width in SIZES.items():
        # Resize
        resized = resize_image(image, max_width)
        
        # Add watermark (skip for thumbnails)
        if size_name != "thumb":
            resized = add_watermark(resized, WATERMARK_TEXT)
        
        # Convert to RGB for JPEG output
        rgb_image = resized.convert("RGB")
        
        # Save as JPEG
        jpeg_buffer = io.BytesIO()
        rgb_image.save(jpeg_buffer, format="JPEG", quality=85, optimize=True)
        jpeg_buffer.seek(0)
        
        jpeg_key = f"resized/{size_name}/{base_name}.jpg"
        target_client.put_object(jpeg_key, jpeg_buffer.getvalue())
        results.append({"key": jpeg_key, "format": "jpeg", "size": size_name})
        logger.info(f"  Uploaded: {jpeg_key}")
        
        # Save as WebP
        webp_buffer = io.BytesIO()
        resized.save(webp_buffer, format="WebP", quality=80)
        webp_buffer.seek(0)
        
        webp_key = f"webp/{size_name}/{base_name}.webp"
        target_client.put_object(webp_key, webp_buffer.getvalue())
        results.append({"key": webp_key, "format": "webp", "size": size_name})
        logger.info(f"  Uploaded: {webp_key}")
    
    return results

def handler(event, context):
    """
    EventBridge trigger handler.
    Receives CloudEvent when an image is uploaded to OSS.
    """
    logger.info(f"Event received: {event}")
    
    # Parse the CloudEvent from EventBridge
    evt = json.loads(event)
    
    # Extract bucket and object key from the event
    data = evt.get("data", {})
    bucket_name = data.get("bucket", {}).get("name", SOURCE_BUCKET)
    object_key = unquote(data.get("object", {}).get("key", ""))
    
    if not object_key:
        logger.error("No object key in event")
        return {"statusCode": 400, "body": "Missing object key"}
    
    # Validate file extension
    ext = os.path.splitext(object_key)[1].lower()
    if ext not in (".jpg", ".jpeg", ".png", ".webp"):
        logger.info(f"Skipping non-image file: {object_key}")
        return {"statusCode": 200, "body": "Skipped: not an image"}
    
    # Create OSS clients
    source_client = get_oss_client(context, bucket_name)
    target_client = get_oss_client(context, TARGET_BUCKET)
    
    # Process the image
    try:
        results = process_image(source_client, target_client, object_key)
        
        response = {
            "statusCode": 200,
            "body": json.dumps({
                "source": object_key,
                "processed": results,
                "count": len(results)
            })
        }
        logger.info(f"Processed {len(results)} variants for {object_key}")
        return response
        
    except Exception as e:
        logger.error(f"Error processing {object_key}: {str(e)}", exc_info=True)
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e), "source": object_key})
        }

1
2
3

# code/requirements.txt
Pillow>=10.0.0
oss2>=2.18.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# s.yaml
edition: 3.0.0
name: image-pipeline
access: default

resources:
  image-processor:
    component: fc3
    props:
      region: cn-beijing
      functionName: image-processor
      description: "Event-driven image processing pipeline"
      runtime: python3.10
      handler: index.handler
      memorySize: 1024
      timeout: 120
      code: ./code
      layers:
        - acs:fc:cn-beijing:123456789:layers/image-processing-deps/versions/1
      environmentVariables:
        SOURCE_BUCKET: upload-images
        TARGET_BUCKET: processed-images
        OSS_ENDPOINT: https://oss-cn-beijing-internal.aliyuncs.com
        WATERMARK_TEXT: "chenk.top"
      role: acs:ram::123456789:role/fc-oss-access
      triggers:
        - triggerName: eb-oss-upload
          triggerType: eventbridge
          triggerConfig:
            triggerEnable: true
            asyncInvocationType: true
            eventRuleFilterPattern: >
              {
                "source": ["acs.oss"],
                "type": [
                  "oss:ObjectCreated:PutObject",
                  "oss:ObjectCreated:PostObject",
                  "oss:ObjectCreated:CompleteMultipartUpload"
                ]
              }              

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

# main.tf — Infrastructure for the image processing pipeline

terraform {
  required_providers {
    alicloud = {
      source  = "aliyun/alicloud"
      version = "~> 1.220"
    }
  }
}

provider "alicloud" {
  region = "cn-beijing"
}

# --- OSS Buckets ---

resource "alicloud_oss_bucket" "source" {
  bucket = "upload-images-${var.env}"
  acl    = "private"

  cors_rule {
    allowed_origins = ["https://example.com"]
    allowed_methods = ["PUT", "POST"]
    allowed_headers = ["*"]
    max_age_seconds = 3600
  }

  lifecycle_rule {
    id      = "cleanup-uploads"
    enabled = true
    prefix  = "uploads/"

    expiration {
      days = 30
    }
  }
}

resource "alicloud_oss_bucket" "target" {
  bucket = "processed-images-${var.env}"
  acl    = "private"

  lifecycle_rule {
    id      = "archive-old"
    enabled = true

    transition {
      days          = 90
      storage_class = "IA"
    }
  }
}

# --- RAM Role for Function Compute ---

resource "alicloud_ram_role" "fc_role" {
  name     = "fc-image-processor-role"
  document = jsonencode({
    Version   = "1"
    Statement = [
      {
        Action    = "sts:AssumeRole"
        Effect    = "Allow"
        Principal = {
          Service = ["fc.aliyuncs.com"]
        }
      }
    ]
  })
}

resource "alicloud_ram_policy" "fc_oss_policy" {
  policy_name = "fc-oss-readwrite"
  policy_document = jsonencode({
    Version   = "1"
    Statement = [
      {
        Action = [
          "oss:GetObject",
          "oss:PutObject",
          "oss:ListObjects"
        ]
        Effect   = "Allow"
        Resource = [
          "acs:oss:*:*:upload-images-${var.env}/*",
          "acs:oss:*:*:processed-images-${var.env}/*"
        ]
      }
    ]
  })
}

resource "alicloud_ram_role_policy_attachment" "fc_oss" {
  role_name   = alicloud_ram_role.fc_role.name
  policy_name = alicloud_ram_policy.fc_oss_policy.policy_name
  policy_type = "Custom"
}

# --- Function Compute ---

resource "alicloud_fcv3_function" "image_processor" {
  function_name = "image-processor"
  description   = "Event-driven image processing pipeline"
  handler       = "index.handler"
  runtime       = "python3.10"
  memory_size   = 1024
  timeout       = 120
  role          = alicloud_ram_role.fc_role.arn

  code {
    zip_file = filebase64("${path.module}/code.zip")
  }

  environment_variables = {
    SOURCE_BUCKET  = alicloud_oss_bucket.source.bucket
    TARGET_BUCKET  = alicloud_oss_bucket.target.bucket
    OSS_ENDPOINT   = "https://oss-cn-beijing-internal.aliyuncs.com"
    WATERMARK_TEXT  = "chenk.top"
  }
}

# --- EventBridge Rule ---

resource "alicloud_event_bridge_rule" "oss_to_fc" {
  event_bus_name = "default"
  rule_name      = "oss-upload-to-image-processor"
  description    = "Route OSS uploads to image processor function"

  filter_pattern = jsonencode({
    source  = ["acs.oss"]
    type    = [
      "oss:ObjectCreated:PutObject",
      "oss:ObjectCreated:PostObject",
      "oss:ObjectCreated:CompleteMultipartUpload"
    ]
    subject = ["acs:oss:cn-beijing:*:${alicloud_oss_bucket.source.bucket}"]
  })

  targets {
    target_id = "fc-image-processor"
    type      = "acs.fc.function"
    endpoint  = alicloud_fcv3_function.image_processor.arn

    param_list {
      resource_key = "Body"
      form         = "ORIGINAL"
    }
  }
}

# --- Variables ---

variable "env" {
  description = "Environment name (dev, staging, prod)"
  type        = string
  default     = "dev"
}

# --- Outputs ---

output "source_bucket" {
  value = alicloud_oss_bucket.source.bucket
}

output "target_bucket" {
  value = alicloud_oss_bucket.target.bucket
}

output "function_name" {
  value = alicloud_fcv3_function.image_processor.function_name
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# Deploy infrastructure with Terraform
cd terraform
terraform init
terraform plan -var="env=dev"
terraform apply -var="env=dev" -auto-approve

# Package and deploy the function
cd ../image-pipeline
pip install -r code/requirements.txt -t code/ --no-cache-dir
s deploy

# Test: upload an image to the source bucket
aliyun oss cp test-image.jpg \
  oss://upload-images-dev/uploads/test-image.jpg \
  --endpoint oss-cn-beijing.aliyuncs.com

# Wait a few seconds for the pipeline to process
sleep 5

# Verify the processed images exist
aliyun oss ls oss://processed-images-dev/resized/ \
  --endpoint oss-cn-beijing.aliyuncs.com

aliyun oss ls oss://processed-images-dev/webp/ \
  --endpoint oss-cn-beijing.aliyuncs.com

# Check function logs
s logs --tail

1
2
3
4
5
6

resized/large/test-image.jpg      (1200px wide, watermarked)
resized/medium/test-image.jpg     (600px wide, watermarked)
resized/thumb/test-image.jpg      (150px wide, no watermark)
webp/large/test-image.webp        (1200px wide, watermarked)
webp/medium/test-image.webp       (600px wide, watermarked)
webp/thumb/test-image.webp        (150px wide, no watermark)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# Create an MNS queue for failed events
aliyun mns CreateQueue \
  --QueueName image-processor-dlq \
  --DelaySeconds 0 \
  --MaximumMessageSize 65536 \
  --MessageRetentionPeriod 1209600 \
  --VisibilityTimeout 60

# Configure async invocation with retry and DLQ
aliyun fc PutAsyncInvokeConfig \
  --functionName image-processor \
  --qualifier LATEST \
  --destinationConfig '{
    "onFailure": {
      "destination": "acs:mns:cn-beijing:123456789:/queues/image-processor-dlq/messages"
    }
  }' \
  --maxAsyncRetryAttempts 2 \
  --maxAsyncEventAgeInSeconds 3600

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

Invocations:
  5,000,000 - 1,000,000 (free) = 4,000,000 billable
  4,000,000 / 10,000 * 0.0133 = 5.32 CNY

Execution duration:
  5,000,000 * 0.2s * 0.5 GiB = 500,000 GiB-seconds
  500,000 - 400,000 (free) = 100,000 billable
  100,000 * 0.000110592 = 11.06 CNY

Total: 5.32 + 11.06 = 16.38 CNY/month