1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

✅ 1. RTP packet loss monitoring (Section 2.2)
   → Alert if loss > 0.01%
   
✅ 2. RTP jitter monitoring (Section 2.2)
   → Alert if jitter > 1ms
   
✅ 3. PTP offset monitoring (Section 2.6)
   → Alert if offset > 10μs
   
✅ 4. Basic Grafana dashboard (Section 5)
   → Visibility into streams

1
2
3
4
5
6
7
8

✅ 5. SMPTE 2022-7 health (Section 2.4)
   → Ensure redundancy works
   
✅ 6. Buffer level monitoring (Section 8.1)
   → Prevent frame drops
   
✅ 7. Alerting (Section 6)
   → Get notified before viewers complain

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

✅ 8. Audio monitoring (Section 2.3)
   → Sample rate, A/V sync
   
✅ 9. Ancillary data (Section 2.5)
   → Closed captions (FCC compliance!)
   
✅ 10. Network switches (Section 4.3)
   → gNMI for switch health
   
✅ 11. NMOS control plane (Section 10.1)
   → Monitor registry and connections

1
2
3
4
5

✅ 12. Security hardening (Section 8.1)
✅ 13. CI/CD pipeline (Section 11.9)
✅ 14. Synthetic monitoring (Section 11.10)
✅ 15. Log correlation (Section 11.11)
✅ 16. Scale strategies (Section 10.4)

1
2
3

RTP Loss Rate: 0.02% (2 packets per 10,000)
Visual Impact: Intermittent blocking artifacts
Business Impact: Viewer complaints, social media backlash

1
2
3

Camera 1 PTP Offset: +5μs (normal)
Camera 7 PTP Offset: +42,000μs (42ms drift!)
Result: Audio/video sync issues across camera switches

1
2
3

Expected Bandwidth: 2.5 Gbps (documented streams)
Actual Bandwidth: 8.7 Gbps (unknown multicast sources!)
Result: Network congestion, dropped frames, failed production

1
2
3
4
5
6
7
8

Scenario: Camera configured as "Narrow" but network has jitter

Expected: Constant packet arrival (easy for receiver buffer)
Reality: Packets arrive in bursts (buffer underruns!)

Result: Frame drops despite 0% packet loss!

Monitoring Need: Detect when stream class doesn't match network behavior

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# Capture single RTP packet for analysis
tcpdump -i eth0 -nn -X -c 1 'udp dst port 20000'

# Output:
# 12:34:56.789012 IP 10.1.1.100.50000 > 239.1.1.10.20000: UDP, length 1460
# 0x0000:  4500 05dc 1234 4000 4011 abcd 0a01 0164  E....4@.@......d
# 0x0010:  ef01 010a c350 4e20 05c8 5678 8060 006f  .....PN...Vx.`.o
#                                    ↑↑↑↑ ↑↑↑↑
#                                    RTP  Seq
#          ↑↑↑↑ ↑↑↑↑ ↑↑↑↑ ↑↑↑↑
#          Ver  DSCP  Total Len

# Parse with tshark for detailed RTP info
tshark -i eth0 -Y "rtp" -T fields \
    -e rtp.seq -e rtp.timestamp -e rtp.ssrc -e rtp.p_type

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

type RTPStreamMetrics struct {
    // Total packets expected based on sequence numbers
    PacketsExpected   uint64
    // Packets actually received
    PacketsReceived   uint64
    // Calculated loss
    PacketLoss        float64 // percentage
    // Loss by category
    SinglePacketLoss  uint64  // 1 packet lost
    BurstLoss         uint64  // 2+ consecutive lost
}

