diff --git a/.gitignore b/.gitignore index 299d43f..cee17b9 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ build/ out/ !**/src/main/**/out/ !**/src/test/**/out/ +.junie ### Kotlin ### .kotlin diff --git a/build.gradle b/build.gradle index ceb986e..11de33b 100644 --- a/build.gradle +++ b/build.gradle @@ -61,7 +61,10 @@ jacoco { def jacocoExcludes = [ '**/LogArchiver.class', '**/KafkaPublisher.class', - '**/Main.class' + '**/Main.class', + '**/GpuflAgent.class', + '**/SessionWatcher.class', + '**/TailerManager.class' ] jacocoTestReport { diff --git a/src/main/java/com/gpuflight/agent/GpuflAgent.java b/src/main/java/com/gpuflight/agent/GpuflAgent.java new file mode 100644 index 0000000..4869e51 --- /dev/null +++ b/src/main/java/com/gpuflight/agent/GpuflAgent.java @@ -0,0 +1,189 @@ +package com.gpuflight.agent; + +import com.gpuflight.agent.config.ConfigLoader; +import com.gpuflight.agent.config.HttpConfig; +import com.gpuflight.agent.config.KafkaConfig; +import com.gpuflight.agent.config.StreamUploadSettings; +import com.gpuflight.agent.filter.DeviceMetricDeduplicator; +import com.gpuflight.agent.model.AgentConfig; +import com.gpuflight.agent.model.DiscoveredSession; +import com.gpuflight.agent.model.LogSourceConfig; +import com.gpuflight.agent.publisher.Publisher; +import com.gpuflight.agent.publisher.PublisherFactory; +import com.gpuflight.agent.service.SessionWatcher; +import com.gpuflight.agent.service.TailerManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.nio.file.Path; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; + +public class GpuflAgent { + + private static final Logger log = LoggerFactory.getLogger(GpuflAgent.class); + + private final AgentConfig config; + private final String[] args; + private final Map env; + + private ExecutorService executor; + private Publisher publisher; + private TailerManager tailerManager; + private final Map> watchedFolders = new LinkedHashMap<>(); + + public GpuflAgent(AgentConfig config, String[] args, Map env) { + this.config = config; + this.args = args; + this.env = env; + } + + public void start() throws Exception { + publisher = PublisherFactory.create(config.publisher()); + log.info("Publisher: {}", config.publisher().getClass().getSimpleName()); + + boolean exitWhenDrained = ConfigLoader.parseExitWhenDrained(args, env); + + resolveWatchedFolders(); + + if (watchedFolders.isEmpty()) { + System.err.println("ERROR: No log sources configured (set --folder, --folders, or GPUFL_SOURCE_FOLDERS)"); + System.exit(1); + } + + String cursorFile = ConfigLoader.resolve(args, "cursor-file", "GPUFL_CURSOR_FILE", "./cursor.json", env); + var cursorMgr = new CursorManager(new File(cursorFile)); + 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()) { + log.info("HTTP upload mode: stream maxLines={} maxBytes={}", + streamUploadSettings.maxLines(), streamUploadSettings.maxBytes()); + } + + executor = Executors.newVirtualThreadPerTaskExecutor(); + var deduplicator = new DeviceMetricDeduplicator(); + + tailerManager = new TailerManager(executor, publisher, cursorMgr, consumedFilesQueue, + deduplicator, streamUploadSettings, topicPrefix); + + // Initial discovery + spawn + for (var entry : watchedFolders.entrySet()) { + for (DiscoveredSession s : SessionWatcher.discoverSources(entry.getKey(), entry.getValue())) { + tailerManager.spawnSessionTailers(s); + } + } + + // Start watchers + for (var entry : watchedFolders.entrySet()) { + new SessionWatcher(entry.getKey(), entry.getValue(), tailerManager::spawnSessionTailers) + .start(executor); + } + + if (config.archiver() != null) { + var archiver = new LogArchiver(config.archiver()); + executor.submit(() -> { + while (!Thread.currentThread().isInterrupted()) { + try { + Path path = consumedFilesQueue.take(); + String objectKey = ConfigLoader.buildArchiveKey(config.archiver().prefix(), path); + archiver.archive(path, objectKey); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } catch (Exception e) { + System.err.println("[archiver] Error: " + e.getMessage()); + } + } + }); + } + + Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown)); + + if (exitWhenDrained) { + log.info("exit-when-drained mode enabled"); + awaitDrainThenExit(watchedFolders.keySet(), tailerManager.getActiveTailers()); + log.info("all sessions drained - exiting"); + shutdown(); + return; + } + + // Daemon mode + new CountDownLatch(1).await(); + } + + private void resolveWatchedFolders() { + if (config.source() != null) { + log.info("Source folder: {} types={}", config.source().folder(), config.source().logTypes()); + watchedFolders.putIfAbsent(new File(config.source().folder()), config.source().logTypes()); + } + if (config.sources() != null) { + for (LogSourceConfig s : config.sources()) { + log.info("Source folder: {} types={}", s.folder(), s.logTypes()); + watchedFolders.putIfAbsent(new File(s.folder()), s.logTypes()); + } + } + + String foldersRaw = ConfigLoader.resolve(args, "folders", "GPUFL_SOURCE_FOLDERS", null, env); + List folderLogTypes = ConfigLoader.logTypesOrDefault(args, env); + if (foldersRaw != null) { + for (String folderPath : foldersRaw.split(",")) { + folderPath = folderPath.trim(); + if (!folderPath.isEmpty()) { + watchedFolders.putIfAbsent(new File(folderPath), folderLogTypes); + } + } + } + } + + private void shutdown() { + log.info("Shutting down..."); + if (executor != null) executor.shutdownNow(); + try { if (publisher != null) publisher.close(); } catch (Exception ignored) {} + } + + void awaitDrainThenExit(Collection folders, AtomicInteger activeTailers) + throws InterruptedException { + boolean sawActive = false; + int clean = 0; + while (true) { + Thread.sleep(1000); + boolean activeNow = anyActiveSession(folders); + if (activeNow) sawActive = true; + boolean drained = sawActive && activeTailers.get() == 0 && !activeNow; + clean = drained ? clean + 1 : 0; + if (clean >= 2) return; + } + } + + 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; + } + + static String topicPrefix(AgentConfig config) { + return switch (config.publisher()) { + case KafkaConfig kafka -> kafka.topicPrefix(); + case HttpConfig ignored -> "gpu-trace"; + }; + } +} diff --git a/src/main/java/com/gpuflight/agent/Main.java b/src/main/java/com/gpuflight/agent/Main.java index 8e6a520..bd8643d 100644 --- a/src/main/java/com/gpuflight/agent/Main.java +++ b/src/main/java/com/gpuflight/agent/Main.java @@ -1,632 +1,12 @@ package com.gpuflight.agent; -import com.gpuflight.agent.config.HttpConfig; -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.config.ConfigLoader; import com.gpuflight.agent.model.AgentConfig; -import com.gpuflight.agent.model.ArchiverConfig; -import com.gpuflight.agent.model.LogSourceConfig; -import com.gpuflight.agent.filter.DeviceMetricDeduplicator; -import com.gpuflight.agent.publisher.Publisher; -import com.gpuflight.agent.publisher.PublisherFactory; - -import java.io.File; -import java.nio.file.Path; -import java.util.*; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.regex.Matcher; -import java.util.regex.Pattern; public class Main { - - static void main(String[] args) throws Exception { - String configPath = parseConfigArg(args); - AgentConfig config = configPath != null - ? loadExternalConfig(configPath) - : loadFromArgs(args); - - 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 - // watch with its channel filter (logTypes); we discover + tail the - // sessions under each, both now and on the periodic rescan below. - var watchedFolders = new java.util.LinkedHashMap>(); - if (config.source() != null) { - System.out.println("[agent] Source folder: " + config.source().folder() - + " types=" + config.source().logTypes()); - watchedFolders.putIfAbsent(new File(config.source().folder()), config.source().logTypes()); - } - if (config.sources() != null) { - for (LogSourceConfig s : config.sources()) { - System.out.println("[agent] Source folder: " + s.folder() - + " types=" + s.logTypes()); - watchedFolders.putIfAbsent(new File(s.folder()), s.logTypes()); - } - } - - String foldersRaw = resolve(args, "folders", "GPUFL_SOURCE_FOLDERS", null); - List folderLogTypes = logTypesOrDefault(args, System.getenv()); - if (foldersRaw != null) { - for (String folderPath : foldersRaw.split(",")) { - folderPath = folderPath.trim(); - if (!folderPath.isEmpty()) { - watchedFolders.putIfAbsent(new File(folderPath), folderLogTypes); - } - } - } - - if (watchedFolders.isEmpty()) { - System.err.println("ERROR: No log sources configured (set --folder, --folders, or GPUFL_SOURCE_FOLDERS)"); - System.exit(1); - } - - String cursorFile = resolve(args, "cursor-file", "GPUFL_CURSOR_FILE", "./cursor.json"); - var cursorMgr = new CursorManager(new File(cursorFile)); - 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(); - - // Set of "::" keys already being tailed. - // ConcurrentHashMap.newKeySet() so the watcher threads and the - // 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 - // 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. - java.util.function.Consumer spawnSessionTailers = (session) -> { - String key = session.folder().getAbsolutePath() + "::" + session.sessionId(); - if (!startedSessions.add(key)) return; - System.out.println("[agent] Tailing session \"" + session.sessionId() - + "\" in " + session.folder() + " types=" + session.logTypes()); - // Count this session's per-channel tailers. When the LAST one finishes - // NATURALLY (session drained + every batch accepted), tell the backend the - // upload is complete so it can finalize without waiting out its grace - // window. A tailer that exits via interruption (agent shutdown) or never - // finishes (a channel whose file never appears) does NOT count toward the - // signal — so a partial upload is never declared complete; the backend's - // grace path covers those. - var remaining = new java.util.concurrent.atomic.AtomicInteger(session.logTypes().size()); - String sid = session.sessionId(); - for (String type : session.logTypes()) { - activeTailers.incrementAndGet(); - executor.submit(() -> { - 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(); - } - }); - } - }; - - // Initial discovery + spawn for every watched folder. - for (var entry : watchedFolders.entrySet()) { - for (DiscoveredSession s : discoverSources(entry.getKey(), entry.getValue())) { - spawnSessionTailers.accept(s); - } - } - - // New-session watcher. The agent must notice sessions that start AFTER - // 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 - // listing per folder per tick. - for (var entry : watchedFolders.entrySet()) { - File folder = entry.getKey(); - List types = entry.getValue(); - executor.submit(() -> { - while (!Thread.currentThread().isInterrupted()) { - try { - for (DiscoveredSession s : discoverSources(folder, types)) { - spawnSessionTailers.accept(s); - } - Thread.sleep(2_000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } catch (Exception e) { - System.err.println("[watcher] error scanning " + - folder + ": " + e.getMessage()); - } - } - }); - } - - if (config.archiver() != null) { - var archiver = new LogArchiver(config.archiver()); - executor.submit(() -> { - while (!Thread.currentThread().isInterrupted()) { - try { - Path path = consumedFilesQueue.take(); - String objectKey = buildArchiveKey(config.archiver().prefix(), path); - archiver.archive(path, objectKey); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } catch (Exception e) { - System.err.println("[archiver] Error: " + e.getMessage()); - } - } - }); - } - - Runtime.getRuntime().addShutdownHook(new Thread(() -> { - System.out.println("Shutting down..."); - executor.shutdownNow(); - try { publisher.close(); } catch (Exception ignored) {} - })); - - 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} - * returns true for terminal outcomes (2xx, or a 4xx like 404 from an older - * backend) and false only for transient 5xx/network errors. If it never gets - * through, the backend's grace path still finalizes the session, so this is - * a pure latency optimization — never a correctness dependency. - */ - private static void signalSessionComplete(Publisher publisher, String sessionId) { - for (int attempt = 1; attempt <= 3; attempt++) { - try { - if (publisher.publishSessionComplete(sessionId)) return; - } catch (Exception e) { - System.err.println("[agent] session-complete signal error for " - + sessionId + ": " + e.getMessage()); - } - try { - Thread.sleep(2000L); - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - return; - } - } - System.out.println("[agent] session-complete signal gave up for " + sessionId - + " - backend grace finalize will apply"); - } - - // ------------------------------------------------------------------------- - // Config resolution - // ------------------------------------------------------------------------- - - /** - * Builds AgentConfig from CLI flags and/or environment variables. - * 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) { - return loadFromArgs(args, System.getenv()); - } - - static AgentConfig loadFromArgs(String[] args, Map env) { - boolean hasAnyConfig = Arrays.stream(args).anyMatch(a -> a.startsWith("--")) - || env.keySet().stream().anyMatch(k -> k.startsWith("GPUFL_")); - - if (!hasAnyConfig) { - 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("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 - String type = require(args, "type", "GPUFL_PUBLISHER_TYPE", env); - - // Reject the legacy --url / GPUFL_HTTP_URL flag with a migration - // hint. Removed May 2026 in favor of --host + --api-version (see - // 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 - // less obvious about WHICH legacy flag is at fault. - if (resolve(args, "url", "GPUFL_HTTP_URL", null, env) != null) { - 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."); - System.exit(1); - } - PublisherConfig publisher = switch (type.toLowerCase()) { - case "http" -> new HttpConfig( - 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)), - 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("ERROR: Unknown publisher type: '" + type + "' (expected: http, kafka)"); - printUsage(); - System.exit(1); - yield null; // unreachable - } - }; - - ArchiverConfig archiver = null; - String archiverEndpoint = resolve(args, "archiver-endpoint", "GPUFL_ARCHIVER_ENDPOINT", null, env); - if (archiverEndpoint != null) { - archiver = new ArchiverConfig( - archiverEndpoint, - require(args, "archiver-bucket", "GPUFL_ARCHIVER_BUCKET", env), - resolve(args, "archiver-region", "GPUFL_ARCHIVER_REGION", null, env), - require(args, "archiver-access-key", "GPUFL_ARCHIVER_ACCESS_KEY", env), - require(args, "archiver-secret-key", "GPUFL_ARCHIVER_SECRET_KEY", env), - resolve(args, "archiver-prefix", "GPUFL_ARCHIVER_PREFIX", null, env), - Boolean.parseBoolean(resolve(args, "archiver-delete", "GPUFL_ARCHIVER_DELETE", "false", env))); - } - - LogSourceConfig source = folder != null - ? new LogSourceConfig(folder, logTypes) : null; - return new AgentConfig(source, null, publisher, archiver); - } - - static List parseLogTypes(String raw) { - if (raw == null) return null; - List parsed = Arrays.stream(raw.split(",")) - .map(String::trim) - .filter(s -> !s.isEmpty()) - .toList(); - return parsed.isEmpty() ? null : parsed; - } - - static List logTypesOrDefault(String[] args, Map env) { - List parsed = parseLogTypes(resolve(args, "log-types", "GPUFL_LOG_TYPES", null, env)); - return parsed == null ? DEFAULT_LOG_TYPES : parsed; - } - - /** - * Resolves a config value: checks --flag=value in args first, then the env var, then the default. - */ - static String resolve(String[] args, String flag, String envVar, String defaultValue) { - return resolve(args, flag, envVar, defaultValue, System.getenv()); - } - - static String resolve(String[] args, String flag, String envVar, String defaultValue, Map env) { - String prefix = "--" + flag + "="; - for (String arg : args) { - if (arg.startsWith(prefix)) return arg.substring(prefix.length()); - } - String envVal = env.get(envVar); - if (envVal != null && !envVal.isBlank()) return envVal; - return defaultValue; - } - - /** Like resolve(), but exits with an error if the value is absent. */ - static String require(String[] args, String flag, String envVar) { - return require(args, flag, envVar, System.getenv()); - } - - 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("ERROR: --" + flag + " (or env " + envVar + ") is required"); - printUsage(); - System.exit(1); - } - return val; - } - - /** - * Per-session-subdirectory channel files under v1.2: - * //{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|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 - * this it would re-print the same hint every cycle for as long as the - * stale flat files sit in the folder. Warn once per folder instead. */ - private static final java.util.Set warnedLegacyFolders = - java.util.concurrent.ConcurrentHashMap.newKeySet(); - - /** 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 - * that the pre-v1.2 LogSourceConfig.filePrefix used to overload. */ - 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 - * 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). - * - *

