اللاخوادم والعمليات المدفوعة بالأحداث

مشروع: خط أنابيب مدفوع بالأحداث

18 دقيقة الدرس 10 من 28

مشروع: خط أنابيب مدفوع بالأحداث

يجمع هذا الدرس الختامي كل مفاهيم البرنامج التعليمي — Lambda ومصادر الأحداث وDLQs والرصد والبنية التحتية كرمز وStep Functions — في نظام واحد جاهز للإنتاج. الهدف: خط أنابيب لمعالجة الصور يقبل الرفع عبر S3، ويمرره عبر ثلاث مراحل غير متزامنة، ويحفظ البيانات الوصفية في DynamoDB، ويوصل النتائج إلى المستهلكين في المصب عبر SNS. كل حالة فشل تُلتقط في قائمة انتظار الرسائل الميتة وتُرفع تنبيهاتها. كل تنفيذ قابل للتتبع من البداية إلى النهاية.

هذا هو نمط التصميم وراء الأنظمة الحقيقية على نطاق واسع: خط استيعاب الأصول في Netflix، ومعالجة الوسائط في Shopify، وتنسيق Rekognition في AWS نفسها — كلها تتبع متغيرات من هذه البنية. نفس الهيكل ينطبق على ترميز الفيديو، وتحليل المستندات، وسلاسل استنتاج التعلم الآلي، ومعالجة الأحداث المالية.

بنية خط الأنابيب

يحتوي خط الأنابيب على خمس طبقات منطقية:

الاستيعاب — يُشغّل حدث S3 PutObject دالة Lambda المنسِّقة عبر EventBridge Pipes، التي تبدأ تنفيذ Step Functions Express Workflow لكل ملف.
التحقق — تتحقق الحالة الأولى من نوع MIME وحجم الملف (أقل من 50 ميغابايت) وتجزئة مكافحة الفيروسات عبر استدعاء Lambda متزامن؛ تُوجَّه عمليات الرفض إلى DLQ.
التحويل — تُوزّع حالة Map ثلاث دوال Lambda للمعالجة المتوازية (الصور المصغرة، والعلامة المائية، واستخراج البيانات الوصفية) بقيمة MaxConcurrency: 10.
الحفظ — تكتب تكامل SDK سجل البيانات الوصفية المُخصَّبة في DynamoDB في عملية conditional-put واحدة؛ لا يلزم أي Lambda لهذه الخطوة.
النشر — يُخطر التوزيع المتشعب عبر SNS المشتركين في المصب (إبطال CDN، وفهرسة البحث، وسجل التدقيق).

خط الأنابيب الكامل المدفوع بالأحداث: يتدفق حدث S3 عبر EventBridge Pipes إلى Step Functions Express Workflow؛ تُوزَّع الملفات المُتحقق منها على دوال Lambda متوازية؛ تُحفظ النتائج في DynamoDB وSNS؛ تُوجَّه جميع مسارات الفشل إلى DLQs؛ يلتقط CloudWatch السجلات والمقاييس والتتبعات الموزعة عبر كل طبقة.

البنية التحتية كرمز — أساس Terraform

يُعرَّف خط الأنابيب بالكامل في Terraform. نقاط التصميم الحرجة: دور تنفيذ Step Functions له أدنى صلاحية لكل مورد؛ تشترك دوال Lambda في مجموعة أمان واحدة لكنها تستخدم أدوار IAM لكل دالة؛ تمتلك DLQs قيمة message_retention_seconds = 1209600 (14 يومًا، الحد الأقصى) حتى لا تُفقد أي رسالة فاشلة بصمت قبل أن تتمكن من التحقيق فيها.

# pipeline.tf — الموارد الأساسية (تم حذف حاوية الحالة وقفل DynamoDB للإيجاز)

locals {
  prefix = "img-pipeline-${var.env}"
}

# ──────────────────────────────────────────────
# Dead Letter Queues
# ──────────────────────────────────────────────
resource "aws_sqs_queue" "validation_dlq" {
  name                       = "${local.prefix}-validation-dlq"
  message_retention_seconds  = 1209600
  kms_master_key_id          = aws_kms_key.pipeline.arn
  tags                       = local.common_tags
}