// Acceptable thresholds
const (
    ThresholdPacketLossWarning  = 0.001  // 0.001% = 1 in 100,000
    ThresholdPacketLossCritical = 0.01   // 0.01% = 1 in 10,000
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type JitterMetrics struct {
    // RFC 3550 jitter calculation
    InterarrivalJitter float64  // in microseconds
    // Max jitter observed
    MaxJitter          float64
    // Jitter histogram (distribution)
    JitterHistogram    map[int]int  // bucket -> count
}

// Thresholds for 1080p60
const (
    ThresholdJitterWarning  = 500   // 500μs
    ThresholdJitterCritical = 1000  // 1ms
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

type ArrivalMetrics struct {
    // Packets per second (should match stream spec)
    PacketsPerSecond  float64
    // Expected rate (from SDP)
    ExpectedPPS       float64
    // Deviation
    RateDeviation     float64  // percentage
}

// Example: 1080p60 4:2:2 10-bit
const (
    Expected1080p60PPS = 90000  // ~90K packets/second
)

1
2
3
4
5
6
7
8

type TimestampMetrics struct {
    // Timestamp jumps (discontinuities)
    TimestampJumps    uint64
    // Clock rate (90kHz for video, 48kHz for audio)
    ClockRate         uint32
    // Timestamp drift vs PTP
    TimestampDrift    float64  // microseconds
}

 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

// ST 2110-22 specific metrics
type ST2110_22Metrics struct {
    // Transmission Mode
    TransmissionMode    string  // "gapped" (2110-20) or "linear" (2110-22)
    
    // TRS (Transmission Rate Scheduler) Compliance
    TRSCompliant        bool
    TRSViolations       uint64
    TRSMode             string  // "2110TPNL" for narrow linear
    
    // Drain Timing (critical for CBR)
    DrainPeriodNs       int64   // Expected drain period (e.g., 13468 ns for 1080p60)
    ActualDrainNs       int64   // Measured drain period
    DrainVarianceNs     int64   // Variance from expected (should be < 100ns)
    DrainJitter         float64 // Jitter in drain timing
    
    // N and Nmax (packets per line)
    PacketsPerLine      int     // Actual packets per video line
    MaxPacketsPerLine   int     // Maximum allowed (from SDP)
    NViolations         uint64  // Times N exceeded Nmax
    
    // TFSM (Time of First Scheduled Packet) for each frame
    TFSMOffset          int64   // Nanoseconds from frame boundary
    TFSMVariance        int64   // Should be constant
    
    // Read Point (Rₚ) tracking
    ReadPointOffset     int64   // Offset from PTP epoch
    ReadPointDrift      float64 // Drift over time
    
    // Packet gaps (should be uniform in linear mode)
    InterPacketGap      int64   // Nanoseconds between packets
    GapVariance         int64   // Should be minimal in linear mode
}

// Thresholds for ST 2110-22
const (
    MaxDrainVarianceNs  = 100    // 100ns max variance
    MaxTFSMVarianceNs   = 50     // 50ns max TFSM variance
    MaxGapVarianceNs    = 200    // 200ns max inter-packet gap variance
)

  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
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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

// rtp/st2110_22.go
package rtp

import (
    "fmt"
    "math"
    "time"
)

type ST2110_22Analyzer struct {
    metrics          ST2110_22Metrics
    
    // State for drain timing
    lastPacketTime   time.Time
    lastFrameStart   time.Time
    packetsThisFrame int
    
    // Expected values (from SDP)
    expectedDrainNs  int64
    expectedNmax     int
    
    // Running statistics
    drainSamples     []int64
    gapSamples       []int64
}

func NewST2110_22Analyzer(width, height int, fps float64) *ST2110_22Analyzer {
    // Calculate expected drain for linear mode
    // For 1080p60: drain = 1/60 / 1125 lines ≈ 13468 ns per line
    frameTimeNs := int64(1e9 / fps)
    totalLines := height + (height / 10) // Active + blanking
    drainPeriodNs := frameTimeNs / int64(totalLines)
    
    return &ST2110_22Analyzer{
        expectedDrainNs: drainPeriodNs,
        metrics: ST2110_22Metrics{
            TransmissionMode: "linear",
            TRSMode:         "2110TPNL",
        },
    }
}

func (a *ST2110_22Analyzer) AnalyzePacket(packet *RTPPacket, arrivalTime time.Time) {
    now := arrivalTime
    
    // Check if new frame (marker bit or timestamp wrap)
    if packet.Marker || a.isNewFrame(packet) {
        // Validate previous frame
        a.validateFrame()
        
        // Reset for new frame
        a.lastFrameStart = now
        a.packetsThisFrame = 0
    }
    
    a.packetsThisFrame++
    
    // Measure inter-packet gap (should be uniform in linear mode)
    if !a.lastPacketTime.IsZero() {
        gap := now.Sub(a.lastPacketTime).Nanoseconds()
        a.gapSamples = append(a.gapSamples, gap)
        a.metrics.InterPacketGap = gap
        
        // Calculate gap variance
        if len(a.gapSamples) > 100 {
            a.metrics.GapVariance = a.calculateVariance(a.gapSamples)
            
            // Alert if variance too high (non-linear transmission!)
            if a.metrics.GapVariance > MaxGapVarianceNs {
                fmt.Printf("WARNING: High gap variance %dns (expected linear mode)\n", 
                    a.metrics.GapVariance)
            }
            
            // Keep only recent samples
            a.gapSamples = a.gapSamples[len(a.gapSamples)-100:]
        }
    }
    a.lastPacketTime = now
    
    // Extract TFSM (Time of First Scheduled Packet) from RTP extension
    if tfsm := a.extractTFSM(packet); tfsm != 0 {
        a.metrics.TFSMOffset = tfsm
        
        // Validate TFSM is consistent across frames
        // (should be same offset from frame boundary)
        if a.metrics.TFSMVariance > MaxTFSMVarianceNs {
            a.metrics.TRSViolations++
            a.metrics.TRSCompliant = false
        }
    }
}

func (a *ST2110_22Analyzer) validateFrame() {
    if a.packetsThisFrame == 0 {
        return
    }
    
    // Calculate actual drain period
    frameDuration := time.Since(a.lastFrameStart).Nanoseconds()
    actualDrain := frameDuration / int64(a.packetsThisFrame)
    
    a.metrics.ActualDrainNs = actualDrain
    a.drainSamples = append(a.drainSamples, actualDrain)
    
    // Calculate drain variance
    if len(a.drainSamples) > 100 {
        variance := a.calculateVariance(a.drainSamples)
        a.metrics.DrainVarianceNs = variance
        
        // Check TRS compliance (drain must be constant within tolerance)
        if variance > MaxDrainVarianceNs {
            a.metrics.TRSViolations++
            a.metrics.TRSCompliant = false
            
            fmt.Printf("TRS VIOLATION: Drain variance %dns (max: %dns)\n",
                variance, MaxDrainVarianceNs)
        } else {
            a.metrics.TRSCompliant = true
        }
        
        // Keep only recent samples
        a.drainSamples = a.drainSamples[len(a.drainSamples)-100:]
    }
    
    // Validate N (packets per line) doesn't exceed Nmax
    a.metrics.PacketsPerLine = a.packetsThisFrame
    if a.expectedNmax > 0 && a.packetsThisFrame > a.expectedNmax {
        a.metrics.NViolations++
        fmt.Printf("N VIOLATION: %d packets (Nmax: %d)\n",
            a.packetsThisFrame, a.expectedNmax)
    }
}

func (a *ST2110_22Analyzer) calculateVariance(samples []int64) int64 {
    if len(samples) == 0 {
        return 0
    }
    
    // Calculate mean
    var sum int64
    for _, v := range samples {
        sum += v
    }
    mean := float64(sum) / float64(len(samples))
    
    // Calculate variance
    var variance float64
    for _, v := range samples {
        diff := float64(v) - mean
        variance += diff * diff
    }
    variance /= float64(len(samples))
    
    return int64(math.Sqrt(variance))
}

func (a *ST2110_22Analyzer) extractTFSM(packet *RTPPacket) int64 {
    // Parse RTP header extension for TFSM (if present)
    // ST 2110-22 uses RTP extension ID 1 for timing info
    // Implementation depends on actual packet structure
    return 0  // Placeholder
}

func (a *ST2110_22Analyzer) isNewFrame(packet *RTPPacket) bool {
    // Detect frame boundaries (timestamp increment)
    // For 1080p60: timestamp increments by 1500 (90kHz / 60fps)
    return false  // Placeholder
}

// Prometheus metrics for ST 2110-22
func (e *ST2110Exporter) registerST2110_22Metrics() {
    e.trsCompliant = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "st2110_22_trs_compliant",
            Help: "TRS compliance status (1=compliant, 0=violation)",
        },
        []string{"stream_id"},
    )
    
    e.drainVariance = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "st2110_22_drain_variance_nanoseconds",
            Help: "Drain timing variance in nanoseconds",
        },
        []string{"stream_id"},
    )
    
    e.trsViolations = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "st2110_22_trs_violations_total",
            Help: "Total TRS compliance violations",
        },
        []string{"stream_id"},
    )
    
    prometheus.MustRegister(e.trsCompliant, e.drainVariance, e.trsViolations)
}

 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

