streamtop 1.5.3

Terminal HLS, DASH, and IPTV stream monitor with wire probes, TR 101 290, SCTE-35, and Prometheus metrics
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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
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

use chrono::Utc;

use crate::engine::channel_stats::record_channel_drop;
use crate::engine::g2g::{compute_g2g, wall_now_unix_ms};
use crate::engine::linter::SpecLinter;
use crate::engine::metrics::update_metrics;
use crate::engine::middlebox::classify_transport_failure;
use crate::engine::redirect::RedirectLoopError;
use crate::models::{
    DiagCategory, DiagSeverity, DiagnosticFinding, DiagnosticReasonCode, LogLevel, StreamEvent,
    WireProbeInfo,
};

use super::segment_fetch::SegmentFetch;
use super::ManifestPoller;

impl ManifestPoller {
    pub(super) fn send_event(&self, event: StreamEvent) {
        if let Some(m) = &self.metrics {
            if let Ok(mut snap) = m.write() {
                update_metrics(&mut snap, &event);
                if let Some(otel) = &self.otel {
                    otel.record_metrics_snapshot(&snap);
                }
            }
        }
        if let Some((reg, id)) = &self.agent_metrics {
            if let Ok(mut guard) = reg.write() {
                if let Some(snap) = guard.streams.get_mut(id) {
                    update_metrics(snap, &event);
                }
            }
        }
        // Bounded: drop when UI/webhook cannot keep up (prefer liveness over backlog).
        let dropped = self.tx.try_send(event.clone()).is_err()
            || self
                .hook_tx
                .as_ref()
                .is_some_and(|h| h.try_send(event).is_err());
        if dropped {
            record_channel_drop();
            if let Some((reg, _)) = &self.agent_metrics {
                if let Ok(mut guard) = reg.write() {
                    guard.dropped_events = guard.dropped_events.saturating_add(1);
                }
            }
        }
    }
    pub(super) fn emit_g2g(
        &self,
        wire: &WireProbeInfo,
        pdt: Option<chrono::DateTime<Utc>>,
        dash_avail_ms: Option<i64>,
        ttfb_ms: u64,
    ) {
        let g2g = compute_g2g(
            wire.timing.prft_ntp_unix_ms,
            pdt.as_ref(),
            dash_avail_ms,
            Some(ttfb_ms),
            wall_now_unix_ms(),
        );
        if !g2g.is_empty() {
            if let Some(otel) = &self.otel {
                otel.record_g2g(&g2g);
            }
            self.send_event(StreamEvent::G2g(g2g));
        }
    }
    pub(super) fn record_segment_otel(&self, fetch: &SegmentFetch) {
        if let Some(exporter) = &self.otel {
            exporter.record_network("http.ttfb", &fetch.network, &fetch.segment_url);
            exporter.record_segment_download(
                &fetch.segment_url,
                &fetch.network,
                fetch.download_ms,
                fetch.http_status,
                fetch.chunked_transfer,
            );
        }
    }
    pub(super) fn emit_transport_failure(&self, err: &reqwest::Error) {
        if err.is_redirect() {
            let reason = if err.to_string().contains("cycle") {
                RedirectLoopError::CycleDetected
            } else {
                RedirectLoopError::LimitExceeded
            };
            self.send_event(StreamEvent::Finding(DiagnosticFinding::with_reason_code(
                DiagCategory::Rfc,
                DiagSeverity::Error,
                "redirect_loop",
                reason.message(),
                RedirectLoopError::reason_code(),
            )));
        }
        let io_error = err.is_connect() || err.is_request() || err.is_timeout();
        let hints = classify_transport_failure(true, 0, io_error);
        if hints.tcp_reset_suspected {
            self.send_event(StreamEvent::Finding(DiagnosticFinding::with_reason_code(
                DiagCategory::Rfc,
                DiagSeverity::Error,
                "tcp_io_reset",
                "TCP I/O drop after connect (passive heuristic)",
                DiagnosticReasonCode::ErrTcpIoReset,
            )));
        }
    }

    pub(super) fn emit_log(
        &self,
        level: LogLevel,
        category: DiagCategory,
        message: impl Into<String>,
    ) {
        self.send_event(StreamEvent::Log {
            level,
            category,
            message: message.into(),
        });
    }

    pub(super) fn flush_findings(&self, linter: &mut SpecLinter) {
        for finding in linter.take_findings() {
            let level = match finding.severity {
                DiagSeverity::Info => LogLevel::Info,
                DiagSeverity::Warn => LogLevel::Warn,
                DiagSeverity::Error => LogLevel::Error,
            };
            let log_msg = if let Some(reason) = &finding.reason {
                format!("[{reason}] {}", finding.message)
            } else {
                finding.message.clone()
            };
            self.emit_log(level, finding.category, log_msg);
            self.send_event(StreamEvent::Finding(finding));
        }
    }
}