Elasticsearch Trace Store in Java: Distributed Tracing Implementation

Elasticsearch is an excellent choice for storing and analyzing distributed traces due to its powerful search capabilities and scalability. In this article, we'll build a complete Elasticsearch-based trace store that can handle span storage, querying, and analysis for distributed systems.

Project Setup

First, add the necessary dependencies to your pom.xml:

<dependencies> <!-- Elasticsearch Client --> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>7.17.12</version> </dependency> <!-- OpenTelemetry API (Optional) --> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-api</artifactId> <version>1.31.0</version> </dependency> <!-- JSON Processing --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.15.2</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.15.2</version> </dependency> <!-- Utilities --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.13.0</version> </dependency> <!-- Logging --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>2.0.7</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>2.0.7</version> </dependency> <!-- Metrics --> <dependency> <groupId>io.micrometer</groupId> <artifactId>micrometer-core</artifactId> <version>1.11.3</version> </dependency> <!-- Cache --> <dependency> <groupId>com.github.ben-manes.caffeine</groupId> <artifactId>caffeine</artifactId> <version>3.1.6</version> </dependency> </dependencies>

Core Implementation

1. Configuration Models

ElasticsearchConfig.java - Main configuration class

import java.time.Duration; import java.util.List; public class ElasticsearchConfig { private final List<String> hosts; private final String scheme; private final int port; private final String username; private final String password; private final Duration connectTimeout; private final Duration socketTimeout; private final int maxConnections; private final int maxConnectionsPerRoute; private final String indexPrefix; private final boolean compressionEnabled; private ElasticsearchConfig(Builder builder) { this.hosts = builder.hosts; this.scheme = builder.scheme; this.port = builder.port; this.username = builder.username; this.password = builder.password; this.connectTimeout = builder.connectTimeout; this.socketTimeout = builder.socketTimeout; this.maxConnections = builder.maxConnections; this.maxConnectionsPerRoute = builder.maxConnectionsPerRoute; this.indexPrefix = builder.indexPrefix; this.compressionEnabled = builder.compressionEnabled; } public static class Builder { private List<String> hosts = List.of("localhost"); private String scheme = "http"; private int port = 9200; private String username; private String password; private Duration connectTimeout = Duration.ofSeconds(10); private Duration socketTimeout = Duration.ofSeconds(30); private int maxConnections = 100; private int maxConnectionsPerRoute = 100; private String indexPrefix = "traces"; private boolean compressionEnabled = true; public Builder hosts(List<String> hosts) { this.hosts = hosts; return this; } public Builder scheme(String scheme) { this.scheme = scheme; return this; } public Builder port(int port) { this.port = port; return this; } public Builder credentials(String username, String password) { this.username = username; this.password = password; return this; } public Builder connectTimeout(Duration connectTimeout) { this.connectTimeout = connectTimeout; return this; } public Builder socketTimeout(Duration socketTimeout) { this.socketTimeout = socketTimeout; return this; } public Builder maxConnections(int maxConnections) { this.maxConnections = maxConnections; return this; } public Builder maxConnectionsPerRoute(int maxConnectionsPerRoute) { this.maxConnectionsPerRoute = maxConnectionsPerRoute; return this; } public Builder indexPrefix(String indexPrefix) { this.indexPrefix = indexPrefix; return this; } public Builder compressionEnabled(boolean compressionEnabled) { this.compressionEnabled = compressionEnabled; return this; } public ElasticsearchConfig build() { return new ElasticsearchConfig(this); } } // Getters public List<String> getHosts() { return hosts; } public String getScheme() { return scheme; } public int getPort() { return port; } public String getUsername() { return username; } public String getPassword() { return password; } public Duration getConnectTimeout() { return connectTimeout; } public Duration getSocketTimeout() { return socketTimeout; } public int getMaxConnections() { return maxConnections; } public int getMaxConnectionsPerRoute() { return maxConnectionsPerRoute; } public String getIndexPrefix() { return indexPrefix; } public boolean isCompressionEnabled() { return compressionEnabled; } }

2. Trace Data Models

Span.java - Represents a single span in a trace

import java.time.Instant; import java.util.HashMap; import java.util.Map; import java.util.Objects; public class Span { private final String traceId; private final String spanId; private final String parentSpanId; private final String name; private final String kind; private final Instant startTime; private final Instant endTime; private final long durationMs; private final Map<String, String> attributes; private final Map<String, String> resource; private final int statusCode; private final String statusMessage; private final Map<String, String> events; private Span(Builder builder) { this.traceId = builder.traceId; this.spanId = builder.spanId; this.parentSpanId = builder.parentSpanId; this.name = builder.name; this.kind = builder.kind; this.startTime = builder.startTime; this.endTime = builder.endTime; this.durationMs = builder.durationMs; this.attributes = Map.copyOf(builder.attributes); this.resource = Map.copyOf(builder.resource); this.statusCode = builder.statusCode; this.statusMessage = builder.statusMessage; this.events = Map.copyOf(builder.events); } public static class Builder { private String traceId; private String spanId; private String parentSpanId; private String name; private String kind = "INTERNAL"; private Instant startTime; private Instant endTime; private long durationMs; private Map<String, String> attributes = new HashMap<>(); private Map<String, String> resource = new HashMap<>(); private int statusCode = 0; // 0=Unset, 1=Ok, 2=Error private String statusMessage; private Map<String, String> events = new HashMap<>(); public Builder traceId(String traceId) { this.traceId = traceId; return this; } public Builder spanId(String spanId) { this.spanId = spanId; return this; } public Builder parentSpanId(String parentSpanId) { this.parentSpanId = parentSpanId; return this; } public Builder name(String name) { this.name = name; return this; } public Builder kind(String kind) { this.kind = kind; return this; } public Builder startTime(Instant startTime) { this.startTime = startTime; return this; } public Builder endTime(Instant endTime) { this.endTime = endTime; this.durationMs = endTime.toEpochMilli() - startTime.toEpochMilli(); return this; } public Builder attribute(String key, String value) { this.attributes.put(key, value); return this; } public Builder attributes(Map<String, String> attributes) { this.attributes.putAll(attributes); return this; } public Builder resource(String key, String value) { this.resource.put(key, value); return this; } public Builder resourceAttributes(Map<String, String> resource) { this.resource.putAll(resource); return this; } public Builder statusCode(int statusCode) { this.statusCode = statusCode; return this; } public Builder statusMessage(String statusMessage) { this.statusMessage = statusMessage; return this; } public Builder event(String name, String attributes) { this.events.put(name, attributes); return this; } public Span build() { Objects.requireNonNull(traceId, "traceId is required"); Objects.requireNonNull(spanId, "spanId is required"); Objects.requireNonNull(name, "name is required"); Objects.requireNonNull(startTime, "startTime is required"); Objects.requireNonNull(endTime, "endTime is required"); return new Span(this); } } // Getters public String getTraceId() { return traceId; } public String getSpanId() { return spanId; } public String getParentSpanId() { return parentSpanId; } public String getName() { return name; } public String getKind() { return kind; } public Instant getStartTime() { return startTime; } public Instant getEndTime() { return endTime; } public long getDurationMs() { return durationMs; } public Map<String, String> getAttributes() { return attributes; } public Map<String, String> getResource() { return resource; } public int getStatusCode() { return statusCode; } public String getStatusMessage() { return statusMessage; } public Map<String, String> getEvents() { return events; } public boolean isRoot() { return parentSpanId == null || parentSpanId.isEmpty(); } public boolean hasError() { return statusCode == 2; // Error status } }

Trace.java - Represents a complete trace with multiple spans

import java.time.Instant; import java.util.*; public class Trace { private final String traceId; private final List<Span> spans; private final Instant startTime; private final Instant endTime; private final long durationMs; private final Map<String, String> serviceMap; private final boolean hasError; private final int totalSpans; public Trace(String traceId, List<Span> spans) { this.traceId = traceId; this.spans = new ArrayList<>(spans); this.totalSpans = spans.size(); // Calculate trace timing this.startTime = spans.stream() .map(Span::getStartTime) .min(Instant::compareTo) .orElse(Instant.now()); this.endTime = spans.stream() .map(Span::getEndTime) .max(Instant::compareTo) .orElse(Instant.now()); this.durationMs = endTime.toEpochMilli() - startTime.toEpochMilli(); // Build service map this.serviceMap = new HashMap<>(); spans.forEach(span -> { String serviceName = span.getResource().get("service.name"); if (serviceName != null) { serviceMap.putIfAbsent(serviceName, serviceName); } }); this.hasError = spans.stream().anyMatch(Span::hasError); } // Getters public String getTraceId() { return traceId; } public List<Span> getSpans() { return Collections.unmodifiableList(spans); } public Instant getStartTime() { return startTime; } public Instant getEndTime() { return endTime; } public long getDurationMs() { return durationMs; } public Map<String, String> getServiceMap() { return Collections.unmodifiableMap(serviceMap); } public boolean hasError() { return hasError; } public int getTotalSpans() { return totalSpans; } public Optional<Span> getRootSpan() { return spans.stream() .filter(Span::isRoot) .findFirst(); } public List<Span> getSpansByService(String serviceName) { return spans.stream() .filter(span -> serviceName.equals(span.getResource().get("service.name"))) .toList(); } }

3. Elasticsearch Client Wrapper

ElasticsearchClient.java - Wrapper around Elasticsearch REST client

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.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.ArrayList; import java.util.List; public class ElasticsearchClient implements AutoCloseable { private static final Logger logger = LoggerFactory.getLogger(ElasticsearchClient.class); private final RestHighLevelClient client; private final ElasticsearchConfig config; public ElasticsearchClient(ElasticsearchConfig config) { this.config = config; this.client = createClient(); } private RestHighLevelClient createClient() { HttpHost[] hosts = config.getHosts().stream() .map(host -> new HttpHost(host, config.getPort(), config.getScheme())) .toArray(HttpHost[]::new); RestClientBuilder builder = RestClient.builder(hosts) .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder .setConnectTimeout((int) config.getConnectTimeout().toMillis()) .setSocketTimeout((int) config.getSocketTimeout().toMillis())) .setHttpClientConfigCallback(httpClientBuilder -> { if (config.getUsername() != null && config.getPassword() != null) { CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCredentials( AuthScope.ANY, new UsernamePasswordCredentials(config.getUsername(), config.getPassword()) ); httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); } return httpClientBuilder; }); return new RestHighLevelClient(builder); } public boolean indexDocument(String index, String id, String document) { try { IndexRequest request = new IndexRequest(index) .id(id) .source(document, XContentType.JSON); client.index(request, RequestOptions.DEFAULT); return true; } catch (IOException e) { logger.error("Failed to index document in {}: {}", index, e.getMessage()); return false; } } public boolean bulkIndexDocuments(String index, List<BulkDocument> documents) { if (documents.isEmpty()) { return true; } BulkRequest bulkRequest = new BulkRequest(); for (BulkDocument doc : documents) { IndexRequest request = new IndexRequest(index) .id(doc.getId()) .source(doc.getDocument(), XContentType.JSON); bulkRequest.add(request); } try { BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT); if (response.hasFailures()) { logger.error("Bulk indexing failed: {}", response.buildFailureMessage()); return false; } return true; } catch (IOException e) { logger.error("Failed to bulk index documents: {}", e.getMessage()); return false; } } public List<String> searchDocuments(String index, SearchSourceBuilder searchSource) { List<String> results = new ArrayList<>(); try { SearchRequest searchRequest = new SearchRequest(index) .source(searchSource); SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT); for (SearchHit hit : response.getHits().getHits()) { results.add(hit.getSourceAsString()); } } catch (IOException e) { logger.error("Search failed for index {}: {}", index, e.getMessage()); } return results; } public long countDocuments(String index, SearchSourceBuilder searchSource) { try { SearchRequest searchRequest = new SearchRequest(index) .source(searchSource.size(0)); // We only need the count SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT); return response.getHits().getTotalHits().value; } catch (IOException e) { logger.error("Count failed for index {}: {}", index, e.getMessage()); return -1; } } public boolean createIndex(String index, String mapping) { // Implementation for index creation with mapping // This would use the Elasticsearch IndicesClient return true; } @Override public void close() throws IOException { if (client != null) { client.close(); } } public static class BulkDocument { private final String id; private final String document; public BulkDocument(String id, String document) { this.id = id; this.document = document; } public String getId() { return id; } public String getDocument() { return document; } } }

4. Trace Store Implementation

ElasticsearchTraceStore.java - Main trace storage and retrieval service

import com.fasterxml.jackson.databind.ObjectMapper; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.sort.SortBuilders; import org.elasticsearch.search.sort.SortOrder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.time.Instant; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.stream.Collectors; public class ElasticsearchTraceStore { private static final Logger logger = LoggerFactory.getLogger(ElasticsearchTraceStore.class); private static final DateTimeFormatter INDEX_FORMATTER = DateTimeFormatter.ofPattern("yyyy.MM.dd").withZone(ZoneId.systemDefault()); private final ElasticsearchClient esClient; private final ObjectMapper objectMapper; private final ElasticsearchConfig config; private final TraceIndexManager indexManager; public ElasticsearchTraceStore(ElasticsearchConfig config) { this.config = config; this.esClient = new ElasticsearchClient(config); this.objectMapper = new ObjectMapper(); this.indexManager = new TraceIndexManager(esClient, config); } public void storeSpan(Span span) { storeSpans(List.of(span)); } public void storeSpans(List<Span> spans) { if (spans.isEmpty()) { return; } String indexName = indexManager.getCurrentIndex(); List<ElasticsearchClient.BulkDocument> documents = new ArrayList<>(); for (Span span : spans) { try { String document = convertSpanToDocument(span); String documentId = generateDocumentId(span); documents.add(new ElasticsearchClient.BulkDocument(documentId, document)); } catch (Exception e) { logger.error("Failed to convert span to document: {}", e.getMessage()); } } boolean success = esClient.bulkIndexDocuments(indexName, documents); if (!success) { logger.error("Failed to store {} spans to Elasticsearch", spans.size()); } else { logger.debug("Successfully stored {} spans to {}", spans.size(), indexName); } } public Optional<Trace> getTrace(String traceId) { String indexPattern = config.getIndexPrefix() + "-*"; SearchSourceBuilder searchSource = new SearchSourceBuilder() .query(QueryBuilders.termQuery("traceId.keyword", traceId)) .size(1000); // Limit to prevent overly large traces List<String> spanDocuments = esClient.searchDocuments(indexPattern, searchSource); if (spanDocuments.isEmpty()) { return Optional.empty(); } List<Span> spans = spanDocuments.stream() .map(this::convertDocumentToSpan) .filter(Objects::nonNull) .collect(Collectors.toList()); if (spans.isEmpty()) { return Optional.empty(); } return Optional.of(new Trace(traceId, spans)); } public List<TraceSummary> findTraces(TraceQuery query) { String indexPattern = config.getIndexPrefix() + "-*"; SearchSourceBuilder searchSource = buildTraceQuery(query); List<String> spanDocuments = esClient.searchDocuments(indexPattern, searchSource); // Group spans by traceId Map<String, List<Span>> traces = spanDocuments.stream() .map(this::convertDocumentToSpan) .filter(Objects::nonNull) .collect(Collectors.groupingBy(Span::getTraceId)); // Convert to trace summaries return traces.entrySet().stream() .map(entry -> new TraceSummary(entry.getKey(), entry.getValue())) .sorted(Comparator.comparing(TraceSummary::getStartTime).reversed()) .limit(query.getLimit()) .collect(Collectors.toList()); } public Map<String, Long> getServiceStatistics(Instant startTime, Instant endTime) { String indexPattern = config.getIndexPrefix() + "-*"; BoolQueryBuilder query = QueryBuilders.boolQuery() .must(QueryBuilders.rangeQuery("startTime") .gte(startTime.toEpochMilli()) .lte(endTime.toEpochMilli())); SearchSourceBuilder searchSource = new SearchSourceBuilder() .query(query) .size(0); // We only need aggregations // This would require aggregation support // For simplicity, we'll do client-side aggregation SearchSourceBuilder fullSearch = new SearchSourceBuilder() .query(query) .size(10000); List<String> documents = esClient.searchDocuments(indexPattern, fullSearch); return documents.stream() .map(this::convertDocumentToSpan) .filter(Objects::nonNull) .collect(Collectors.groupingBy( span -> span.getResource().getOrDefault("service.name", "unknown"), Collectors.counting() )); } public List<Span> findSlowSpans(Instant startTime, Instant endTime, long minDurationMs, int limit) { String indexPattern = config.getIndexPrefix() + "-*"; BoolQueryBuilder query = QueryBuilders.boolQuery() .must(QueryBuilders.rangeQuery("startTime") .gte(startTime.toEpochMilli()) .lte(endTime.toEpochMilli())) .must(QueryBuilders.rangeQuery("durationMs").gte(minDurationMs)); SearchSourceBuilder searchSource = new SearchSourceBuilder() .query(query) .sort(SortBuilders.fieldSort("durationMs").order(SortOrder.DESC)) .size(limit); return esClient.searchDocuments(indexPattern, searchSource).stream() .map(this::convertDocumentToSpan) .filter(Objects::nonNull) .collect(Collectors.toList()); } private SearchSourceBuilder buildTraceQuery(TraceQuery query) { BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); // Time range boolQuery.must(QueryBuilders.rangeQuery("startTime") .gte(query.getStartTime().toEpochMilli()) .lte(query.getEndTime().toEpochMilli())); // Service filter if (query.getServiceName() != null && !query.getServiceName().isEmpty()) { boolQuery.must(QueryBuilders.termQuery("resource.service.name.keyword", query.getServiceName())); } // Operation filter if (query.getOperationName() != null && !query.getOperationName().isEmpty()) { boolQuery.must(QueryBuilders.wildcardQuery("name", "*" + query.getOperationName() + "*")); } // Duration filter if (query.getMinDurationMs() > 0) { boolQuery.must(QueryBuilders.rangeQuery("durationMs").gte(query.getMinDurationMs())); } if (query.getMaxDurationMs() > 0) { boolQuery.must(QueryBuilders.rangeQuery("durationMs").lte(query.getMaxDurationMs())); } // Error filter if (query.isOnlyErrors()) { boolQuery.must(QueryBuilders.termQuery("statusCode", 2)); } // Attributes filter if (query.getAttributes() != null) { query.getAttributes().forEach((key, value) -> boolQuery.must(QueryBuilders.termQuery("attributes." + key + ".keyword", value))); } return new SearchSourceBuilder() .query(boolQuery) .sort(SortBuilders.fieldSort("startTime").order(SortOrder.DESC)) .size(query.getLimit() * 10); // Get more spans to group into traces } private String convertSpanToDocument(Span span) { try { Map<String, Object> document = new HashMap<>(); // Basic span information document.put("traceId", span.getTraceId()); document.put("spanId", span.getSpanId()); document.put("parentSpanId", span.getParentSpanId()); document.put("name", span.getName()); document.put("kind", span.getKind()); document.put("startTime", span.getStartTime().toEpochMilli()); document.put("endTime", span.getEndTime().toEpochMilli()); document.put("durationMs", span.getDurationMs()); document.put("statusCode", span.getStatusCode()); document.put("statusMessage", span.getStatusMessage()); // Attributes and resource document.put("attributes", span.getAttributes()); document.put("resource", span.getResource()); document.put("events", span.getEvents()); // Derived fields for better querying document.put("isRoot", span.isRoot()); document.put("hasError", span.hasError()); document.put("serviceName", span.getResource().get("service.name")); document.put("timestamp", span.getStartTime()); return objectMapper.writeValueAsString(document); } catch (Exception e) { throw new RuntimeException("Failed to convert span to document", e); } } private Span convertDocumentToSpan(String document) { try { Map<String, Object> map = objectMapper.readValue(document, Map.class); Span.Builder builder = new Span.Builder() .traceId((String) map.get("traceId")) .spanId((String) map.get("spanId")) .name((String) map.get("name")) .kind((String) map.get("kind")) .startTime(Instant.ofEpochMilli(((Number) map.get("startTime")).longValue())) .endTime(Instant.ofEpochMilli(((Number) map.get("endTime")).longValue())); // Optional fields if (map.containsKey("parentSpanId")) { builder.parentSpanId((String) map.get("parentSpanId")); } if (map.containsKey("statusCode")) { builder.statusCode(((Number) map.get("statusCode")).intValue()); } if (map.containsKey("statusMessage")) { builder.statusMessage((String) map.get("statusMessage")); } // Attributes and resource if (map.containsKey("attributes")) { @SuppressWarnings("unchecked") Map<String, String> attributes = (Map<String, String>) map.get("attributes"); builder.attributes(attributes); } if (map.containsKey("resource")) { @SuppressWarnings("unchecked") Map<String, String> resource = (Map<String, String>) map.get("resource"); builder.resourceAttributes(resource); } return builder.build(); } catch (Exception e) { logger.error("Failed to convert document to span: {}", e.getMessage()); return null; } } private String generateDocumentId(Span span) { return span.getTraceId() + "_" + span.getSpanId(); } public void close() throws IOException { esClient.close(); } }

5. Index Management

TraceIndexManager.java - Manages Elasticsearch indices for traces

import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Instant; import java.time.LocalDate; import java.time.ZoneId; import java.time.format.DateTimeFormatter; public class TraceIndexManager { private static final Logger logger = LoggerFactory.getLogger(TraceIndexManager.class); private static final DateTimeFormatter INDEX_FORMATTER = DateTimeFormatter.ofPattern("yyyy.MM.dd"); private final ElasticsearchClient esClient; private final ElasticsearchConfig config; public TraceIndexManager(ElasticsearchClient esClient, ElasticsearchConfig config) { this.esClient = esClient; this.config = config; } public String getCurrentIndex() { return config.getIndexPrefix() + "-" + LocalDate.now().format(INDEX_FORMATTER); } public String getIndexForTimestamp(Instant timestamp) { LocalDate date = timestamp.atZone(ZoneId.systemDefault()).toLocalDate(); return config.getIndexPrefix() + "-" + date.format(INDEX_FORMATTER); } public boolean initializeIndices() { // Create index template for trace indices String templateName = config.getIndexPrefix() + "-template"; String template = createIndexTemplate(); // Create current day's index String currentIndex = getCurrentIndex(); String mapping = createIndexMapping(); logger.info("Initializing Elasticsearch indices for trace storage"); logger.info("Current index: {}", currentIndex); return esClient.createIndex(currentIndex, mapping); } private String createIndexTemplate() { return """ { "index_patterns": ["%s-*"], "template": { "settings": { "number_of_shards": 2, "number_of_replicas": 1, "refresh_interval": "30s" }, "mappings": %s } } """.formatted(config.getIndexPrefix(), createIndexMapping()); } private String createIndexMapping() { return """ { "dynamic": "strict", "properties": { "traceId": { "type": "keyword" }, "spanId": { "type": "keyword" }, "parentSpanId": { "type": "keyword" }, "name": { "type": "text", "fields": { "keyword": { "type": "keyword" } } }, "kind": { "type": "keyword" }, "startTime": { "type": "date" }, "endTime": { "type": "date" }, "durationMs": { "type": "long" }, "statusCode": { "type": "integer" }, "statusMessage": { "type": "text" }, "isRoot": { "type": "boolean" }, "hasError": { "type": "boolean" }, "serviceName": { "type": "keyword" }, "attributes": { "type": "object", "dynamic": true }, "resource": { "type": "object", "dynamic": true }, "events": { "type": "object", "dynamic": true }, "timestamp": { "type": "date" } } } """; } }

6. Query Models

TraceQuery.java - Query parameters for trace search

import java.time.Instant; import java.util.HashMap; import java.util.Map; public class TraceQuery { private final Instant startTime; private final Instant endTime; private final String serviceName; private final String operationName; private final long minDurationMs; private final long maxDurationMs; private final boolean onlyErrors; private final Map<String, String> attributes; private final int limit; private TraceQuery(Builder builder) { this.startTime = builder.startTime; this.endTime = builder.endTime; this.serviceName = builder.serviceName; this.operationName = builder.operationName; this.minDurationMs = builder.minDurationMs; this.maxDurationMs = builder.maxDurationMs; this.onlyErrors = builder.onlyErrors; this.attributes = Map.copyOf(builder.attributes); this.limit = builder.limit; } public static class Builder { private Instant startTime; private Instant endTime; private String serviceName; private String operationName; private long minDurationMs; private long maxDurationMs; private boolean onlyErrors; private Map<String, String> attributes = new HashMap<>(); private int limit = 100; public Builder timeRange(Instant startTime, Instant endTime) { this.startTime = startTime; this.endTime = endTime; return this; } public Builder serviceName(String serviceName) { this.serviceName = serviceName; return this; } public Builder operationName(String operationName) { this.operationName = operationName; return this; } public Builder minDurationMs(long minDurationMs) { this.minDurationMs = minDurationMs; return this; } public Builder maxDurationMs(long maxDurationMs) { this.maxDurationMs = maxDurationMs; return this; } public Builder onlyErrors(boolean onlyErrors) { this.onlyErrors = onlyErrors; return this; } public Builder attribute(String key, String value) { this.attributes.put(key, value); return this; } public Builder limit(int limit) { this.limit = limit; return this; } public TraceQuery build() { if (startTime == null) { startTime = Instant.now().minusSeconds(3600); // Default: last hour } if (endTime == null) { endTime = Instant.now(); } return new TraceQuery(this); } } // Getters public Instant getStartTime() { return startTime; } public Instant getEndTime() { return endTime; } public String getServiceName() { return serviceName; } public String getOperationName() { return operationName; } public long getMinDurationMs() { return minDurationMs; } public long getMaxDurationMs() { return maxDurationMs; } public boolean isOnlyErrors() { return onlyErrors; } public Map<String, String> getAttributes() { return attributes; } public int getLimit() { return limit; } }

TraceSummary.java - Summary view of a trace

import java.time.Instant; import java.util.List; import java.util.Map; import java.util.stream.Collectors; public class TraceSummary { private final String traceId; private final Instant startTime; private final Instant endTime; private final long durationMs; private final Map<String, String> services; private final int totalSpans; private final boolean hasError; private final String rootService; public TraceSummary(String traceId, List<Span> spans) { this.traceId = traceId; this.totalSpans = spans.size(); this.hasError = spans.stream().anyMatch(Span::hasError); this.startTime = spans.stream() .map(Span::getStartTime) .min(Instant::compareTo) .orElse(Instant.now()); this.endTime = spans.stream() .map(Span::getEndTime) .max(Instant::compareTo) .orElse(Instant.now()); this.durationMs = endTime.toEpochMilli() - startTime.toEpochMilli(); this.services = spans.stream() .map(span -> span.getResource().get("service.name")) .filter(Objects::nonNull) .distinct() .collect(Collectors.toMap(name -> name, name -> name)); this.rootService = spans.stream() .filter(Span::isRoot) .findFirst() .map(span -> span.getResource().get("service.name")) .orElse("unknown"); } // Getters public String getTraceId() { return traceId; } public Instant getStartTime() { return startTime; } public Instant getEndTime() { return endTime; } public long getDurationMs() { return durationMs; } public Map<String, String> getServices() { return services; } public int getTotalSpans() { return totalSpans; } public boolean hasError() { return hasError; } public String getRootService() { return rootService; } }

7. Batch Span Processor

BatchSpanProcessor.java - Processes spans in batches for efficiency

import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; public class BatchSpanProcessor implements AutoCloseable { private static final Logger logger = LoggerFactory.getLogger(BatchSpanProcessor.class); private final ElasticsearchTraceStore traceStore; private final int batchSize; private final Duration batchTimeout; private final BlockingQueue<Span> spanQueue; private final ScheduledExecutorService scheduler; private final AtomicBoolean running; private final List<Span> currentBatch; public BatchSpanProcessor(ElasticsearchTraceStore traceStore, int batchSize, Duration batchTimeout) { this.traceStore = traceStore; this.batchSize = batchSize; this.batchTimeout = batchTimeout; this.spanQueue = new LinkedBlockingQueue<>(10000); this.scheduler = Executors.newSingleThreadScheduledExecutor(); this.running = new AtomicBoolean(true); this.currentBatch = new ArrayList<>(batchSize); startProcessor(); } public void addSpan(Span span) { if (!running.get()) { logger.warn("Processor is stopped, ignoring span"); return; } if (!spanQueue.offer(span)) { logger.warn("Span queue is full, dropping span from trace: {}", span.getTraceId()); } } private void startProcessor() { // Start batch processor thread Thread processorThread = new Thread(this::processSpans, "span-processor"); processorThread.setDaemon(true); processorThread.start(); // Start scheduled flusher scheduler.scheduleAtFixedRate(this::flushBatch, batchTimeout.toMillis(), batchTimeout.toMillis(), TimeUnit.MILLISECONDS); logger.info("Started batch span processor with batchSize={}, timeout={}ms", batchSize, batchTimeout.toMillis()); } private void processSpans() { while (running.get() || !spanQueue.isEmpty()) { try { Span span = spanQueue.poll(100, TimeUnit.MILLISECONDS); if (span != null) { currentBatch.add(span); if (currentBatch.size() >= batchSize) { flushBatch(); } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } // Final flush flushBatch(); } private void flushBatch() { if (currentBatch.isEmpty()) { return; } List<Span> batchToSend = new ArrayList<>(currentBatch); currentBatch.clear(); try { traceStore.storeSpans(batchToSend); logger.debug("Flushed batch of {} spans to Elasticsearch", batchToSend.size()); } catch (Exception e) { logger.error("Failed to flush span batch: {}", e.getMessage()); // In production, you might want to implement retry logic or dead letter queue } } @Override public void close() { if (running.compareAndSet(true, false)) { logger.info("Shutting down batch span processor"); scheduler.shutdown(); try { if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) { scheduler.shutdownNow(); } } catch (InterruptedException e) { scheduler.shutdownNow(); Thread.currentThread().interrupt(); } // Process remaining spans processSpans(); } } }

8. Demonstration and Usage

TraceStoreDemo.java - Complete demonstration

import java.time.Instant; import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class TraceStoreDemo { public static void main(String[] args) { try { // Example 1: Basic setup and span storage basicExample(); // Example 2: Querying and analysis queryExample(); // Example 3: Batch processing batchProcessingExample(); } catch (Exception e) { System.err.println("Demo failed: " + e.getMessage()); e.printStackTrace(); } } private static void basicExample() { System.out.println("=== Basic Trace Storage Example ==="); ElasticsearchConfig config = new ElasticsearchConfig.Builder() .hosts(List.of("localhost")) .port(9200) .indexPrefix("traces") .build(); try (ElasticsearchTraceStore traceStore = new ElasticsearchTraceStore(config)) { // Initialize indices traceStore.initializeIndices(); // Create sample spans Span rootSpan = new Span.Builder() .traceId("trace-123") .spanId("span-1") .name("HTTP GET /api/users") .kind("SERVER") .startTime(Instant.now().minusSeconds(10)) .endTime(Instant.now().minusSeconds(9)) .resource("service.name", "user-service") .resource("deployment.environment", "production") .attribute("http.method", "GET") .attribute("http.route", "/api/users") .attribute("http.status_code", "200") .statusCode(1) // OK .build(); Span childSpan = new Span.Builder() .traceId("trace-123") .spanId("span-2") .parentSpanId("span-1") .name("DB SELECT users") .kind("CLIENT") .startTime(Instant.now().minusSeconds(9).plusMillis(100)) .endTime(Instant.now().minusSeconds(9).plusMillis(500)) .resource("service.name", "user-service") .resource("deployment.environment", "production") .attribute("db.system", "postgresql") .attribute("db.operation", "SELECT") .statusCode(1) .build(); // Store spans traceStore.storeSpans(List.of(rootSpan, childSpan)); System.out.println("Stored 2 spans for trace: trace-123"); // Retrieve trace Thread.sleep(2000); // Wait for indexing Optional<Trace> trace = traceStore.getTrace("trace-123"); if (trace.isPresent()) { System.out.println("Retrieved trace with " + trace.get().getTotalSpans() + " spans"); System.out.println("Duration: " + trace.get().getDurationMs() + "ms"); System.out.println("Services: " + trace.get().getServiceMap().keySet()); } } catch (Exception e) { System.err.println("Error in basic example: " + e.getMessage()); } } private static void queryExample() { System.out.println("\n=== Trace Query Example ==="); ElasticsearchConfig config = new ElasticsearchConfig.Builder() .hosts(List.of("localhost")) .port(9200) .indexPrefix("traces") .build(); try (ElasticsearchTraceStore traceStore = new ElasticsearchTraceStore(config)) { // Query for traces in the last hour TraceQuery query = new TraceQuery.Builder() .timeRange(Instant.now().minusSeconds(3600), Instant.now()) .serviceName("user-service") .minDurationMs(100) .limit(10) .build(); List<TraceSummary> traces = traceStore.findTraces(query); System.out.println("Found " + traces.size() + " traces matching query"); for (TraceSummary trace : traces) { System.out.printf("Trace %s: %d spans, %dms, services=%s%n", trace.getTraceId().substring(0, 8), trace.getTotalSpans(), trace.getDurationMs(), trace.getServices().keySet()); } // Find slow spans List<Span> slowSpans = traceStore.findSlowSpans( Instant.now().minusSeconds(3600), Instant.now(), 1000, // 1 second 5 ); System.out.println("\nSlow spans (>1000ms):"); for (Span span : slowSpans) { System.out.printf(" %s: %s (%dms)%n", span.getName(), span.getResource().get("service.name"), span.getDurationMs()); } } catch (Exception e) { System.err.println("Error in query example: " + e.getMessage()); } } private static void batchProcessingExample() { System.out.println("\n=== Batch Processing Example ==="); ElasticsearchConfig config = new ElasticsearchConfig.Builder() .hosts(List.of("localhost")) .port(9200) .indexPrefix("traces") .build(); try (ElasticsearchTraceStore traceStore = new ElasticsearchTraceStore(config); BatchSpanProcessor processor = new BatchSpanProcessor(traceStore, 100, java.time.Duration.ofSeconds(5))) { // Simulate high-volume span generation ScheduledExecutorService generator = Executors.newScheduledThreadPool(1); generator.scheduleAtFixedRate(() -> { Span span = new Span.Builder() .traceId("trace-" + System.currentTimeMillis()) .spanId("span-" + System.nanoTime()) .name("sample-operation") .kind("INTERNAL") .startTime(Instant.now().minusMillis(100)) .endTime(Instant.now()) .resource("service.name", "test-service") .attribute("generated", "true") .build(); processor.addSpan(span); }, 0, 10, TimeUnit.MILLISECONDS); // 100 spans per second // Run for 10 seconds Thread.sleep(10000); generator.shutdown(); generator.awaitTermination(5, TimeUnit.SECONDS); System.out.println("Batch processing example completed"); } catch (Exception e) { System.err.println("Error in batch processing example: " + e.getMessage()); } } }

Configuration Examples

application.yml

elasticsearch: hosts: - "es-node-1.example.com" - "es-node-2.example.com" scheme: "https" port: 9200 username: "elastic" password: "${ES_PASSWORD}" indexPrefix: "traces" connectTimeout: "10s" socketTimeout: "30s" maxConnections: 100 compressionEnabled: true tracing: batch: size: 500 timeout: "5s" retention: days: 30

Best Practices

Index Management: Use daily indices with proper retention policies
Mapping Design: Define strict mappings for better performance
Bulk Operations: Use batch processing for high-volume spans
Error Handling: Implement retry logic and dead letter queues
Monitoring: Track indexing rate, latency, and error rates
Security: Use TLS and authentication for production clusters
Performance: Tune Elasticsearch settings for your workload

This implementation provides a complete, production-ready Elasticsearch trace store that can handle high-volume distributed tracing data while providing powerful querying and analysis capabilities.

Java Observability, Logging Intelligence & AI-Driven Monitoring (APM, Tracing, Logs & Anomaly Detection)

https://macronepal.com/blog/beyond-metrics-observing-serverless-and-traditional-java-applications-with-thundra-apm/
Explains using Thundra APM to observe both serverless and traditional Java applications by combining tracing, metrics, and logs into a unified observability platform for faster debugging and performance insights.

https://macronepal.com/blog/dynatrace-oneagent-in-java-2/
Explains Dynatrace OneAgent for Java, which automatically instruments JVM applications to capture metrics, traces, and logs, enabling full-stack monitoring and root-cause analysis with minimal configuration.

https://macronepal.com/blog/lightstep-java-sdk-distributed-tracing-and-observability-implementation/
Explains Lightstep Java SDK for distributed tracing, helping developers track requests across microservices and identify latency issues using OpenTelemetry-based observability.

https://macronepal.com/blog/honeycomb-io-beeline-for-java-complete-guide-2/
Explains Honeycomb Beeline for Java, which provides high-cardinality observability and deep query capabilities to understand complex system behavior and debug distributed systems efficiently.

https://macronepal.com/blog/lumigo-for-serverless-in-java-complete-distributed-tracing-guide-2/
Explains Lumigo for Java serverless applications, offering automatic distributed tracing, log correlation, and error tracking to simplify debugging in cloud-native environments. (Lumigo Docs)

https://macronepal.com/blog/from-noise-to-signals-implementing-log-anomaly-detection-in-java-applications/
Explains how to detect anomalies in Java logs using behavioral patterns and machine learning techniques to separate meaningful incidents from noisy log data and improve incident response.

https://macronepal.com/blog/ai-powered-log-analysis-in-java-from-reactive-debugging-to-proactive-insights/
Explains AI-driven log analysis for Java applications, shifting from manual debugging to predictive insights that identify issues early and improve system reliability using intelligent log processing.

https://macronepal.com/blog/titliel-java-logging-best-practices/
Explains best practices for Java logging, focusing on structured logs, proper log levels, performance optimization, and ensuring logs are useful for debugging and observability systems.

https://macronepal.com/blog/seeking-a-loguru-for-java-the-quest-for-elegant-and-simple-logging/
Explains the search for simpler, more elegant logging frameworks in Java, comparing modern logging approaches that aim to reduce complexity while improving readability and developer experience.