# alerts/st2110_22.yml
groups:
  - name: st2110_22_cbr
    interval: 1s
    rules:
      # TRS compliance violation
      - alert: ST2110_22_TRSViolation
        expr: st2110_22_trs_compliant == 0
        for: 5s
        labels:
          severity: critical
        annotations:
          summary: "TRS compliance violation on {{ $labels.stream_id }}"
          description: "Stream not maintaining constant bitrate transmission"
      
      # Excessive drain variance
      - alert: ST2110_22_HighDrainVariance
        expr: st2110_22_drain_variance_nanoseconds > 100
        for: 10s
        labels:
          severity: warning
        annotations:
          summary: "High drain variance on {{ $labels.stream_id }}"
          description: "Drain variance {{ $value }}ns (max: 100ns)"
      
      # N exceeded Nmax
      - alert: ST2110_22_NExceeded
        expr: increase(st2110_22_n_violations_total[1m]) > 0
        labels:
          severity: critical
        annotations:
          summary: "Packets per line exceeded Nmax on {{ $labels.stream_id }}"
          description: "ST 2110-22 N constraint violated"

 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

// Audio-specific metrics
type AudioStreamMetrics struct {
    // Sample rate validation
    SampleRate          int     // 48000, 96000, etc.
    ActualSampleRate    float64 // Measured (should match declared)
    SampleRateDrift     float64 // ppm (parts per million)
    
    // Channel mapping
    DeclaredChannels    int     // From SDP
    ActualChannels      int     // Detected in stream
    ChannelMappingOK    bool
    
    // Audio-specific timing
    PacketsPerSecond    float64 // e.g., 1000 for 1ms packets
    SamplesPerPacket    int     // e.g., 48 samples for 48kHz/1ms
    
    // A/V Sync (relative to video stream)
    VideoStreamID       string
    AudioDelayMs        float64 // Audio ahead (+) or behind (-) video
    LipSyncError        bool    // > 40ms is noticeable
    
    // AES67 compliance
    AES67Compliant      bool
    AES67Profile        string  // "High", "Medium", "Low"
    
    // Audio quality indicators
    SilenceDetected     bool
    ClippingDetected    bool    // Audio > 0dBFS
    PhaseIssues         bool    // L/R channel phase problems
    
    // ST 2110-31 (HDR audio) specific
    BitDepth            int     // 16, 24, 32 bit
    DynamicRange        float64 // dB
}

