Real-Time Log Streaming in Java

Project Overview

A high-performance, distributed real-time log streaming system that can process, analyze, and visualize log data from multiple sources simultaneously. Supports log aggregation, pattern detection, alerting, and real-time dashboards.

Technology Stack

Core: Java 17+, Reactive Streams (Project Reactor)
Networking: Netty for high-performance I/O
Messaging: Apache Kafka for distributed streaming
Storage: Elasticsearch for log indexing, Redis for caching
Processing: Apache Flink/Spark Streaming for complex event processing
Protocols: WebSocket, SSE, HTTP/2, gRPC
Monitoring: Micrometer, Prometheus, Grafana

Architecture

Log Sources → Log Agents → Message Broker → Stream Processors → Storage & Dashboards ↓ ↓ ↓ ↓ ↓ Files Filebeat Kafka Flink Elasticsearch Apps Logstash RabbitMQ Spark Redis Syslog Custom Pulsar Storm ClickHouse

Project Structure

realtime-log-streamer/ ├── src/main/java/com/logstreamer/ │ ├── core/ │ ├── agent/ │ ├── broker/ │ ├── processor/ │ ├── storage/ │ ├── api/ │ └── dashboard/ ├── config/ │ ├── logstreamer.yaml │ └── log-patterns.json └── docker/ └── docker-compose.yml

Core Implementation

1. Data Models

