Реалізація Serverless Scheduled Tasks (CloudWatch Events / Cron)

Реалізація Serverless Scheduled Tasks (CloudWatch Events / Cron)

Scheduled Tasks у serverless замінює традиційні cron jobs без постійно запущеного сервера. EventBridge Scheduler (AWS) або Cloud Scheduler (GCP) запускають Lambda за розписанням. Переваги: немає обслуговування сервера, вбудований retry, централізоване управління розписанням.

AWS EventBridge Scheduler vs CloudWatch Events

CloudWatch Events (старий): працює, але обмежений. Немає управління concurrency, немає retry за замовчуванням.

EventBridge Scheduler (2022, рекомендований): для нових проектів:

• Flexible time window (запустити у будь-яку мить протягом N хвилин)
• Retry policy з backoff
• DLQ для невдалих invocations
• Підтримка часового поясу для розписання
• Одноразові розписання (at-invokes)

resource "aws_scheduler_schedule" "daily_report" {
  name = "daily-sales-report"

  flexible_time_window {
    mode                      = "FLEXIBLE"
    maximum_window_in_minutes = 15
  }

  schedule_expression          = "cron(0 8 * * ? *)"
  schedule_expression_timezone = "Europe/Moscow"

  target {
    arn      = aws_lambda_function.generate_report.arn
    role_arn = aws_iam_role.scheduler_role.arn

    input = jsonencode({
      report_type = "daily_sales"
      format      = "pdf"
    })

    retry_policy {
      maximum_event_age_in_seconds = 3600
      maximum_retry_attempts       = 3
    }

    dead_letter_config {
      arn = aws_sqs_queue.scheduler_dlq.arn
    }
  }
}

AWS Cron синтаксис

AWS використовує 6-полевий cron (рік додано):

cron(хвилини години день-місяця місяць день-тижня рік)

cron(0 12 * * ? *)        # Щодня о 12:00 UTC
cron(0 */6 * * ? *)       # Кожні 6 годин
cron(0 9 ? * MON-FRI *)   # Кожен робочий день о 9:00
cron(0 0 1 * ? *)         # Перший день місяця о 00:00
cron(0 8 ? * MON *)       # Кожен понеділок о 8:00

rate(5 minutes)            # Кожні 5 хвилин
rate(1 hour)               # Кожну годину

Lambda обробник для scheduled task

import json
import logging
from datetime import datetime

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

def handler(event, context):
    task_type = event.get('task_type', 'unknown')
    started_at = datetime.utcnow().isoformat()

    logger.info(f"Starting scheduled task: {task_type} at {started_at}")

    try:
        if task_type == 'cleanup_expired_sessions':
            result = cleanup_expired_sessions()
        elif task_type == 'send_weekly_digest':
            result = send_weekly_digest()
        elif task_type == 'sync_inventory':
            result = sync_inventory_from_erp()
        else:
            raise ValueError(f"Unknown task type: {task_type}")

        logger.info(f"Task completed: {json.dumps(result)}")
        return {'status': 'success', 'task': task_type, **result}

    except Exception as e:
        logger.error(f"Task failed: {e}", exc_info=True)
        raise

Ідемпотентність для scheduled tasks

Якщо задача запустилась двічі (flexible window або retry) — результат мусить бути однаковим:

import boto3
import hashlib
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
task_locks = dynamodb.Table('task_locks')

def run_with_idempotency(task_fn, task_id: str, time_window_minutes: int = 60):
    """Гарантує, що задача виконається лише один раз за вікно"""
    window_key = f"{task_id}:{datetime.utcnow().strftime('%Y%m%d%H')}"

    try:
        task_locks.put_item(
            Item={
                'task_key': window_key,
                'ttl': int(time.time()) + time_window_minutes * 60
            },
            ConditionExpression='attribute_not_exists(task_key)'
        )
    except task_locks.meta.client.exceptions.ConditionalCheckFailedException:
        print(f"Task {task_id} already ran in this window, skipping")
        return None

    return task_fn()

Моніторинг scheduled tasks

Heartbeat паттерн — кожна успішна задача пінгує моніторинг. Якщо ping не прийшов — задача не запустилась:

import requests

def send_healthcheck_ping(check_id: str):
    """Послати ping у Healthchecks.io або UptimeRobot"""
    requests.get(
        f"https://hc-ping.com/{check_id}",
        timeout=5
    )

def handler(event, context):
    result = perform_task()

    # Уведомити моніторинг про успіх
    send_healthcheck_ping(os.environ['HEALTHCHECK_ID'])

    return result

Healthchecks.io — спеціалізований сервіс для моніторингу cron. Якщо ping не прийшов в очікуваний час — уведомлення.

Кілька окремень

Scheduled tasks часто не потрібні в staging/dev:

resource "aws_scheduler_schedule" "daily_report" {
  count = var.environment == "production" ? 1 : 0

  name = "daily-sales-report"
  ...
}

Або використовувати state = "DISABLED" для non-production.

Типові scheduled tasks

• Очистка застарілих сесій / тимчасових файлів
• Генерація звітів та дайджестів
• Синхронізація даних з зовнішніми API (раз на годину)
• Перевірка та оновлення SSL-сертифікатів
• Резервне копіювання даних
• Оновлення кешів (сайтмап, категорії, лічильники)
• Обробка відкладених задач (відправити заплановані email)

Графік впровадження

• EventBridge Scheduler + базова Lambda — 1-2 дні
• Ідемпотентність + retry policy — 1 день
• Моніторинг (Healthchecks.io + алерти) — 0.5-1 день
• Кілька задач + тестування — 1-2 дні