lablink_allocator_service.routes.vm_telemetry¶

`lablink_allocator_service.routes.vm_telemetry` ¶

VM telemetry: client-VM writes and the operator's read side.

Both directions over the same columns live here on purpose — POST and GET /api/vm-status are two ends of one column, and splitting them by auth gate would scatter one concern across two files.

Auth differs per direction, so read the decorators carefully: @require_client_secret for anything a client VM pushes, @auth.login_required for anything an operator reads.

Attributes¶

`MAX_LOG_SIZE = 1 * 1024 * 1024` `module-attribute` ¶

`bp = Blueprint('vm_telemetry', name)` `module-attribute` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

Functions:¶

`get_all_vm_status()` ¶

Get the status of all VMs.

Returns a JSON object mapping each VM hostname to its current status. Used by the admin dashboard.

Success Response:

Code: 200 OK

Content:

{
  "lablink-vm-prod-1": "running",
  "lablink-vm-prod-2": "initializing"
}

Error Response:

Code: 404 Not Found if no VMs are found in the database.
Code: 500 Internal Server Error on failure.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/vm-status", methods=["GET"])
@auth.login_required
def get_all_vm_status():
    """Get the status of all VMs.

    Returns a JSON object mapping each VM hostname to its current status.
    Used by the admin dashboard.

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:**
      ```json
      {
        "lablink-vm-prod-1": "running",
        "lablink-vm-prod-2": "initializing"
      }
      ```

    **Error Response:**

    - **Code:** `404 Not Found` if no VMs are found in the database.
    - **Code:** `500 Internal Server Error` on failure.
    """
    from lablink_allocator_service import main

    try:
        vm_status = main.database.get_all_vm_status()
        if not vm_status:
            return jsonify({"error": "No VMs found."}), 404

        return jsonify(vm_status), 200
    except Exception as e:
        logger.error(f"Error getting all VM status: {e}")
        return jsonify({"error": "Failed to get VM status."}), 500

`get_vm_logs_by_hostname(hostname)` ¶

Get the stored logs for a specific VM.

Used by the admin log viewer page.

URL Parameters:

hostname (string, required): The hostname of the VM.

Query Parameters:

errors_only (boolean, optional, default false): When true, both cloud_init_logs and docker_logs are reduced to their error lines — lines carrying an uppercase level token (ERROR, CRITICAL, FATAL, plus cloudflared's ERR/FTL and postgres' PANIC), together with any traceback that follows them. Matching is case-sensitive so the client startup script's --retry-all-errors and status=error lines are not treated as errors, and a short denylist suppresses client reporting-path failures that do not affect the VM's usability (GPU-health and in-use-status report failures, and session-metrics push failures). A stream with no error lines becomes "", not null.

Success Response:

Code: 200 OK

Content:

{
  "hostname": "lablink-vm-prod-1",
  "cloud_init_logs": "Starting cloud-init...\n...",
  "docker_logs": "[start] ALLOCATOR_HOST: ...\n...",
  "logs": "Starting cloud-init...\n..."
}

Error Response:

