diff --git a/README.md b/README.md index 8f00139..35407a3 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,25 @@ # gpufl-agent -Uploads the session logs that gpufl-client writes to the GPUFlight backend. +Uploads the session logs that gpufl-client, the gpufl-serving session writer +or any other producer writes to the GPUFlight backend. + +## Channels + +The agent uploads every channel it finds in a session directory: a channel is +a name with a published window sidecar (`.gpufl-window...json`), so +a new producer or channel needs no agent change. A bare `*.log` file, such as +an application's `server.log`, is never a channel and is never uploaded. The +native channels `device`, `scope`, `system` and `sass` are also recognized by +their bare window files, for clients that predate sidecars. A session may have +at most 16 channels. + +`--log-types=` (`GPUFL_LOG_TYPES`, or `logTypes` in a config file) is +an optional filter: only those channels are uploaded, and a directory without +any of them is left alone. + +[docs/session-layout.md](docs/session-layout.md) is the layout contract a +producer writes: session directories, the lock file, `.tmp/`, window files and +sidecars, ACK tombstones and markers. ## Running the agent as a separate user diff --git a/docs/session-layout.md b/docs/session-layout.md new file mode 100644 index 0000000..f3201b4 --- /dev/null +++ b/docs/session-layout.md @@ -0,0 +1,216 @@ +# Session layout + +This is the on-disk layout the agent consumes. Any producer that writes it (the +native gpufl-client, the Python gpufl-serving session writer, or a new tool) +has its channels uploaded with no agent change. Every statement here comes +from the agent code (`SessionWatcher`, `TailerManager`, `LogTailer`, +`SessionOwnership`, `AcknowledgedWindowCleaner`, `HttpPublisher`) or from the +native writer (`include/gpufl/core/logger/` in gpufl-client). Items marked +**Tolerated** work today but a producer should not rely on them. + +## Session directories + +- The agent watches the folders given by `--folder`, `--folders` or the + config file. A session is a directory either directly under a watched folder + (`//`) or one grouping level below it + (`///`, as multi-pass runs use + `run--`). Deeper directories are not found, and names + starting with `.` are skipped. +- The directory name is the session id. The agent sends it as + `X-GpuFlight-Session-Id` and requires every sidecar's `session_id` to equal + it. The native writer uses `/`. +- A directory is a session once it holds at least one channel (next section). + The agent scans every 2 seconds. +- With `--ignore-preexisting`, session directories last modified before the + agent started are ignored. + +## Channels + +- A channel name matches `[A-Za-z0-9_-]{1,64}`, the rule the backend's stream + endpoint accepts. It cannot contain a dot. +- A name becomes a channel when the session directory holds a sidecar + `.gpufl-window...json` for it: the sidecar of a published window, + or the same file kept as an ACK tombstone after the payload is deleted. The + sidecar counts once it parses as a JSON object with integer `payload_bytes` + and `payload_crc32`; the rest is checked when the window is uploaded. +- A bare `*.log` file never makes a channel. An application log next to the + windows (`server.log`, `server.1.log`) is never read or uploaded, and a + directory holding only such files is not a session. +- **Tolerated (legacy):** the four native names `device`, `scope`, `system` + and `sass` are also recognized by `.log`, `..log` or + `..log.gz` without a sidecar, for clients that predate sidecars. + Do not use these four names for anything else, and do not put an unrelated + file with such a name in a session directory. +- At most 16 channels per session. A session with more has the extra channels + left untailed; the agent never sends session-complete for it, marks it + `.failed` and never prunes it. +- `--log-types` (`GPUFL_LOG_TYPES`, `logTypes` in the config file) is an + optional filter. Without it every channel found is uploaded. With it only the + listed channels are uploaded, and a directory counts as a session only if a + listed channel is found in it. Such an agent still settles a directory once + its listed channels finish, so channels outside the filter in the same + directory are not uploaded by it or by a later agent. +- A channel whose first window appears while the session is live is tailed + from the next scan. Before sending session-complete the agent scans the + directory once more, so a channel published just before the producer + finished is uploaded too. + +## Lock file + +- `.gpufl-session.lock` in the session directory. The producer creates it + before it writes anything else, holds an exclusive OS lock on it for its + whole lifetime, and releases it last: after every window is published and + `.tmp/` is removed (or deliberately left behind). +- The native writer opens it with `O_CREAT | O_RDWR`, mode `0660` masked by the + umask, and takes `fcntl(F_SETLK)` with `F_WRLCK` over the whole file; on + Windows it uses `CreateFileW(OPEN_ALWAYS)` with read, write and delete + sharing and `LockFileEx` exclusive on the first byte. +- The agent probes it by opening it read-only and trying a shared lock over + the whole file (PR #30): not acquired means the session is active; acquired + means the writer is gone (the agent releases it at once). A shared lock + conflicts with the writer's exclusive one (`fcntl` on POSIX, `LockFileEx` on + Windows), so the agent's user only needs read access to the file. If the + agent cannot open it or the probe fails, it treats the session as active; + it warns once when it cannot open the file. +- On POSIX, lock with an `fcntl` record lock (for example Python + `fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)`). The agent's probe is an + `fcntl` lock, and Linux keeps `flock(2)` locks separate from those, so an + `flock` holder would look gone to the agent. +- Finished windows upload while the lock is held; only completion and the + orphan decision depend on the probe. +- **Tolerated (legacy):** a session without a lock file. Completion then + depends on `.tmp/` alone, and a payload without a sidecar is sent without + its window identity. With a lock file present, a payload whose sidecar is + missing is not uploaded; the agent retries every 5 seconds. + +## `.tmp/` staging + +- `/.tmp/` holds the files still being written. The native writer + keeps the active `.tmp/.log`, retired `.tmp/..log` and + gzip staging files there. The agent never reads files in `.tmp/`; it checks + only whether the directory exists and, without a lock file, how recently it + or its direct entries changed. +- Create `.tmp/` before the first window. Remove it only after the last window + of every channel is published: its removal tells the agent that no more + windows will appear. Then release the lock. +- Per channel, the agent finishes as follows: + - `.tmp/` present and lock held (or probe failed): keep waiting, with no + time limit. + - `.tmp/` present and lock released: the producer died without finishing. + The channel finishes and the session is marked `.failed`. + - `.tmp/` present, no lock file, and nothing in it changed for 30 seconds: + the same. + - `.tmp/` gone and lock held: keep waiting. + - `.tmp/` gone and lock released (or no lock file): wait 4.5 seconds, look + for the next window once more, then finish. +- An `--exit-when-drained` agent does not exit while a session's `.tmp/` + changed within the last 30 seconds. + +## Window files + +- A window is `..log.gz` in the session directory: a gzip file of + NDJSON, one JSON record per line. +- `N` starts at 1 and grows by 1 per channel, with no gaps. The agent sends a + channel's windows strictly in order and waits at the first `N` it cannot + find. A window after a gap is never sent: the channel finishes at the gap + when the session ends. The only exception is an `N` whose ACK tombstone + proves it was already sent. +- Publish the sidecar first, then the payload. Make each appear atomically: + write it elsewhere (the native writer stages in `.tmp/`) and rename it into + place without replacing an existing file. A visible window is complete and + never changes; the agent may read it at once. +- The agent deletes acknowledged payloads, so derive the next `N` from the + sidecars as well as the payloads. The native writer takes the highest index + found in `.tmp/`, in the session directory and in the sidecars, plus one. +- The agent reads a whole window into memory and sends it in one request, so + keep windows bounded. The native writer rotates them by size and time and + never publishes an empty one. +- **Tolerated:** a plain `..log` (uncompressed NDJSON, from native + builds without a compressor). The agent prefers it over a `.log.gz` with the + same `N`, reads it line by line and sends batches without the window + identity, the CRC check, the ACK or payload cleanup. + +## Sidecars + +`.gpufl-window...json`, one per window, a single JSON object that +never changes once published. The native writer writes it to +`.part.` and moves it into place without replacing; that staging +name is not matched by the agent. Unknown fields are ignored. + +| Field | Value | Agent check | +|---|---|---| +| `schema_version` | `1` | must be 1 | +| `type` | `"transport_window"` | must match | +| `window_id` | UUID, new for every window | must not be blank; sent as `X-GpuFlight-Window-Id` | +| `session_id` | the session directory name | must match | +| `channel` | the channel in the file names | must match; sent as `X-GpuFlight-Window-Channel` | +| `window_sequence` | `N` | must match; sent as `X-GpuFlight-Window-Sequence` | +| `opened_mono_ms`, `closed_mono_ms` | producer monotonic clock when the window opened and closed, valid within that run only (native writes -1 when unknown) | not checked | +| `created_wall_ms` | wall clock, ms since the epoch, when the sidecar was made | not checked | +| `payload_file` | the published payload's file name (`serving.3.log.gz`), never a staging name | must match | +| `payload_bytes` | size of the payload file in bytes (the gzip bytes) | must be >= 0 and equal the bytes read | +| `payload_crc32` | CRC-32 (zlib, as `java.util.zip.CRC32`) of the payload file bytes, unsigned decimal | must equal; sent as `X-GpuFlight-Window-CRC32` | + +A sidecar that is malformed or does not match its payload is a permanent +failure: the agent writes `.gpufl-transport-loss.agent-..json`, +keeps the payload and sidecar, stops that channel and never sends +session-complete. The backend rejects a reused `window_id` with different bytes +(409 `window_identity_conflict`); the agent keeps the window and retries. + +## ACK tombstones + +- When the backend answers a window with `"window_acknowledged": true`, the + agent writes `.gpufl-window-ack..` holding the `window_id` and a + newline, synced to disk, before it moves on. +- It then deletes the payload, unless `--retain-acked-payloads` is set, but + only if the sidecar matches the payload and the ACK file holds the sidecar's + `window_id`. The sidecar stays as the tombstone: it keeps `N` used and still + marks the channel for discovery. +- A producer must never delete or rewrite sidecars or `.gpufl-window-ack.*` + files, never reuse an `N`, and must expect payloads to disappear. + +## Markers + +- `.uploaded`: the agent finished every channel cleanly. `.failed`: it + finished off an orphaned `.tmp/`, found a transport-loss marker, or the + session was over the channel cap. Every later scan, by any agent, skips a + directory with either marker. With `--prune-failed` the agent deletes such a + session instead of marking it `.failed`, except over the channel cap. + Producers never create these files and never reuse a settled directory. +- `.gpufl-transport-loss.*`: any file with this prefix means data was lost. + The agent then never sends session-complete and marks the session `.failed`. + The native writer writes `.gpufl-transport-loss...json` + (`schema_version`, `type: "transport_window_loss"`, `channel`, + `window_index`, `reason`) when it drops data; a producer that drops records + should write one too, so the session is not reported complete. +- The agent ignores every other file in the session directory, including the + native writer's `.gpufl-control..json` and `.gpufl-control-ack..json`. + +## Completion + +In stream mode the agent sends session-complete once: after every channel it +tails has finished and a last scan found no new channel. It does not send it +when a transport-loss marker exists or the session is over the channel cap. It +does send it for an orphaned session (lock released with `.tmp/` left), which +it then marks `.failed`. It tries three times with backoff; after that the +backend's grace finalize applies. + +## Record contents + +In stream mode (the HTTP default) the agent sends each gzip window +byte-for-byte; it never decompresses it or reads its records. The backend +routes each record by its `type` field. + +**Tolerated:** plain `.log` windows, `--upload-mode=legacy` and the Kafka +publisher take the line path instead: the agent decompresses the window, parses +every line as JSON, drops malformed lines, reads `type` (which picks the +endpoint in legacy mode) and filters `device_metric_batch` rows on the `system` +channel. A producer whose records must stay unread by the agent publishes gzip +windows and relies on stream mode. + +## Lifecycle rows + +The native writer writes a `job_start` row when the session starts and a +`shutdown` row as its last row before it closes, both to every channel. The +agent never reads or checks them; they are for the backend. A producer should +write `job_start` first and `shutdown` last in each of its channels. diff --git a/src/main/java/com/gpuflight/agent/GpuflAgent.java b/src/main/java/com/gpuflight/agent/GpuflAgent.java index b7ad7ab..5d38523 100644 --- a/src/main/java/com/gpuflight/agent/GpuflAgent.java +++ b/src/main/java/com/gpuflight/agent/GpuflAgent.java @@ -141,12 +141,13 @@ public void start() throws Exception { private void resolveWatchedFolders() { if (config.source() != null) { - log.info("Source folder: {} types={}", config.source().folder(), config.source().logTypes()); + log.info("Source folder: {} types={}", config.source().folder(), + describeLogTypes(config.source().logTypes())); watchedFolders.putIfAbsent(new File(config.source().folder()), config.source().logTypes()); } if (config.sources() != null) { for (LogSourceConfig s : config.sources()) { - log.info("Source folder: {} types={}", s.folder(), s.logTypes()); + log.info("Source folder: {} types={}", s.folder(), describeLogTypes(s.logTypes())); watchedFolders.putIfAbsent(new File(s.folder()), s.logTypes()); } } @@ -163,6 +164,11 @@ private void resolveWatchedFolders() { } } + // An empty filter tails every channel found; say so rather than log "[]". + private static String describeLogTypes(List logTypes) { + return logTypes.isEmpty() ? "all channels found" : String.join(",", logTypes); + } + private void shutdown() { log.info("Shutting down..."); if (executor != null) executor.shutdownNow(); diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java index 3bbc1db..65b4ae1 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -17,7 +17,10 @@ public class ConfigLoader { private static final Logger log = LoggerFactory.getLogger(ConfigLoader.class); - public static final List DEFAULT_LOG_TYPES = List.of("device", "scope", "system", "sass"); + /** No --log-types filter: every channel a producer publishes in a session is tailed, + * so a new producer (the Python serving writer's "serving", or a channel this agent + * has never heard of) needs no agent change. A list narrows that to its channels. */ + public static final List DEFAULT_LOG_TYPES = List.of(); public static AgentConfig load(String[] args) { String configPath = parseConfigArg(args); @@ -239,7 +242,7 @@ Config file (overrides all flags): Source (at least one of --folder or --folders is required): --folder= Log folder path [GPUFL_SOURCE_FOLDER] --folders= Auto-discover folders [GPUFL_SOURCE_FOLDERS] - --log-types= Log channels to tail [GPUFL_LOG_TYPES] default: device,scope,system,sass + --log-types= Only tail these channels [GPUFL_LOG_TYPES] default: every channel found --cursor-file= Cursor state file [GPUFL_CURSOR_FILE] default: ./cursor.json Publisher (required): diff --git a/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java b/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java index 6a26def..fb05c25 100644 --- a/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java +++ b/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java @@ -5,6 +5,19 @@ /** * A session found under a watched folder: the folder, the session_id - * (the subdir name), and the channels to tail. + * (the subdir name), the channels found in it to tail, and the --log-types + * filter they were found under (empty = none), which a later rescan of the + * session for new channels applies again. */ -public record DiscoveredSession(File folder, String sessionId, List logTypes) {} +public record DiscoveredSession( + File folder, String sessionId, List logTypes, List channelFilter) { + + public DiscoveredSession { + if (channelFilter == null) channelFilter = List.of(); + } + + /** A session limited to {@code logTypes}: a rescan looks for those channels only. */ + public DiscoveredSession(File folder, String sessionId, List logTypes) { + this(folder, sessionId, logTypes, logTypes); + } +} diff --git a/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java b/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java index 50fde06..d320d60 100644 --- a/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java +++ b/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java @@ -1,6 +1,7 @@ package com.gpuflight.agent.model; import com.fasterxml.jackson.annotation.JsonProperty; +import com.gpuflight.agent.config.ConfigLoader; import java.util.List; @@ -8,7 +9,8 @@ * A folder to watch for gpufl sessions. In v1.2 each run writes its logs * under {@code //.log[.gz]}, so a source is * just a folder — sessions are auto-discovered (the {@code session_id} is the - * subdir name). {@code logTypes} optionally restricts which channels to tail. + * subdir name). {@code logTypes} optionally restricts which channels to tail; + * without it every channel found in a session is tailed. * *