// Thresholds
const (
    MaxSampleRateDriftPPM  = 10     // 10 ppm max drift
    MaxAudioDelayMs        = 40     // 40ms lip sync tolerance
    MaxSilenceDurationMs   = 5000   // 5 seconds of silence = alert
)

  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
140
141
142
143
144
145
146
147
148
149
150
151

// audio/analyzer.go
package audio

import (
    "fmt"
    "math"
    "time"
)

type AudioAnalyzer struct {
    metrics        AudioStreamMetrics
    videoAnalyzer  *VideoAnalyzer  // For A/V sync calculation
    
    // Sample rate measurement
    lastTimestamp  uint32
    lastPacketTime time.Time
    sampleCount    uint64
    
    // Silence detection
    silenceStart   time.Time
    isSilent       bool
    
    // Channel validation
    channelData    [][]int16  // Per-channel samples
}

func NewAudioAnalyzer(sampleRate, channels int) *AudioAnalyzer {
    return &AudioAnalyzer{
        metrics: AudioStreamMetrics{
            SampleRate:       sampleRate,
            DeclaredChannels: channels,
        },
        channelData: make([][]int16, channels),
    }
}

func (a *AudioAnalyzer) AnalyzePacket(packet *RTPPacket, arrivalTime time.Time) {
    // Extract audio samples from RTP payload
    samples := a.extractSamples(packet)
    
    // Measure actual sample rate
    if a.lastTimestamp != 0 {
        timestampDiff := packet.Timestamp - a.lastTimestamp
        timeDiff := arrivalTime.Sub(a.lastPacketTime).Seconds()
        
        if timeDiff > 0 {
            actualRate := float64(timestampDiff) / timeDiff
            a.metrics.ActualSampleRate = actualRate
            
            // Calculate drift in ppm
            expectedRate := float64(a.metrics.SampleRate)
            drift := (actualRate - expectedRate) / expectedRate * 1e6
            a.metrics.SampleRateDrift = drift
            
            if math.Abs(drift) > MaxSampleRateDriftPPM {
                fmt.Printf("AUDIO DRIFT: %.2f ppm (max: %d)\n", drift, MaxSampleRateDriftPPM)
            }
        }
    }
    
    a.lastTimestamp = packet.Timestamp
    a.lastPacketTime = arrivalTime
    a.sampleCount += uint64(len(samples))
    
    // Detect silence (all samples near zero)
    if a.isSilenceFrame(samples) {
        if !a.isSilent {
            a.silenceStart = arrivalTime
            a.isSilent = true
        }
        
        silenceDuration := arrivalTime.Sub(a.silenceStart)
        if silenceDuration.Milliseconds() > MaxSilenceDurationMs {
            a.metrics.SilenceDetected = true
            fmt.Printf("SILENCE DETECTED: %dms\n", silenceDuration.Milliseconds())
        }
    } else {
        a.isSilent = false
        a.metrics.SilenceDetected = false
    }
    
    // Detect clipping (samples at max/min values)
    if a.detectClipping(samples) {
        a.metrics.ClippingDetected = true
    }
    
    // Validate channel count
    channels := len(samples) / a.metrics.SamplesPerPacket
    if channels != a.metrics.DeclaredChannels {
        a.metrics.ChannelMappingOK = false
        fmt.Printf("CHANNEL MISMATCH: Expected %d, got %d\n",
            a.metrics.DeclaredChannels, channels)
    }
}

