|
|
|
@@ -69,6 +69,7 @@ type EngineStats struct { |
|
|
|
MaxWriteMs float64 `json:"maxWriteMs,omitempty"` |
|
|
|
Queue output.QueueStats `json:"queue"` |
|
|
|
RuntimeIndicator RuntimeIndicator `json:"runtimeIndicator"` |
|
|
|
RuntimeAlert string `json:"runtimeAlert,omitempty"` |
|
|
|
} |
|
|
|
|
|
|
|
type RuntimeIndicator string |
|
|
|
@@ -350,6 +351,7 @@ func (e *Engine) Stats() EngineStats { |
|
|
|
|
|
|
|
queue := e.frameQueue.Stats() |
|
|
|
lateBuffers := e.lateBuffers.Load() |
|
|
|
ri := runtimeIndicator(queue.Health, lateBuffers) |
|
|
|
return EngineStats{ |
|
|
|
State: state.String(), |
|
|
|
ChunksProduced: e.chunksProduced.Load(), |
|
|
|
@@ -363,7 +365,8 @@ func (e *Engine) Stats() EngineStats { |
|
|
|
MaxUpsampleMs: durationMs(e.maxUpsampleNs.Load()), |
|
|
|
MaxWriteMs: durationMs(e.maxWriteNs.Load()), |
|
|
|
Queue: queue, |
|
|
|
RuntimeIndicator: runtimeIndicator(queue.Health, lateBuffers), |
|
|
|
RuntimeIndicator: ri, |
|
|
|
RuntimeAlert: runtimeAlert(queue.Health, lateBuffers), |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@@ -378,6 +381,19 @@ func runtimeIndicator(queueHealth output.QueueHealth, lateBuffers uint64) Runtim |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func runtimeAlert(queueHealth output.QueueHealth, lateBuffers uint64) string { |
|
|
|
switch { |
|
|
|
case queueHealth == output.QueueHealthCritical: |
|
|
|
return "queue health critical" |
|
|
|
case lateBuffers > 0: |
|
|
|
return "late buffers" |
|
|
|
case queueHealth == output.QueueHealthLow: |
|
|
|
return "queue health low" |
|
|
|
default: |
|
|
|
return "" |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
func (e *Engine) run(ctx context.Context) { |
|
|
|
e.wg.Add(1) |
|
|
|
go e.writerLoop(ctx) |
|
|
|
|