223 lines
10 KiB
Python
223 lines
10 KiB
Python
"""Standalone entrypoint для tradein-scraper контейнера (#1182).
|
||
|
||
Зачем: API-деплой (docker restart tradein-backend) не должен прерывать
|
||
бегущие scraper sweep'ы (avito/cian/rosreestr). Вынесен в отдельный
|
||
контейнер tradein-scraper с тем же backend-образом, но другой командой.
|
||
|
||
Запуск: python -m app.scheduler_main
|
||
|
||
Zombie-reap: reap_zombies() вызывается в начале каждого тика scheduler_loop()
|
||
(см. app/services/scheduler.py), дублировать здесь не нужно.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import os
|
||
import signal
|
||
import sys
|
||
from contextlib import suppress
|
||
|
||
from app.core.config import settings
|
||
from app.core.shutdown import request_shutdown, shutdown_requested, wait_for_shutdown
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Safety net: после запроса drain'а ждём добровольного выхода задачи не дольше
|
||
# этого порога, который ОБЯЗАН быть < docker stop_grace_period (120s). Если
|
||
# некооперативная задача не вышла за это время — hard-cancel, чтобы процесс не
|
||
# подвис до SIGKILL за пределами grace. Module-level → monkeypatch'абельно в тестах.
|
||
_DRAIN_TIMEOUT_S = 100.0
|
||
|
||
# Мониторинг ошибок — GlitchTip (Sentry-совместимый, #396).
|
||
# Только integrations без Starlette/FastAPI (у нас нет ASGI-приложения здесь).
|
||
# SqlalchemyIntegration + HttpxIntegration + LoggingIntegration покрывают scraper-слой.
|
||
if settings.glitchtip_dsn:
|
||
import sentry_sdk
|
||
from sentry_sdk.integrations.httpx import HttpxIntegration
|
||
from sentry_sdk.integrations.logging import LoggingIntegration
|
||
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
|
||
|
||
from app.observability.sentry_scrub import scrub_pii_event
|
||
|
||
sentry_sdk.init(
|
||
dsn=settings.glitchtip_dsn,
|
||
environment=settings.environment,
|
||
release=os.getenv("GIT_SHA") or os.getenv("SENTRY_RELEASE") or "unknown",
|
||
traces_sample_rate=0.0,
|
||
send_default_pii=False,
|
||
before_send=scrub_pii_event,
|
||
integrations=[
|
||
SqlalchemyIntegration(),
|
||
HttpxIntegration(),
|
||
LoggingIntegration(level=logging.INFO, event_level=logging.ERROR),
|
||
],
|
||
)
|
||
logger.info("GlitchTip monitoring enabled (scheduler_main)")
|
||
|
||
|
||
def _should_run() -> bool:
|
||
"""Проверить kill-switch SCHEDULER_ENABLE перед запуском asyncio loop."""
|
||
return settings.scheduler_enable
|
||
|
||
|
||
async def _await_scheduler(task: asyncio.Task[None]) -> None:
|
||
"""Дождаться завершения scheduler-задачи с кооперативным SIGTERM-drain'ом.
|
||
|
||
- Нет shutdown: scheduler_loop бесконечен → задача никогда не завершается, ждём её
|
||
как есть (процесс просто работает).
|
||
- shutdown запрошен, задача ещё бежит: bounded `wait_for(_DRAIN_TIMEOUT_S)` —
|
||
кооперативная задача докоммитит текущий unit на ближайшем checkpoint'е и выйдет
|
||
сама. Если превысила grace → hard-cancel + suppress CancelledError, чтобы
|
||
некооперативная задача не подвесила процесс за пределами docker stop_grace_period.
|
||
"""
|
||
# Гонка «задача завершилась сама» против «пришёл SIGTERM»: отсчёт safety-net'а
|
||
# должен стартовать от МОМЕНТА запроса drain'а, а не от старта процесса.
|
||
shutdown_waiter = asyncio.create_task(wait_for_shutdown())
|
||
try:
|
||
await asyncio.wait({task, shutdown_waiter}, return_when=asyncio.FIRST_COMPLETED)
|
||
finally:
|
||
shutdown_waiter.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await shutdown_waiter
|
||
|
||
if task.done():
|
||
# Задача вышла сама (drained на checkpoint'е; без сигнала сюда не попадаем).
|
||
# .result() пробрасывает реальное исключение, если scheduler_loop неожиданно
|
||
# упал (сохраняем прежнюю loud-crash семантику `await task`), а не глушит его.
|
||
task.result()
|
||
logger.info("scheduler_main: scheduler task exited cleanly")
|
||
return
|
||
|
||
logger.info(
|
||
"scheduler_main: SIGTERM-drain — waiting up to %.0fs for in-flight unit to commit",
|
||
_DRAIN_TIMEOUT_S,
|
||
)
|
||
try:
|
||
await asyncio.wait_for(task, timeout=_DRAIN_TIMEOUT_S)
|
||
logger.info("scheduler_main: scheduler drained and exited cleanly")
|
||
except TimeoutError:
|
||
logger.warning(
|
||
"scheduler_main: drain exceeded %.0fs grace — hard-cancelling scheduler task",
|
||
_DRAIN_TIMEOUT_S,
|
||
)
|
||
task.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await task
|
||
|
||
|
||
async def _run_kit_scheduler() -> None:
|
||
"""Kit-путь (#2192): собрать SchedulerContext + registry и крутить kit scheduler_loop.
|
||
|
||
Инжектируем продуктовые адаптеры (RealScraperConfig/Matcher/Enrichment/SessionFactory) +
|
||
боевой runs-модуль (app.services.scrape_runs) + текущий shutdown_requested callable, чтобы
|
||
kit-loop делил ту же concurrency/lifecycle-семантику с боевым scheduler'ом. Продуктовые
|
||
НЕ-sweep source'ы приходят из build_product_handlers; kit-native sweeps — из build_registry.
|
||
|
||
SIGTERM-drain: kit scheduler_loop сам следит за ctx.shutdown_requested() и дренит
|
||
in-flight задачи (ctx.drain_inflight) — та же кооперативная семантика, что у боевого.
|
||
"""
|
||
from scraper_kit.orchestration.scheduler import (
|
||
SchedulerContext,
|
||
build_registry,
|
||
)
|
||
from scraper_kit.orchestration.scheduler import (
|
||
scheduler_loop as kit_scheduler_loop,
|
||
)
|
||
|
||
from app.services import scrape_runs as runs_mod
|
||
from app.services.product_handlers import build_product_handlers
|
||
from app.services.scraper_adapters import (
|
||
RealEnrichmentJobs,
|
||
RealMatcherAdapter,
|
||
RealProxyProvider,
|
||
RealScraperConfig,
|
||
RealSessionFactory,
|
||
)
|
||
|
||
# Proxy-пул (#2163 curl / #2164 P4 browser) инжектируется ТОЛЬКО когда включён любой
|
||
# из флагов. Оба дефолтно False (ship-dark) → proxy_provider=None → curl/browser берут
|
||
# прокси из env, прод не меняется. Один RealProxyProvider обслуживает оба пути (curl
|
||
# через curl_proxy_url, browser через BrowserFetcher → POST /fetch{proxy}).
|
||
use_pool = settings.use_proxy_pool_curl or settings.use_proxy_pool_browser
|
||
proxy_provider = RealProxyProvider() if use_pool else None
|
||
|
||
ctx = SchedulerContext(
|
||
config=RealScraperConfig(),
|
||
matcher=RealMatcherAdapter(),
|
||
enrichment=RealEnrichmentJobs(),
|
||
session_factory=RealSessionFactory(),
|
||
shutdown_requested=shutdown_requested,
|
||
runs=runs_mod,
|
||
proxy_provider=proxy_provider,
|
||
)
|
||
registry = build_registry(product_handlers=build_product_handlers(ctx))
|
||
logger.info("scheduler_main: kit-scheduler registry built (%d source handlers)", len(registry))
|
||
await kit_scheduler_loop(ctx, registry)
|
||
|
||
|
||
async def _run() -> None:
|
||
"""Async entrypoint: scheduler_loop() с кооперативным SIGTERM/SIGINT-drain'ом.
|
||
|
||
SIGTERM/SIGINT → request_shutdown() (НЕ task.cancel()): бегущий scrape-unit
|
||
докоммитит текущую карточку и выйдет сам на ближайшем between-unit checkpoint'е
|
||
(scheduler_loop / avito_detail_backfill / import_rosreestr_dkp). Safety-net в
|
||
_await_scheduler гарантирует выход в пределах docker grace.
|
||
|
||
Развилка #2192 (ship-dark): USE_KIT_SCHEDULER=false (дефолт) → боевой
|
||
app.services.scheduler.scheduler_loop без изменений; =true → kit-путь.
|
||
"""
|
||
if settings.use_kit_scheduler:
|
||
logger.info("scheduler_main: USE_KIT_SCHEDULER=true — running kit-scheduler path")
|
||
task = asyncio.create_task(_run_kit_scheduler())
|
||
else:
|
||
from app.services.scheduler import scheduler_loop
|
||
|
||
task = asyncio.create_task(scheduler_loop())
|
||
|
||
loop = asyncio.get_running_loop()
|
||
|
||
def _on_signal(signum: int) -> None:
|
||
logger.info("scheduler_main: signal %d received — requesting cooperative drain", signum)
|
||
request_shutdown()
|
||
|
||
try:
|
||
loop.add_signal_handler(signal.SIGTERM, lambda: _on_signal(signal.SIGTERM))
|
||
loop.add_signal_handler(signal.SIGINT, lambda: _on_signal(signal.SIGINT))
|
||
except NotImplementedError:
|
||
# Windows dev: signal handlers через loop не поддерживаются
|
||
logger.warning("scheduler_main: loop.add_signal_handler not supported (Windows dev)")
|
||
|
||
await _await_scheduler(task)
|
||
|
||
if shutdown_requested():
|
||
logger.info("scheduler_main: scheduler drained cleanly (SIGTERM)")
|
||
else:
|
||
logger.info("scheduler_main: scheduler task exited")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if not _should_run():
|
||
logger.warning("scheduler_main: scheduler disabled via SCHEDULER_ENABLE — exiting")
|
||
sys.exit(0)
|
||
|
||
# Best-effort FDW bootstrap: чтобы ensure_fdw_user_mapping отработал при старте
|
||
# scraper-контейнера (postgres_fdw нужен для rosreestr_dkp_import).
|
||
try:
|
||
from app.core.db import SessionLocal
|
||
from app.core.fdw import ensure_fdw_user_mapping
|
||
|
||
with SessionLocal() as db:
|
||
ensure_fdw_user_mapping(db)
|
||
except Exception:
|
||
logger.exception(
|
||
"scheduler_main: FDW user mapping bootstrap failed — cadastral queries may fail"
|
||
)
|
||
|
||
asyncio.run(_run())
|