Своя cloud-функция integrity-проверки

Большинство рецептов раздела «Соло-разработка с ИИ» сводятся к одному паттерну:

«Раз в N минут (или по моему запросу) запусти небольшой код в облаке, который что-то проверит, и если результат не такой, как ожидается — пришли мне push».

В Yandex Cloud это решается одной Cloud Function с timer-trigger. Здесь — универсальный скелет, который можно скопировать и адаптировать под:

• синтетического пользователя (логин в приложение, поиск, заказ);
• проверку RAG end-to-end (запрос → embeddings → Qdrant → LLM → ответ);
• верификацию идемпотентности webhook-обработчика;
• сверку счётчиков в YDB и Postgres между сервисами;
• проверку, что cron-job отработал и положил файл в S3.

Скелет: один файл index.py

import os, time, json, requests, traceback

NOTIFLY_URL   = os.environ["NOTIFLY_URL"]
NOTIFLY_TOKEN = os.environ["NOTIFLY_TOKEN"]
APP_BASE      = os.environ.get("APP_BASE", "https://app.example.com")

def notify(title, msg, prio=8):
    requests.post(f"{NOTIFLY_URL}/message",
                  params={"token": NOTIFLY_TOKEN},
                  json={"title": title, "message": msg, "priority": prio},
                  timeout=5)

# --- набор проверок ----------------------------------------------------

def check_login():
    r = requests.post(f"{APP_BASE}/api/login",
                      json={"user": "synthetic@example.com", "pass": os.environ["CANARY_PASS"]},
                      timeout=10)
    assert r.status_code == 200, f"login: {r.status_code}"
    assert r.json().get("token"), "no token in response"

def check_rag_query():
    t0 = time.time()
    r = requests.post(f"{APP_BASE}/api/ask",
                      json={"q": "What is our refund policy?"},
                      timeout=20)
    assert r.status_code == 200, f"ask: {r.status_code}"
    body = r.json()
    assert "refund" in body.get("answer", "").lower(), "answer drifted"
    assert (time.time() - t0) < 8, "rag too slow"

def check_db_counters():
    # пример: число заказов за вчера должно совпадать с числом записей в реплике
    a = int(requests.get(f"{APP_BASE}/internal/orders/yesterday/count",
                         headers={"X-Probe": os.environ["PROBE_KEY"]}, timeout=5).text)
    b = int(requests.get(f"{APP_BASE}/internal/replica/orders/yesterday/count",
                         headers={"X-Probe": os.environ["PROBE_KEY"]}, timeout=5).text)
    assert a == b, f"counter drift: app={a} replica={b}"

CHECKS = {
    "login":         check_login,
    "rag-query":     check_rag_query,
    "db-counters":   check_db_counters,
}

# --- entrypoint --------------------------------------------------------

def handler(event, context):
    """
    YC Function entrypoint.
    Вызывается по timer-trigger или вручную (yc fn invoke).
    Можно ограничить набор проверок: payload {"checks": ["login"]}
    """
    payload = event if isinstance(event, dict) else {}
    if isinstance(event, dict) and "body" in event:        # http-вызов
        try:    payload = json.loads(event["body"] or "{}")
        except Exception: payload = {}

    selected = payload.get("checks") or list(CHECKS)
    failed   = []

    for name in selected:
        fn = CHECKS.get(name)
        if not fn:
            continue
        try:
            fn()
        except Exception as e:
            failed.append((name, e, traceback.format_exc()))

    if failed:
        body = "\n\n".join(f"❌ {n}: {type(e).__name__}: {e}" for n, e, _ in failed)
        notify(f"🩺 integrity: {len(failed)} fail", body, priority=9)

    return {"statusCode": 200, "body": json.dumps({
        "ok":      len(failed) == 0,
        "failed":  [n for n, _, _ in failed],
        "checked": selected,
    })}

Деплой

cd integrity-check
zip -qr ../fn.zip .

yc serverless function version create \
  --function-name notifly-integrity \
  --runtime python311 \
  --entrypoint index.handler \
  --memory 256MB --execution-timeout 30s \
  --environment NOTIFLY_URL=...,NOTIFLY_TOKEN=...,APP_BASE=...,CANARY_PASS=...,PROBE_KEY=... \
  --source-path ../fn.zip

Timer-trigger:

yc serverless trigger create timer \
  --name notifly-integrity-timer \
  --cron-expression '* * * * ? *' \
  --invoke-function-name notifly-integrity \
  --invoke-function-service-account-id <sa-id>

После этого функция дергается раз в минуту, и при поломке — push в Notifly.

Когда вызывать вручную из приложения

Тот же endpoint можно дёрнуть из вашего сервиса в подозрительные моменты: после деплоя, после миграции, после массового импорта данных. Запрос запустит проверки немедленно, не дожидаясь следующего timer-tick:

requests.post(
    "https://functions.yandexcloud.net/d4e.../",
    json={"checks": ["login", "rag-query"]},
    headers={"Authorization": f"Bearer {iam_token}"},
)

Чем это лучше «просто curl /health»

• Реальные проверки бизнес-логики, а не «процесс жив».
• Один центральный файл, в котором лежат все ваши «знания» о том, что должно работать. Удобно для соло-разработчика — это и тесты, и мониторинг.
• Запускается из облака — не зависит от вашего ноута, VPN и Wi-Fi в кафе.
• Дешёво — при * * * * ? * это ≤ 43 200 запусков в месяц по 1–2 секунды, free-tier Cloud Functions на это рассчитан.

Что положить в текст алёрта

• имя проваленной проверки;
• exception class + первые строки traceback;
• timestamp начала падения (для дедупликации, если падает 60 раз подряд);
• ссылка на dashboard функции (https://console.yandex.cloud/.../functions/...).