// Calculate A/V sync offset
func (a *AudioAnalyzer) CalculateAVSync() {
    if a.videoAnalyzer == nil {
        return
    }
    
    // Get audio timestamp (in samples)
    audioTimestampNs := int64(a.lastTimestamp) * 1e9 / int64(a.metrics.SampleRate)
    
    // Get video timestamp (in 90kHz units)
    videoTimestampNs := int64(a.videoAnalyzer.lastTimestamp) * 1e9 / 90000
    
    // Calculate offset
    offsetNs := audioTimestampNs - videoTimestampNs
    a.metrics.AudioDelayMs = float64(offsetNs) / 1e6
    
    // Check lip sync error
    if math.Abs(a.metrics.AudioDelayMs) > MaxAudioDelayMs {
        a.metrics.LipSyncError = true
        fmt.Printf("LIP SYNC ERROR: Audio %+.1fms (max: ±%dms)\n",
            a.metrics.AudioDelayMs, MaxAudioDelayMs)
    } else {
        a.metrics.LipSyncError = false
    }
}

func (a *AudioAnalyzer) isSilenceFrame(samples []int16) bool {
    // Check if all samples are below threshold (e.g., -60dBFS)
    threshold := int16(32)  // Very quiet
    
    for _, sample := range samples {
        if sample > threshold || sample < -threshold {
            return false
        }
    }
    return true
}

func (a *AudioAnalyzer) detectClipping(samples []int16) bool {
    // Check if any samples are at max/min (clipping)
    maxVal := int16(32767)
    minVal := int16(-32768)
    
    for _, sample := range samples {
        if sample >= maxVal-10 || sample <= minVal+10 {
            return true
        }
    }
    return false
}

func (a *AudioAnalyzer) extractSamples(packet *RTPPacket) []int16 {
    // Parse L16, L24, or L32 audio from RTP payload
    // Implementation depends on bit depth
    return nil  // Placeholder
}

 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

# alerts/audio.yml
groups:
  - name: st2110_audio
    interval: 1s
    rules:
      # Sample rate drift
      - alert: ST2110AudioSampleRateDrift
        expr: abs(st2110_audio_sample_rate_drift_ppm) > 10
        for: 5s
        labels:
          severity: critical
        annotations:
          summary: "Audio sample rate drift on {{ $labels.stream_id }}"
          description: "Drift: {{ $value }} ppm (max: 10 ppm)"
      
      # Lip sync error
      - alert: ST2110LipSyncError
        expr: abs(st2110_audio_delay_milliseconds) > 40
        for: 10s
        labels:
          severity: critical
        annotations:
          summary: "Lip sync error on {{ $labels.stream_id }}"
          description: "Audio offset: {{ $value }}ms (max: ±40ms)"
      
      # Prolonged silence
      - alert: ST2110AudioSilence
        expr: st2110_audio_silence_detected == 1
        for: 5s
        labels:
          severity: warning
        annotations:
          summary: "Prolonged silence on {{ $labels.stream_id }}"
          description: "No audio signal detected for > 5 seconds"
      
      # Audio clipping
      - alert: ST2110AudioClipping
        expr: st2110_audio_clipping_detected == 1
        for: 1s
        labels:
          severity: warning
        annotations:
          summary: "Audio clipping on {{ $labels.stream_id }}"
          description: "Audio levels exceeding 0dBFS (distortion)"
      
      # Channel mapping error
      - alert: ST2110AudioChannelMismatch
        expr: st2110_audio_channel_mapping_ok == 0
        for: 5s
        labels:
          severity: critical
        annotations:
          summary: "Audio channel mismatch on {{ $labels.stream_id }}"
          description: "Declared vs actual channel count mismatch"

1
2
3
4
5
6
7
8
9

⚠️  CRITICAL: Path Diversity < 50%
Shared Hops: core-switch-1.local
Risk: Single point of failure detected!

Action Required:
1. Reconfigure backup path via Core B
2. Verify with traceroute:
   Main:   hop1→CoreA→hop3
   Backup: hop1→CoreB→hop3

 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