resource "aws_sqs_queue" "transform_dlq" {
  name                       = "${local.prefix}-transform-dlq"
  message_retention_seconds  = 1209600
  kms_master_key_id          = aws_kms_key.pipeline.arn
  tags                       = local.common_tags
}

# تنبيه CloudWatch: أي رسالة في DLQ = تنبيه PagerDuty
resource "aws_cloudwatch_metric_alarm" "dlq_not_empty" {
  for_each            = toset(["validation", "transform"])
  alarm_name          = "${local.prefix}-${each.key}-dlq-not-empty"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = 60
  statistic           = "Sum"
  threshold           = 0
  alarm_actions       = [var.pagerduty_sns_arn]
  dimensions = {
    QueueName = "${local.prefix}-${each.key}-dlq"
  }
}

# ──────────────────────────────────────────────
# دوال Lambda
# ──────────────────────────────────────────────
resource "aws_lambda_function" "validate" {
  function_name = "${local.prefix}-validate"
  filename      = data.archive_file.validate_zip.output_path
  handler       = "index.handler"
  runtime       = "nodejs22.x"
  role          = aws_iam_role.validate_exec.arn
  timeout       = 30
  memory_size   = 512

  dead_letter_config {
    target_arn = aws_sqs_queue.validation_dlq.arn
  }

  tracing_config {
    mode = "Active"
  }

  environment {
    variables = {
      MAX_BYTES     = "52428800"
      ALLOWED_TYPES = "image/jpeg,image/png,image/webp"
    }
  }

  reserved_concurrent_executions = 200
  tags = local.common_tags
}

# ──────────────────────────────────────────────
# Step Functions — Express Workflow
# ──────────────────────────────────────────────
resource "aws_sfn_state_machine" "pipeline" {
  name       = "${local.prefix}-state-machine"
  role_arn   = aws_iam_role.sfn_exec.arn
  type       = "EXPRESS"
  definition = templatefile("${path.module}/pipeline.asl.json", {
    validate_arn   = aws_lambda_function.validate.arn
    thumbnail_arn  = aws_lambda_function.thumbnail.arn
    watermark_arn  = aws_lambda_function.watermark.arn
    meta_arn       = aws_lambda_function.metadata.arn
    dynamo_table   = aws_dynamodb_table.assets.name
    sns_topic_arn  = aws_sns_topic.asset_events.arn
  })

  logging_configuration {
    level                  = "ALL"
    include_execution_data = true
    log_destination        = "${aws_cloudwatch_log_group.sfn.arn}:*"
  }

  tracing_configuration {
    enabled = true
  }

  tags = local.common_tags
}

لماذا Express Workflow هنا؟ يكتمل كل ملف في أقل من 30 ثانية. بمعدل 100 ألف رفع يوميًا، يكلف Standard Workflow حوالي 2,500 دولار شهريًا (انتقالات الحالة بسعر 0.025 دولار لكل 1000)؛ بينما يكلف Express حوالي 8 دولارات شهريًا. بالنسبة لخطوط الأنابيب التي تكتمل بسرعة وتتحمل دلالات at-least-once، فإن Express هو دائمًا الخيار الصحيح. يُحتفظ بـ Standard للسير العمل متعددة الأيام أو معاملات الأعمال exactly-once.

دليل تشغيل DLQ

DLQ ليست مقبرة — إنها قائمة انتظار احتجاز للرسائل التي تحتاج إلى معالجة بشرية أو آلية. على نطاق المؤسسات الكبرى تحتاج إلى ثلاثة أشياء: تنبيه آلي (تم بالفعل أعلاه)، وآلية إعادة تشغيل، ومسار تفصيلي. يغطي مقتطف AWS CLI أدناه عمليات DLQ الروتينية:

# ── فحص الرسائل دون حذفها ──
aws sqs receive-message \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/img-pipeline-prod-transform-dlq \
  --max-number-of-messages 10 \
  --visibility-timeout 30 \
  --attribute-names All \
  | jq '.Messages[] | {id:.MessageId, body:(.Body|fromjson), received:.Attributes.ApproximateReceiveCount}'

