[Core] LogMonitor never archives zero-byte worker logs and delays new log discovery

[acake88]

What happened + What you expected to happen

LogMonitor discovers worker-*.out/.err files and puts them in
closed_file_infos. A file is opened only after its size grows. Worker logs are
currently moved to logs/old/ only from _close_all_files(), which operates on
files that were previously opened.

As a result, a worker stdout/stderr file that remains zero bytes until its worker
exits is never opened and therefore never reaches the existing archival path. It
stays indefinitely in the active logs/ directory and in closed_file_infos.

On a long-running cluster this accumulated to:

• 771,932 files directly under one session's logs/ directory;
• 483,413 matching worker-*.out/.err files, with hundreds of thousands of
  zero-byte files in the directory;
• 4.05-4.34 seconds for the six filename glob operations in one
  update_log_filenames()-equivalent scan;
• about 37-38 seconds between creation of a new job log and the corresponding
  Beginning to track file message.

There is a second scaling problem in the same backlog path:
open_closed_files() repeatedly calls closed_file_infos.pop(0). Removing
500,000 elements from the front of a Python list took 15.91 seconds in an
isolated benchmark, while collections.deque.popleft() took 0.015 seconds.
Because run() invokes update_log_filenames() and open_closed_files() in the
same loop, time spent draining this queue delays the next filename-discovery
iteration.

Expected behavior:

1. stale zero-byte worker logs whose worker process has exited should complete
   the existing lifecycle and move to logs/old/;
2. processing the closed-file queue should remain linear as the queue grows;
3. active, recent, non-worker, or concurrently growing files must not be
   archived.

This is narrower than #62189 and #62961. Those issues propose general retention
or deletion, including cleanup of logs/old/. This issue is about a lifecycle
gap in the existing worker-log archival path and its direct impact on timely log
discovery. No deletion or general retention policy is proposed here.

Versions / Dependencies

• Reproduced against Ray master at commit
  167681c2b2485fffc670ddd9edd40ed428854e01.
• The production observation came from a downstream Ray 3.0-based image that
  retains the same LogMonitor behavior.
• Production runtime: Linux, Python 3.10.

Reproduction script

import os
import tempfile
import time
from pathlib import Path

from ray._private.log_monitor import LogMonitor

with tempfile.TemporaryDirectory() as tmp:
    logs = Path(tmp) / "logs"
    logs.mkdir()
    (logs / "old").mkdir()
    stale = time.time() - 120

    for pid in range(10000, 15000):
        path = logs / f"worker-{'a' * 56}-01000000-{pid}.out"
        path.touch()
        os.utime(path, (stale, stale))

    monitor = LogMonitor(
        "127.0.0.1",
        str(logs),
        None,
        lambda _: False,
        max_files_open=200,
    )
    monitor.update_log_filenames()
    monitor.open_closed_files()

    print("active worker logs:", len(list(logs.glob("worker-*.out"))))
    print("closed_file_infos:", len(monitor.closed_file_infos))
    print("archived worker logs:", len(list((logs / "old").iterdir())))

On unpatched master, all 5,000 zero-byte files remain active and tracked as
closed files, and none are archived.

Issue Severity

Medium: It is a significant difficulty but I can work around it.

[acake88]

I'd like to work on this focused bug. I will keep general log retention and
deletion out of scope: the patch will complete the existing archive-to-old/
lifecycle for stale zero-byte logs from dead workers, replace the closed-file
queue's quadratic front removal with deque.popleft(), and add regression tests
for the safety conditions and the file-growth race.

[MengjinYan]

Thanks for reporting the issue and thanks for offering the help, @acake88! Please feel free to submit a PR and we can review that once it is ready.