From 11ff296a042bbaeb08d84efb0a7441d328f1e538 Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Tue, 30 Jun 2026 11:16:28 -0700 Subject: [PATCH 1/2] feat(agent): scope --upload to one run + settle/prune finished sessions --- .../java/com/gpuflight/agent/GpuflAgent.java | 43 +++++-- .../java/com/gpuflight/agent/LogTailer.java | 35 +++++- .../gpuflight/agent/config/ConfigLoader.java | 22 ++++ .../agent/service/SessionWatcher.java | 59 ++++++++-- .../agent/service/TailerManager.java | 55 ++++++++- .../java/com/gpuflight/agent/util/Delays.java | 10 ++ .../java/com/gpuflight/agent/MainTest.java | 29 ++++- .../agent/service/SessionWatcherTest.java | 108 ++++++++++++++++++ .../agent/service/TailerManagerTest.java | 94 +++++++++++++++ 9 files changed, 427 insertions(+), 28 deletions(-) create mode 100644 src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java create mode 100644 src/test/java/com/gpuflight/agent/service/TailerManagerTest.java diff --git a/src/main/java/com/gpuflight/agent/GpuflAgent.java b/src/main/java/com/gpuflight/agent/GpuflAgent.java index f1af78a..afba39d 100644 --- a/src/main/java/com/gpuflight/agent/GpuflAgent.java +++ b/src/main/java/com/gpuflight/agent/GpuflAgent.java @@ -17,6 +17,7 @@ import org.slf4j.LoggerFactory; import java.io.File; +import java.lang.management.ManagementFactory; import java.nio.file.Path; import java.util.Collection; import java.util.LinkedHashMap; @@ -26,6 +27,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; +import java.util.function.Consumer; public class GpuflAgent { @@ -51,6 +53,13 @@ public void start() throws Exception { log.info("Publisher: {}", config.publisher().getClass().getSimpleName()); boolean exitWhenDrained = ConfigLoader.parseExitWhenDrained(args, env); + // A launcher-spawned --upload agent uploads only THIS run's sessions. The JVM + // start time predates the target's session (the launcher spawns the agent + // before it forks the target), so any session dir older than it is from an + // earlier run. 0 = standalone agent: no cutoff, upload everything. + long sinceMs = ConfigLoader.parseIgnorePreexisting(args, env) + ? ManagementFactory.getRuntimeMXBean().getStartTime() + : 0L; resolveWatchedFolders(); @@ -78,18 +87,27 @@ public void start() throws Exception { tailerManager = new TailerManager(executor, publisher, cursorMgr, consumedFilesQueue, deduplicator, streamUploadSettings, topicPrefix); + tailerManager.setPruneFailed(ConfigLoader.parsePruneFailed(args, env)); + + // A launcher-spawned --upload agent uploads only sessions newer than sinceMs + // (this run); a standalone agent (sinceMs == 0) uploads everything. + Consumer spawn = s -> { + if (sinceMs > 0 && new File(s.folder(), s.sessionId()).lastModified() < sinceMs) { + return; // belongs to an earlier run - not ours to upload + } + tailerManager.spawnSessionTailers(s); + }; // Initial discovery + spawn for (var entry : watchedFolders.entrySet()) { for (DiscoveredSession s : SessionWatcher.discoverSources(entry.getKey(), entry.getValue())) { - tailerManager.spawnSessionTailers(s); + spawn.accept(s); } } // Start watchers for (var entry : watchedFolders.entrySet()) { - new SessionWatcher(entry.getKey(), entry.getValue(), tailerManager::spawnSessionTailers) - .start(executor); + new SessionWatcher(entry.getKey(), entry.getValue(), spawn).start(executor); } if (config.archiver() != null) { @@ -114,7 +132,7 @@ public void start() throws Exception { if (exitWhenDrained) { log.info("exit-when-drained mode enabled"); - awaitDrainThenExit(watchedFolders.keySet(), tailerManager); + awaitDrainThenExit(watchedFolders.keySet(), tailerManager, sinceMs); log.info("all sessions drained - exiting"); shutdown(); return; @@ -154,7 +172,7 @@ private void shutdown() { try { if (publisher != null) publisher.close(); } catch (Exception ignored) {} } - void awaitDrainThenExit(Collection folders, TailerManager tailers) { + void awaitDrainThenExit(Collection folders, TailerManager tailers, long sinceMs) { int clean = 0; while (true) { if (!Delays.sleep(Delays.DRAIN_CHECK_POLL)) break; @@ -162,19 +180,28 @@ void awaitDrainThenExit(Collection folders, TailerManager tailers) { // .tmp/ marker: a short trace finalizes that marker between our 1s polls, which // left the old sawActive gate spinning forever even after the upload drained. boolean started = tailers.hasStartedAnySession(); - boolean idle = tailers.getActiveTailers().get() == 0 && !anyActiveSession(folders); + boolean idle = tailers.getActiveTailers().get() == 0 && !anyActiveSession(folders, sinceMs); clean = (started && idle) ? clean + 1 : 0; if (clean >= 2) return; } } - static boolean anyActiveSession(Collection folders) { + static boolean anyActiveSession(Collection folders, long sinceMs) { for (File folder : folders) { File[] subdirs = folder.listFiles(File::isDirectory); if (subdirs == null) continue; for (File subdir : subdirs) { if (subdir.getName().startsWith(".")) continue; - if (new File(subdir, ".tmp").isDirectory()) return true; + // Under --upload scope, a session older than this run belongs to another + // run; its .tmp/ must not keep this run's agent from exiting. + if (sinceMs > 0 && subdir.lastModified() < sinceMs) continue; + File tmp = new File(subdir, ".tmp"); + if (!tmp.isDirectory()) continue; + // A frozen .tmp/ (crashed/killed client) is not active - the same stale + // grace the tailer uses, so an orphaned .tmp/ can't block the drain. + if (System.currentTimeMillis() - LogTailer.newestMtime(tmp) + > Delays.STALE_TMP_GRACE.toMillis()) continue; + return true; } } return false; diff --git a/src/main/java/com/gpuflight/agent/LogTailer.java b/src/main/java/com/gpuflight/agent/LogTailer.java index 2d1431a..7af0cc8 100644 --- a/src/main/java/com/gpuflight/agent/LogTailer.java +++ b/src/main/java/com/gpuflight/agent/LogTailer.java @@ -215,7 +215,7 @@ private static long headSignature(File f) { * dir is gone - the client removes it once every channel has closed - and no * further window has appeared. No partial-file tailing, no rotation-shift guess. */ - public void tail(Publisher publisher) { + public boolean tail(Publisher publisher) { var cursor = cursorMgr.get(streamKey); // fileIndex = window index in progress (1-based; 0 = none yet); offset = // bytes already sent within it (mid-window resume after a crash). @@ -242,7 +242,18 @@ public void tail(Publisher publisher) { } // Window not published yet. Is the session still writing? - if (sessionTmpDir().exists()) { + File tmp = sessionTmpDir(); + if (tmp.exists()) { + // A live client keeps appending to .tmp/ (system sampling alone + // advances its mtime); a client that crashed or was killed leaves + // .tmp/ frozen and never removed. Once it has sat unchanged past the + // grace, treat it as orphaned and finish - one dead session must not + // strand the agent's drain (and the 60s --upload cap) forever. + if (System.currentTimeMillis() - newestMtime(tmp) > Delays.STALE_TMP_GRACE.toMillis()) { + System.out.println("[" + logType + "] Session finished (.tmp stale > " + + Delays.STALE_TMP_GRACE.toSeconds() + "s - client gone; last sent " + (idx - 1) + ")."); + return true; // orphaned: producer never removed .tmp/ + } if (!Delays.sleep(Delays.LOG_TAILER_POLL)) break; continue; } @@ -254,8 +265,9 @@ public void tail(Publisher publisher) { if (resolveRotated(idx) != null) continue; System.out.println("[" + logType + "] Session finished (.tmp gone, no window " + idx + "; last sent " + (idx - 1) + ")."); - return; + return false; // clean: producer closed and removed .tmp/ } + return false; // interrupted before finishing } /** The client's per-session working dir {@code //.tmp}: active @@ -265,6 +277,23 @@ private File sessionTmpDir() { return new File(sessionDir(), ".tmp"); } + /** Newest last-modified time of {@code dir} and its immediate entries. The + * client appends to {@code .tmp/.log} while writing, so this advances + * as long as the session is live and freezes once the client is gone. + * Package-private so the drain-exit gate (GpuflAgent) shares the tailer's + * staleness notion. */ + static long newestMtime(File dir) { + long newest = dir.lastModified(); + File[] kids = dir.listFiles(); + if (kids != null) { + for (File k : kids) { + long m = k.lastModified(); + if (m > newest) newest = m; + } + } + return newest; + } + /** * Drain a compressed rotated file from uncompressed {@code startOffset}, * publishing each line. Returns {@code -1} when the file is fully drained diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java index 13b5071..cb1c00b 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -149,6 +149,28 @@ public static boolean parseExitWhenDrained(String[] args, Map en return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); } + /** + * Scope a launcher-spawned {@code --upload} agent to THIS run: ignore every + * session that already exists when the agent starts, uploading only sessions + * that appear afterwards. A standalone agent (no flag) keeps scanning all + * sessions so "watch a parent, run many times" still works. + */ + public static boolean parseIgnorePreexisting(String[] args, Map env) { + String v = resolve(args, "ignore-preexisting", "GPUFL_AGENT_IGNORE_PREEXISTING", null, env); + return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); + } + + /** + * Delete a session finished off an orphaned {@code .tmp/} (producer crashed or + * was killed) instead of leaving a {@code .failed} marker - its windows are + * already uploaded, so the local dir is just leftover. Off by default (keep the + * marker so the failure can be inspected). + */ + public static boolean parsePruneFailed(String[] args, Map env) { + String v = resolve(args, "prune-failed", "GPUFL_AGENT_PRUNE_FAILED", null, env); + return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); + } + public static String buildArchiveKey(String prefix, Path path) { return prefix + path.toFile().getName(); } diff --git a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java index f308d11..6e7dea9 100644 --- a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -23,6 +23,16 @@ public class SessionWatcher { private static final Pattern CHANNEL_FILE_PATTERN = 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. */ + public static final String UPLOADED_MARKER = ".uploaded"; + + /** Marker for a session that finished off an orphaned {@code .tmp/} (producer + * crashed or was killed and never removed it). Discovery skips it like + * UPLOADED_MARKER; an agent with pruning on deletes such a session instead. */ + public static final String FAILED_MARKER = ".failed"; + private static final Set warnedLegacyFolders = ConcurrentHashMap.newKeySet(); @@ -84,19 +94,48 @@ public static List discoverSources(File folder, List for (File subdir : subdirs) { if (subdir.getName().startsWith(".")) continue; - File[] inside = subdir.listFiles(); - if (inside == null) continue; - boolean looksLikeSession = false; - for (File child : inside) { - if (child.isFile() && - CHANNEL_FILE_PATTERN.matcher(child.getName()).matches()) { - looksLikeSession = true; - break; + if (looksLikeSession(subdir)) { + // 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)); + } + continue; + } + // One grouping level: /// - a multi-pass + // run nests its passes under a "run--" folder. Descend + // one level (and no further) so each pass session is still discovered. + File[] grandchildren = subdir.listFiles(File::isDirectory); + if (grandchildren == null) continue; + Arrays.sort(grandchildren, Comparator.comparing(File::getName)); + for (File leaf : grandchildren) { + if (leaf.getName().startsWith(".")) continue; + if (looksLikeSession(leaf) && !isSettled(leaf)) { + result.add(new DiscoveredSession(subdir, leaf.getName(), types)); } } - if (!looksLikeSession) continue; - result.add(new DiscoveredSession(folder, subdir.getName(), types)); } return result; } + + // A settled session (fully uploaded, or finished+marked failed) is skipped by a + // re-scan so it is never re-tailed or re-signalled. + private static boolean isSettled(File sessionDir) { + return new File(sessionDir, UPLOADED_MARKER).exists() + || new File(sessionDir, FAILED_MARKER).exists(); + } + + // A directory is a session if it directly contains a channel log file + // (device/scope/system/sass[.N].log[.gz]). + private static boolean looksLikeSession(File dir) { + File[] inside = dir.listFiles(); + if (inside == null) return false; + for (File child : inside) { + if (child.isFile() && + CHANNEL_FILE_PATTERN.matcher(child.getName()).matches()) { + 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 d279f37..e9a6430 100644 --- a/src/main/java/com/gpuflight/agent/service/TailerManager.java +++ b/src/main/java/com/gpuflight/agent/service/TailerManager.java @@ -10,12 +10,15 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.File; +import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; public class TailerManager { @@ -32,6 +35,7 @@ public class TailerManager { private final Set startedSessions = ConcurrentHashMap.newKeySet(); private final AtomicInteger activeTailers = new AtomicInteger(0); + private volatile boolean pruneFailed = false; public TailerManager(ExecutorService executor, Publisher publisher, @@ -55,7 +59,9 @@ public void spawnSessionTailers(DiscoveredSession session) { 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(); @@ -65,11 +71,15 @@ public void spawnSessionTailers(DiscoveredSession session) { var tailer = new LogTailer(session.folder(), session.sessionId(), type, topicPrefix, cursorMgr, consumedFilesQueue, dedup, streamUploadSettings); - tailer.tail(publisher); + if (tailer.tail(publisher)) orphaned.set(true); // finished off a stale .tmp/ if (!Thread.currentThread().isInterrupted() - && remaining.decrementAndGet() == 0 - && streamUploadSettings.enabled()) { - signalSessionComplete(publisher, sid); + && remaining.decrementAndGet() == 0) { + // 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()) { + signalSessionComplete(publisher, sid); + } + settleSession(sessionDir, orphaned.get()); } } finally { activeTailers.decrementAndGet(); @@ -78,6 +88,37 @@ public void spawnSessionTailers(DiscoveredSession session) { } } + /** 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) { + // Windows are already uploaded; drop the local orphan. A locked .tmp/ can + // block deletion - fall back to marking it failed. + if (deleteRecursively(sessionDir)) { + log.info("pruned orphaned session {}", sessionDir.getName()); + return; + } + log.warn("could not prune orphaned session {} - marking failed instead", sessionDir.getName()); + } + String marker = orphaned ? SessionWatcher.FAILED_MARKER : SessionWatcher.UPLOADED_MARKER; + try { + Files.writeString(new File(sessionDir, marker).toPath(), ""); + } catch (Exception e) { + log.warn("could not write {} marker in {}: {}", marker, sessionDir, e.getMessage()); + } + } + + private static boolean deleteRecursively(File f) { + File[] kids = f.listFiles(); + if (kids != null) { + for (File k : kids) deleteRecursively(k); + } + return f.delete(); + } + private void signalSessionComplete(Publisher publisher, String sessionId) { long delayMs = Delays.SESSION_COMPLETE_RETRY.toMillis(); for (int attempt = 1; attempt <= 3; attempt++) { @@ -96,6 +137,12 @@ public AtomicInteger getActiveTailers() { return activeTailers; } + /** Enable deleting orphaned sessions (finished off a stale .tmp/) instead of + * leaving a {@code .failed} marker - their windows are already uploaded. */ + public void setPruneFailed(boolean prune) { + this.pruneFailed = prune; + } + /** True once any session has been discovered and tailed. Cumulative - never cleared. */ public boolean hasStartedAnySession() { return !startedSessions.isEmpty(); diff --git a/src/main/java/com/gpuflight/agent/util/Delays.java b/src/main/java/com/gpuflight/agent/util/Delays.java index 5856651..d7143b7 100644 --- a/src/main/java/com/gpuflight/agent/util/Delays.java +++ b/src/main/java/com/gpuflight/agent/util/Delays.java @@ -32,6 +32,16 @@ private Delays() {} */ public static final Duration SESSION_END_GRACE_PERIOD = Duration.ofMillis(4500); + /** + * How long a session's {@code .tmp/} working dir may sit unchanged - no new + * window, no write - before the tailer treats it as ORPHANED (left behind by a + * client that crashed or was killed and never removed it) and finishes anyway. + * A live client keeps writing {@code .tmp/} - system sampling alone advances it + * well inside this window - so this only fires on genuinely dead sessions, and + * stops one of them from stranding the agent's drain forever. + */ + public static final Duration STALE_TMP_GRACE = Duration.ofSeconds(30); + /** * Sleep for the specified duration. * @return true if the sleep finished normally, false if it was interrupted. diff --git a/src/test/java/com/gpuflight/agent/MainTest.java b/src/test/java/com/gpuflight/agent/MainTest.java index d59bf53..c95e2e3 100644 --- a/src/test/java/com/gpuflight/agent/MainTest.java +++ b/src/test/java/com/gpuflight/agent/MainTest.java @@ -481,7 +481,7 @@ void parseExitWhenDrained_offWhenAbsentOrFalsey() { void anyActiveSession_trueWhenSessionStillWriting(@TempDir Path dir) throws IOException { File folder = dir.toFile(); Files.createDirectories(new File(new File(folder, "sess-1"), ".tmp").toPath()); // .tmp = active - assertTrue(GpuflAgent.anyActiveSession(List.of(folder))); + assertTrue(GpuflAgent.anyActiveSession(List.of(folder), 0L)); } @Test @@ -490,8 +490,31 @@ void anyActiveSession_falseWhenDrainedOrEmpty(@TempDir Path dir) throws IOExcept File session = new File(folder, "sess-1"); Files.createDirectories(session.toPath()); Files.writeString(new File(session, "device.1.log.gz").toPath(), "x"); // finished window, no .tmp - assertFalse(GpuflAgent.anyActiveSession(List.of(folder))); - assertFalse(GpuflAgent.anyActiveSession(List.of(new File(folder, "missing")))); // non-existent folder + assertFalse(GpuflAgent.anyActiveSession(List.of(folder), 0L)); + assertFalse(GpuflAgent.anyActiveSession(List.of(new File(folder, "missing")), 0L)); // non-existent folder + } + + @Test + void anyActiveSession_ignoresSessionsOlderThanCutoff(@TempDir Path dir) throws IOException { + File folder = dir.toFile(); + File active = new File(folder, "sess-1"); + Files.createDirectories(new File(active, ".tmp").toPath()); // .tmp = active session + // Backdate it below the cutoff: a --upload agent (sinceMs > 0) treats it as an + // earlier run's session and must not let its .tmp/ block this run's exit. + long cutoff = System.currentTimeMillis(); + active.setLastModified(cutoff - 60_000L); + assertFalse(GpuflAgent.anyActiveSession(List.of(folder), cutoff)); + assertTrue(GpuflAgent.anyActiveSession(List.of(folder), 0L)); // no cutoff -> still active + } + + @Test + void anyActiveSession_falseWhenTmpIsStale(@TempDir Path dir) throws IOException { + File folder = dir.toFile(); + File tmp = new File(new File(folder, "sess-1"), ".tmp"); + Files.createDirectories(tmp.toPath()); + // An orphaned .tmp/ frozen well past the stale grace must not read as active. + tmp.setLastModified(System.currentTimeMillis() - 120_000L); // 120s >> 30s grace + assertFalse(GpuflAgent.anyActiveSession(List.of(folder), 0L)); } // ---- printUsage() — smoke test ---- diff --git a/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java b/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java new file mode 100644 index 0000000..d679a88 --- /dev/null +++ b/src/test/java/com/gpuflight/agent/service/SessionWatcherTest.java @@ -0,0 +1,108 @@ +package com.gpuflight.agent.service; + +import com.gpuflight.agent.model.DiscoveredSession; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * SessionWatcher discovers sessions both flat under the watched folder + * ({@code //} - embedded gpufl and single-pass `gpufl trace`) + * and nested one grouping level deep + * ({@code /run--//} - a multi-pass run). + * Both must be found so "watch a parent, run many times" keeps working across + * single- and multi-pass runs. + */ +class SessionWatcherTest { + + private static final List TYPES = List.of("device", "scope", "system", "sass"); + + private static void session(Path dir, String channelFile) throws IOException { + Files.createDirectories(dir); + Files.writeString(dir.resolve(channelFile), "{}\n"); + } + + @Test + void discoversFlatSessionDirectlyUnderFolder(@TempDir Path root) throws IOException { + session(root.resolve("sess-aaa"), "device.log"); + + List found = SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertEquals(1, found.size()); + assertEquals("sess-aaa", found.get(0).sessionId()); + assertEquals(root.toFile(), found.get(0).folder()); + } + + @Test + void discoversNestedSessionsUnderRunFolder(@TempDir Path root) throws IOException { + Path group = root.resolve("run-myapp-abc12345"); + session(group.resolve("sess-pass0"), "device.log"); + session(group.resolve("sess-pass1"), "sass.1.log.gz"); + + List found = SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertEquals(2, found.size()); + assertEquals(List.of("sess-pass0", "sess-pass1"), + found.stream().map(DiscoveredSession::sessionId).sorted().toList()); + // / must resolve to the real pass directory. + for (DiscoveredSession s : found) { + assertTrue(new File(s.folder(), s.sessionId()).isDirectory()); + } + } + + @Test + void discoversFlatAndNestedTogether(@TempDir Path root) throws IOException { + session(root.resolve("flat-sess"), "system.log"); + session(root.resolve("run-app-deadbeef").resolve("grouped-sess"), "device.2.log"); + + List found = SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertEquals(2, found.size()); + assertEquals(List.of("flat-sess", "grouped-sess"), + found.stream().map(DiscoveredSession::sessionId).sorted().toList()); + } + + @Test + void ignoresDirsWithoutChannelFilesAndDotDirs(@TempDir Path root) throws IOException { + Files.createDirectories(root.resolve("not-a-session").resolve("random")); + Files.writeString(root.resolve("not-a-session").resolve("notes.txt"), "x"); + session(root.resolve(".hidden"), "device.log"); // dot dir is skipped + session(root.resolve("real"), "scope.log"); + + List found = SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertEquals(1, found.size()); + assertEquals("real", found.get(0).sessionId()); + } + + @Test + void doesNotRecurseBeyondOneGroupingLevel(@TempDir Path root) throws IOException { + // /a/b// is two grouping levels deep - out of scope. + session(root.resolve("a").resolve("b").resolve("too-deep"), "device.log"); + + List found = SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertTrue(found.isEmpty()); + } + + @Test + void skipsSessionsMarkedUploadedOrFailed(@TempDir Path root) throws IOException { + session(root.resolve("done"), "device.log"); + Files.writeString(root.resolve("done").resolve(SessionWatcher.UPLOADED_MARKER), ""); + session(root.resolve("failed"), "device.log"); + Files.writeString(root.resolve("failed").resolve(SessionWatcher.FAILED_MARKER), ""); + session(root.resolve("pending"), "device.log"); + + List found = SessionWatcher.discoverSources(root.toFile(), TYPES); + + assertEquals(1, found.size()); + assertEquals("pending", found.get(0).sessionId()); + } +} diff --git a/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java new file mode 100644 index 0000000..ff4407d --- /dev/null +++ b/src/test/java/com/gpuflight/agent/service/TailerManagerTest.java @@ -0,0 +1,94 @@ +package com.gpuflight.agent.service; + +import com.gpuflight.agent.CursorManager; +import com.gpuflight.agent.config.StreamUploadSettings; +import com.gpuflight.agent.model.DiscoveredSession; +import com.gpuflight.agent.model.LogWrapper; +import com.gpuflight.agent.publisher.Publisher; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Once a session's tailers finish, the manager "settles" it so a re-scan skips it: + * a producer that closed cleanly (.tmp/ removed) → {@code .uploaded}; one left with a + * stale/orphaned .tmp/ (producer crashed) → {@code .failed}, or deleted when pruning + * is enabled (its windows are already uploaded). + */ +class TailerManagerTest { + + /** Accepts everything via the legacy path; stream/session-complete use interface defaults. */ + static final class NoopPublisher implements Publisher { + @Override public boolean publish(String topic, String key, LogWrapper log) { return true; } + @Override public void close() {} + } + + private TailerManager managerIn(ExecutorService ex, Path root) { + return new TailerManager(ex, new NoopPublisher(), + new CursorManager(new File(root.toFile(), "cursor.json")), + new LinkedBlockingQueue<>(), null, StreamUploadSettings.DISABLED, "gpu-trace"); + } + + private static void runToCompletion(ExecutorService ex) throws InterruptedException { + ex.shutdown(); + assertTrue(ex.awaitTermination(20, TimeUnit.SECONDS), "tailers did not finish"); + } + + @Test + void cleanFinishMarksUploaded(@TempDir Path root) throws Exception { + File sess = new File(root.toFile(), "sess-clean"); + Files.createDirectories(sess.toPath()); + Files.writeString(new File(sess, "device.1.log").toPath(), "x\n"); // final window, no .tmp/ + + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + managerIn(ex, root).spawnSessionTailers( + new DiscoveredSession(root.toFile(), "sess-clean", List.of("device"))); + runToCompletion(ex); + + assertTrue(new File(sess, SessionWatcher.UPLOADED_MARKER).exists()); + assertFalse(new File(sess, SessionWatcher.FAILED_MARKER).exists()); + } + + @Test + void orphanedFinishMarksFailed(@TempDir Path root) throws Exception { + File sess = new File(root.toFile(), "sess-orphan"); + Files.createDirectories(new File(sess, ".tmp").toPath()); + Files.writeString(new File(sess, "device.1.log").toPath(), "x\n"); + new File(sess, ".tmp").setLastModified(System.currentTimeMillis() - 120_000L); // orphaned + + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + managerIn(ex, root).spawnSessionTailers( + new DiscoveredSession(root.toFile(), "sess-orphan", List.of("device"))); + runToCompletion(ex); + + assertTrue(new File(sess, SessionWatcher.FAILED_MARKER).exists()); + assertFalse(new File(sess, SessionWatcher.UPLOADED_MARKER).exists()); + } + + @Test + void orphanedFinishPrunesWhenEnabled(@TempDir Path root) throws Exception { + File sess = new File(root.toFile(), "sess-prune"); + Files.createDirectories(new File(sess, ".tmp").toPath()); + Files.writeString(new File(sess, "device.1.log").toPath(), "x\n"); + new File(sess, ".tmp").setLastModified(System.currentTimeMillis() - 120_000L); + + ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); + TailerManager m = managerIn(ex, root); + m.setPruneFailed(true); + m.spawnSessionTailers(new DiscoveredSession(root.toFile(), "sess-prune", List.of("device"))); + runToCompletion(ex); + + assertFalse(sess.exists(), "orphaned session should be pruned"); + } +} From 2ce8b76457176b167bd0a62edfaa305244b0a639 Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Tue, 30 Jun 2026 12:52:54 -0700 Subject: [PATCH 2/2] feat(agent): exit one-shot upload immediately when nothing to ship --- .../java/com/gpuflight/agent/GpuflAgent.java | 6 +++ .../gpuflight/agent/config/ConfigLoader.java | 10 ++++ .../agent/config/ConfigLoaderTest.java | 51 +++++++++++++++++++ 3 files changed, 67 insertions(+) create mode 100644 src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java diff --git a/src/main/java/com/gpuflight/agent/GpuflAgent.java b/src/main/java/com/gpuflight/agent/GpuflAgent.java index afba39d..395099f 100644 --- a/src/main/java/com/gpuflight/agent/GpuflAgent.java +++ b/src/main/java/com/gpuflight/agent/GpuflAgent.java @@ -53,6 +53,7 @@ public void start() throws Exception { log.info("Publisher: {}", config.publisher().getClass().getSimpleName()); boolean exitWhenDrained = ConfigLoader.parseExitWhenDrained(args, env); + boolean exitIfEmpty = ConfigLoader.parseExitIfEmpty(args, env); // A launcher-spawned --upload agent uploads only THIS run's sessions. The JVM // start time predates the target's session (the launcher spawns the agent // before it forks the target), so any session dir older than it is from an @@ -131,6 +132,11 @@ public void start() throws Exception { Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown)); if (exitWhenDrained) { + if (exitIfEmpty && !tailerManager.hasStartedAnySession()) { + log.info("nothing to upload - exiting"); + shutdown(); + return; + } log.info("exit-when-drained mode enabled"); awaitDrainThenExit(watchedFolders.keySet(), tailerManager, sinceMs); log.info("all sessions drained - exiting"); diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java index cb1c00b..0c3433d 100644 --- a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -171,6 +171,16 @@ public static boolean parsePruneFailed(String[] args, Map env) { return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); } + /** + * One-shot: exit right away if no session is found at startup (nothing to + * upload) instead of waiting for one to appear. Set for `gpufl upload`, NOT for + * trace/monitor whose session is created after the agent starts. + */ + public static boolean parseExitIfEmpty(String[] args, Map env) { + String v = resolve(args, "exit-if-empty", "GPUFL_AGENT_EXIT_IF_EMPTY", null, env); + return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); + } + public static String buildArchiveKey(String prefix, Path path) { return prefix + path.toFile().getName(); } diff --git a/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java b/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java new file mode 100644 index 0000000..cb3d93e --- /dev/null +++ b/src/test/java/com/gpuflight/agent/config/ConfigLoaderTest.java @@ -0,0 +1,51 @@ +package com.gpuflight.agent.config; + +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; + +class ConfigLoaderTest { + + private static final Map NO_ENV = Map.of(); + + @Test + void booleanFlags_offByDefault_onViaFlagOrEnv() { + // default off + assertFalse(ConfigLoader.parseExitWhenDrained(new String[]{}, NO_ENV)); + assertFalse(ConfigLoader.parseIgnorePreexisting(new String[]{}, NO_ENV)); + assertFalse(ConfigLoader.parsePruneFailed(new String[]{}, NO_ENV)); + assertFalse(ConfigLoader.parseExitIfEmpty(new String[]{}, NO_ENV)); + + // on via flag + assertTrue(ConfigLoader.parsePruneFailed(new String[]{"--prune-failed=1"}, NO_ENV)); + assertTrue(ConfigLoader.parseExitIfEmpty(new String[]{"--exit-if-empty=true"}, NO_ENV)); + + // on via env + assertTrue(ConfigLoader.parseExitWhenDrained(new String[]{}, Map.of("GPUFL_AGENT_EXIT_WHEN_DRAINED", "1"))); + assertTrue(ConfigLoader.parseIgnorePreexisting(new String[]{}, Map.of("GPUFL_AGENT_IGNORE_PREEXISTING", "1"))); + + // explicit false / 0 -> off + assertFalse(ConfigLoader.parsePruneFailed(new String[]{"--prune-failed=false"}, NO_ENV)); + assertFalse(ConfigLoader.parseExitIfEmpty(new String[]{}, Map.of("GPUFL_AGENT_EXIT_IF_EMPTY", "0"))); + } + + @Test + void resolve_flagBeatsEnvBeatsDefault() { + assertEquals("flag", + ConfigLoader.resolve(new String[]{"--host=flag"}, "host", "ENV", "def", Map.of("ENV", "e"))); + assertEquals("e", + ConfigLoader.resolve(new String[]{}, "host", "ENV", "def", Map.of("ENV", "e"))); + assertEquals("def", + ConfigLoader.resolve(new String[]{}, "host", "ENV", "def", NO_ENV)); + } + + @Test + void parseLogTypes_splitsTrimsAndDefaultsToNull() { + assertNull(ConfigLoader.parseLogTypes(null)); + assertNull(ConfigLoader.parseLogTypes(" ")); + assertEquals(List.of("a", "b"), ConfigLoader.parseLogTypes(" a , b ")); + } +}