Code: 404 Not Found if the VM is not found.
Code: 503 Service Unavailable if the logs are not yet available because the VM's log shipper has not started reporting yet.
Code: 500 Internal Server Error on failure.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/vm-logs/<hostname>", methods=["GET"])
@auth.login_required
def get_vm_logs_by_hostname(hostname):
    """Get the stored logs for a specific VM.

    Used by the admin log viewer page.

    **URL Parameters:**

    - `hostname` (string, required): The hostname of the VM.

    **Query Parameters:**

    - `errors_only` (boolean, optional, default `false`): When `true`, both
      `cloud_init_logs` and `docker_logs` are reduced to their error lines —
      lines carrying an uppercase level token (`ERROR`, `CRITICAL`, `FATAL`,
      plus cloudflared's `ERR`/`FTL` and postgres' `PANIC`), together with
      any traceback that follows them. Matching is case-sensitive so the
      client startup script's `--retry-all-errors` and `status=error` lines
      are not treated as errors, and a short denylist suppresses client
      reporting-path failures that do not affect the VM's usability
      (GPU-health and in-use-status report failures, and session-metrics
      push failures). A stream with no error lines becomes `""`, not `null`.

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:**
      ```json
      {
        "hostname": "lablink-vm-prod-1",
        "cloud_init_logs": "Starting cloud-init...\\n...",
        "docker_logs": "[start] ALLOCATOR_HOST: ...\\n...",
        "logs": "Starting cloud-init...\\n..."
      }
      ```

    **Error Response:**

    - **Code:** `404 Not Found` if the VM is not found.
    - **Code:** `503 Service Unavailable` if the logs are not yet available
      because the VM's log shipper has not started reporting yet.
    - **Code:** `500 Internal Server Error` on failure.
    """
    from lablink_allocator_service import main

    try:
        if not main.database.vm_exists(hostname):
            logger.error(f"VM with hostname {hostname} not found.")
            return jsonify({"error": "VM not found."}), 404

        # If the logs are empty but the vm is initializing, return a 503 status
        logs_data = main.database.get_vm_logs(hostname=hostname)
        status = main.database.get_status_by_hostname(hostname)
        if logs_data is None and status == "initializing":
            return jsonify({"error": "VM is initializing."}), 503

        cloud_init_logs = (logs_data or {}).get("cloud_init_logs")
        docker_logs = (logs_data or {}).get("docker_logs")

        # Same spelling as routes/metrics.py:122's include_logs flag.
        if request.args.get("errors_only", "false").lower() == "true":
            cloud_init_logs = filter_errors(cloud_init_logs)
            docker_logs = filter_errors(docker_logs)

        return jsonify({
            "hostname": hostname,
            "cloud_init_logs": cloud_init_logs,
            "docker_logs": docker_logs,
            "logs": "\n".join(filter(None, [cloud_init_logs, docker_logs])) or None,
        }), 200
    except Exception as e:
        logger.error(f"Error getting VM logs: {e}")
        return jsonify({"error": "Failed to get VM logs."}), 500

`heartbeat()` ¶

Record a client-VM liveness heartbeat.

Periodic liveness ping. Lets the allocator detect silent failures — a dead container, a broken network, a hung host, an OOM, or an out-of-band VM termination — that no other endpoint would reveal. The body also carries cheap health signals for early warning.

Auth: Client secret (resolved from the vm_id field).

Request Body: application/json

{
  "vm_id": "lablink-vm-prod-1",
  "boot_id": "…",
  "disk_free_pct": 62
}

Client Usage: Sent by the client's heartbeat service every HEARTBEAT_INTERVAL_SECONDS. A client that stops heartbeating for longer than the staleness window becomes eligible for automatic recovery.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/heartbeat", methods=["POST"])
@require_client_secret
def heartbeat():
    """Record a client-VM liveness heartbeat.

    Periodic liveness ping. Lets the allocator detect silent failures — a
    dead container, a broken network, a hung host, an OOM, or an out-of-band
    VM termination — that no other endpoint would reveal. The body also
    carries cheap health signals for early warning.

    Auth: Client secret (resolved from the `vm_id` field).

    **Request Body:** `application/json`

    ```json
    {
      "vm_id": "lablink-vm-prod-1",
      "boot_id": "…",
      "disk_free_pct": 62
    }
    ```

    **Client Usage:** Sent by the client's `heartbeat` service every
    `HEARTBEAT_INTERVAL_SECONDS`. A client that stops heartbeating for longer
    than the staleness window becomes eligible for automatic recovery.
    """
    from lablink_allocator_service import main

    data = request.get_json() or {}
    hostname = data.get("vm_id")
    if not hostname:
        return jsonify({"error": "vm_id is required."}), 400

    boot_id = data.get("boot_id")
    disk_free_pct = data.get("disk_free_pct")

    try:
        ok = main.database.record_heartbeat(
            hostname=hostname,
            boot_id=boot_id,
            disk_free_pct=disk_free_pct,
        )
        if not ok:
            return jsonify({"error": "Unknown hostname."}), 404
        return jsonify({"ok": True}), 200
    except Exception as e:
        logger.error(f"Error recording heartbeat for {hostname}: {e}")
        return jsonify({"error": "Failed to record heartbeat."}), 500

`receive_vm_logs(hostname)` ¶

Receive and store logs pushed from a VM.

Receives batched log lines pushed from a VM by the log_shipper.sh script, which tails the VM's cloud-init output and the client container's Docker logs.

URL Parameters:

