المختبر: إنشاء واختبار وظائف Lambda

هدف المختبر

إنشاء وتكوين واختبار وظائف Lambda مع مشغلات مختلفة. يستخدم هذا المختبر LocalStack لمحاكاة بيئة AWS.

الجزء 1: وظيفة Hello World

تمرين 1.1: إنشاء الوظيفة

الكود المراد إنشاؤه: hello_world.py

import json
from datetime import datetime

def lambda_handler(event, context):
    """
    وظيفة Hello World مع تسجيل
    """
    # تسجيل الحدث المستلم
    print(f"Event received: {json.dumps(event)}")
    print(f"Context: {context.function_name}, {context.aws_request_id}")
    
    # استخراج الاسم
    name = event.get('name', 'World')
    timestamp = datetime.now().isoformat()
    
    # بناء الاستجابة
    response = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        },
        'body': json.dumps({
            'message': f'Hello, {name}!',
            'timestamp': timestamp,
            'function': context.function_name
        })
    }
    
    print(f"Response: {json.dumps(response)}")
    return response

تمرين 1.2: نشر LocalStack

# 1. بدء تشغيل LocalStack (إذا لم يتم ذلك بالفعل)
docker-compose up -d

# 2. إنشاء دور IAM
aws --endpoint-url=http://localhost:4566 iam create-role \
    --role-name lambda-role \
    --assume-role-policy-document '{
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Principal": {"Service": "lambda.amazonaws.com"},
            "Action": "sts:AssumeRole"
        }]
    }'

# 3. ضغط الكود
zip hello_world.zip hello_world.py

# 4. إنشاء الوظيفة
aws --endpoint-url=http://localhost:4566 lambda create-function \
    --function-name hello-world \
    --runtime python3.9 \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --handler hello_world.lambda_handler \
    --zip-file fileb://hello_world.zip

# 5. اختبار
aws --endpoint-url=http://localhost:4566 lambda invoke \
    --function-name hello-world \
    --payload '{"name": "LocalStack"}' \
    response.json

cat response.json

أسئلة

1. ما هو محتوى response.json؟
2. قم بتعديل الكود لإضافة عداد للاستدعاءات. أين ستخزن هذا العداد؟
3. ماذا يحدث إذا قمت بتمرير حمولة غير صالحة (ليست JSON)؟

💡 الحلول

1. محتوى response.json:

{
  "statusCode": 200,
  "headers": {"Content-Type": "application/json"},
  "body": "{\"message\": \"Hello, LocalStack!\", \"timestamp\": \"...\", \"function\": \"hello-world\"}"
}

2. عداد الاستدعاءات:

   • DynamoDB للتخزين الدائم
   • متغير عالمي (إعادة التعيين عند كل بدء بارد)
   • CloudWatch Metrics

3. حمولة غير صالحة:

   • خطأ في تحليل JSON
   • الوظيفة تستقبل السلسلة الخام
   • التعامل مع try/except

الجزء 2: وظيفة مع متغيرات البيئة

تمرين 2.1: التكوين

# config_demo.py
import json
import os

def lambda_handler(event, context):
    """
    عرض توضيحي لمتغيرات البيئة
    """
    # قراءة متغيرات البيئة
    config = {
        'environment': os.environ.get('ENVIRONMENT', 'development'),
        'log_level': os.environ.get('LOG_LEVEL', 'INFO'),
        'api_endpoint': os.environ.get('API_ENDPOINT', 'http://localhost'),
        'max_retries': int(os.environ.get('MAX_RETRIES', '3'))
    }
    
    print(f"Configuration loaded: {json.dumps(config)}")
    
    # محاكاة سلوك مختلف حسب البيئة
    if config['environment'] == 'production':
        message = "Running in PRODUCTION mode"
    else:
        message = f"Running in {config['environment']} mode"
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'message': message,
            'config': config
        })
    }

تمرين 2.2: النشر مع المتغيرات

# إنشاء مع متغيرات البيئة
aws --endpoint-url=http://localhost:4566 lambda create-function \
    --function-name config-demo \
    --runtime python3.9 \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --handler config_demo.lambda_handler \
    --zip-file fileb://config_demo.zip \
    --environment 'Variables={
        ENVIRONMENT=staging,
        LOG_LEVEL=DEBUG,
        API_ENDPOINT=https://api.example.com,
        MAX_RETRIES=5
    }'