# ── إعادة تشغيل جميع رسائل DLQ إلى قائمة الانتظار المصدر ──
aws sqs start-message-move-task \
  --source-arn arn:aws:sqs:us-east-1:123456789012:img-pipeline-prod-transform-dlq \
  --destination-arn arn:aws:sqs:us-east-1:123456789012:img-pipeline-prod-transform-queue \
  --max-number-of-messages-per-second 50

# مراقبة مهمة الانتقال
aws sqs list-message-move-tasks \
  --source-arn arn:aws:sqs:us-east-1:123456789012:img-pipeline-prod-transform-dlq

# ── تنظيف DLQ بعد تأكيد نشر الإصلاح ──
aws sqs purge-queue \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/img-pipeline-prod-transform-dlq

# ── إيجاد تتبع X-Ray لتنفيذ فاشل ──
aws stepfunctions get-execution-history \
  --execution-arn arn:aws:states:us-east-1:123456789012:execution:img-pipeline-prod-state-machine:abc123 \
  --query 'events[?type==`ExecutionFailed`].executionFailedEventDetails'

الرصد الشامل من البداية إلى النهاية

التتبع الموزع أمر لا يمكن التنازل عنه في خط الأنابيب غير المتزامن. يُنشر X-Ray عبر _X_AMZN_TRACE_ID في متغيرات بيئة Lambda إلى Step Functions وصولًا إلى استدعاءات SDK. كل PutItem في DynamoDB وكل Publish في SNS يظهران كشريحة تتبع. مجموعة المقاييس التشغيلية الرئيسية لهذا خط الأنابيب:

زمن الاستجابة P99 لخط الأنابيب — aws cloudwatch get-metric-statistics على ExecutionTime في نطاق AWS/States؛ تنبيه إذا تجاوز P99 45 ثانية.
عمق DLQ — ApproximateNumberOfMessagesVisible على كلا DLQs؛ تنبيه على أي قيمة غير صفرية.
معدل أخطاء Lambda — Errors / Invocations لكل دالة؛ تنبيه إذا تجاوز 1% على مدى 5 دقائق.
التزامن في التحويل — ConcurrentExecutions لدوال Lambda الخاصة بالصور المصغرة والعلامة المائية؛ تنبيه إذا اقترب من حد الحساب (1,000 افتراضيًا) لطلب زيادة الحصة بشكل استباقي.
نسبة البدء البارد — مقياس EMF مخصص تُصدره كل Lambda: { "cold_start": 1 } عند التهيئة الجديدة؛ يُتتبع كنسبة مئوية من جميع الاستدعاءات.

السجلات المنظمة عبر كل Lambda: أصدر سطر سجل JSON بالصيغة {"level":"INFO","traceId":"...","executionName":"...","fileKey":"...","durationMs":120} في نهاية كل معالج. استعلام CloudWatch Insights واحد يمتد بعد ذلك عبر خط الأنابيب بالكامل: fields @timestamp, traceId, fileKey, durationMs | filter executionName like /img-pipeline/ | sort @timestamp desc.

أنماط الفشل في الإنتاج والتخفيف منها

يبني المهندسون المتقدمون الأنظمة بتصنيف الفشل في الاعتبار قبل شحن السطر الأول من الكود. بالنسبة لهذا خط الأنابيب:

