b8f74bd027
- Changed terminology in `config.yaml` from "SLA deadline" to "readiness target" to clarify that jobs still run after the specified time. - Updated `README.md` to reflect the new readiness target concept, emphasizing that jobs started after the target are still executed and reported. - Modified `pipeline` logic to incorporate the new readiness target, ensuring accurate tracking of jobs that begin after the target time. - Enhanced reporting in the run report to include counts of jobs that started after the readiness target, improving transparency in job execution outcomes.
248 lines
9.5 KiB
Python
248 lines
9.5 KiB
Python
"""Batch orchestrator: discover jobs, run them across a bounded thread pool,
|
|
and write one run report per day.
|
|
|
|
One `ThreadPoolExecutor` runs every discovered (field, disease) job so a slow
|
|
crop cannot serialise the others; `pipeline.job.run_job` isolates each job's
|
|
failures from the rest. This module owns everything above that: worklist
|
|
discovery + CLI filters, the `--skip-existing-advice` resume check, the
|
|
09:00 readiness target, and writing the run report.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import date, datetime
|
|
from datetime import time as dtime
|
|
|
|
from pipeline.config import Settings
|
|
from pipeline.db import fetch_all, qualify_app_sql
|
|
from pipeline.job import JobResult, run_job
|
|
from pipeline.output import write_job_output, write_run_report
|
|
from pipeline.prompts import PromptRegistry
|
|
from pipeline.resources import Resources
|
|
from pipeline.vocab import Vocabulary
|
|
from pipeline.worklist import Job, build_jobs
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM {advisories} WHERE issuance_date = ?"
|
|
|
|
|
|
def _parse_deadline(deadline_str: str) -> datetime:
|
|
"""
|
|
Build today's readiness-target datetime from an "HH:MM" string.
|
|
|
|
Deliberately anchored to the wall-clock date the batch is actually
|
|
running on, not `as_of`: the target ("advice ready by 09:00") is about
|
|
when the run happens, and in production `as_of` is always today anyway.
|
|
Anchoring to `as_of` instead would make every historical/backfill run
|
|
(`--as-of` in the past) look like its target had already passed.
|
|
Jobs that start after the target still run.
|
|
"""
|
|
try:
|
|
hour_str, minute_str = deadline_str.split(":", 1)
|
|
return datetime.combine(date.today(), dtime(hour=int(hour_str), minute=int(minute_str)))
|
|
except (ValueError, TypeError) as exc:
|
|
raise ValueError(f"Invalid deadline '{deadline_str}'. Expected HH:MM.") from exc
|
|
|
|
|
|
def _deadline_notice(
|
|
now: datetime,
|
|
deadline: datetime,
|
|
batch_start: datetime,
|
|
) -> str | None:
|
|
"""Classify `now` against the readiness target. Never means "skip".
|
|
|
|
"late" — the target has passed; the job should still run.
|
|
"budget_80" — at least 80% of the time before the target has been used.
|
|
None — neither. A batch that has already started past the target is
|
|
"late"; the 80% branch does not run in that case.
|
|
"""
|
|
if now >= deadline:
|
|
return "late"
|
|
total_budget_s = (deadline - batch_start).total_seconds()
|
|
if total_budget_s <= 0:
|
|
return None
|
|
elapsed_s = (now - batch_start).total_seconds()
|
|
if elapsed_s >= 0.8 * total_budget_s:
|
|
return "budget_80"
|
|
return None
|
|
|
|
|
|
def filter_jobs(
|
|
jobs: list[Job],
|
|
*,
|
|
crop: str | None = None,
|
|
disease: str | None = None,
|
|
field_ids: list[int] | None = None,
|
|
) -> list[Job]:
|
|
"""Narrow the discovered worklist to CLI-selected crop / disease / field IDs."""
|
|
filtered = jobs
|
|
if crop:
|
|
filtered = [j for j in filtered if j.crop_english.lower() == crop.strip().lower()]
|
|
if disease:
|
|
filtered = [j for j in filtered if j.disease_english.lower() == disease.strip().lower()]
|
|
if field_ids:
|
|
wanted = set(field_ids)
|
|
filtered = [j for j in filtered if j.field_id in wanted]
|
|
return filtered
|
|
|
|
|
|
def _existing_advice_keys(resources: Resources, as_of: date) -> set[tuple[int, str]]:
|
|
"""One bulk SELECT instead of one per job, for --skip-existing-advice."""
|
|
conn = resources.sql_connection()
|
|
rows = fetch_all(
|
|
conn,
|
|
qualify_app_sql(resources.settings.sql, _EXISTING_ADVICE_SQL),
|
|
(as_of,),
|
|
)
|
|
return {(int(row.field_id), str(row.disease)) for row in rows}
|
|
|
|
|
|
def run_batch(
|
|
settings: Settings,
|
|
crop_vocab: Vocabulary,
|
|
disease_vocab: Vocabulary,
|
|
as_of: date,
|
|
*,
|
|
crop: str | None = None,
|
|
disease: str | None = None,
|
|
field_ids: list[int] | None = None,
|
|
deadline_override: str | None = None,
|
|
skip_existing_advice: bool = False,
|
|
dry_run: bool = False,
|
|
workers_override: int | None = None,
|
|
) -> int:
|
|
"""Run every discovered job and write the run report. Returns the process exit code."""
|
|
prompts = PromptRegistry(settings.prompts_dir)
|
|
resources = Resources(settings, prompts)
|
|
try:
|
|
jobs = build_jobs(
|
|
resources.sql_connection(),
|
|
settings.worklist,
|
|
crop_vocab,
|
|
disease_vocab,
|
|
)
|
|
jobs = filter_jobs(jobs, crop=crop, disease=disease, field_ids=field_ids)
|
|
logger.info("Worklist discovered %d job(s) for as_of=%s", len(jobs), as_of.isoformat())
|
|
|
|
# Unique pairs only: one warning per crop/disease, not one per field.
|
|
# An empty job list validates nothing.
|
|
prompt_pairs = list(dict.fromkeys((job.crop_english, job.disease_english) for job in jobs))
|
|
fallbacks = prompts.validate(prompt_pairs)
|
|
if fallbacks:
|
|
logger.warning(
|
|
"%d crop/disease pair(s) have no dedicated advice prompt and will fall back "
|
|
"to prompts/advice/_default/: %s",
|
|
len(fallbacks),
|
|
fallbacks,
|
|
)
|
|
|
|
if not jobs:
|
|
logger.warning(
|
|
"No jobs to run for as_of=%s; nothing matched pat_mod_AI_model = 1 "
|
|
"and anmod_AI_advice = 1 (or the CLI filters).",
|
|
as_of.isoformat(),
|
|
)
|
|
write_run_report(settings.output_dir, as_of, [])
|
|
return 0
|
|
|
|
resources.load_product_index()
|
|
|
|
existing: set[tuple[int, str]] = set()
|
|
if skip_existing_advice:
|
|
existing = _existing_advice_keys(resources, as_of)
|
|
logger.info(
|
|
"%d field/disease pair(s) already have advice for %s and will be skipped",
|
|
len(existing),
|
|
as_of.isoformat(),
|
|
)
|
|
|
|
deadline_str = deadline_override or settings.schedule.deadline
|
|
deadline = _parse_deadline(deadline_str)
|
|
batch_start = datetime.now()
|
|
warned_80pct = False
|
|
warned_late = False
|
|
late_starts = 0
|
|
warn_lock = threading.Lock()
|
|
|
|
def _run_one(job: Job) -> JobResult:
|
|
nonlocal warned_80pct, warned_late, late_starts
|
|
|
|
now = datetime.now()
|
|
notice = _deadline_notice(now, deadline, batch_start)
|
|
started_late = notice == "late"
|
|
if notice == "late":
|
|
with warn_lock:
|
|
first_late = not warned_late
|
|
warned_late = True
|
|
if first_late:
|
|
logger.warning(
|
|
"Target %s has passed; continuing remaining jobs so every "
|
|
"field still receives advice.",
|
|
deadline_str,
|
|
)
|
|
elif notice == "budget_80":
|
|
with warn_lock:
|
|
first_80, warned_80pct = not warned_80pct, True
|
|
if first_80:
|
|
logger.warning(
|
|
"80%% of the time before the %s target has been used; "
|
|
"the run may finish after the target, and remaining jobs "
|
|
"will still run.",
|
|
deadline_str,
|
|
)
|
|
|
|
if (job.field_id, job.disease_english) in existing:
|
|
return JobResult(job=job, status="skipped_existing")
|
|
|
|
if started_late:
|
|
with warn_lock:
|
|
late_starts += 1
|
|
return run_job(resources, job, as_of, dry_run=dry_run)
|
|
|
|
# Worker threads start with an empty OpenTelemetry context
|
|
# (`contextvars` do not cross `ThreadPoolExecutor.submit`). That is
|
|
# intentional: `job_span` opens inside the worker and becomes a root
|
|
# span, one independent trace per (field, crop, disease, as_of). Do
|
|
# not wrap this submit in a tracing context manager from this thread
|
|
# — it would not reach the workers. See pipeline/observability.py.
|
|
workers = workers_override or settings.concurrency.workers
|
|
results: list[JobResult] = []
|
|
with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:
|
|
futures = {pool.submit(_run_one, job): job for job in jobs}
|
|
for future in as_completed(futures):
|
|
job = futures[future]
|
|
try:
|
|
result = future.result()
|
|
except BaseException as exc: # noqa: BLE001 - defense in depth; run_job already isolates failures
|
|
logger.exception(
|
|
"Unexpected error scheduling job field=%s crop=%s disease=%s",
|
|
job.field_id,
|
|
job.crop_english,
|
|
job.disease_english,
|
|
)
|
|
result = JobResult(
|
|
job=job,
|
|
status="failed",
|
|
error_type=type(exc).__name__,
|
|
error_message=str(exc),
|
|
)
|
|
|
|
results.append(result)
|
|
if result.payload is not None:
|
|
write_job_output(settings.output_dir, as_of, result.payload)
|
|
|
|
report = write_run_report(
|
|
settings.output_dir,
|
|
as_of,
|
|
results,
|
|
jobs_started_after_deadline=late_starts,
|
|
)
|
|
logger.info("Batch complete: %s", {k: v for k, v in report["totals"].items() if v})
|
|
return 0 if report["totals"].get("failed", 0) == 0 else 1
|
|
finally:
|
|
resources.close()
|