diff --git a/.gitignore b/.gitignore
index e8bc9e4..299d43f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,7 +47,7 @@ bin/
.DS_Store
## Local config
-config
+/config/
cursor.json
run-local.sh
diff --git a/src/main/java/com/gpuflight/agent/CursorPosition.java b/src/main/java/com/gpuflight/agent/CursorPosition.java
index 117ed11..66f5123 100644
--- a/src/main/java/com/gpuflight/agent/CursorPosition.java
+++ b/src/main/java/com/gpuflight/agent/CursorPosition.java
@@ -14,8 +14,8 @@
*
* Two identity signals, because the client both renames AND compresses on
* rotation:
- * - {@code fileKey} (inode on POSIX) survives a rename/shift (active →
- * {@code .1.log} → {@code .2.log}) but NOT compression — gzip writes a new
+ * - {@code fileKey} (inode on POSIX) survives a rename/shift (active ->
+ * {@code .1.log} -> {@code .2.log}) but NOT compression - gzip writes a new
* file ({@code .1.log.gz}) with a different inode. Fast path on Linux while
* the rotated file is still uncompressed.
* - {@code headSig} (CRC32 of the first bytes of the *uncompressed* content)
@@ -24,7 +24,7 @@
* {@code .gz}.
*
*
Identity fields are nullable/zero and default that way, so cursor files
- * written before this change still deserialize cleanly (identity unknown →
+ * written before this change still deserialize cleanly (identity unknown ->
* legacy behavior).
*/
public record CursorPosition(
@@ -34,7 +34,7 @@ public record CursorPosition(
@JsonProperty("fileSize") long fileSize, // size at last update (sanity check)
@JsonProperty("headSig") long headSig // CRC32 of first uncompressed bytes; survives compression. 0 = unset.
) {
- /** Backward-compatible constructor — identity unknown. */
+ /** Backward-compatible constructor - identity unknown. */
public CursorPosition(int fileIndex, long offset) {
this(fileIndex, offset, null, 0L, 0L);
}
diff --git a/src/main/java/com/gpuflight/agent/LogArchiver.java b/src/main/java/com/gpuflight/agent/LogArchiver.java
index fa22b1c..bda40f0 100644
--- a/src/main/java/com/gpuflight/agent/LogArchiver.java
+++ b/src/main/java/com/gpuflight/agent/LogArchiver.java
@@ -32,7 +32,7 @@ public void archive(Path file, String objectKey) {
.build();
// Blocking S3 upload is fine on a virtual thread.
s3.putObject(request, file);
- System.out.println("[archiver] Uploaded " + file.getFileName() + " → s3://" + config.bucket() + "/" + objectKey);
+ System.out.println("[archiver] Uploaded " + file.getFileName() + " -> s3://" + config.bucket() + "/" + objectKey);
if (config.deleteAfterUpload()) {
file.toFile().delete();
System.out.println("[archiver] Deleted " + file.getFileName());
diff --git a/src/main/java/com/gpuflight/agent/LogTailer.java b/src/main/java/com/gpuflight/agent/LogTailer.java
index 55ce998..3dd4386 100644
--- a/src/main/java/com/gpuflight/agent/LogTailer.java
+++ b/src/main/java/com/gpuflight/agent/LogTailer.java
@@ -2,6 +2,7 @@
import tools.jackson.databind.JsonNode;
import com.gpuflight.agent.config.JsonSettings;
+import com.gpuflight.agent.config.StreamUploadSettings;
import com.gpuflight.agent.filter.DeviceMetricDeduplicator;
import com.gpuflight.agent.model.LogWrapper;
import com.gpuflight.agent.publisher.Publisher;
@@ -18,6 +19,8 @@
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.zip.CRC32;
import java.util.zip.GZIPInputStream;
@@ -28,11 +31,11 @@
* v1.2 file naming (one LogTailer per session/channel pair):
* Active file : //.log
* Rotated files: //.1.log[.gz] (newest rotated)
- * //.2.log[.gz] (older), …
+ * //.2.log[.gz] (older), ...
*
* Cursor fileIndex semantics:
- * 0 → currently tailing the active file
- * N≥1 → finishing the just-rotated file ..log[.gz] after a rotation event,
+ * 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
@@ -56,6 +59,7 @@ public class LogTailer {
private final CursorManager cursorMgr;
private final BlockingQueue consumedFilesQueue; // nullable
private final DeviceMetricDeduplicator deduplicator; // nullable
+ private final StreamUploadSettings streamUploadSettings;
private final String streamKey;
/** Max rotated index scanned during restart reconciliation. Matches the
@@ -72,6 +76,14 @@ public LogTailer(File folder, String sessionId, String logType, String topicPref
public LogTailer(File folder, String sessionId, String logType, String topicPrefix,
CursorManager cursorMgr, BlockingQueue consumedFilesQueue,
DeviceMetricDeduplicator deduplicator) {
+ this(folder, sessionId, logType, topicPrefix, cursorMgr, consumedFilesQueue,
+ deduplicator, StreamUploadSettings.DISABLED);
+ }
+
+ public LogTailer(File folder, String sessionId, String logType, String topicPrefix,
+ CursorManager cursorMgr, BlockingQueue consumedFilesQueue,
+ DeviceMetricDeduplicator deduplicator,
+ StreamUploadSettings streamUploadSettings) {
this.folder = folder;
this.sessionId = sessionId;
this.logType = logType;
@@ -79,9 +91,12 @@ public LogTailer(File folder, String sessionId, String logType, String topicPref
this.cursorMgr = cursorMgr;
this.consumedFilesQueue = consumedFilesQueue;
this.deduplicator = deduplicator;
+ this.streamUploadSettings = streamUploadSettings == null
+ ? StreamUploadSettings.DISABLED
+ : streamUploadSettings;
// Cursor key namespaces by session_id so two concurrent sessions'
// device channels (say `s1.device` and `s2.device`) don't share
- // a cursor position. Survives restart — sessions in
+ // a cursor position. Survives restart - sessions in
// cursor.json that are no longer on disk get pruned lazily.
this.streamKey = sessionId + "." + logType;
}
@@ -98,9 +113,9 @@ private File activeFile() {
/**
* The compressed-active file: {@code //.log.gz}.
- * On a CLEAN SHUTDOWN the client compresses the active file IN PLACE — the
+ * 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
+ * .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
@@ -111,7 +126,7 @@ private File activeGzFile() {
}
/**
- * The rotated file for {@code index} (≥1). Prefers the uncompressed
+ * The rotated file for {@code index} (>=1). Prefers the uncompressed
* {@code .index.log}; falls back to {@code .index.log.gz} once the client
* has compressed it. Returns null if neither exists.
*/
@@ -122,10 +137,82 @@ 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");
}
+ private static final class StreamBatch {
+ private final int maxLines;
+ private final long maxBytes;
+ private final List lines = new ArrayList<>();
+ private long bytes;
+ private long startOffset = -1L;
+ private long endOffset;
+
+ StreamBatch(StreamUploadSettings settings) {
+ this.maxLines = Math.max(1, settings.maxLines());
+ this.maxBytes = Math.max(1L, settings.maxBytes());
+ }
+
+ boolean isEmpty() {
+ return lines.isEmpty();
+ }
+
+ long startOffset() {
+ return startOffset;
+ }
+
+ long endOffset() {
+ return endOffset;
+ }
+
+ int lineCount() {
+ return lines.size();
+ }
+
+ long bytes() {
+ return bytes;
+ }
+
+ void add(String line, long lineStart, long lineEnd) {
+ if (lines.isEmpty()) {
+ startOffset = lineStart;
+ }
+ lines.add(line);
+ bytes += line.getBytes(StandardCharsets.UTF_8).length + 1L;
+ endOffset = lineEnd;
+ }
+
+ boolean shouldFlush() {
+ return lines.size() >= maxLines || bytes >= maxBytes;
+ }
+
+ List lines() {
+ return List.copyOf(lines);
+ }
+
+ void clear() {
+ lines.clear();
+ bytes = 0L;
+ startOffset = -1L;
+ endOffset = 0L;
+ }
+ }
+
// ---- File identity (D) ----
private record Identity(String fileKey, long size, long headSig) {}
@@ -147,7 +234,7 @@ private static Identity identityOf(File f) {
* CRC32 of the first {@link #HEAD_SIG_BYTES} bytes of the file's
* UNCOMPRESSED content (transparently decompresses a .gz). Because it is
* computed over content, it survives both the client's rename AND its
- * compression — it is the signal that lets a restart find the file it was
+ * compression - it is the signal that lets a restart find the file it was
* reading even after it became a .gz. Returns 0 when unreadable (the
* "unset" sentinel); never returns 0 for real content.
*/
@@ -175,7 +262,7 @@ private static boolean identityMatches(CursorPosition saved, Identity cand) {
return false;
}
- /** Locate the index (0=active, ≥1=rotated) whose current file matches the
+ /** 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();
@@ -195,7 +282,7 @@ private int locateByIdentity(CursorPosition saved) {
}
/**
- * Blocking tail loop — designed to run on a virtual thread.
+ * Blocking tail loop - designed to run on a virtual thread.
* Thread.sleep() parks the virtual thread cheaply; no carrier thread is blocked.
*/
public void tail(Publisher publisher) {
@@ -205,7 +292,7 @@ public void tail(Publisher publisher) {
// ---- 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
+ // 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
@@ -218,10 +305,10 @@ public void tail(Publisher publisher) {
idx = located;
cursorMgr.update(streamKey, idx, offset);
}
- // located == -1 → file not found anywhere; keep cursor, legacy fallback applies.
+ // located == -1 -> file not found anywhere; keep cursor, legacy fallback applies.
}
- System.out.println("[" + logType + "] Starting – fileIndex=" + idx + ", offset=" + offset);
+ 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));
@@ -250,23 +337,33 @@ public void tail(Publisher publisher) {
// 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
+ // cold (offset 0 -> whole file) or were live-tailing
// .log when it was compressed at shutdown (offset =
- // bytes already published → just the tail).
+ // 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.");
+ + 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.
+ // 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;
}
@@ -275,14 +372,31 @@ public void tail(Publisher publisher) {
if (idx > 0 && isGz(file)) {
long resume = drainGz(file, offset, publisher, idx);
if (resume < 0) {
- System.out.println("[" + logType + "] Finished " + file.getName() + " (gz), resuming active file.");
+ 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()) {
+ idx = 0;
+ offset = 0;
+ cursorMgr.update(streamKey, idx, offset);
+ 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);
} else {
- // Publish failed mid-gz — back off and retry from persisted offset.
+ // Publish failed mid-gz - back off and retry from persisted offset.
offset = resume;
sleep(5000);
}
@@ -301,17 +415,56 @@ public void tail(Publisher publisher) {
if (file.length() > offset) {
String line;
boolean publishFailed = false;
- // 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;
+ 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();
}
- offset = reader.getFilePointer();
}
cursorMgr.update(streamKey, idx, offset, id.fileKey(), id.size(), id.headSig());
if (publishFailed) {
@@ -332,12 +485,12 @@ public void tail(Publisher publisher) {
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 —
+ // 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);
+ 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 {
@@ -345,11 +498,29 @@ else if (rotated != null && (isGz(rotated) || rotated.length() >= offset)) {
}
} else {
// Finished draining an uncompressed rotated file.
- System.out.println("[" + logType + "] Finished " + file.getName() + ", resuming active 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()) {
+ idx = 0;
+ offset = 0;
+ cursorMgr.update(streamKey, idx, offset);
+ 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);
@@ -370,19 +541,58 @@ else if (rotated != null && (isGz(rotated) || rotated.length() >= offset)) {
* (EOF); otherwise the offset to resume from after a publish failure. The
* cursor is updated as progress is made, so a crash mid-drain resumes
* correctly. gz streams have no random seek, so a retry re-opens and
- * re-skips — bounded by the 50 MB rotated-file size.
+ * 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)))) {
long start = skipFully(in, startOffset);
long[] consumed = { start };
+ if (streamUploadSettings.enabled()) {
+ StreamBatch batch = new StreamBatch(streamUploadSettings);
+ while (!Thread.currentThread().isInterrupted()) {
+ long lineStart = consumed[0];
+ String line = readLineCounting(in, consumed);
+ if (line == null) {
+ if (!batch.isEmpty()) {
+ if (!flushStreamBatch(batch, publisher)) {
+ System.out.println("[" + logType + "] Stream publish FAILED (gz) - will retry in 5s");
+ cursorMgr.update(streamKey, idx, batch.startOffset(),
+ id.fileKey(), id.size(), id.headSig());
+ return batch.startOffset();
+ }
+ cursorMgr.update(streamKey, idx, batch.endOffset(),
+ id.fileKey(), id.size(), id.headSig());
+ }
+ return -1; // EOF -> fully drained
+ }
+
+ String prepared = prepareStreamLine(line);
+ if (prepared == null) {
+ cursorMgr.update(streamKey, idx, consumed[0], id.fileKey(), id.size(), id.headSig());
+ continue;
+ }
+ batch.add(prepared, lineStart, consumed[0]);
+ if (batch.shouldFlush()) {
+ if (!flushStreamBatch(batch, publisher)) {
+ System.out.println("[" + logType + "] Stream publish FAILED (gz) - will retry in 5s");
+ cursorMgr.update(streamKey, idx, batch.startOffset(),
+ id.fileKey(), id.size(), id.headSig());
+ return batch.startOffset();
+ }
+ cursorMgr.update(streamKey, idx, batch.endOffset(),
+ id.fileKey(), id.size(), id.headSig());
+ batch.clear();
+ }
+ }
+ return batch.isEmpty() ? consumed[0] : batch.startOffset();
+ }
while (!Thread.currentThread().isInterrupted()) {
long lineStart = consumed[0];
String line = readLineCounting(in, consumed);
- if (line == null) return -1; // EOF → fully drained
+ if (line == null) return -1; // EOF -> fully drained
if (!handleLine(line, publisher)) {
- System.out.println("[" + logType + "] Publish FAILED (gz) – will retry in 5s");
+ System.out.println("[" + logType + "] Publish FAILED (gz) - will retry in 5s");
cursorMgr.update(streamKey, idx, lineStart, id.fileKey(), id.size(), id.headSig());
return lineStart;
}
@@ -395,6 +605,47 @@ private long drainGz(File gz, long startOffset, Publisher publisher, int idx) {
}
}
+ private boolean flushStreamBatch(StreamBatch batch, Publisher publisher) {
+ if (batch.isEmpty()) {
+ return true;
+ }
+ System.out.println("[" + logType + "] Sending stream batch: session=" + sessionId
+ + " lines=" + batch.lineCount()
+ + " bytes=" + batch.bytes()
+ + " offset=" + batch.startOffset() + "->" + batch.endOffset());
+ boolean accepted = publisher.publishStream(sessionId, batch.lines());
+ if (accepted) {
+ System.out.println("[" + logType + "] Stream batch accepted: session=" + sessionId
+ + " lines=" + batch.lineCount()
+ + " nextOffset=" + batch.endOffset());
+ } else {
+ System.out.println("[" + logType + "] Stream batch rejected: session=" + sessionId
+ + " retryOffset=" + batch.startOffset());
+ }
+ return accepted;
+ }
+
+ /**
+ * Validate and optionally deduplicate one raw NDJSON line for stream upload.
+ * Returns null when the line should be skipped.
+ */
+ private String prepareStreamLine(String line) {
+ if (line.isBlank()) return null;
+ try {
+ JsonNode node = JsonSettings.MAPPER.readTree(line);
+ String type = node.path("type").asText("unknown");
+ if (deduplicator != null && "device_metric_batch".equals(type)) {
+ return deduplicator.filterBatch(line);
+ }
+ return line;
+ } catch (Exception e) {
+ // Occasional parse failures are expected when tailing an actively-written
+ // file - the reader may see a partial line before the writer flushes.
+ System.err.println("[LogTailer] Skipping malformed line: " + e.getMessage());
+ return null;
+ }
+ }
+
/**
* Parse, dedup, and publish one raw line. Returns true if the line was
* published or intentionally skipped (blank / malformed / all rows
@@ -403,7 +654,7 @@ private long drainGz(File gz, long startOffset, Publisher publisher, int idx) {
private boolean handleLine(String line, Publisher publisher) {
if (line.isBlank()) return true;
LogWrapper wrapper = processLine(line);
- if (wrapper == null) return true; // malformed — skip (matches prior behavior)
+ if (wrapper == null) return true; // malformed - skip (matches prior behavior)
if (deduplicator != null && "device_metric_batch".equals(wrapper.type())) {
String filtered = deduplicator.filterBatch(wrapper.data());
if (filtered == null) return true; // all rows suppressed
@@ -429,7 +680,7 @@ private LogWrapper processLine(String rawLine) {
);
} catch (Exception e) {
// Occasional parse failures are expected when tailing an actively-written
- // file — the reader may see a partial line before the writer flushes.
+ // file - the reader may see a partial line before the writer flushes.
System.err.println("[LogTailer] Skipping malformed line: " + e.getMessage());
return null;
}
diff --git a/src/main/java/com/gpuflight/agent/Main.java b/src/main/java/com/gpuflight/agent/Main.java
index f8fdb60..45f316d 100644
--- a/src/main/java/com/gpuflight/agent/Main.java
+++ b/src/main/java/com/gpuflight/agent/Main.java
@@ -4,6 +4,7 @@
import com.gpuflight.agent.config.JsonSettings;
import com.gpuflight.agent.config.KafkaConfig;
import com.gpuflight.agent.config.PublisherConfig;
+import com.gpuflight.agent.config.StreamUploadSettings;
import com.gpuflight.agent.model.AgentConfig;
import com.gpuflight.agent.model.ArchiverConfig;
import com.gpuflight.agent.model.LogSourceConfig;
@@ -31,7 +32,7 @@ static void main(String[] args) throws Exception {
Publisher publisher = PublisherFactory.create(config.publisher());
System.out.println("[agent] Publisher: " + config.publisher().getClass().getSimpleName());
- // In v1.2 a "source" is just a FOLDER to watch — each run writes its
+ // 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
// watch with its channel filter (logTypes); we discover + tail the
@@ -62,7 +63,7 @@ static void main(String[] args) throws Exception {
}
if (watchedFolders.isEmpty()) {
- System.err.println("❌ No log sources configured (set --folder, --folders, or GPUFL_SOURCE_FOLDERS)");
+ System.err.println("ERROR: No log sources configured (set --folder, --folders, or GPUFL_SOURCE_FOLDERS)");
System.exit(1);
}
@@ -71,6 +72,15 @@ static void main(String[] args) throws Exception {
var consumedFilesQueue = new LinkedBlockingQueue();
String topicPrefix = topicPrefix(config);
+ StreamUploadSettings streamUploadSettings = switch (config.publisher()) {
+ case HttpConfig http -> StreamUploadSettings.from(http);
+ default -> StreamUploadSettings.DISABLED;
+ };
+ if (streamUploadSettings.enabled()) {
+ System.out.println("[agent] HTTP upload mode: stream"
+ + " maxLines=" + streamUploadSettings.maxLines()
+ + " maxBytes=" + streamUploadSettings.maxBytes());
+ }
var executor = Executors.newVirtualThreadPerTaskExecutor();
var deduplicator = new DeviceMetricDeduplicator();
@@ -80,12 +90,12 @@ static void main(String[] args) throws Exception {
// initial spawn can all insert without losing races. A session is
// announced + spawned exactly once; the 2s rescan is then a no-op for
// it. (With the compressed-active drain in LogTailer, a finished
- // session's tailers exit on their own once drained — they are never
+ // session's tailers exit on their own once drained - they are never
// re-watched, so this set also marks "done" for the agent's lifetime.)
var startedSessions = java.util.concurrent.ConcurrentHashMap.newKeySet();
// Spawn the per-channel tailer set for one discovered session.
- // Idempotent — a second call for the same (folder, session_id) is a
+ // Idempotent - a second call for the same (folder, session_id) is a
// no-op, so the rescan can call it freely without re-announcing.
java.util.function.Consumer spawnSessionTailers = (session) -> {
String key = session.folder().getAbsolutePath() + "::" + session.sessionId();
@@ -97,7 +107,8 @@ static void main(String[] args) throws Exception {
// 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);
+ topicPrefix, cursorMgr, consumedFilesQueue, dedup,
+ streamUploadSettings);
tailer.tail(publisher);
});
}
@@ -111,7 +122,7 @@ static void main(String[] args) throws Exception {
}
// New-session watcher. The agent must notice sessions that start AFTER
- // it booted — without this, a long-running agent would only ship the
+ // it booted - without this, a long-running agent would only ship the
// sessions that existed at startup. We poll each watched folder (2s)
// instead of inotify/ReadDirectoryChangesW for portability: the latency
// is negligible vs. typical session lifetimes, and the cost is one
@@ -156,7 +167,7 @@ static void main(String[] args) throws Exception {
}
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- System.out.println("Shutting down…");
+ System.out.println("Shutting down...");
executor.shutdownNow();
try { publisher.close(); } catch (Exception ignored) {}
}));
@@ -171,7 +182,7 @@ static void main(String[] args) throws Exception {
/**
* Builds AgentConfig from CLI flags and/or environment variables.
- * Resolution order per field: CLI flag → env var → default.
+ * Resolution order per field: CLI flag -> env var -> default.
* Falls back to the bundled local.json when no flags or GPUFL_* vars are present.
*/
static AgentConfig loadFromArgs(String[] args) {
@@ -183,19 +194,19 @@ static AgentConfig loadFromArgs(String[] args, Map env) {
|| env.keySet().stream().anyMatch(k -> k.startsWith("GPUFL_"));
if (!hasAnyConfig) {
- System.out.println("No flags or GPUFL_* env vars found — using bundled local.json");
+ System.out.println("No flags or GPUFL_* env vars found - using bundled local.json");
return loadClasspathConfig("config/local.json");
}
String folder = resolve(args, "folder", "GPUFL_SOURCE_FOLDER", null, env);
String foldersEnv = resolve(args, "folders", "GPUFL_SOURCE_FOLDERS", null, env);
if (folder == null && foldersEnv == null) {
- System.err.println("❌ --folder (or env GPUFL_SOURCE_FOLDER) or --folders (or env GPUFL_SOURCE_FOLDERS) is required");
+ System.err.println("ERROR: --folder (or env GPUFL_SOURCE_FOLDER) or --folders (or env GPUFL_SOURCE_FOLDERS) is required");
printUsage();
System.exit(1);
}
String logTypesRaw = resolve(args, "log-types", "GPUFL_LOG_TYPES", null, env);
- List logTypes = parseLogTypes(logTypesRaw); // null → LogSourceConfig applies the default
+ List logTypes = parseLogTypes(logTypesRaw); // null -> LogSourceConfig applies the default
String type = require(args, "type", "GPUFL_PUBLISHER_TYPE", env);
// Reject the legacy --url / GPUFL_HTTP_URL flag with a migration
@@ -203,10 +214,10 @@ static AgentConfig loadFromArgs(String[] args, Map env) {
// HttpConfig javadoc). Without this explicit check the user
// would silently lose the URL value (resolve() would return
// null for the new --host name) and hit the IllegalArgumentException
- // out of HttpConfig's compact constructor — same outcome but
+ // out of HttpConfig's compact constructor - same outcome but
// less obvious about WHICH legacy flag is at fault.
if (resolve(args, "url", "GPUFL_HTTP_URL", null, env) != null) {
- System.err.println("❌ --url / GPUFL_HTTP_URL is no longer supported.");
+ System.err.println("ERROR: --url / GPUFL_HTTP_URL is no longer supported.");
System.err.println(" Use --host= (or env GPUFL_HTTP_HOST)");
System.err.println(" --api-version= (or env GPUFL_HTTP_API_VERSION, default: v1)");
System.err.println(" The /api/{version}/events/ path is now built automatically.");
@@ -217,14 +228,17 @@ static AgentConfig loadFromArgs(String[] args, Map env) {
require(args, "host", "GPUFL_HTTP_HOST", env),
resolve(args, "api-version", "GPUFL_HTTP_API_VERSION", HttpConfig.DEFAULT_API_VERSION, env),
resolve(args, "token", "GPUFL_HTTP_TOKEN", null, env),
- Long.parseLong(resolve(args, "timeout", "GPUFL_HTTP_TIMEOUT_SEC", "10", env)));
+ Long.parseLong(resolve(args, "timeout", "GPUFL_HTTP_TIMEOUT_SEC", "10", env)),
+ resolve(args, "upload-mode", "GPUFL_AGENT_UPLOAD_MODE", HttpConfig.DEFAULT_UPLOAD_MODE, env),
+ Integer.parseInt(resolve(args, "stream-max-lines", "GPUFL_AGENT_STREAM_MAX_LINES", "0", env)),
+ Long.parseLong(resolve(args, "stream-max-bytes", "GPUFL_AGENT_STREAM_MAX_BYTES", "0", env)));
case "kafka" -> new KafkaConfig(
require(args, "brokers", "GPUFL_KAFKA_BROKERS", env),
resolve(args, "topic-prefix", "GPUFL_KAFKA_TOPIC_PREFIX", null, env),
resolve(args, "compression", "GPUFL_KAFKA_COMPRESSION", null, env),
Integer.parseInt(resolve(args, "kafka-linger-ms", "GPUFL_KAFKA_LINGER_MS", "0", env)));
default -> {
- System.err.println("❌ Unknown publisher type: '" + type + "' (expected: http, kafka)");
+ System.err.println("ERROR: Unknown publisher type: '" + type + "' (expected: http, kafka)");
printUsage();
System.exit(1);
yield null; // unreachable
@@ -288,7 +302,7 @@ static String require(String[] args, String flag, String envVar) {
static String require(String[] args, String flag, String envVar, Map env) {
String val = resolve(args, flag, envVar, null, env);
if (val == null) {
- System.err.println("❌ --" + flag + " (or env " + envVar + ") is required");
+ System.err.println("ERROR: --" + flag + " (or env " + envVar + ") is required");
printUsage();
System.exit(1);
}
@@ -297,14 +311,14 @@ static String require(String[] args, String flag, String envVar, Map//{device,scope,system}.log[.N.log[.gz]]
+ * //{device,scope,system,sass}.log[.N.log[.gz]]
*
* Pattern matches the filenames INSIDE a session subdir (no prefix
* component). Used as a quick check that a candidate subdir looks
* like a gpufl session (vs. an unrelated user-created folder).
*/
private static final Pattern CHANNEL_FILE_PATTERN =
- Pattern.compile("^(device|scope|system)(?:\\.\\d+)?\\.log(?:\\.gz)?$");
+ Pattern.compile("^(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$");
/** Folders we've already emitted the pre-v1.2 migration warning for.
* {@link #discoverSources} runs on a 2-second rescan loop, so without
@@ -313,8 +327,11 @@ static String require(String[] args, String flag, String envVar, Map warnedLegacyFolders =
java.util.concurrent.ConcurrentHashMap.newKeySet();
- /** Default channels to tail when a source doesn't restrict them. */
- private static final List DEFAULT_LOG_TYPES = List.of("device", "scope", "system");
+ /** Default channels to tail when a source doesn't restrict them.
+ * "sass" carries the bulky SASS-disassembly / source-content artifacts
+ * split out of device.log; it must be tailed or those artifacts miss
+ * live upload. */
+ private static final List DEFAULT_LOG_TYPES = List.of("device", "scope", "system", "sass");
/** A session found under a watched folder: the folder, the session_id
* (the subdir name), and the channels to tail. Carries the session_id
@@ -323,7 +340,7 @@ record DiscoveredSession(File folder, String sessionId, List logTypes) {
/**
* v1.2: scan {@code folder} for session subdirectories. Each subdir is a
- * session — its name is the session_id; its contents are channel files
+ * session - its name is the session_id; its contents are channel files
* matching {@link #CHANNEL_FILE_PATTERN}. Returns one
* {@link DiscoveredSession} per discovered session, each carrying the
* given {@code logTypes} (or the default channels when null/empty).
@@ -333,7 +350,7 @@ record DiscoveredSession(File folder, String sessionId, List logTypes) {
* that match the pre-v1.2 flat pattern {@code ..log}
* trigger a one-time warning so a user upgrading from v1.1 sees the
* migration hint. The per-session announcement happens once in
- * spawnSessionTailers, NOT here — this method runs on a 2s rescan loop, so
+ * spawnSessionTailers, NOT here - this method runs on a 2s rescan loop, so
* logging a "discovered" line per session every tick would just spam.
*/
static List discoverSources(File folder, List logTypes) {
@@ -343,16 +360,16 @@ static List discoverSources(File folder, List logType
List types = (logTypes == null || logTypes.isEmpty()) ? DEFAULT_LOG_TYPES : logTypes;
// Legacy-format warning: if there are pre-v1.2 flat files at
- // the folder's top level, the agent won't see them — the new
+ // the folder's top level, the agent won't see them - the new
// discovery walks subdirs only.
Pattern legacyTop = Pattern.compile(
- "^(.+)\\.(device|scope|system)(?:\\.\\d+)?\\.log(?:\\.gz)?$");
+ "^(.+)\\.(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$");
File[] topFiles = folder.listFiles();
if (topFiles != null) {
for (File f : topFiles) {
if (f.isFile() && legacyTop.matcher(f.getName()).matches()) {
// Warn only the first time we see legacy files in this
- // folder — the rescan loop calls this every 2s.
+ // folder - the rescan loop calls this every 2s.
if (warnedLegacyFolders.add(folder.getAbsolutePath())) {
System.err.println(
"[agent] warning: found pre-v1.2 flat log file '" +
@@ -414,7 +431,7 @@ static String topicPrefix(AgentConfig config) {
private static AgentConfig loadExternalConfig(String path) {
var file = new File(path);
if (!file.exists()) {
- System.err.println("❌ Config file not found: '" + path + "'");
+ System.err.println("ERROR: Config file not found: '" + path + "'");
System.exit(1);
}
try {
@@ -429,7 +446,7 @@ private static AgentConfig loadExternalConfig(String path) {
private static AgentConfig loadClasspathConfig(String resourcePath) {
try (var stream = Main.class.getClassLoader().getResourceAsStream(resourcePath)) {
if (stream == null) {
- System.err.println("❌ Bundled config not found: '" + resourcePath + "'");
+ System.err.println("ERROR: Bundled config not found: '" + resourcePath + "'");
System.exit(1);
}
return JsonSettings.MAPPER.readValue(stream, AgentConfig.class);
@@ -455,7 +472,7 @@ Config file (overrides all flags):
Source (at least one of --folder or --folders is required):
--folder= Log folder path [GPUFL_SOURCE_FOLDER]
--folders= Auto-discover folders [GPUFL_SOURCE_FOLDERS]
- --log-types= Log channels to tail [GPUFL_LOG_TYPES] default: device,scope,system
+ --log-types= Log channels to tail [GPUFL_LOG_TYPES] default: device,scope,system,sass
--cursor-file= Cursor state file [GPUFL_CURSOR_FILE] default: ./cursor.json
Publisher (required):
@@ -466,6 +483,9 @@ Config file (overrides all flags):
--api-version= Backend API version [GPUFL_HTTP_API_VERSION] default: v1
--token= Bearer auth token [GPUFL_HTTP_TOKEN]
--timeout= Request timeout [GPUFL_HTTP_TIMEOUT_SEC] default: 10
+ --upload-mode= Upload protocol [GPUFL_AGENT_UPLOAD_MODE] default: stream
+ --stream-max-lines= Stream batch line limit [GPUFL_AGENT_STREAM_MAX_LINES] default: 5000
+ --stream-max-bytes= Stream batch byte limit [GPUFL_AGENT_STREAM_MAX_BYTES] default: 1000000
Kafka publisher:
--brokers= Bootstrap servers [GPUFL_KAFKA_BROKERS]
@@ -473,7 +493,7 @@ Config file (overrides all flags):
--compression= Compression codec [GPUFL_KAFKA_COMPRESSION] default: snappy
--kafka-linger-ms= Producer linger.ms [GPUFL_KAFKA_LINGER_MS] default: 100
- Archiver (optional — disabled if --archiver-endpoint is absent):
+ Archiver (optional - disabled if --archiver-endpoint is absent):
--archiver-endpoint= S3-compatible endpoint [GPUFL_ARCHIVER_ENDPOINT]
--archiver-bucket= S3 bucket [GPUFL_ARCHIVER_BUCKET]
--archiver-region= Region [GPUFL_ARCHIVER_REGION] default: nyc3
@@ -492,7 +512,7 @@ Config file (overrides all flags):
# JSON config file
gpufl-agent --config=/etc/gpuflight/agent.json
- # No args → uses bundled local.json (development only)
+ # No args -> uses bundled local.json (development only)
gpufl-agent
""");
}
diff --git a/src/main/java/com/gpuflight/agent/config/HttpConfig.java b/src/main/java/com/gpuflight/agent/config/HttpConfig.java
index 4ab1eaa..6fc1b8a 100644
--- a/src/main/java/com/gpuflight/agent/config/HttpConfig.java
+++ b/src/main/java/com/gpuflight/agent/config/HttpConfig.java
@@ -2,22 +2,24 @@
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Locale;
+
/**
* Configuration for the HTTP publisher.
*
- * The destination is split across {@code hostUrl} + {@code apiVersion} —
+ *
The destination is split across {@code hostUrl} + {@code apiVersion};
* the full path is built internally via {@link #endpointFor(String)} so
- * the user only specifies what changes per-deployment (the host) and
- * occasionally per-rollout (the version). The {@code /api/{version}/events/}
+ * the user only specifies what changes per deployment (the host) and
+ * occasionally per rollout (the version). The {@code /api/{version}/events/}
* path is a structural contract baked into this record.
*
*
Examples:
*
* hostUrl="https://api.gpuflight.com" apiVersion="v1"
- * → endpointFor("init") = "https://api.gpuflight.com/api/v1/events/init"
+ * endpointFor("init") = "https://api.gpuflight.com/api/v1/events/init"
*
* hostUrl="http://localhost:8080" apiVersion="v2"
- * → endpointFor("init") = "http://localhost:8080/api/v2/events/init"
+ * endpointFor("init") = "http://localhost:8080/api/v2/events/init"
*
*
* Migration note (May 2026): previous releases used a
@@ -33,23 +35,53 @@ public record HttpConfig(
@JsonProperty("hostUrl") String hostUrl,
@JsonProperty("apiVersion") String apiVersion,
@JsonProperty("authToken") String authToken,
- @JsonProperty("timeoutSeconds") long timeoutSeconds
+ @JsonProperty("timeoutSeconds") long timeoutSeconds,
+ @JsonProperty("uploadMode") String uploadMode,
+ @JsonProperty("streamMaxLines") int streamMaxLines,
+ @JsonProperty("streamMaxBytes") long streamMaxBytes
) implements PublisherConfig {
/** Default API version when the user doesn't specify one. */
public static final String DEFAULT_API_VERSION = "v1";
+ public static final String DEFAULT_UPLOAD_MODE = "stream";
+ public static final int DEFAULT_STREAM_MAX_LINES = 5_000;
+ public static final long DEFAULT_STREAM_MAX_BYTES = 1_000_000L;
+
+ private static final String LEGACY_UPLOAD_MODE = "legacy";
+
+ public HttpConfig(String hostUrl, String apiVersion, String authToken, long timeoutSeconds) {
+ this(hostUrl, apiVersion, authToken, timeoutSeconds, DEFAULT_UPLOAD_MODE, 0, 0);
+ }
public HttpConfig {
// Field-level defaults / normalization. Run inside the compact
- // constructor so they apply to BOTH the JSON-deserialized path
- // (via Jackson) and any direct callers (tests, CLI plumbing).
-
+ // constructor so they apply to both JSON deserialization and direct callers.
if (timeoutSeconds <= 0) {
timeoutSeconds = 10L;
}
if (apiVersion == null || apiVersion.isBlank()) {
apiVersion = DEFAULT_API_VERSION;
}
+ if (uploadMode == null || uploadMode.isBlank()) {
+ uploadMode = DEFAULT_UPLOAD_MODE;
+ }
+ uploadMode = uploadMode.trim().toLowerCase(Locale.ROOT);
+ if (!DEFAULT_UPLOAD_MODE.equals(uploadMode) && !LEGACY_UPLOAD_MODE.equals(uploadMode)) {
+ throw new IllegalArgumentException(
+ "HttpConfig.uploadMode must be 'stream' or 'legacy', got: " + uploadMode);
+ }
+ if (DEFAULT_UPLOAD_MODE.equals(uploadMode)) {
+ if (streamMaxLines <= 0) {
+ streamMaxLines = DEFAULT_STREAM_MAX_LINES;
+ }
+ if (streamMaxBytes <= 0) {
+ streamMaxBytes = DEFAULT_STREAM_MAX_BYTES;
+ }
+ } else {
+ streamMaxLines = 0;
+ streamMaxBytes = 0L;
+ }
+
// Hard reject blank/missing host with a migration-aware error.
// Jackson silently sets unknown fields to null (we disabled
// FAIL_ON_UNKNOWN_PROPERTIES in JsonSettings), so a config that
@@ -73,11 +105,19 @@ public record HttpConfig(
}
}
+ public boolean isStreamMode() {
+ return DEFAULT_UPLOAD_MODE.equals(uploadMode);
+ }
+
+ public String streamEndpoint() {
+ return endpointFor(DEFAULT_UPLOAD_MODE);
+ }
+
/**
* Build the full upload URL for a given event type.
*
* @param type the event-type segment, e.g. {@code "init"},
- * {@code "kernel"}, {@code "metric"} — appended after
+ * {@code "kernel"}, {@code "metric"}; appended after
* the {@code /api/{version}/events/} prefix.
*/
public String endpointFor(String type) {
diff --git a/src/main/java/com/gpuflight/agent/config/StreamUploadSettings.java b/src/main/java/com/gpuflight/agent/config/StreamUploadSettings.java
new file mode 100644
index 0000000..550dbc9
--- /dev/null
+++ b/src/main/java/com/gpuflight/agent/config/StreamUploadSettings.java
@@ -0,0 +1,24 @@
+package com.gpuflight.agent.config;
+
+/**
+ * Per-tailer batching knobs for stream upload mode. When {@code enabled}, the
+ * tailer collects lines into a batch and hands the whole batch to
+ * {@code Publisher.publishStream} - flushing when {@code maxLines} or
+ * {@code maxBytes} (uncompressed) is reached, or at EOF (no artificial
+ * latency: the moment the file has no more data, whatever is buffered ships).
+ * The cursor advances only after a flush is accepted, so a crash or rejected
+ * batch re-reads from the last accepted offset.
+ *
+ *
{@code DISABLED} keeps the original one-publish-per-line flow (legacy
+ * mode, Kafka, and existing tests).
+ */
+public record StreamUploadSettings(boolean enabled, int maxLines, long maxBytes) {
+
+ public static final StreamUploadSettings DISABLED = new StreamUploadSettings(false, 0, 0);
+
+ public static StreamUploadSettings from(HttpConfig http) {
+ return http.isStreamMode()
+ ? new StreamUploadSettings(true, http.streamMaxLines(), http.streamMaxBytes())
+ : DISABLED;
+ }
+}
diff --git a/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java b/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java
index 833fcc3..50fde06 100644
--- a/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java
+++ b/src/main/java/com/gpuflight/agent/model/LogSourceConfig.java
@@ -20,6 +20,8 @@ public record LogSourceConfig(
) {
public LogSourceConfig {
if (folder == null) folder = ".";
- if (logTypes == null || logTypes.isEmpty()) logTypes = List.of("device", "scope", "system");
+ // "sass" carries the SASS-disassembly / source-content artifacts split
+ // out of device.log; tail it by default or they miss live upload.
+ if (logTypes == null || logTypes.isEmpty()) logTypes = List.of("device", "scope", "system", "sass");
}
}
diff --git a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java
index 42bc29e..d2979ae 100644
--- a/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java
+++ b/src/main/java/com/gpuflight/agent/publisher/HttpPublisher.java
@@ -4,11 +4,16 @@
import com.gpuflight.agent.config.JsonSettings;
import com.gpuflight.agent.model.LogWrapper;
+import java.io.ByteArrayOutputStream;
+import java.net.InetAddress;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.util.List;
+import java.util.zip.GZIPOutputStream;
public class HttpPublisher implements Publisher {
private final HttpConfig config;
@@ -25,11 +30,15 @@ public HttpPublisher(HttpConfig config) {
@Override
public boolean publish(String topic, String key, LogWrapper log) {
try {
- // Path is structural — assembled inside HttpConfig from
+ // Path is structural - assembled inside HttpConfig from
// hostUrl + apiVersion + event-type segment. See
// HttpConfig.endpointFor() for the template.
String url = config.endpointFor(log.type());
String message = JsonSettings.MAPPER.writeValueAsString(log);
+ System.out.println("[agent] HTTP event POST starting: url=" + url
+ + " topic=" + topic
+ + " key=" + key
+ + " type=" + log.type());
var requestBuilder = HttpRequest.newBuilder()
.uri(URI.create(url))
@@ -38,23 +47,23 @@ public boolean publish(String topic, String key, LogWrapper log) {
.header("X-Key", key)
.POST(HttpRequest.BodyPublishers.ofString(message));
- if (config.authToken() != null) {
- requestBuilder.header("Authorization", "Bearer " + config.authToken());
- }
+ addAuthHeader(requestBuilder);
- // Blocking send is fine on a virtual thread — the carrier thread parks, not blocks.
+ // Blocking send is fine on a virtual thread - the carrier thread parks, not blocks.
var response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofString());
if (response.statusCode() >= 200 && response.statusCode() <= 299) {
+ System.out.println("[agent] HTTP event POST accepted: status="
+ + response.statusCode() + " type=" + log.type());
return true;
}
if (response.statusCode() == 402) {
- // GPU limit exceeded — permanent failure, do not retry
+ // GPU limit exceeded - permanent failure, do not retry
System.err.println("[GPUFL] ========================================");
System.err.println("[GPUFL] GPU limit exceeded for this workspace.");
System.err.println("[GPUFL] " + response.body());
System.err.println("[GPUFL] Upgrade at https://gpuflight.com/pricing");
System.err.println("[GPUFL] ========================================");
- return true; // advance cursor — retrying won't help
+ return true; // advance cursor - retrying won't help
}
System.out.println("HTTP Publish failed [" + url + "]: " + response.statusCode() + " - " + response.body());
return false;
@@ -67,6 +76,62 @@ public boolean publish(String topic, String key, LogWrapper log) {
}
}
+ @Override
+ public boolean publishStream(String sessionId, List ndjsonLines) {
+ if (ndjsonLines == null || ndjsonLines.isEmpty()) {
+ return true;
+ }
+ 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);
+
+ 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));
+
+ 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);
+ return true;
+ }
+ System.out.println("HTTP stream publish failed [" + url + "]: "
+ + response.statusCode() + " - " + response.body());
+ return false;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ } catch (Exception e) {
+ System.out.println("HTTP stream connection error: " + e);
+ return false;
+ }
+ }
+
+ private void addAuthHeader(HttpRequest.Builder requestBuilder) {
+ if (config.authToken() != null && !config.authToken().isBlank()) {
+ requestBuilder.header("Authorization", "Bearer " + config.authToken());
+ }
+ }
+
+ private static byte[] gzip(byte[] content) throws Exception {
+ var out = new ByteArrayOutputStream();
+ try (var gzip = new GZIPOutputStream(out)) {
+ gzip.write(content);
+ }
+ return out.toByteArray();
+ }
+
@Override
public void close() {
// Java HttpClient has no explicit close; resources are reclaimed on GC.
diff --git a/src/main/java/com/gpuflight/agent/publisher/KafkaPublisher.java b/src/main/java/com/gpuflight/agent/publisher/KafkaPublisher.java
index b502278..4c8c8a4 100644
--- a/src/main/java/com/gpuflight/agent/publisher/KafkaPublisher.java
+++ b/src/main/java/com/gpuflight/agent/publisher/KafkaPublisher.java
@@ -27,7 +27,7 @@ public boolean publish(String topic, String key, LogWrapper log) {
try {
String message = JsonSettings.MAPPER.writeValueAsString(log.data());
// producer.send() is async and only blocks when the internal buffer is full.
- // Blocking on a virtual thread is fine — carrier thread parks, not blocks.
+ // Blocking on a virtual thread is fine - carrier thread parks, not blocks.
producer.send(new ProducerRecord<>(topic, key, message)).get();
return true;
} catch (Exception e) {
diff --git a/src/main/java/com/gpuflight/agent/publisher/Publisher.java b/src/main/java/com/gpuflight/agent/publisher/Publisher.java
index 545d76e..8228d41 100644
--- a/src/main/java/com/gpuflight/agent/publisher/Publisher.java
+++ b/src/main/java/com/gpuflight/agent/publisher/Publisher.java
@@ -3,8 +3,14 @@
import com.gpuflight.agent.model.LogWrapper;
import java.io.Closeable;
+import java.util.List;
public interface Publisher extends Closeable {
/** Returns true if the event was accepted by the backend, false on failure. */
boolean publish(String topic, String key, LogWrapper log);
+
+ /** Returns true if the NDJSON batch was accepted by the backend. */
+ default boolean publishStream(String sessionId, List ndjsonLines) {
+ return false;
+ }
}
diff --git a/src/test/java/com/gpuflight/agent/LogTailerTest.java b/src/test/java/com/gpuflight/agent/LogTailerTest.java
index 50ce7ce..bdd2a72 100644
--- a/src/test/java/com/gpuflight/agent/LogTailerTest.java
+++ b/src/test/java/com/gpuflight/agent/LogTailerTest.java
@@ -1,5 +1,6 @@
package com.gpuflight.agent;
+import com.gpuflight.agent.config.StreamUploadSettings;
import com.gpuflight.agent.model.LogWrapper;
import com.gpuflight.agent.publisher.Publisher;
import org.junit.jupiter.api.Test;
@@ -54,6 +55,26 @@ public boolean publish(String topic, String key, LogWrapper log) {
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 publishStream(String sessionId, List ndjsonLines) {
+ sessionIds.add(sessionId);
+ batches.add(List.copyOf(ndjsonLines));
+ return true;
+ }
+
+ @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);
@@ -78,6 +99,28 @@ private static void awaitEvents(List list, int count, long timeoutMs) throws
// ---- 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());
+ }
+
@Test
void tail_readsExistingLinesAndPublishes(@TempDir Path tempDir) throws Exception {
// v1.2: log files live under //.log.
@@ -192,6 +235,65 @@ void tail_persistsCursorAfterReading(@TempDir Path tempDir) throws Exception {
assertTrue(pos.offset() > 0, "Cursor offset should have advanced past the line");
}
+ @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");
+ }
+
+ @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\""));
+ }
+
// ---- waits when active file is absent ----
@Test
diff --git a/src/test/java/com/gpuflight/agent/MainTest.java b/src/test/java/com/gpuflight/agent/MainTest.java
index 3885d7a..891f853 100644
--- a/src/test/java/com/gpuflight/agent/MainTest.java
+++ b/src/test/java/com/gpuflight/agent/MainTest.java
@@ -97,7 +97,7 @@ void logTypesOrDefault_usesExplicitFolderLogTypes() {
@Test
void logTypesOrDefault_usesDefaultWhenAbsent() {
- assertEquals(List.of("device", "scope", "system"),
+ assertEquals(List.of("device", "scope", "system", "sass"),
Main.logTypesOrDefault(new String[]{}, Collections.emptyMap()));
}
diff --git a/src/test/java/com/gpuflight/agent/config/HttpConfigUploadModeTest.java b/src/test/java/com/gpuflight/agent/config/HttpConfigUploadModeTest.java
new file mode 100644
index 0000000..ced183f
--- /dev/null
+++ b/src/test/java/com/gpuflight/agent/config/HttpConfigUploadModeTest.java
@@ -0,0 +1,52 @@
+package com.gpuflight.agent.config;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/** Upload-mode + stream-batch fields on HttpConfig. */
+class HttpConfigUploadModeTest {
+
+ @Test
+ void defaults_streamModeWithBatchLimits() {
+ HttpConfig c = new HttpConfig("https://api.gpuflight.com", "v1", null, 10,
+ null, 0, 0L);
+ assertEquals("stream", c.uploadMode());
+ assertTrue(c.isStreamMode());
+ assertEquals(HttpConfig.DEFAULT_STREAM_MAX_LINES, c.streamMaxLines());
+ assertEquals(HttpConfig.DEFAULT_STREAM_MAX_BYTES, c.streamMaxBytes());
+ assertEquals("https://api.gpuflight.com/api/v1/events/stream", c.streamEndpoint());
+ }
+
+ @Test
+ void legacyFourFieldConstructor_getsStreamDefaults() {
+ HttpConfig c = new HttpConfig("https://api.gpuflight.com", "v1", null, 10);
+ assertTrue(c.isStreamMode());
+ assertEquals(HttpConfig.DEFAULT_STREAM_MAX_LINES, c.streamMaxLines());
+ }
+
+ @Test
+ void legacyMode_disablesStream() {
+ HttpConfig c = new HttpConfig("https://api.gpuflight.com", "v1", null, 10,
+ "legacy", 0, 0L);
+ assertFalse(c.isStreamMode());
+ assertEquals(StreamUploadSettings.DISABLED, StreamUploadSettings.from(c));
+ }
+
+ @Test
+ void streamMode_settingsCarryLimits() {
+ HttpConfig c = new HttpConfig("https://api.gpuflight.com", "v1", null, 10,
+ "stream", 250, 1_000_000L);
+ StreamUploadSettings s = StreamUploadSettings.from(c);
+ assertTrue(s.enabled());
+ assertEquals(250, s.maxLines());
+ assertEquals(1_000_000L, s.maxBytes());
+ }
+
+ @Test
+ void unknownUploadMode_rejectedAtConstruction() {
+ assertThrows(IllegalArgumentException.class, () ->
+ new HttpConfig("https://api.gpuflight.com", "v1", null, 10,
+ "carrier-pigeon", 0, 0L));
+ }
+}
diff --git a/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java b/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java
index a959a9e..326e9d6 100644
--- a/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java
+++ b/src/test/java/com/gpuflight/agent/model/LogSourceConfigTest.java
@@ -17,7 +17,7 @@ void nullFolder_defaultsToDot() {
@Test
void nullLogTypes_defaultsToAllChannels() {
LogSourceConfig config = new LogSourceConfig("/logs", null);
- assertEquals(List.of("device", "scope", "system"), config.logTypes());
+ assertEquals(List.of("device", "scope", "system", "sass"), config.logTypes());
}
@Test
diff --git a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java
index a07cd52..ff245a5 100644
--- a/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java
+++ b/src/test/java/com/gpuflight/agent/publisher/HttpPublisherTest.java
@@ -7,10 +7,14 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.GZIPInputStream;
import static org.junit.jupiter.api.Assertions.*;
@@ -21,13 +25,26 @@ class HttpPublisherTest {
private final AtomicInteger statusToReturn = new AtomicInteger(200);
private final AtomicReference lastBody = new AtomicReference<>();
private final AtomicReference lastAuthHeader = new AtomicReference<>();
+ private final AtomicReference lastContentEncoding = new AtomicReference<>();
+ private final AtomicReference lastContentType = new AtomicReference<>();
+ private final AtomicReference lastPath = new AtomicReference<>();
+ private final AtomicReference lastSessionId = new AtomicReference<>();
@BeforeEach
void startServer() throws IOException {
server = HttpServer.create(new InetSocketAddress(0), 0);
server.createContext("/api/", exchange -> {
- lastBody.set(new String(exchange.getRequestBody().readAllBytes()));
- lastAuthHeader.set(exchange.getRequestHeaders().getFirst("Authorization"));
+ var headers = exchange.getRequestHeaders();
+ byte[] body = exchange.getRequestBody().readAllBytes();
+ String encoding = headers.getFirst("Content-Encoding");
+ lastBody.set("gzip".equalsIgnoreCase(encoding)
+ ? gunzip(body)
+ : new String(body, StandardCharsets.UTF_8));
+ lastAuthHeader.set(headers.getFirst("Authorization"));
+ lastContentEncoding.set(encoding);
+ lastContentType.set(headers.getFirst("Content-Type"));
+ lastPath.set(exchange.getRequestURI().getPath());
+ lastSessionId.set(headers.getFirst("X-GpuFlight-Session-Id"));
exchange.sendResponseHeaders(statusToReturn.get(), 0);
exchange.close();
});
@@ -55,6 +72,12 @@ private LogWrapper sampleLog(String type) {
return new LogWrapper(System.currentTimeMillis(), "{\"key\":\"val\"}", type, "testhost", "127.0.0.1");
}
+ private static String gunzip(byte[] body) throws IOException {
+ try (var gzip = new GZIPInputStream(new ByteArrayInputStream(body))) {
+ return new String(gzip.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ }
+
@Test
void publish_sendsPostRequest() throws InterruptedException {
HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5);
@@ -90,6 +113,36 @@ void publish_noToken_doesNotSendAuthHeader() throws InterruptedException {
assertNull(lastAuthHeader.get());
}
+ @Test
+ void publish_blankToken_doesNotSendAuthHeader() {
+ HttpConfig config = new HttpConfig(hostUrl(), "v1", "", 5);
+ HttpPublisher pub = new HttpPublisher(config);
+
+ pub.publish("topic", "system", sampleLog("system_event"));
+
+ assertNull(lastAuthHeader.get());
+ }
+
+ @Test
+ void publishStream_sendsGzippedNdjsonBatch() {
+ statusToReturn.set(202);
+ HttpConfig config = new HttpConfig(hostUrl(), "v1", null, 5,
+ "stream", 100, 1_000_000L);
+ HttpPublisher pub = new HttpPublisher(config);
+
+ boolean ok = pub.publishStream("session-1", List.of(
+ "{\"type\":\"kernel_event\",\"session_id\":\"session-1\"}",
+ "{\"type\":\"scope_event\",\"session_id\":\"session-1\"}"));
+
+ assertTrue(ok);
+ assertEquals("/api/v1/events/stream", lastPath.get());
+ assertEquals("gzip", lastContentEncoding.get());
+ assertEquals("application/x-ndjson", lastContentType.get());
+ assertEquals("session-1", lastSessionId.get());
+ assertTrue(lastBody.get().contains("\"kernel_event\""));
+ assertTrue(lastBody.get().endsWith("\n"));
+ }
+
@Test
void publish_nonSuccessResponse_doesNotThrow() throws InterruptedException {
statusToReturn.set(500);