Реализация 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 для неудачных вызовов
• Часовой пояс для расписания
• Одноразовые расписания (at-запуски)

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

  flexible_time_window {
    mode                      = "FLEXIBLE"
    maximum_window_in_minutes = 15  # Запустить в течение 15 минут от указанного времени
  }

  schedule_expression          = "cron(0 8 * * ? *)"  # Каждый день в 8:00 UTC
  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  # Повторять до 1 часа
      maximum_retry_attempts       = 3
    }

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

Cron синтаксис для AWS

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 expression)
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  # Позволяет retry policy сработать

Идемпотентность 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):
    """Гарантирует, что задача не выполнится дважды в окне time_window_minutes"""
    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 jobs. Если 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 дня