From bb949d18b7b88efaa1cfcc216a474913bfb4e19a Mon Sep 17 00:00:00 2001 From: root Date: Fri, 11 Sep 2026 10:58:50 +0000 Subject: [PATCH] feat(admin): show each stream's producer, CPU and memory on the streams page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A new "Resources" column between Stream Info and Actions: which process is producing the channel (the fanout daemon's native remuxer, ffmpeg, or PHP for the LLOD segmenter / loopback relay), the CPU it is burning and the memory it holds. With the native remuxer now an option per stream, "what does this channel actually cost" and "which backend is it on" are the two questions the list could not answer. ProcessManager reads both from /proc/PID/stat (fields 14/15 and 24, with the page size derived rather than assumed — 64K pages are normal on arm64). CPU there is cumulative, so a percentage needs two readings: cron:streams samples each producer once a pass and folds cpu/mem/producer into the stream's progress_info, carrying the previous reading in the same JSON to subtract from. cpuPercent() returns null rather than a wild figure when the pair says nothing — no previous sample, same instant, or a counter that went backwards because the producer restarted. Only the node running a stream can read its own /proc, so the sampling happens there and reaches the panel in the row the cron already writes; MAIN just renders it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UZP7fc2Hz36wbPmuLd9F9o --- docs/en/guides/process-management.md | 22 +++ src/Cli/CronJobs/StreamsCronJob.php | 42 ++++++ src/Core/Localization/lang/bg.ini | 1 + src/Core/Localization/lang/de.ini | 1 + src/Core/Localization/lang/en.ini | 1 + src/Core/Localization/lang/es.ini | 1 + src/Core/Localization/lang/fr.ini | 1 + src/Core/Localization/lang/pt.ini | 1 + src/Core/Localization/lang/ru.ini | 1 + src/Core/Process/ProcessManager.php | 133 ++++++++++++++++++ .../Controllers/Admin/TableController.php | 21 ++- src/Public/Views/admin/streams.php | 41 ++++++ tests/Unit/ProcessResourceUsageTest.php | 61 ++++++++ 13 files changed, 326 insertions(+), 1 deletion(-) create mode 100644 tests/Unit/ProcessResourceUsageTest.php diff --git a/docs/en/guides/process-management.md b/docs/en/guides/process-management.md index 602c442b..cfe52487 100644 --- a/docs/en/guides/process-management.md +++ b/docs/en/guides/process-management.md @@ -67,6 +67,28 @@ around?". --- +## Per-stream CPU and memory + +```php +ProcessManager::resourceSample($pid): ?array // ['ticks' => CPU ticks so far, 'rss' => bytes, 'at' => microtime] +ProcessManager::cpuPercent(array $now, array $prev): ?float // percent of ONE core between two samples +ProcessManager::producerKind($pid): ?string // 'fanout' (xc_fanout remux) | 'ffmpeg' | 'php' +``` + +Read straight out of `/proc/PID/stat` (fields 14/15 for CPU, 24 for RSS; the page size is derived +from this process's own `statm` vs `status`, since 64K pages are normal on arm64). CPU in `/proc` +is cumulative, so a percentage needs **two** samples: `cpuPercent()` returns `null` when the pair +says nothing — no previous reading, two readings from the same instant, or a counter that went +backwards because the producer restarted under the same stream. + +`cron:streams` samples each running stream's producer once per pass and folds the result into that +stream's `progress_info` JSON (`cpu`, `mem`, `producer`, plus `cpu_t` / `cpu_at` carrying the +reading the next pass subtracts from). Only the node running a stream can read its own `/proc`, so +the sampling happens there and travels to the panel in the row the cron already writes; the admin +streams list renders it as the **Resources** column (producer badge, CPU %, RAM). + +--- + ## Process Termination ```php diff --git a/src/Cli/CronJobs/StreamsCronJob.php b/src/Cli/CronJobs/StreamsCronJob.php index 42734204..88a1b4d5 100644 --- a/src/Cli/CronJobs/StreamsCronJob.php +++ b/src/Cli/CronJobs/StreamsCronJob.php @@ -66,6 +66,47 @@ class StreamsCronJob implements CommandInterface { return (strpos($rExe, 'ffmpeg') === 0 && FanoutClient::daemonStreamMissing(intval($rStream['stream_id']))) ? 1 : 0; } + /** + * Fold the producer's CPU, memory and kind into the stream's progress JSON — + * what the admin streams list shows per stream. + * + * Only this node can read its own /proc, so the reading is taken here and + * travels to the panel in the row the cron already writes. CPU is a delta + * against the previous pass's reading (carried in the same JSON), so it is + * the average over the pass rather than ffmpeg's lifetime average, which for + * a channel running for days says nothing about now. + * + * @param string $rProgressJson This pass's progress report. + * @param string|null $rPreviousJson Last pass's, for the CPU delta. + * @param int $rPID The producer's pid. + * @return string The JSON to store. + */ + private static function withResourceUsage(string $rProgressJson, ?string $rPreviousJson, int $rPID): string { + $rProgress = json_decode($rProgressJson, true); + if (!is_array($rProgress)) { + $rProgress = array(); + } + $rSample = ProcessManager::resourceSample($rPID); + if ($rSample === null) { + unset($rProgress['cpu'], $rProgress['mem'], $rProgress['producer'], $rProgress['cpu_t'], $rProgress['cpu_at']); + return json_encode($rProgress); + } + + $rPrevious = json_decode((string) $rPreviousJson, true); + $rCPU = null; + if (is_array($rPrevious) && isset($rPrevious['cpu_t'], $rPrevious['cpu_at'])) { + $rCPU = ProcessManager::cpuPercent($rSample, array('ticks' => (int) $rPrevious['cpu_t'], 'at' => (float) $rPrevious['cpu_at'])); + } + + $rProgress['cpu'] = $rCPU; // null on the first pass and after a restart + $rProgress['mem'] = $rSample['rss']; + $rProgress['producer'] = ProcessManager::producerKind($rPID); + $rProgress['cpu_t'] = $rSample['ticks']; + $rProgress['cpu_at'] = round($rSample['at'], 3); + + return json_encode($rProgress); + } + private function loadCron(): void { $rRedis = SettingsManager::getBool('redis_handler'); global $db; @@ -250,6 +291,7 @@ class StreamsCronJob implements CommandInterface { } else { $rProgress = $rStream['progress_info']; } + $rProgress = self::withResourceUsage((string) $rProgress, $rStream['progress_info'], $rPID); // A supervised stream's codecs, resolution and bitrate come // from the daemon, measured off the bytes (reconcileSupervised // wrote them above); recomputing them here from a stream_info diff --git a/src/Core/Localization/lang/bg.ini b/src/Core/Localization/lang/bg.ini index 3604c0c4..c180c700 100644 --- a/src/Core/Localization/lang/bg.ini +++ b/src/Core/Localization/lang/bg.ini @@ -644,6 +644,7 @@ status = "Статус" stream = "Стрийм" stream_errors = "Грешки в стрийма" stream_info = "Информация за стрийма" +stream_usage = "Ресурси" stream_name = "Име на стрийма" stream_source = "Източник на стрийма" stream_success = "Стриймовете са импортирани и обработени." diff --git a/src/Core/Localization/lang/de.ini b/src/Core/Localization/lang/de.ini index 41807319..66a3d8ad 100644 --- a/src/Core/Localization/lang/de.ini +++ b/src/Core/Localization/lang/de.ini @@ -643,6 +643,7 @@ status = "Status" stream = "Stream" stream_errors = "Stream Errors" stream_info = "Stream Info" +stream_usage = "Ressourcen" stream_name = "Stream Name" stream_source = "Stream Source" stream_success = "Streams have been imported and processed." diff --git a/src/Core/Localization/lang/en.ini b/src/Core/Localization/lang/en.ini index 5d2dd5b4..9703bfa2 100644 --- a/src/Core/Localization/lang/en.ini +++ b/src/Core/Localization/lang/en.ini @@ -702,6 +702,7 @@ status = "Status" stream = "Stream" stream_errors = "Stream Errors" stream_info = "Stream Info" +stream_usage = "Resources" stream_name = "Stream Name" stream_source = "Stream Source" stream_success = "Streams have been imported and processed." diff --git a/src/Core/Localization/lang/es.ini b/src/Core/Localization/lang/es.ini index d1cfca9b..e530db7a 100644 --- a/src/Core/Localization/lang/es.ini +++ b/src/Core/Localization/lang/es.ini @@ -643,6 +643,7 @@ status = "Estado" stream = "Arroyo" stream_errors = "Errores de transmisión" stream_info = "Información de transmisión" +stream_usage = "Recursos" stream_name = "Nombre de la transmisión" stream_source = "Fuente de transmisión" stream_success = "Los canales han sido importados y procesados." diff --git a/src/Core/Localization/lang/fr.ini b/src/Core/Localization/lang/fr.ini index 5d4c2721..9b9ea9f8 100644 --- a/src/Core/Localization/lang/fr.ini +++ b/src/Core/Localization/lang/fr.ini @@ -643,6 +643,7 @@ status = "Statut" stream = "Flux" stream_errors = "Erreurs de flux" stream_info = "Informations sur le flux" +stream_usage = "Ressources" stream_name = "Nom du flux" stream_source = "Source du flux" stream_success = "Streams have been imported and processed." diff --git a/src/Core/Localization/lang/pt.ini b/src/Core/Localization/lang/pt.ini index 0b0dd63a..5920d951 100644 --- a/src/Core/Localization/lang/pt.ini +++ b/src/Core/Localization/lang/pt.ini @@ -643,6 +643,7 @@ status = "Status" stream = "Fluxo" stream_errors = "Erros de fluxo" stream_info = "Informações da transmissão" +stream_usage = "Recursos" stream_name = "Nome do fluxo" stream_source = "Fonte de transmissão" stream_success = "Os fluxos foram importados e processados." diff --git a/src/Core/Localization/lang/ru.ini b/src/Core/Localization/lang/ru.ini index 7dbc968d..569209f6 100644 --- a/src/Core/Localization/lang/ru.ini +++ b/src/Core/Localization/lang/ru.ini @@ -643,6 +643,7 @@ status = "Статус" stream = "Транслировать" stream_errors = "Ошибки потока" stream_info = "Информация о потоке" +stream_usage = "Ресурсы" stream_name = "Имя потока" stream_source = "Источник потока" stream_success = "Потоки были импортированы и обработаны." diff --git a/src/Core/Process/ProcessManager.php b/src/Core/Process/ProcessManager.php index f9541df1..6c894b5e 100644 --- a/src/Core/Process/ProcessManager.php +++ b/src/Core/Process/ProcessManager.php @@ -173,6 +173,139 @@ class ProcessManager { return false; } + /** + * What kind of producer a stream's pid is, for the panel's stream list. + * + * @param int $pid Process ID + * @return string|null 'fanout' (xc_fanout remux), 'ffmpeg', 'php' (the LLOD + * segmenter / loopback relay), or null when it cannot be read. + */ + public static function producerKind($pid) { + $pid = (int)$pid; + + if ($pid <= 0 || !self::procExists($pid) || !is_readable('/proc/' . $pid . '/exe')) { + return null; + } + + $exe = @basename(@readlink('/proc/' . $pid . '/exe')); + + if (strpos($exe, 'xc_fanout') === 0) { + $cmdline = (string) @file_get_contents('/proc/' . $pid . '/cmdline'); + return strpos($cmdline, "\0remux\0") !== false ? 'fanout' : null; + } + if (strpos($exe, 'ffmpeg') === 0) { + return 'ffmpeg'; + } + if (strpos($exe, 'php') === 0) { + return 'php'; + } + + return null; + } + + /** + * One CPU/memory reading for a process, from /proc/PID/stat. + * + * CPU is cumulative (the ticks the process has burned since it started), so a + * percentage needs two readings — see cpuPercent(). Memory is resident set + * size in bytes, which is the figure that matters for a box running hundreds + * of encoders: what they actually hold in RAM. + * + * @param int $pid Process ID + * @return array{ticks:int,rss:int,at:float}|null Null when the process is gone. + */ + public static function resourceSample($pid) { + $pid = (int)$pid; + + if ($pid <= 1) { + return null; + } + + $stat = @file_get_contents('/proc/' . $pid . '/stat'); + if (!is_string($stat) || $stat === '') { + return null; + } + + // Field 2 (comm) is parenthesised and may itself contain spaces and + // brackets — "(ffmpeg (x))" is a legal name — so everything before the + // LAST ')' is skipped and the split starts at field 3 (state). + $rClose = strrpos($stat, ')'); + if ($rClose === false) { + return null; + } + $rFields = preg_split('/\s+/', trim(substr($stat, $rClose + 1))); + if (!is_array($rFields) || count($rFields) < 22) { + return null; + } + + // $rFields[$i] is /proc/PID/stat field $i + 3: utime 14, stime 15, rss 24. + return [ + 'ticks' => (int) $rFields[11] + (int) $rFields[12], + 'rss' => (int) $rFields[21] * self::pageSize(), + 'at' => microtime(true), + ]; + } + + /** + * CPU use between two resourceSample() readings, in percent of one core. + * + * @param array $now The newer sample. + * @param array $prev The older one (from the previous pass). + * @return float|null Null when the pair says nothing: no previous reading, a + * restarted process (its counter went backwards), or two + * readings from the same instant. + */ + public static function cpuPercent(array $now, array $prev) { + if (!isset($now['ticks'], $now['at'], $prev['ticks'], $prev['at'])) { + return null; + } + + $rSeconds = (float) $now['at'] - (float) $prev['at']; + $rTicks = (int) $now['ticks'] - (int) $prev['ticks']; + if ($rSeconds <= 0 || $rTicks < 0) { + return null; + } + + // /proc reports CPU time in USER_HZ, which is 100 on Linux whatever the + // kernel's own tick rate — it is part of the /proc ABI, not CONFIG_HZ. + return round(($rTicks / 100) / $rSeconds * 100, 1); + } + + /** + * The kernel page size, in bytes — /proc/PID/stat counts RSS in pages. + * + * Derived from this process's own two views of its memory (statm in pages, + * status in kB) rather than assumed, because 64K pages are normal on arm64. + * + * @return int + */ + protected static function pageSize() { + static $rSize = null; + if ($rSize !== null) { + return $rSize; + } + + $rSize = 4096; + $rStatm = @file_get_contents('/proc/self/statm'); + $rStatus = @file_get_contents('/proc/self/status'); + if (is_string($rStatm) && is_string($rStatus) && preg_match('/^VmRSS:\s+(\d+) kB/m', $rStatus, $rMatch)) { + $rPages = (int) (preg_split('/\s+/', trim($rStatm))[1] ?? 0); + if ($rPages > 0) { + // The two files are read a moment apart, so snap the ratio to a + // real page size instead of trusting it to the byte. + $rRatio = ((int) $rMatch[1] * 1024) / $rPages; + foreach ([4096, 8192, 16384, 65536] as $rCandidate) { + if ($rRatio >= $rCandidate * 0.75 && $rRatio <= $rCandidate * 1.25) { + $rSize = $rCandidate; + break; + } + } + } + } + + return $rSize; + } + // ─────────────────────────────────────────────────────────── // Process Control // ─────────────────────────────────────────────────────────── diff --git a/src/Public/Controllers/Admin/TableController.php b/src/Public/Controllers/Admin/TableController.php index 6f5572ba..7c255ff2 100644 --- a/src/Public/Controllers/Admin/TableController.php +++ b/src/Public/Controllers/Admin/TableController.php @@ -769,7 +769,10 @@ class TableController extends BaseAdminController { } $rCategories = CategoryService::getAllByType("live"); // Leading false, false = Responsive control + bulk-select checkbox columns (Bootstrap 5). - $rOrder = [false, false, "`streams`.`id`", "`streams`.`stream_icon`", "`streams`.`stream_display_name`", "`streams_servers`.`current_source`", "`clients`", "`streams_servers`.`stream_started`", false, false, false, "`streams_servers`.`bitrate`"]; + // One entry per column of the streams table (admin/streams.php), in order: + // control, select, id, icon, title, server, connections, status, player, + // EPG, stream info, usage, actions. false = not sortable in SQL. + $rOrder = [false, false, "`streams`.`id`", "`streams`.`stream_icon`", "`streams`.`stream_display_name`", "`streams_servers`.`current_source`", "`clients`", "`streams_servers`.`stream_started`", false, false, false, false, "`streams_servers`.`bitrate`"]; if (RequestManager::has("order") && 0 < strlen(RequestManager::get("order")[0]["column"] ?? '')) { $rOrderRow = (int) (RequestManager::get("order")[0]["column"] ?? 0); } else { @@ -1142,6 +1145,21 @@ class TableController extends BaseAdminController { $rPlayerVideo = strtoupper((string) ($rVideo["codec_name"] ?? "")); } + // What the producer costs this node, and which producer it is + // (ffmpeg, or the fanout daemon's native remuxer). Sampled + // from /proc by cron:streams ON the server that runs the + // stream — only it can read its own processes — and carried + // here in the progress report it writes anyway. + $rUsage = null; + $rUsageInfo = json_decode($rRow["progress_info"] ?? '', true); + if (is_array($rUsageInfo) && (isset($rUsageInfo["mem"]) || isset($rUsageInfo["producer"]))) { + $rUsage = [ + "cpu" => isset($rUsageInfo["cpu"]) ? (float) $rUsageInfo["cpu"] : null, + "mem" => isset($rUsageInfo["mem"]) ? (int) $rUsageInfo["mem"] : null, + "producer" => $rUsageInfo["producer"] ?? null, + ]; + } + // EPG availability + player codec compatibility. $rEPG = file_exists(EPG_PATH . "stream_" . $rRow["id"]) ? "available" : ($rRow["channel_id"] ? "pending" : "none"); $rPlayerOk = false; @@ -1176,6 +1194,7 @@ class TableController extends BaseAdminController { "notes" => !empty($rRow["notes"]) ? $rRow["notes"] : null, "player_ok" => $rPlayerOk, "info" => $rInfo, + "usage" => $rUsage, ]; } } diff --git a/src/Public/Views/admin/streams.php b/src/Public/Views/admin/streams.php index df306ee9..e4991fb8 100644 --- a/src/Public/Views/admin/streams.php +++ b/src/Public/Views/admin/streams.php @@ -128,6 +128,7 @@ $rStatusFilters = [ EPG + @@ -265,6 +266,26 @@ renderUnifiedLayoutFooter('admin'); var running = function(row) { return row.status === 1 || row.status === 2 || row.status === 3 || row.status === 5 || row.on_demand; }; + var fmtBytes = function(b) { + if (b == null) { + return '—'; + } + return b >= 1073741824 ? (b / 1073741824).toFixed(1) + ' GB' : Math.round(b / 1048576) + ' MB'; + }; + // CPU is percent of ONE core, so a transcode legitimately passes 100. + var cpuTone = function(c) { + if (c == null) { + return 'secondary'; + } + return c < 50 ? 'success' : (c < 150 ? 'warning' : 'danger'); + }; + // Which process produces the stream: the fanout daemon's native remuxer, + // ffmpeg, or PHP (the LLOD segmenter / loopback relay). + var PRODUCER = { + fanout: ['info', 'fanout'], + ffmpeg: ['secondary', 'ffmpeg'], + php: ['secondary', 'php'] + }; var selected = {}; var updateBulk = function() { @@ -467,6 +488,26 @@ renderUnifiedLayoutFooter('admin'); '' + esc(d.fps) + ''; } }, + { + data: 'usage', + orderable: false, + searchable: false, + className: 'text-nowrap', + render: function(d) { + if (!d) { + return '—'; + } + var p = PRODUCER[d.producer] || null; + var html = '
'; + if (p) { + html += '' + p[1] + ''; + } + html += '' + + (d.cpu == null ? '—' : Number(d.cpu).toFixed(1) + '%') + '' + + '' + esc(fmtBytes(d.mem)) + '
'; + return html; + } + }, { data: null, orderable: false, diff --git a/tests/Unit/ProcessResourceUsageTest.php b/tests/Unit/ProcessResourceUsageTest.php new file mode 100644 index 00000000..2e2e9941 --- /dev/null +++ b/tests/Unit/ProcessResourceUsageTest.php @@ -0,0 +1,61 @@ +assertIsArray($rSample); + $this->assertArrayHasKey('ticks', $rSample); + $this->assertGreaterThan(0, $rSample['rss'], 'a running process holds memory'); + $this->assertGreaterThan(0, $rSample['at']); + // PHP itself is a few MB at least, and nothing here is a gigabyte: a + // wrong page size or a misread field shows up as an absurd figure. + $this->assertGreaterThan(1 << 20, $rSample['rss']); + $this->assertLessThan(4 << 30, $rSample['rss']); + } + + public function testAProcessThatDoesNotExistReadsNothing(): void { + $rMax = (int) @file_get_contents('/proc/sys/kernel/pid_max'); + $this->assertNull(ProcessManager::resourceSample($rMax > 0 ? $rMax + 1 : 4194305)); + $this->assertNull(ProcessManager::resourceSample(0)); + $this->assertNull(ProcessManager::resourceSample(-1)); + } + + public function testCpuIsPercentOfOneCore(): void { + // /proc counts CPU time in USER_HZ = 100, so 150 ticks is 1.5 s of CPU; + // spent over 3 s of wall clock that is half a core. + $this->assertSame(50.0, ProcessManager::cpuPercent( + ['ticks' => 150, 'at' => 103.0], + ['ticks' => 0, 'at' => 100.0] + )); + // A transcode on several cores legitimately passes 100%. + $this->assertSame(250.0, ProcessManager::cpuPercent( + ['ticks' => 500, 'at' => 102.0], + ['ticks' => 0, 'at' => 100.0] + )); + } + + public function testAPairThatSaysNothingReportsNothing(): void { + // Restarted producer: the pid is new, its counter starts over. Reporting + // the difference would show a wild negative percentage. + $this->assertNull(ProcessManager::cpuPercent(['ticks' => 5, 'at' => 200.0], ['ticks' => 900, 'at' => 100.0])); + // Two readings from the same instant divide by zero. + $this->assertNull(ProcessManager::cpuPercent(['ticks' => 10, 'at' => 100.0], ['ticks' => 5, 'at' => 100.0])); + // No previous reading at all (the first pass after a start). + $this->assertNull(ProcessManager::cpuPercent(['ticks' => 10, 'at' => 100.0], [])); + } + + public function testProducerKind(): void { + $this->assertSame('php', ProcessManager::producerKind(getmypid())); + $rMax = (int) @file_get_contents('/proc/sys/kernel/pid_max'); + $this->assertNull(ProcessManager::producerKind($rMax > 0 ? $rMax + 1 : 4194305)); + } +}