Subdirectories that don't look like sessions (no channel files - * inside) are silently skipped. Files at the top level of {@code folder} - * 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 - * logging a "discovered" line per session every tick would just spam. - */ - static List discoverSources(File folder, List logTypes) { - List result = new ArrayList<>(); - if (!folder.isDirectory()) return result; - - 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 - // discovery walks subdirs only. - Pattern legacyTop = Pattern.compile( - "^(.+)\\.(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. - if (warnedLegacyFolders.add(folder.getAbsolutePath())) { - System.err.println( - "[agent] warning: found pre-v1.2 flat log file '" + - f.getName() + "' at top level of " + folder + - ". v1.2 expects /.log " + - "under that folder. Re-run with a v1.2 gpufl client " + - "or migrate old files into a session subdirectory."); - } - break; // one warning is enough - } - } - } - - File[] subdirs = folder.listFiles(File::isDirectory); - if (subdirs == null) return result; - java.util.Arrays.sort(subdirs, java.util.Comparator.comparing(File::getName)); - - for (File subdir : subdirs) { - if (subdir.getName().startsWith(".")) continue; // skip dotdirs - File[] inside = subdir.listFiles(); - if (inside == null) continue; - boolean looksLikeSession = false; - for (File child : inside) { - if (child.isFile() && - CHANNEL_FILE_PATTERN.matcher(child.getName()).matches()) { - looksLikeSession = true; - break; - } - } - if (!looksLikeSession) continue; - result.add(new DiscoveredSession(folder, subdir.getName(), types)); - } - return result; - } - - static String parseConfigArg(String[] args) { - for (String arg : args) { - if (arg.startsWith("--config=")) return arg.substring("--config=".length()); - } - return null; - } - - static String buildArchiveKey(String prefix, Path path) { - return prefix + path.toFile().getName(); - } - - /** Derives the topic/topic-prefix from whichever publisher is configured. */ - static String topicPrefix(AgentConfig config) { - return switch (config.publisher()) { - case KafkaConfig kafka -> kafka.topicPrefix(); - case HttpConfig ignored -> "gpu-trace"; - }; - } - - // ------------------------------------------------------------------------- - // Loaders - // ------------------------------------------------------------------------- - - private static AgentConfig loadExternalConfig(String path) { - var file = new File(path); - if (!file.exists()) { - System.err.println("ERROR: Config file not found: '" + path + "'"); - System.exit(1); - } - try { - return JsonSettings.MAPPER.readValue(file, AgentConfig.class); - } catch (Exception e) { - System.err.println("Failed to parse config file: " + e.getMessage()); - System.exit(1); - return null; // unreachable - } - } - - private static AgentConfig loadClasspathConfig(String resourcePath) { - try (var stream = Main.class.getClassLoader().getResourceAsStream(resourcePath)) { - if (stream == null) { - System.err.println("ERROR: Bundled config not found: '" + resourcePath + "'"); - System.exit(1); - } - return JsonSettings.MAPPER.readValue(stream, AgentConfig.class); - } catch (Exception e) { - System.err.println("Failed to parse bundled config: " + e.getMessage()); - System.exit(1); - return null; // unreachable - } - } - - // ------------------------------------------------------------------------- - // Usage - // ------------------------------------------------------------------------- - - static void printUsage() { - System.err.println(""" - - Usage: gpufl-agent [options] - - Config file (overrides all flags): - --config= Load full JSON config from file - - Source (at least one of --folder or --folders is required): - --folder= Log folder path [GPUFL_SOURCE_FOLDER] - --folders= Auto-discover folders [GPUFL_SOURCE_FOLDERS] - --log-types= Log channels to tail [GPUFL_LOG_TYPES] default: device,scope,system,sass - --cursor-file= Cursor state file [GPUFL_CURSOR_FILE] default: ./cursor.json - - Publisher (required): - --type= Publisher type [GPUFL_PUBLISHER_TYPE] - - HTTP publisher: - --host= Backend host [GPUFL_HTTP_HOST] e.g. https://api.gpuflight.com - --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] - --topic-prefix= Topic prefix [GPUFL_KAFKA_TOPIC_PREFIX] default: gpu-trace - --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-endpoint= S3-compatible endpoint [GPUFL_ARCHIVER_ENDPOINT] - --archiver-bucket= S3 bucket [GPUFL_ARCHIVER_BUCKET] - --archiver-region= Region [GPUFL_ARCHIVER_REGION] default: nyc3 - --archiver-access-key= Access key [GPUFL_ARCHIVER_ACCESS_KEY] - --archiver-secret-key= Secret key [GPUFL_ARCHIVER_SECRET_KEY] - --archiver-prefix= Object key prefix [GPUFL_ARCHIVER_PREFIX] default: raw-events/ - --archiver-delete= Delete after upload [GPUFL_ARCHIVER_DELETE] default: false - - Examples: - # CLI flags - gpufl-agent --folder=/var/log/gpuflight --type=http --host=https://api.gpuflight.com - - # Env vars (systemd / Docker) - GPUFL_SOURCE_FOLDER=/var/log GPUFL_PUBLISHER_TYPE=http GPUFL_HTTP_HOST=https://api.gpuflight.com gpufl-agent - - # JSON config file - gpufl-agent --config=/etc/gpuflight/agent.json - - # No args -> uses bundled local.json (development only) - gpufl-agent - """); + public static void main(String[] args) throws Exception { + AgentConfig config = ConfigLoader.load(args); + GpuflAgent agent = new GpuflAgent(config, args, System.getenv()); + agent.start(); } } diff --git a/src/main/java/com/gpuflight/agent/config/ConfigLoader.java b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java new file mode 100644 index 0000000..13b5071 --- /dev/null +++ b/src/main/java/com/gpuflight/agent/config/ConfigLoader.java @@ -0,0 +1,240 @@ +package com.gpuflight.agent.config; + +import com.gpuflight.agent.model.AgentConfig; +import com.gpuflight.agent.model.ArchiverConfig; +import com.gpuflight.agent.model.LogSourceConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.InputStream; +import java.nio.file.Path; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +public class ConfigLoader { + + private static final Logger log = LoggerFactory.getLogger(ConfigLoader.class); + + public static final List DEFAULT_LOG_TYPES = List.of("device", "scope", "system", "sass"); + + public static AgentConfig load(String[] args) { + String configPath = parseConfigArg(args); + if (configPath != null) { + return loadExternalConfig(configPath); + } + return loadFromArgs(args); + } + + public static String parseConfigArg(String[] args) { + for (String arg : args) { + if (arg.startsWith("--config=")) return arg.substring("--config=".length()); + } + return null; + } + + public static AgentConfig loadFromArgs(String[] args) { + return loadFromArgs(args, System.getenv()); + } + + public static AgentConfig loadFromArgs(String[] args, Map env) { + boolean hasAnyConfig = Arrays.stream(args).anyMatch(a -> a.startsWith("--")) + || env.keySet().stream().anyMatch(k -> k.startsWith("GPUFL_")); + + if (!hasAnyConfig) { + log.info("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("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 + String type = require(args, "type", "GPUFL_PUBLISHER_TYPE", env); + + if (resolve(args, "url", "GPUFL_HTTP_URL", null, env) != null) { + 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."); + System.exit(1); + } + + PublisherConfig publisher = switch (type.toLowerCase()) { + case "http" -> new HttpConfig( + 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)), + 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("ERROR: Unknown publisher type: '" + type + "' (expected: http, kafka)"); + printUsage(); + System.exit(1); + yield null; // unreachable + } + }; + + ArchiverConfig archiver = null; + String archiverEndpoint = resolve(args, "archiver-endpoint", "GPUFL_ARCHIVER_ENDPOINT", null, env); + if (archiverEndpoint != null) { + archiver = new ArchiverConfig( + archiverEndpoint, + require(args, "archiver-bucket", "GPUFL_ARCHIVER_BUCKET", env), + resolve(args, "archiver-region", "GPUFL_ARCHIVER_REGION", null, env), + require(args, "archiver-access-key", "GPUFL_ARCHIVER_ACCESS_KEY", env), + require(args, "archiver-secret-key", "GPUFL_ARCHIVER_SECRET_KEY", env), + resolve(args, "archiver-prefix", "GPUFL_ARCHIVER_PREFIX", null, env), + Boolean.parseBoolean(resolve(args, "archiver-delete", "GPUFL_ARCHIVER_DELETE", "false", env))); + } + + LogSourceConfig source = folder != null + ? new LogSourceConfig(folder, logTypes) : null; + return new AgentConfig(source, null, publisher, archiver); + } + + public static List parseLogTypes(String raw) { + if (raw == null) return null; + List parsed = Arrays.stream(raw.split(",")) + .map(String::trim) + .filter(s -> !s.isEmpty()) + .toList(); + return parsed.isEmpty() ? null : parsed; + } + + public static List logTypesOrDefault(String[] args, Map env) { + List parsed = parseLogTypes(resolve(args, "log-types", "GPUFL_LOG_TYPES", null, env)); + return parsed == null ? DEFAULT_LOG_TYPES : parsed; + } + + public static String resolve(String[] args, String flag, String envVar, String defaultValue) { + return resolve(args, flag, envVar, defaultValue, System.getenv()); + } + + public static String resolve(String[] args, String flag, String envVar, String defaultValue, Map env) { + String prefix = "--" + flag + "="; + for (String arg : args) { + if (arg.startsWith(prefix)) return arg.substring(prefix.length()); + } + String envVal = env.get(envVar); + if (envVal != null && !envVal.isBlank()) return envVal; + return defaultValue; + } + + public 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("ERROR: --" + flag + " (or env " + envVar + ") is required"); + printUsage(); + System.exit(1); + } + return val; + } + + public 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"); + } + + public static String buildArchiveKey(String prefix, Path path) { + return prefix + path.toFile().getName(); + } + + private static AgentConfig loadExternalConfig(String path) { + File file = new File(path); + if (!file.exists()) { + System.err.println("ERROR: Config file not found: '" + path + "'"); + System.exit(1); + } + try { + return JsonSettings.MAPPER.readValue(file, AgentConfig.class); + } catch (Exception e) { + System.err.println("Failed to parse config file: " + e.getMessage()); + System.exit(1); + return null; // unreachable + } + } + + private static AgentConfig loadClasspathConfig(String resourcePath) { + try (InputStream stream = ConfigLoader.class.getClassLoader().getResourceAsStream(resourcePath)) { + if (stream == null) { + System.err.println("ERROR: Bundled config not found: '" + resourcePath + "'"); + System.exit(1); + } + return JsonSettings.MAPPER.readValue(stream, AgentConfig.class); + } catch (Exception e) { + System.err.println("Failed to parse bundled config: " + e.getMessage()); + System.exit(1); + return null; // unreachable + } + } + + public static void printUsage() { + System.err.println(""" + + Usage: gpufl-agent [options] + + Config file (overrides all flags): + --config= Load full JSON config from file + + Source (at least one of --folder or --folders is required): + --folder= Log folder path [GPUFL_SOURCE_FOLDER] + --folders= Auto-discover folders [GPUFL_SOURCE_FOLDERS] + --log-types= Log channels to tail [GPUFL_LOG_TYPES] default: device,scope,system,sass + --cursor-file= Cursor state file [GPUFL_CURSOR_FILE] default: ./cursor.json + + Publisher (required): + --type= Publisher type [GPUFL_PUBLISHER_TYPE] + + HTTP publisher: + --host= Backend host [GPUFL_HTTP_HOST] e.g. https://api.gpuflight.com + --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] + --topic-prefix= Topic prefix [GPUFL_KAFKA_TOPIC_PREFIX] default: gpu-trace + --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-endpoint= S3-compatible endpoint [GPUFL_ARCHIVER_ENDPOINT] + --archiver-bucket= S3 bucket [GPUFL_ARCHIVER_BUCKET] + --archiver-region= Region [GPUFL_ARCHIVER_REGION] default: nyc3 + --archiver-access-key= Access key [GPUFL_ARCHIVER_ACCESS_KEY] + --archiver-secret-key= Secret key [GPUFL_ARCHIVER_SECRET_KEY] + --archiver-prefix= Object key prefix [GPUFL_ARCHIVER_PREFIX] default: raw-events/ + --archiver-delete= Delete after upload [GPUFL_ARCHIVER_DELETE] default: false + + Examples: + # CLI flags + gpufl-agent --folder=/var/log/gpuflight --type=http --host=https://api.gpuflight.com + + # Env vars (systemd / Docker) + GPUFL_SOURCE_FOLDER=/var/log GPUFL_PUBLISHER_TYPE=http GPUFL_HTTP_HOST=https://api.gpuflight.com gpufl-agent + + # JSON config file + gpufl-agent --config=/etc/gpuflight/agent.json + + # No args -> uses bundled local.json (development only) + gpufl-agent + """); + } +} diff --git a/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java b/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java new file mode 100644 index 0000000..6a26def --- /dev/null +++ b/src/main/java/com/gpuflight/agent/model/DiscoveredSession.java @@ -0,0 +1,10 @@ +package com.gpuflight.agent.model; + +import java.io.File; +import java.util.List; + +/** + * A session found under a watched folder: the folder, the session_id + * (the subdir name), and the channels to tail. + */ +public record DiscoveredSession(File folder, String sessionId, List logTypes) {} diff --git a/src/main/java/com/gpuflight/agent/service/SessionWatcher.java b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java new file mode 100644 index 0000000..ac75e9e --- /dev/null +++ b/src/main/java/com/gpuflight/agent/service/SessionWatcher.java @@ -0,0 +1,105 @@ +package com.gpuflight.agent.service; + +import com.gpuflight.agent.config.ConfigLoader; +import com.gpuflight.agent.model.DiscoveredSession; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.function.Consumer; +import java.util.regex.Pattern; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class SessionWatcher { + + private static final Logger log = LoggerFactory.getLogger(SessionWatcher.class); + + private static final Pattern CHANNEL_FILE_PATTERN = + Pattern.compile("^(device|scope|system|sass)(?:\\.\\d+)?\\.log(?:\\.gz)?$"); + + private static final Set warnedLegacyFolders = + ConcurrentHashMap.newKeySet(); + + private final File folder; + private final List logTypes; + private final Consumer onSessionDiscovered; + + public SessionWatcher(File folder, List logTypes, Consumer onSessionDiscovered) { + this.folder = folder; + this.logTypes = logTypes; + this.onSessionDiscovered = onSessionDiscovered; + } + + public void start(ExecutorService executor) { + executor.submit(() -> { + while (!Thread.currentThread().isInterrupted()) { + try { + for (DiscoveredSession s : discoverSources(folder, logTypes)) { + onSessionDiscovered.accept(s); + } + Thread.sleep(2_000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } catch (Exception e) { + System.err.println("[watcher] error scanning " + + folder + ": " + e.getMessage()); + } + } + }); + } + + public static List discoverSources(File folder, List logTypes) { + List result = new ArrayList<>(); + if (!folder.isDirectory()) return result; + + List types = (logTypes == null || logTypes.isEmpty()) ? ConfigLoader.DEFAULT_LOG_TYPES : logTypes; + + // Legacy-format warning + Pattern legacyTop = Pattern.compile( + "^(.+)\\.(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()) { + if (warnedLegacyFolders.add(folder.getAbsolutePath())) { + System.err.println( + "[agent] warning: found pre-v1.2 flat log file '" + + f.getName() + "' at top level of " + folder + + ". v1.2 expects /.log " + + "under that folder. Re-run with a v1.2 gpufl client " + + "or migrate old files into a session subdirectory."); + } + break; + } + } + } + + File[] subdirs = folder.listFiles(File::isDirectory); + if (subdirs == null) return result; + Arrays.sort(subdirs, Comparator.comparing(File::getName)); + + for (File subdir : subdirs) { + if (subdir.getName().startsWith(".")) continue; + File[] inside = subdir.listFiles(); + if (inside == null) continue; + boolean looksLikeSession = false; + for (File child : inside) { + if (child.isFile() && + CHANNEL_FILE_PATTERN.matcher(child.getName()).matches()) { + looksLikeSession = true; + break; + } + } + if (!looksLikeSession) continue; + result.add(new DiscoveredSession(folder, subdir.getName(), types)); + } + return result; + } +} diff --git a/src/main/java/com/gpuflight/agent/service/TailerManager.java b/src/main/java/com/gpuflight/agent/service/TailerManager.java new file mode 100644 index 0000000..251fd2a --- /dev/null +++ b/src/main/java/com/gpuflight/agent/service/TailerManager.java @@ -0,0 +1,100 @@ +package com.gpuflight.agent.service; + +import com.gpuflight.agent.CursorManager; +import com.gpuflight.agent.LogTailer; +import com.gpuflight.agent.config.StreamUploadSettings; +import com.gpuflight.agent.filter.DeviceMetricDeduplicator; +import com.gpuflight.agent.model.DiscoveredSession; +import com.gpuflight.agent.publisher.Publisher; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.file.Path; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +public class TailerManager { + + private static final Logger log = LoggerFactory.getLogger(TailerManager.class); + + private final ExecutorService executor; + private final Publisher publisher; + private final CursorManager cursorMgr; + private final BlockingQueue consumedFilesQueue; + private final DeviceMetricDeduplicator deduplicator; + private final StreamUploadSettings streamUploadSettings; + private final String topicPrefix; + + private final Set startedSessions = ConcurrentHashMap.newKeySet(); + private final AtomicInteger activeTailers = new AtomicInteger(0); + + public TailerManager(ExecutorService executor, + Publisher publisher, + CursorManager cursorMgr, + BlockingQueue consumedFilesQueue, + DeviceMetricDeduplicator deduplicator, + StreamUploadSettings streamUploadSettings, + String topicPrefix) { + this.executor = executor; + this.publisher = publisher; + this.cursorMgr = cursorMgr; + this.consumedFilesQueue = consumedFilesQueue; + this.deduplicator = deduplicator; + this.streamUploadSettings = streamUploadSettings; + this.topicPrefix = topicPrefix; + } + + public void spawnSessionTailers(DiscoveredSession session) { + String key = session.folder().getAbsolutePath() + "::" + session.sessionId(); + if (!startedSessions.add(key)) return; + log.info("Tailing session \"{}\" in {} types={}", + session.sessionId(), session.folder(), session.logTypes()); + + var remaining = new AtomicInteger(session.logTypes().size()); + String sid = session.sessionId(); + for (String type : session.logTypes()) { + activeTailers.incrementAndGet(); + executor.submit(() -> { + try { + var dedup = "system".equals(type) ? deduplicator : null; + var tailer = new LogTailer(session.folder(), session.sessionId(), type, + topicPrefix, cursorMgr, consumedFilesQueue, dedup, + streamUploadSettings); + tailer.tail(publisher); + if (!Thread.currentThread().isInterrupted() + && remaining.decrementAndGet() == 0 + && streamUploadSettings.enabled()) { + signalSessionComplete(publisher, sid); + } + } finally { + activeTailers.decrementAndGet(); + } + }); + } + } + + private void signalSessionComplete(Publisher publisher, String sessionId) { + for (int attempt = 1; attempt <= 3; attempt++) { + try { + if (publisher.publishSessionComplete(sessionId)) return; + } catch (Exception e) { + System.err.println("[agent] session-complete signal error for " + + sessionId + ": " + e.getMessage()); + } + try { + Thread.sleep(2000L); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } + } + log.warn("session-complete signal gave up for {} - backend grace finalize will apply", sessionId); + } + + public AtomicInteger getActiveTailers() { + return activeTailers; + } +} diff --git a/src/test/java/com/gpuflight/agent/MainTest.java b/src/test/java/com/gpuflight/agent/MainTest.java index 057f2f4..d59bf53 100644 --- a/src/test/java/com/gpuflight/agent/MainTest.java +++ b/src/test/java/com/gpuflight/agent/MainTest.java @@ -1,5 +1,6 @@ package com.gpuflight.agent; +import com.gpuflight.agent.config.ConfigLoader; import com.gpuflight.agent.config.HttpConfig; import com.gpuflight.agent.config.KafkaConfig; import com.gpuflight.agent.model.AgentConfig; @@ -26,38 +27,38 @@ class MainTest { @Test void resolve_returnsFlagValue() { String[] args = {"--host=http://example.com", "--token=abc"}; - assertEquals("http://example.com", Main.resolve(args, "host", "GPUFL_HTTP_HOST", null, Collections.emptyMap())); + assertEquals("http://example.com", ConfigLoader.resolve(args, "host", "GPUFL_HTTP_HOST", null, Collections.emptyMap())); } @Test void resolve_returnsSecondFlagValue() { String[] args = {"--host=http://example.com", "--token=abc"}; - assertEquals("abc", Main.resolve(args, "token", "GPUFL_HTTP_TOKEN", null, Collections.emptyMap())); + assertEquals("abc", ConfigLoader.resolve(args, "token", "GPUFL_HTTP_TOKEN", null, Collections.emptyMap())); } @Test void resolve_returnsDefaultWhenAbsent() { String[] args = {}; - assertEquals("default-val", Main.resolve(args, "missing-flag", "GPUFL_MISSING_VAR_XYZ123", "default-val", Collections.emptyMap())); + assertEquals("default-val", ConfigLoader.resolve(args, "missing-flag", "GPUFL_MISSING_VAR_XYZ123", "default-val", Collections.emptyMap())); } @Test void resolve_returnsNullDefaultWhenAbsent() { String[] args = {}; - assertNull(Main.resolve(args, "missing-flag", "GPUFL_MISSING_VAR_XYZ123", null, Collections.emptyMap())); + assertNull(ConfigLoader.resolve(args, "missing-flag", "GPUFL_MISSING_VAR_XYZ123", null, Collections.emptyMap())); } @Test void resolve_flagTakesPrecedenceOverDefault() { String[] args = {"--folder=/from-cli"}; - assertEquals("/from-cli", Main.resolve(args, "folder", "GPUFL_SOURCE_FOLDER_NOTSET_XYZ", "/default", Collections.emptyMap())); + assertEquals("/from-cli", ConfigLoader.resolve(args, "folder", "GPUFL_SOURCE_FOLDER_NOTSET_XYZ", "/default", Collections.emptyMap())); } @Test void resolve_partialMatchDoesNotReturn() { // --folderExtra should not match --folder= String[] args = {"--folderExtra=/wrong"}; - assertNull(Main.resolve(args, "folder", "GPUFL_MISSING_XYZ123", null, Collections.emptyMap())); + assertNull(ConfigLoader.resolve(args, "folder", "GPUFL_MISSING_XYZ123", null, Collections.emptyMap())); } // ---- env var support ---- @@ -66,39 +67,39 @@ void resolve_partialMatchDoesNotReturn() { void resolve_returnsEnvVarWhenFlagAbsent() { String[] args = {}; Map env = Map.of("GPUFL_HTTP_HOST", "http://env-url.com"); - assertEquals("http://env-url.com", Main.resolve(args, "host", "GPUFL_HTTP_HOST", null, env)); + assertEquals("http://env-url.com", ConfigLoader.resolve(args, "host", "GPUFL_HTTP_HOST", null, env)); } @Test void resolve_flagTakesPrecedenceOverEnvVar() { String[] args = {"--host=http://cli-url.com"}; Map env = Map.of("GPUFL_HTTP_HOST", "http://env-url.com"); - assertEquals("http://cli-url.com", Main.resolve(args, "host", "GPUFL_HTTP_HOST", null, env)); + assertEquals("http://cli-url.com", ConfigLoader.resolve(args, "host", "GPUFL_HTTP_HOST", null, env)); } // ---- log type parsing ---- @Test void parseLogTypes_trimsAndDropsEmptyValues() { - assertEquals(List.of("system", "device"), Main.parseLogTypes(" system, ,device,")); + assertEquals(List.of("system", "device"), ConfigLoader.parseLogTypes(" system, ,device,")); } @Test void parseLogTypes_returnsNullForMissingOrEmptyValues() { - assertNull(Main.parseLogTypes(null)); - assertNull(Main.parseLogTypes(" , ")); + assertNull(ConfigLoader.parseLogTypes(null)); + assertNull(ConfigLoader.parseLogTypes(" , ")); } @Test void logTypesOrDefault_usesExplicitFolderLogTypes() { String[] args = {"--folders=/logs", "--log-types=system"}; - assertEquals(List.of("system"), Main.logTypesOrDefault(args, Collections.emptyMap())); + assertEquals(List.of("system"), ConfigLoader.logTypesOrDefault(args, Collections.emptyMap())); } @Test void logTypesOrDefault_usesDefaultWhenAbsent() { assertEquals(List.of("device", "scope", "system", "sass"), - Main.logTypesOrDefault(new String[]{}, Collections.emptyMap())); + ConfigLoader.logTypesOrDefault(new String[]{}, Collections.emptyMap())); } // ---- require() ---- @@ -106,7 +107,7 @@ void logTypesOrDefault_usesDefaultWhenAbsent() { @Test void require_returnsValueWhenPresent() { String[] args = {"--folder=/tmp/logs"}; - assertEquals("/tmp/logs", Main.require(args, "folder", "GPUFL_SOURCE_FOLDER", Collections.emptyMap())); + assertEquals("/tmp/logs", ConfigLoader.require(args, "folder", "GPUFL_SOURCE_FOLDER", Collections.emptyMap())); } // ---- parseConfigArg() ---- @@ -114,18 +115,18 @@ void require_returnsValueWhenPresent() { @Test void parseConfigArg_returnsPath() { String[] args = {"--config=/etc/agent.json", "--other=val"}; - assertEquals("/etc/agent.json", Main.parseConfigArg(args)); + assertEquals("/etc/agent.json", ConfigLoader.parseConfigArg(args)); } @Test void parseConfigArg_returnsNullWhenAbsent() { String[] args = {"--folder=/tmp"}; - assertNull(Main.parseConfigArg(args)); + assertNull(ConfigLoader.parseConfigArg(args)); } @Test void parseConfigArg_emptyArgs() { - assertNull(Main.parseConfigArg(new String[]{})); + assertNull(ConfigLoader.parseConfigArg(new String[]{})); } // ---- buildArchiveKey() ---- @@ -133,19 +134,19 @@ void parseConfigArg_emptyArgs() { @Test void buildArchiveKey_concatenatesPrefixAndFilename() { Path path = Path.of("/var/log/gpuflight/device.1.log"); - assertEquals("raw-events/device.1.log", Main.buildArchiveKey("raw-events/", path)); + assertEquals("raw-events/device.1.log", ConfigLoader.buildArchiveKey("raw-events/", path)); } @Test void buildArchiveKey_emptyPrefix() { Path path = Path.of("/some/dir/file.log"); - assertEquals("file.log", Main.buildArchiveKey("", path)); + assertEquals("file.log", ConfigLoader.buildArchiveKey("", path)); } @Test void buildArchiveKey_nestedPath() { Path path = Path.of("/a/b/c/my.scope.2.log"); - assertEquals("prefix/my.scope.2.log", Main.buildArchiveKey("prefix/", path)); + assertEquals("prefix/my.scope.2.log", ConfigLoader.buildArchiveKey("prefix/", path)); } // ---- loadFromArgs() — no flags → classpath fallback ---- @@ -154,7 +155,7 @@ void buildArchiveKey_nestedPath() { void loadFromArgs_noArgsUsesClasspathConfig() { boolean hasGpuflEnv = System.getenv().keySet().stream().anyMatch(k -> k.startsWith("GPUFL_")); if (!hasGpuflEnv) { - AgentConfig config = Main.loadFromArgs(new String[]{}); + AgentConfig config = ConfigLoader.loadFromArgs(new String[]{}); assertNotNull(config); assertNotNull(config.source()); assertEquals(".", config.source().folder()); @@ -171,7 +172,7 @@ void loadFromArgs_httpPublisher_basic() { "--folder=/var/log", "--type=http", "--host=http://localhost:8080" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); assertNotNull(config); assertEquals("/var/log", config.source().folder()); assertInstanceOf(HttpConfig.class, config.publisher()); @@ -190,7 +191,7 @@ void loadFromArgs_httpPublisher_withToken() { "--folder=/logs", "--type=http", "--host=http://collector:8080", "--token=tok123" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); HttpConfig http = (HttpConfig) config.publisher(); assertEquals("tok123", http.authToken()); } @@ -201,7 +202,7 @@ void loadFromArgs_httpPublisher_customTimeout() { "--folder=/logs", "--type=http", "--host=http://collector:8080", "--timeout=30" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); HttpConfig http = (HttpConfig) config.publisher(); assertEquals(30L, http.timeoutSeconds()); } @@ -215,7 +216,7 @@ void loadFromArgs_kafkaPublisher_withOptions() { "--brokers=broker1:9092,broker2:9092", "--topic-prefix=my-topic", "--compression=lz4" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); assertInstanceOf(KafkaConfig.class, config.publisher()); KafkaConfig kafka = (KafkaConfig) config.publisher(); assertEquals("broker1:9092,broker2:9092", kafka.bootstrapServers()); @@ -229,7 +230,7 @@ void loadFromArgs_kafkaPublisher_defaultsApplied() { "--folder=/var/log", "--type=kafka", "--brokers=localhost:9092" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); KafkaConfig kafka = (KafkaConfig) config.publisher(); // KafkaConfig compact constructor replaces null with defaults assertEquals("gpu-trace", kafka.topicPrefix()); @@ -247,7 +248,7 @@ void loadFromArgs_withArchiver_defaults() { "--archiver-access-key=AKIAKEY", "--archiver-secret-key=SECRET" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); assertNotNull(config.archiver()); assertEquals("http://minio:9000", config.archiver().endpoint()); assertEquals("my-bucket", config.archiver().bucket()); @@ -270,7 +271,7 @@ void loadFromArgs_withArchiver_explicitOptions() { "--archiver-prefix=logs/", "--archiver-delete=true" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); ArchiverConfig archiver = config.archiver(); assertNotNull(archiver); assertEquals("us-east-1", archiver.region()); @@ -283,7 +284,7 @@ void loadFromArgs_noArchiver_whenEndpointAbsent() { String[] args = { "--folder=/logs", "--type=http", "--host=http://localhost" }; - AgentConfig config = Main.loadFromArgs(args, Collections.emptyMap()); + AgentConfig config = ConfigLoader.loadFromArgs(args, Collections.emptyMap()); assertNull(config.archiver()); } @@ -302,7 +303,7 @@ void parseConfigArg_extractsPathFromConfigFlag(@TempDir Path tempDir) throws IOE // Invoke via parseConfigArg + direct public method chain String[] args = {"--config=" + configFile.getAbsolutePath()}; - String configPath = Main.parseConfigArg(args); + String configPath = ConfigLoader.parseConfigArg(args); assertNotNull(configPath); assertEquals(configFile.getAbsolutePath(), configPath); } @@ -320,7 +321,7 @@ void loadExternalConfig_loadsValidJsonFile(@TempDir Path tempDir) throws Excepti File configFile = tempDir.resolve("ext.json").toFile(); Files.writeString(configFile.toPath(), json); - Method m = Main.class.getDeclaredMethod("loadExternalConfig", String.class); + Method m = ConfigLoader.class.getDeclaredMethod("loadExternalConfig", String.class); m.setAccessible(true); AgentConfig config = (AgentConfig) m.invoke(null, configFile.getAbsolutePath()); @@ -340,7 +341,7 @@ void loadExternalConfig_withKafkaPublisher(@TempDir Path tempDir) throws Excepti File configFile = tempDir.resolve("kafka.json").toFile(); Files.writeString(configFile.toPath(), json); - Method m = Main.class.getDeclaredMethod("loadExternalConfig", String.class); + Method m = ConfigLoader.class.getDeclaredMethod("loadExternalConfig", String.class); m.setAccessible(true); AgentConfig config = (AgentConfig) m.invoke(null, configFile.getAbsolutePath()); @@ -367,7 +368,7 @@ void loadExternalConfig_withArchiver(@TempDir Path tempDir) throws Exception { File configFile = tempDir.resolve("arch.json").toFile(); Files.writeString(configFile.toPath(), json); - Method m = Main.class.getDeclaredMethod("loadExternalConfig", String.class); + Method m = ConfigLoader.class.getDeclaredMethod("loadExternalConfig", String.class); m.setAccessible(true); AgentConfig config = (AgentConfig) m.invoke(null, configFile.getAbsolutePath()); @@ -392,7 +393,7 @@ void loadExternalConfig_sourcesArray(@TempDir Path tempDir) throws Exception { File configFile = tempDir.resolve("multi.json").toFile(); Files.writeString(configFile.toPath(), json); - Method m = Main.class.getDeclaredMethod("loadExternalConfig", String.class); + Method m = ConfigLoader.class.getDeclaredMethod("loadExternalConfig", String.class); m.setAccessible(true); AgentConfig config = (AgentConfig) m.invoke(null, configFile.getAbsolutePath()); @@ -418,7 +419,7 @@ void loadExternalConfig_sourceAndSourcesMerged(@TempDir Path tempDir) throws Exc File configFile = tempDir.resolve("both.json").toFile(); Files.writeString(configFile.toPath(), json); - Method m = Main.class.getDeclaredMethod("loadExternalConfig", String.class); + Method m = ConfigLoader.class.getDeclaredMethod("loadExternalConfig", String.class); m.setAccessible(true); AgentConfig config = (AgentConfig) m.invoke(null, configFile.getAbsolutePath()); @@ -434,26 +435,26 @@ void loadExternalConfig_sourceAndSourcesMerged(@TempDir Path tempDir) throws Exc @Test void topicPrefix_httpConfig_returnsGpuTrace() { - AgentConfig config = Main.loadFromArgs(new String[]{ + AgentConfig config = ConfigLoader.loadFromArgs(new String[]{ "--folder=/logs", "--type=http", "--host=http://localhost" }); - assertEquals("gpu-trace", Main.topicPrefix(config)); + assertEquals("gpu-trace", GpuflAgent.topicPrefix(config)); } @Test void topicPrefix_kafkaConfig_returnsConfiguredPrefix() { - AgentConfig config = Main.loadFromArgs(new String[]{ + AgentConfig config = ConfigLoader.loadFromArgs(new String[]{ "--folder=/logs", "--type=kafka", "--brokers=localhost:9092", "--topic-prefix=my-prefix" }); - assertEquals("my-prefix", Main.topicPrefix(config)); + assertEquals("my-prefix", GpuflAgent.topicPrefix(config)); } // ---- loadClasspathConfig (private) via reflection ---- @Test void loadClasspathConfig_loadsLocalJson() throws Exception { - Method m = Main.class.getDeclaredMethod("loadClasspathConfig", String.class); + Method m = ConfigLoader.class.getDeclaredMethod("loadClasspathConfig", String.class); m.setAccessible(true); AgentConfig config = (AgentConfig) m.invoke(null, "config/local.json"); @@ -465,22 +466,22 @@ void loadClasspathConfig_loadsLocalJson() throws Exception { @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())); + assertTrue(ConfigLoader.parseExitWhenDrained(new String[]{}, Map.of("GPUFL_AGENT_EXIT_WHEN_DRAINED", "1"))); + assertTrue(ConfigLoader.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())); + assertFalse(ConfigLoader.parseExitWhenDrained(new String[]{}, Collections.emptyMap())); + assertFalse(ConfigLoader.parseExitWhenDrained(new String[]{}, Map.of("GPUFL_AGENT_EXIT_WHEN_DRAINED", "0"))); + assertFalse(ConfigLoader.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))); + assertTrue(GpuflAgent.anyActiveSession(List.of(folder))); } @Test @@ -489,8 +490,8 @@ void anyActiveSession_falseWhenDrainedOrEmpty(@TempDir Path dir) throws IOExcept File session = new File(folder, "sess-1"); Files.createDirectories(session.toPath()); Files.writeString(new File(session, "device.1.log.gz").toPath(), "x"); // finished window, no .tmp - assertFalse(Main.anyActiveSession(List.of(folder))); - assertFalse(Main.anyActiveSession(List.of(new File(folder, "missing")))); // non-existent folder + assertFalse(GpuflAgent.anyActiveSession(List.of(folder))); + assertFalse(GpuflAgent.anyActiveSession(List.of(new File(folder, "missing")))); // non-existent folder } // ---- printUsage() — smoke test ---- @@ -500,7 +501,7 @@ void printUsage_doesNotThrow() { PrintStream original = System.err; try { System.setErr(new PrintStream(PrintStream.nullOutputStream())); - Main.printUsage(); + ConfigLoader.printUsage(); } finally { System.setErr(original); }