File indexing completed on 2026-07-21 08:46:27
0001 """Publish a bounded epicprod PanDA activity projection to Snapper."""
0002
0003 import json
0004 from dataclasses import dataclass
0005 from datetime import datetime
0006 from typing import Optional
0007
0008 from django.db import connections, transaction
0009 from django.utils import timezone
0010
0011 from snapper_ai.services import (
0012 ComponentNotFound,
0013 ComponentUpdate,
0014 publish_component,
0015 register_component,
0016 )
0017
0018 from .panda.constants import JOB_STATUS_CATEGORIES, PANDA_SCHEMA
0019 from .panda.queries import get_activity
0020
0021
0022 PUBLISHER_IDENTITY = "swf-monitor:panda-activity"
0023 ASSESSMENT_POLICY_VERSION = "swf-panda-activity-24h-v2"
0024 EVENT_RESOLVER = "swf-panda-activity-history"
0025 WINDOW_DAYS = 1
0026 WINDOW_HOURS = 24
0027 MAX_STATUSES = 32
0028 MAX_SITES = 32
0029 MAX_TASK_TYPES = 32
0030 MAX_SERIALIZED_BYTES = 64 * 1024
0031 TASK_TERMINAL_STATUSES = (
0032 "done",
0033 "finished",
0034 "failed",
0035 "broken",
0036 "aborted",
0037 "exhausted",
0038 "passed",
0039 )
0040
0041 PANDA_REGISTRATION = {
0042 "title": "Curated epicprod PanDA activity",
0043 "description": (
0044 "Five-minute observations of bounded one-day PanDA job and task "
0045 "activity, together with all current in-flight job and task states, "
0046 "target sites, and running cores."
0047 ),
0048 "visibility": "public",
0049 "owning_subsystem": "SWF PanDA production monitor",
0050 "assessment_policy": ASSESSMENT_POLICY_VERSION,
0051 "max_serialized_bytes": MAX_SERIALIZED_BYTES,
0052 "quantities": {
0053 "window_hours": {
0054 "path": "window_hours",
0055 "type": "integer",
0056 "required": True,
0057 "kind": "window",
0058 "description": "Trailing activity window in hours.",
0059 },
0060 "jobs_total_24h": {
0061 "path": "jobs.total_24h",
0062 "type": "integer",
0063 "required": True,
0064 "kind": "window_count",
0065 "description": "Jobs modified in the trailing 24-hour window.",
0066 },
0067 "jobs_by_status_24h": {
0068 "path": "jobs.by_status_24h",
0069 "type": "object",
0070 "required": True,
0071 "kind": "bounded_map",
0072 "max_items": MAX_STATUSES,
0073 "description": "Trailing 24-hour job counts by PanDA status.",
0074 },
0075 "in_flight_jobs_now": {
0076 "path": "jobs.in_flight_now.total",
0077 "type": "integer",
0078 "required": True,
0079 "kind": "gauge",
0080 "description": "Jobs currently in any in-flight PanDA state.",
0081 },
0082 "in_flight_by_status_now": {
0083 "path": "jobs.in_flight_now.by_status",
0084 "type": "object",
0085 "required": True,
0086 "kind": "bounded_map",
0087 "max_items": MAX_STATUSES,
0088 "description": "Current in-flight job counts by PanDA status.",
0089 },
0090 "running_jobs_now": {
0091 "path": "jobs.in_flight_now.running_jobs",
0092 "type": "integer",
0093 "required": True,
0094 "kind": "gauge",
0095 "description": "Jobs currently in PanDA running state.",
0096 },
0097 "running_cores_now": {
0098 "path": "jobs.in_flight_now.running_cores",
0099 "type": "integer",
0100 "required": True,
0101 "kind": "gauge",
0102 "description": "Cores currently allocated to running PanDA jobs.",
0103 },
0104 "sites": {
0105 "path": "jobs.sites",
0106 "type": "object",
0107 "required": True,
0108 "kind": "bounded_map",
0109 "max_items": MAX_SITES,
0110 "description": (
0111 "Top PanDA target sites with trailing job outcomes and current "
0112 "in-flight status counts and running jobs and cores."
0113 ),
0114 },
0115 "tasks_total_24h": {
0116 "path": "tasks.total_24h",
0117 "type": "integer",
0118 "required": True,
0119 "kind": "window_count",
0120 "description": "Tasks modified in the trailing 24-hour window.",
0121 },
0122 "tasks_by_status_24h": {
0123 "path": "tasks.by_status_24h",
0124 "type": "object",
0125 "required": True,
0126 "kind": "bounded_map",
0127 "max_items": MAX_STATUSES,
0128 "description": "Trailing 24-hour task counts by PanDA status.",
0129 },
0130 "tasks_by_type_24h": {
0131 "path": "tasks.by_type_24h",
0132 "type": "object",
0133 "required": True,
0134 "kind": "bounded_map",
0135 "max_items": MAX_TASK_TYPES,
0136 "description": "Trailing 24-hour task counts by processing type.",
0137 },
0138 "in_flight_tasks_now": {
0139 "path": "tasks.in_flight_now.total",
0140 "type": "integer",
0141 "required": True,
0142 "kind": "gauge",
0143 "description": "JEDI tasks currently in any nonterminal state.",
0144 },
0145 "in_flight_tasks_by_status_now": {
0146 "path": "tasks.in_flight_now.by_status",
0147 "type": "object",
0148 "required": True,
0149 "kind": "bounded_map",
0150 "max_items": MAX_STATUSES,
0151 "description": "Current nonterminal JEDI task counts by status.",
0152 },
0153 "task_sites": {
0154 "path": "tasks.sites",
0155 "type": "object",
0156 "required": True,
0157 "kind": "bounded_map",
0158 "max_items": MAX_SITES,
0159 "description": (
0160 "Current nonterminal JEDI task counts by target site and "
0161 "status."
0162 ),
0163 },
0164 },
0165 "event_sources": [
0166 {
0167 "name": "panda-task-job-activity",
0168 "resolver": EVENT_RESOLVER,
0169 "owner": "ePIC PanDA production",
0170 "event_kind": "panda-task-job-activity",
0171 "event_time_field": "modificationtime",
0172 "visibility": "public",
0173 }
0174 ],
0175 }
0176
0177
0178 @dataclass(frozen=True)
0179 class PandaPublication:
0180 registration_update: ComponentUpdate
0181 update: ComponentUpdate
0182 projection: dict
0183 observed_at: datetime
0184
0185
0186 def _in_flight_activity() -> list[dict]:
0187 statuses = JOB_STATUS_CATEGORIES["active"]
0188 placeholders = ", ".join(["%s"] * len(statuses))
0189 sql = f"""
0190 SELECT "jobstatus",
0191 COALESCE("computingsite", 'unknown'),
0192 COUNT(*),
0193 COALESCE(SUM(
0194 COALESCE("actualcorecount", "corecount", 1)
0195 ), 0)
0196 FROM "{PANDA_SCHEMA}"."jobsactive4"
0197 WHERE "jobstatus" IN ({placeholders})
0198 GROUP BY "jobstatus", COALESCE("computingsite", 'unknown')
0199 ORDER BY "jobstatus", 3 DESC
0200 """
0201 with connections["panda"].cursor() as cursor:
0202 cursor.execute(sql, statuses)
0203 return [
0204 {
0205 "status": str(status or "unknown"),
0206 "site": str(site or "unknown"),
0207 "jobs": int(jobs or 0),
0208 "cores": int(cores or 0),
0209 }
0210 for status, site, jobs, cores in cursor.fetchall()
0211 ]
0212
0213
0214 def _in_flight_task_activity() -> list[dict]:
0215 placeholders = ", ".join(["%s"] * len(TASK_TERMINAL_STATUSES))
0216 sql = f"""
0217 SELECT COALESCE("status", 'unknown'),
0218 COALESCE("site", 'unknown'),
0219 COUNT(*)
0220 FROM "{PANDA_SCHEMA}"."jedi_tasks"
0221 WHERE "status" IS NULL OR "status" NOT IN ({placeholders})
0222 GROUP BY COALESCE("status", 'unknown'), COALESCE("site", 'unknown')
0223 ORDER BY 1, 3 DESC
0224 """
0225 with connections["panda"].cursor() as cursor:
0226 cursor.execute(sql, TASK_TERMINAL_STATUSES)
0227 return [
0228 {
0229 "status": str(status or "unknown"),
0230 "site": str(site or "unknown"),
0231 "tasks": int(task_count or 0),
0232 }
0233 for status, site, task_count in cursor.fetchall()
0234 ]
0235
0236
0237 def _count_map(values, label, maximum):
0238 result = {
0239 str(key or "unknown"): int(value or 0)
0240 for key, value in (values or {}).items()
0241 }
0242 if len(result) > maximum:
0243 raise ValueError(f"{label} exceeds {maximum} entries")
0244 return result
0245
0246
0247 def panda_projection(
0248 activity: Optional[dict] = None,
0249 in_flight_activity: Optional[list[dict]] = None,
0250 in_flight_task_activity: Optional[list[dict]] = None,
0251 ) -> tuple[dict, datetime]:
0252 """Build the bounded revision-driving PanDA activity projection."""
0253 activity = activity if activity is not None else get_activity(days=WINDOW_DAYS)
0254 if not isinstance(activity, dict) or activity.get("error"):
0255 raise ValueError(
0256 f"PanDA activity query failed: "
0257 f"{activity.get('error') if isinstance(activity, dict) else activity!r}"
0258 )
0259 jobs = activity.get("jobs") or {}
0260 tasks = activity.get("tasks") or {}
0261 in_flight_rows = (
0262 in_flight_activity
0263 if in_flight_activity is not None
0264 else _in_flight_activity()
0265 )
0266 in_flight_task_rows = (
0267 in_flight_task_activity
0268 if in_flight_task_activity is not None
0269 else _in_flight_task_activity()
0270 )
0271
0272 jobs_by_status = _count_map(
0273 jobs.get("by_status"), "job statuses", MAX_STATUSES
0274 )
0275 tasks_by_status = _count_map(
0276 tasks.get("by_status"), "task statuses", MAX_STATUSES
0277 )
0278 task_types = {}
0279 for row in tasks.get("by_type") or []:
0280 label = str(row.get("type") or "unknown")
0281 task_types[label] = task_types.get(label, 0) + int(row.get("count") or 0)
0282 if len(task_types) > MAX_TASK_TYPES:
0283 task_types = dict(
0284 sorted(task_types.items(), key=lambda item: (-item[1], item[0]))[
0285 :MAX_TASK_TYPES
0286 ]
0287 )
0288
0289 recent_sites = {
0290 str(row.get("site") or "unknown"): row
0291 for row in jobs.get("by_site") or []
0292 }
0293 in_flight_by_status = {}
0294 current_sites = {}
0295 for row in in_flight_rows:
0296 status = str(row.get("status") or "unknown")
0297 site = str(row.get("site") or "unknown")
0298 jobs_now = int(row.get("jobs") or 0)
0299 cores_now = int(row.get("cores") or 0)
0300 in_flight_by_status[status] = (
0301 in_flight_by_status.get(status, 0) + jobs_now
0302 )
0303 site_state = current_sites.setdefault(
0304 site,
0305 {
0306 "in_flight_jobs": 0,
0307 "by_status": {},
0308 "running_jobs": 0,
0309 "running_cores": 0,
0310 },
0311 )
0312 site_state["in_flight_jobs"] += jobs_now
0313 site_state["by_status"][status] = (
0314 site_state["by_status"].get(status, 0) + jobs_now
0315 )
0316 if status == "running":
0317 site_state["running_jobs"] += jobs_now
0318 site_state["running_cores"] += cores_now
0319 if len(in_flight_by_status) > MAX_STATUSES:
0320 raise ValueError(
0321 f"in-flight job statuses exceed {MAX_STATUSES} entries"
0322 )
0323
0324 in_flight_tasks_by_status = {}
0325 current_task_sites = {}
0326 for row in in_flight_task_rows:
0327 status = str(row.get("status") or "unknown")
0328 site = str(row.get("site") or "unknown")
0329 tasks_now = int(row.get("tasks") or 0)
0330 in_flight_tasks_by_status[status] = (
0331 in_flight_tasks_by_status.get(status, 0) + tasks_now
0332 )
0333 site_state = current_task_sites.setdefault(
0334 site,
0335 {"in_flight_tasks": 0, "by_status": {}},
0336 )
0337 site_state["in_flight_tasks"] += tasks_now
0338 site_state["by_status"][status] = (
0339 site_state["by_status"].get(status, 0) + tasks_now
0340 )
0341 if len(in_flight_tasks_by_status) > MAX_STATUSES:
0342 raise ValueError(
0343 f"in-flight task statuses exceed {MAX_STATUSES} entries"
0344 )
0345
0346 ranked_task_sites = sorted(
0347 current_task_sites,
0348 key=lambda name: (
0349 -int(current_task_sites[name]["in_flight_tasks"]),
0350 name,
0351 ),
0352 )[:MAX_SITES]
0353 task_sites = {
0354 name: {
0355 "in_flight_tasks_now": int(
0356 current_task_sites[name]["in_flight_tasks"]
0357 ),
0358 "by_status_now": current_task_sites[name]["by_status"],
0359 }
0360 for name in ranked_task_sites
0361 }
0362
0363 site_names = set(recent_sites) | set(current_sites)
0364 ranked_sites = sorted(
0365 site_names,
0366 key=lambda name: (
0367 -int((current_sites.get(name) or {}).get("in_flight_jobs") or 0),
0368 -int((recent_sites.get(name) or {}).get("total") or 0),
0369 name,
0370 ),
0371 )[:MAX_SITES]
0372 sites = {}
0373 for name in ranked_sites:
0374 recent = recent_sites.get(name) or {}
0375 current = current_sites.get(name) or {}
0376 sites[name] = {
0377 "jobs_24h": int(recent.get("total") or 0),
0378 "finished_24h": int(recent.get("finished") or 0),
0379 "failed_24h": int(recent.get("failed") or 0),
0380 "in_flight_jobs_now": int(current.get("in_flight_jobs") or 0),
0381 "by_status_now": current.get("by_status") or {},
0382 "running_jobs_now": int(current.get("running_jobs") or 0),
0383 "running_cores_now": int(current.get("running_cores") or 0),
0384 }
0385
0386 in_flight_total = sum(in_flight_by_status.values())
0387 running_jobs = in_flight_by_status.get("running", 0)
0388 running_cores = sum(
0389 int(row.get("cores") or 0)
0390 for row in in_flight_rows
0391 if str(row.get("status") or "unknown") == "running"
0392 )
0393 projection = {
0394 "window_hours": WINDOW_HOURS,
0395 "jobs": {
0396 "total_24h": int(jobs.get("total") or 0),
0397 "by_status_24h": jobs_by_status,
0398 "in_flight_now": {
0399 "total": in_flight_total,
0400 "by_status": in_flight_by_status,
0401 "running_jobs": running_jobs,
0402 "running_cores": running_cores,
0403 },
0404 "sites": sites,
0405 },
0406 "tasks": {
0407 "total_24h": int(tasks.get("total") or 0),
0408 "by_status_24h": tasks_by_status,
0409 "by_type_24h": task_types,
0410 "in_flight_now": {
0411 "total": sum(in_flight_tasks_by_status.values()),
0412 "by_status": in_flight_tasks_by_status,
0413 },
0414 "sites": task_sites,
0415 },
0416 }
0417 return projection, timezone.now()
0418
0419
0420 def publish_panda_activity() -> PandaPublication:
0421 """Query, curate, and atomically publish epicprod PanDA activity."""
0422 projection, observed_at = panda_projection()
0423 with transaction.atomic():
0424 try:
0425 update = publish_component(
0426 scope="epicprod",
0427 name="panda",
0428 publisher_identity=PUBLISHER_IDENTITY,
0429 data=projection,
0430 assessed_at=observed_at,
0431 source_as_of=observed_at,
0432 assessment_policy_version=ASSESSMENT_POLICY_VERSION,
0433 )
0434 except ComponentNotFound:
0435 registration_update = register_component(
0436 scope="epicprod",
0437 name="panda",
0438 publisher_identity=PUBLISHER_IDENTITY,
0439 registration=PANDA_REGISTRATION,
0440 component_schema_version=3,
0441 )
0442 update = publish_component(
0443 scope="epicprod",
0444 name="panda",
0445 publisher_identity=PUBLISHER_IDENTITY,
0446 data=projection,
0447 assessed_at=observed_at,
0448 source_as_of=observed_at,
0449 assessment_policy_version=ASSESSMENT_POLICY_VERSION,
0450 )
0451 else:
0452
0453
0454 registration_update = register_component(
0455 scope="epicprod",
0456 name="panda",
0457 publisher_identity=PUBLISHER_IDENTITY,
0458 registration=PANDA_REGISTRATION,
0459 component_schema_version=3,
0460 )
0461 return PandaPublication(
0462 registration_update=registration_update,
0463 update=update,
0464 projection=projection,
0465 observed_at=observed_at,
0466 )
0467
0468
0469 def compact_panda_publication_report(publication: PandaPublication) -> str:
0470 return json.dumps(
0471 {
0472 "scope": publication.update.scope,
0473 "component": publication.update.name,
0474 "revision": max(
0475 publication.update.revision,
0476 publication.registration_update.revision,
0477 ),
0478 "content_changed": publication.update.content_changed,
0479 "registration_changed": (
0480 publication.registration_update.registration_changed
0481 ),
0482 "jobs_24h": publication.projection["jobs"]["total_24h"],
0483 "in_flight_jobs": publication.projection["jobs"]["in_flight_now"][
0484 "total"
0485 ],
0486 "running_jobs": publication.projection["jobs"]["in_flight_now"][
0487 "running_jobs"
0488 ],
0489 "running_cores": publication.projection["jobs"]["in_flight_now"][
0490 "running_cores"
0491 ],
0492 "tasks_24h": publication.projection["tasks"]["total_24h"],
0493 "in_flight_tasks": publication.projection["tasks"][
0494 "in_flight_now"
0495 ]["total"],
0496 "observed_at": publication.observed_at.isoformat(),
0497 },
0498 indent=2,
0499 sort_keys=True,
0500 )