// Validate true path diversity for SMPTE 2022-7
func (a *ST2022_7Analyzer) ValidatePathDiversity() error {
    // Traceroute both streams
    mainPath := traceroute(a.mainStreamIP)
    backupPath := traceroute(a.backupStreamIP)
    
    // Find shared hops
    sharedHops := []string{}
    for _, hop := range mainPath {
        if contains(backupPath, hop) {
            sharedHops = append(sharedHops, hop)
        }
    }
    
    // Calculate diversity percentage
    totalHops := len(mainPath) + len(backupPath)
    uniqueHops := totalHops - (2 * len(sharedHops))
    diversity := float64(uniqueHops) / float64(totalHops)
    
    a.metrics.PathDiversity = diversity
    a.metrics.SharedHops = sharedHops
    
    // Alert if diversity too low
    if diversity < MinPathDiversity {
        return fmt.Errorf(
            "CRITICAL: Path diversity %.1f%% < %.1f%%. Shared hops: %v",
            diversity*100, MinPathDiversity*100, sharedHops,
        )
    }
    
    return nil
}

 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

// ST 2022-7 detailed metrics
type ST2022_7Metrics struct {
    // Stream status
    MainStreamActive       bool
    BackupStreamActive     bool
    BothStreamsHealthy     bool
    
    // Per-stream health
    MainPacketLoss         float64
    BackupPacketLoss       float64
    MainJitter             float64
    BackupJitter           float64
    MainLastSeenMs         int64   // Milliseconds since last packet
    BackupLastSeenMs       int64
    
    // Protection switching
    CurrentActiveStream    string  // "main", "backup", "both"
    SwitchingMode          string  // "seamless" or "manual"
    LastSwitchTime         time.Time
    SwitchingEvents        uint64
    
    // Seamless switching performance
    LastSwitchDuration     time.Duration  // How long switch took
    PacketsLostDuringSwitch uint64        // Should be ZERO for seamless
    
    // Path diversity validation
    MainNetworkPath        []string  // IP addresses in path (traceroute)
    BackupNetworkPath      []string
    PathDiversity          float64   // Percentage of different hops
    SharedHops             []string  // Common points of failure
    
    // Timing alignment
    StreamTimingOffset     int64     // Nanoseconds between main/backup
    TimingWithinTolerance  bool      // < 1ms offset required
    
    // Packet merger stats
    DuplicatePacketsRx     uint64    // Both streams received same packet
    UniqueFromMain         uint64    // Only main had packet
    UniqueFromBackup       uint64    // Only backup had packet (switch events)
    MergerBufferUsage      float64   // Percentage of merger buffer used
}

