From 9a27d4b46edb8015f8bdf7dec52001011aa5b1ba Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Fri, 19 Jun 2026 10:23:17 -0700 Subject: [PATCH 1/2] fix: tail rotated log windows in order and upload each window whole LogTailer assumed logrotate-style rotation (.1 = newest) with a live active file at the session root, but the client writes monotonic incrementing windows (..log[.gz], 1 = oldest) with the active file hidden in /.tmp/. Live-tailing a session large enough to rotate past one window dropped every window after the first (~50% kernel loss on large sessions; on-disk data was intact, only the live upload was truncated). Rewrite tail() to the window model: send each complete window whole in ascending index order and finish once the session's .tmp/ dir is gone (the client removes it after every channel closes) plus a short straggler grace. In stream mode a published .gz window is already the exact gzipped-NDJSON wire format, so upload it verbatim as one chunk (publishStreamGz) instead of decompressing and re-batching into <=1MB POSTs: the agent no longer holds a whole decompressed window and a large session uploads in a handful of chunks. The cursor advances immediately after the durable 2xx so a crash cannot re-send the whole window. Drop the dead RandomAccessFile-based readLineUtf8 left from the old tailer. --- .../java/com/gpuflight/agent/LogTailer.java | 431 ++-------- .../agent/publisher/HttpPublisher.java | 55 +- .../gpuflight/agent/publisher/Publisher.java | 11 + .../com/gpuflight/agent/LogTailerTest.java | 778 +++++------------- .../agent/publisher/HttpPublisherTest.java | 46 ++ 5 files changed, 369 insertions(+), 952 deletions(-) diff --git a/src/main/java/com/gpuflight/agent/LogTailer.java b/src/main/java/com/gpuflight/agent/LogTailer.java index bea17a4..d5d0007 100644 --- a/src/main/java/com/gpuflight/agent/LogTailer.java +++ b/src/main/java/com/gpuflight/agent/LogTailer.java @@ -13,7 +13,6 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; -import java.io.RandomAccessFile; import java.nio.charset.StandardCharsets; import java.net.InetAddress; import java.nio.file.Files; @@ -26,30 +25,19 @@ import java.util.zip.GZIPInputStream; /** - * Tails log files written by the C++ gpufl client. + * Tails the log files written by the C++ gpufl client and ships them to the backend. * - * v1.2 file naming (one LogTailer per session/channel pair): - * Active file : //.log - * Rotated files: //.1.log[.gz] (newest rotated) - * //.2.log[.gz] (older), ... + * The client publishes each rotation as a COMPLETE, immutable window file in the + * session dir: {@code //..log[.gz]}, where the + * index increments (1 = oldest, higher = newer). The live active file is written + * inside {@code /.tmp/} and is never tailed directly, so this tailer + * simply sends each finished window WHOLE, in increasing index order, and finishes + * when the session's {@code .tmp/} dir is gone (the client removes it once every + * channel has closed) and no further window has appeared. * - * Cursor fileIndex semantics: - * 0 -> currently tailing the active file - * N>=1 -> finishing the just-rotated file ..log[.gz] after a rotation event, - * then returning to index 0 for the new active file - * - * Two robustness features guard against the client compressing a rotated file - * before the agent finished reading it: - * A) the agent can read a rotated file whether it is .log OR .log.gz, and - * D) on restart it locates the file it was reading by a content signature - * (which survives both rename and compression) instead of blindly - * resetting to the new active file. - * - * Migration from v1.1: pre-v1.2 used a flat layout - * /..log[.N.log[.gz]] (one tailer per - * . pair). v1.2 nests each session in its own - * subdirectory; one tailer per . pair instead. The - * tailer's API is unchanged, just the path construction. + * Cursor: {@code fileIndex} = the window index in progress, {@code offset} = bytes + * already sent within it (for mid-window resume after a crash). One LogTailer per + * (sessionId, channel) pair. */ public class LogTailer { private final File folder; @@ -62,9 +50,6 @@ public class LogTailer { private final StreamUploadSettings streamUploadSettings; private final String streamKey; - /** Max rotated index scanned during restart reconciliation. Matches the - * client's LogRotationOptions.max_files default. */ - private static final int MAX_ROTATED_SCAN = 100; /** Bytes of uncompressed head used for the content signature. */ private static final int HEAD_SIG_BYTES = 512; @@ -106,25 +91,6 @@ private File sessionDir() { return new File(folder, sessionId); } - /** v1.2 active file: //.log */ - private File activeFile() { - return new File(sessionDir(), logType + ".log"); - } - - /** - * The compressed-active file: {@code //.log.gz}. - * On a CLEAN SHUTDOWN the client compresses the active file IN PLACE - the - * name keeps the channel base with NO rotation index (active {@code - * .log} -> {@code .log.gz}), distinct from a rotated - * {@code ..log.gz}. A finished session's only data for a channel - * therefore lives here, and there is never a plain {@code .log} - * alongside it. The tail loop falls back to this when the active file is - * gone so finished sessions are still ingested. - */ - private File activeGzFile() { - return new File(sessionDir(), logType + ".log.gz"); - } - /** * The rotated file for {@code index} (>=1). Prefers the uncompressed * {@code .index.log}; falls back to {@code .index.log.gz} once the client @@ -137,20 +103,6 @@ private File resolveRotated(int index) { return gz.exists() ? gz : null; } - private int highestRotatedIndex() { - for (int i = MAX_ROTATED_SCAN; i >= 1; i--) { - if (resolveRotated(i) != null) return i; - } - return -1; - } - - private int previousRotatedIndex(int current) { - for (int i = current - 1; i >= 1; i--) { - if (resolveRotated(i) != null) return i; - } - return -1; - } - private static boolean isGz(File f) { return f.getName().endsWith(".gz"); } @@ -255,290 +207,64 @@ private static long headSignature(File f) { } } - /** Whether {@code cand} is the file described by the saved cursor identity. */ - private static boolean identityMatches(CursorPosition saved, Identity cand) { - if (saved.headSig() != 0L && cand.headSig() == saved.headSig()) return true; // compression-proof - if (saved.fileKey() != null && saved.fileKey().equals(cand.fileKey())) return true; // uncompressed fast path - return false; - } - - /** Locate the index (0=active, >=1=rotated) whose current file matches the - * saved identity. Returns -1 if no file matches. */ - private int locateByIdentity(CursorPosition saved) { - File active = activeFile(); - if (active.exists() && identityMatches(saved, identityOf(active))) return 0; - // The active file we were reading may have been compressed in place to - // .log.gz at shutdown. Its content signature is unchanged - // (headSignature decompresses transparently), so it still resolves to - // the active slot (index 0) and the loop's active branch drains it. - File activeGz = activeGzFile(); - if (activeGz.exists() && identityMatches(saved, identityOf(activeGz))) return 0; - for (int i = 1; i <= MAX_ROTATED_SCAN; i++) { - File r = resolveRotated(i); - if (r == null) continue; - if (identityMatches(saved, identityOf(r))) return i; - } - return -1; - } - /** - * Blocking tail loop - designed to run on a virtual thread. - * Thread.sleep() parks the virtual thread cheaply; no carrier thread is blocked. + * Blocking tail loop (virtual thread). The client publishes each rotation as a + * COMPLETE, immutable window file {@code ..log[.gz]} (index 1 = + * oldest, monotonic incrementing). We send each finished window WHOLE, in + * increasing index order, and finish when the session's {@code .tmp/} working + * 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) { var cursor = cursorMgr.get(streamKey); - int idx = cursor.fileIndex(); - long offset = cursor.offset(); - - // ---- D: restart reconciliation ---- - // A brief downtime may have rotated (and compressed) the file we were - // reading out of its slot - active -> .1.log -> .1.log.gz, or a rotated - // file shifted to a higher index. If we recorded an identity, find the - // file that currently holds it (by content signature, which survives - // compression) and resume from there instead of resetting to the new - // active file and losing the unread tail. - if (cursor.headSig() != 0L || cursor.fileKey() != null) { - int located = locateByIdentity(cursor); - if (located >= 0 && located != idx) { - System.out.println("[" + logType + "] Restart: saved file is now index " + located - + " (cursor said " + idx + "); resuming there at offset " + offset); - idx = located; - cursorMgr.update(streamKey, idx, offset); - } - // located == -1 -> file not found anywhere; keep cursor, legacy fallback applies. - } - - System.out.println("[" + logType + "] Starting - fileIndex=" + idx + ", offset=" + offset); - System.out.println("[" + logType + "] Active file: " + activeFile().getAbsolutePath()); - System.out.println("[" + logType + "] Active file exists: " + activeFile().exists() + ", size: " + (activeFile().exists() ? activeFile().length() : -1)); - - // Signature of the most recently fully-consumed rotated file. Prevents - // re-switching to (and re-reading) the same rotated file after we drain - // it and return to the active file while it is still on disk (it lingers - // until the client's next rotation shifts it, or the archiver deletes it). - long lastConsumedSig = 0L; + // fileIndex = window index in progress (1-based; 0 = none yet); offset = + // bytes already sent within it (mid-window resume after a crash). + int idx = Math.max(1, cursor.fileIndex()); + long offset = cursor.fileIndex() >= 1 ? cursor.offset() : 0L; + System.out.println("[" + logType + "] Starting (window mode) - window index=" + idx + ", offset=" + offset); while (!Thread.currentThread().isInterrupted()) { - File file = (idx == 0) ? activeFile() : resolveRotated(idx); - - if (idx > 0 && file == null) { - // The rotated file has aged out (e.g. agent restarted long after - // the rotation and the file was deleted). Fall back to active. - System.out.println("[" + logType + "] Rotated index " + idx + " gone, resuming active file."); - idx = 0; - offset = 0; - cursorMgr.update(streamKey, idx, offset); - continue; - } - if (idx == 0 && !file.exists()) { - // The active .log is gone. On a clean shutdown the - // client compresses the active file IN PLACE to - // .log.gz (no rotation index), so a finished session's - // only data for this channel lives there. Drain it once from - // our current offset: drainGz skips already-read uncompressed - // bytes, so this is correct whether we discovered the session - // cold (offset 0 -> whole file) or were live-tailing - // .log when it was compressed at shutdown (offset = - // bytes already published -> just the tail). - File activeGz = activeGzFile(); - if (activeGz.exists()) { - long resume = drainGz(activeGz, offset, publisher, 0); - if (resume < 0) { - System.out.println("[" + logType + "] Drained compressed-active " - + activeGz.getName() + " - session finished."); - if (consumedFilesQueue != null) consumedFilesQueue.offer(activeGz.toPath()); - return; // compressed-active is terminal: no more data will arrive - } - // Publish failed mid-drain - back off, retry from persisted offset. - offset = resume; - sleep(5000); - continue; - } - int oldestRotated = highestRotatedIndex(); - if (oldestRotated > 0) { - File rotated = resolveRotated(oldestRotated); - System.out.println("[" + logType + "] Active file is absent; draining rotated " - + rotated.getName() + " from offset 0."); - idx = oldestRotated; - offset = 0; - cursorMgr.update(streamKey, idx, offset); - continue; - } - sleep(1000); - continue; - } - - // ---- A: drain a compressed rotated file ---- - if (idx > 0 && isGz(file)) { - long resume = drainGz(file, offset, publisher, idx); + File window = resolveRotated(idx); + if (window != null) { + long resume = drainWindow(window, offset, publisher, idx); if (resume < 0) { - System.out.println("[" + logType + "] Finished " + file.getName() + " (gz)."); - if (consumedFilesQueue != null) consumedFilesQueue.offer(file.toPath()); - lastConsumedSig = headSignature(file); - int nextRotated = activeFile().exists() ? -1 : previousRotatedIndex(idx); - if (nextRotated > 0) { - idx = nextRotated; - offset = 0; - System.out.println("[" + logType + "] Draining next rotated file " - + resolveRotated(idx).getName() + "."); - cursorMgr.update(streamKey, idx, offset); - continue; - } - if (!activeFile().exists()) { - // Session finished. Leave the cursor at this drained file's - // EOF (drainGz already persisted {idx, endOffset, identity}) - // so a restart resumes HERE, skips to EOF and exits. Do NOT - // reset to offset 0 - that keeps the file identity but wipes - // progress, so the next run re-locates this file by content - // signature and re-uploads the whole session (backend 409s it). - System.out.println("[" + logType + "] No active file remains; session finished."); - return; - } - System.out.println("[" + logType + "] Resuming active file."); - idx = 0; - offset = 0; + System.out.println("[" + logType + "] Sent window " + window.getName() + "."); + if (consumedFilesQueue != null) consumedFilesQueue.offer(window.toPath()); + idx++; + offset = 0L; cursorMgr.update(streamKey, idx, offset); } else { - // Publish failed mid-gz - back off and retry from persisted offset. + // Publish failed mid-window; the cursor already holds the resume offset. offset = resume; sleep(5000); } continue; } - // ---- plain .log path (active file, or an uncompressed rotated file) ---- - Identity id = identityOf(file); - try (var reader = new RandomAccessFile(file, "r")) { - if (offset > file.length()) offset = 0; - reader.seek(offset); - - boolean fileIsDone = false; - - while (!Thread.currentThread().isInterrupted() && !fileIsDone) { - if (file.length() > offset) { - String line; - boolean publishFailed = false; - if (streamUploadSettings.enabled()) { - StreamBatch batch = new StreamBatch(streamUploadSettings); - while (true) { - long lineStart = reader.getFilePointer(); - line = readLineUtf8(reader); - if (line == null) break; - long lineEnd = reader.getFilePointer(); - - String prepared = prepareStreamLine(line); - if (prepared == null) { - offset = lineEnd; - continue; - } - batch.add(prepared, lineStart, lineEnd); - if (batch.shouldFlush()) { - if (!flushStreamBatch(batch, publisher)) { - System.out.println("[" + logType + "] Stream publish FAILED - will retry in 5s"); - reader.seek(batch.startOffset()); - offset = batch.startOffset(); - publishFailed = true; - break; - } - offset = batch.endOffset(); - batch.clear(); - } - } - if (!publishFailed && !batch.isEmpty()) { - if (!flushStreamBatch(batch, publisher)) { - System.out.println("[" + logType + "] Stream publish FAILED - will retry in 5s"); - reader.seek(batch.startOffset()); - offset = batch.startOffset(); - publishFailed = true; - } else { - offset = batch.endOffset(); - batch.clear(); - } - } - } else { - // NOTE: RandomAccessFile.readLine() decodes bytes as ISO-8859-1, - // corrupting multi-byte UTF-8 characters. Use readLineUtf8(). - while ((line = readLineUtf8(reader)) != null) { - long lineStart = offset; - if (!handleLine(line, publisher)) { - System.out.println("[" + logType + "] Publish FAILED - will retry in 5s"); - reader.seek(lineStart); // retry this line next iteration - publishFailed = true; - break; - } - offset = reader.getFilePointer(); - } - } - cursorMgr.update(streamKey, idx, offset, id.fileKey(), id.size(), id.headSig()); - if (publishFailed) { - sleep(5000); - } - } else { - // No new bytes in this file. - if (idx == 0) { - // Tailing the active file: check whether rotation just happened. - // Break to close the reader so rotation can proceed on Windows. - fileIsDone = true; - - File rotated = resolveRotated(1); - // Don't re-consume a rotated file we already fully drained: it - // lingers on disk until the client's next rotation or the archiver - // removes it. Identify it by content signature. - long rsig = (rotated != null) ? headSignature(rotated) : 0L; - if (rotated != null && rsig != 0L && rsig == lastConsumedSig) { - sleep(200); // already consumed this exact rotated content; wait for a real new rotation - } - // A compressed rotated file (.gz) is always valid to switch to - - // its uncompressed size isn't comparable to our byte offset, so we - // rely on skip-to-offset in drainGz. For an uncompressed .log we keep - // the length>=offset guard (it must be at least the old active size). - else if (rotated != null && (isGz(rotated) || rotated.length() >= offset)) { - System.out.println("[" + logType + "] Rotation detected - switching to " + rotated.getName() + " at offset " + offset); - idx = 1; - cursorMgr.update(streamKey, idx, offset, id.fileKey(), id.size(), id.headSig()); - } else { - sleep(200); - } - } else { - // Finished draining an uncompressed rotated file. - System.out.println("[" + logType + "] Finished " + file.getName() + "."); - if (consumedFilesQueue != null) { - consumedFilesQueue.offer(file.toPath()); - } - lastConsumedSig = id.headSig(); - int nextRotated = activeFile().exists() ? -1 : previousRotatedIndex(idx); - if (nextRotated > 0) { - idx = nextRotated; - offset = 0; - System.out.println("[" + logType + "] Draining next rotated file " - + resolveRotated(idx).getName() + "."); - cursorMgr.update(streamKey, idx, offset); - fileIsDone = true; - continue; - } - if (!activeFile().exists()) { - // Session finished. Leave the cursor at this drained - // file's EOF (updated above to {idx, offset, identity}) - // so a restart skips to EOF and exits. Do NOT reset to - // offset 0 - that re-uploads the whole session next run. - System.out.println("[" + logType + "] No active file remains; session finished."); - return; - } - System.out.println("[" + logType + "] Resuming active file."); - idx = 0; - offset = 0; - cursorMgr.update(streamKey, idx, offset); - fileIsDone = true; - } - } - } - } catch (IOException e) { - System.out.println("[" + logType + "] IO error: " + e.getMessage()); + // Window not published yet. Is the session still writing? + if (sessionTmpDir().exists()) { sleep(2000); + continue; } + + // .tmp/ gone -> the client closed every channel. Each channel's last + // window is published BEFORE .tmp/ is removed, so wait a moment then do + // one final check for a straggler before finishing. + sleep(4500); + if (resolveRotated(idx) != null) continue; + System.out.println("[" + logType + "] Session finished (.tmp gone, no window " + + idx + "; last sent " + (idx - 1) + ")."); + return; } } + /** The client's per-session working dir {@code //.tmp}: active + * files live here while writing; the client removes it once every channel has + * closed, which is our "no more windows" signal. */ + private File sessionTmpDir() { + return new File(sessionDir(), ".tmp"); + } + /** * Drain a compressed rotated file from uncompressed {@code startOffset}, * publishing each line. Returns {@code -1} when the file is fully drained @@ -547,9 +273,32 @@ else if (rotated != null && (isGz(rotated) || rotated.length() >= offset)) { * correctly. gz streams have no random seek, so a retry re-opens and * re-skips - bounded by the 50 MB rotated-file size. */ - private long drainGz(File gz, long startOffset, Publisher publisher, int idx) { - Identity id = identityOf(gz); - try (InputStream in = new BufferedInputStream(new GZIPInputStream(new FileInputStream(gz)))) { + private long drainWindow(File window, long startOffset, Publisher publisher, int idx) { + // A published window is a COMPLETE gzipped NDJSON file - exactly the stream + // wire format - so in stream mode send it VERBATIM as one chunk (no decompress + // / re-batch / re-gzip). Returns -1 (sent) or 0 (retry from the start). Plain + // (.log) windows from a no-compressor build fall through to the line path below. + if (streamUploadSettings.enabled() && isGz(window)) { + try { + byte[] gz = Files.readAllBytes(window.toPath()); + if (publisher.publishStreamGz(sessionId, gz)) { + // Persist the window-done advance right after the durable 2xx so a crash + // before tail() advances won't re-send the whole window on restart (the + // line path likewise persists the cursor after each accepted batch). + cursorMgr.update(streamKey, idx + 1, 0L); + return -1L; + } + System.out.println("[" + logType + "] window publish FAILED - retry in 5s"); + return 0L; + } catch (IOException e) { + System.out.println("[" + logType + "] read error on " + window.getName() + ": " + e.getMessage()); + return 0L; + } + } + Identity id = identityOf(window); + try (InputStream in = isGz(window) + ? new BufferedInputStream(new GZIPInputStream(new FileInputStream(window))) + : new BufferedInputStream(new FileInputStream(window))) { long start = skipFully(in, startOffset); long[] consumed = { start }; if (streamUploadSettings.enabled()) { @@ -604,7 +353,7 @@ private long drainGz(File gz, long startOffset, Publisher publisher, int idx) { } return consumed[0]; // interrupted } catch (IOException e) { - System.out.println("[" + logType + "] gz read error on " + gz.getName() + ": " + e.getMessage()); + System.out.println("[" + logType + "] window read error on " + window.getName() + ": " + e.getMessage()); return startOffset; // retry from the start of this attempt } } @@ -690,34 +439,6 @@ private LogWrapper processLine(String rawLine) { } } - /** - * Read a line from a RandomAccessFile, decoding bytes as UTF-8. - * Handles both {@code \n} and {@code \r\n} line endings. - * Returns null at EOF or on an incomplete (unterminated) line. - */ - private static String readLineUtf8(RandomAccessFile raf) throws IOException { - long startPos = raf.getFilePointer(); - var buf = new ByteArrayOutputStream(512); - int b; - boolean foundNewline = false; - while ((b = raf.read()) != -1) { - if (b == '\n') { foundNewline = true; break; } - if (b == '\r') { - foundNewline = true; - long pos = raf.getFilePointer(); - int next = raf.read(); - if (next != '\n' && next != -1) raf.seek(pos); - break; - } - buf.write(b); - } - if (!foundNewline) { - raf.seek(startPos); - return null; - } - return buf.toString(StandardCharsets.UTF_8); - } - /** * Read one UTF-8 line from a plain InputStream, advancing {@code consumed[0]} * by the exact number of bytes read (including the terminating {@code \n}). diff --git a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java index 00f1cf2..695b9ca 100644 --- a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java +++ b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java @@ -84,47 +84,66 @@ public boolean publishStream(String sessionId, List ndjsonLines) { try { String ndjson = String.join("\n", ndjsonLines) + "\n"; byte[] body = gzip(ndjson.getBytes(StandardCharsets.UTF_8)); - String url = config.streamEndpoint(); - System.out.println("[agent] HTTP stream POST starting: url=" + url - + " session=" + sessionId - + " lines=" + ndjsonLines.size() - + " gzipBytes=" + body.length); + System.out.println("[agent] HTTP stream POST starting: url=" + config.streamEndpoint() + + " session=" + sessionId + " lines=" + ndjsonLines.size() + " gzipBytes=" + body.length); + return postGzStream(sessionId, body); + } catch (Exception e) { + System.out.println("HTTP stream connection error: " + e); + return false; + } + } + @Override + public boolean publishStreamGz(String sessionId, byte[] gzBody) { + if (gzBody == null || gzBody.length == 0) { + return true; + } + System.out.println("[agent] HTTP stream POST starting (gz window): url=" + config.streamEndpoint() + + " session=" + sessionId + " gzipBytes=" + gzBody.length); + return postGzStream(sessionId, gzBody); + } + + /** + * POST an already-gzipped NDJSON body to the stream endpoint and interpret the + * status. 2xx (accepted), 409 (already finalized on a restart) and 402 (limit) + * all advance; anything else is a retryable failure. + */ + private boolean postGzStream(String sessionId, byte[] gzBody) { + try { + String url = config.streamEndpoint(); var requestBuilder = HttpRequest.newBuilder() .uri(URI.create(url)) .header("Content-Type", "application/x-ndjson") .header("Content-Encoding", "gzip") .header("X-GpuFlight-Session-Id", sessionId) .header("X-GpuFlight-Hostname", InetAddress.getLocalHost().getHostName()) - .POST(HttpRequest.BodyPublishers.ofByteArray(body)); + .POST(HttpRequest.BodyPublishers.ofByteArray(gzBody)); addAuthHeader(requestBuilder); var response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString()); - if (response.statusCode() >= 200 && response.statusCode() <= 299) { - System.out.println("[agent] HTTP stream POST accepted: status=" - + response.statusCode() + " session=" + sessionId); + int sc = response.statusCode(); + if (sc >= 200 && sc <= 299) { + System.out.println("[agent] HTTP stream POST accepted: status=" + sc + " session=" + sessionId); return true; } - if (response.statusCode() == 409) { + if (sc == 409) { // Session already finalized on the backend - typically this agent - // re-tailing a finished session after a restart. The data is - // already stored and re-profiling is required to replace it, so - // retrying NEVER succeeds. Advance like a 2xx (mirrors the 402 - // handling in publish()) so the tailer drains to EOF and exits - // instead of re-POSTing the same batch every 5s forever. + // re-tailing a finished session after a restart. The data is already + // stored and re-profiling is required to replace it, so retrying NEVER + // succeeds. Advance like a 2xx so the tailer finishes instead of + // re-POSTing forever. System.out.println("[agent] HTTP stream POST: session already uploaded (409) - " + "skipping, session=" + sessionId); return true; } - if (response.statusCode() == 402) { + if (sc == 402) { // GPU/workspace limit exceeded - permanent; retrying won't help. System.err.println("[GPUFL] GPU limit exceeded for this workspace - " + "skipping session=" + sessionId + ". " + response.body()); return true; } - System.out.println("HTTP stream publish failed [" + url + "]: " - + response.statusCode() + " - " + response.body()); + System.out.println("HTTP stream publish failed [" + url + "]: " + sc + " - " + response.body()); return false; } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/src/main/java/com/gpuflight/agent/publisher/Publisher.java b/src/main/java/com/gpuflight/agent/publisher/Publisher.java index ea530ac..b329242 100644 --- a/src/main/java/com/gpuflight/agent/publisher/Publisher.java +++ b/src/main/java/com/gpuflight/agent/publisher/Publisher.java @@ -14,6 +14,17 @@ default boolean publishStream(String sessionId, List ndjsonLines) { return false; } + /** + * Upload a complete, already-gzipped NDJSON window verbatim as one chunk - the + * client's {@code ..log.gz} sent byte-for-byte. It is exactly the + * gzipped-NDJSON wire format {@link #publishStream} produces, so this skips the + * decompress + re-batch + re-gzip round-trip. Default unsupported - the window + * fast-path is gated to the HTTP stream publisher. + */ + default boolean publishStreamGz(String sessionId, byte[] gzBody) { + return false; + } + /** * Signal the backend that EVERY channel of {@code sessionId} has finished * uploading — the agent has drained all per-channel tailers and every batch diff --git a/src/test/java/com/gpuflight/agent/LogTailerTest.java b/src/test/java/com/gpuflight/agent/LogTailerTest.java index 94af7a0..38cb9c9 100644 --- a/src/test/java/com/gpuflight/agent/LogTailerTest.java +++ b/src/test/java/com/gpuflight/agent/LogTailerTest.java @@ -7,12 +7,10 @@ import org.junit.jupiter.api.io.TempDir; import java.io.File; - import java.io.IOException; 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.LinkedBlockingQueue; @@ -21,68 +19,63 @@ import static org.junit.jupiter.api.Assertions.*; +/** + * Window-model tailing: the client publishes each rotation as a COMPLETE, + * immutable file {@code ..log[.gz]} (index 1 = oldest, higher = + * newer). The tailer sends each finished window WHOLE, in increasing index order, + * and finishes once the session's {@code .tmp/} dir is gone and no further window + * has appeared. There is no live {@code .log} to tail. + */ class LogTailerTest { - /** Simple in-memory publisher for testing. */ static class CapturingPublisher implements Publisher { final List events = new CopyOnWriteArrayList<>(); - - @Override - public boolean publish(String topic, String key, LogWrapper log) { - events.add(log); - return true; - } - - @Override - public void close() {} - } - - /** Publisher that rejects any line whose data contains a marker (simulates - * a backend that won't accept a specific event), accepting everything else. */ - static class RejectingPublisher implements Publisher { - final List events = new CopyOnWriteArrayList<>(); - final String reject; - RejectingPublisher(String reject) { this.reject = reject; } - - @Override - public boolean publish(String topic, String key, LogWrapper log) { - if (reject != null && log.data().contains(reject)) return false; - events.add(log); - return true; - } - - @Override - public void close() {} + @Override public boolean publish(String topic, String key, LogWrapper log) { events.add(log); return true; } + @Override public void close() {} } static class CapturingStreamPublisher implements Publisher { final List> batches = new CopyOnWriteArrayList<>(); final List sessionIds = new CopyOnWriteArrayList<>(); - - @Override - public boolean publish(String topic, String key, LogWrapper log) { - return false; + @Override public boolean publish(String topic, String key, LogWrapper log) { return false; } + @Override public boolean publishStream(String sessionId, List ndjsonLines) { + sessionIds.add(sessionId); batches.add(List.copyOf(ndjsonLines)); return true; } - - @Override - public boolean publishStream(String sessionId, List ndjsonLines) { + @Override public boolean publishStreamGz(String sessionId, byte[] gzBody) { + // Mirror the backend: decompress the verbatim window gz and record its lines. sessionIds.add(sessionId); - batches.add(List.copyOf(ndjsonLines)); + try (var in = new java.util.zip.GZIPInputStream(new java.io.ByteArrayInputStream(gzBody))) { + String text = new String(in.readAllBytes(), StandardCharsets.UTF_8); + batches.add(java.util.Arrays.stream(text.split("\n")).filter(s -> !s.isEmpty()).toList()); + } catch (Exception e) { return false; } return true; } - - @Override - public void close() {} + @Override public void close() {} } - /** gzip src → dest (used to simulate the C++ client compressing a rotated file). */ private static void gzip(Path src, Path dest) throws IOException { - try (var in = Files.newInputStream(src); - var out = new GZIPOutputStream(Files.newOutputStream(dest))) { + try (var in = Files.newInputStream(src); var out = new GZIPOutputStream(Files.newOutputStream(dest))) { in.transferTo(out); } } + /** Publish a complete window {@code /..log}. */ + private static Path window(Path dir, String session, String channel, int index, String content) throws IOException { + Files.createDirectories(dir.resolve(session)); + Path p = dir.resolve(session + "/" + channel + "." + index + ".log"); + Files.writeString(p, content); + return p; + } + + /** Publish a compressed window {@code /..log.gz}. */ + private static Path gzWindow(Path dir, String session, String channel, int index, String content) throws IOException { + Path plain = window(dir, session, channel, index, content); + Path gz = dir.resolve(session + "/" + channel + "." + index + ".log.gz"); + gzip(plain, gz); + Files.delete(plain); + return gz; + } + private static Thread startTailer(LogTailer tailer, Publisher publisher) { Thread t = new Thread(() -> tailer.tail(publisher)); t.setDaemon(true); @@ -92,593 +85,220 @@ private static Thread startTailer(LogTailer tailer, Publisher publisher) { private static void awaitEvents(List list, int count, long timeoutMs) throws InterruptedException { long deadline = System.currentTimeMillis() + timeoutMs; - while (list.size() < count && System.currentTimeMillis() < deadline) { - Thread.sleep(50); - } + while (list.size() < count && System.currentTimeMillis() < deadline) Thread.sleep(50); } - // ---- happy-path: reads existing lines ---- - - @Test - void tail_sassChannel_publishes(@TempDir Path tempDir) throws Exception { - // sass.log carries SASS-disassembly / source-content artifacts split - // out of device.log; the agent must tail it like any other channel - // or those artifacts silently miss live upload. - Files.createDirectories(tempDir.resolve("app")); - Files.writeString(tempDir.resolve("app/sass.log"), - "{\"type\":\"cubin_disassembly\"}\n"); - - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "sass", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(1, publisher.events.size()); - assertEquals("cubin_disassembly", publisher.events.get(0).type()); + private static LogTailer tailer(Path dir, String session, String channel, CursorManager cur) { + return new LogTailer(dir.toFile(), session, channel, "gpu-trace", cur, new LinkedBlockingQueue<>()); } - @Test - void tail_readsExistingLinesAndPublishes(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - Path logFile = tempDir.resolve("app/device.log"); - Files.writeString(logFile, - "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n" + - "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n" - ); - - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 2, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(2, publisher.events.size()); - assertEquals("kernel_event", publisher.events.get(0).type()); - assertEquals("kernel_event", publisher.events.get(1).type()); + private static LogTailer streamTailer(Path dir, String session, String channel, CursorManager cur) { + return new LogTailer(dir.toFile(), session, channel, "gpu-trace", cur, new LinkedBlockingQueue<>(), + null, new StreamUploadSettings(true, 10, 1_000_000L)); } + // ---- a complete window is sent whole ---- + @Test - void tail_skipsBlankLines(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - Path logFile = tempDir.resolve("app/scope.log"); - Files.writeString(logFile, - "{\"type\":\"scope_event\"}\n" + - "\n" + - " \n" + - "{\"type\":\"scope_event\"}\n" - ); - - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "scope", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 2, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(2, publisher.events.size()); + void window_readsAndPublishes(@TempDir Path dir) throws Exception { + window(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "device", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.events, 2, 3000); + t.interrupt(); t.join(2000); + assertEquals(2, pub.events.size()); + assertEquals("kernel_event", pub.events.get(0).type()); } @Test - void tail_invalidJsonLine_doesNotCrash(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - Path logFile = tempDir.resolve("app/system.log"); - Files.writeString(logFile, - "not-json-at-all\n" + - "{\"type\":\"system_event\"}\n" - ); - - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "system", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 3000); - t.interrupt(); - t.join(2000); - - // Only the valid JSON line is published - assertEquals(1, publisher.events.size()); - assertEquals("system_event", publisher.events.get(0).type()); + void window_sassChannel_publishes(@TempDir Path dir) throws Exception { + window(dir, "app", "sass", 1, "{\"type\":\"cubin_disassembly\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "sass", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.events, 1, 3000); + t.interrupt(); t.join(2000); + assertEquals(1, pub.events.size()); + assertEquals("cubin_disassembly", pub.events.get(0).type()); } - // ---- cursor persistence ---- - @Test - void tail_persistsCursorAfterReading(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - Path logFile = tempDir.resolve("app/device.log"); - Files.writeString(logFile, "{\"type\":\"kernel_event\"}\n"); - - File cursorFile = tempDir.resolve("cursor.json").toFile(); - CursorManager cursorMgr = new CursorManager(cursorFile); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 3000); - t.interrupt(); - t.join(2000); - - // Reload cursor and verify it advanced - CursorManager mgr2 = new CursorManager(cursorFile); - CursorPosition pos = mgr2.get("app.device"); - assertEquals(0, pos.fileIndex()); - assertTrue(pos.offset() > 0, "Cursor offset should have advanced past the line"); + void window_skipsBlankLines(@TempDir Path dir) throws Exception { + window(dir, "app", "scope", 1, "{\"type\":\"scope_event\"}\n\n \n{\"type\":\"scope_event\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "scope", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.events, 2, 3000); + t.interrupt(); t.join(2000); + assertEquals(2, pub.events.size()); } @Test - void tail_streamMode_publishesNdjsonBatchAndPersistsCursor(@TempDir Path tempDir) throws Exception { - Files.createDirectories(tempDir.resolve("app")); - Path logFile = tempDir.resolve("app/device.log"); - Files.writeString(logFile, - "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" + - "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n" - ); - - File cursorFile = tempDir.resolve("cursor.json").toFile(); - CursorManager cursorMgr = new CursorManager(cursorFile); - CapturingStreamPublisher publisher = new CapturingStreamPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", - "gpu-trace", cursorMgr, new LinkedBlockingQueue<>(), null, - new StreamUploadSettings(true, 10, 1_000_000L)); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.batches, 1, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(List.of("app"), publisher.sessionIds); - assertEquals(1, publisher.batches.size()); - assertEquals(2, publisher.batches.get(0).size()); - assertTrue(publisher.batches.get(0).get(0).contains("\"k1\"")); - - CursorPosition pos = new CursorManager(cursorFile).get("app.device"); - assertEquals(0, pos.fileIndex()); - assertTrue(pos.offset() > 0, "Cursor offset should advance after stream batch accept"); + void window_invalidJsonLine_doesNotCrash(@TempDir Path dir) throws Exception { + window(dir, "app", "system", 1, "not-json-at-all\n{\"type\":\"system_event\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "system", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.events, 1, 3000); + t.interrupt(); t.join(2000); + assertEquals(1, pub.events.size()); + assertEquals("system_event", pub.events.get(0).type()); } @Test - void tail_streamMode_drainsRotatedGzWhenActiveFileMissing(@TempDir Path tempDir) throws Exception { - Files.createDirectories(tempDir.resolve("app")); - String content = - "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" + - "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n"; - Path plain = tempDir.resolve("app/device.1.log"); - Files.writeString(plain, content); - Path gz = tempDir.resolve("app/device.1.log.gz"); - gzip(plain, gz); - Files.delete(plain); - - File cursorFile = tempDir.resolve("cursor.json").toFile(); - CapturingStreamPublisher publisher = new CapturingStreamPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", - "gpu-trace", new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null, - new StreamUploadSettings(true, 10, 1_000_000L)); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.batches, 1, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(1, publisher.batches.size()); - assertEquals(2, publisher.batches.get(0).size()); - assertTrue(publisher.batches.get(0).get(1).contains("\"k2\"")); + void window_nullQueue_doesNotThrow(@TempDir Path dir) throws Exception { + window(dir, "app", "device", 1, "{\"type\":\"kernel_event\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(new LogTailer(dir.toFile(), "app", "device", "gpu-trace", + new CursorManager(dir.resolve("cursor.json").toFile()), null), pub); + awaitEvents(pub.events, 1, 3000); + t.interrupt(); t.join(2000); + assertEquals(1, pub.events.size()); } - // ---- waits when active file is absent ---- + // ---- multiple windows sent in increasing index order ---- @Test - void tail_waitsForActiveFileToAppear(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - // No log file exists initially — tailer should wait - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - Thread.sleep(300); // let it enter the waiting branch - - // Now create the file with content - Path logFile = tempDir.resolve("app/device.log"); - Files.writeString(logFile, "{\"type\":\"kernel_event\"}\n"); - - awaitEvents(publisher.events, 1, 5000); - t.interrupt(); - t.join(2000); - - assertEquals(1, publisher.events.size()); + void windows_sentInAscendingOrder(@TempDir Path dir) throws Exception { + window(dir, "app", "device", 1, "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"); + window(dir, "app", "device", 2, "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"); + window(dir, "app", "device", 3, "{\"type\":\"kernel_event\",\"name\":\"k3\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "device", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.events, 3, 4000); + t.interrupt(); t.join(2000); + assertEquals(3, pub.events.size()); + assertTrue(pub.events.get(0).data().contains("k1")); + assertTrue(pub.events.get(1).data().contains("k2")); + assertTrue(pub.events.get(2).data().contains("k3")); } - // ---- null consumedFilesQueue is handled gracefully ---- + // ---- gz windows + mid-window resume ---- @Test - void tail_nullQueue_doesNotThrow(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - Path logFile = tempDir.resolve("app/device.log"); - Files.writeString(logFile, "{\"type\":\"kernel_event\"}\n"); - - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - // Pass null for the queue — should not throw - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, null); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(1, publisher.events.size()); + void gzWindow_reads(@TempDir Path dir) throws Exception { + gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "device", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.events, 2, 3000); + t.interrupt(); t.join(2000); + assertEquals(2, pub.events.size()); } - // ---- log file rotation ---- - @Test - void tail_detects_rotation_and_offersToQueue(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - // Set up: active file exists; then we rename it to .1.log (simulating C++ rotation) - // and create a new empty active file. - Path activeFile = tempDir.resolve("app/device.log"); - Files.writeString(activeFile, "{\"type\":\"kernel_event\"}\n"); - - LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, queue); - - Thread t = startTailer(tailer, publisher); - // Wait for tailer to read the line and advance the cursor - awaitEvents(publisher.events, 1, 3000); - - // Stop the tailer thread to release the file lock - t.interrupt(); - t.join(5000); - - // Simulate rotation: rename active → .1.log, create new empty active - Path rotatedFile = tempDir.resolve("app/device.1.log"); - Files.move(activeFile, rotatedFile); - Files.createFile(activeFile); // new empty active file - - // Restart tailer to see if it detects the rotation from the persisted cursor - t = startTailer(tailer, publisher); - - // The tailer should detect rotation and switch to index 1, - // then drain the rotated file (it's already fully read, offset == length) - // then offer the rotated path to the queue. - Path consumed = queue.poll(5, TimeUnit.SECONDS); - t.interrupt(); - t.join(2000); - - assertNotNull(consumed, "Rotated file should have been offered to the archive queue"); - assertEquals(rotatedFile.toAbsolutePath(), consumed.toAbsolutePath()); + void gzWindow_resumesFromOffset(@TempDir Path dir) throws Exception { + String l1 = "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"; + gzWindow(dir, "app", "device", 1, l1 + "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"); + long off = l1.getBytes(StandardCharsets.UTF_8).length; + File cursorFile = dir.resolve("cursor.json").toFile(); + Files.writeString(cursorFile.toPath(), "{\"streams\":{\"app.device\":{\"fileIndex\":1,\"offset\":" + off + "}}}"); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "device", new CursorManager(cursorFile)), pub); + awaitEvents(pub.events, 1, 3000); + t.interrupt(); t.join(2000); + assertEquals(1, pub.events.size(), "only the tail after the offset should be read"); + assertTrue(pub.events.get(0).data().contains("k2")); } - // ---- rotated file gone on restart ---- + // ---- cursor advances past a sent window ---- @Test - void tail_rotatedFileGone_fallsBackToActiveFile(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - // Simulate a cursor that points to fileIndex=1 (rotated), but the file no longer exists - File cursorFile = tempDir.resolve("cursor.json").toFile(); - // Write a cursor that says we were reading rotated file index 1 at offset 0 - java.nio.file.Files.writeString(cursorFile.toPath(), """ - { - "streams": { - "app.device": { "fileIndex": 1, "offset": 0 } - } - } - """); - - // Create only the active file (rotated .1.log is gone) - Path activeFile = tempDir.resolve("app/device.log"); - Files.writeString(activeFile, "{\"type\":\"kernel_event\"}\n"); - - CursorManager cursorMgr = new CursorManager(cursorFile); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 5000); - t.interrupt(); - t.join(2000); - - // Should recover and eventually read from the active file - assertFalse(publisher.events.isEmpty(), "Should fall back to active file after rotated file is gone"); + void cursor_advancesAfterWindow(@TempDir Path dir) throws Exception { + window(dir, "app", "device", 1, "{\"type\":\"kernel_event\"}\n"); + File cursorFile = dir.resolve("cursor.json").toFile(); + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "device", new CursorManager(cursorFile)), pub); + awaitEvents(pub.events, 1, 3000); + Thread.sleep(200); // let the loop advance the cursor past window 1 + t.interrupt(); t.join(2000); + CursorPosition pos = new CursorManager(cursorFile).get("app.device"); + assertTrue(pos.fileIndex() >= 1, "cursor should sit on/past window 1"); } - // ---- A: read compressed (.gz) rotated files ---- - @Test - void tail_readsGzRotatedFile(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - String content = - "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n" + - "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"; - Path plain = tempDir.resolve("app/device.1.log"); - Files.writeString(plain, content); - Path gz = tempDir.resolve("app/device.1.log.gz"); - gzip(plain, gz); - Files.delete(plain); // only the .gz remains - Files.createFile(tempDir.resolve("app/device.log")); // empty active so tailer doesn't just wait - - File cursorFile = tempDir.resolve("cursor.json").toFile(); - Files.writeString(cursorFile.toPath(), - "{\"streams\":{\"app.device\":{\"fileIndex\":1,\"offset\":0}}}"); - - CursorManager cursorMgr = new CursorManager(cursorFile); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 2, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(2, publisher.events.size(), "Both lines should be read from the .gz"); + void streamMode_publishesBatch(@TempDir Path dir) throws Exception { + window(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" + + "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n"); + var pub = new CapturingStreamPublisher(); + var t = startTailer(streamTailer(dir, "app", "device", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + awaitEvents(pub.batches, 1, 3000); + t.interrupt(); t.join(2000); + assertEquals(List.of("app"), pub.sessionIds); + assertEquals(2, pub.batches.get(0).size()); + assertTrue(pub.batches.get(0).get(0).contains("\"k1\"")); } @Test - void tail_gzResumeFromOffset(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - String l1 = "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"; - String l2 = "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"; - Path plain = tempDir.resolve("app/device.1.log"); - Files.writeString(plain, l1 + l2); - Path gz = tempDir.resolve("app/device.1.log.gz"); - gzip(plain, gz); - Files.delete(plain); - Files.createFile(tempDir.resolve("app/device.log")); - - long off = l1.getBytes(StandardCharsets.UTF_8).length; // start after the first line - File cursorFile = tempDir.resolve("cursor.json").toFile(); - Files.writeString(cursorFile.toPath(), - "{\"streams\":{\"app.device\":{\"fileIndex\":1,\"offset\":" + off + "}}}"); - - CursorManager cursorMgr = new CursorManager(cursorFile); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - cursorMgr, new LinkedBlockingQueue<>()); + void streamMode_gzWindow_sentAsOneChunk(@TempDir Path dir) throws Exception { + // A .gz window is shipped verbatim as ONE chunk (no decompress/re-batch), + // and the cursor advances past it so a crash/restart never re-sends it. + gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"); + File cursorFile = dir.resolve("cursor.json").toFile(); + var pub = new CapturingStreamPublisher(); + var t = startTailer(streamTailer(dir, "app", "device", new CursorManager(cursorFile)), pub); + awaitEvents(pub.batches, 1, 3000); + Thread.sleep(200); // let the window-done advance persist + t.interrupt(); t.join(2000); + assertEquals(1, pub.batches.size(), "the whole gz window is sent as ONE chunk"); + assertEquals(2, pub.batches.get(0).size()); + assertTrue(pub.batches.get(0).get(0).contains("k1")); + assertTrue(pub.batches.get(0).get(1).contains("k2")); + assertTrue(new CursorManager(cursorFile).get("app.device").fileIndex() >= 2, + "cursor must advance past the sent window so a restart does not re-send it"); + } - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 3000); - t.interrupt(); - t.join(2000); + // ---- waits while the session is still writing (.tmp/ present) ---- - assertEquals(1, publisher.events.size(), "Only the line after the offset should be read"); - assertTrue(publisher.events.get(0).data().contains("k2")); + @Test + void waitsWhileTmpPresent_thenSendsLateWindow(@TempDir Path dir) throws Exception { + Files.createDirectories(dir.resolve("app/.tmp")); // session writing, no window yet + var pub = new CapturingPublisher(); + var t = startTailer(tailer(dir, "app", "device", new CursorManager(dir.resolve("cursor.json").toFile())), pub); + Thread.sleep(300); + assertTrue(pub.events.isEmpty(), "no window published yet -> nothing sent"); + window(dir, "app", "device", 1, "{\"type\":\"kernel_event\"}\n"); + awaitEvents(pub.events, 1, 5000); + t.interrupt(); t.join(2000); + assertEquals(1, pub.events.size()); } - // ---- D: restart recovers the unread tail after the client compressed it ---- + // ---- a finished session (no .tmp/) exits on its own and is not re-uploaded ---- @Test - void tail_restartAfterCompression_recoversUnreadTail(@TempDir Path tempDir) throws Exception { - // v1.2: log files live under //.log. - // Create the session subdir so the tests' writeString calls don't - // hit "parent dir doesn't exist" errors. The "app" name is used - // throughout as the session_id below (it was the filePrefix in - // pre-v1.2 tests — same string, repurposed). - Files.createDirectories(tempDir.resolve("app")); - Path active = tempDir.resolve("app/device.log"); - String k1 = "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"; - String k2 = "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"; - Files.writeString(active, k1 + k2); - - File cursorFile = tempDir.resolve("cursor.json").toFile(); - - // Run 1: publisher accepts k1, rejects k2 → the agent stays "behind" at the - // offset after k1, having recorded the file's identity in the cursor. - RejectingPublisher pub1 = new RejectingPublisher("k2"); - LogTailer tailer1 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - new CursorManager(cursorFile), new LinkedBlockingQueue<>()); - Thread t1 = startTailer(tailer1, pub1); - awaitEvents(pub1.events, 1, 3000); // k1 published - Thread.sleep(400); // let it attempt+fail k2 and persist the cursor - t1.interrupt(); - t1.join(2000); - - CursorPosition saved = new CursorManager(cursorFile).get("app.device"); - assertEquals(0, saved.fileIndex()); - assertEquals(k1.getBytes(StandardCharsets.UTF_8).length, saved.offset(), - "cursor should sit just after k1"); - assertTrue(saved.headSig() != 0L, "file identity (headSig) should be captured"); - - // Simulate the client rotating AND immediately compressing: the old active - // content lands in .1.log.gz, the active file is replaced with fresh content. - Path gz = tempDir.resolve("app/device.1.log.gz"); - gzip(active, gz); - Files.delete(active); - String k3 = "{\"type\":\"kernel_event\",\"name\":\"k3\"}\n"; - Files.writeString(active, k3); - - // Run 2: without A+D the k2 tail (now only inside the .gz) would be lost. - CapturingPublisher pub2 = new CapturingPublisher(); - LogTailer tailer2 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - new CursorManager(cursorFile), new LinkedBlockingQueue<>()); - Thread t2 = startTailer(tailer2, pub2); - awaitEvents(pub2.events, 2, 5000); // k2 (recovered from gz) + k3 (new active) - t2.interrupt(); - t2.join(2000); - - List datas = pub2.events.stream().map(LogWrapper::data).toList(); - assertTrue(datas.stream().anyMatch(d -> d.contains("k2")), - "k2 must be recovered from the compressed rotated file"); - assertTrue(datas.stream().anyMatch(d -> d.contains("k3")), - "k3 from the new active file should also be read"); + void finishedSession_exitsAndNotReuploaded(@TempDir Path dir) throws Exception { + gzWindow(dir, "app", "device", 1, + "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" + + "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n"); + File cursorFile = dir.resolve("cursor.json").toFile(); + + // Run 1: no .tmp/ -> send the one window, then exit on its own (after the finish grace). + var pub1 = new CapturingStreamPublisher(); + var t1 = startTailer(streamTailer(dir, "app", "device", new CursorManager(cursorFile)), pub1); + t1.join(8000); + assertFalse(t1.isAlive(), "tailer should exit on its own once the finished session is drained"); + assertEquals(1, pub1.batches.size(), "run 1 sends the finished session once"); + + // Run 2: same files + persisted cursor -> nothing re-sent. + var pub2 = new CapturingStreamPublisher(); + var t2 = startTailer(streamTailer(dir, "app", "device", new CursorManager(cursorFile)), pub2); + t2.join(8000); + assertFalse(t2.isAlive(), "restart of a finished session should exit"); + assertTrue(pub2.batches.isEmpty(), "a finished session must not be re-uploaded on restart"); } - // ---- compressed-active: a cleanly-shut-down session whose active file - // was compressed IN PLACE to .log.gz (no rotation index) ---- + // ---- the sent window is offered to the archive queue ---- @Test - void tail_drainsCompressedActiveFile(@TempDir Path tempDir) throws Exception { - // On a clean shutdown the client compresses the active .log in - // place to .log.gz and removes the .log — there is no .log and - // no .N.log.gz. The tailer must still ingest this finished session. - Files.createDirectories(tempDir.resolve("app")); - String content = - "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n" + - "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"; - Path plain = tempDir.resolve("app/device.log"); - Files.writeString(plain, content); - Path gz = tempDir.resolve("app/device.log.gz"); - gzip(plain, gz); - Files.delete(plain); // only the compressed-active .log.gz remains - + void sentWindow_offeredToQueue(@TempDir Path dir) throws Exception { + Path gz = gzWindow(dir, "app", "device", 1, "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"); LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); - CursorManager cursorMgr = new CursorManager(tempDir.resolve("cursor.json").toFile()); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", cursorMgr, queue); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 2, 3000); + var pub = new CapturingPublisher(); + var t = startTailer(new LogTailer(dir.toFile(), "app", "device", "gpu-trace", + new CursorManager(dir.resolve("cursor.json").toFile()), queue), pub); Path consumed = queue.poll(3, TimeUnit.SECONDS); - t.interrupt(); - t.join(2000); - - assertEquals(2, publisher.events.size(), - "Both lines should be read from the compressed-active .log.gz"); - assertNotNull(consumed, "Compressed-active file should be offered to the archive queue"); + t.interrupt(); t.join(2000); + assertNotNull(consumed, "the sent window should be offered to the archive queue"); assertEquals(gz.toAbsolutePath(), consumed.toAbsolutePath()); } - - @Test - void tail_compressedActive_resumesFromOffset(@TempDir Path tempDir) throws Exception { - // Live-tail handoff: we'd already published k1 from .log when the - // client compressed it in place. Resuming from the saved offset must yield - // only the unread tail (k2) — no duplicate of k1. - Files.createDirectories(tempDir.resolve("app")); - String l1 = "{\"type\":\"kernel_event\",\"name\":\"k1\"}\n"; - String l2 = "{\"type\":\"kernel_event\",\"name\":\"k2\"}\n"; - Path plain = tempDir.resolve("app/device.log"); - Files.writeString(plain, l1 + l2); - Path gz = tempDir.resolve("app/device.log.gz"); - gzip(plain, gz); - Files.delete(plain); - - long off = l1.getBytes(StandardCharsets.UTF_8).length; // resume after k1 - File cursorFile = tempDir.resolve("cursor.json").toFile(); - Files.writeString(cursorFile.toPath(), - "{\"streams\":{\"app.device\":{\"fileIndex\":0,\"offset\":" + off + "}}}"); - - CursorManager cursorMgr = new CursorManager(cursorFile); - CapturingPublisher publisher = new CapturingPublisher(); - LogTailer tailer = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - cursorMgr, new LinkedBlockingQueue<>()); - - Thread t = startTailer(tailer, publisher); - awaitEvents(publisher.events, 1, 3000); - t.interrupt(); - t.join(2000); - - assertEquals(1, publisher.events.size(), "Only the tail after the offset should be read"); - assertTrue(publisher.events.get(0).data().contains("k2")); - } - - // ---- A finished session is not re-uploaded when the agent restarts ---- - - @Test - void tail_finishedRotatedGzSession_notReuploadedOnRestart(@TempDir Path tempDir) throws Exception { - // A finished session whose channel rotated: only .1.log.gz - // remains (no active .log). Run 1 drains it fully and tail() - // returns. Run 2 over the SAME files + cursor must NOT re-publish: the - // cursor left at the gz's EOF makes the restart skip-to-EOF and exit. - // - // Regression: the terminal "session finished" branch used to reset the - // cursor to offset 0 while keeping the file identity, so run 2 re-located - // the gz by content signature and re-uploaded the whole session - which - // the backend then rejected with 409 "session already uploaded", looping - // forever. - Files.createDirectories(tempDir.resolve("app")); - String content = - "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k1\"}\n" + - "{\"type\":\"kernel_event\",\"session_id\":\"app\",\"name\":\"k2\"}\n"; - Path plain = tempDir.resolve("app/device.1.log"); - Files.writeString(plain, content); - Path gz = tempDir.resolve("app/device.1.log.gz"); - gzip(plain, gz); - Files.delete(plain); // finished: only the rotated .gz remains, no active .log - - File cursorFile = tempDir.resolve("cursor.json").toFile(); - - // Run 1: drains the gz fully; tail() returns on "session finished". - CapturingStreamPublisher pub1 = new CapturingStreamPublisher(); - LogTailer tailer1 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null, - new StreamUploadSettings(true, 10, 1_000_000L)); - Thread t1 = startTailer(tailer1, pub1); - awaitEvents(pub1.batches, 1, 3000); - t1.join(3000); // tail() exits by itself once the finished session is drained - assertFalse(t1.isAlive(), "tailer should exit on its own after draining a finished session"); - assertEquals(1, pub1.batches.size(), "run 1 publishes the finished session once"); - - // Run 2: same files, same persisted cursor. Must NOT re-publish anything. - CapturingStreamPublisher pub2 = new CapturingStreamPublisher(); - LogTailer tailer2 = new LogTailer(tempDir.toFile(), "app", "device", "gpu-trace", - new CursorManager(cursorFile), new LinkedBlockingQueue<>(), null, - new StreamUploadSettings(true, 10, 1_000_000L)); - Thread t2 = startTailer(tailer2, pub2); - t2.join(3000); // should skip-to-EOF and exit immediately - assertFalse(t2.isAlive(), "restart of a finished session should exit immediately"); - assertTrue(pub2.batches.isEmpty(), - "a finished session must NOT be re-uploaded on restart (got " + pub2.batches + ")"); - } } diff --git a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java index 2fa1536..24d81f0 100644 --- a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java +++ b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java @@ -78,6 +78,14 @@ private static String gunzip(byte[] body) throws IOException { } } + private static byte[] gzipBytes(String s) throws IOException { + var out = new java.io.ByteArrayOutputStream(); + try (var gz = new java.util.zip.GZIPOutputStream(out)) { + gz.write(s.getBytes(StandardCharsets.UTF_8)); + } + return out.toByteArray(); + } + @Test void publish_sendsPostRequest() throws InterruptedException { HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5); @@ -185,6 +193,44 @@ void publishStream_serverError500_signalsRetry() { assertFalse(ok, "5xx is transient and must be retried"); } + @Test + void publishStreamGz_sendsVerbatimGzAsOneChunk() throws IOException { + // A complete .gz window is POSTed byte-for-byte - same endpoint/headers as + // publishStream, but no decompress/re-gzip on the agent side. + statusToReturn.set(202); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + + byte[] gz = gzipBytes( + "{\"type\":\"kernel_event\",\"session_id\":\"s2\"}\n{\"type\":\"scope_event\",\"session_id\":\"s2\"}\n"); + boolean ok = pub.publishStreamGz("s2", gz); + + assertTrue(ok); + assertEquals("/api/v1/events/stream", lastPath.get()); + assertEquals("gzip", lastContentEncoding.get()); + assertEquals("application/x-ndjson", lastContentType.get()); + assertEquals("s2", lastSessionId.get()); + assertTrue(lastBody.get().contains("\"kernel_event\"")); + assertTrue(lastBody.get().endsWith("\n")); + } + + @Test + void publishStreamGz_emptyBody_returnsTrueWithoutPosting() { + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + assertTrue(pub.publishStreamGz("s2", new byte[0]), "empty window is a no-op success"); + assertNull(lastPath.get(), "no POST for an empty body"); + } + + @Test + void publishStreamGz_alreadyUploaded409_advancesAsAccepted() throws IOException { + statusToReturn.set(409); + HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5, "stream", 100, 1_000_000L); + HttpPublisher pub = new HttpPublisher(config); + assertTrue(pub.publishStreamGz("s2", gzipBytes("{\"type\":\"kernel_event\"}\n")), + "409 already-uploaded must advance, not retry forever"); + } + @Test void publish_nonSuccessResponse_doesNotThrow() throws InterruptedException { statusToReturn.set(500); From f109aa9846e0933fc373fedffb367a45d7778547 Mon Sep 17 00:00:00 2001 From: Myoungho Shin Date: Sat, 20 Jun 2026 20:26:23 -0700 Subject: [PATCH 2/2] feat: agent exits when drained so the launcher can wait for a clean upload --- src/main/java/com/gpuflight/agent/Main.java | 96 ++++++++++++++++--- .../java/com/gpuflight/agent/MainTest.java | 32 +++++++ 2 files changed, 115 insertions(+), 13 deletions(-) diff --git a/src/main/java/com/gpuflight/agent/Main.java b/src/main/java/com/gpuflight/agent/Main.java index 9c7b9ec..8e6a520 100644 --- a/src/main/java/com/gpuflight/agent/Main.java +++ b/src/main/java/com/gpuflight/agent/Main.java @@ -32,6 +32,14 @@ static void main(String[] args) throws Exception { Publisher publisher = PublisherFactory.create(config.publisher()); System.out.println("[agent] Publisher: " + config.publisher().getClass().getSimpleName()); + // One-shot mode for the launcher-driven `gpufl trace` upload: exit once every + // discovered session has fully drained, so the launcher can wait for a clean finish + // instead of hard-killing the agent mid-upload (which dropped late windows). + boolean exitWhenDrained = parseExitWhenDrained(args, System.getenv()); + if (exitWhenDrained) { + System.out.println("[agent] exit-when-drained mode enabled"); + } + // In v1.2 a "source" is just a FOLDER to watch - each run writes its // logs under //.log[.gz], so sessions are // auto-discovered (no per-session config). Collect every folder to @@ -94,6 +102,11 @@ static void main(String[] args) throws Exception { // re-watched, so this set also marks "done" for the agent's lifetime.) var startedSessions = java.util.concurrent.ConcurrentHashMap.newKeySet(); + // Live per-channel tailer tasks across ALL sessions. The exit-when-drained monitor + // exits the agent once this returns to 0 (after >=1 session was seen and nothing is + // still being written). + var activeTailers = new java.util.concurrent.atomic.AtomicInteger(0); + // Spawn the per-channel tailer set for one discovered session. // Idempotent - a second call for the same (folder, session_id) is a // no-op, so the rescan can call it freely without re-announcing. @@ -112,19 +125,24 @@ static void main(String[] args) throws Exception { var remaining = new java.util.concurrent.atomic.AtomicInteger(session.logTypes().size()); String sid = session.sessionId(); for (String type : session.logTypes()) { + activeTailers.incrementAndGet(); executor.submit(() -> { - // Only the "system" channel carries device_metric_batch events. - var dedup = "system".equals(type) ? deduplicator : null; - var tailer = new LogTailer(session.folder(), session.sessionId(), type, - topicPrefix, cursorMgr, consumedFilesQueue, dedup, - streamUploadSettings); - tailer.tail(publisher); - // Reached only when tail() returns. Skip on interruption (agent - // shutdown) so an in-flight upload is never declared complete. - if (!Thread.currentThread().isInterrupted() - && remaining.decrementAndGet() == 0 - && streamUploadSettings.enabled()) { - signalSessionComplete(publisher, sid); + try { + // Only the "system" channel carries device_metric_batch events. + var dedup = "system".equals(type) ? deduplicator : null; + var tailer = new LogTailer(session.folder(), session.sessionId(), type, + topicPrefix, cursorMgr, consumedFilesQueue, dedup, + streamUploadSettings); + tailer.tail(publisher); + // Reached only when tail() returns. Skip on interruption (agent + // shutdown) so an in-flight upload is never declared complete. + if (!Thread.currentThread().isInterrupted() + && remaining.decrementAndGet() == 0 + && streamUploadSettings.enabled()) { + signalSessionComplete(publisher, sid); + } + } finally { + activeTailers.decrementAndGet(); } }); } @@ -188,10 +206,62 @@ static void main(String[] args) throws Exception { try { publisher.close(); } catch (Exception ignored) {} })); - // Block main thread until SIGTERM/Ctrl+C triggers the shutdown hook. + if (exitWhenDrained) { + awaitDrainThenExit(watchedFolders.keySet(), startedSessions, activeTailers); + System.out.println("[agent] all sessions drained - exiting"); + executor.shutdownNow(); + try { publisher.close(); } catch (Exception ignored) {} + return; + } + + // Daemon mode: block main thread until SIGTERM/Ctrl+C triggers the shutdown hook. new CountDownLatch(1).await(); } + /** + * Block until every discovered session has fully drained (no live tailers) and no + * session is still being written (no {@code .tmp/} dir), after at least one session + * was seen. Lets the launcher-driven one-shot {@code gpufl trace} upload wait for a + * clean finish instead of hard-killing the agent mid-upload. Two consecutive clean + * checks guard against a session discovered between ticks. + */ + private static void awaitDrainThenExit(Collection folders, + Set startedSessions, + java.util.concurrent.atomic.AtomicInteger activeTailers) + throws InterruptedException { + int clean = 0; + while (true) { + Thread.sleep(1000); + boolean drained = !startedSessions.isEmpty() + && activeTailers.get() == 0 + && !anyActiveSession(folders); + clean = drained ? clean + 1 : 0; + if (clean >= 2) return; + } + } + + /** True if any session subdir under {@code folders} still has a {@code .tmp/} working + * dir - i.e., a run is still writing it (the client removes .tmp once every channel + * closes). Guards exit-when-drained against a run that hasn't finished yet. */ + static boolean anyActiveSession(Collection folders) { + 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; + } + } + return false; + } + + /** Whether the launcher asked the agent to exit once its session(s) drain + * (GPUFL_AGENT_EXIT_WHEN_DRAINED / --exit-when-drained). */ + static boolean parseExitWhenDrained(String[] args, Map env) { + String v = resolve(args, "exit-when-drained", "GPUFL_AGENT_EXIT_WHEN_DRAINED", null, env); + return v != null && !v.equalsIgnoreCase("false") && !v.equals("0"); + } + /** * Tell the backend that every channel of a session has finished uploading. * Best-effort with a few bounded retries: {@code publishSessionComplete} diff --git a/src/test/java/com/gpuflight/agent/MainTest.java b/src/test/java/com/gpuflight/agent/MainTest.java index 891f853..057f2f4 100644 --- a/src/test/java/com/gpuflight/agent/MainTest.java +++ b/src/test/java/com/gpuflight/agent/MainTest.java @@ -461,6 +461,38 @@ void loadClasspathConfig_loadsLocalJson() throws Exception { assertEquals(".", config.source().folder()); } + // ---- exit-when-drained helpers ---- + + @Test + void parseExitWhenDrained_onForEnvOrFlag() { + assertTrue(Main.parseExitWhenDrained(new String[]{}, Map.of("GPUFL_AGENT_EXIT_WHEN_DRAINED", "1"))); + assertTrue(Main.parseExitWhenDrained(new String[]{"--exit-when-drained=true"}, Collections.emptyMap())); + } + + @Test + void parseExitWhenDrained_offWhenAbsentOrFalsey() { + assertFalse(Main.parseExitWhenDrained(new String[]{}, Collections.emptyMap())); + assertFalse(Main.parseExitWhenDrained(new String[]{}, Map.of("GPUFL_AGENT_EXIT_WHEN_DRAINED", "0"))); + assertFalse(Main.parseExitWhenDrained(new String[]{"--exit-when-drained=false"}, Collections.emptyMap())); + } + + @Test + 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(Main.anyActiveSession(List.of(folder))); + } + + @Test + void anyActiveSession_falseWhenDrainedOrEmpty(@TempDir Path dir) throws IOException { + File folder = dir.toFile(); + 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(Main.anyActiveSession(List.of(folder))); + assertFalse(Main.anyActiveSession(List.of(new File(folder, "missing")))); // non-existent folder + } + // ---- printUsage() — smoke test ---- @Test