hostname (string, required): The hostname of the VM.

Request Body: application/json

{
  "log_group": "/aws/ec2/lablink",
  "log_stream": "lablink-vm-prod-1",
  "messages": ["log line 1", "log line 2"]
}

Success Response:

Code: 200 OK
Content: {"message": "VM logs posted successfully."}

Error Response:

Code: 400 Bad Request if required fields are missing.
Code: 404 Not Found if the VM does not exist.
Code: 500 Internal Server Error on failure.

Client Usage: Not called directly from packages/client code. The log_shipper.sh script, installed on each VM by the OpenTofu user-data script, tails the VM's cloud-init output log and the client container's Docker logs, batches new lines, and POSTs them here. ANSI escapes are stripped and each log type is capped at 1MB.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/vm-logs/<hostname>", methods=["POST"])
@require_client_secret
def receive_vm_logs(hostname):
    """Receive and store logs pushed from a VM.

    Receives batched log lines pushed from a VM by the `log_shipper.sh`
    script, which tails the VM's `cloud-init` output and the client
    container's Docker logs.

    **URL Parameters:**

    - `hostname` (string, required): The hostname of the VM.

    **Request Body:** `application/json`

    ```json
    {
      "log_group": "/aws/ec2/lablink",
      "log_stream": "lablink-vm-prod-1",
      "messages": ["log line 1", "log line 2"]
    }
    ```

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:** `{"message": "VM logs posted successfully."}`

    **Error Response:**

    - **Code:** `400 Bad Request` if required fields are missing.
    - **Code:** `404 Not Found` if the VM does not exist.
    - **Code:** `500 Internal Server Error` on failure.

    **Client Usage:** Not called directly from `packages/client` code. The
    `log_shipper.sh` script, installed on each VM by the OpenTofu user-data
    script, tails the VM's `cloud-init` output log and the client
    container's Docker logs, batches new lines, and POSTs them here. ANSI
    escapes are stripped and each log type is capped at 1MB.
    """
    from lablink_allocator_service import main

    try:
        data = request.get_json()
        log_group = data.get("log_group")
        messages = data.get("messages", [])

        if not log_group or not messages:
            return (
                jsonify({"error": "Log group and messages are required."}),
                400,
            )

        # Check if the VM exists in the database
        if not main.database.vm_exists(hostname):
            logger.error(f"VM with hostname {hostname} does not exist.")
            return jsonify({"error": "VM not found."}), 404

        logger.debug(
            f"Received logs for {log_group}/{hostname}: {len(messages)} messages"
        )

        # Strip ANSI escape codes and drop empty lines
        messages = [strip_ansi(m) for m in messages]
        messages = [m for m in messages if m.strip()]

        if not messages:
            return jsonify({"message": "No log messages after filtering."}), 200

        # Determine log type from log_group
        log_type = "docker" if log_group.endswith("-docker") else "cloud_init"

        # Save the logs to the database atomically (cap at 1MB per log type)
        new_logs = "\n".join(messages)
        main.database.append_logs_by_hostname(
            hostname=hostname,
            new_logs=new_logs,
            log_type=log_type,
            max_size=MAX_LOG_SIZE,
        )

        return jsonify({"message": "VM logs posted successfully."}), 200
    except Exception as e:
        logger.error(f"Error receiving VM logs: {e}")
        return jsonify({"error": "Failed to post VM logs."}), 500

`receive_vm_metrics(hostname)` ¶

Receive and store startup metrics from a VM.

Called by the client VM to post timing metrics for cloud-init and container startup.

URL Parameters:

hostname (string, required): The hostname of the VM reporting metrics.

Request Body: application/json

{
  "cloud_init_start": 1678886400,
  "cloud_init_end": 1678886460,
  "cloud_init_duration_seconds": 60
}

Success Response:

Code: 200 OK
Content: {"message": "VM metrics posted successfully."}

Error Response:

Code: 404 Not Found if the VM does not exist.
Code: 500 Internal Server Error on failure.

