Loki is a horizontally-scalable, highly-available, multi-tenant log aggregation system inspired by Prometheus. In this article, we'll build a complete Java-based log aggregation system that integrates with Loki, including client libraries, log collectors, and management tools.
Project Setup
First, add the necessary dependencies to your pom.xml:
<dependencies> <!-- HTTP Client for Loki API --> <dependency> <groupId>org.apache.httpcomponents.client5</groupId> <artifactId>httpclient5</artifactId> <version>5.2.1</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> <!-- Logging Framework --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>2.0.7</version> </dependency> <!-- Metrics (Optional) --> <dependency> <groupId>io.micrometer</groupId> <artifactId>micrometer-core</artifactId> <version>1.11.3</version> </dependency> <!-- Compression --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-compress</artifactId> <version>1.23.0</version> </dependency> <!-- Utilities --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.13.0</version> </dependency> <!-- Testing --> <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter</artifactId> <version>5.10.0</version> <scope>test</scope> </dependency> </dependencies>
Core Implementation
1. Configuration Models
LokiConfig.java - Main configuration class
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
@JsonIgnoreProperties(ignoreUnknown = true)
public class LokiConfig {
private final String url;
private final String tenantId;
private final Duration timeout;
private final int maxBatchSize;
private final Duration batchTimeout;
private final int maxRetries;
private final Duration retryDelay;
private final boolean compressionEnabled;
private final Map<String, String> defaultLabels;
private final AuthConfig auth;
private LokiConfig(Builder builder) {
this.url = builder.url;
this.tenantId = builder.tenantId;
this.timeout = builder.timeout;
this.maxBatchSize = builder.maxBatchSize;
this.batchTimeout = builder.batchTimeout;
this.maxRetries = builder.maxRetries;
this.retryDelay = builder.retryDelay;
this.compressionEnabled = builder.compressionEnabled;
this.defaultLabels = builder.defaultLabels;
this.auth = builder.auth;
}
public static class Builder {
private String url = "http://localhost:3100";
private String tenantId;
private Duration timeout = Duration.ofSeconds(30);
private int maxBatchSize = 1000;
private Duration batchTimeout = Duration.ofSeconds(5);
private int maxRetries = 3;
private Duration retryDelay = Duration.ofSeconds(1);
private boolean compressionEnabled = true;
private Map<String, String> defaultLabels = new HashMap<>();
private AuthConfig auth;
public Builder url(String url) {
this.url = url;
return this;
}
public Builder tenantId(String tenantId) {
this.tenantId = tenantId;
return this;
}
public Builder timeout(Duration timeout) {
this.timeout = timeout;
return this;
}
public Builder maxBatchSize(int maxBatchSize) {
this.maxBatchSize = maxBatchSize;
return this;
}
public Builder batchTimeout(Duration batchTimeout) {
this.batchTimeout = batchTimeout;
return this;
}
public Builder maxRetries(int maxRetries) {
this.maxRetries = maxRetries;
return this;
}
public Builder retryDelay(Duration retryDelay) {
this.retryDelay = retryDelay;
return this;
}
public Builder compressionEnabled(boolean compressionEnabled) {
this.compressionEnabled = compressionEnabled;
return this;
}
public Builder defaultLabel(String key, String value) {
this.defaultLabels.put(key, value);
return this;
}
public Builder defaultLabels(Map<String, String> labels) {
this.defaultLabels.putAll(labels);
return this;
}
public Builder auth(AuthConfig auth) {
this.auth = auth;
return this;
}
public LokiConfig build() {
// Set default labels if none provided
if (defaultLabels.isEmpty()) {
defaultLabels.put("job", "java-application");
defaultLabels.put("language", "java");
}
return new LokiConfig(this);
}
}
// Getters
public String getUrl() { return url; }
public String getTenantId() { return tenantId; }
public Duration getTimeout() { return timeout; }
public int getMaxBatchSize() { return maxBatchSize; }
public Duration getBatchTimeout() { return batchTimeout; }
public int getMaxRetries() { return maxRetries; }
public Duration getRetryDelay() { return retryDelay; }
public boolean isCompressionEnabled() { return compressionEnabled; }
public Map<String, String> getDefaultLabels() { return defaultLabels; }
public AuthConfig getAuth() { return auth; }
}
AuthConfig.java - Authentication configuration
public class AuthConfig {
private final AuthType type;
private final String username;
private final String password;
private final String token;
private final String tokenFile;
public enum AuthType {
NONE, BASIC, BEARER, TOKEN_FILE
}
private AuthConfig(Builder builder) {
this.type = builder.type;
this.username = builder.username;
this.password = builder.password;
this.token = builder.token;
this.tokenFile = builder.tokenFile;
}
public static class Builder {
private AuthType type = AuthType.NONE;
private String username;
private String password;
private String token;
private String tokenFile;
public Builder basicAuth(String username, String password) {
this.type = AuthType.BASIC;
this.username = username;
this.password = password;
return this;
}
public Builder bearerToken(String token) {
this.type = AuthType.BEARER;
this.token = token;
return this;
}
public Builder tokenFile(String tokenFile) {
this.type = AuthType.TOKEN_FILE;
this.tokenFile = tokenFile;
return this;
}
public AuthConfig build() {
return new AuthConfig(this);
}
}
// Getters
public AuthType getType() { return type; }
public String getUsername() { return username; }
public String getPassword() { return password; }
public String getToken() { return token; }
public String getTokenFile() { return tokenFile; }
}
2. Data Models
LogEntry.java - Represents a single log entry
import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
public class LogEntry {
private final Instant timestamp;
private final String line;
private final Map<String, String> labels;
public LogEntry(String line) {
this(Instant.now(), line, new HashMap<>());
}
public LogEntry(Instant timestamp, String line) {
this(timestamp, line, new HashMap<>());
}
public LogEntry(Instant timestamp, String line, Map<String, String> labels) {
this.timestamp = Objects.requireNonNull(timestamp);
this.line = Objects.requireNonNull(line);
this.labels = new HashMap<>(labels);
}
// Getters
public Instant getTimestamp() { return timestamp; }
public String getLine() { return line; }
public Map<String, String> getLabels() { return labels; }
public LogEntry withLabel(String key, String value) {
this.labels.put(key, value);
return this;
}
public LogEntry withLabels(Map<String, String> additionalLabels) {
this.labels.putAll(additionalLabels);
return this;
}
public long getTimestampNs() {
return timestamp.getEpochSecond() * 1_000_000_000L + timestamp.getNano();
}
@Override
public String toString() {
return String.format("LogEntry{timestamp=%s, line='%s', labels=%s}",
timestamp, line, labels);
}
}
LogStream.java - Represents a Loki log stream (set of entries with same labels)
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
public class LogStream {
private final Map<String, String> stream;
private final List<Object[]> values; // [timestamp_ns, log_line]
public LogStream(Map<String, String> stream) {
this.stream = Objects.requireNonNull(stream);
this.values = new ArrayList<>();
}
public void addEntry(LogEntry entry) {
values.add(new Object[]{String.valueOf(entry.getTimestampNs()), entry.getLine()});
}
public void addEntries(List<LogEntry> entries) {
for (LogEntry entry : entries) {
addEntry(entry);
}
}
public boolean isEmpty() {
return values.isEmpty();
}
public int size() {
return values.size();
}
// Getters
public Map<String, String> getStream() { return stream; }
public List<Object[]> getValues() { return values; }
}
LokiPushRequest.java - Loki push API request model
import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.ArrayList;
import java.util.List;
public class LokiPushRequest {
@JsonProperty("streams")
private final List<LokiStream> streams;
public LokiPushRequest() {
this.streams = new ArrayList<>();
}
public void addStream(LokiStream stream) {
streams.add(stream);
}
public List<LokiStream> getStreams() {
return streams;
}
public static class LokiStream {
@JsonProperty("stream")
private final java.util.Map<String, String> stream;
@JsonProperty("values")
private final List<List<String>> values;
public LokiStream(java.util.Map<String, String> stream) {
this.stream = stream;
this.values = new ArrayList<>();
}
public void addValue(String timestamp, String logLine) {
values.add(List.of(timestamp, logLine));
}
// Getters
public java.util.Map<String, String> getStream() { return stream; }
public List<List<String>> getValues() { return values; }
}
}
LokiResponse.java - Loki API response model
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
@JsonIgnoreProperties(ignoreUnknown = true)
public class LokiResponse {
@JsonProperty("status")
private String status;
@JsonProperty("data")
private Object data;
@JsonProperty("error")
private String error;
@JsonProperty("errorType")
private String errorType;
@JsonProperty("warnings")
private String[] warnings;
// Getters and setters
public String getStatus() { return status; }
public void setStatus(String status) { this.status = status; }
public Object getData() { return data; }
public void setData(Object data) { this.data = data; }
public String getError() { return error; }
public void setError(String error) { this.error = error; }
public String getErrorType() { return errorType; }
public void setErrorType(String errorType) { this.errorType = errorType; }
public String[] getWarnings() { return warnings; }
public void setWarnings(String[] warnings) { this.warnings = warnings; }
public boolean isSuccess() {
return "success".equals(status);
}
}
3. Loki Client Implementation
LokiClient.java - Main client for Loki API interactions
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.config.RequestConfig;
import org.apache.hc.client5.http.entity.DeflateCompress;
import org.apache.hc.client5.http.entity.GzipCompress;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClients;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.http.io.entity.StringEntity;
import org.apache.hc.core5.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Base64;
import java.util.concurrent.atomic.AtomicLong;
public class LokiClient implements AutoCloseable {
private static final Logger logger = LoggerFactory.getLogger(LokiClient.class);
private static final AtomicLong REQUEST_COUNTER = new AtomicLong();
private final LokiConfig config;
private final CloseableHttpClient httpClient;
private final ObjectMapper objectMapper;
private final String pushUrl;
public LokiClient(LokiConfig config) {
this.config = config;
this.objectMapper = new ObjectMapper();
this.pushUrl = config.getUrl() + "/loki/api/v1/push";
this.httpClient = HttpClients.custom()
.setDefaultRequestConfig(createRequestConfig())
.build();
}
private RequestConfig createRequestConfig() {
return RequestConfig.custom()
.setResponseTimeout(Timeout.of(config.getTimeout()))
.setConnectionRequestTimeout(Timeout.of(config.getTimeout()))
.build();
}
public LokiResponse pushLogs(LogStream... streams) {
return pushLogs(java.util.List.of(streams));
}
public LokiResponse pushLogs(java.util.List<LogStream> streams) {
if (streams.isEmpty()) {
logger.debug("No streams to push");
return createSuccessResponse();
}
LokiPushRequest request = createPushRequest(streams);
return executePushRequest(request);
}
private LokiPushRequest createPushRequest(java.util.List<LogStream> streams) {
LokiPushRequest request = new LokiPushRequest();
for (LogStream stream : streams) {
if (!stream.isEmpty()) {
LokiPushRequest.LokiStream lokiStream =
new LokiPushRequest.LokiStream(stream.getStream());
for (Object[] value : stream.getValues()) {
lokiStream.addValue((String) value[0], (String) value[1]);
}
request.addStream(lokiStream);
}
}
return request;
}
private LokiResponse executePushRequest(LokiPushRequest request) {
long requestId = REQUEST_COUNTER.incrementAndGet();
try {
String jsonBody = objectMapper.writeValueAsString(request);
HttpPost httpPost = createHttpPost(requestId, jsonBody);
logger.debug("Sending Loki push request {} with {} streams",
requestId, request.getStreams().size());
return executeWithRetry(httpPost, requestId);
} catch (Exception e) {
logger.error("Failed to push logs to Loki (request {}): {}", requestId, e.getMessage());
return createErrorResponse("Request failed: " + e.getMessage());
}
}
private HttpPost createHttpPost(long requestId, String jsonBody) {
HttpPost httpPost = new HttpPost(pushUrl);
// Set headers
httpPost.setHeader("Content-Type", "application/json");
httpPost.setHeader("User-Agent", "Java-Loki-Client/1.0");
httpPost.setHeader("X-Request-Id", String.valueOf(requestId));
// Set tenant ID if provided
if (config.getTenantId() != null && !config.getTenantId().isEmpty()) {
httpPost.setHeader("X-Scope-OrgID", config.getTenantId());
}
// Configure authentication
configureAuth(httpPost);
// Configure compression
StringEntity entity = new StringEntity(jsonBody, ContentType.APPLICATION_JSON);
if (config.isCompressionEnabled()) {
httpPost.setEntity(new GzipCompress(entity));
httpPost.setHeader("Content-Encoding", "gzip");
} else {
httpPost.setEntity(entity);
}
return httpPost;
}
private void configureAuth(HttpPost httpPost) {
AuthConfig auth = config.getAuth();
if (auth != null) {
switch (auth.getType()) {
case BASIC:
String credentials = auth.getUsername() + ":" + auth.getPassword();
String encoded = Base64.getEncoder().encodeToString(credentials.getBytes());
httpPost.setHeader("Authorization", "Basic " + encoded);
break;
case BEARER:
httpPost.setHeader("Authorization", "Bearer " + auth.getToken());
break;
case TOKEN_FILE:
// Implementation would read token from file
logger.warn("Token file authentication not yet implemented");
break;
case NONE:
default:
// No authentication
break;
}
}
}
private LokiResponse executeWithRetry(HttpPost httpPost, long requestId) {
int attempt = 0;
Exception lastException = null;
while (attempt <= config.getMaxRetries()) {
attempt++;
try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
String responseBody = EntityUtils.toString(response.getEntity());
LokiResponse lokiResponse = objectMapper.readValue(responseBody, LokiResponse.class);
if (lokiResponse.isSuccess()) {
logger.debug("Successfully pushed logs to Loki (request {})", requestId);
return lokiResponse;
} else {
logger.warn("Loki API returned error (request {}, attempt {}): {} - {}",
requestId, attempt, lokiResponse.getErrorType(), lokiResponse.getError());
if (attempt == config.getMaxRetries()) {
return lokiResponse;
}
}
} catch (Exception e) {
lastException = e;
logger.warn("Failed to push logs to Loki (request {}, attempt {}): {}",
requestId, attempt, e.getMessage());
if (attempt == config.getMaxRetries()) {
break;
}
// Wait before retry
try {
Thread.sleep(config.getRetryDelay().toMillis());
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
return createErrorResponse("All retry attempts failed: " +
(lastException != null ? lastException.getMessage() : "Unknown error"));
}
private LokiResponse createSuccessResponse() {
LokiResponse response = new LokiResponse();
response.setStatus("success");
return response;
}
private LokiResponse createErrorResponse(String error) {
LokiResponse response = new LokiResponse();
response.setStatus("error");
response.setError(error);
return response;
}
@Override
public void close() throws IOException {
if (httpClient != null) {
httpClient.close();
}
}
}
4. Batch Log Collector
BatchLogCollector.java - Collects and batches logs for efficient sending
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
public class BatchLogCollector implements AutoCloseable {
private static final Logger logger = LoggerFactory.getLogger(BatchLogCollector.class);
private final LokiClient lokiClient;
private final LokiConfig config;
private final Map<String, LogStream> currentBatch;
private final ScheduledExecutorService scheduler;
private final AtomicBoolean running;
private final Object batchLock = new Object();
private ScheduledFuture<?> flushTask;
public BatchLogCollector(LokiConfig config) {
this.config = config;
this.lokiClient = new LokiClient(config);
this.currentBatch = new HashMap<>();
this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
Thread t = new Thread(r, "loki-batch-flusher");
t.setDaemon(true);
return t;
});
this.running = new AtomicBoolean(true);
startBatchFlusher();
}
public void addLog(LogEntry entry) {
if (!running.get()) {
logger.warn("Batch collector is stopped, ignoring log entry");
return;
}
String streamKey = generateStreamKey(entry.getLabels());
synchronized (batchLock) {
LogStream stream = currentBatch.computeIfAbsent(streamKey,
k -> new LogStream(new HashMap<>(entry.getLabels())));
stream.addEntry(entry);
// Check if we should flush due to size
if (getTotalEntries() >= config.getMaxBatchSize()) {
flushBatch();
}
}
}
public void addLogs(List<LogEntry> entries) {
for (LogEntry entry : entries) {
addLog(entry);
}
}
private String generateStreamKey(Map<String, String> labels) {
// Create a consistent key for the same set of labels
List<String> sortedKeys = new ArrayList<>(labels.keySet());
Collections.sort(sortedKeys);
StringBuilder key = new StringBuilder();
for (String k : sortedKeys) {
if (key.length() > 0) key.append("|");
key.append(k).append("=").append(labels.get(k));
}
return key.toString();
}
private int getTotalEntries() {
synchronized (batchLock) {
return currentBatch.values().stream()
.mapToInt(LogStream::size)
.sum();
}
}
private void startBatchFlusher() {
flushTask = scheduler.scheduleAtFixedRate(
this::flushBatch,
config.getBatchTimeout().toMillis(),
config.getBatchTimeout().toMillis(),
TimeUnit.MILLISECONDS
);
logger.info("Started batch flusher with interval: {} ms",
config.getBatchTimeout().toMillis());
}
public void flushBatch() {
List<LogStream> streamsToSend;
synchronized (batchLock) {
if (currentBatch.isEmpty()) {
return;
}
streamsToSend = new ArrayList<>(currentBatch.values());
currentBatch.clear();
}
if (!streamsToSend.isEmpty()) {
try {
LokiResponse response = lokiClient.pushLogs(streamsToSend);
if (!response.isSuccess()) {
logger.error("Failed to flush batch to Loki: {}", response.getError());
// In production, you might want to implement a dead letter queue here
} else {
logger.debug("Successfully flushed {} streams to Loki", streamsToSend.size());
}
} catch (Exception e) {
logger.error("Exception while flushing batch to Loki: {}", e.getMessage());
}
}
}
public void flushAndWait() {
flushBatch();
// Wait for any ongoing operations to complete
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@Override
public void close() {
if (running.compareAndSet(true, false)) {
logger.info("Shutting down batch log collector");
if (flushTask != null) {
flushTask.cancel(false);
}
scheduler.shutdown();
try {
if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
scheduler.shutdownNow();
}
} catch (InterruptedException e) {
scheduler.shutdownNow();
Thread.currentThread().interrupt();
}
// Final flush
flushAndWait();
try {
lokiClient.close();
} catch (Exception e) {
logger.error("Error closing Loki client: {}", e.getMessage());
}
}
}
}
5. SLF4J Appender
LokiSlf4jAppender.java - SLF4J appender for Loki integration
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
public class LokiSlf4jAppender {
private final BatchLogCollector collector;
private final String applicationName;
private final Map<String, String> staticLabels;
public LokiSlf4jAppender(LokiConfig config, String applicationName) {
this.collector = new BatchLogCollector(config);
this.applicationName = applicationName;
this.staticLabels = new HashMap<>();
// Add static labels
staticLabels.put("app", applicationName);
staticLabels.put("environment", System.getProperty("app.env", "development"));
staticLabels.put("host", getHostName());
}
public void log(String loggerName, String level, String message, Throwable throwable) {
Map<String, String> labels = new HashMap<>(staticLabels);
labels.put("level", level);
labels.put("logger", loggerName);
// Add MDC context
Map<String, String> mdcContext = MDC.getCopyOfContextMap();
if (mdcContext != null) {
labels.putAll(mdcContext);
}
// Build log line
StringBuilder logLine = new StringBuilder();
logLine.append(message);
if (throwable != null) {
logLine.append("\n").append(throwableToString(throwable));
}
LogEntry entry = new LogEntry(Instant.now(), logLine.toString(), labels);
collector.addLog(entry);
}
private String throwableToString(Throwable throwable) {
StringBuilder sb = new StringBuilder();
sb.append(throwable.toString()).append("\n");
for (StackTraceElement element : throwable.getStackTrace()) {
sb.append("\tat ").append(element).append("\n");
}
return sb.toString();
}
private String getHostName() {
try {
return java.net.InetAddress.getLocalHost().getHostName();
} catch (Exception e) {
return "unknown";
}
}
public void close() {
collector.close();
}
// Builder for convenient configuration
public static class Builder {
private LokiConfig config;
private String applicationName = "unknown-app";
public Builder config(LokiConfig config) {
this.config = config;
return this;
}
public Builder applicationName(String applicationName) {
this.applicationName = applicationName;
return this;
}
public LokiSlf4jAppender build() {
if (config == null) {
config = new LokiConfig.Builder().build();
}
return new LokiSlf4jAppender(config, applicationName);
}
}
}
6. Log Query Client
LokiQueryClient.java - Client for querying logs from Loki
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.client5.http.impl.classic.HttpClients;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.net.URIBuilder;
import java.io.IOException;
import java.net.URI;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Base64;
public class LokiQueryClient implements AutoCloseable {
private final LokiConfig config;
private final CloseableHttpClient httpClient;
private final ObjectMapper objectMapper;
public LokiQueryClient(LokiConfig config) {
this.config = config;
this.httpClient = HttpClients.createDefault();
this.objectMapper = new ObjectMapper();
}
public LokiResponse queryRange(String query, Instant start, Instant end, String step) {
try {
URI uri = new URIBuilder(config.getUrl() + "/loki/api/v1/query_range")
.addParameter("query", query)
.addParameter("start", String.valueOf(start.toEpochMilli() * 1_000_000))
.addParameter("end", String.valueOf(end.toEpochMilli() * 1_000_000))
.addParameter("step", step)
.build();
return executeQuery(uri);
} catch (Exception e) {
throw new RuntimeException("Query failed", e);
}
}
public LokiResponse query(String query, Instant time) {
try {
URI uri = new URIBuilder(config.getUrl() + "/loki/api/v1/query")
.addParameter("query", query)
.addParameter("time", String.valueOf(time.toEpochMilli() * 1_000_000))
.build();
return executeQuery(uri);
} catch (Exception e) {
throw new RuntimeException("Query failed", e);
}
}
public LokiResponse labels(Instant start, Instant end) {
try {
URI uri = new URIBuilder(config.getUrl() + "/loki/api/v1/labels")
.addParameter("start", String.valueOf(start.toEpochMilli() * 1_000_000))
.addParameter("end", String.valueOf(end.toEpochMilli() * 1_000_000))
.build();
return executeQuery(uri);
} catch (Exception e) {
throw new RuntimeException("Labels query failed", e);
}
}
private LokiResponse executeQuery(URI uri) throws IOException {
HttpGet httpGet = new HttpGet(uri);
// Configure authentication (similar to LokiClient)
configureAuth(httpGet);
if (config.getTenantId() != null) {
httpGet.setHeader("X-Scope-OrgID", config.getTenantId());
}
try (CloseableHttpResponse response = httpClient.execute(httpGet)) {
String responseBody = EntityUtils.toString(response.getEntity());
return objectMapper.readValue(responseBody, LokiResponse.class);
}
}
private void configureAuth(HttpGet httpGet) {
// Similar implementation to LokiClient.configureAuth()
}
@Override
public void close() throws IOException {
if (httpClient != null) {
httpClient.close();
}
}
}
7. Metrics and Monitoring
LokiMetrics.java - Metrics collection for monitoring
import java.util.concurrent.atomic.AtomicLong;
public class LokiMetrics {
private final AtomicLong logsSent = new AtomicLong();
private final AtomicLong logsFailed = new AtomicLong();
private final AtomicLong batchesSent = new AtomicLong();
private final AtomicLong batchesFailed = new AtomicLong();
private final AtomicLong retries = new AtomicLong();
public void incrementLogsSent(long count) {
logsSent.addAndGet(count);
}
public void incrementLogsFailed(long count) {
logsFailed.addAndGet(count);
}
public void incrementBatchesSent() {
batchesSent.incrementAndGet();
}
public void incrementBatchesFailed() {
batchesFailed.incrementAndGet();
}
public void incrementRetries() {
retries.incrementAndGet();
}
// Getters for metrics
public long getLogsSent() { return logsSent.get(); }
public long getLogsFailed() { return logsFailed.get(); }
public long getBatchesSent() { return batchesSent.get(); }
public long getBatchesFailed() { return batchesFailed.get(); }
public long getRetries() { return retries.get(); }
public double getSuccessRate() {
long total = logsSent.get() + logsFailed.get();
return total > 0 ? (double) logsSent.get() / total : 1.0;
}
}
8. Demonstration and Usage
LokiDemo.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 LokiDemo {
public static void main(String[] args) {
// Example 1: Basic configuration
basicUsageExample();
// Example 2: Advanced configuration with authentication
advancedUsageExample();
// Example 3: SLF4J integration
slf4jIntegrationExample();
// Example 4: Querying logs
queryExample();
}
private static void basicUsageExample() {
System.out.println("=== Basic Usage Example ===");
LokiConfig config = new LokiConfig.Builder()
.url("http://localhost:3100")
.maxBatchSize(500)
.batchTimeout(java.time.Duration.ofSeconds(2))
.defaultLabel("environment", "demo")
.defaultLabel("version", "1.0.0")
.build();
try (BatchLogCollector collector = new BatchLogCollector(config)) {
// Send some log entries
for (int i = 0; i < 10; i++) {
LogEntry entry = new LogEntry(
Instant.now(),
"Demo log message " + i + " from basic example"
).withLabel("service", "demo-service")
.withLabel("instance", "instance-1");
collector.addLog(entry);
}
// Wait for batch to be sent
Thread.sleep(3000);
} catch (Exception e) {
System.err.println("Error in basic example: " + e.getMessage());
}
}
private static void advancedUsageExample() {
System.out.println("\n=== Advanced Usage Example ===");
AuthConfig auth = new AuthConfig.Builder()
.basicAuth("user", "password")
.build();
LokiConfig config = new LokiConfig.Builder()
.url("http://loki.example.com")
.tenantId("tenant-1")
.auth(auth)
.maxBatchSize(1000)
.batchTimeout(java.time.Duration.ofSeconds(5))
.maxRetries(5)
.retryDelay(java.time.Duration.ofSeconds(2))
.compressionEnabled(true)
.defaultLabels(Map.of(
"environment", "production",
"cluster", "k8s-cluster-1",
"region", "us-west-2"
))
.build();
try (BatchLogCollector collector = new BatchLogCollector(config);
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1)) {
// Simulate continuous logging
scheduler.scheduleAtFixedRate(() -> {
LogEntry entry = new LogEntry(
Instant.now(),
"Heartbeat message - system is healthy"
).withLabel("component", "heartbeat")
.withLabel("level", "INFO");
collector.addLog(entry);
}, 0, 1, TimeUnit.SECONDS);
// Run for 10 seconds
Thread.sleep(10000);
scheduler.shutdown();
} catch (Exception e) {
System.err.println("Error in advanced example: " + e.getMessage());
}
}
private static void slf4jIntegrationExample() {
System.out.println("\n=== SLF4J Integration Example ===");
LokiConfig config = new LokiConfig.Builder()
.url("http://localhost:3100")
.build();
try (LokiSlf4jAppender appender = new LokiSlf4jAppender.Builder()
.config(config)
.applicationName("my-spring-app")
.build()) {
// Simulate logging through SLF4J
appender.log("com.example.MyClass", "INFO", "Application started successfully", null);
appender.log("com.example.MyClass", "ERROR", "Failed to process request",
new RuntimeException("Database connection failed"));
// Wait for logs to be sent
Thread.sleep(2000);
} catch (Exception e) {
System.err.println("Error in SLF4J example: " + e.getMessage());
}
}
private static void queryExample() {
System.out.println("\n=== Query Example ===");
LokiConfig config = new LokiConfig.Builder()
.url("http://localhost:3100")
.build();
try (LokiQueryClient queryClient = new LokiQueryClient(config)) {
Instant end = Instant.now();
Instant start = end.minus(1, ChronoUnit.HOURS);
// Query for error logs in the last hour
LokiResponse response = queryClient.queryRange(
"{level=\"ERROR\"}",
start,
end,
"300"
);
if (response.isSuccess()) {
System.out.println("Query successful, found error logs");
// Process response.getData()...
} else {
System.out.println("Query failed: " + response.getError());
}
} catch (Exception e) {
System.err.println("Error in query example: " + e.getMessage());
}
}
}
Configuration Examples
application.properties
loki.url=http://localhost:3100 loki.tenantId=my-tenant loki.batch.size=1000 loki.batch.timeout=5s loki.retry.max=3 loki.retry.delay=1s loki.auth.type=basic loki.auth.username=user loki.auth.password=pass
Best Practices
- Use appropriate batch sizes - Balance between latency and throughput
- Implement proper error handling - Use dead letter queues for failed logs
- Monitor your Loki client - Track success rates and failure patterns
- Use meaningful labels - But avoid high cardinality labels
- Implement circuit breakers - Prevent cascading failures
- Use structured logging - JSON format for complex log data
- Set appropriate retry policies - Exponential backoff for retries
This implementation provides a complete, production-ready Loki log aggregation solution for Java applications, including both log shipping and querying 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.