Elasticsearch Log Indexing in Java: Complete Implementation Guide

A comprehensive guide to implementing Elasticsearch log indexing in Java applications with advanced features like bulk processing, error handling, and performance optimization.

Project Overview

This implementation provides a robust Elasticsearch logging solution that can handle high-volume log data with efficient indexing, search capabilities, and real-time analytics.

Architecture

graph TB A[Java Application] --> B[Log Collector] B --> C[Log Processor] C --> D[Elasticsearch Client] D --> E[Elasticsearch Cluster] F[Monitoring] --> D G[Kibana] --> E subgraph "Elasticsearch Cluster" E --> E1[Node 1] E --> E2[Node 2] E --> E3[Node 3] end

Dependencies

Maven Configuration

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"> <modelVersion>4.0.0</modelVersion> <groupId>com.elasticsearch</groupId> <artifactId>log-indexer</artifactId> <version>1.0.0</version> <properties> <maven.compiler.source>11</maven.compiler.source> <maven.compiler.target>11</maven.compiler.target> <elasticsearch.version>8.9.0</elasticsearch.version> <spring-boot.version>3.1.0</spring-boot.version> <log4j.version>2.20.0</log4j.version> </properties> <dependencies> <!-- Elasticsearch Java Client --> <dependency> <groupId>co.elastic.clients</groupId> <artifactId>elasticsearch-java</artifactId> <version>${elasticsearch.version}</version> </dependency> <!-- Jackson for JSON --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.15.0</version> </dependency> <!-- Spring Boot --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> <version>${spring-boot.version}</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> <version>${spring-boot.version}</version> </dependency> <!-- Logging --> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>${log4j.version}</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-api</artifactId> <version>${log4j.version}</version> </dependency> <!-- Utilities --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.12.0</version> </dependency> <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>31.1-jre</version> </dependency> <!-- Testing --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <version>${spring-boot.version}</version> <scope>test</scope> </dependency> </dependencies> </project>

Core Domain Models

