src/monitor_app/snapper_delivery.py

0001 """Publish the campaign delivered-data record to Snapper.
0002
0003 The delivery record of CAMPAIGN_DELIVERY.md (swf-epicprod docs): per
0004 producing/current campaign, one leaf per physics configuration keyed
0005 by its pc label — events available where events/file is configured,
0006 files and bytes placed always, and the expected-events denominator
0007 with its provenance tier (the recorded chain: the edition's curated
0008 target, else the largest PC-anchored request, else absent). Lenses are
0009 not baked in; they project from the leaves at series-extraction time.
0010 """
0011
0012 import json
0013 from dataclasses import dataclass
0014 from typing import Optional
0015
0016 from django.db import transaction
0017 from django.utils import timezone
0018
0019 from snapper_ai.services import (
0020     ComponentNotFound,
0021     ComponentUpdate,
0022     publish_component,
0023     register_component,
0024 )
0025
0026 PUBLISHER_IDENTITY = "swf-monitor:campaign-delivery"
0027 ASSESSMENT_POLICY_VERSION = "swf-campaign-delivery-v1"
0028 MAX_CAMPAIGNS = 8
0029 MAX_LEAVES = 1024
0030 MAX_SERIALIZED_BYTES = 192 * 1024
0031
0032 DELIVERY_REGISTRATION = {
0033     "title": "Campaign delivered data",
0034     "description": (
0035         "Per producing/current campaign, one leaf per physics "
0036         "configuration (keyed by pc label): events available where "
0037         "events/file is configured, files and bytes placed, and the "
0038         "expected-events denominator with its provenance tier "
0039         "(included / requested / derived). The delivered-data record "
0040         "of CAMPAIGN_DELIVERY.md; categorization lenses project from "
0041         "the leaves at read time."
0042     ),
0043     "visibility": "public",
0044     "owning_subsystem": "SWF production catalog",
0045     "assessment_policy": ASSESSMENT_POLICY_VERSION,
0046     "max_serialized_bytes": MAX_SERIALIZED_BYTES,
0047     "quantities": {
0048         "campaigns": {
0049             "path": "campaigns",
0050             "type": "object",
0051             "required": True,
0052             "kind": "bounded_map",
0053             "max_items": MAX_CAMPAIGNS,
0054             "description": (
0055                 "Campaign name to {totals, leaves}; leaves map pc "
0056                 "labels to {events, expected, tier, files, bytes, "
0057                 "complete}."
0058             ),
0059         },
0060     },
0061 }
0062
0063
0064 @dataclass
0065 class DeliveryPublication:
0066     registration_update: ComponentUpdate
0067     update: ComponentUpdate
0068     projection: dict
0069     observed_at: object
0070
0071
0072 def _campaign_leaves(campaign_name):
0073     """Leaves for one campaign, keyed by pc label."""
0074     from pcs.models import Dataset, ProdTask
0075     from pcs.services import pc_request_projection
0076
0077     heads = list(
0078         Dataset.objects.filter(campaign__name=campaign_name)
0079         .select_related("physics_config")
0080         .order_by("composed_name", "block_num", "pk")
0081         .distinct("composed_name"))
0082     projection = pc_request_projection(heads)
0083     # Placement from the task output records with the progress
0084     # accounting rule: unique output DID, most recently checked record
0085     # wins (duplicate DIDs across task rows are the documented PCS
0086     # multiplicity), each unique output attributed to the winning
0087     # record's physics configuration. Verified to reproduce the
0088     # progress rollup's figures exactly. Outputs without a DID cannot
0089     # be placement-checked and are skipped.
0090     task_rows = list(
0091         ProdTask.objects.filter(dataset__campaign__name=campaign_name)
0092         .select_related("dataset__physics_config", "prod_config"))
0093     tasks = {task.dataset.composed_name: task for task in task_rows}
0094     # Attribution goes through the HEAD row's configuration per
0095     # composed name: multi-row past-output identities can resolve to
0096     # different configurations across their physical rows (per-row
0097     # metadata feeds the evgen resolution), and leaves are keyed by the
0098     # head's resolution.
0099     head_pc = {head.composed_name: head.physics_config.label
0100                for head in heads if head.physics_config_id}
0101     unique = {}
0102     for task in task_rows:
0103         pc = (head_pc.get(task.dataset.composed_name)
0104               or (task.dataset.physics_config.label
0105                   if task.dataset.physics_config_id else None))
0106         if pc is None:
0107             continue
0108         for output in task.outputs:
0109             did = str(output.get("did") or "").strip()
0110             if not did:
0111                 continue
0112             checked = str(output.get("checked_at") or "")
0113             prior = unique.get(did)
0114             if prior is None or checked >= prior[0]:
0115                 unique[did] = (checked, pc, output)
0116     placed_by_pc = {}
0117     for _checked, pc, output in unique.values():
0118         slot = placed_by_pc.setdefault(
0119             pc, {"files": 0, "bytes": 0, "complete": True})
0120         slot["files"] += int(output.get("file_count") or 0)
0121         slot["bytes"] += int(output.get("bytes") or 0)
0122         if not output.get("complete", True):
0123             slot["complete"] = False
0124
0125     # One leaf per distinct physics configuration: a campaign can hold
0126     # several version editions of one PC, and per-edition iteration
0127     # would count the PC's placement once per edition.
0128     heads_by_pc = {}
0129     for head in heads:
0130         if head.physics_config_id:
0131             heads_by_pc.setdefault(
0132                 head.physics_config.label, []).append(head)
0133
0134     leaves = {}
0135     totals = {"configs": 0, "with_target": 0, "events": 0,
0136               "expected": 0, "files": 0, "bytes": 0}
0137     for pc, pc_heads in heads_by_pc.items():
0138         expected = tier = None
0139         for head in pc_heads:
0140             if head.expected_events is not None:
0141                 expected = head.expected_events
0142                 tier = head.expected_events_source
0143                 break
0144         if expected is None:
0145             anchored = [r.nevents
0146                         for head in pc_heads
0147                         for r in projection.get(head.composed_name, ())
0148                         if r.nevents]
0149             if anchored:
0150                 expected, tier = max(anchored), "requested"
0151         # One payload job generates events_per_job events into one
0152         # output file, so the configured events_per_job IS the events
0153         # per file. (events_per_file is not a key any config carries.)
0154         events_per_file = None
0155         for head in pc_heads:
0156             task = tasks.get(head.composed_name)
0157             if task is None:
0158                 continue
0159             config = task.get_effective_config()
0160             try:
0161                 events_per_file = int(
0162                     (config.get("data") or {}).get("events_per_job"))
0163                 break
0164             except (TypeError, ValueError):
0165                 events_per_file = None
0166         placed = placed_by_pc.get(pc) or {}
0167         files = int(placed.get("files") or 0)
0168         bytes_placed = int(placed.get("bytes") or 0)
0169         events = files * events_per_file if events_per_file else None
0170         leaves[pc] = {
0171             "events": events,
0172             "expected": expected,
0173             "tier": tier or "",
0174             "files": files,
0175             "bytes": bytes_placed,
0176             "complete": bool(placed.get("complete", True)),
0177         }
0178         totals["configs"] += 1
0179         if expected is not None:
0180             totals["with_target"] += 1
0181             totals["expected"] += expected
0182         if events:
0183             totals["events"] += events
0184         totals["files"] += files
0185         totals["bytes"] += bytes_placed
0186     if len(leaves) > MAX_LEAVES:
0187         raise ValueError(
0188             f"{campaign_name}: {len(leaves)} leaves exceed {MAX_LEAVES}")
0189     return {"totals": totals, "leaves": leaves}
0190
0191
0192 def delivery_projection():
0193     from swf_epicprod.analytics.rollup import resolve_target_campaigns
0194
0195     campaigns = resolve_target_campaigns()[:MAX_CAMPAIGNS]
0196     projection = {
0197         "campaigns": {name: _campaign_leaves(name) for name in campaigns},
0198     }
0199     serialized = len(json.dumps(projection, separators=(",", ":")))
0200     if serialized > MAX_SERIALIZED_BYTES:
0201         raise ValueError(
0202             f"delivery projection {serialized} bytes exceeds "
0203             f"{MAX_SERIALIZED_BYTES}")
0204     return projection, timezone.now()
0205
0206
0207 def publish_delivery() -> DeliveryPublication:
0208     """Assemble and atomically publish the campaign delivery record."""
0209     projection, observed_at = delivery_projection()
0210     with transaction.atomic():
0211         try:
0212             update = publish_component(
0213                 scope="epicprod",
0214                 name="delivery",
0215                 publisher_identity=PUBLISHER_IDENTITY,
0216                 data=projection,
0217                 assessed_at=observed_at,
0218                 source_as_of=observed_at,
0219                 assessment_policy_version=ASSESSMENT_POLICY_VERSION,
0220             )
0221             registration_update = register_component(
0222                 scope="epicprod",
0223                 name="delivery",
0224                 publisher_identity=PUBLISHER_IDENTITY,
0225                 registration=DELIVERY_REGISTRATION,
0226                 component_schema_version=1,
0227             )
0228         except ComponentNotFound:
0229             registration_update = register_component(
0230                 scope="epicprod",
0231                 name="delivery",
0232                 publisher_identity=PUBLISHER_IDENTITY,
0233                 registration=DELIVERY_REGISTRATION,
0234                 component_schema_version=1,
0235             )
0236             update = publish_component(
0237                 scope="epicprod",
0238                 name="delivery",
0239                 publisher_identity=PUBLISHER_IDENTITY,
0240                 data=projection,
0241                 assessed_at=observed_at,
0242                 source_as_of=observed_at,
0243                 assessment_policy_version=ASSESSMENT_POLICY_VERSION,
0244             )
0245     return DeliveryPublication(
0246         registration_update=registration_update,
0247         update=update,
0248         projection=projection,
0249         observed_at=observed_at,
0250     )
0251
0252
0253 def compact_delivery_publication_report(
0254         publication: DeliveryPublication) -> str:
0255     campaigns = publication.projection["campaigns"]
0256     return json.dumps(
0257         {
0258             "scope": publication.update.scope,
0259             "component": publication.update.name,
0260             "revision": publication.update.revision,
0261             "content_changed": publication.update.content_changed,
0262             "campaigns": {
0263                 name: block["totals"] for name, block in campaigns.items()
0264             },
0265             "observed_at": publication.observed_at.isoformat(),
0266         },
0267         indent=2,
0268         sort_keys=True,
0269     )