Client Usage: user_data.sh posts cloud-init timings during boot, and the client container's start.sh entrypoint posts its own start/end times at the end of its startup sequence. This helps in monitoring startup duration.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/vm-metrics/<hostname>", methods=["POST"])
@require_client_secret
def receive_vm_metrics(hostname):
    """Receive and store startup metrics from a VM.

    Called by the client VM to post timing metrics for `cloud-init` and
    container startup.

    **URL Parameters:**

    - `hostname` (string, required): The hostname of the VM reporting
      metrics.

    **Request Body:** `application/json`

    ```json
    {
      "cloud_init_start": 1678886400,
      "cloud_init_end": 1678886460,
      "cloud_init_duration_seconds": 60
    }
    ```

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:** `{"message": "VM metrics posted successfully."}`

    **Error Response:**

    - **Code:** `404 Not Found` if the VM does not exist.
    - **Code:** `500 Internal Server Error` on failure.

    **Client Usage:** `user_data.sh` posts cloud-init timings during boot,
    and the client container's `start.sh` entrypoint posts its own start/end
    times at the end of its startup sequence. This helps in monitoring
    startup duration.
    """
    from lablink_allocator_service import main

    try:
        data = request.get_json()

        if not main.database.vm_exists(hostname=hostname):
            logger.error(f"VM with hostname {hostname} does not exist.")
            return jsonify({"error": "VM not found."}), 404

        main.database.touch_last_seen(hostname=hostname)
        # Update VM metrics and calculate total startup time atomically
        # This combines two database operations into one for better performance
        main.database.update_vm_metrics_atomic(hostname=hostname, metrics=data)

        logger.debug(f"Received metrics for {hostname}")
        return jsonify({"message": "VM metrics posted successfully."}), 200

    except Exception as e:
        logger.error(f"Error receiving VM metrics for {hostname}: {e}", exc_info=True)
        return jsonify({"error": "Failed to post VM metrics."}), 500

`update_gpu_health()` ¶

Update the GPU health status of a VM.

Request Body: application/json

{
  "hostname": "lablink-vm-prod-1",
  "gpu_status": "healthy"
}

Success Response:

Code: 200 OK
Content: {"message": "GPU health status updated successfully."}

Error Response:

Code: 400 Bad Request if hostname or gpu_status is missing.
Code: 500 Internal Server Error on failure.

Client Usage: Called by the check_gpu service, which is started by start.sh and runs continuously. The service periodically runs nvidia-smi, determines the GPU status (Healthy, Unhealthy, or N/A), and reports whenever the status changes. The value lands in the Healthy column — the auto-reboot service decides what to do about an unhealthy VM.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/gpu_health", methods=["POST"])
@require_client_secret
def update_gpu_health():
    """Update the GPU health status of a VM.

    **Request Body:** `application/json`

    ```json
    {
      "hostname": "lablink-vm-prod-1",
      "gpu_status": "healthy"
    }
    ```

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:** `{"message": "GPU health status updated successfully."}`

    **Error Response:**

    - **Code:** `400 Bad Request` if `hostname` or `gpu_status` is missing.
    - **Code:** `500 Internal Server Error` on failure.

    **Client Usage:** Called by the `check_gpu` service, which is started by
    `start.sh` and runs continuously. The service periodically runs
    `nvidia-smi`, determines the GPU status (`Healthy`, `Unhealthy`, or
    `N/A`), and reports whenever the status changes. The value lands in the
    `Healthy` column — the auto-reboot service decides what to do about an
    unhealthy VM.
    """
    from lablink_allocator_service import main

    data = request.get_json()
    gpu_status = data.get("gpu_status")
    hostname = data.get("hostname")
    if gpu_status is None or hostname is None:
        return jsonify({"error": "GPU status and hostname are required."}), 400

    try:
        main.database.touch_last_seen(hostname=hostname)
        main.database.update_health(hostname=hostname, healthy=gpu_status)
        logger.debug(f"Updated GPU health status for {hostname} to {gpu_status}")
        return jsonify({"message": "GPU health status updated successfully."}), 200
    except Exception as e:
        logger.error(f"Error updating GPU health status: {e}")
        return jsonify({"error": "Failed to update GPU health status."}), 500

`update_inuse_status()` ¶

Update the in-use status of a VM.

Called by the client VM to indicate whether a user is actively using it.

Request Body: application/json

{
  "hostname": "lablink-vm-prod-1",
  "status": true
}

Success Response:

Code: 200 OK
Content: {"message": "In-use status updated successfully."}

Error Response:

