google-site-verification: google61fe8ba583a51912.html
MACRO NEPAL
Vector Log Agent in Java
awiskar acharya
November 17, 2025
10:54 am

Introduction

Vector is a high-performance observability data pipeline. This Java implementation provides a lightweight, high-performance log agent that can collect, transform, and route log data to various destinations with minimal resource overhead.

Architecture Overview

Vector Agent Architecture

public class VectorLogAgent { /** * Vector Log Agent Components: * 1. Sources - Collect logs from various inputs * 2. Transforms - Process and enrich log data * 3. Sinks - Route logs to destinations * 4. Buffer - Manage backpressure and batching * 5. Metrics - Monitor performance and errors */ private List<LogSource> sources; private List<LogTransform> transforms; private List<LogSink> sinks; private BufferManager bufferManager; private MetricsCollector metricsCollector; public void start() { sources.forEach(LogSource::start); bufferManager.start(); metricsCollector.start(); } }

Core Components

Log Event Model

public class LogEvent { private final String id; private final String message; private final Map<String, Object> fields; private final Map<String, String> metadata; private final Instant timestamp; private final LogLevel level; private final String source; public LogEvent(String message, Map<String, Object> fields, Map<String, String> metadata, Instant timestamp, LogLevel level, String source) { this.id = UUID.randomUUID().toString(); this.message = message; this.fields = fields != null ? new HashMap<>(fields) : new HashMap<>(); this.metadata = metadata != null ? new HashMap<>(metadata) : new HashMap<>(); this.timestamp = timestamp != null ? timestamp : Instant.now(); this.level = level != null ? level : LogLevel.INFO; this.source = source; } // Builder pattern for easy creation public static class Builder { private String message; private Map<String, Object> fields = new HashMap<>(); private Map<String, String> metadata = new HashMap<>(); private Instant timestamp; private LogLevel level; private String source; public Builder message(String message) { this.message = message; return this; } public Builder field(String key, Object value) { this.fields.put(key, value); return this; } public Builder metadata(String key, String value) { this.metadata.put(key, value); return this; } public Builder timestamp(Instant timestamp) { this.timestamp = timestamp; return this; } public Builder level(LogLevel level) { this.level = level; return this; } public Builder source(String source) { this.source = source; return this; } public LogEvent build() { return new LogEvent(message, fields, metadata, timestamp, level, source); } } // Getters public String getId() { return id; } public String getMessage() { return message; } public Map<String, Object> getFields() { return fields; } public Instant getTimestamp() { return timestamp; } public void addField(String key, Object value) { fields.put(key, value); } public void addMetadata(String key, String value) { metadata.put(key, value); } public String toJSON() { ObjectMapper mapper = new ObjectMapper(); try { Map<String, Object> jsonMap = new HashMap<>(); jsonMap.put("id", id); jsonMap.put("message", message); jsonMap.put("timestamp", timestamp.toString()); jsonMap.put("level", level.name()); jsonMap.put("source", source); jsonMap.put("fields", fields); jsonMap.put("metadata", metadata); return mapper.writeValueAsString(jsonMap); } catch (JsonProcessingException e) { throw new RuntimeException("Failed to serialize log event to JSON", e); } } } public enum LogLevel { TRACE, DEBUG, INFO, WARN, ERROR, FATAL }

Source Interface and Implementations

public interface LogSource { String getName(); void start(); void stop(); boolean isRunning(); void addListener(LogEventListener listener); void removeListener(LogEventListener listener); } public interface LogEventListener { void onEvent(LogEvent event); } @Component public class FileLogSource implements LogSource { private static final Logger logger = LoggerFactory.getLogger(FileLogSource.class); private final String name; private final Path filePath; private final Charset charset; private final Set<LogEventListener> listeners; private volatile boolean running; private Thread watchThread; private long lastPosition; public FileLogSource(String name, String filePath, Charset charset) { this.name = name; this.filePath = Paths.get(filePath); this.charset = charset; this.listeners = ConcurrentHashMap.newKeySet(); this.lastPosition = 0; } @Override public String getName() { return name; } @Override public void start() { if (running) { logger.warn("File log source {} is already running", name); return; } running = true; watchThread = new Thread(this::watchFile, "file-watcher-" + name); watchThread.setDaemon(true); watchThread.start(); logger.info("Started file log source: {} watching {}", name, filePath); } @Override public void stop() { running = false; if (watchThread != null) { watchThread.interrupt(); try { watchThread.join(5000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } logger.info("Stopped file log source: {}", name); } @Override public boolean isRunning() { return running; } @Override public void addListener(LogEventListener listener) { listeners.add(listener); } @Override public void removeListener(LogEventListener listener) { listeners.remove(listener); } private void watchFile() { try { // Initialize file position if (Files.exists(filePath)) { lastPosition = Files.size(filePath); } while (running) { if (Files.exists(filePath)) { readNewLines(); } Thread.sleep(1000); // Check every second } } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.debug("File watcher interrupted for {}", name); } catch (Exception e) { logger.error("Error watching file {}", filePath, e); } } private void readNewLines() throws IOException { long currentSize = Files.size(filePath); if (currentSize < lastPosition) { // File was truncated lastPosition = 0; } if (currentSize > lastPosition) { try (FileChannel channel = FileChannel.open(filePath, StandardOpenOption.READ); BufferedReader reader = new BufferedReader( Channels.newReader(channel, charset.name()))) { channel.position(lastPosition); String line; int lineNumber = 0; while ((line = reader.readLine()) != null) { LogEvent event = new LogEvent.Builder() .message(line) .source(name) .field("file_path", filePath.toString()) .field("line_number", ++lineNumber) .build(); notifyListeners(event); } lastPosition = channel.position(); } } } private void notifyListeners(LogEvent event) { for (LogEventListener listener : listeners) { try { listener.onEvent(event); } catch (Exception e) { logger.error("Error notifying listener for event {}", event.getId(), e); } } } } @Component public class SyslogSource implements LogSource { private static final Logger logger = LoggerFactory.getLogger(SyslogSource.class); private final String name; private final int port; private final Set<LogEventListener> listeners; private volatile boolean running; private ServerSocket serverSocket; private ExecutorService executor; public SyslogSource(String name, int port) { this.name = name; this.port = port; this.listeners = ConcurrentHashMap.newKeySet(); } @Override public String getName() { return name; } @Override public void start() { if (running) { return; } try { serverSocket = new ServerSocket(port); executor = Executors.newFixedThreadPool(10); running = true; executor.submit(this::acceptConnections); logger.info("Started syslog source {} on port {}", name, port); } catch (IOException e) { throw new RuntimeException("Failed to start syslog source on port " + port, e); } } @Override public void stop() { running = false; if (serverSocket != null) { try { serverSocket.close(); } catch (IOException e) { logger.warn("Error closing syslog server socket", e); } } if (executor != null) { executor.shutdown(); } logger.info("Stopped syslog source: {}", name); } @Override public boolean isRunning() { return running; } @Override public void addListener(LogEventListener listener) { listeners.add(listener); } @Override public void removeListener(LogEventListener listener) { listeners.remove(listener); } private void acceptConnections() { while (running) { try { Socket clientSocket = serverSocket.accept(); executor.submit(() -> handleClient(clientSocket)); } catch (IOException e) { if (running) { logger.error("Error accepting syslog connection", e); } } } } private void handleClient(Socket clientSocket) { try (BufferedReader reader = new BufferedReader( new InputStreamReader(clientSocket.getInputStream()))) { String line; while ((line = reader.readLine()) != null && running) { LogEvent event = parseSyslogMessage(line); if (event != null) { notifyListeners(event); } } } catch (IOException e) { logger.debug("Syslog client disconnected", e); } finally { try { clientSocket.close(); } catch (IOException e) { logger.debug("Error closing client socket", e); } } } private LogEvent parseSyslogMessage(String message) { // Basic syslog parsing (RFC 3164) try { Pattern pattern = Pattern.compile("<(\\d+)>(\\w{3}\\s+\\d+\\s+\\d+:\\d+:\\d+)\\s+(\\S+)\\s+(.+)"); Matcher matcher = pattern.matcher(message); if (matcher.matches()) { int priority = Integer.parseInt(matcher.group(1)); String timestamp = matcher.group(2); String hostname = matcher.group(3); String content = matcher.group(4); LogLevel level = parseSyslogLevel(priority); return new LogEvent.Builder() .message(content) .source(name) .level(level) .field("syslog_priority", priority) .field("syslog_timestamp", timestamp) .field("syslog_hostname", hostname) .build(); } } catch (Exception e) { logger.debug("Failed to parse syslog message: {}", message, e); } // Fallback: treat as raw message return new LogEvent.Builder() .message(message) .source(name) .build(); } private LogLevel parseSyslogLevel(int priority) { int severity = priority & 0x07; switch (severity) { case 0: case 1: return LogLevel.FATAL; case 2: case 3: return LogLevel.ERROR; case 4: return LogLevel.WARN; case 5: return LogLevel.INFO; case 6: return LogLevel.DEBUG; case 7: return LogLevel.TRACE; default: return LogLevel.INFO; } } private void notifyListeners(LogEvent event) { for (LogEventListener listener : listeners) { try { listener.onEvent(event); } catch (Exception e) { logger.error("Error notifying syslog listener", e); } } } } @Component public class JournaldSource implements LogSource { // Implementation for systemd journal (would require native integration) // This is a simplified version @Override public String getName() { return "journald"; } @Override public void start() { // Start journald monitoring } @Override public void stop() { // Stop journald monitoring } @Override public boolean isRunning() { return false; } @Override public void addListener(LogEventListener listener) { // Add listener } @Override public void removeListener(LogEventListener listener) { // Remove listener } }

Transform Interface and Implementations

public interface LogTransform { String getName(); LogEvent transform(LogEvent event); void initialize(); void shutdown(); } @Component public class GrokParserTransform implements LogTransform { private static final Logger logger = LoggerFactory.getLogger(GrokParserTransform.class); private final String name; private final String pattern; private final Map<String, String> fieldMappings; private Pattern compiledPattern; public GrokParserTransform(String name, String pattern, Map<String, String> fieldMappings) { this.name = name; this.pattern = pattern; this.fieldMappings = fieldMappings; } @Override public String getName() { return name; } @Override public void initialize() { try { // Convert grok pattern to regex String regex = convertGrokToRegex(pattern); compiledPattern = Pattern.compile(regex); logger.info("Initialized Grok parser with pattern: {}", pattern); } catch (Exception e) { throw new RuntimeException("Failed to initialize Grok parser with pattern: " + pattern, e); } } @Override public void shutdown() { // Cleanup resources } @Override public LogEvent transform(LogEvent event) { if (compiledPattern == null) { logger.warn("Grok parser not initialized"); return event; } String message = event.getMessage(); if (message == null) { return event; } Matcher matcher = compiledPattern.matcher(message); if (matcher.matches()) { LogEvent.Builder builder = new LogEvent.Builder() .message(event.getMessage()) .timestamp(event.getTimestamp()) .level(event.getLevel()) .source(event.getSource()); // Copy existing fields event.getFields().forEach(builder::field); // Add parsed fields for (Map.Entry<String, String> mapping : fieldMappings.entrySet()) { try { String fieldName = mapping.getKey(); String groupName = mapping.getValue(); String value = matcher.group(groupName); if (value != null) { builder.field(fieldName, value); } } catch (Exception e) { logger.debug("Failed to extract group {} from pattern", mapping.getValue(), e); } } return builder.build(); } return event; } private String convertGrokToRegex(String grokPattern) { // Simplified Grok pattern conversion // In production, use a proper Grok library Map<String, String> patterns = new HashMap<>(); patterns.put("WORD", "\\b\\w+\\b"); patterns.put("NUMBER", "\\b\\d+\\b"); patterns.put("IP", "\\b(?:[0-9]{1,3}\\.){3}[0-9]{1,3}\\b"); patterns.put("TIMESTAMP_ISO8601", "\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}"); String regex = grokPattern; for (Map.Entry<String, String> entry : patterns.entrySet()) { regex = regex.replace("%{" + entry.getKey() + "}", "(" + entry.getValue() + ")"); } return regex; } } @Component public class AddFieldsTransform implements LogTransform { private final String name; private final Map<String, Object> additionalFields; public AddFieldsTransform(String name, Map<String, Object> additionalFields) { this.name = name; this.additionalFields = additionalFields; } @Override public String getName() { return name; } @Override public void initialize() { // No initialization needed } @Override public void shutdown() { // No cleanup needed } @Override public LogEvent transform(LogEvent event) { LogEvent.Builder builder = new LogEvent.Builder() .message(event.getMessage()) .timestamp(event.getTimestamp()) .level(event.getLevel()) .source(event.getSource()); // Copy existing fields event.getFields().forEach(builder::field); // Add additional fields additionalFields.forEach(builder::field); return builder.build(); } } @Component public class FilterTransform implements LogTransform { private final String name; private final Predicate<LogEvent> filter; public FilterTransform(String name, Predicate<LogEvent> filter) { this.name = name; this.filter = filter; } @Override public String getName() { return name; } @Override public void initialize() { // No initialization needed } @Override public void shutdown() { // No cleanup needed } @Override public LogEvent transform(LogEvent event) { return filter.test(event) ? event : null; } } @Component public class SamplingTransform implements LogTransform { private final String name; private final double sampleRate; private final Random random; public SamplingTransform(String name, double sampleRate) { this.name = name; this.sampleRate = sampleRate; this.random = new Random(); } @Override public String getName() { return name; } @Override public void initialize() { // No initialization needed } @Override public void shutdown() { // No cleanup needed } @Override public LogEvent transform(LogEvent event) { return random.nextDouble() <= sampleRate ? event : null; } }

Sink Interface and Implementations

public interface LogSink { String getName(); void start(); void stop(); boolean isRunning(); void emit(LogEvent event) throws IOException; void emitBatch(List<LogEvent> events) throws IOException; } @Component public class ElasticsearchSink implements LogSink { private static final Logger logger = LoggerFactory.getLogger(ElasticsearchSink.class); private final String name; private final String endpoint; private final String indexPattern; private final RestHighLevelClient client; private final BulkProcessor bulkProcessor; private volatile boolean running; public ElasticsearchSink(String name, String endpoint, String indexPattern, int bulkSize, Duration flushInterval) { this.name = name; this.endpoint = endpoint; this.indexPattern = indexPattern; RestClientBuilder restClientBuilder = RestClient.builder( HttpHost.create(endpoint) ); this.client = new RestHighLevelClient(restClientBuilder); this.bulkProcessor = BulkProcessor.builder( (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), new BulkProcessor.Listener() { @Override public void beforeBulk(long executionId, BulkRequest request) { logger.debug("Executing bulk request with {} actions", request.numberOfActions()); } @Override public void afterBulk(long executionId, BulkRequest request, BulkResponse response) { if (response.hasFailures()) { logger.warn("Bulk request had failures: {}", response.buildFailureMessage()); } else { logger.debug("Bulk request completed successfully"); } } @Override public void afterBulk(long executionId, BulkRequest request, Throwable failure) { logger.error("Bulk request failed", failure); } }) .setBulkActions(bulkSize) .setFlushInterval(flushInterval) .build(); } @Override public String getName() { return name; } @Override public void start() { running = true; logger.info("Started Elasticsearch sink: {} -> {}", name, endpoint); } @Override public void stop() { running = false; try { bulkProcessor.awaitClose(30, TimeUnit.SECONDS); client.close(); } catch (Exception e) { logger.warn("Error closing Elasticsearch client", e); } logger.info("Stopped Elasticsearch sink: {}", name); } @Override public boolean isRunning() { return running; } @Override public void emit(LogEvent event) throws IOException { if (!running) { throw new IllegalStateException("Sink is not running"); } IndexRequest request = createIndexRequest(event); bulkProcessor.add(request); } @Override public void emitBatch(List<LogEvent> events) throws IOException { if (!running) { throw new IllegalStateException("Sink is not running"); } for (LogEvent event : events) { IndexRequest request = createIndexRequest(event); bulkProcessor.add(request); } } private IndexRequest createIndexRequest(LogEvent event) { String indexName = resolveIndexName(event); Map<String, Object> source = new HashMap<>(); source.put("message", event.getMessage()); source.put("timestamp", event.getTimestamp()); source.put("level", event.getLevel().name()); source.put("source", event.getSource()); source.put("fields", event.getFields()); source.put("metadata", event.getMetadata()); return new IndexRequest(indexName) .source(source, XContentType.JSON) .id(event.getId()); } private String resolveIndexName(LogEvent event) { // Replace date patterns in index name String indexName = indexPattern; DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy.MM.dd"); String dateStr = event.getTimestamp().atZone(ZoneId.systemDefault()).format(formatter); indexName = indexName.replace("%{+yyyy.MM.dd}", dateStr); return indexName; } } @Component public class KafkaSink implements LogSink { private static final Logger logger = LoggerFactory.getLogger(KafkaSink.class); private final String name; private final String topic; private final KafkaProducer<String, String> producer; private volatile boolean running; public KafkaSink(String name, String bootstrapServers, String topic) { this.name = name; this.topic = topic; Properties props = new Properties(); props.put("bootstrap.servers", bootstrapServers); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("acks", "1"); props.put("retries", 3); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); this.producer = new KafkaProducer<>(props); } @Override public String getName() { return name; } @Override public void start() { running = true; logger.info("Started Kafka sink: {} -> topic: {}", name, topic); } @Override public void stop() { running = false; producer.close(); logger.info("Stopped Kafka sink: {}", name); } @Override public boolean isRunning() { return running; } @Override public void emit(LogEvent event) throws IOException { if (!running) { throw new IllegalStateException("Sink is not running"); } String key = event.getId(); String value = event.toJSON(); ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value); producer.send(record, (metadata, exception) -> { if (exception != null) { logger.error("Failed to send log event to Kafka", exception); } }); } @Override public void emitBatch(List<LogEvent> events) throws IOException { for (LogEvent event : events) { emit(event); } } } @Component public class FileSink implements LogSink { private static final Logger logger = LoggerFactory.getLogger(FileSink.class); private final String name; private final Path filePath; private final Charset charset; private final boolean append; private volatile boolean running; private BufferedWriter writer; public FileSink(String name, String filePath, Charset charset, boolean append) { this.name = name; this.filePath = Paths.get(filePath); this.charset = charset; this.append = append; } @Override public String getName() { return name; } @Override public void start() { try { Files.createDirectories(filePath.getParent()); writer = Files.newBufferedWriter(filePath, charset, append ? StandardOpenOption.CREATE : StandardOpenOption.CREATE, StandardOpenOption.WRITE, append ? StandardOpenOption.APPEND : StandardOpenOption.TRUNCATE_EXISTING); running = true; logger.info("Started file sink: {} -> {}", name, filePath); } catch (IOException e) { throw new RuntimeException("Failed to start file sink: " + filePath, e); } } @Override public void stop() { running = false; if (writer != null) { try { writer.close(); } catch (IOException e) { logger.warn("Error closing file sink writer", e); } } logger.info("Stopped file sink: {}", name); } @Override public boolean isRunning() { return running; } @Override public void emit(LogEvent event) throws IOException { if (!running) { throw new IllegalStateException("Sink is not running"); } String line = event.toJSON() + System.lineSeparator(); writer.write(line); writer.flush(); } @Override public void emitBatch(List<LogEvent> events) throws IOException { for (LogEvent event : events) { emit(event); } } } @Component public class ConsoleSink implements LogSink { private static final Logger logger = LoggerFactory.getLogger(ConsoleSink.class); private final String name; private final boolean prettyPrint; private volatile boolean running; public ConsoleSink(String name, boolean prettyPrint) { this.name = name; this.prettyPrint = prettyPrint; } @Override public String getName() { return name; } @Override public void start() { running = true; logger.info("Started console sink: {}", name); } @Override public void stop() { running = false; logger.info("Stopped console sink: {}", name); } @Override public boolean isRunning() { return running; } @Override public void emit(LogEvent event) throws IOException { if (!running) { throw new IllegalStateException("Sink is not running"); } if (prettyPrint) { ObjectMapper mapper = new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT); try { String json = mapper.writeValueAsString(convertToMap(event)); System.out.println(json); } catch (JsonProcessingException e) { System.out.println(event.toJSON()); } } else { System.out.println(event.toJSON()); } } @Override public void emitBatch(List<LogEvent> events) throws IOException { for (LogEvent event : events) { emit(event); } } private Map<String, Object> convertToMap(LogEvent event) { Map<String, Object> map = new HashMap<>(); map.put("id", event.getId()); map.put("timestamp", event.getTimestamp()); map.put("level", event.getLevel()); map.put("source", event.getSource()); map.put("message", event.getMessage()); map.put("fields", event.getFields()); return map; } }

Buffer Management

High-Performance Buffer

@Component public class BufferManager { private static final Logger logger = LoggerFactory.getLogger(BufferManager.class); private final int capacity; private final BlockingQueue<LogEvent> buffer; private final List<LogTransform> transforms; private final List<LogSink> sinks; private final MetricsCollector metrics; private final ExecutorService processor; private volatile boolean running; public BufferManager(int capacity, List<LogTransform> transforms, List<LogSink> sinks, MetricsCollector metrics) { this.capacity = capacity; this.buffer = new ArrayBlockingQueue<>(capacity); this.transforms = transforms; this.sinks = sinks; this.metrics = metrics; this.processor = Executors.newFixedThreadPool(Math.max(2, Runtime.getRuntime().availableProcessors())); } public void start() { running = true; // Start sink workers for (int i = 0; i < sinks.size(); i++) { processor.submit(this::sinkWorker); } logger.info("Started buffer manager with capacity: {}", capacity); } public void stop() { running = false; processor.shutdown(); try { if (!processor.awaitTermination(30, TimeUnit.SECONDS)) { processor.shutdownNow(); } } catch (InterruptedException e) { processor.shutdownNow(); Thread.currentThread().interrupt(); } logger.info("Stopped buffer manager"); } public boolean offer(LogEvent event) { boolean accepted = buffer.offer(event); if (!accepted) { metrics.recordBufferFull(); logger.warn("Buffer is full, dropping event: {}", event.getId()); } else { metrics.recordBufferOffer(); } return accepted; } public boolean offer(List<LogEvent> events) { boolean allAccepted = true; for (LogEvent event : events) { if (!offer(event)) { allAccepted = false; } } return allAccepted; } private void sinkWorker() { while (running) { try { LogEvent event = buffer.poll(1, TimeUnit.SECONDS); if (event != null) { processEvent(event); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { logger.error("Error in sink worker", e); } } } private void processEvent(LogEvent event) { try { // Apply transforms LogEvent transformed = applyTransforms(event); if (transformed == null) { metrics.recordEventFiltered(); return; } // Emit to all sinks for (LogSink sink : sinks) { if (sink.isRunning()) { try { sink.emit(transformed); metrics.recordEventEmitted(sink.getName()); } catch (Exception e) { logger.error("Failed to emit event to sink: {}", sink.getName(), e); metrics.recordEmitError(sink.getName()); } } } } catch (Exception e) { logger.error("Error processing event: {}", event.getId(), e); metrics.recordProcessingError(); } } private LogEvent applyTransforms(LogEvent event) { LogEvent current = event; for (LogTransform transform : transforms) { try { current = transform.transform(current); if (current == null) { return null; // Event was filtered out } } catch (Exception e) { logger.error("Error applying transform: {}", transform.getName(), e); metrics.recordTransformError(transform.getName()); } } return current; } public int getBufferSize() { return buffer.size(); } public int getRemainingCapacity() { return buffer.remainingCapacity(); } public double getBufferUtilization() { return (double) buffer.size() / capacity; } }

Metrics and Monitoring

Comprehensive Metrics Collection

@Component public class MetricsCollector { private final MeterRegistry meterRegistry; private final Map<String, Counter> eventCounters = new ConcurrentHashMap<>(); private final Map<String, Counter> errorCounters = new ConcurrentHashMap<>(); private final Map<String, Timer> processingTimers = new ConcurrentHashMap<>(); public MetricsCollector(MeterRegistry meterRegistry) { this.meterRegistry = meterRegistry; } public void recordEventReceived(String source) { Counter counter = eventCounters.computeIfAbsent(source, s -> Counter.builder("vector.events.received") .tag("source", s) .register(meterRegistry)); counter.increment(); } public void recordEventEmitted(String sink) { Counter counter = eventCounters.computeIfAbsent(sink, s -> Counter.builder("vector.events.emitted") .tag("sink", s) .register(meterRegistry)); counter.increment(); } public void recordEventFiltered() { Counter.builder("vector.events.filtered") .register(meterRegistry) .increment(); } public void recordBufferOffer() { Counter.builder("vector.buffer.offers") .register(meterRegistry) .increment(); } public void recordBufferFull() { Counter.builder("vector.buffer.full") .register(meterRegistry) .increment(); } public void recordProcessingError() { Counter.builder("vector.processing.errors") .register(meterRegistry) .increment(); } public void recordTransformError(String transform) { Counter counter = errorCounters.computeIfAbsent(transform, t -> Counter.builder("vector.transform.errors") .tag("transform", t) .register(meterRegistry)); counter.increment(); } public void recordEmitError(String sink) { Counter counter = errorCounters.computeIfAbsent(sink, s -> Counter.builder("vector.emit.errors") .tag("sink", s) .register(meterRegistry)); counter.increment(); } public void recordProcessingTime(String operation, long durationMs) { Timer timer = processingTimers.computeIfAbsent(operation, op -> Timer.builder("vector.processing.time") .tag("operation", op) .register(meterRegistry)); timer.record(durationMs, TimeUnit.MILLISECONDS); } public void registerBufferGauge(Supplier<Number> bufferSizeSupplier) { Gauge.builder("vector.buffer.size", bufferSizeSupplier) .register(meterRegistry); } public Map<String, Object> getMetricsSnapshot() { Map<String, Object> snapshot = new HashMap<>(); eventCounters.forEach((key, counter) -> { snapshot.put("events." + key, counter.count()); }); errorCounters.forEach((key, counter) -> { snapshot.put("errors." + key, counter.count()); }); return snapshot; } }

Vector Agent Core

Main Agent Implementation

@Component public class VectorAgent { private static final Logger logger = LoggerFactory.getLogger(VectorAgent.class); private final List<LogSource> sources; private final List<LogTransform> transforms; private final List<LogSink> sinks; private final BufferManager bufferManager; private final MetricsCollector metricsCollector; private final VectorConfig config; public VectorAgent(VectorConfig config) { this.config = config; this.metricsCollector = new MetricsCollector(config.getMeterRegistry()); this.sources = initializeSources(config); this.transforms = initializeTransforms(config); this.sinks = initializeSinks(config); this.bufferManager = new BufferManager( config.getBufferCapacity(), transforms, sinks, metricsCollector); setupSourceListeners(); } public void start() { logger.info("Starting Vector agent..."); // Initialize transforms transforms.forEach(LogTransform::initialize); // Start sinks sinks.forEach(LogSink::start); // Start buffer manager bufferManager.start(); // Start sources sources.forEach(LogSource::start); // Register metrics metricsCollector.registerBufferGauge(bufferManager::getBufferSize); logger.info("Vector agent started successfully"); } public void stop() { logger.info("Stopping Vector agent..."); // Stop sources sources.forEach(LogSource::stop); // Stop buffer manager bufferManager.stop(); // Stop sinks sinks.forEach(LogSink::stop); // Shutdown transforms transforms.forEach(LogTransform::shutdown); logger.info("Vector agent stopped"); } private List<LogSource> initializeSources(VectorConfig config) { List<LogSource> sources = new ArrayList<>(); // File sources config.getFileSources().forEach((name, sourceConfig) -> { FileLogSource source = new FileLogSource( name, sourceConfig.getPath(), Charset.forName(sourceConfig.getCharset()) ); sources.add(source); }); // Syslog sources config.getSyslogSources().forEach((name, sourceConfig) -> { SyslogSource source = new SyslogSource(name, sourceConfig.getPort()); sources.add(source); }); return sources; } private List<LogTransform> initializeTransforms(VectorConfig config) { List<LogTransform> transforms = new ArrayList<>(); // Grok parsers config.getGrokTransforms().forEach((name, transformConfig) -> { GrokParserTransform transform = new GrokParserTransform( name, transformConfig.getPattern(), transformConfig.getFieldMappings() ); transforms.add(transform); }); // Add fields transforms config.getAddFieldsTransforms().forEach((name, transformConfig) -> { AddFieldsTransform transform = new AddFieldsTransform( name, transformConfig.getAdditionalFields() ); transforms.add(transform); }); // Filter transforms config.getFilterTransforms().forEach((name, transformConfig) -> { FilterTransform transform = new FilterTransform( name, transformConfig.getFilter() ); transforms.add(transform); }); return transforms; } private List<LogSink> initializeSinks(VectorConfig config) { List<LogSink> sinks = new ArrayList<>(); // Elasticsearch sinks config.getElasticsearchSinks().forEach((name, sinkConfig) -> { ElasticsearchSink sink = new ElasticsearchSink( name, sinkConfig.getEndpoint(), sinkConfig.getIndexPattern(), sinkConfig.getBulkSize(), sinkConfig.getFlushInterval() ); sinks.add(sink); }); // Kafka sinks config.getKafkaSinks().forEach((name, sinkConfig) -> { KafkaSink sink = new KafkaSink( name, sinkConfig.getBootstrapServers(), sinkConfig.getTopic() ); sinks.add(sink); }); // File sinks config.getFileSinks().forEach((name, sinkConfig) -> { FileSink sink = new FileSink( name, sinkConfig.getPath(), Charset.forName(sinkConfig.getCharset()), sinkConfig.isAppend() ); sinks.add(sink); }); // Console sink (for debugging) if (config.isEnableConsoleSink()) { ConsoleSink consoleSink = new ConsoleSink("console", true); sinks.add(consoleSink); } return sinks; } private void setupSourceListeners() { for (LogSource source : sources) { source.addListener(event -> { metricsCollector.recordEventReceived(source.getName()); bufferManager.offer(event); }); } } public Map<String, Object> getStatus() { Map<String, Object> status = new HashMap<>(); status.put("sources", sources.stream() .collect(Collectors.toMap(LogSource::getName, LogSource::isRunning))); status.put("sinks", sinks.stream() .collect(Collectors.toMap(LogSink::getName, LogSink::isRunning))); status.put("buffer", Map.of( "size", bufferManager.getBufferSize(), "capacity", bufferManager.getRemainingCapacity(), "utilization", bufferManager.getBufferUtilization() )); status.put("metrics", metricsCollector.getMetricsSnapshot()); return status; } }

Configuration Management

YAML Configuration

@ConfigurationProperties(prefix = "vector") @Data public class VectorConfig { private int bufferCapacity = 10000; private boolean enableConsoleSink = false; private MeterRegistry meterRegistry; private Map<String, FileSourceConfig> fileSources = new HashMap<>(); private Map<String, SyslogSourceConfig> syslogSources = new HashMap<>(); private Map<String, GrokTransformConfig> grokTransforms = new HashMap<>(); private Map<String, AddFieldsTransformConfig> addFieldsTransforms = new HashMap<>(); private Map<String, FilterTransformConfig> filterTransforms = new HashMap<>(); private Map<String, ElasticsearchSinkConfig> elasticsearchSinks = new HashMap<>(); private Map<String, KafkaSinkConfig> kafkaSinks = new HashMap<>(); private Map<String, FileSinkConfig> fileSinks = new HashMap<>(); @Data public static class FileSourceConfig { private String path; private String charset = "UTF-8"; private boolean follow = true; } @Data public static class SyslogSourceConfig { private int port = 514; private String protocol = "tcp"; } @Data public static class GrokTransformConfig { private String pattern; private Map<String, String> fieldMappings = new HashMap<>(); } @Data public static class AddFieldsTransformConfig { private Map<String, Object> additionalFields = new HashMap<>(); } @Data public static class FilterTransformConfig { private String condition; // Would be parsed into Predicate<LogEvent> private Predicate<LogEvent> getFilter() { // Parse condition string into predicate return event -> true; // Simplified } } @Data public static class ElasticsearchSinkConfig { private String endpoint; private String indexPattern = "logs-%{+yyyy.MM.dd}"; private int bulkSize = 1000; private Duration flushInterval = Duration.ofSeconds(1); } @Data public static class KafkaSinkConfig { private String bootstrapServers; private String topic; } @Data public static class FileSinkConfig { private String path; private String charset = "UTF-8"; private boolean append = true; } } @Configuration @EnableConfigurationProperties(VectorConfig.class) public class VectorConfiguration { @Bean @ConditionalOnProperty(name = "vector.enabled", havingValue = "true") public VectorAgent vectorAgent(VectorConfig config) { return new VectorAgent(config); } @Bean public MeterRegistry meterRegistry() { return new SimpleMeterRegistry(); } }

REST API for Management

Management Endpoints

@RestController @RequestMapping("/api/vector") public class VectorManagementController { private final VectorAgent vectorAgent; public VectorManagementController(VectorAgent vectorAgent) { this.vectorAgent = vectorAgent; } @GetMapping("/status") public Map<String, Object> getStatus() { return vectorAgent.getStatus(); } @PostMapping("/start") public ResponseEntity<String> startAgent() { vectorAgent.start(); return ResponseEntity.ok("Vector agent started"); } @PostMapping("/stop") public ResponseEntity<String> stopAgent() { vectorAgent.stop(); return ResponseEntity.ok("Vector agent stopped"); } @GetMapping("/health") public ResponseEntity<Map<String, Object>> health() { Map<String, Object> status = vectorAgent.getStatus(); boolean healthy = (boolean) ((Map<?, ?>) status.get("buffer")).get("utilization") < 0.9; Map<String, Object> health = new HashMap<>(); health.put("status", healthy ? "HEALTHY" : "UNHEALTHY"); health.put("timestamp", Instant.now()); health.put("details", status); return healthy ? ResponseEntity.ok(health) : ResponseEntity.status(503).body(health); } @GetMapping("/metrics") public Map<String, Object> getMetrics() { return vectorAgent.getStatus().get("metrics"); } }

Testing Framework

Comprehensive Test Suite

@SpringBootTest @TestPropertySource(properties = { "vector.enabled=true", "vector.buffer-capacity=1000", "vector.enable-console-sink=true" }) public class VectorAgentTest { @Autowired private VectorAgent vectorAgent; @TempDir Path tempDir; @Test void testFileSourceToConsoleSink() throws Exception { // Create test log file Path logFile = tempDir.resolve("test.log"); Files.write(logFile, "Test log message\n".getBytes()); vectorAgent.start(); // Add more log entries Files.write(logFile, "Another log message\n".getBytes(), StandardOpenOption.APPEND); // Wait for processing Thread.sleep(2000); vectorAgent.stop(); // Verify metrics show events processed Map<String, Object> status = vectorAgent.getStatus(); assertNotNull(status); } @Test void testBufferBackpressure() throws Exception { vectorAgent.start(); // Create many events quickly to fill buffer for (int i = 0; i < 1500; i++) { LogEvent event = new LogEvent.Builder() .message("Test message " + i) .source("test") .build(); // This would normally come from sources } // Verify buffer utilization is high Map<String, Object> status = vectorAgent.getStatus(); Map<?, ?> bufferStatus = (Map<?, ?>) status.get("buffer"); double utilization = (Double) bufferStatus.get("utilization"); assertTrue(utilization > 0.5); vectorAgent.stop(); } @Test void testTransformPipeline() { // Test Grok parsing GrokParserTransform grok = new GrokParserTransform( "test-grok", "%{WORD:method} %{NUMBER:status}", Map.of("http_method", "method", "http_status", "status") ); grok.initialize(); LogEvent event = new LogEvent.Builder() .message("GET 200") .source("test") .build(); LogEvent transformed = grok.transform(event); assertNotNull(transformed); assertEquals("GET", transformed.getFields().get("http_method")); assertEquals("200", transformed.getFields().get("http_status")); } } public class PerformanceTest { @Test void testHighThroughput() throws Exception { VectorConfig config = createHighPerfConfig(); VectorAgent agent = new VectorAgent(config); agent.start(); long startTime = System.currentTimeMillis(); long eventCount = 100000; // Simulate high event rate // This would normally come from sources long duration = System.currentTimeMillis() - startTime; double eventsPerSecond = eventCount / (duration / 1000.0); System.out.printf("Processed %d events in %d ms (%.2f events/sec)%n", eventCount, duration, eventsPerSecond); assertTrue(eventsPerSecond > 1000, "Should process at least 1000 events/sec"); agent.stop(); } private VectorConfig createHighPerfConfig() { VectorConfig config = new VectorConfig(); config.setBufferCapacity(50000); // Add high-performance sources and sinks return config; } }

Conclusion

This Vector Log Agent implementation provides:

  1. Multiple Source Types - File, Syslog, Journald, etc.
  2. Flexible Transforms - Grok parsing, field addition, filtering, sampling
  3. Various Sink Destinations - Elasticsearch, Kafka, files, console
  4. High-Performance Buffer - Backpressure management and batching
  5. Comprehensive Metrics - Monitoring and observability
  6. Configuration Management - YAML-based configuration
  7. REST API - Management and monitoring endpoints
  8. Testing Framework - Unit and performance tests

The agent is designed for high-throughput log processing with minimal resource overhead, making it suitable for production deployment in observability pipelines.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper