From f352339bf8a251c0eed741009afec995e85d00bc Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Mon, 26 Oct 2026 19:33:25 -0700 Subject: [PATCH 1/4] Discover and upload serving-only session directories The Python serving writer produces sessions whose only channel is serving: serving..log.gz windows with sidecars, a .tmp/ staging dir and the session lock, but no device, scope, system or sass file. Discovery matched only the four native channels, so such a directory was never tailed, and serving was not tailed by default either. Add serving to the default log types and derive the discovery patterns from that one list, so a known channel can no longer be tailed without being discovered. LogSourceConfig now uses the same constant instead of its own copy. Window records stay opaque; the agent only transports them. --- .../gpuflight/agent/config/ConfigLoader.java | 8 +- .../agent/model/LogSourceConfig.java | 8 +- .../agent/service/SessionWatcher.java | 11 ++- .../java/com/gpuflight/agent/MainTest.java | 2 +- .../agent/model/LogSourceConfigTest.java | 2 +- .../agent/service/TailerManagerTest.java | 98 +++++++++++++++++++ 6 files changed, 118 insertions(+), 11 deletions(-) diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java index 3bbc1db..6d8f5ab 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -17,7 +17,11 @@ 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"); + /** Every channel the agent knows; discovery recognizes a session directory by any + * of them. "serving" carries the records the Python serving writer emits, and its + * session directories hold no native channel at all. */ + public static final List DEFAULT_LOG_TYPES = + List.of("device", "scope", "system", "sass", "serving"); public static AgentConfig load(String[] args) { String configPath = parseConfigArg(args); @@ -239,7 +243,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= Log channels to tail [GPUFL_LOG_TYPES] default: device,scope,system,sass,serving --cursor-file= Cursor state file [GPUFL_CURSOR_FILE] default: ./cursor.json Publisher (required): diff --git a/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java b/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java index 50fde06..727b7f1 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; @@ -20,8 +21,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"); + // Tail every known channel by default: "sass" carries the SASS-disassembly / + // source-content artifacts split out of device.log, and "serving" the LLM-serving + // records; either would otherwise miss live upload. + 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..6a529a6 100644 --- a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -20,11 +20,14 @@ public class SessionWatcher { private static final Logger log = LoggerFactory.getLogger(SessionWatcher.class); + // Derived from the known channels, so a directory holding only serving windows + // (the Python serving writer's sessions) is discovered like a native one. + private static final String CHANNELS = String.join("|", ConfigLoader.DEFAULT_LOG_TYPES); private static final Pattern CHANNEL_FILE_PATTERN = - Pattern.compile("^(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$"); + Pattern.compile("^(" + CHANNELS + ")(?:\\.\\d+)?\\.log(?:\\.gz)?$"); private static final Pattern WINDOW_TOMBSTONE_PATTERN = Pattern.compile("^\\.gpufl-window\\." - + "(device|scope|system|sass)\\.[1-9][0-9]*\\.json$"); + + "(" + CHANNELS + ")\\.[1-9][0-9]*\\.json$"); /** 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 @@ -128,8 +131,8 @@ private static boolean isSettled(File sessionDir) { || 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]). + // A directory is a session if it directly contains a known channel's log file + // ([.N].log[.gz]) or window tombstone. private static boolean looksLikeSession(File dir) { File[] inside = dir.listFiles(); if (inside == null) return false; diff --git a/src/test/java/com/gpuflight/agent/MainTest.java b/src/test/java/com/gpuflight/agent/MainTest.java index 6dfed4a..dafa6de 100644 --- a/src/test/java/com/gpuflight/agent/MainTest.java +++ b/src/test/java/com/gpuflight/agent/MainTest.java @@ -99,7 +99,7 @@ void logTypesOrDefault_usesExplicitFolderLogTypes() { @Test void logTypesOrDefault_usesDefaultWhenAbsent() { - assertEquals(List.of("device", "scope", "system", "sass"), + assertEquals(List.of("device", "scope", "system", "sass", "serving"), 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..c2291d3 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("device", "scope", "system", "sass", "serving"), config.logTypes()); } @Test diff --git a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java index 4393dfd..e3c0f40 100644 --- a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java +++ b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java @@ -1,22 +1,40 @@ 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.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.zip.CRC32; +import java.util.zip.GZIPOutputStream; import static org.junit.jupiter.api.Assertions.*; @@ -139,4 +157,84 @@ 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")); + byte[] gz = gzip("{\"type\":\"serving_request\",\"session_id\":\"serving-sess\"}\n"); + Files.write(sess.resolve("serving.1.log.gz"), gz); + CRC32 crc = new CRC32(); + crc.update(gz); + Files.writeString(sess.resolve(".gpufl-window.serving.1.json"), + JsonSettings.MAPPER.writeValueAsString(new WindowMetadata( + 1, "transport_window", "11111111-2222-4333-8444-555555555555", + "serving-sess", "serving", 1, 10, 20, 30, + "serving.1.log.gz", gz.length, crc.getValue()))); + Path lock = Files.writeString(sess.resolve(SessionOwnership.LOCK_FILE), ""); + + // Fake backend: acknowledges the window identity and the completion signal. + List channels = new CopyOnWriteArrayList<>(); + AtomicInteger completions = new AtomicInteger(); + CountDownLatch uploaded = new CountDownLatch(1); + HttpServer backend = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + backend.createContext("/api/v1/events/", exchange -> { + exchange.getRequestBody().readAllBytes(); + String path = exchange.getRequestURI().getPath(); + String body = "{}"; + if (path.endsWith("/stream")) { + channels.add(exchange.getRequestHeaders().getFirst("X-GpuFlight-Window-Channel")); + body = "{\"window_acknowledged\":true}"; + uploaded.countDown(); + } else if (path.endsWith("/session-complete")) { + completions.incrementAndGet(); + } + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + backend.start(); + try { + HttpConfig http = new HttpConfig( + "http://127.0.0.1:" + backend.getAddress().getPort(), "v1", null, 5); + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + TailerManager manager = new TailerManager(ex, new HttpPublisher(http), + new CursorManager(new File(root.toFile(), "cursor.json")), + new LinkedBlockingQueue<>(), null, StreamUploadSettings.from(http), + "gpu-trace"); + + try (FileChannel writer = FileChannel.open( + lock, StandardOpenOption.READ, StandardOpenOption.WRITE); + FileLock ignored = writer.lock()) { + List found = SessionWatcher.discoverSources( + root.toFile(), ConfigLoader.DEFAULT_LOG_TYPES); + assertEquals(List.of("serving-sess"), + found.stream().map(DiscoveredSession::sessionId).toList()); + manager.spawnSessionTailers(found.get(0)); + + // A finished window uploads while the writer still holds the lock... + assertTrue(uploaded.await(10, TimeUnit.SECONDS), "serving window not uploaded"); + assertEquals(List.of("serving"), channels); + assertEquals(0, completions.get()); + Files.delete(sess.resolve(".tmp")); + } + // ...and the session completes once the writer releases it. + runToCompletion(ex); + assertEquals(1, completions.get()); + assertTrue(new File(sess.toFile(), SessionWatcher.UPLOADED_MARKER).exists()); + } finally { + backend.stop(0); + } + } + + 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(); + } } From 36419a1dddad8de66be83178ad3bc21ffe75f40c Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Mon, 26 Oct 2026 19:36:55 -0700 Subject: [PATCH 2/4] Recognize sessions only by the channels this agent tails Discovery matched every known channel while the tailers follow the configured log types. The launcher still starts the agent with device,scope,system,sass, so such an agent would adopt a serving-only directory, tail none of its windows, signal session-complete and mark it .uploaded once the lock is released, and a later agent that does tail serving would then skip it: a new way to lose serving data. Build the discovery patterns from the configured types, with each name quoted since it comes from configuration. An agent that does not tail serving now leaves a serving-only directory alone; the default types still include serving. --- .../gpuflight/agent/config/ConfigLoader.java | 6 ++-- .../agent/service/SessionWatcher.java | 33 ++++++++++--------- .../agent/service/TailerManagerTest.java | 22 +++++++++++++ 3 files changed, 43 insertions(+), 18 deletions(-) diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java index 6d8f5ab..fa302db 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -17,9 +17,9 @@ public class ConfigLoader { private static final Logger log = LoggerFactory.getLogger(ConfigLoader.class); - /** Every channel the agent knows; discovery recognizes a session directory by any - * of them. "serving" carries the records the Python serving writer emits, and its - * session directories hold no native channel at all. */ + /** Every channel the agent knows, tailed when none are configured. "serving" carries + * the records the Python serving writer emits, and its session directories hold no + * native channel at all. */ public static final List DEFAULT_LOG_TYPES = List.of("device", "scope", "system", "sass", "serving"); diff --git a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java index 6a529a6..53e1da1 100644 --- a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -13,6 +13,7 @@ import java.util.concurrent.ExecutorService; import java.util.function.Consumer; import java.util.regex.Pattern; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -20,15 +21,6 @@ public class SessionWatcher { private static final Logger log = LoggerFactory.getLogger(SessionWatcher.class); - // Derived from the known channels, so a directory holding only serving windows - // (the Python serving writer's sessions) is discovered like a native one. - private static final String CHANNELS = String.join("|", ConfigLoader.DEFAULT_LOG_TYPES); - private static final Pattern CHANNEL_FILE_PATTERN = - Pattern.compile("^(" + CHANNELS + ")(?:\\.\\d+)?\\.log(?:\\.gz)?$"); - private static final Pattern WINDOW_TOMBSTONE_PATTERN = - Pattern.compile("^\\.gpufl-window\\." - + "(" + CHANNELS + ")\\.[1-9][0-9]*\\.json$"); - /** 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 * re-tails or re-signals an already-finished session. */ @@ -74,6 +66,16 @@ public static List discoverSources(File folder, List 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. + String channels = types.stream().map(Pattern::quote).collect(Collectors.joining("|")); + Pattern channelFile = + Pattern.compile("^(" + channels + ")(?:\\.\\d+)?\\.log(?:\\.gz)?$"); + Pattern windowTombstone = + Pattern.compile("^\\.gpufl-window\\.(" + channels + ")\\.[1-9][0-9]*\\.json$"); + // Legacy-format warning Pattern legacyTop = Pattern.compile( "^(.+)\\.(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$"); @@ -100,7 +102,7 @@ public static List discoverSources(File folder, List for (File subdir : subdirs) { if (subdir.getName().startsWith(".")) continue; - if (looksLikeSession(subdir)) { + if (looksLikeSession(subdir, channelFile, windowTombstone)) { // Flat: // - embedded gpufl and single-pass // `gpufl trace` write the session directly under the watched dir. if (!isSettled(subdir)) { @@ -116,7 +118,7 @@ public static List discoverSources(File folder, List Arrays.sort(grandchildren, Comparator.comparing(File::getName)); for (File leaf : grandchildren) { if (leaf.getName().startsWith(".")) continue; - if (looksLikeSession(leaf) && !isSettled(leaf)) { + if (looksLikeSession(leaf, channelFile, windowTombstone) && !isSettled(leaf)) { result.add(new DiscoveredSession(subdir, leaf.getName(), types)); } } @@ -131,15 +133,16 @@ private static boolean isSettled(File sessionDir) { || new File(sessionDir, FAILED_MARKER).exists(); } - // A directory is a session if it directly contains a known channel's log file + // A directory is a session if it directly contains a tailed channel's log file // ([.N].log[.gz]) or window tombstone. - private static boolean looksLikeSession(File dir) { + private static boolean looksLikeSession( + File dir, Pattern channelFile, Pattern windowTombstone) { 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( + (channelFile.matcher(child.getName()).matches() + || windowTombstone.matcher( child.getName()).matches())) { return true; } diff --git a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java index e3c0f40..0d2978f 100644 --- a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java +++ b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java @@ -230,6 +230,28 @@ void servingOnlySessionUploadsAndCompletesOnceItsLockIsReleased( } } + @Test + void agentNotTailingServingNeverAdoptsServingOnlySession(@TempDir Path root) + throws Exception { + Path sess = Files.createDirectories(root.resolve("serving-sess")); + Files.write(sess.resolve("serving.1.log.gz"), gzip("{\"type\":\"serving_request\"}\n")); + Files.writeString(sess.resolve(".gpufl-window.serving.1.json"), "{}"); + + 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 byte[] gzip(String text) throws IOException { var out = new ByteArrayOutputStream(); try (var gz = new GZIPOutputStream(out)) { From 948a6e13780696bfe296bed036dc7d4fd637c6e7 Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Tue, 27 Oct 2026 16:43:59 -0700 Subject: [PATCH 3/4] Tail every channel a producer publishes, not a fixed list The agent is meant to be a channel-agnostic transport: the native client, the Python serving writer and future producers write the session layout, and the agent ships whatever channels they publish, with no agent or client change for each new producer. Discovery now finds a session's channels. A name is a channel once the directory holds a window sidecar for it (JSON carrying the payload size and CRC32 the tailer verifies), which also stays as the ACK tombstone. A bare *.log file never makes a channel, so an application log next to the windows, which could hold prompts, is never read or uploaded. The four native channels keep their legacy recognition by bare window files. Names follow the backend's [A-Za-z0-9_-]{1,64} rule, and a session is capped at 16 channels so a broken producer cannot spawn unbounded tailers; a session over the cap is never reported complete, is marked .failed and is not pruned. --log-types becomes an optional filter, empty by default. When set, only those channels are tailed and a directory is recognized only by them, so a filtered agent still never adopts a directory it cannot upload. A channel whose first window appears while the session is live gets a tailer on the next scan, and the last tailer to finish rescans once more before session-complete, so a channel published just before the producer finished is not missed. --- .../java/com/gpuflight/agent/GpuflAgent.java | 10 +- .../gpuflight/agent/config/ConfigLoader.java | 11 +- .../agent/model/DiscoveredSession.java | 17 +- .../agent/model/LogSourceConfig.java | 9 +- .../agent/service/SessionWatcher.java | 113 +++++--- .../agent/service/TailerManager.java | 168 ++++++++--- .../java/com/gpuflight/agent/MainTest.java | 4 +- .../agent/model/LogSourceConfigTest.java | 2 +- .../agent/service/SessionWatcherTest.java | 4 +- .../agent/service/TailerManagerTest.java | 263 ++++++++++++++---- 10 files changed, 451 insertions(+), 150 deletions(-) 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 fa302db..65b4ae1 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -17,11 +17,10 @@ public class ConfigLoader { private static final Logger log = LoggerFactory.getLogger(ConfigLoader.class); - /** Every channel the agent knows, tailed when none are configured. "serving" carries - * the records the Python serving writer emits, and its session directories hold no - * native channel at all. */ - public static final List DEFAULT_LOG_TYPES = - List.of("device", "scope", "system", "sass", "serving"); + /** 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); @@ -243,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,serving + --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 727b7f1..d320d60 100644 --- a/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java +++ b/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java @@ -9,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 .}), @@ -21,9 +22,9 @@ public record LogSourceConfig( ) { public LogSourceConfig { if (folder == null) folder = "."; - // Tail every known channel by default: "sass" carries the SASS-disassembly / - // source-content artifacts split out of device.log, and "serving" the LLM-serving - // records; either would otherwise miss live upload. + // 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 53e1da1..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,18 +10,37 @@ 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 java.util.stream.Collectors; 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); + /** 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 * re-tails or re-signals an already-finished session. */ @@ -60,21 +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. - String channels = types.stream().map(Pattern::quote).collect(Collectors.joining("|")); - Pattern channelFile = - Pattern.compile("^(" + channels + ")(?:\\.\\d+)?\\.log(?:\\.gz)?$"); - Pattern windowTombstone = - Pattern.compile("^\\.gpufl-window\\.(" + channels + ")\\.[1-9][0-9]*\\.json$"); + List filter = logTypes == null ? ConfigLoader.DEFAULT_LOG_TYPES : logTypes; // Legacy-format warning Pattern legacyTop = Pattern.compile( @@ -102,12 +122,13 @@ public static List discoverSources(File folder, List for (File subdir : subdirs) { if (subdir.getName().startsWith(".")) continue; - if (looksLikeSession(subdir, channelFile, windowTombstone)) { + // 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 @@ -117,36 +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, channelFile, windowTombstone) && !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 tailed channel's log file - // ([.N].log[.gz]) or window tombstone. - private static boolean looksLikeSession( - File dir, Pattern channelFile, Pattern windowTombstone) { - File[] inside = dir.listFiles(); - if (inside == null) return false; - for (File child : inside) { - if (child.isFile() && - (channelFile.matcher(child.getName()).matches() - || windowTombstone.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 dafa6de..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", "serving"), + 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 c2291d3..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", "serving"), 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 0d2978f..f5a660f 100644 --- a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java +++ b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java @@ -26,13 +26,12 @@ 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.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.CRC32; import java.util.zip.GZIPOutputStream; @@ -164,69 +163,133 @@ void servingOnlySessionUploadsAndCompletesOnceItsLockIsReleased( // 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")); - byte[] gz = gzip("{\"type\":\"serving_request\",\"session_id\":\"serving-sess\"}\n"); - Files.write(sess.resolve("serving.1.log.gz"), gz); - CRC32 crc = new CRC32(); - crc.update(gz); - Files.writeString(sess.resolve(".gpufl-window.serving.1.json"), - JsonSettings.MAPPER.writeValueAsString(new WindowMetadata( - 1, "transport_window", "11111111-2222-4333-8444-555555555555", - "serving-sess", "serving", 1, 10, 20, 30, - "serving.1.log.gz", gz.length, crc.getValue()))); + publishWindow(sess, "serving", 1); Path lock = Files.writeString(sess.resolve(SessionOwnership.LOCK_FILE), ""); - // Fake backend: acknowledges the window identity and the completion signal. - List channels = new CopyOnWriteArrayList<>(); - AtomicInteger completions = new AtomicInteger(); - CountDownLatch uploaded = new CountDownLatch(1); - HttpServer backend = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); - backend.createContext("/api/v1/events/", exchange -> { - exchange.getRequestBody().readAllBytes(); - String path = exchange.getRequestURI().getPath(); - String body = "{}"; - if (path.endsWith("/stream")) { - channels.add(exchange.getRequestHeaders().getFirst("X-GpuFlight-Window-Channel")); - body = "{\"window_acknowledged\":true}"; - uploaded.countDown(); - } else if (path.endsWith("/session-complete")) { - completions.incrementAndGet(); - } - byte[] bytes = body.getBytes(StandardCharsets.UTF_8); - exchange.sendResponseHeaders(200, bytes.length); - exchange.getResponseBody().write(bytes); - exchange.close(); - }); - backend.start(); - try { - HttpConfig http = new HttpConfig( - "http://127.0.0.1:" + backend.getAddress().getPort(), "v1", null, 5); + try (FakeBackend backend = new FakeBackend()) { ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); - TailerManager manager = new TailerManager(ex, new HttpPublisher(http), - new CursorManager(new File(root.toFile(), "cursor.json")), - new LinkedBlockingQueue<>(), null, StreamUploadSettings.from(http), - "gpu-trace"); + TailerManager manager = backend.managerIn(ex, root); try (FileChannel writer = FileChannel.open( lock, StandardOpenOption.READ, StandardOpenOption.WRITE); FileLock ignored = writer.lock()) { - List found = SessionWatcher.discoverSources( - root.toFile(), ConfigLoader.DEFAULT_LOG_TYPES); - assertEquals(List.of("serving-sess"), - found.stream().map(DiscoveredSession::sessionId).toList()); - manager.spawnSessionTailers(found.get(0)); + // No configuration: the serving channel is found, not listed. + manager.spawnSessionTailers(discoverOnly(root)); // A finished window uploads while the writer still holds the lock... - assertTrue(uploaded.await(10, TimeUnit.SECONDS), "serving window not uploaded"); - assertEquals(List.of("serving"), channels); - assertEquals(0, completions.get()); + 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(1, completions.get()); + assertEquals(List.of("serving", COMPLETE), backend.events); assertTrue(new File(sess.toFile(), SessionWatcher.UPLOADED_MARKER).exists()); - } finally { - backend.stop(0); + } + } + + @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"); } } @@ -234,8 +297,7 @@ void servingOnlySessionUploadsAndCompletesOnceItsLockIsReleased( void agentNotTailingServingNeverAdoptsServingOnlySession(@TempDir Path root) throws Exception { Path sess = Files.createDirectories(root.resolve("serving-sess")); - Files.write(sess.resolve("serving.1.log.gz"), gzip("{\"type\":\"serving_request\"}\n")); - Files.writeString(sess.resolve(".gpufl-window.serving.1.json"), "{}"); + publishWindow(sess, "serving", 1); CompletionPublisher publisher = new CompletionPublisher(); ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); @@ -252,6 +314,101 @@ void agentNotTailingServingNeverAdoptsServingOnlySession(@TempDir Path root) 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)) { From 547869fa7bc20e33656023da3ce15237bee8792b Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Tue, 27 Oct 2026 16:44:11 -0700 Subject: [PATCH 4/4] Document the session layout producers write With the agent shipping any channel it finds, the on-disk layout is the contract between it and every producer. Write it down: session directories, the lock file and how the agent probes it, .tmp/ and what its removal means, window naming and numbering, the sidecar fields and CRC32, ACK tombstones, the markers, completion, record contents and the lifecycle rows. Each statement is taken from the agent code or the native writer, and what the agent only tolerates is marked so a new producer does not depend on it. Link it from the README, which now describes channel discovery and the --log-types filter. --- README.md | 21 +++- docs/session-layout.md | 216 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+), 1 deletion(-) create mode 100644 docs/session-layout.md 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.