Merge pull request #355 from opensensor/qoe-services

Implement Stream Health & QoE Monitoring (Phase 1)

Author: matteius

CMakeLists.txt

@@ -433,6 +433,7 @@ list(APPEND VIDEO_SOURCES ${FFMPEG_LEAK_DETECTOR_SOURCES})
 
 # Add HLS sources
 file(GLOB HLS_SOURCES "src/video/hls/*.c")
+file(GLOB TELEMETRY_SOURCES "src/telemetry/*.c")
 
 # Add unified HLS thread sources explicitly to ensure they're included
 set(HLS_UNIFIED_THREAD_SOURCES
@@ -462,6 +463,7 @@ set(SOURCES
         ${VIDEO_SOURCES}
         ${HLS_SOURCES}
         ${HLS_UNIFIED_THREAD_SOURCES}
+        ${TELEMETRY_SOURCES}
         ${EZXML_SOURCES}
 )
 

include/telemetry/player_telemetry.h

@@ -0,0 +1,65 @@
+/**
+ * @file player_telemetry.h
+ * @brief Client-side playback QoE telemetry ingestion
+ *
+ * Accepts QoE events from the web player (TTFF, rebuffer, WebRTC RTT)
+ * via POST /api/telemetry/player and stores them in a ring buffer for
+ * Prometheus exposition and health reporting.
+ */
+
+#ifndef LIGHTNVR_PLAYER_TELEMETRY_H
+#define LIGHTNVR_PLAYER_TELEMETRY_H
+
+#include <stdint.h>
+#include <time.h>
+#include "core/config.h"
+
+#define PLAYER_TELEMETRY_RING_SIZE 256
+
+/**
+ * Single player telemetry event
+ */
+typedef struct {
+    char stream_name[MAX_STREAM_NAME];
+    char session_id[64];
+    char transport[16];       /* "webrtc", "hls", "mse" */
+    double ttff_ms;           /* time-to-first-frame */
+    int rebuffer_count;
+    double rebuffer_duration_ms;
+    int resolution_switches;
+    double webrtc_rtt_ms;
+    time_t timestamp;
+} player_telemetry_event_t;
+
+/**
+ * Initialize the player telemetry subsystem
+ */
+void player_telemetry_init(void);
+
+/**
+ * Shutdown the player telemetry subsystem
+ */
+void player_telemetry_shutdown(void);
+
+/**
+ * Record a player telemetry event
+ *
+ * @param event Pointer to the event to record (copied into ring buffer)
+ */
+void player_telemetry_record(const player_telemetry_event_t *event);
+
+/**
+ * Snapshot recent player telemetry events
+ *
+ * @param out       Pre-allocated array to receive events
+ * @param max_count Size of out array
+ * @return Number of events copied (most recent first)
+ */
+int player_telemetry_snapshot(player_telemetry_event_t *out, int max_count);
+
+/**
+ * Get the number of events currently stored
+ */
+int player_telemetry_count(void);
+
+#endif /* LIGHTNVR_PLAYER_TELEMETRY_H */

include/telemetry/stream_metrics.h

@@ -0,0 +1,233 @@
+/**
+ * @file stream_metrics.h
+ * @brief In-process stream health metric counters and ring buffers
+ *
+ * Provides per-stream QoS/QoE metric collection for Prometheus exposition,
+ * JSON health summaries, and UI sparklines. Zero external dependencies.
+ *
+ * Thread safety: metric slots use pthread_rwlock_t. Writer threads (recording,
+ * HLS, stream_state) acquire write locks briefly to update counters. The HTTP
+ * thread acquires read locks to snapshot metrics for API responses.
+ */
+
+#ifndef LIGHTNVR_STREAM_METRICS_H
+#define LIGHTNVR_STREAM_METRICS_H
+
+#include <stdint.h>
+#include <stdbool.h>
+#include <stdatomic.h>
+#include <time.h>
+#include <pthread.h>
+#include "core/config.h"
+
+/* Ring buffer: 5-second sample interval, 720 samples = 1 hour of sparkline data */
+#define METRICS_RING_SIZE 720
+
+/* Thresholds for health status classification */
+#define STREAM_HEALTH_FRAME_TIMEOUT_UP    5   /* seconds: no frame → not UP */
+#define STREAM_HEALTH_FRAME_TIMEOUT_DOWN  10  /* seconds: no frame → DOWN */
+#define STREAM_HEALTH_FPS_RATIO           0.5 /* current/configured < this → degraded */
+
+/* Gap detection threshold */
+#define RECORDING_GAP_THRESHOLD_SEC 5
+
+/**
+ * Stream health status
+ */
+typedef enum {
+    STREAM_HEALTH_UP = 0,       /* Receiving frames within thresholds */
+    STREAM_HEALTH_DEGRADED,     /* Receiving frames but below FPS threshold */
+    STREAM_HEALTH_DOWN          /* No frames received recently */
+} stream_health_status_t;
+
+/**
+ * Single ring buffer sample for sparkline rendering
+ */
+typedef struct {
+    float fps;
+    float bitrate_kbps;
+    time_t timestamp;
+} metrics_ring_sample_t;
+
+/**
+ * Per-stream metric slot
+ *
+ * Gauges use _Atomic for lock-free reads from the HTTP thread.
+ * The rwlock protects the ring buffer and multi-field updates.
+ */
+typedef struct {
+    /* Identity */
+    char stream_name[MAX_STREAM_NAME];
+    bool active;                          /* slot in use */
+
+    /* Health status (recomputed by sampler thread) */
+    atomic_int health_status;             /* stream_health_status_t */
+    atomic_int stream_up;                 /* 1 if UP, 0 otherwise */
+
+    /* Gauges */
+    atomic_int_fast64_t last_frame_ts;    /* unix epoch seconds of last frame */
+    double current_fps;                   /* computed FPS (updated by sampler) */
+    double current_bitrate_bps;           /* computed bitrate in bits/s (updated by sampler) */
+    double connection_latency_ms;         /* last (re)connect latency */
+    double configured_fps;               /* configured FPS for health threshold */
+
+    /* Monotonic counters */
+    atomic_uint_fast64_t frames_total;
+    atomic_uint_fast64_t frames_dropped;
+    atomic_uint_fast64_t bytes_received;
+    atomic_uint_fast64_t reconnects_total;
+    atomic_uint_fast64_t uptime_seconds;
+
+    /* Error counters by type */
+    atomic_uint_fast64_t error_decode;
+    atomic_uint_fast64_t error_timeout;
+    atomic_uint_fast64_t error_protocol;
+    atomic_uint_fast64_t error_io;
+
+    /* Recording metrics */
+    atomic_int recording_active;          /* 1 if recording, 0 otherwise */
+    atomic_uint_fast64_t recording_bytes_written;
+    atomic_uint_fast64_t recording_segments_total;
+    atomic_uint_fast64_t recording_gaps_total;
+
+    /* Ring buffer for sparkline data (protected by rwlock) */
+    metrics_ring_sample_t ring[METRICS_RING_SIZE];
+    int ring_head;                        /* next write position */
+    int ring_count;                       /* number of valid samples */
+
+    /* Gap detection state */
+    time_t last_segment_end_time;
+    int expected_segment_duration;
+
+    /* Stream start time for uptime calculation */
+    time_t stream_start_time;
+
+    /* FPS/bitrate computation state (used internally by sampler) */
+    uint64_t prev_frames_total;
+    uint64_t prev_bytes_received;
+    time_t prev_sample_time;
+
+    /* Read-write lock for this slot */
+    pthread_rwlock_t lock;
+} stream_metrics_t;
+
+/**
+ * Initialize the metrics subsystem
+ *
+ * @param max_streams Maximum number of stream slots to allocate
+ * @return 0 on success, -1 on failure
+ */
+int metrics_init(int max_streams);
+
+/**
+ * Shutdown the metrics subsystem and free resources
+ */
+void metrics_shutdown(void);
+
+/**
+ * Get or create a metric slot for a stream.
+ * Returns NULL if all slots are full and the name doesn't match any existing slot.
+ *
+ * @param stream_name Stream name
+ * @return Pointer to the stream's metric slot, or NULL
+ */
+stream_metrics_t *metrics_get_slot(const char *stream_name);
+
+/**
+ * Release (deactivate) a metric slot for a stream
+ *
+ * @param stream_name Stream name
+ */
+void metrics_release_slot(const char *stream_name);
+
+/**
+ * Record a received frame (video or audio)
+ *
+ * @param stream_name Stream name
+ * @param bytes       Packet size in bytes
+ * @param is_video    true for video frames, false for audio
+ */
+void metrics_record_frame(const char *stream_name, int bytes, bool is_video);
+
+/**
+ * Record a dropped frame
+ *
+ * @param stream_name Stream name
+ */
+void metrics_record_drop(const char *stream_name);
+
+/**
+ * Record a stream error
+ *
+ * @param stream_name Stream name
+ * @param error_type  One of: "decode", "timeout", "protocol", "io"
+ */
+void metrics_record_error(const char *stream_name, const char *error_type);
+
+/**
+ * Record a reconnection event
+ *
+ * @param stream_name Stream name
+ */
+void metrics_record_reconnect(const char *stream_name);
+
+/**
+ * Record a completed recording segment (for gap detection)
+ *
+ * @param stream_name Stream name
+ * @param start_time  Segment start timestamp
+ * @param end_time    Segment end timestamp
+ * @param bytes       Bytes written in this segment
+ */
+void metrics_record_segment_complete(const char *stream_name, time_t start_time,
+                                     time_t end_time, uint64_t bytes);
+
+/**
+ * Set recording active state for a stream
+ *
+ * @param stream_name Stream name
+ * @param active      true if recording is active
+ */
+void metrics_set_recording_active(const char *stream_name, bool active);
+
+/**
+ * Set the connection latency for the last (re)connect
+ *
+ * @param stream_name Stream name
+ * @param latency_ms  Latency in milliseconds
+ */
+void metrics_set_connection_latency(const char *stream_name, double latency_ms);
+
+/**
+ * Set the configured FPS for a stream (used for health threshold computation)
+ *
+ * @param stream_name Stream name
+ * @param fps         Configured FPS value
+ */
+void metrics_set_configured_fps(const char *stream_name, double fps);
+
+/**
+ * Thread-safe snapshot of all active stream metrics
+ *
+ * @param out_array   Pre-allocated array to receive snapshots
+ * @param max_count   Size of out_array
+ * @return Number of active streams copied
+ */
+int metrics_snapshot_all(stream_metrics_t *out_array, int max_count);
+
+/**
+ * Get ring buffer data for a stream's sparkline
+ *
+ * @param stream_name Stream name
+ * @param out         Pre-allocated array to receive samples
+ * @param max_count   Size of out array
+ * @return Number of samples copied (oldest first)
+ */
+int metrics_get_ring_data(const char *stream_name, metrics_ring_sample_t *out, int max_count);
+
+/**
+ * Get the total number of allocated metric slots
+ */
+int metrics_get_max_streams(void);
+
+#endif /* LIGHTNVR_STREAM_METRICS_H */

include/video/mp4_writer_thread.h

@@ -38,6 +38,9 @@ typedef struct {
     // If set, the next segment should start by writing this packet first.
     // This intentionally duplicates the boundary keyframe to bias toward overlap (no gaps).
     AVPacket *pending_video_keyframe;
+
+    // Stream name for telemetry instrumentation
+    char stream_name[MAX_STREAM_NAME];
 } segment_info_t;
 
 /**

include/web/api_handlers_metrics.h

@@ -0,0 +1,24 @@
+/**
+ * @file api_handlers_metrics.h
+ * @brief Prometheus metrics endpoint and player telemetry ingest handlers
+ */
+
+#ifndef LIGHTNVR_API_HANDLERS_METRICS_H
+#define LIGHTNVR_API_HANDLERS_METRICS_H
+
+#include "web/request_response.h"
+
+/**
+ * GET /api/metrics
+ * Returns all stream health metrics in Prometheus exposition text format.
+ */
+void handle_get_metrics(const http_request_t *req, http_response_t *res);
+
+/**
+ * POST /api/telemetry/player
+ * Accepts client-side playback QoE events from the web player.
+ * Returns 204 No Content.
+ */
+void handle_post_player_telemetry(const http_request_t *req, http_response_t *res);
+
+#endif /* LIGHTNVR_API_HANDLERS_METRICS_H */

src/core/main.c

@@ -43,6 +43,8 @@
 #include "video/ffmpeg_leak_detector.h"
 #include "video/onvif_motion_recording.h"
 #include "core/curl_init.h"
+#include "telemetry/stream_metrics.h"
+#include "telemetry/player_telemetry.h"
 
 // Include go2rtc headers if USE_GO2RTC is defined
 #ifdef USE_GO2RTC
@@ -804,6 +806,13 @@ int main(int argc, char *argv[]) {
         goto cleanup;
     }
 
+    // Initialize telemetry subsystem
+    if (metrics_init(config.max_streams) != 0) {
+        log_error("Failed to initialize metrics subsystem");
+        goto cleanup;
+    }
+    player_telemetry_init();
+
     // Initialize go2rtc integration if enabled
     #ifdef USE_GO2RTC
     if (!config.go2rtc_enabled) {
@@ -1204,6 +1213,11 @@ int main(int argc, char *argv[]) {
     log_info("Stopping health check system...");
     cleanup_health_check_system();
 
+    // Shutdown telemetry subsystem
+    log_info("Shutting down telemetry...");
+    metrics_shutdown();
+    player_telemetry_shutdown();
+
     // Now that we're in the main thread (not signal handler), we can safely
     // call initiate_shutdown() which uses mutexes and logging
     initiate_shutdown();