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:
- Multiple Source Types - File, Syslog, Journald, etc.
- Flexible Transforms - Grok parsing, field addition, filtering, sampling
- Various Sink Destinations - Elasticsearch, Kafka, files, console
- High-Performance Buffer - Backpressure management and batching
- Comprehensive Metrics - Monitoring and observability
- Configuration Management - YAML-based configuration
- REST API - Management and monitoring endpoints
- 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.