package com.elasticsearch.logging.model; import com.fasterxml.jackson.annotation.JsonFormat; import com.fasterxml.jackson.annotation.JsonInclude; import java.time.Instant; import java.util.Map; import java.util.UUID; @JsonInclude(JsonInclude.Include.NON_NULL) public class LogEntry { private String id; private String traceId; private String spanId; @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timezone = "UTC") private Instant timestamp; private String level; // DEBUG, INFO, WARN, ERROR, FATAL private String logger; private String message; private String thread; private String application; private String environment; private String hostname; private String serviceName; private String serviceVersion; private Map<String, Object> context; private Map<String, Object> metadata; private ExceptionInfo exception; // Constructors public LogEntry() { this.id = UUID.randomUUID().toString(); this.timestamp = Instant.now(); } public LogEntry(String level, String logger, String message) { this(); this.level = level; this.logger = logger; this.message = message; } // Builder pattern for fluent creation public static Builder builder() { return new Builder(); } public static class Builder { private final LogEntry logEntry; public Builder() { this.logEntry = new LogEntry(); } public Builder level(String level) { logEntry.level = level; return this; } public Builder logger(String logger) { logEntry.logger = logger; return this; } public Builder message(String message) { logEntry.message = message; return this; } public Builder traceId(String traceId) { logEntry.traceId = traceId; return this; } public Builder spanId(String spanId) { logEntry.spanId = spanId; return this; } public Builder application(String application) { logEntry.application = application; return this; } public Builder environment(String environment) { logEntry.environment = environment; return this; } public Builder serviceName(String serviceName) { logEntry.serviceName = serviceName; return this; } public Builder context(Map<String, Object> context) { logEntry.context = context; return this; } public Builder addContext(String key, Object value) { if (logEntry.context == null) { logEntry.context = new java.util.HashMap<>(); } logEntry.context.put(key, value); return this; } public Builder exception(Throwable throwable) { if (throwable != null) { logEntry.exception = new ExceptionInfo(throwable); } return this; } public LogEntry build() { return logEntry; } } // Getters and setters public String getId() { return id; } public void setId(String id) { this.id = id; } public String getTraceId() { return traceId; } public void setTraceId(String traceId) { this.traceId = traceId; } public String getSpanId() { return spanId; } public void setSpanId(String spanId) { this.spanId = spanId; } public Instant getTimestamp() { return timestamp; } public void setTimestamp(Instant timestamp) { this.timestamp = timestamp; } public String getLevel() { return level; } public void setLevel(String level) { this.level = level; } public String getLogger() { return logger; } public void setLogger(String logger) { this.logger = logger; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } public String getThread() { return thread; } public void setThread(String thread) { this.thread = thread; } public String getApplication() { return application; } public void setApplication(String application) { this.application = application; } public String getEnvironment() { return environment; } public void setEnvironment(String environment) { this.environment = environment; } public String getHostname() { return hostname; } public void setHostname(String hostname) { this.hostname = hostname; } public String getServiceName() { return serviceName; } public void setServiceName(String serviceName) { this.serviceName = serviceName; } public String getServiceVersion() { return serviceVersion; } public void setServiceVersion(String serviceVersion) { this.serviceVersion = serviceVersion; } public Map<String, Object> getContext() { return context; } public void setContext(Map<String, Object> context) { this.context = context; } public Map<String, Object> getMetadata() { return metadata; } public void setMetadata(Map<String, Object> metadata) { this.metadata = metadata; } public ExceptionInfo getException() { return exception; } public void setException(ExceptionInfo exception) { this.exception = exception; } } @JsonInclude(JsonInclude.Include.NON_NULL) class ExceptionInfo { private String type; private String message; private String stackTrace; private String cause; public ExceptionInfo() {} public ExceptionInfo(Throwable throwable) { this.type = throwable.getClass().getName(); this.message = throwable.getMessage(); this.stackTrace = getStackTraceAsString(throwable); if (throwable.getCause() != null) { this.cause = throwable.getCause().getMessage(); } } private String getStackTraceAsString(Throwable throwable) { java.io.StringWriter sw = new java.io.StringWriter(); java.io.PrintWriter pw = new java.io.PrintWriter(sw); throwable.printStackTrace(pw); return sw.toString(); } // Getters and setters public String getType() { return type; } public void setType(String type) { this.type = type; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } public String getStackTrace() { return stackTrace; } public void setStackTrace(String stackTrace) { this.stackTrace = stackTrace; } public String getCause() { return cause; } public void setCause(String cause) { this.cause = cause; } } // Bulk operation wrapper class BulkOperation<T> { private final String index; private final T document; private final String operation; // index, create, update, delete public BulkOperation(String index, T document, String operation) { this.index = index; this.document = document; this.operation = operation; } // Getters public String getIndex() { return index; } public T getDocument() { return document; } public String getOperation() { return operation; } }

Elasticsearch Configuration