The pre-v1.2 {@code filePrefix} field was removed: it no longer maps to * anything on disk (sessions are nested, not flat {@code .}), @@ -20,8 +22,9 @@ public record LogSourceConfig( ) { public LogSourceConfig { if (folder == null) folder = "."; - // "sass" carries the SASS-disassembly / source-content artifacts split - // out of device.log; tail it by default or they miss live upload. - if (logTypes == null || logTypes.isEmpty()) logTypes = List.of("device", "scope", "system", "sass"); + // No filter by default: every channel a producer publishes is tailed, including + // "sass" (the SASS-disassembly / source-content artifacts split out of device.log) + // and "serving"; a fixed list would miss a producer's new channel. + if (logTypes == null || logTypes.isEmpty()) logTypes = ConfigLoader.DEFAULT_LOG_TYPES; } } diff --git a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java index 342443a..a2795d5 100644 --- a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -1,6 +1,7 @@ package com.gpuflight.agent.service; import com.gpuflight.agent.config.ConfigLoader; +import com.gpuflight.agent.config.JsonSettings; import com.gpuflight.agent.model.DiscoveredSession; import com.gpuflight.agent.util.Delays; import java.io.File; @@ -9,22 +10,36 @@ import java.util.Comparator; import java.util.List; import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.function.Consumer; +import java.util.regex.Matcher; import java.util.regex.Pattern; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import tools.jackson.databind.JsonNode; public class SessionWatcher { private static final Logger log = LoggerFactory.getLogger(SessionWatcher.class); - private static final Pattern CHANNEL_FILE_PATTERN = - Pattern.compile("^(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$"); - private static final Pattern WINDOW_TOMBSTONE_PATTERN = - Pattern.compile("^\\.gpufl-window\\." - + "(device|scope|system|sass)\\.[1-9][0-9]*\\.json$"); + /** Most channels the agent tails in one session. A producer adds a channel by + * publishing a window for it, so this bounds the tailers a broken producer can make + * the agent run; a session with more is never reported complete. */ + public static final int MAX_CHANNELS_PER_SESSION = 16; + + // A published window's sidecar, which stays as its tombstone once the payload is + // ACKed and deleted. The channel name rule is the one the backend's stream endpoint + // accepts. + private static final Pattern WINDOW_SIDECAR = Pattern.compile( + "^\\.gpufl-window\\.([A-Za-z0-9_-]{1,64})\\.[1-9][0-9]*\\.json$"); + + // The native channels of clients that predate sidecars are still recognized by their + // bare log files. No other name is: an application log next to the windows + // (server.log) must never become a channel. + private static final Pattern LEGACY_NATIVE_FILE = Pattern.compile( + "^(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$"); /** Marker file the agent drops in a session dir once every channel has been * fully uploaded. Discovery skips a session that has it, so a re-scan never @@ -65,11 +80,21 @@ public void start(ExecutorService executor) { }); } + /** + * The unsettled sessions under {@code folder}, each with the channels found in it. + * {@code logTypes} is the --log-types filter: empty (or null) tails every channel a + * producer publishes; otherwise only the named ones are tailed, and a directory is a + * session only if one of them is found in it. + */ public static List discoverSources(File folder, List logTypes) { List result = new ArrayList<>(); if (!folder.isDirectory()) return result; - List types = (logTypes == null || logTypes.isEmpty()) ? ConfigLoader.DEFAULT_LOG_TYPES : logTypes; + // Recognize a session only by a channel this agent tails. Adopting a directory + // whose channels all go untailed (a serving-only session under + // --log-types=device,scope,system,sass) would complete and mark it .uploaded + // with its windows never sent, so a later agent that does tail them skips it. + List filter = logTypes == null ? ConfigLoader.DEFAULT_LOG_TYPES : logTypes; // Legacy-format warning Pattern legacyTop = Pattern.compile( @@ -97,12 +122,13 @@ public static List discoverSources(File folder, List for (File subdir : subdirs) { if (subdir.getName().startsWith(".")) continue; - if (looksLikeSession(subdir)) { + // Only a session is ever settled, so skip one without reading its sidecars. + if (isSettled(subdir)) continue; + List channels = discoverChannels(subdir, filter); + if (!channels.isEmpty()) { // Flat: // - embedded gpufl and single-pass // `gpufl trace` write the session directly under the watched dir. - if (!isSettled(subdir)) { - result.add(new DiscoveredSession(folder, subdir.getName(), types)); - } + result.add(new DiscoveredSession(folder, subdir.getName(), channels, filter)); continue; } // One grouping level: /// - a multi-pass @@ -112,35 +138,64 @@ public static List discoverSources(File folder, List if (grandchildren == null) continue; Arrays.sort(grandchildren, Comparator.comparing(File::getName)); for (File leaf : grandchildren) { - if (leaf.getName().startsWith(".")) continue; - if (looksLikeSession(leaf) && !isSettled(leaf)) { - result.add(new DiscoveredSession(subdir, leaf.getName(), types)); + if (leaf.getName().startsWith(".") || isSettled(leaf)) continue; + List leafChannels = discoverChannels(leaf, filter); + if (!leafChannels.isEmpty()) { + result.add(new DiscoveredSession(subdir, leaf.getName(), leafChannels, filter)); } } } return result; } + /** + * The channels {@code sessionDir} holds, sorted; at most one more than + * {@link #MAX_CHANNELS_PER_SESSION}, enough to tell that a session is over it. A + * name is a channel once a window sidecar for it is published (or kept as an ACK + * tombstone); for the four native channels a bare window file also counts. + * {@code filter} limits the result to the channels it names, unless it is empty. + */ + public static List discoverChannels(File sessionDir, List filter) { + String[] names = sessionDir.list(); + if (names == null) return List.of(); + Arrays.sort(names); + Set channels = new TreeSet<>(); + for (String name : names) { + Matcher match = LEGACY_NATIVE_FILE.matcher(name); + boolean legacy = match.matches(); + if (!legacy) { + match = WINDOW_SIDECAR.matcher(name); + if (!match.matches()) continue; + } + String channel = match.group(1); + if (channels.contains(channel) + || !(filter.isEmpty() || filter.contains(channel))) continue; + File file = new File(sessionDir, name); + if (legacy ? file.isFile() : isWindowSidecar(file)) { + channels.add(channel); + if (channels.size() > MAX_CHANNELS_PER_SESSION) break; + } + } + return List.copyOf(channels); + } + + // A sidecar counts once it parses and carries the payload size and CRC32 the tailer + // verifies; the tailer checks the rest of its identity before it uploads the window. + private static boolean isWindowSidecar(File file) { + try { + JsonNode sidecar = JsonSettings.MAPPER.readTree(file); + return sidecar.isObject() + && sidecar.path("payload_bytes").isIntegralNumber() + && sidecar.path("payload_crc32").isIntegralNumber(); + } catch (RuntimeException unreadable) { + return false; + } + } + // A settled session (fully uploaded, or finished+marked failed) is skipped by a // re-scan so it is never re-tailed or re-signalled. private static boolean isSettled(File sessionDir) { return new File(sessionDir, UPLOADED_MARKER).exists() || new File(sessionDir, FAILED_MARKER).exists(); } - - // A directory is a session if it directly contains a channel log file - // (device/scope/system/sass[.N].log[.gz]). - private static boolean looksLikeSession(File dir) { - File[] inside = dir.listFiles(); - if (inside == null) return false; - for (File child : inside) { - if (child.isFile() && - (CHANNEL_FILE_PATTERN.matcher(child.getName()).matches() - || WINDOW_TOMBSTONE_PATTERN.matcher( - child.getName()).matches())) { - return true; - } - } - return false; - } } diff --git a/src/main/java/com/gpuflight/agent/service/TailerManager.java b/src/main/java/com/gpuflight/agent/service/TailerManager.java index 836430d..20bdac8 100644 --- a/src/main/java/com/gpuflight/agent/service/TailerManager.java +++ b/src/main/java/com/gpuflight/agent/service/TailerManager.java @@ -15,6 +15,10 @@ import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; @@ -34,10 +38,27 @@ public class TailerManager { private final StreamUploadSettings streamUploadSettings; private final String topicPrefix; - private final Set startedSessions = ConcurrentHashMap.newKeySet(); + private final Map sessions = new ConcurrentHashMap<>(); private final AtomicInteger activeTailers = new AtomicInteger(0); private volatile boolean pruneFailed = false; + /** One session's tailers. Its monitor guards the channel set and the completion + * decision, since the watcher can add a channel while the last tailer completes. */ + private static final class SessionRun { + final DiscoveredSession session; + final File dir; + final Set channels = new HashSet<>(); + final AtomicBoolean orphaned = new AtomicBoolean(false); + int running; + boolean overCap; + boolean completing; + + SessionRun(DiscoveredSession session) { + this.session = session; + this.dir = new File(session.folder(), session.sessionId()); + } + } + public TailerManager(ExecutorService executor, Publisher publisher, CursorManager cursorMgr, @@ -55,61 +76,116 @@ public TailerManager(ExecutorService executor, } /** - * Starts this session exactly once. + * Starts tailing this session once, and later tails each channel found in it that + * has no tailer yet: a channel's first window can be published while the session + * is live. * - * @return {@code true} only when this invocation submitted the tailers; {@code false} when - * an asynchronous watcher or a terminal-drain rescan had already started the session. + * @return {@code true} only when this invocation submitted tailers, for a new session + * or a new channel; {@code false} when an asynchronous watcher or a terminal-drain + * rescan had already started the session and every channel found in it. */ public boolean spawnSessionTailers(DiscoveredSession session) { String key = session.folder().getAbsolutePath() + "::" + session.sessionId(); - if (!startedSessions.add(key)) return false; - log.info("Tailing session \"{}\" in {} types={}", - session.sessionId(), session.folder(), session.logTypes()); - - File sessionDir = new File(session.folder(), session.sessionId()); - var remaining = new AtomicInteger(session.logTypes().size()); - var orphaned = new AtomicBoolean(false); - String sid = session.sessionId(); - for (String type : session.logTypes()) { - activeTailers.incrementAndGet(); - executor.submit(() -> { - try { - var dedup = "system".equals(type) ? deduplicator : null; - var tailer = new LogTailer(session.folder(), session.sessionId(), type, - topicPrefix, cursorMgr, consumedFilesQueue, dedup, - streamUploadSettings); - if (tailer.tail(publisher)) orphaned.set(true); // finished off a stale .tmp/ - if (!Thread.currentThread().isInterrupted() - && remaining.decrementAndGet() == 0) { - boolean transportLoss = - SessionOwnership.hasTransportLoss(sessionDir.toPath()); - if (transportLoss) { - orphaned.set(true); - log.error("session {} has a durable transport-loss " - + "marker; refusing session-complete", sid); - } - // Last channel done: every available window is uploaded. Signal the - // backend (stream mode), then settle the session so a re-scan skips it. - if (streamUploadSettings.enabled() && !transportLoss) { - signalSessionComplete(publisher, sid); - } - settleSession(sessionDir, orphaned.get()); - } - } finally { - activeTailers.decrementAndGet(); + SessionRun created = new SessionRun(session); + SessionRun run = sessions.putIfAbsent(key, created); + if (run == null) { + run = created; + log.info("Tailing session \"{}\" in {} channels={}", + session.sessionId(), session.folder(), session.logTypes()); + } + synchronized (run) { + List added = tailNewChannels(run, session.logTypes()); + if (run != created && !added.isEmpty()) { + log.info("Tailing channels {} published in live session \"{}\"", + added, session.sessionId()); + } + return run == created || !added.isEmpty(); + } + } + + // Starts a tailer for each channel in found that has none yet, up to the per-session + // cap. The caller holds the run's monitor. + private List tailNewChannels(SessionRun run, List found) { + List added = new ArrayList<>(); + if (run.completing) return added; + for (String channel : found) { + if (run.channels.contains(channel)) continue; + if (run.channels.size() >= SessionWatcher.MAX_CHANNELS_PER_SESSION) { + if (!run.overCap) { + log.error("session {} has more than {} channels; channel {} and any " + + "later one are not uploaded and the session will " + + "not be reported complete", run.session.sessionId(), + SessionWatcher.MAX_CHANNELS_PER_SESSION, channel); } - }); + run.overCap = true; + break; + } + run.channels.add(channel); + added.add(channel); + startTailer(run, channel); + } + return added; + } + + private void startTailer(SessionRun run, String type) { + run.running++; + activeTailers.incrementAndGet(); + executor.submit(() -> { + try { + var dedup = "system".equals(type) ? deduplicator : null; + var tailer = new LogTailer(run.session.folder(), run.session.sessionId(), type, + topicPrefix, cursorMgr, consumedFilesQueue, dedup, + streamUploadSettings); + if (tailer.tail(publisher)) run.orphaned.set(true); // finished off a stale .tmp/ + if (!Thread.currentThread().isInterrupted()) channelFinished(run); + } finally { + activeTailers.decrementAndGet(); + } + }); + } + + private void channelFinished(SessionRun run) { + synchronized (run) { + if (--run.running > 0) return; + // Every tailer has seen the producer finish. A channel whose first window it + // published just before that has no tailer if the watcher has not polled + // since, so look once more; its tailer completes the session instead. + List late = tailNewChannels(run, SessionWatcher.discoverChannels( + run.dir, run.session.channelFilter())); + if (!late.isEmpty()) { + log.info("Tailing channels {} found in session \"{}\" before completing it", + late, run.session.sessionId()); + return; + } + run.completing = true; + } + String sid = run.session.sessionId(); + boolean transportLoss = SessionOwnership.hasTransportLoss(run.dir.toPath()); + if (transportLoss) { + run.orphaned.set(true); + log.error("session {} has a durable transport-loss " + + "marker; refusing session-complete", sid); + } + if (run.overCap) { + log.error("session {} has channels that were not uploaded; " + + "refusing session-complete", sid); + } + // Last channel done: every available window is uploaded. Signal the + // backend (stream mode), then settle the session so a re-scan skips it. + if (streamUploadSettings.enabled() && !transportLoss && !run.overCap) { + signalSessionComplete(publisher, sid); } - return true; + settleSession(run.dir, run.orphaned.get() || run.overCap, run.overCap); } /** Settle a finished session so a later scan skips it. A producer that closed * cleanly (.tmp/ removed) is marked {@code .uploaded}; one finished off a stale * .tmp/ (producer crashed or was killed) is marked {@code .failed} - or deleted - * when pruning is on, since its windows are already uploaded. The marker's - * presence is the signal, so an empty file is enough. */ - private void settleSession(File sessionDir, boolean orphaned) { - if (orphaned && pruneFailed) { + * when pruning is on, since its windows are already uploaded. A session still + * holding windows that were never sent (over the channel cap) is never pruned. + * The marker's presence is the signal, so an empty file is enough. */ + private void settleSession(File sessionDir, boolean orphaned, boolean holdsUnsentWindows) { + if (orphaned && pruneFailed && !holdsUnsentWindows) { // Windows are already uploaded; drop the local orphan. A locked .tmp/ can // block deletion - fall back to marking it failed. if (deleteRecursively(sessionDir)) { @@ -160,6 +236,6 @@ public void setPruneFailed(boolean prune) { /** True once any session has been discovered and tailed. Cumulative - never cleared. */ public boolean hasStartedAnySession() { - return !startedSessions.isEmpty(); + return !sessions.isEmpty(); } } diff --git a/src/test/java/com/gpuflight/agent/MainTest.java b/src/test/java/com/gpuflight/agent/MainTest.java index 6dfed4a..e2be509 100644 --- a/src/test/java/com/gpuflight/agent/MainTest.java +++ b/src/test/java/com/gpuflight/agent/MainTest.java @@ -98,8 +98,8 @@ void logTypesOrDefault_usesExplicitFolderLogTypes() { } @Test - void logTypesOrDefault_usesDefaultWhenAbsent() { - assertEquals(List.of("device", "scope", "system", "sass"), + void logTypesOrDefault_isNoFilterWhenAbsent() { + assertEquals(List.of(), ConfigLoader.logTypesOrDefault(new String[]{}, Collections.emptyMap())); } diff --git a/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java b/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java index 326e9d6..8e7f0dd 100644 --- a/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java +++ b/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java @@ -17,7 +17,7 @@ void nullFolder_defaultsToDot() { @Test void nullLogTypes_defaultsToAllChannels() { LogSourceConfig config = new LogSourceConfig("/logs", null); - assertEquals(List.of("device", "scope", "system", "sass"), config.logTypes()); + assertEquals(List.of(), config.logTypes()); // no filter: every channel found } @Test diff --git a/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java b/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java index 8b8e9c2..3d22ab5 100644 --- a/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java +++ b/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java @@ -73,8 +73,8 @@ void discoversFlatAndNestedTogether(@TempDir Path root) throws IOException { void discoversSessionAfterAckCleanupLeavesOnlyWindowTombstone( @TempDir Path root) throws IOException { Path session = Files.createDirectories(root.resolve("acked-not-settled")); - Files.writeString( - session.resolve(".gpufl-window.device.1.json"), "{}"); + Files.writeString(session.resolve(".gpufl-window.device.1.json"), + "{\"payload_bytes\":20,\"payload_crc32\":12345}"); List found = SessionWatcher.discoverSources(root.toFile(), TYPES); diff --git a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java index 4393dfd..f5a660f 100644 --- a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java +++ b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java @@ -1,22 +1,39 @@ package com.gpuflight.agent.service; import com.gpuflight.agent.CursorManager; +import com.gpuflight.agent.SessionOwnership; +import com.gpuflight.agent.config.ConfigLoader; +import com.gpuflight.agent.config.HttpConfig; +import com.gpuflight.agent.config.JsonSettings; import com.gpuflight.agent.config.StreamUploadSettings; import com.gpuflight.agent.model.DiscoveredSession; import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.model.WindowMetadata; +import com.gpuflight.agent.publisher.HttpPublisher; import com.gpuflight.agent.publisher.Publisher; +import com.sun.net.httpserver.HttpServer; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardOpenOption; import java.util.List; +import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.zip.CRC32; +import java.util.zip.GZIPOutputStream; import static org.junit.jupiter.api.Assertions.*; @@ -139,4 +156,264 @@ void transportLossNeverSignalsSessionComplete(@TempDir Path root) assertTrue(new File(sess, SessionWatcher.FAILED_MARKER).exists()); assertFalse(new File(sess, SessionWatcher.UPLOADED_MARKER).exists()); } + + @Test + void servingOnlySessionUploadsAndCompletesOnceItsLockIsReleased( + @TempDir Path root) throws Exception { + // The Python serving writer's layout: serving windows only, no native channel. + Path sess = Files.createDirectories(root.resolve("serving-sess")); + Files.createDirectories(sess.resolve(".tmp")); + publishWindow(sess, "serving", 1); + Path lock = Files.writeString(sess.resolve(SessionOwnership.LOCK_FILE), ""); + + try (FakeBackend backend = new FakeBackend()) { + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + TailerManager manager = backend.managerIn(ex, root); + + try (FileChannel writer = FileChannel.open( + lock, StandardOpenOption.READ, StandardOpenOption.WRITE); + FileLock ignored = writer.lock()) { + // No configuration: the serving channel is found, not listed. + manager.spawnSessionTailers(discoverOnly(root)); + + // A finished window uploads while the writer still holds the lock... + backend.awaitEvent("serving"); + assertEquals(List.of("serving"), backend.events); + Files.delete(sess.resolve(".tmp")); + } + // ...and the session completes once the writer releases it. + runToCompletion(ex); + assertEquals(List.of("serving", COMPLETE), backend.events); + assertTrue(new File(sess.toFile(), SessionWatcher.UPLOADED_MARKER).exists()); + } + } + + @Test + void bareApplicationLogIsNeverUploadedOrTakenForAChannel(@TempDir Path root) + throws Exception { + // An application log next to the windows can hold prompts. It has no sidecar, + // so it is no channel, and a directory holding only such logs is no session. + String prompt = "{\"type\":\"request\",\"prompt\":\"do not upload\"}\n"; + Path sess = Files.createDirectories(root.resolve("serving-sess")); + publishWindow(sess, "serving", 1); + Files.writeString(sess.resolve("server.log"), prompt); + Files.writeString(sess.resolve("server.1.log"), prompt); + Path appOnly = Files.createDirectories(root.resolve("app-only")); + Files.writeString(appOnly.resolve("server.log"), prompt); + + try (FakeBackend backend = new FakeBackend()) { + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + TailerManager manager = backend.managerIn(ex, root); + DiscoveredSession found = discoverOnly(root); + assertEquals("serving-sess", found.sessionId()); + assertEquals(List.of("serving"), found.logTypes()); + manager.spawnSessionTailers(found); + runToCompletion(ex); + + assertEquals(List.of("serving", COMPLETE), backend.events); + assertFalse(new File(appOnly.toFile(), SessionWatcher.UPLOADED_MARKER).exists()); + } + } + + @Test + void nativeSessionUploadsExactlyItsFourChannels(@TempDir Path root) throws Exception { + Path sess = Files.createDirectories(root.resolve("native-sess")); + for (String channel : List.of("device", "scope", "system", "sass")) { + publishWindow(sess, channel, 1); + } + + try (FakeBackend backend = new FakeBackend()) { + uploadWithNoConfiguration(backend, root); + assertEquals(List.of("device", "sass", "scope", "system"), backend.windows()); + assertEquals(1, backend.completions()); + } + } + + @Test + void unknownProducerChannelUploadsWithNoConfiguration(@TempDir Path root) + throws Exception { + Path sess = Files.createDirectories(root.resolve("foo-sess")); + publishWindow(sess, "foo", 1); + publishWindow(sess, "foo", 2); + + try (FakeBackend backend = new FakeBackend()) { + uploadWithNoConfiguration(backend, root); + assertEquals(List.of("foo", "foo", COMPLETE), backend.events); + } + } + + @Test + void channelsPublishedLateAreUploadedBeforeCompletion(@TempDir Path root) + throws Exception { + Path sess = Files.createDirectories(root.resolve("late-sess")); + Files.createDirectories(sess.resolve(".tmp")); + publishWindow(sess, "serving", 1); + Path lock = Files.writeString(sess.resolve(SessionOwnership.LOCK_FILE), ""); + + try (FakeBackend backend = new FakeBackend()) { + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + TailerManager manager = backend.managerIn(ex, root); + try (FileChannel writer = FileChannel.open( + lock, StandardOpenOption.READ, StandardOpenOption.WRITE); + FileLock ignored = writer.lock()) { + assertTrue(manager.spawnSessionTailers(discoverOnly(root))); + backend.awaitEvent("serving"); + + // A channel's first window published while the session is live: the + // next scan gives it a tailer. + publishWindow(sess, "foo", 1); + assertTrue(manager.spawnSessionTailers(discoverOnly(root))); + backend.awaitEvent("foo"); + assertFalse(manager.spawnSessionTailers(discoverOnly(root))); + + // One published just before the producer finishes, with no scan after it. + publishWindow(sess, "bar", 1); + Files.delete(sess.resolve(".tmp")); + } + awaitUploaded(sess); + runToCompletion(ex); + assertEquals(List.of("serving", "foo", "bar", COMPLETE), backend.events); + } + } + + @Test + void sessionOverTheChannelCapIsNeverReportedComplete(@TempDir Path root) + throws Exception { + Path sess = Files.createDirectories(root.resolve("broken-sess")); + for (int i = 0; i <= SessionWatcher.MAX_CHANNELS_PER_SESSION; i++) { + publishWindow(sess, "ch" + (char) ('a' + i), 1); + } + + try (FakeBackend backend = new FakeBackend()) { + uploadWithNoConfiguration(backend, root); + assertEquals(SessionWatcher.MAX_CHANNELS_PER_SESSION, backend.windows().size()); + assertEquals(0, backend.completions()); + assertTrue(new File(sess.toFile(), SessionWatcher.FAILED_MARKER).exists()); + assertTrue(Files.exists(sess.resolve("chq.1.log.gz")), "unsent window was removed"); + } + } + + @Test + void agentNotTailingServingNeverAdoptsServingOnlySession(@TempDir Path root) + throws Exception { + Path sess = Files.createDirectories(root.resolve("serving-sess")); + publishWindow(sess, "serving", 1); + + CompletionPublisher publisher = new CompletionPublisher(); + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + TailerManager manager = streamManagerIn(ex, root, publisher); + // The launcher's log types today: native channels only. + for (DiscoveredSession s : SessionWatcher.discoverSources( + root.toFile(), List.of("device", "scope", "system", "sass"))) { + manager.spawnSessionTailers(s); + } + runToCompletion(ex); + + assertFalse(manager.hasStartedAnySession()); + assertEquals(0, publisher.completionSignals); + assertFalse(new File(sess.toFile(), SessionWatcher.UPLOADED_MARKER).exists()); + } + + private static final String COMPLETE = "session-complete"; + + /** Fake backend: acknowledges every window identity and records, in arrival order, + * each window's channel and each session-complete signal. */ + private static final class FakeBackend implements AutoCloseable { + final List events = new CopyOnWriteArrayList<>(); + private final HttpServer server; + + FakeBackend() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/api/v1/events/", exchange -> { + exchange.getRequestBody().readAllBytes(); + String path = exchange.getRequestURI().getPath(); + String body = "{}"; + if (path.endsWith("/stream")) { + String channel = exchange.getRequestHeaders() + .getFirst("X-GpuFlight-Window-Channel"); + events.add(channel == null ? "batch-without-window-identity" : channel); + body = "{\"window_acknowledged\":true}"; + } else if (path.endsWith("/session-complete")) { + events.add(COMPLETE); + } + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.start(); + } + + TailerManager managerIn(ExecutorService ex, Path root) { + HttpConfig http = new HttpConfig( + "http://127.0.0.1:" + server.getAddress().getPort(), "v1", null, 5); + return new TailerManager(ex, new HttpPublisher(http), + new CursorManager(new File(root.toFile(), "cursor.json")), + new LinkedBlockingQueue<>(), null, StreamUploadSettings.from(http), + "gpu-trace"); + } + + List windows() { + return events.stream().filter(e -> !COMPLETE.equals(e)).sorted().toList(); + } + + long completions() { + return events.stream().filter(COMPLETE::equals).count(); + } + + void awaitEvent(String event) throws InterruptedException { + for (int i = 0; i < 100 && !events.contains(event); i++) Thread.sleep(100); + assertTrue(events.contains(event), event + " not received"); + } + + @Override + public void close() { + server.stop(0); + } + } + + // A producer's window: its sidecar first, then the gzip payload the sidecar describes. + private static void publishWindow(Path sess, String channel, int sequence) + throws IOException { + String sid = sess.getFileName().toString(); + byte[] gz = gzip("{\"type\":\"" + channel + "_record\",\"session_id\":\"" + sid + "\"}\n"); + CRC32 crc = new CRC32(); + crc.update(gz); + String payload = channel + "." + sequence + ".log.gz"; + Files.writeString(sess.resolve(".gpufl-window." + channel + "." + sequence + ".json"), + JsonSettings.MAPPER.writeValueAsString(new WindowMetadata( + 1, "transport_window", UUID.randomUUID().toString(), sid, channel, + sequence, 10, 20, 30, payload, gz.length, crc.getValue()))); + Files.write(sess.resolve(payload), gz); + } + + private static DiscoveredSession discoverOnly(Path root) { + List found = SessionWatcher.discoverSources( + root.toFile(), ConfigLoader.DEFAULT_LOG_TYPES); + assertEquals(1, found.size()); + return found.get(0); + } + + // Discovers and tails the one session under root as an agent with no --log-types. + private static void uploadWithNoConfiguration(FakeBackend backend, Path root) + throws InterruptedException { + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + backend.managerIn(ex, root).spawnSessionTailers(discoverOnly(root)); + runToCompletion(ex); + } + + // Waits with the executor still open, since completing may start a late tailer. + private static void awaitUploaded(Path sess) throws InterruptedException { + Path marker = sess.resolve(SessionWatcher.UPLOADED_MARKER); + for (int i = 0; i < 300 && !Files.exists(marker); i++) Thread.sleep(100); + assertTrue(Files.exists(marker), "session not settled"); + } + + private static byte[] gzip(String text) throws IOException { + var out = new ByteArrayOutputStream(); + try (var gz = new GZIPOutputStream(out)) { + gz.write(text.getBytes(StandardCharsets.UTF_8)); + } + return out.toByteArray(); + } }