Skip to content

lablink_allocator_service.db.metrics

Persistence for the Tier-1 session-metrics columns.

A column-family split, not a table split: the SessionMetrics / Seconds / Gpu* columns live on the VM table itself, which is why this class needs table_name where OperationsDatabase needs only a pool. Nothing outside this module reads or writes those columns, with one soft exception — VmDatabase.get_all_vms_for_export does SELECT * via get_column_names() and therefore returns them too. That is a schema-shape dependency, not a code one.

Persistence for session-metrics columns on the VM table.

Parameters:

Name Type Description Default
pool

A psycopg2 connection pool, shared with the other db classes.

 required
table_name str

The VM table name (VM_TABLE_NAME in production).

 required
Source code in packages/allocator/src/lablink_allocator_service/db/metrics.py 
78
79
80
def __init__(self, pool, table_name: str):
    self._pool = pool
    self.table_name = table_name

Seal every unsealed VM (called from the destroy paths).

Returns:

Name Type Description
int int

number of rows sealed.
Source code in packages/allocator/src/lablink_allocator_service/db/metrics.py 
166
167
168
169
170
171
172
173
174
175
176
177
def bulk_seal_session_metrics(self) -> int:
    """Seal every unsealed VM (called from the destroy paths).

    Returns:
        int: number of rows sealed.
    """
    with self._cursor as cursor:
        cursor.execute(
            f"UPDATE {self.table_name} SET SessionMetricsSealedAt = NOW() "
            "WHERE SessionMetricsSealedAt IS NULL"
        )
        return cursor.rowcount or 0

Aggregate the cohort summary for the admin page.

The AS aliases below ARE the row keys _build_summary reads — the dict-rows cursor keys each row by them (Postgres folds the unquoted CamelCase column names to bare lowercase, hence the explicit snake_case aliases).

Source code in packages/allocator/src/lablink_allocator_service/db/metrics.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def get_session_metrics_summary(self) -> dict:
    """Aggregate the cohort summary for the admin page.

    The AS aliases below ARE the row keys `_build_summary` reads —
    the dict-rows cursor keys each row by them (Postgres folds the
    unquoted CamelCase column names to bare lowercase, hence the
    explicit snake_case aliases).
    """
    with self._cursor as cursor:
        cursor.execute(
            f"""
            SELECT HostName                 AS host_name,
                   SessionMetricsStartedAt  AS session_metrics_started_at,
                   SecondsToFirstSleapLabel AS seconds_to_first_sleap_label,
                   SecondsToFirstSleapTrain AS seconds_to_first_sleap_train,
                   SecondsToFirstSleapTrack AS seconds_to_first_sleap_track,
                   SecondsInSubjectSoftware AS seconds_in_subject_software,
                   GpuActiveSeconds         AS gpu_active_seconds,
                   MaxLabeledFrames         AS max_labeled_frames,
                   TrainingEpochsCompleted  AS training_epochs_completed
            FROM {self.table_name}
            """
        )
        rows = cursor.fetchall()
    return _build_summary(rows)

Last-write-wins UPDATE of session-metrics columns.

Atomic with respect to seal: the sealed-row check is folded into the UPDATE's WHERE clause, so a concurrent bulk_seal_session_metrics cannot land between a separate SELECT and a separate UPDATE. When the UPDATE affects zero rows, a follow-up existence SELECT classifies the failure as LookupError (no such row) or ValueError (row exists but is sealed).

Returns:

Type Description

The row's authoritative SessionStartedAt (datetime or

None), via RETURNING on the same UPDATE — the route echoes

it so the client can heal a lost session anchor.

Raises:

Type Description
LookupError

if hostname unknown.
ValueError

if the row is already sealed.
Source code in packages/allocator/src/lablink_allocator_service/db/metrics.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def update_session_metrics(self, hostname: str, payload: dict):
    """Last-write-wins UPDATE of session-metrics columns.

    Atomic with respect to seal: the sealed-row check is folded into
    the UPDATE's WHERE clause, so a concurrent ``bulk_seal_session_metrics``
    cannot land between a separate SELECT and a separate UPDATE.
    When the UPDATE affects zero rows, a follow-up existence SELECT
    classifies the failure as ``LookupError`` (no such row) or
    ``ValueError`` (row exists but is sealed).

    Returns:
        The row's authoritative ``SessionStartedAt`` (datetime or
        None), via RETURNING on the same UPDATE — the route echoes
        it so the client can heal a lost session anchor.

    Raises:
        LookupError: if hostname unknown.
        ValueError: if the row is already sealed.
    """
    counters = payload.get("counters", {})
    with self._cursor as cursor:
        cursor.execute(
            f"""
            UPDATE {self.table_name} SET
              SessionMetricsStartedAt      = COALESCE(SessionMetricsStartedAt, %s),
              SessionMetricsLastReportedAt = NOW(),
              SecondsInSubjectSoftware     = %s,
              SecondsInTerminal            = %s,
              SecondsInBrowser             = %s,
              SecondsInOther               = %s,
              GpuActiveSeconds             = %s,
              GpuUtilPeak                  = %s,
              VramUsedPeakMb               = %s,
              SecondsToFirstSleapLabel     = %s,
              SecondsToFirstSleapTrain     = %s,
              SecondsToFirstSleapTrack     = %s,
              MaxLabeledFrames             = %s,
              TrainingEpochsCompleted      = %s,
              TrainingFinalLoss            = %s,
              SessionMetricsRaw            = %s
            WHERE HostName = %s AND SessionMetricsSealedAt IS NULL
            RETURNING SessionStartedAt AS session_started_at
            """,
            (
                payload.get("session_started_at"),
                counters.get("seconds_in_subject_software"),
                counters.get("seconds_in_terminal"),
                counters.get("seconds_in_browser"),
                counters.get("seconds_in_other"),
                counters.get("gpu_active_seconds"),
                counters.get("gpu_util_peak"),
                counters.get("vram_used_peak_mb"),
                counters.get("seconds_to_first_sleap_label"),
                counters.get("seconds_to_first_sleap_train"),
                counters.get("seconds_to_first_sleap_track"),
                counters.get("max_labeled_frames"),
                counters.get("training_epochs_completed"),
                counters.get("training_final_loss"),
                Json(counters),
                hostname,
            ),
        )
        if cursor.rowcount >= 1:
            row = cursor.fetchone()
            return row["session_started_at"] if row else None

        # UPDATE matched zero rows — classify so the route can return
        # 404 vs 409. HostName is PRIMARY KEY on vms, so this SELECT
        # can return at most one row.
        cursor.execute(
            f"SELECT 1 FROM {self.table_name} WHERE HostName = %s",
            (hostname,),
        )
        if cursor.fetchone() is None:
            raise LookupError(f"VM {hostname} not found")
        raise ValueError(f"VM {hostname} session is sealed")