# اختبار
aws --endpoint-url=http://localhost:4566 lambda invoke \
    --function-name config-demo \
    response.json

# تعديل متغير
aws --endpoint-url=http://localhost:4566 lambda update-function-configuration \
    --function-name config-demo \
    --environment 'Variables={
        ENVIRONMENT=production,
        LOG_LEVEL=WARN,
        API_ENDPOINT=https://api.example.com,
        MAX_RETRIES=3
    }'

أسئلة

أكمل الجدول:

متغير	قيمة Dev	قيمة Staging	قيمة Prod
ENVIRONMENT	?	?	?
LOG_LEVEL	?	?	?
MAX_RETRIES	?	?	?

💡 الحل

متغير	قيمة Dev	قيمة Staging	قيمة Prod
ENVIRONMENT	development	staging	production
LOG_LEVEL	DEBUG	DEBUG	WARN
MAX_RETRIES	3	5	3

الجزء 3: مشغل S3 (محاكى)

تمرين 3.1: معالج الملفات

# s3_processor.py
import json

def lambda_handler(event, context):
    """
    معالجة أحداث S3 (تحميل الملفات)
    """
    results = []
    
    for record in event.get('Records', []):
        # استخراج معلومات S3
        s3_info = record.get('s3', {})
        bucket = s3_info.get('bucket', {}).get('name', 'unknown')
        key = s3_info.get('object', {}).get('key', 'unknown')
        size = s3_info.get('object', {}).get('size', 0)
        event_type = record.get('eventName', 'unknown')
        
        print(f"Processing: s3://{bucket}/{key}")
        print(f"Event: {event_type}, Size: {size} bytes")
        
        # محاكاة المعالجة
        result = {
            'bucket': bucket,
            'key': key,
            'size': size,
            'event': event_type,
            'status': 'processed'
        }
        
        # منطق شرطي بناءً على نوع الملف
        if key.endswith('.jpg') or key.endswith('.png'):
            result['action'] = 'image_processing'
        elif key.endswith('.json'):
            result['action'] = 'data_import'
        elif key.endswith('.csv'):
            result['action'] = 'csv_processing'
        else:
            result['action'] = 'archive'
        
        results.append(result)
        print(f"Result: {json.dumps(result)}")
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'processed': len(results),
            'results': results
        })
    }

تمرين 3.2: محاكاة حدث S3

# إنشاء ملف حدث S3 محاكى
cat > s3_event.json << 'EOF'
{
  "Records": [
    {
      "eventName": "ObjectCreated:Put",
      "s3": {
        "bucket": {"name": "my-uploads"},
        "object": {
          "key": "images/photo.jpg",
          "size": 1024000
        }
      }
    },
    {
      "eventName": "ObjectCreated:Put",
      "s3": {
        "bucket": {"name": "my-uploads"},
        "object": {
          "key": "data/report.csv",
          "size": 5000
        }
      }
    }
  ]
}
EOF

# نشر واختبار
zip s3_processor.zip s3_processor.py

aws --endpoint-url=http://localhost:4566 lambda create-function \
    --function-name s3-processor \
    --runtime python3.9 \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --handler s3_processor.lambda_handler \
    --zip-file fileb://s3_processor.zip

aws --endpoint-url=http://localhost:4566 lambda invoke \
    --function-name s3-processor \
    --payload file://s3_event.json \
    response.json

cat response.json | python -m json.tool

أسئلة

1. ما الإجراء الذي تم تعيينه لـ photo.jpg؟
2. ما الإجراء الذي تم تعيينه لـ report.csv؟
3. كيف ستعدل الكود لمعالجة الملفات التي تزيد عن 1 ميجابايت بشكل مختلف؟

💡 الحلول

1. photo.jpg → الإجراء: image_processing
2. report.csv → الإجراء: csv_processing
3. إضافة شرط:

if size > 1_000_000:  # > 1 MB
    result['processing_mode'] = 'async'  # معالجة غير متزامنة
else:
    result['processing_mode'] = 'sync'

