Agent drain robustness

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,14 @@ 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
+        // earlier run. 0 = standalone agent: no cutoff, upload everything.
+        long sinceMs = ConfigLoader.parseIgnorePreexisting(args, env)
+                ? ManagementFactory.getRuntimeMXBean().getStartTime()
+                : 0L;
 
         resolveWatchedFolders();
 
@@ -78,18 +88,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<DiscoveredSession> 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) {
@@ -113,8 +132,13 @@ 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);
+            awaitDrainThenExit(watchedFolders.keySet(), tailerManager, sinceMs);
             log.info("all sessions drained - exiting");
             shutdown();
             return;
@@ -154,27 +178,36 @@ private void shutdown() {
         try { if (publisher != null) publisher.close(); } catch (Exception ignored) {}
     }
 
-    void awaitDrainThenExit(Collection<File> folders, TailerManager tailers) {
+    void awaitDrainThenExit(Collection<File> folders, TailerManager tailers, long sinceMs) {
         int clean = 0;
         while (true) {
             if (!Delays.sleep(Delays.DRAIN_CHECK_POLL)) break;
             // Gate on "a session was discovered" (cumulative), not on observing the live
             // .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<File> folders) {
+    static boolean anyActiveSession(Collection<File> 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;

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 <idx> 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 <folder>/<sessionId>/.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/<channel>.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

src/main/java/com/gpuflight/agent/config/ConfigLoader.java

@@ -149,6 +149,38 @@ public static boolean parseExitWhenDrained(String[] args, Map<String, String> 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<String, String> 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<String, String> env) {
+        String v = resolve(args, "prune-failed", "GPUFL_AGENT_PRUNE_FAILED", null, 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<String, String> 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();
     }

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<String> warnedLegacyFolders =
             ConcurrentHashMap.newKeySet();
 
@@ -84,19 +94,48 @@ public static List<DiscoveredSession> discoverSources(File folder, List<String>
 
         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: <folder>/<session_id>/ - 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: <folder>/<group>/<session_id>/ - a multi-pass
+            // run nests its passes under a "run-<app>-<analysis_id>" 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;
+    }
 }