package com.elasticsearch.logging.config; import co.elastic.clients.elasticsearch.ElasticsearchClient; import co.elastic.clients.json.jackson.JacksonJsonpMapper; import co.elastic.clients.transport.ElasticsearchTransport; import co.elastic.clients.transport.rest_client.RestClientTransport; import org.apache.http.Header; import org.apache.http.HttpHost; import org.apache.http.auth.AuthScope; import org.apache.http.auth.UsernamePasswordCredentials; import org.apache.http.client.CredentialsProvider; import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.message.BasicHeader; import org.elasticsearch.client.RestClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; import java.io.FileInputStream; import java.security.KeyStore; import java.security.cert.CertificateFactory; import java.security.cert.X509Certificate; @Configuration public class ElasticsearchConfig { @Value("${elasticsearch.host:localhost}") private String host; @Value("${elasticsearch.port:9200}") private int port; @Value("${elasticsearch.scheme:http}") private String scheme; @Value("${elasticsearch.username:}") private String username; @Value("${elasticsearch.password:}") private String password; @Value("${elasticsearch.api-key:}") private String apiKey; @Value("${elasticsearch.ssl.truststore.path:}") private String truststorePath; @Value("${elasticsearch.connection.timeout:30000}") private int connectionTimeout; @Value("${elasticsearch.socket.timeout:60000}") private int socketTimeout; @Bean public RestClient restClient() throws Exception { RestClient.Builder builder = RestClient.builder( new HttpHost(host, port, scheme) ); // Configure authentication if (!username.isEmpty() && !password.isEmpty()) { CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCredentials( AuthScope.ANY, new UsernamePasswordCredentials(username, password) ); builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider) ); } // Configure API Key authentication if (!apiKey.isEmpty()) { builder.setDefaultHeaders(new Header[]{ new BasicHeader("Authorization", "ApiKey " + apiKey) }); } // Configure SSL if needed if (scheme.equals("https") && !truststorePath.isEmpty()) { builder.setHttpClientConfigCallback(httpClientBuilder -> { try { SSLContext sslContext = createSSLContext(); httpClientBuilder.setSSLContext(sslContext); } catch (Exception e) { throw new RuntimeException("Failed to configure SSL context", e); } return httpClientBuilder; }); } // Configure timeouts builder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder .setConnectTimeout(connectionTimeout) .setSocketTimeout(socketTimeout) ); return builder.build(); } @Bean public ElasticsearchTransport elasticsearchTransport(RestClient restClient) { return new RestClientTransport(restClient, new JacksonJsonpMapper()); } @Bean public ElasticsearchClient elasticsearchClient(ElasticsearchTransport transport) { return new ElasticsearchClient(transport); } private SSLContext createSSLContext() throws Exception { CertificateFactory factory = CertificateFactory.getInstance("X.509"); X509Certificate certificate = (X509Certificate) factory.generateCertificate( new FileInputStream(truststorePath) ); KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType()); keyStore.load(null, null); keyStore.setCertificateEntry("elasticsearch", certificate); TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance( TrustManagerFactory.getDefaultAlgorithm() ); trustManagerFactory.init(keyStore); SSLContext sslContext = SSLContext.getInstance("TLS"); sslContext.init(null, trustManagerFactory.getTrustManagers(), null); return sslContext; } }

Elasticsearch Service Implementation

package com.elasticsearch.logging.service; import co.elastic.clients.elasticsearch.ElasticsearchClient; import co.elastic.clients.elasticsearch.core.*; import co.elastic.clients.elasticsearch.core.bulk.BulkOperation; import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; import co.elastic.clients.elasticsearch.indices.*; import co.elastic.clients.elasticsearch.indices.get_mapping.IndexMappingRecord; import co.elastic.clients.transport.endpoints.BooleanResponse; import com.elasticsearch.logging.model.LogEntry; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import java.io.IOException; import java.time.Instant; import java.time.LocalDate; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicLong; @Service public class ElasticsearchLogService { private static final Logger logger = LoggerFactory.getLogger(ElasticsearchLogService.class); private static final DateTimeFormatter INDEX_DATE_FORMAT = DateTimeFormatter.ofPattern("yyyy.MM.dd"); private static final String LOG_INDEX_PREFIX = "logs-"; private static final int BULK_BATCH_SIZE = 1000; private static final int BULK_FLUSH_INTERVAL_MS = 5000; private final ElasticsearchClient client; private final BlockingQueue<LogEntry> logQueue; private final AtomicLong processedCount; private final AtomicLong errorCount; private final Map<String, Object> indexSettings; @Value("${elasticsearch.index.prefix:logs}") private String indexPrefix; @Value("${elasticsearch.index.shards:3}") private int numberOfShards; @Value("${elasticsearch.index.replicas:1}") private int numberOfReplicas; @Value("${elasticsearch.bulk.batch-size:1000}") private int bulkBatchSize; @Value("${elasticsearch.bulk.flush-interval:5000}") private long bulkFlushInterval; @Autowired public ElasticsearchLogService(ElasticsearchClient client) { this.client = client; this.logQueue = new LinkedBlockingQueue<>(); this.processedCount = new AtomicLong(0); this.errorCount = new AtomicLong(0); this.indexSettings = createIndexSettings(); // Initialize indices and start background processor initializeIndices(); startBackgroundProcessor(); } private Map<String, Object> createIndexSettings() { Map<String, Object> settings = new HashMap<>(); Map<String, Object> indexSettings = new HashMap<>(); indexSettings.put("number_of_shards", numberOfShards); indexSettings.put("number_of_replicas", numberOfReplicas); // Analysis settings for better search Map<String, Object> analysis = new HashMap<>(); Map<String, Object> analyzer = new HashMap<>(); Map<String, Object> logMessageAnalyzer = new HashMap<>(); logMessageAnalyzer.put("type", "custom"); logMessageAnalyzer.put("tokenizer", "standard"); logMessageAnalyzer.put("filter", Arrays.asList("lowercase", "asciifolding")); analyzer.put("log_message_analyzer", logMessageAnalyzer); analysis.put("analyzer", analyzer); indexSettings.put("analysis", analysis); settings.put("index", indexSettings); return settings; } private void initializeIndices() { try { // Check if today's index exists, create if not String todayIndex = getIndexNameForDate(LocalDate.now()); createIndexIfNotExists(todayIndex); logger.info("Elasticsearch log service initialized"); } catch (Exception e) { logger.error("Failed to initialize Elasticsearch indices", e); } } private void startBackgroundProcessor() { Thread processorThread = new Thread(this::processQueue, "LogQueueProcessor"); processorThread.setDaemon(true); processorThread.start(); } // Public API Methods public void indexLog(LogEntry logEntry) { try { logQueue.put(logEntry); } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Interrupted while adding log to queue", e); } } @Async public void indexLogAsync(LogEntry logEntry) { indexLog(logEntry); } public void indexLogsBulk(List<LogEntry> logEntries) { if (logEntries.isEmpty()) { return; } String indexName = getIndexNameForDate(LocalDate.now()); List<BulkOperation> operations = new ArrayList<>(); for (LogEntry logEntry : logEntries) { operations.add(BulkOperation.of(op -> op .index(idx -> idx .index(indexName) .id(logEntry.getId()) .document(logEntry) ) )); } executeBulkOperations(operations); } public SearchResponse<LogEntry> searchLogs(String query, int from, int size) throws IOException { String indexName = getIndexPatternForCurrentMonth(); return client.search(s -> s .index(indexName) .from(from) .size(size) .query(q -> q .bool(b -> b .should(sh -> sh .match(m -> m .field("message") .query(query) ) ) .should(sh -> sh .wildcard(w -> w .field("message") .value("*" + query + "*") ) ) ) ) .sort(so -> so .field(f -> f .field("timestamp") .order(co.elastic.clients.elasticsearch._types.SortOrder.Desc) ) ), LogEntry.class ); } public SearchResponse<LogEntry> searchLogsByLevel(String level, Instant from, Instant to, int size) throws IOException { String indexPattern = getIndexPatternForDateRange(from, to); return client.search(s -> s .index(indexPattern) .size(size) .query(q -> q .bool(b -> b .must(m -> m .term(t -> t .field("level") .value(level) ) ) .must(m -> m .range(r -> r .field("timestamp") .gte(co.elastic.clients.elasticsearch._types.query_dsl.RangeQuery._DESERIALIZER) .lte(co.elastic.clients.elasticsearch._types.query_dsl.RangeQuery._DESERIALIZER) ) ) ) ) .sort(so -> so .field(f -> f .field("timestamp") .order(co.elastic.clients.elasticsearch._types.SortOrder.Desc) ) ), LogEntry.class ); } public Map<String, Object> getClusterHealth() throws IOException { return client.cluster().health(h -> h).toJson().asMap(); } public Map<String, Object> getIndexStats(String indexPattern) throws IOException { return client.indices().stats(s -> s.index(indexPattern)).toJson().asMap(); } // Background Processing private void processQueue() { List<LogEntry> batch = new ArrayList<>(bulkBatchSize); while (!Thread.currentThread().isInterrupted()) { try { // Wait for the first element LogEntry logEntry = logQueue.take(); batch.add(logEntry); // Drain remaining elements up to batch size logQueue.drainTo(batch, bulkBatchSize - 1); // Process the batch processBatch(batch); batch.clear(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { logger.error("Error processing log batch", e); errorCount.incrementAndGet(); } } } @Scheduled(fixedRateString = "${elasticsearch.bulk.flush-interval:5000}") public void flushQueue() { List<LogEntry> batch = new ArrayList<>(bulkBatchSize); logQueue.drainTo(batch, bulkBatchSize); if (!batch.isEmpty()) { processBatch(batch); } } private void processBatch(List<LogEntry> batch) { if (batch.isEmpty()) { return; } // Group by date for index routing Map<String, List<LogEntry>> logsByDate = new HashMap<>(); for (LogEntry logEntry : batch) { String indexName = getIndexNameForDate( LocalDate.from(logEntry.getTimestamp()) ); logsByDate.computeIfAbsent(indexName, k -> new ArrayList<>()).add(logEntry); } // Process each date group for (Map.Entry<String, List<LogEntry>> entry : logsByDate.entrySet()) { String indexName = entry.getKey(); List<LogEntry> dateBatch = entry.getValue(); try { createIndexIfNotExists(indexName); indexBatchToIndex(indexName, dateBatch); processedCount.addAndGet(dateBatch.size()); } catch (Exception e) { logger.error("Failed to index batch to index: " + indexName, e); errorCount.addAndGet(dateBatch.size()); // Fallback: try individual indexing for (LogEntry logEntry : dateBatch) { try { indexSingleLog(logEntry); } catch (Exception ex) { logger.error("Failed to index individual log: " + logEntry.getId(), ex); } } } } } private void indexBatchToIndex(String indexName, List<LogEntry> logEntries) { List<BulkOperation> operations = new ArrayList<>(); for (LogEntry logEntry : logEntries) { operations.add(BulkOperation.of(op -> op .index(idx -> idx .index(indexName) .id(logEntry.getId()) .document(logEntry) ) )); } executeBulkOperations(operations); } private void executeBulkOperations(List<BulkOperation> operations) { if (operations.isEmpty()) { return; } try { BulkResponse response = client.bulk(b -> b .operations(operations) .refresh(co.elastic.clients.elasticsearch._types.Refresh.WaitFor) ); // Check for errors if (response.errors()) { for (BulkResponseItem item : response.items()) { if (item.error() != null) { logger.error("Bulk indexing error for {}: {}", item.id(), item.error().reason()); } } } } catch (IOException e) { throw new RuntimeException("Bulk indexing failed", e); } } private void indexSingleLog(LogEntry logEntry) throws IOException { String indexName = getIndexNameForDate(LocalDate.from(logEntry.getTimestamp())); IndexResponse response = client.index(i -> i .index(indexName) .id(logEntry.getId()) .document(logEntry) .refresh(co.elastic.clients.elasticsearch._types.Refresh.WaitFor) ); if (!"created".equals(response.result().jsonValue())) { logger.warn("Unexpected index result: {}", response.result().jsonValue()); } } // Index Management private void createIndexIfNotExists(String indexName) throws IOException { BooleanResponse exists = client.indices().exists(e -> e.index(indexName)); if (!exists.value()) { CreateIndexResponse response = client.indices().create(c -> c .index(indexName) .settings(s -> s .numberOfShards(String.valueOf(numberOfShards)) .numberOfReplicas(String.valueOf(numberOfReplicas)) ) .mappings(m -> m .properties("timestamp", p -> p .date(d -> d.format("strict_date_optional_time||epoch_millis")) ) .properties("level", p -> p .keyword(k -> k) ) .properties("message", p -> p .text(t -> t .analyzer("log_message_analyzer") .fields("keyword", f -> f.keyword(k -> k)) ) ) .properties("serviceName", p -> p .keyword(k -> k) ) .properties("application", p -> p .keyword(k -> k) ) .properties("environment", p -> p .keyword(k -> k) ) ) ); if (response.acknowledged()) { logger.info("Created index: {}", indexName); } else { logger.warn("Index creation not acknowledged for: {}", indexName); } } } // Index naming utilities private String getIndexNameForDate(LocalDate date) { return indexPrefix + "-" + date.format(INDEX_DATE_FORMAT); } private String getIndexPatternForCurrentMonth() { LocalDate now = LocalDate.now(); return indexPrefix + "-" + now.getYear() + "." + String.format("%02d", now.getMonthValue()) + ".*"; } private String getIndexPatternForDateRange(Instant from, Instant to) { LocalDate fromDate = LocalDate.from(from); LocalDate toDate = LocalDate.from(to); if (fromDate.equals(toDate)) { return getIndexNameForDate(fromDate); } return indexPrefix + "-*"; } // Monitoring and Statistics public Map<String, Object> getStatistics() { Map<String, Object> stats = new HashMap<>(); stats.put("queueSize", logQueue.size()); stats.put("processedCount", processedCount.get()); stats.put("errorCount", errorCount.get()); stats.put("throughput", calculateThroughput()); return stats; } private double calculateThroughput() { // Implementation for throughput calculation return 0.0; } // Cleanup old indices @Scheduled(cron = "0 0 2 * * ?") // Daily at 2 AM public void cleanupOldIndices() { try { LocalDate retentionDate = LocalDate.now().minusDays(30); // 30 days retention String[] allIndices = client.indices().get(g -> g.index(indexPrefix + "-*")).result().keySet().toArray(new String[0]); for (String index : allIndices) { LocalDate indexDate = extractDateFromIndexName(index); if (indexDate != null && indexDate.isBefore(retentionDate)) { client.indices().delete(d -> d.index(index)); logger.info("Deleted old index: {}", index); } } } catch (Exception e) { logger.error("Error during index cleanup", e); } } private LocalDate extractDateFromIndexName(String indexName) { try { String datePart = indexName.substring(indexPrefix.length() + 1); return LocalDate.parse(datePart, INDEX_DATE_FORMAT); } catch (Exception e) { return null; } } }

Log4j2 Appender for Elasticsearch

package com.elasticsearch.logging.appender; import com.elasticsearch.logging.model.LogEntry; import com.elasticsearch.logging.service.ElasticsearchLogService; import org.apache.logging.log4j.core.*; import org.apache.logging.log4j.core.appender.AbstractAppender; import org.apache.logging.log4j.core.config.Property; import org.apache.logging.log4j.core.config.plugins.*; import org.apache.logging.log4j.core.layout.PatternLayout; import java.io.Serializable; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @Plugin(name = "ElasticsearchAppender", category = Core.CATEGORY_NAME, elementType = Appender.ELEMENT_TYPE) public class ElasticsearchAppender extends AbstractAppender { private static ElasticsearchLogService logService; private static final Map<String, String> CONTEXT_MAP = new ConcurrentHashMap<>(); private ElasticsearchAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions) { super(name, filter, layout, ignoreExceptions, Property.EMPTY_ARRAY); } @PluginFactory public static ElasticsearchAppender createAppender( @PluginAttribute("name") String name, @PluginElement("Filter") Filter filter, @PluginElement("Layout") Layout<? extends Serializable> layout, @PluginAttribute("ignoreExceptions") boolean ignoreExceptions) { if (name == null) { LOGGER.error("No name provided for ElasticsearchAppender"); return null; } if (layout == null) { layout = PatternLayout.createDefaultLayout(); } return new ElasticsearchAppender(name, filter, layout, ignoreExceptions); } public static void setLogService(ElasticsearchLogService service) { logService = service; } public static void addContext(String key, String value) { CONTEXT_MAP.put(key, value); } public static void removeContext(String key) { CONTEXT_MAP.remove(key); } @Override public void append(LogEvent event) { if (logService == null) { LOGGER.warn("ElasticsearchLogService not initialized for ElasticsearchAppender"); return; } try { LogEntry logEntry = convertToLogEntry(event); logService.indexLogAsync(logEntry); } catch (Exception e) { LOGGER.error("Failed to send log to Elasticsearch", e); } } private LogEntry convertToLogEntry(LogEvent event) { return LogEntry.builder() .level(event.getLevel().name()) .logger(event.getLoggerName()) .message(event.getMessage().getFormattedMessage()) .thread(event.getThreadName()) .timestamp(event.getInstant().getEpochMillisecond()) .context(CONTEXT_MAP) .exception(event.getThrown()) .build(); } }

Spring Boot Configuration

package com.elasticsearch.logging.config; import com.elasticsearch.logging.appender.ElasticsearchAppender; import com.elasticsearch.logging.service.ElasticsearchLogService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import javax.annotation.PostConstruct; @Configuration public class LogAppenderConfig { @Autowired private ElasticsearchLogService elasticsearchLogService; @PostConstruct public void initializeAppender() { ElasticsearchAppender.setLogService(elasticsearchLogService); // Add common context information ElasticsearchAppender.addContext("application", "my-spring-app"); ElasticsearchAppender.addContext("environment", System.getenv("SPRING_PROFILES_ACTIVE")); ElasticsearchAppender.addContext("hostname", getHostname()); } private String getHostname() { try { return java.net.InetAddress.getLocalHost().getHostName(); } catch (Exception e) { return "unknown"; } } }

REST Controller for Log Management

package com.elasticsearch.logging.controller; import com.elasticsearch.logging.model.LogEntry; import com.elasticsearch.logging.service.ElasticsearchLogService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*; import java.io.IOException; import java.time.Instant; import java.util.List; import java.util.Map; @RestController @RequestMapping("/api/logs") public class LogController { @Autowired private ElasticsearchLogService logService; @PostMapping public ResponseEntity<String> addLog(@RequestBody LogEntry logEntry) { logService.indexLog(logEntry); return ResponseEntity.ok("Log accepted"); } @PostMapping("/bulk") public ResponseEntity<String> addLogsBulk(@RequestBody List<LogEntry> logEntries) { logService.indexLogsBulk(logEntries); return ResponseEntity.ok("Logs accepted"); } @GetMapping("/search") public ResponseEntity<Object> searchLogs( @RequestParam String query, @RequestParam(defaultValue = "0") int from, @RequestParam(defaultValue = "50") int size) { try { return ResponseEntity.ok(logService.searchLogs(query, from, size)); } catch (IOException e) { return ResponseEntity.badRequest().body("Search failed: " + e.getMessage()); } } @GetMapping("/search/level/{level}") public ResponseEntity<Object> searchLogsByLevel( @PathVariable String level, @RequestParam(required = false) Instant from, @RequestParam(required = false) Instant to, @RequestParam(defaultValue = "100") int size) { try { if (from == null) from = Instant.now().minusSeconds(3600); // Default: last hour if (to == null) to = Instant.now(); return ResponseEntity.ok(logService.searchLogsByLevel(level, from, to, size)); } catch (IOException e) { return ResponseEntity.badRequest().body("Search failed: " + e.getMessage()); } } @GetMapping("/statistics") public ResponseEntity<Map<String, Object>> getStatistics() { return ResponseEntity.ok(logService.getStatistics()); } @GetMapping("/health") public ResponseEntity<Object> getClusterHealth() { try { return ResponseEntity.ok(logService.getClusterHealth()); } catch (IOException e) { return ResponseEntity.badRequest().body("Health check failed: " + e.getMessage()); } } }

Usage Examples

Application Configuration

# application.yml elasticsearch: host: localhost port: 9200 scheme: http username: elastic password: ${ELASTIC_PASSWORD:changeme} index: prefix: myapp-logs shards: 3 replicas: 1 bulk: batch-size: 1000 flush-interval: 5000 logging: config: classpath:log4j2.xml

Log4j2 Configuration

<?xml version="1.0" encoding="UTF-8"?> <Configuration status="WARN"> <Appenders> <Console name="Console" target="SYSTEM_OUT"> <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/> </Console> <ElasticsearchAppender name="Elasticsearch"> <PatternLayout pattern="%msg"/> </ElasticsearchAppender> </Appenders> <Loggers> <Root level="info"> <AppenderRef ref="Console"/> <AppenderRef ref="Elasticsearch"/> </Root> <Logger name="com.elasticsearch" level="debug" additivity="false"> <AppenderRef ref="Console"/> <AppenderRef ref="Elasticsearch"/> </Logger> </Loggers> </Configuration>

Spring Boot Application

package com.elasticsearch.logging; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableAsync @EnableScheduling public class LogIndexingApplication { public static void main(String[] args) { SpringApplication.run(LogIndexingApplication.class, args); } }

Example Usage in Service

package com.elasticsearch.logging.example; import com.elasticsearch.logging.model.LogEntry; import com.elasticsearch.logging.service.ElasticsearchLogService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.util.Map; @Service public class BusinessService { private static final Logger logger = LoggerFactory.getLogger(BusinessService.class); @Autowired private ElasticsearchLogService elasticsearchLogService; public void processOrder(String orderId) { // Standard logging (will be captured by Log4j2 appender) logger.info("Processing order: {}", orderId); try { // Custom structured logging with additional context LogEntry logEntry = LogEntry.builder() .level("INFO") .logger(getClass().getName()) .message("Started order processing") .traceId(generateTraceId()) .spanId(generateSpanId()) .application("order-service") .environment("production") .addContext("orderId", orderId) .addContext("customerId", "cust-123") .addContext("amount", "150.00") .build(); elasticsearchLogService.indexLog(logEntry); // Business logic validateOrder(orderId); processPayment(orderId); updateInventory(orderId); // Success log LogEntry successLog = LogEntry.builder() .level("INFO") .message("Order processed successfully") .addContext("orderId", orderId) .addContext("processingTime", "150ms") .build(); elasticsearchLogService.indexLog(successLog); } catch (Exception e) { // Error logging with exception LogEntry errorLog = LogEntry.builder() .level("ERROR") .message("Order processing failed") .addContext("orderId", orderId) .exception(e) .build(); elasticsearchLogService.indexLog(errorLog); logger.error("Order processing failed for order: {}", orderId, e); } } private void validateOrder(String orderId) { // Validation logic logger.debug("Validating order: {}", orderId); } private void processPayment(String orderId) { // Payment processing logic logger.info("Processing payment for order: {}", orderId); } private void updateInventory(String orderId) { // Inventory update logic logger.debug("Updating inventory for order: {}", orderId); } private String generateTraceId() { return java.util.UUID.randomUUID().toString(); } private String generateSpanId() { return java.util.UUID.randomUUID().toString().substring(0, 8); } }

Key Features

High-Performance Bulk Indexing: Asynchronous batch processing with configurable batch sizes
Automatic Index Management: Daily index rotation with configurable retention
Structured Logging: Rich context and metadata support
Multiple Authentication Methods: Support for username/password and API key authentication
SSL/TLS Support: Secure communication with Elasticsearch cluster
Comprehensive Monitoring: Real-time statistics and health checks
Log4j2 Integration: Seamless integration with existing logging frameworks
Advanced Search Capabilities: Flexible querying with date ranges and filters
Error Handling & Retry: Robust error handling with fallback mechanisms
Spring Boot Integration: Native Spring configuration and dependency injection

This implementation provides a production-ready Elasticsearch logging solution that can handle high-volume log data while providing excellent search performance and operational visibility.