الجزء 4: إدارة الأخطاء

تمرين 4.1: وظيفة مع أخطاء

# error_handling.py
import json

class ValidationError(Exception):
    """خطأ تحقق مخصص"""
    pass

def validate_input(event):
    """التحقق من صحة بيانات الإدخال"""
    if 'name' not in event:
        raise ValidationError("الحقل 'name' مطلوب")
    if len(event['name']) < 2:
        raise ValidationError("يجب أن يحتوي الاسم على حرفين على الأقل")
    if 'age' in event and not isinstance(event['age'], int):
        raise ValidationError("يجب أن يكون العمر عددًا صحيحًا")
    return True

def lambda_handler(event, context):
    """
    وظيفة مع إدارة كاملة للأخطاء
    """
    print(f"Input: {json.dumps(event)}")
    
    try:
        # التحقق من صحة الإدخال
        validate_input(event)
        
        # المعالجة
        name = event['name']
        age = event.get('age', 'غير محدد')
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'Bonjour {name}!',
                'age': age,
                'status': 'success'
            })
        }
        
    except ValidationError as e:
        print(f"Validation error: {str(e)}")
        return {
            'statusCode': 400,
            'body': json.dumps({
                'error': 'Validation Error',
                'message': str(e),
                'status': 'failed'
            })
        }
        
    except Exception as e:
        print(f"Unexpected error: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': 'Internal Server Error',
                'message': 'حدث خطأ غير متوقع',
                'status': 'failed'
            })
        }

تمرين 4.2: اختبارات السيناريوهات

اختبر مع حمولات مختلفة وأكمل الجدول:

الحمولة	رمز الحالة	نوع الخطأ
{"name": "Alice"}	?	?
{}	?	?
{"name": "A"}	?	?
{"name": "Bob", "age": "vingt"}	?	?
{"name": "Charlie", "age": 25}	?	?

💡 الحلول

الحمولة	رمز الحالة	نوع الخطأ
{"name": "Alice"}	200	لا يوجد
{}	400	ValidationError (name مطلوب)
{"name": "A"}	400	ValidationError (min 2 chars)
{"name": "Bob", "age": "vingt"}	400	ValidationError (age int)
{"name": "Charlie", "age": 25}	200	لا يوجد

الجزء 5: تحليل الأداء

تمرين 5.1: قياس أوقات التنفيذ

# performance_test.py
import json
import time

def lambda_handler(event, context):
    """
    وظيفة لقياس الأداء
    """
    start_time = time.time()
    
    # محاكاة أعباء عمل مختلفة
    workload = event.get('workload', 'light')
    
    if workload == 'light':
        # عمل خفيف
        result = sum(range(1000))
        
    elif workload == 'medium':
        # عمل متوسط
        result = sum(range(100000))
        
    elif workload == 'heavy':
        # عمل ثقيل
        result = sum(range(10000000))
    
    else:
        result = 0
    
    execution_time = (time.time() - start_time) * 1000  # بالمللي ثانية
    remaining_time = context.get_remaining_time_in_millis()
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'workload': workload,
            'result': result,
            'execution_time_ms': round(execution_time, 2),
            'remaining_time_ms': remaining_time,
            'memory_limit_mb': context.memory_limit_in_mb
        })
    }

أسئلة التحليل

1. قم بالتنفيذ مع workload: light، medium، وheavy. لاحظ الأوقات.
2. ماذا يحدث إذا تجاوز وقت التنفيذ المهلة؟
3. كيف يمكنك تحسين وظيفة بطيئة؟

التسليم النهائي

الوثيقة المطلوبة

1. كود الوظائف الأربع (hello، config، s3، errors)
2. نتائج الاختبارات (الحمولات والاستجابات)
3. جدول تحليل الأخطاء
4. توصيات التحسين

معايير التقييم

المعيار	النقاط
الوظائف التي تم إنشاؤها واختبارها	/30
إدارة الأخطاء	/20
متغيرات البيئة	/15
تحليل أحداث S3	/15
جودة الكود	/10
التوثيق	/10
المجموع	/100

---

الأسبوع القادم: MODULE 12 - استكشاف أخطاء Lambda وإصلاحها