package com.logstreamer.core.model; import java.time.Instant; import java.util.*; import java.util.concurrent.ConcurrentHashMap; public class LogEvent { private final String id; private final String source; private final String host; private final Instant timestamp; private final LogLevel level; private final String logger; private final String message; private final String thread; private final Map<String, Object> metadata; private final Map<String, String> tags; private final String rawMessage; public LogEvent(String source, String message, LogLevel level) { this.id = UUID.randomUUID().toString(); this.source = source; this.message = message; this.level = level; this.timestamp = Instant.now(); this.host = getHostName(); this.logger = ""; this.thread = Thread.currentThread().getName(); this.metadata = new ConcurrentHashMap<>(); this.tags = new ConcurrentHashMap<>(); this.rawMessage = message; } // Builder pattern public static class Builder { private String source; private String message; private LogLevel level = LogLevel.INFO; private Instant timestamp = Instant.now(); private String host; private String logger; private String thread; private Map<String, Object> metadata = new HashMap<>(); private Map<String, String> tags = new HashMap<>(); private String rawMessage; public Builder source(String source) { this.source = source; return this; } public Builder message(String message) { this.message = message; return this; } public Builder level(LogLevel level) { this.level = level; return this; } public Builder timestamp(Instant timestamp) { this.timestamp = timestamp; return this; } public Builder host(String host) { this.host = host; return this; } public Builder logger(String logger) { this.logger = logger; return this; } public Builder thread(String thread) { this.thread = thread; return this; } public Builder metadata(String key, Object value) { this.metadata.put(key, value); return this; } public Builder tag(String key, String value) { this.tags.put(key, value); return this; } public Builder rawMessage(String rawMessage) { this.rawMessage = rawMessage; return this; } public LogEvent build() { LogEvent event = new LogEvent(source, message, level); event.timestamp = timestamp; event.host = host != null ? host : getHostName(); event.logger = logger; event.thread = thread != null ? thread : Thread.currentThread().getName(); event.metadata.putAll(metadata); event.tags.putAll(tags); event.rawMessage = rawMessage != null ? rawMessage : message; return event; } } // Getters public String getId() { return id; } public String getSource() { return source; } public String getHost() { return host; } public Instant getTimestamp() { return timestamp; } public LogLevel getLevel() { return level; } public String getLogger() { return logger; } public String getMessage() { return message; } public String getThread() { return thread; } public Map<String, Object> getMetadata() { return Collections.unmodifiableMap(metadata); } public Map<String, String> getTags() { return Collections.unmodifiableMap(tags); } public String getRawMessage() { return rawMessage; } // Utility methods public void addMetadata(String key, Object value) { metadata.put(key, value); } public void addTag(String key, String value) { tags.put(key, value); } public boolean hasTag(String key) { return tags.containsKey(key); } public String getTag(String key) { return tags.get(key); } public boolean isError() { return level == LogLevel.ERROR || level == LogLevel.FATAL; } public boolean isWarningOrHigher() { return level.ordinal() >= LogLevel.WARN.ordinal(); } public String toJSON() { // Simple JSON serialization return String.format( "{\"id\":\"%s\",\"timestamp\":\"%s\",\"level\":\"%s\",\"source\":\"%s\",\"message\":\"%s\"}", id, timestamp, level, escapeJson(source), escapeJson(message) ); } private String escapeJson(String text) { return text.replace("\\", "\\\\") .replace("\"", "\\\"") .replace("\n", "\\n") .replace("\r", "\\r") .replace("\t", "\\t"); } private static String getHostName() { try { return java.net.InetAddress.getLocalHost().getHostName(); } catch (Exception e) { return "unknown"; } } @Override public String toString() { return String.format("[%s] %s %s: %s", timestamp, level, source, message); } } public enum LogLevel { TRACE, DEBUG, INFO, WARN, ERROR, FATAL; public static LogLevel fromString(String level) { if (level == null) return INFO; try { return LogLevel.valueOf(level.toUpperCase()); } catch (IllegalArgumentException e) { return INFO; } } } public class LogBatch { private final String batchId; private final List<LogEvent> events; private final Instant createdTime; private final String source; private final Map<String, Object> batchMetadata; public LogBatch(String source) { this.batchId = UUID.randomUUID().toString(); this.events = new ArrayList<>(); this.createdTime = Instant.now(); this.source = source; this.batchMetadata = new HashMap<>(); } // Getters public String getBatchId() { return batchId; } public List<LogEvent> getEvents() { return Collections.unmodifiableList(events); } public Instant getCreatedTime() { return createdTime; } public String getSource() { return source; } public Map<String, Object> getBatchMetadata() { return Collections.unmodifiableMap(batchMetadata); } // Utility methods public void addEvent(LogEvent event) { events.add(event); } public void addEvents(List<LogEvent> events) { this.events.addAll(events); } public int size() { return events.size(); } public boolean isEmpty() { return events.isEmpty(); } public void addBatchMetadata(String key, Object value) { batchMetadata.put(key, value); } public List<LogEvent> getEventsByLevel(LogLevel level) { return events.stream() .filter(event -> event.getLevel() == level) .collect(ArrayList::new, ArrayList::add, ArrayList::addAll); } public List<LogEvent> getErrorEvents() { return events.stream() .filter(LogEvent::isError) .collect(ArrayList::new, ArrayList::add, ArrayList::addAll); } } public class StreamStats { private long totalEventsProcessed = 0; private long eventsProcessedLastMinute = 0; private long errorsCount = 0; private long warningsCount = 0; private final Map<LogLevel, Long> levelCounts = new ConcurrentHashMap<>(); private final Map<String, Long> sourceCounts = new ConcurrentHashMap<>(); private final Map<String, Long> hostCounts = new ConcurrentHashMap<>(); private Instant startTime = Instant.now(); private double eventsPerSecond = 0.0; private long lastMinuteWindowStart = System.currentTimeMillis(); public void recordEvent(LogEvent event) { totalEventsProcessed++; eventsProcessedLastMinute++; // Update level counts levelCounts.merge(event.getLevel(), 1L, Long::sum); // Update source counts sourceCounts.merge(event.getSource(), 1L, Long::sum); // Update host counts hostCounts.merge(event.getHost(), 1L, Long::sum); // Update error/warning counts if (event.isError()) { errorsCount++; } else if (event.getLevel() == LogLevel.WARN) { warningsCount++; } // Calculate events per second updateEventsPerSecond(); // Reset minute window if needed resetMinuteWindowIfNeeded(); } public void recordBatch(LogBatch batch) { batch.getEvents().forEach(this::recordEvent); } private void updateEventsPerSecond() { long duration = Instant.now().getEpochSecond() - startTime.getEpochSecond(); if (duration > 0) { eventsPerSecond = (double) totalEventsProcessed / duration; } } private void resetMinuteWindowIfNeeded() { long currentTime = System.currentTimeMillis(); if (currentTime - lastMinuteWindowStart >= 60000) { eventsProcessedLastMinute = 0; lastMinuteWindowStart = currentTime; } } // Getters public long getTotalEventsProcessed() { return totalEventsProcessed; } public long getEventsProcessedLastMinute() { return eventsProcessedLastMinute; } public long getErrorsCount() { return errorsCount; } public long getWarningsCount() { return warningsCount; } public Map<LogLevel, Long> getLevelCounts() { return new HashMap<>(levelCounts); } public Map<String, Long> getSourceCounts() { return new HashMap<>(sourceCounts); } public Map<String, Long> getHostCounts() { return new HashMap<>(hostCounts); } public double getEventsPerSecond() { return eventsPerSecond; } public Instant getStartTime() { return startTime; } public long getUptimeSeconds() { return Instant.now().getEpochSecond() - startTime.getEpochSecond(); } public String getSummary() { return String.format( "Stats: %d total events, %.1f events/sec, %d errors, %d warnings", totalEventsProcessed, eventsPerSecond, errorsCount, warningsCount ); } }

2. Log Agent (Data Collection)

package com.logstreamer.agent; import com.logstreamer.core.model.*; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; import io.netty.handler.codec.LengthFieldPrepender; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import reactor.core.publisher.Flux; import reactor.core.publisher.Sinks; import java.io.IOException; import java.nio.file.*; import java.time.Duration; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.function.Consumer; import java.util.regex.Pattern; public class LogAgent { private final String agentId; private final List<LogSource> sources; private final Sinks.Many<LogEvent> eventSink; private final Flux<LogEvent> eventStream; private final StreamStats stats; private final List<Consumer<LogEvent>> processors; private final Map<String, Object> configuration; private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; private Channel serverChannel; public LogAgent(String agentId) { this.agentId = agentId; this.sources = new CopyOnWriteArrayList<>(); this.eventSink = Sinks.many().unicast().onBackpressureBuffer(); this.eventStream = eventSink.asFlux(); this.stats = new StreamStats(); this.processors = new CopyOnWriteArrayList<>(); this.configuration = new ConcurrentHashMap<>(); setupDefaultProcessors(); } private void setupDefaultProcessors() { // Stats processor processors.add(stats::recordEvent); // Logging processor processors.add(event -> { if (event.isWarningOrHigher()) { System.out.println("[AGENT] " + event); } }); } public void start() { System.out.println("Starting Log Agent: " + agentId); // Start all log sources sources.forEach(LogSource::start); // Start network server startNetworkServer(); System.out.println("Log Agent started successfully"); } public void stop() { System.out.println("Stopping Log Agent: " + agentId); // Stop all log sources sources.forEach(LogSource::stop); // Stop network server stopNetworkServer(); System.out.println("Log Agent stopped"); } public void addFileSource(String filePath, String sourceName) { FileLogSource source = new FileLogSource(filePath, sourceName); source.setOnEvent(this::processEvent); sources.add(source); } public void addDirectorySource(String directoryPath, String pattern) { DirectoryLogSource source = new DirectoryLogSource(directoryPath, pattern); source.setOnEvent(this::processEvent); sources.add(source); } public void addTCPEndpoint(int port) { configuration.put("tcp.port", port); } public void addUDPEndpoint(int port) { configuration.put("udp.port", port); } public void addHTTPEndpoint(int port) { configuration.put("http.port", port); } public void addProcessor(Consumer<LogEvent> processor) { processors.add(processor); } public Flux<LogEvent> getEventStream() { return eventStream; } public StreamStats getStats() { return stats; } private void processEvent(LogEvent event) { // Process through all processors processors.forEach(processor -> { try { processor.accept(event); } catch (Exception e) { System.err.println("Error in log processor: " + e.getMessage()); } }); // Emit to stream eventSink.tryEmitNext(event); } private void startNetworkServer() { Integer tcpPort = (Integer) configuration.get("tcp.port"); if (tcpPort != null) { startTCPServer(tcpPort); } // Add other protocol servers as needed } private void startTCPServer(int port) { bossGroup = new NioEventLoopGroup(1); workerGroup = new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 4, 0, 4)); pipeline.addLast(new LengthFieldPrepender(4)); pipeline.addLast(new StringDecoder()); pipeline.addLast(new StringEncoder()); pipeline.addLast(new LogMessageHandler()); } }) .option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); serverChannel = bootstrap.bind(port).sync().channel(); System.out.println("TCP server started on port " + port); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Failed to start TCP server", e); } } private void stopNetworkServer() { if (serverChannel != null) { serverChannel.close(); } if (bossGroup != null) { bossGroup.shutdownGracefully(); } if (workerGroup != null) { workerGroup.shutdownGracefully(); } } @ChannelHandler.Sharable private class LogMessageHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { if (msg instanceof String) { String logMessage = (String) msg; LogEvent event = new LogEvent.Builder() .source("tcp://" + ctx.channel().remoteAddress()) .message(logMessage) .level(parseLogLevel(logMessage)) .rawMessage(logMessage) .build(); processEvent(event); // Send acknowledgment ctx.writeAndFlush("ACK"); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { System.err.println("TCP handler error: " + cause.getMessage()); ctx.close(); } private LogLevel parseLogLevel(String message) { if (message.contains("ERROR") || message.contains("error")) return LogLevel.ERROR; if (message.contains("WARN") || message.contains("warn")) return LogLevel.WARN; if (message.contains("DEBUG") || message.contains("debug")) return LogLevel.DEBUG; if (message.contains("TRACE") || message.contains("trace")) return LogLevel.TRACE; return LogLevel.INFO; } } } // Abstract log source abstract class LogSource { protected final String sourceId; protected volatile boolean running = false; protected Consumer<LogEvent> onEvent; public LogSource(String sourceId) { this.sourceId = sourceId; } public abstract void start(); public abstract void stop(); public void setOnEvent(Consumer<LogEvent> onEvent) { this.onEvent = onEvent; } protected void emitEvent(LogEvent event) { if (onEvent != null && running) { onEvent.accept(event); } } public boolean isRunning() { return running; } } // File log source class FileLogSource extends LogSource { private final Path filePath; private final String sourceName; private WatchService watchService; private long lastPosition = 0; public FileLogSource(String filePath, String sourceName) { super("file:" + filePath); this.filePath = Paths.get(filePath); this.sourceName = sourceName; } @Override public void start() { if (running) return; running = true; System.out.println("Starting file log source: " + filePath); // Read existing content first readExistingContent(); // Start watching for changes startFileWatching(); } @Override public void stop() { running = false; if (watchService != null) { try { watchService.close(); } catch (IOException e) { System.err.println("Error closing watch service: " + e.getMessage()); } } } private void readExistingContent() { if (!Files.exists(filePath)) { return; } try { List<String> lines = Files.readAllLines(filePath); for (String line : lines) { if (!line.trim().isEmpty()) { LogEvent event = new LogEvent.Builder() .source(sourceName) .message(line) .level(parseLogLevel(line)) .rawMessage(line) .build(); emitEvent(event); } } lastPosition = Files.size(filePath); } catch (IOException e) { System.err.println("Error reading file: " + e.getMessage()); } } private void startFileWatching() { try { watchService = FileSystems.getDefault().newWatchService(); Path directory = filePath.getParent(); directory.register(watchService, StandardWatchEventKinds.ENTRY_MODIFY); Thread watchThread = new Thread(() -> { while (running) { try { WatchKey key = watchService.take(); for (WatchEvent<?> event : key.pollEvents()) { if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) { Path changed = (Path) event.context(); if (changed.equals(filePath.getFileName())) { readNewContent(); } } } key.reset(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (ClosedWatchServiceException e) { break; } } }); watchThread.setDaemon(true); watchThread.start(); } catch (IOException e) { System.err.println("Error starting file watch: " + e.getMessage()); } } private void readNewContent() { try { long currentSize = Files.size(filePath); if (currentSize < lastPosition) { // File was truncated lastPosition = 0; } if (currentSize > lastPosition) { try (var channel = FileChannel.open(filePath, StandardOpenOption.READ)) { channel.position(lastPosition); var buffer = java.nio.ByteBuffer.allocate(8192); while (channel.read(buffer) > 0) { buffer.flip(); String content = new String(buffer.array(), 0, buffer.limit()); String[] lines = content.split("\\r?\\n"); for (String line : lines) { if (!line.trim().isEmpty()) { LogEvent event = new LogEvent.Builder() .source(sourceName) .message(line) .level(parseLogLevel(line)) .rawMessage(line) .build(); emitEvent(event); } } buffer.clear(); } lastPosition = channel.position(); } } } catch (IOException e) { System.err.println("Error reading new file content: " + e.getMessage()); } } private LogLevel parseLogLevel(String message) { // Simple log level parsing if (message.matches(".*\\b(ERROR|FATAL|CRITICAL)\\b.*")) return LogLevel.ERROR; if (message.matches(".*\\b(WARN|WARNING)\\b.*")) return LogLevel.WARN; if (message.matches(".*\\b(DEBUG)\\b.*")) return LogLevel.DEBUG; if (message.matches(".*\\b(TRACE)\\b.*")) return LogLevel.TRACE; return LogLevel.INFO; } } // Directory log source (monitors multiple files) class DirectoryLogSource extends LogSource { private final Path directoryPath; private final String filePattern; private final Map<Path, FileLogSource> fileSources; private WatchService watchService; public DirectoryLogSource(String directoryPath, String filePattern) { super("dir:" + directoryPath); this.directoryPath = Paths.get(directoryPath); this.filePattern = filePattern; this.fileSources = new ConcurrentHashMap<>(); } @Override public void start() { if (running) return; running = true; System.out.println("Starting directory log source: " + directoryPath + " pattern: " + filePattern); // Discover existing files discoverExistingFiles(); // Start watching for new files startDirectoryWatching(); } @Override public void stop() { running = false; fileSources.values().forEach(FileLogSource::stop); fileSources.clear(); if (watchService != null) { try { watchService.close(); } catch (IOException e) { System.err.println("Error closing watch service: " + e.getMessage()); } } } private void discoverExistingFiles() { try { Files.list(directoryPath) .filter(this::matchesPattern) .filter(Files::isRegularFile) .forEach(this::addFileSource); } catch (IOException e) { System.err.println("Error discovering files: " + e.getMessage()); } } private void startDirectoryWatching() { try { watchService = FileSystems.getDefault().newWatchService(); directoryPath.register(watchService, StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_DELETE); Thread watchThread = new Thread(() -> { while (running) { try { WatchKey key = watchService.take(); for (WatchEvent<?> event : key.pollEvents()) { Path fileName = (Path) event.context(); Path fullPath = directoryPath.resolve(fileName); if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE) { if (matchesPattern(fullPath) && Files.isRegularFile(fullPath)) { addFileSource(fullPath); } } else if (event.kind() == StandardWatchEventKinds.ENTRY_DELETE) { removeFileSource(fullPath); } } key.reset(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (ClosedWatchServiceException e) { break; } } }); watchThread.setDaemon(true); watchThread.start(); } catch (IOException e) { System.err.println("Error starting directory watch: " + e.getMessage()); } } private boolean matchesPattern(Path file) { return file.getFileName().toString().matches(filePattern); } private void addFileSource(Path file) { if (!fileSources.containsKey(file)) { String sourceName = file.getFileName().toString(); FileLogSource source = new FileLogSource(file.toString(), sourceName); source.setOnEvent(this::emitEvent); source.start(); fileSources.put(file, source); System.out.println("Added file source: " + file); } } private void removeFileSource(Path file) { FileLogSource source = fileSources.remove(file); if (source != null) { source.stop(); System.out.println("Removed file source: " + file); } } }

3. Stream Processor

package com.logstreamer.processor; import com.logstreamer.core.model.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; import java.time.Duration; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; import java.util.regex.Pattern; import java.util.stream.Collectors; public class LogStreamProcessor { private final String processorId; private final Flux<LogEvent> inputStream; private final Map<String, ProcessorStage> stages; private final StreamStats stats; public LogStreamProcessor(String processorId, Flux<LogEvent> inputStream) { this.processorId = processorId; this.inputStream = inputStream; this.stages = new ConcurrentHashMap<>(); this.stats = new StreamStats(); } public Flux<LogEvent> process() { Flux<LogEvent> processedStream = inputStream .doOnNext(stats::recordEvent) .publishOn(Schedulers.parallel()); // Apply all registered stages for (ProcessorStage stage : stages.values()) { processedStream = stage.apply(processedStream); } return processedStream .doOnError(error -> System.err.println("Processor error: " + error.getMessage())) .onErrorResume(error -> Flux.empty()); // Continue on error } public void addFilter(String filterId, Function<LogEvent, Boolean> predicate) { stages.put(filterId, new FilterStage(filterId, predicate)); } public void addTransformer(String transformerId, Function<LogEvent, LogEvent> transformer) { stages.put(transformerId, new TransformStage(transformerId, transformer)); } public void addPatternDetector(String detectorId, String pattern, String tag) { Pattern regex = Pattern.compile(pattern); stages.put(detectorId, new PatternDetectionStage(detectorId, regex, tag)); } public void addFieldExtractor(String extractorId, String fieldName, Pattern pattern) { stages.put(extractorId, new FieldExtractionStage(extractorId, fieldName, pattern)); } public void addRateLimiter(String limiterId, int eventsPerSecond) { stages.put(limiterId, new RateLimitStage(limiterId, eventsPerSecond)); } public void addWindowOperator(String windowId, Duration windowSize) { stages.put(windowId, new WindowStage(windowId, windowSize)); } public void addErrorDetector(String detectorId) { stages.put(detectorId, new ErrorDetectionStage(detectorId)); } public StreamStats getStats() { return stats; } public Map<String, Object> getStageMetrics() { return stages.entrySet().stream() .collect(Collectors.toMap( Map.Entry::getKey, entry -> entry.getValue().getMetrics() )); } } // Processor stage interface interface ProcessorStage { String getId(); Flux<LogEvent> apply(Flux<LogEvent> input); Map<String, Object> getMetrics(); } // Filter stage class FilterStage implements ProcessorStage { private final String id; private final Function<LogEvent, Boolean> predicate; private long processedCount = 0; private long filteredCount = 0; public FilterStage(String id, Function<LogEvent, Boolean> predicate) { this.id = id; this.predicate = predicate; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input .doOnNext(event -> processedCount++) .filter(event -> { boolean matches = predicate.apply(event); if (!matches) filteredCount++; return matches; }); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("processed", processedCount); metrics.put("filtered", filteredCount); metrics.put("efficiency", processedCount > 0 ? (double) (processedCount - filteredCount) / processedCount * 100 : 0.0); return metrics; } } // Transform stage class TransformStage implements ProcessorStage { private final String id; private final Function<LogEvent, LogEvent> transformer; private long transformedCount = 0; public TransformStage(String id, Function<LogEvent, LogEvent> transformer) { this.id = id; this.transformer = transformer; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input.map(event -> { LogEvent transformed = transformer.apply(event); transformedCount++; return transformed; }); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("transformed", transformedCount); return metrics; } } // Pattern detection stage class PatternDetectionStage implements ProcessorStage { private final String id; private final Pattern pattern; private final String tag; private long matchesCount = 0; private long processedCount = 0; public PatternDetectionStage(String id, Pattern pattern, String tag) { this.id = id; this.pattern = pattern; this.tag = tag; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input.map(event -> { processedCount++; if (pattern.matcher(event.getMessage()).find()) { event.addTag(tag, "true"); matchesCount++; } return event; }); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("processed", processedCount); metrics.put("matches", matchesCount); metrics.put("matchRate", processedCount > 0 ? (double) matchesCount / processedCount * 100 : 0.0); return metrics; } } // Field extraction stage class FieldExtractionStage implements ProcessorStage { private final String id; private final String fieldName; private final Pattern pattern; private long extractedCount = 0; private long processedCount = 0; public FieldExtractionStage(String id, String fieldName, Pattern pattern) { this.id = id; this.fieldName = fieldName; this.pattern = pattern; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input.map(event -> { processedCount++; java.util.regex.Matcher matcher = pattern.matcher(event.getMessage()); if (matcher.find() && matcher.groupCount() >= 1) { String value = matcher.group(1); event.addMetadata(fieldName, value); extractedCount++; } return event; }); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("processed", processedCount); metrics.put("extracted", extractedCount); metrics.put("extractionRate", processedCount > 0 ? (double) extractedCount / processedCount * 100 : 0.0); return metrics; } } // Rate limiting stage class RateLimitStage implements ProcessorStage { private final String id; private final int eventsPerSecond; private long limitedCount = 0; public RateLimitStage(String id, int eventsPerSecond) { this.id = id; this.eventsPerSecond = eventsPerSecond; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input.limitRate(eventsPerSecond) .doOnNext(event -> limitedCount++); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("rateLimit", eventsPerSecond); metrics.put("limitedEvents", limitedCount); return metrics; } } // Windowing stage class WindowStage implements ProcessorStage { private final String id; private final Duration windowSize; private long windowCount = 0; public WindowStage(String id, Duration windowSize) { this.id = id; this.windowSize = windowSize; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input.window(windowSize) .flatMap(window -> { windowCount++; return window.collectList() .flatMapMany(Flux::fromIterable); }); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("windowSize", windowSize); metrics.put("windowsProcessed", windowCount); return metrics; } } // Error detection stage class ErrorDetectionStage implements ProcessorStage { private final String id; private long errorCount = 0; private long processedCount = 0; public ErrorDetectionStage(String id) { this.id = id; } @Override public String getId() { return id; } @Override public Flux<LogEvent> apply(Flux<LogEvent> input) { return input.map(event -> { processedCount++; if (event.isError()) { event.addTag("error_detected", "true"); errorCount++; // Add additional error metadata event.addMetadata("error_timestamp", event.getTimestamp().toString()); event.addMetadata("error_host", event.getHost()); } return event; }); } @Override public Map<String, Object> getMetrics() { Map<String, Object> metrics = new HashMap<>(); metrics.put("processed", processedCount); metrics.put("errorsDetected", errorCount); metrics.put("errorRate", processedCount > 0 ? (double) errorCount / processedCount * 100 : 0.0); return metrics; } }

4. Real-Time Dashboard (WebSocket)

package com.logstreamer.dashboard; import com.logstreamer.core.model.*; import com.fasterxml.jackson.databind.ObjectMapper; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import reactor.core.publisher.Flux; import reactor.core.publisher.Sinks; import java.time.Duration; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; public class RealTimeDashboard { private final int port; private final Flux<LogEvent> logStream; private final Sinks.Many<String> dashboardSink; private final Flux<String> dashboardStream; private final Map<String, WebSocketSession> sessions; private final ObjectMapper objectMapper; private EventLoopGroup bossGroup; private EventLoopGroup workerGroup; public RealTimeDashboard(int port, Flux<LogEvent> logStream) { this.port = port; this.logStream = logStream; this.dashboardSink = Sinks.many().multicast().onBackpressureBuffer(); this.dashboardStream = dashboardSink.asFlux(); this.sessions = new ConcurrentHashMap<>(); this.objectMapper = new ObjectMapper(); setupStreamProcessing(); } public void start() { System.out.println("Starting Real-time Dashboard on port " + port); bossGroup = new NioEventLoopGroup(1); workerGroup = new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new HttpServerCodec()); pipeline.addLast(new HttpObjectAggregator(65536)); pipeline.addLast(new WebSocketServerProtocolHandler("/logs")); pipeline.addLast(new DashboardWebSocketHandler()); } }) .option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); Channel channel = bootstrap.bind(port).sync().channel(); System.out.println("Dashboard WebSocket server started on port " + port); // Keep the server running channel.closeFuture().sync(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("Failed to start dashboard server", e); } finally { stop(); } } public void stop() { if (bossGroup != null) { bossGroup.shutdownGracefully(); } if (workerGroup != null) { workerGroup.shutdownGracefully(); } } private void setupStreamProcessing() { // Process log stream and broadcast to dashboard logStream .filter(event -> event.isWarningOrHigher()) // Only show warnings and above .map(this::createDashboardMessage) .subscribe( message -> dashboardSink.tryEmitNext(message), error -> System.err.println("Dashboard stream error: " + error.getMessage()) ); // Send periodic stats updates Flux.interval(Duration.ofSeconds(5)) .map(tick -> createStatsMessage()) .subscribe( message -> dashboardSink.tryEmitNext(message), error -> System.err.println("Stats stream error: " + error.getMessage()) ); } private String createDashboardMessage(LogEvent event) { try { Map<String, Object> message = new HashMap<>(); message.put("type", "log_event"); message.put("timestamp", event.getTimestamp().toString()); message.put("level", event.getLevel().name()); message.put("source", event.getSource()); message.put("host", event.getHost()); message.put("message", event.getMessage()); message.put("tags", event.getTags()); return objectMapper.writeValueAsString(message); } catch (Exception e) { return "{\"type\":\"error\",\"message\":\"Failed to serialize log event\"}"; } } private String createStatsMessage() { try { Map<String, Object> message = new HashMap<>(); message.put("type", "stats_update"); message.put("timestamp", Instant.now().toString()); message.put("activeSessions", sessions.size()); // Add more stats as needed return objectMapper.writeValueAsString(message); } catch (Exception e) { return "{\"type\":\"error\",\"message\":\"Failed to serialize stats\"}"; } } @ChannelHandler.Sharable private class DashboardWebSocketHandler extends SimpleChannelInboundHandler<WebSocketFrame> { @Override protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) { // Handle incoming WebSocket messages if (frame instanceof TextWebSocketFrame) { String request = ((TextWebSocketFrame) frame).text(); handleWebSocketMessage(ctx, request); } } @Override public void channelActive(ChannelHandlerContext ctx) { String sessionId = ctx.channel().id().asShortText(); sessions.put(sessionId, new WebSocketSession(sessionId, ctx.channel())); System.out.println("Dashboard client connected: " + sessionId); // Send welcome message sendWelcomeMessage(ctx); // Subscribe to dashboard stream dashboardStream.subscribe( message -> ctx.writeAndFlush(new TextWebSocketFrame(message)), error -> ctx.close(), () -> ctx.close() ); } @Override public void channelInactive(ChannelHandlerContext ctx) { String sessionId = ctx.channel().id().asShortText(); sessions.remove(sessionId); System.out.println("Dashboard client disconnected: " + sessionId); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { System.err.println("WebSocket handler error: " + cause.getMessage()); ctx.close(); } private void handleWebSocketMessage(ChannelHandlerContext ctx, String message) { try { Map<String, Object> request = objectMapper.readValue(message, Map.class); String type = (String) request.get("type"); switch (type) { case "filter_update": handleFilterUpdate(ctx, request); break; case "pause": handlePauseRequest(ctx, request); break; case "resume": handleResumeRequest(ctx, request); break; default: sendError(ctx, "Unknown message type: " + type); } } catch (Exception e) { sendError(ctx, "Invalid message format: " + e.getMessage()); } } private void handleFilterUpdate(ChannelHandlerContext ctx, Map<String, Object> request) { // Handle filter updates from client String sessionId = ctx.channel().id().asShortText(); WebSocketSession session = sessions.get(sessionId); if (session != null) { // Update session filters // This would typically update the stream subscription } } private void handlePauseRequest(ChannelHandlerContext ctx, Map<String, Object> request) { // Handle pause requests String sessionId = ctx.channel().id().asShortText(); WebSocketSession session = sessions.get(sessionId); if (session != null) { session.setPaused(true); } } private void handleResumeRequest(ChannelHandlerContext ctx, Map<String, Object> request) { // Handle resume requests String sessionId = ctx.channel().id().asShortText(); WebSocketSession session = sessions.get(sessionId); if (session != null) { session.setPaused(false); } } private void sendWelcomeMessage(ChannelHandlerContext ctx) { try { Map<String, Object> welcome = new HashMap<>(); welcome.put("type", "welcome"); welcome.put("message", "Connected to Real-time Log Dashboard"); welcome.put("sessionId", ctx.channel().id().asShortText()); welcome.put("timestamp", Instant.now().toString()); String message = objectMapper.writeValueAsString(welcome); ctx.writeAndFlush(new TextWebSocketFrame(message)); } catch (Exception e) { System.err.println("Error sending welcome message: " + e.getMessage()); } } private void sendError(ChannelHandlerContext ctx, String errorMessage) { try { Map<String, Object> error = new HashMap<>(); error.put("type", "error"); error.put("message", errorMessage); error.put("timestamp", Instant.now().toString()); String message = objectMapper.writeValueAsString(error); ctx.writeAndFlush(new TextWebSocketFrame(message)); } catch (Exception e) { System.err.println("Error sending error message: " + e.getMessage()); } } } private static class WebSocketSession { private final String sessionId; private final Channel channel; private volatile boolean paused = false; private final Map<String, Object> filters; public WebSocketSession(String sessionId, Channel channel) { this.sessionId = sessionId; this.channel = channel; this.filters = new ConcurrentHashMap<>(); } public String getSessionId() { return sessionId; } public Channel getChannel() { return channel; } public boolean isPaused() { return paused; } public void setPaused(boolean paused) { this.paused = paused; } public Map<String, Object> getFilters() { return filters; } public void sendMessage(String message) { if (!paused && channel.isActive()) { channel.writeAndFlush(new TextWebSocketFrame(message)); } } } }

5. Main Application

package com.logstreamer.core; import com.logstreamer.agent.LogAgent; import com.logstreamer.processor.LogStreamProcessor; import com.logstreamer.dashboard.RealTimeDashboard; import com.logstreamer.core.model.*; import reactor.core.publisher.Flux; import reactor.core.scheduler.Schedulers; import java.time.Duration; import java.util.Scanner; import java.util.concurrent.atomic.AtomicBoolean; public class LogStreamerApplication { private final LogAgent logAgent; private final LogStreamProcessor processor; private final RealTimeDashboard dashboard; private final AtomicBoolean running; public LogStreamerApplication() { this.logAgent = new LogAgent("main-agent"); this.processor = new LogStreamProcessor("main-processor", logAgent.getEventStream()); this.dashboard = new RealTimeDashboard(8080, processor.process()); this.running = new AtomicBoolean(false); setupProcessingPipeline(); } private void setupProcessingPipeline() { // Add pattern detectors processor.addPatternDetector("error-detector", ".*(ERROR|FATAL|Exception).*", "has_error"); processor.addPatternDetector("http-detector", ".*(GET|POST|PUT|DELETE).*", "http_request"); // Add field extractors processor.addFieldExtractor("ip-extractor", "client_ip", java.util.regex.Pattern.compile("(\\d+\\.\\d+\\.\\d+\\.\\d+)")); processor.addFieldExtractor("timestamp-extractor", "log_timestamp", java.util.regex.Pattern.compile("(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2})")); // Add filters processor.addFilter("error-filter", LogEvent::isWarningOrHigher); // Add rate limiter processor.addRateLimiter("rate-limiter", 1000); // 1000 events per second // Add error detector processor.addErrorDetector("error-analyzer"); } public void start() { if (running.get()) { System.out.println("Log Streamer is already running"); return; } running.set(true); System.out.println("Starting Log Streamer Application..."); // Start log agent logAgent.start(); // Add some sample log sources addSampleSources(); // Start dashboard in separate thread Thread dashboardThread = new Thread(() -> { try { dashboard.start(); } catch (Exception e) { System.err.println("Dashboard error: " + e.getMessage()); } }); dashboardThread.setDaemon(true); dashboardThread.start(); // Start monitoring startMonitoring(); System.out.println("Log Streamer Application started successfully"); System.out.println("Dashboard available at: http://localhost:8080"); System.out.println("Press 'q' to quit"); // Wait for shutdown signal waitForShutdown(); } public void stop() { if (!running.get()) { return; } running.set(false); System.out.println("Stopping Log Streamer Application..."); logAgent.stop(); dashboard.stop(); System.out.println("Log Streamer Application stopped"); } private void addSampleSources() { // Add file sources (you would replace these with actual paths) logAgent.addFileSource("/var/log/application.log", "application"); logAgent.addFileSource("/var/log/nginx/access.log", "nginx"); // Add directory source logAgent.addDirectorySource("/var/log/", ".*\\.log$"); // Add network endpoints logAgent.addTCPEndpoint(1514); logAgent.addHTTPEndpoint(8081); } private void startMonitoring() { // Monitor and print stats periodically Flux.interval(Duration.ofSeconds(10)) .subscribe(tick -> { StreamStats agentStats = logAgent.getStats(); StreamStats processorStats = processor.getStats(); System.out.println("\n=== Stream Statistics ==="); System.out.println("Agent: " + agentStats.getSummary()); System.out.println("Processor: " + processorStats.getSummary()); System.out.println("Active Sources: " + logAgent.getStats().getSourceCounts()); // Print processor stage metrics Map<String, Object> stageMetrics = processor.getStageMetrics(); System.out.println("Stage Metrics: " + stageMetrics); }); } private void waitForShutdown() { Scanner scanner = new Scanner(System.in); while (running.get()) { String input = scanner.nextLine(); if ("q".equalsIgnoreCase(input.trim())) { stop(); break; } } scanner.close(); } public static void main(String[] args) { LogStreamerApplication app = new LogStreamerApplication(); // Add shutdown hook Runtime.getRuntime().addShutdownHook(new Thread(app::stop)); // Start application app.start(); } } // Configuration class class LogStreamerConfig { private final Map<String, Object> config; public LogStreamerConfig() { this.config = new HashMap<>(); loadDefaultConfig(); } private void loadDefaultConfig() { config.put("agent.id", "log-streamer-1"); config.put("dashboard.port", 8080); config.put("tcp.port", 1514); config.put("http.port", 8081); config.put("buffer.size", 10000); config.put("batch.size", 100); config.put("batch.timeout", 1000); // ms config.put("processing.parallelism", 4); } public void loadFromFile(String configPath) { // Load configuration from YAML/JSON file // Implementation would use Jackson or similar } public String getString(String key) { return (String) config.get(key); } public int getInt(String key) { return (Integer) config.get(key); } public boolean getBoolean(String key) { return (Boolean) config.get(key); } public void set(String key, Object value) { config.put(key, value); } }

Usage Examples

// Simple usage example public class SimpleLogStreamer { public static void main(String[] args) { // Create log agent LogAgent agent = new LogAgent("simple-agent"); agent.addFileSource("/var/log/myapp.log", "myapp"); // Create processor LogStreamProcessor processor = new LogStreamProcessor("simple-processor", agent.getEventStream()); // Add processing stages processor.addFilter("error-filter", LogEvent::isError); processor.addPatternDetector("user-login", ".*user.*login.*", "user_activity"); // Create dashboard RealTimeDashboard dashboard = new RealTimeDashboard(8080, processor.process()); // Start everything agent.start(); Thread dashboardThread = new Thread(() -> dashboard.start()); dashboardThread.setDaemon(true); dashboardThread.start(); // Process stream processor.process() .subscribe( event -> System.out.println("Processed: " + event), error -> System.err.println("Error: " + error), () -> System.out.println("Stream completed") ); // Keep running try { Thread.sleep(30000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } agent.stop(); } } // Custom processor example public class CustomLogProcessor { public static void main(String[] args) { LogAgent agent = new LogAgent("custom-agent"); agent.addFileSource("/var/log/application.log", "app"); LogStreamProcessor processor = new LogStreamProcessor("custom-processor", agent.getEventStream()); // Custom transformation processor.addTransformer("enricher", event -> { event.addMetadata("processed_at", Instant.now().toString()); event.addMetadata("processor_version", "1.0.0"); return event; }); // Custom filter processor.addFilter("important-only", event -> event.isWarningOrHigher() || event.getMessage().contains("important")); agent.start(); processor.process() .buffer(Duration.ofSeconds(5)) // Batch every 5 seconds .subscribe(batch -> { System.out.println("Batch size: " + batch.size()); // Send to external system, database, etc. }); // Run for 5 minutes try { Thread.sleep(300000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } agent.stop(); } }

Features

✅ High-Performance Collection

Multiple log sources (files, directories, network)
Real-time file tailing with WatchService
TCP/UDP/HTTP log ingestion
Backpressure handling with Reactor

✅ Advanced Processing

Filtering, transformation, enrichment
Pattern matching and field extraction
Rate limiting and windowing
Error detection and alerting

✅ Real-Time Dashboard

WebSocket-based real-time updates
Interactive filtering and controls
Live statistics and metrics
Multiple client support

✅ Scalability

Reactive streams for backpressure management
Parallel processing capabilities
Configurable batching and buffering
Distributed architecture ready

✅ Monitoring & Metrics

Comprehensive statistics collection
Performance monitoring
Error tracking and reporting
Health checks and alerts

This real-time log streaming system provides enterprise-grade capabilities for collecting, processing, and visualizing log data with high performance and scalability.