// Thresholds
const (
    MaxStreamTimingOffsetMs = 1     // 1ms max between streams
    MaxSwitchDurationMs     = 100   // 100ms max switch time
    MinPathDiversity        = 0.5   // 50% different paths minimum
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

Scenario: Main stream fails due to switch reboot

Without 2022-7:
  T+0ms:    Main stream stops
  T+500ms:  Operator notices
  T+30s:    Manual failover initiated
  T+35s:    Backup stream live
  Result: 35 seconds of BLACK on air ($$$$$)

With 2022-7 (Seamless):
  T+0ms:    Main stream stops
  T+1ms:    Receiver automatically switches to backup
  T+2ms:    Backup stream outputting
  Result: 2ms glitch (invisible to viewers) ✅

  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
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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233

// protection/st2022_7.go
package protection

import (
    "fmt"
    "time"
)

type ST2022_7Analyzer struct {
    metrics ST2022_7Metrics
    
    // Packet merger state
    seenPackets      map[uint16]packetInfo  // Sequence -> info
    mergerBuffer     []MergedPacket
    mergerBufferSize int
    
    // Stream health tracking
    mainHealthCheck  time.Time
    backupHealthCheck time.Time
}

type packetInfo struct {
    source      string  // "main" or "backup"
    timestamp   time.Time
    delivered   bool
}

type MergedPacket struct {
    seqNumber   uint16
    fromMain    bool
    fromBackup  bool
    delivered   string  // Which stream was used
    arrivalDiff time.Duration  // Time difference between streams
}

func NewST2022_7Analyzer(bufferSize int) *ST2022_7Analyzer {
    return &ST2022_7Analyzer{
        seenPackets:      make(map[uint16]packetInfo),
        mergerBuffer:     make([]MergedPacket, 0, bufferSize),
        mergerBufferSize: bufferSize,
        metrics: ST2022_7Metrics{
            SwitchingMode: "seamless",
        },
    }
}

func (a *ST2022_7Analyzer) ProcessPacket(packet *RTPPacket, source string, arrivalTime time.Time) *RTPPacket {
    seq := packet.SequenceNumber
    
    // Update stream health
    if source == "main" {
        a.mainHealthCheck = arrivalTime
        a.metrics.MainStreamActive = true
    } else {
        a.backupHealthCheck = arrivalTime
        a.metrics.BackupStreamActive = true
    }
    
    // Check if we've seen this packet before
    if existing, seen := a.seenPackets[seq]; seen {
        // Duplicate packet (both streams working)
        a.metrics.DuplicatePacketsRx++
        
        // Calculate timing offset between streams
        timeDiff := arrivalTime.Sub(existing.timestamp)
        a.metrics.StreamTimingOffset = timeDiff.Nanoseconds()
        
        if timeDiff.Milliseconds() > MaxStreamTimingOffsetMs {
            a.metrics.TimingWithinTolerance = false
            fmt.Printf("TIMING OFFSET: %dms between main/backup (max: %dms)\n",
                timeDiff.Milliseconds(), MaxStreamTimingOffsetMs)
        }
        
        // Already delivered, discard duplicate
        if existing.delivered {
            return nil
        }
        
        // Update merge record
        a.updateMergeRecord(seq, source, arrivalTime)
        
        return nil  // Discard duplicate
    }
    
    // New packet - record it
    a.seenPackets[seq] = packetInfo{
        source:    source,
        timestamp: arrivalTime,
        delivered: false,
    }
    
    // Update unique packet counters
    if source == "main" {
        a.metrics.UniqueFromMain++
    } else {
        a.metrics.UniqueFromBackup++
        
        // Packet only from backup = main stream had loss!
        // This is a switching event
        if !a.isMainHealthy() {
            a.handleSwitch(arrivalTime)
        }
    }
    
    // Deliver packet
    info := a.seenPackets[seq]
    info.delivered = true
    a.seenPackets[seq] = info
    
    // Update active stream
    a.updateActiveStream(source)
    
    // Clean old packets from map (keep only last 1000)
    if len(a.seenPackets) > 1000 {
        a.cleanOldPackets()
    }
    
    return packet
}

func (a *ST2022_7Analyzer) isMainHealthy() bool {
    // Main considered down if no packets in last 100ms
    return time.Since(a.mainHealthCheck).Milliseconds() < 100
}

func (a *ST2022_7Analyzer) isBackupHealthy() bool {
    return time.Since(a.backupHealthCheck).Milliseconds() < 100
}

func (a *ST2022_7Analyzer) handleSwitch(switchTime time.Time) {
    // Record switch event
    a.metrics.SwitchingEvents++
    
    // Calculate switch duration
    if !a.metrics.LastSwitchTime.IsZero() {
        duration := switchTime.Sub(a.metrics.LastSwitchTime)
        a.metrics.LastSwitchDuration = duration
        
        fmt.Printf("PROTECTION SWITCH: Main → Backup (duration: %dms)\n",
            duration.Milliseconds())
        
        if duration.Milliseconds() > MaxSwitchDurationMs {
            fmt.Printf("SLOW SWITCH: %dms (max: %dms)\n",
                duration.Milliseconds(), MaxSwitchDurationMs)
        }
    }
    
    a.metrics.LastSwitchTime = switchTime
    a.metrics.CurrentActiveStream = "backup"
}

func (a *ST2022_7Analyzer) updateActiveStream(source string) {
    mainHealthy := a.isMainHealthy()
    backupHealthy := a.isBackupHealthy()
    
    a.metrics.BothStreamsHealthy = mainHealthy && backupHealthy
    
    if mainHealthy && backupHealthy {
        a.metrics.CurrentActiveStream = "both"
    } else if mainHealthy {
        a.metrics.CurrentActiveStream = "main"
    } else if backupHealthy {
        a.metrics.CurrentActiveStream = "backup"
    } else {
        a.metrics.CurrentActiveStream = "none"  // Disaster!
    }
}

func (a *ST2022_7Analyzer) updateMergeRecord(seq uint16, source string, arrival time.Time) {
    // Find existing merge record
    for i := range a.mergerBuffer {
        if a.mergerBuffer[i].seqNumber == seq {
            if source == "backup" {
                a.mergerBuffer[i].fromBackup = true
            }
            return
        }
    }
}

func (a *ST2022_7Analyzer) cleanOldPackets() {
    // Remove packets older than 500 (keep recent window)
    minSeq := uint16(0)
    for seq := range a.seenPackets {
        if seq > minSeq {
            minSeq = seq
        }
    }
    
    cutoff := minSeq - 500
    for seq := range a.seenPackets {
        if seq < cutoff {
            delete(a.seenPackets, seq)
        }
    }
}

// Validate path diversity (must use different network paths!)
func (a *ST2022_7Analyzer) ValidatePathDiversity() {
    // This would use traceroute or similar to validate
    // main and backup streams take different physical paths
    
    mainPath := a.metrics.MainNetworkPath
    backupPath := a.metrics.BackupNetworkPath
    
    if len(mainPath) == 0 || len(backupPath) == 0 {
        return
    }
    
    // Count shared hops
    shared := 0
    a.metrics.SharedHops = []string{}
    
    for _, mainHop := range mainPath {
        for _, backupHop := range backupPath {
            if mainHop == backupHop {
                shared++
                a.metrics.SharedHops = append(a.metrics.SharedHops, mainHop)
            }
        }
    }
    
    // Calculate diversity percentage
    totalHops := len(mainPath) + len(backupPath)
    diversity := 1.0 - (float64(shared*2) / float64(totalHops))
    a.metrics.PathDiversity = diversity
    
    if diversity < MinPathDiversity {
        fmt.Printf("LOW PATH DIVERSITY: %.1f%% (min: %.1f%%)\n",
            diversity*100, MinPathDiversity*100)
        fmt.Printf("Shared hops: %v\n", a.metrics.SharedHops)
    }
}

 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

# alerts/st2022_7.yml
groups:
  - name: smpte_2022_7
    interval: 1s
    rules:
      # Both streams down = disaster
      - alert: ST2022_7_BothStreamsDown
        expr: st2110_st2022_7_both_streams_healthy == 0
        for: 1s
        labels:
          severity: critical
          page: "true"
        annotations:
          summary: "DISASTER: Both ST 2022-7 streams down on {{ $labels.stream_id }}"
          description: "Main AND backup streams offline - TOTAL FAILURE"
      
      # Backup stream down (no protection!)
      - alert: ST2022_7_BackupStreamDown
        expr: st2110_st2022_7_backup_stream_active == 0
        for: 30s
        labels:
          severity: warning
        annotations:
          summary: "ST 2022-7 backup stream down on {{ $labels.stream_id }}"
          description: "No protection available if main stream fails!"
      
      # Excessive switching (network instability)
      - alert: ST2022_7_ExcessiveSwitching
        expr: rate(st2110_st2022_7_switching_events[5m]) > 0.1
        labels:
          severity: warning
        annotations:
          summary: "Excessive ST 2022-7 switching on {{ $labels.stream_id }}"
          description: "{{ $value }} switches/sec - indicates network instability"
      
      # Slow switch (> 100ms)
      - alert: ST2022_7_SlowSwitch
        expr: st2110_st2022_7_last_switch_duration_ms > 100
        labels:
          severity: warning
        annotations:
          summary: "Slow ST 2022-7 switch on {{ $labels.stream_id }}"
          description: "Switch took {{ $value }}ms (max: 100ms)"
      
      # Low path diversity (single point of failure)
      - alert: ST2022_7_LowPathDiversity
        expr: st2110_st2022_7_path_diversity < 0.5
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "Low path diversity on {{ $labels.stream_id }}"
          description: "Only {{ $value | humanizePercentage }} path diversity"
      
      # Timing offset too high
      - alert: ST2022_7_TimingOffset
        expr: abs(st2110_st2022_7_stream_timing_offset_ms) > 1
        for: 10s
        labels:
          severity: warning
        annotations:
          summary: "High timing offset between ST 2022-7 streams"
          description: "Offset: {{ $value }}ms (max: 1ms)"

 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

// Ancillary data metrics
type AncillaryDataMetrics struct {
    // Data types present
    ClosedCaptionsPresent bool
    TimecodePresent       bool
    AFDPresent            bool    // Active Format Description
    VITCPresent           bool    // Vertical Interval Timecode
    
    // Closed captions (CEA-608/708)
    CCPacketsReceived     uint64
    CCPacketsLost         uint64
    CCLossRate            float64
    LastCCTimestamp       time.Time
    CCGaps                uint64   // Gaps > 1 second
    
    // Timecode tracking
    Timecode              string   // HH:MM:SS:FF
    TimecodeJumps         uint64   // Discontinuities
    TimecodeDropFrame     bool
    TimecodeFrameRate     float64
    
    // AFD (aspect ratio signaling)
    AFDCode               uint8    // 0-15
    AFDChanged            uint64   // How many times AFD changed
    
    // SCTE-104 (ad insertion triggers)
    SCTE104Present        bool
    AdInsertionTriggers   uint64
}

// Why ancillary monitoring matters
const (
    MaxCCGapMs = 1000  // 1 second without CC = compliance violation (FCC)
)

1
2
3
4
5
6
7

Incident: Lost closed captions for 2 minutes during live news

Root Cause: ST 2110-40 ancillary stream had 0.5% packet loss
Video/Audio: Perfect (0.001% loss)
Result: $50K FCC fine + viewer complaints

Lesson: Monitor ancillary data separately!

  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