Upload every channel a producer publishes

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.<channel>.<N>.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=<a,b,...>` (`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
 

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
+  (`<folder>/<session_id>/`) or one grouping level below it
+  (`<folder>/<group>/<session_id>/`, as multi-pass runs use
+  `run-<app>-<analysis_id>`). 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 `<base_path>/<session_id>`.
+- 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.<channel>.<N>.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 `<name>.log`, `<name>.<N>.log` or
+  `<name>.<N>.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
+
+- `<session>/.tmp/` holds the files still being written. The native writer
+  keeps the active `.tmp/<channel>.log`, retired `.tmp/<channel>.<N>.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 `<channel>.<N>.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 `<channel>.<N>.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.<channel>.<N>.json`, one per window, a single JSON object that
+never changes once published. The native writer writes it to
+`<name>.part.<id>` 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-<channel>.<N>.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.<channel>.<N>` 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.<channel>.<index>.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.<N>.json` and `.gpufl-control-ack.<N>.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.

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<String> logTypes) {
+        return logTypes.isEmpty() ? "all channels found" : String.join(",", logTypes);
+    }
+
     private void shutdown() {
         log.info("Shutting down...");
         if (executor != null) executor.shutdownNow();

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<String> 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<String> 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=<path>              Log folder path          [GPUFL_SOURCE_FOLDER]
               --folders=<p1,p2,...>        Auto-discover folders    [GPUFL_SOURCE_FOLDERS]
-              --log-types=<a,b,...>        Log channels to tail     [GPUFL_LOG_TYPES]      default: device,scope,system,sass
+              --log-types=<a,b,...>        Only tail these channels [GPUFL_LOG_TYPES]      default: every channel found
               --cursor-file=<path>         Cursor state file        [GPUFL_CURSOR_FILE]    default: ./cursor.json
 
             Publisher (required):

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<String> logTypes) {}
+public record DiscoveredSession(
+        File folder, String sessionId, List<String> logTypes, List<String> 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<String> logTypes) {
+        this(folder, sessionId, logTypes, logTypes);
+    }
+}

src/main/java/com/gpuflight/agent/model/LogSourceConfig.java

@@ -1,14 +1,16 @@
 package com.gpuflight.agent.model;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.gpuflight.agent.config.ConfigLoader;
 
 import java.util.List;
 
 /**
  * A folder to watch for gpufl sessions. In v1.2 each run writes its logs
  * under {@code <folder>/<session_id>/<channel>.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.
  *
  * <p>The pre-v1.2 {@code filePrefix} field was removed: it no longer maps to
  * anything on disk (sessions are nested, not flat {@code <prefix>.<channel>}),
@@ -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;
     }
 }