Code: 400 Bad Request if hostname or status is missing.
Code: 500 Internal Server Error on failure.

Client Usage: Called by the update_inuse_status service, which is started by start.sh and runs continuously. The service monitors for the presence of the research software process (e.g., sleap). When the process starts, it sends status: true; when it stops, status: false. This lets the allocator know if a user is actively using the VM.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/update_inuse_status", methods=["POST"])
@require_client_secret
def update_inuse_status():
    """Update the in-use status of a VM.

    Called by the client VM to indicate whether a user is actively using it.

    **Request Body:** `application/json`

    ```json
    {
      "hostname": "lablink-vm-prod-1",
      "status": true
    }
    ```

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:** `{"message": "In-use status updated successfully."}`

    **Error Response:**

    - **Code:** `400 Bad Request` if `hostname` or `status` is missing.
    - **Code:** `500 Internal Server Error` on failure.

    **Client Usage:** Called by the `update_inuse_status` service, which is
    started by `start.sh` and runs continuously. The service monitors for the
    presence of the research software process (e.g., `sleap`). When the
    process starts, it sends `status: true`; when it stops, `status: false`.
    This lets the allocator know if a user is actively using the VM.
    """
    from lablink_allocator_service import main

    data = request.get_json()
    hostname = data.get("hostname")
    in_use = data.get("status")

    logger.debug(f"Updating in-use status for {hostname} to {in_use}")

    if not hostname:
        return jsonify({"error": "Hostname is required."}), 400

    try:
        main.database.update_vm_in_use(hostname=hostname, in_use=in_use)
        return jsonify({"message": "In-use status updated successfully."}), 200
    except Exception as e:
        logger.error(f"Error updating in-use status: {e}")
        return jsonify({"error": "Failed to update in-use status."}), 500

`update_vm_status()` ¶

Update the overall status of a VM.

Called by the client VM during its startup sequence to report its current status (initializing, running, error, rebooting).

Request Body: application/json

{
  "hostname": "lablink-vm-prod-1",
  "status": "running"
}

Success Response:

Code: 200 OK
Content: {"message": "VM status updated successfully."}

Error Response:

Code: 400 Bad Request if hostname or status is missing.
Code: 500 Internal Server Error on failure.

Client Usage: Not called from the packages/client code. It is called by the user_data.sh script during the client VM's initial boot sequence (cloud-init), which reports initializing, running, and error statuses so the allocator can track the VM's progress before the client service container has started. An error trap in user_data.sh automatically sends an error status if any command fails, which can then trigger the auto-reboot service.

Source code in packages/allocator/src/lablink_allocator_service/routes/vm_telemetry.py

@bp.route("/api/vm-status", methods=["POST"])
@require_client_secret
def update_vm_status():
    """Update the overall status of a VM.

    Called by the client VM during its startup sequence to report its
    current status (`initializing`, `running`, `error`, `rebooting`).

    **Request Body:** `application/json`

    ```json
    {
      "hostname": "lablink-vm-prod-1",
      "status": "running"
    }
    ```

    **Success Response:**

    - **Code:** `200 OK`
    - **Content:** `{"message": "VM status updated successfully."}`

    **Error Response:**

    - **Code:** `400 Bad Request` if `hostname` or `status` is missing.
    - **Code:** `500 Internal Server Error` on failure.

    **Client Usage:** Not called from the `packages/client` code. It is
    called by the `user_data.sh` script during the client VM's initial boot
    sequence (cloud-init), which reports `initializing`, `running`, and
    `error` statuses so the allocator can track the VM's progress before the
    client service container has started. An error trap in `user_data.sh`
    automatically sends an `error` status if any command fails, which can
    then trigger the auto-reboot service.
    """
    from lablink_allocator_service import main

    try:
        data = request.get_json()
        hostname = data.get("hostname")
        status = data.get("status")

        if not hostname or status is None:
            return jsonify({"error": "Hostname and status are required."}), 400

        main.database.touch_last_seen(hostname=hostname)
        main.database.update_vm_status(hostname=hostname, status=status)

        return jsonify({"message": "VM status updated successfully."}), 200
    except Exception as e:
        logger.error(f"Error updating VM status: {e}")
        return jsonify({"error": "Failed to update VM status."}), 500

lablink_allocator_service.routes.vm_telemetry¶