تكرار أحداث S3 — يُسلّم S3 بضمان at-least-once. تستخدم عملية PutItem في DynamoDB شرط التعبير attribute_not_exists(fileKey)؛ تُخفق التنفيذات المكررة في خطوة الحفظ مع ConditionalCheckFailedException، تلتقطها آلة الحالة وتوجهها إلى مسار Success (بتصميم idempotent).
تسلسل البدء البارد في Lambda — دفعة مفاجئة (مثلًا رفع 5,000 ملف) ستستنفد حد التزامن. خفّف من ذلك بتقييد maximumConcurrency في EventBridge Pipes وبالتزامن المُخصَّص مسبقًا على دالة validate (إنها عنق الزجاجة — تمر جميع الملفات عبرها).
تراكم DLQ دون إعادة تشغيل — إذا شُحن خلل ووصلت 50 ألف رسالة إلى DLQ قبل اطلاق التنبيه، فإن إعادة التشغيل الساذجة بالسرعة الكاملة ستُشغّل اندفاعة تزامن Lambda. أعد التشغيل دائمًا بـ --max-number-of-messages-per-second مضبوطًا على معدل يمكن لمصبّك تحمّله دون خنق.
فقدان سجل Express Workflow — لا تحتفظ Express Workflows بسجل التنفيذ في وحدة التحكم لأكثر من 90 يومًا. إن logging_configuration { level = "ALL" } في Terraform يحفظ كل انتقال حالة في CloudWatch Logs بشكل دائم. لا تنشر Express Workflows أبدًا دون هذا الإعداد.

فخ إعادة المحاولة الصامتة: تمتلك Lambda سياسة إعادة محاولة مدمجة غير متزامنة (محاولتان مع تراجع أسّي). تمتلك Step Functions أيضًا كتلة Retry الخاصة بها. إذا كان كلاهما نشطَين على نفس الدالة، يمكن لحالة الفشل العابرة أن تُشغّل ما يصل إلى 6 استدعاءات قبل الوصول إلى DLQ — وتُظهر مقاييسك 6 أضعاف عدد الاستدعاءات المتوقعة. بالنسبة لدوال Lambda المُنسَّقة عبر Step Functions، اضبط EventInvokeConfig الخاصة بـ Lambda على صفر محاولات (MaximumRetryAttempts: 0) واجعل آلة الحالة تمتلك جميع منطق إعادة المحاولة.

تجميع كل شيء معًا

انشر خط الأنابيب من البداية إلى النهاية بأمر terraform apply واحد، ثم اختبره ببث رفع حقيقي:

# 1. تطبيق البنية التحتية
cd infra/
terraform init -backend-config="bucket=my-tf-state" -backend-config="key=img-pipeline/prod.tfstate"
terraform plan -var="env=prod" -var="pagerduty_sns_arn=arn:aws:sns:..."
terraform apply -auto-approve -var="env=prod" -var="pagerduty_sns_arn=arn:aws:sns:..."

# 2. رفع صورة اختبارية ومراقبة تنفيذ خط الأنابيب
aws s3 cp ./test-image.jpg s3://img-pipeline-prod-ingest/uploads/test-image.jpg

# 3. الاستعلام عن آلة الحالة للتنفيذ
aws stepfunctions list-executions \
  --state-machine-arn arn:aws:states:us-east-1:123456789012:stateMachine:img-pipeline-prod-state-machine \
  --status-filter RUNNING \
  --query 'executions[0].executionArn' --output text \
  | xargs aws stepfunctions describe-execution --execution-arn

# 4. تأكيد كتابة سجل DynamoDB
aws dynamodb get-item \
  --table-name img-pipeline-prod-assets \
  --key '{"fileKey": {"S": "uploads/test-image.jpg"}}' \
  | jq '.Item'

# 5. تأكيد عدم وجود رسائل DLQ
aws sqs get-queue-attributes \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/img-pipeline-prod-transform-dlq \
  --attribute-names ApproximateNumberOfMessagesVisible \
  | jq '.Attributes.ApproximateNumberOfMessagesVisible'
# المتوقع: "0"

نمط خط الأنابيب هذا — حدث S3، وتصفية EventBridge Pipes، وتنسيق Express Workflow، وتحويلات Lambda المتوازية، وتكاملات SDK المباشرة لكتابة DynamoDB، وتوزيع SNS المتشعب للاقتران المصبّي، وDLQs عند كل حدود الفشل، وتتبعات X-Ray تربط كل النقاط، وTerraform يدير كل ذلك — هو الهيكل الجاهز للإنتاج المستخدم في خطوط أنابيب الصور والفيديو والمستندات المؤسسية في جميع أنحاء الصناعة. قم بتكييف دوال Lambda الخاصة بالتحويل لمجالك؛ يبقى الهيكل التشغيلي متطابقًا.