Skip to content

lablink_client_service.monitoring.aggregator

Rolling counters for the Tier 1 monitoring agent.

The aggregator holds in-memory state only. Each sampler produces a Sample once per tick (default 2 s); apply_sample updates the counters with that sample's effect on the running totals. The counters are serialised to JSON by the pusher and POSTed to the allocator.

There is no per-sample storage. Crashing the process loses at most the last push_interval_seconds of counter state.

One tick of input from the four samplers.

Source code in packages/client/src/lablink_client_service/monitoring/aggregator.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def apply_sample(c: SessionCounters, s: Sample) -> None:
    c.sample_count += 1

    bucket = (
        s.active_window_bucket if s.active_window_bucket in WINDOW_BUCKETS else "other"
    )
    attr = SUBJECT_FIELD if bucket == "subject" else f"seconds_in_{bucket}"
    setattr(c, attr, getattr(c, attr) + s.sample_interval_seconds)

    if s.gpu_util_pct > GPU_ACTIVE_UTIL_THRESHOLD:
        c.gpu_active_seconds += s.sample_interval_seconds
    if s.gpu_util_pct > c.gpu_util_peak:
        c.gpu_util_peak = s.gpu_util_pct
    if s.vram_mb > c.vram_used_peak_mb:
        c.vram_used_peak_mb = s.vram_mb

    elapsed = int((s.ts - c.session_started_at).total_seconds())
    if "sleap-label" in s.processes_seen and c.seconds_to_first_sleap_label is None:
        c.seconds_to_first_sleap_label = elapsed
    if "sleap-train" in s.processes_seen and c.seconds_to_first_sleap_train is None:
        c.seconds_to_first_sleap_train = elapsed
    if "sleap-track" in s.processes_seen and c.seconds_to_first_sleap_track is None:
        c.seconds_to_first_sleap_track = elapsed

    if s.max_labeled_frames is not None and s.max_labeled_frames > c.max_labeled_frames:
        c.max_labeled_frames = s.max_labeled_frames
    if (
        s.training_epochs_completed is not None
        and s.training_epochs_completed > c.training_epochs_completed
    ):
        c.training_epochs_completed = s.training_epochs_completed
    if s.training_final_loss is not None:
        c.training_final_loss = s.training_final_loss
Source code in packages/client/src/lablink_client_service/monitoring/aggregator.py 
63
64
def new_counters(session_started_at: datetime) -> SessionCounters:
    return SessionCounters(session_started_at=session_started_at)