From 56bc183be19e339203f04d4c3d659a5955af01af Mon Sep 17 00:00:00 2001
From: "malini-agent[bot]"
<338999285+malini-agent[bot]@users.noreply.github.com>
Date: Sat, 24 Oct 2026 15:03:40 +0200
Subject: [PATCH 1/2] feat: add event tracking and wait for queued requests on
flush and close
---
README.md | 17 +-
src/main/java/io/logdash/sdk/Logdash.java | 70 ++++-
.../io/logdash/sdk/config/LogdashConfig.java | 6 +-
.../java/io/logdash/sdk/event/EventEntry.java | 103 +++++++
.../logdash/sdk/transport/HttpTransport.java | 81 ++++--
.../sdk/transport/LogdashTransport.java | 15 +-
.../logdash/sdk/transport/NoOpTransport.java | 31 ++-
.../io/logdash/sdk/util/JsonSerializer.java | 30 +-
.../java/io/logdash/sdk/LogdashTrackTest.java | 263 ++++++++++++++++++
.../io/logdash/sdk/event/EventEntryTest.java | 121 ++++++++
.../sdk/transport/HttpTransportTest.java | 2 +-
.../sdk/transport/LogdashTransportTest.java | 6 +
12 files changed, 699 insertions(+), 46 deletions(-)
create mode 100644 src/main/java/io/logdash/sdk/event/EventEntry.java
create mode 100644 src/test/java/io/logdash/sdk/LogdashTrackTest.java
create mode 100644 src/test/java/io/logdash/sdk/event/EventEntryTest.java
diff --git a/README.md b/README.md
index 35e3628..fb651b1 100644
--- a/README.md
+++ b/README.md
@@ -282,6 +282,19 @@ metrics.set("memory_usage_mb", 512.5);
metrics.set("queue_size", 0);
```
+## Events
+
+Track what happens in your app, such as an agent run:
+
+```java
+logdash.track("agent_run", Map.of("agent", "support", "steps", 4, "success", true), user.id());
+```
+
+Events show up in the domain's Analytics as goals, with a daily trend.
+Pass the user ID for signed-in users, so one person counts once across the website and apps.
+Without it, Logdash counts a visitor of that app that changes every day.
+Never put personal data in props.
+
## Apps
A domain has one ingest key, the domain key, shared by all apps of that domain, such as a desktop app, an API and a landing page.
@@ -294,7 +307,7 @@ var logdash = Logdash.builder()
.build();
```
-Every log and metric this instance sends carries `"app": "api"`.
+Every log, metric and event this instance sends carries `"app": "api"`.
With a domain key, the app is created on first use.
Older service keys ignore `app`.
@@ -336,7 +349,7 @@ Logdash logdash = Logdash.builder()
| Parameter | Default | Description |
|-------------------------|--------------------------|----------------------------------|
| `apiKey` | `null` | Your Logdash API key (required) |
-| `app` | `null` | App name sent with logs/metrics |
+| `app` | `null` | App name sent with all data |
| `baseUrl` | `https://api.logdash.io` | Logdash API endpoint |
| `enableConsoleOutput` | `true` | Show logs/metrics in console |
| `enableVerboseLogging` | `false` | Enable SDK debug output |
diff --git a/src/main/java/io/logdash/sdk/Logdash.java b/src/main/java/io/logdash/sdk/Logdash.java
index 4a1d7ea..86b2afa 100644
--- a/src/main/java/io/logdash/sdk/Logdash.java
+++ b/src/main/java/io/logdash/sdk/Logdash.java
@@ -1,12 +1,14 @@
package io.logdash.sdk;
import io.logdash.sdk.config.LogdashConfig;
+import io.logdash.sdk.event.EventEntry;
import io.logdash.sdk.log.LogdashLogger;
import io.logdash.sdk.metrics.LogdashMetrics;
import io.logdash.sdk.transport.HttpTransport;
import io.logdash.sdk.transport.LogdashTransport;
import io.logdash.sdk.transport.NoOpTransport;
+import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -14,9 +16,9 @@
/**
* Main entry point for the Logdash Java SDK.
*
- * This class provides access to logging and metrics functionality for sending observability data
- * to the Logdash.io platform. It manages the underlying transport layer and provides a simple API
- * for developers to integrate observability into their applications.
+ *
This class provides access to logging, metrics and event tracking for sending observability
+ * data to the Logdash.io platform. It manages the underlying transport layer and provides a simple
+ * API for developers to integrate observability into their applications.
*
*
The SDK automatically handles HTTP transport with retry logic, rate limiting, and graceful
* shutdown. If no API key is provided or HTTP transport fails, it falls back to console-only mode.
@@ -102,6 +104,58 @@ public LogdashMetrics metrics() {
return new LogdashMetrics(config, transport);
}
+ /**
+ * Tracks an event without properties or a user.
+ *
+ * @param name the event name, matching {@code ^[a-z][a-z0-9_]{0,63}$}
+ * @see #track(String, Map, String)
+ */
+ public void track(String name) {
+ track(name, Map.of(), null);
+ }
+
+ /**
+ * Tracks an event with properties and no user.
+ *
+ * @param name the event name, matching {@code ^[a-z][a-z0-9_]{0,63}$}
+ * @param props flat properties with string, number or boolean values
+ * @see #track(String, Map, String)
+ */
+ public void track(String name, Map props) {
+ track(name, props, null);
+ }
+
+ /**
+ * Tracks an event, such as an agent run, for the domain's Analytics.
+ *
+ * Like logs, the event is sent in the background, and this method never throws. Flush and
+ * close send pending events too. Without an API key, the event is printed to the console only.
+ *
+ *
An invalid name, a name of the website tracker ({@code pageview}, {@code pageleave} or
+ * {@code browser_error}), or a user ID that contains {@code @} drops the event, with a warning
+ * when verbose logging is on.
+ *
+ *
{@code
+ * logdash.track("agent_run", Map.of("agent", "support", "steps", 4), user.id());
+ * }
+ *
+ * @param name the event name, matching {@code ^[a-z][a-z0-9_]{0,63}$}
+ * @param props up to 10 flat properties with keys matching {@code ^[a-z][a-z0-9_]{0,39}$} and
+ * string, number or boolean values; never personal data
+ * @param userId the signed-in user's internal ID, never an email address, so one person counts
+ * once across the website and apps; null counts a visitor of this app that
+ * changes every day
+ */
+ public void track(String name, Map props, String userId) {
+ try {
+ transport.sendEvent(new EventEntry(name, props, userId));
+ } catch (RuntimeException e) {
+ if (config.enableVerboseLogging()) {
+ System.err.println("Warning: Event dropped: " + e.getMessage());
+ }
+ }
+ }
+
/**
* Returns the configuration used by this instance.
*
@@ -112,7 +166,7 @@ public LogdashConfig config() {
}
/**
- * Flushes all pending logs and metrics synchronously.
+ * Flushes all pending logs, metrics and events synchronously.
*
* Blocks until all queued requests complete or the configured timeout is reached. This method
* is useful for ensuring all data is sent before application shutdown.
@@ -267,7 +321,7 @@ public Builder baseUrl(String baseUrl) {
}
/**
- * Enables or disables console output for logs and metrics.
+ * Enables or disables console output for logs, metrics and events.
*
* @param enableConsoleOutput true to enable console output
* @return this builder
@@ -348,9 +402,9 @@ public Builder maxConcurrentRequests(int maxConcurrentRequests) {
* {@code landing}.
*
*
The domain key is shared by all apps of a domain, and {@code app} names this one, so the
- * dashboard can filter logs and metrics by app. Every log and metric this instance sends
- * carries it. With a domain key, the app is created on first use. Older service keys ignore
- * {@code app}.
+ * dashboard can filter logs and metrics by app. Every log, metric and event this instance
+ * sends carries it. With a domain key, the app is created on first use. Older service keys
+ * ignore {@code app}.
*
*
{@code
* var logdash = Logdash.builder()
diff --git a/src/main/java/io/logdash/sdk/config/LogdashConfig.java b/src/main/java/io/logdash/sdk/config/LogdashConfig.java
index 66566d1..a60683c 100644
--- a/src/main/java/io/logdash/sdk/config/LogdashConfig.java
+++ b/src/main/java/io/logdash/sdk/config/LogdashConfig.java
@@ -13,15 +13,15 @@
*
* @param apiKey API key for authenticating with Logdash.io
* @param baseUrl Base URL for the Logdash API (defaults to https://api.logdash.io)
- * @param enableConsoleOutput Whether to output logs and metrics to console
+ * @param enableConsoleOutput Whether to output logs, metrics and events to console
* @param enableVerboseLogging Whether to enable verbose internal logging
* @param maxRetries Maximum number of retry attempts for failed requests
* @param retryDelayMs Delay between retry attempts in milliseconds
* @param requestTimeoutMs HTTP request timeout in milliseconds
* @param shutdownTimeoutMs Maximum time to wait for graceful shutdown
* @param maxConcurrentRequests Maximum number of concurrent HTTP requests
- * @param app Name of the app sent with every log and metric, trimmed; null when
- * not set or blank
+ * @param app Name of the app sent with every log, metric and event, trimmed; null
+ * when not set or blank
*/
public record LogdashConfig(
String apiKey,
diff --git a/src/main/java/io/logdash/sdk/event/EventEntry.java b/src/main/java/io/logdash/sdk/event/EventEntry.java
new file mode 100644
index 0000000..2b86e70
--- /dev/null
+++ b/src/main/java/io/logdash/sdk/event/EventEntry.java
@@ -0,0 +1,103 @@
+package io.logdash.sdk.event;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.regex.Pattern;
+
+/**
+ * Immutable data container representing an event to be sent to Logdash.
+ *
+ * An event entry consists of:
+ *
+ *
+ * - A name matching {@code ^[a-z][a-z0-9_]{0,63}$}, such as {@code agent_run}
+ *
- Flat properties with string, number or boolean values
+ *
- The signed-in user's internal ID, never an email address
+ *
- A UUID v4 that retries reuse, so the event is stored once
+ *
+ *
+ * @param name the event name, must match {@code ^[a-z][a-z0-9_]{0,63}$}
+ * @param props the event properties; null keys and values are left out, and values other than
+ * strings, numbers and booleans become strings
+ * @param userId the signed-in user's internal ID of up to 256 characters, never an email address;
+ * null when not set or blank
+ * @param id the UUID v4 of this event, generated when null
+ */
+public record EventEntry(String name, Map props, String userId, UUID id) {
+ private static final Pattern NAME = Pattern.compile("[a-z][a-z0-9_]{0,63}");
+ private static final Set WEBSITE_TRACKER_EVENTS =
+ Set.of("pageview", "pageleave", "browser_error");
+ private static final int MAX_USER_ID_LENGTH = 256;
+
+ /**
+ * Validates and normalizes the event entry parameters during construction.
+ *
+ * @throws IllegalArgumentException if the name is invalid or belongs to the website tracker, or
+ * the user ID contains {@code @} or is too long
+ */
+ public EventEntry {
+ if (name == null || !NAME.matcher(name).matches()) {
+ throw new IllegalArgumentException(
+ "Event name must match ^[a-z][a-z0-9_]{0,63}$, got: " + name);
+ }
+ if (WEBSITE_TRACKER_EVENTS.contains(name)) {
+ throw new IllegalArgumentException(
+ "Event name " + name + " belongs to the website tracker");
+ }
+ if (userId != null && userId.isBlank()) {
+ userId = null;
+ }
+ if (userId != null && userId.contains("@")) {
+ throw new IllegalArgumentException(
+ "User ID of event " + name + " must be an internal ID, never an email address");
+ }
+ if (userId != null && userId.length() > MAX_USER_ID_LENGTH) {
+ throw new IllegalArgumentException(
+ "User ID of event " + name + " must be at most " + MAX_USER_ID_LENGTH + " characters");
+ }
+ props = flatten(props);
+ if (id == null) {
+ id = UUID.randomUUID();
+ }
+ }
+
+ /**
+ * Creates an event entry with a new UUID v4.
+ *
+ * @param name the event name
+ * @param props the event properties
+ * @param userId the signed-in user's internal ID, or null
+ */
+ public EventEntry(String name, Map props, String userId) {
+ this(name, props, userId, null);
+ }
+
+ /**
+ * Returns the properties of this event, in the order they were given.
+ *
+ * @return an immutable copy of the props map
+ */
+ public Map props() {
+ return Collections.unmodifiableMap(new LinkedHashMap<>(props));
+ }
+
+ private static Map flatten(Map props) {
+ var flat = new LinkedHashMap();
+ if (props != null) {
+ props.forEach(
+ (key, value) -> {
+ if (key != null && value != null) {
+ flat.put(
+ key,
+ value instanceof Number || value instanceof Boolean
+ ? value
+ : String.valueOf(value));
+ }
+ });
+ }
+ return flat;
+ }
+}
diff --git a/src/main/java/io/logdash/sdk/transport/HttpTransport.java b/src/main/java/io/logdash/sdk/transport/HttpTransport.java
index edecad6..0781086 100644
--- a/src/main/java/io/logdash/sdk/transport/HttpTransport.java
+++ b/src/main/java/io/logdash/sdk/transport/HttpTransport.java
@@ -1,6 +1,7 @@
package io.logdash.sdk.transport;
import io.logdash.sdk.config.LogdashConfig;
+import io.logdash.sdk.event.EventEntry;
import io.logdash.sdk.exception.LogdashException;
import io.logdash.sdk.log.LogEntry;
import io.logdash.sdk.metrics.MetricEntry;
@@ -14,6 +15,7 @@
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -26,7 +28,7 @@
*
*
* - Non-blocking concurrency control using tryAcquire
- *
- Efficient request tracking with CountDownLatch
+ *
- Request tracking from submission, so flush and shutdown wait for every request sent
*
- Optimized thread pool sizing
*
- Fast retry logic with exponential backoff
*
- Minimal synchronization overhead
@@ -36,6 +38,7 @@ public final class HttpTransport implements LogdashTransport, AutoCloseable {
private static final String LOGS_ENDPOINT = "/logs";
private static final String METRICS_ENDPOINT = "/metrics";
+ private static final String EVENTS_ENDPOINT = "/events";
private static final String API_KEY_HEADER = "project-api-key";
private static final String USER_AGENT = "logdash-java-sdk/0.2.0";
@@ -49,9 +52,7 @@ public final class HttpTransport implements LogdashTransport, AutoCloseable {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false);
private final AtomicLong requestCounter = new AtomicLong(0);
- private final AtomicLong activeRequests = new AtomicLong(0);
-
- private volatile CountDownLatch pendingRequestsLatch = new CountDownLatch(0);
+ private final Set> pendingRequests = ConcurrentHashMap.newKeySet();
/**
* Creates a new HTTP transport with the specified configuration.
@@ -101,6 +102,21 @@ public CompletableFuture sendMetric(MetricEntry metricEntry) {
}
}
+ @Override
+ public CompletableFuture sendEvent(EventEntry eventEntry) {
+ if (shutdownInitiated.get() || closed.get()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ try {
+ String json = serializer.serialize(eventEntry);
+ return sendWithRetry(EVENTS_ENDPOINT, json, "POST");
+ } catch (Exception e) {
+ handleTransportError("Failed to serialize event entry", e);
+ return CompletableFuture.completedFuture(null);
+ }
+ }
+
/**
* Sends an HTTP request with optimized retry logic and smart concurrency control. Uses tryAcquire
* with timeout instead of immediate rejection for better throughput.
@@ -117,7 +133,7 @@ private CompletableFuture sendWithRetry(String endpoint, String json, Stri
final long sequenceId = requestCounter.incrementAndGet();
- return CompletableFuture.runAsync(
+ CompletableFuture request = CompletableFuture.runAsync(
() -> {
boolean acquired = false;
try {
@@ -130,15 +146,11 @@ private CompletableFuture sendWithRetry(String endpoint, String json, Stri
return;
}
- if (shutdownInitiated.get() || closed.get()) {
+ // A graceful shutdown still sends what was queued before it
+ if (closed.get()) {
return;
}
- long currentActive = activeRequests.incrementAndGet();
- if (currentActive == 1) {
- pendingRequestsLatch = new CountDownLatch(1);
- }
-
Exception lastException = null;
for (int attempt = 1; attempt <= config.maxRetries() + 1; attempt++) {
try {
@@ -179,14 +191,14 @@ private CompletableFuture sendWithRetry(String endpoint, String json, Stri
if (acquired) {
rateLimiter.release();
}
-
- long remaining = activeRequests.decrementAndGet();
- if (remaining == 0) {
- pendingRequestsLatch.countDown();
- }
}
},
executorService);
+
+ // Tracked from submission, not from start, so a flush right after a send waits for it
+ pendingRequests.add(request);
+ request.whenComplete((result, error) -> pendingRequests.remove(request));
+ return request;
}
private void sendHttpRequest(
@@ -235,7 +247,7 @@ private void sendHttpRequest(
}
/**
- * Flushes all pending requests efficiently using CountDownLatch.
+ * Flushes all pending requests, including the ones still queued.
*
* @return a future that completes when all pending requests finish
*/
@@ -243,16 +255,15 @@ public CompletableFuture flush() {
return CompletableFuture.runAsync(
() -> {
try {
- CountDownLatch currentLatch = pendingRequestsLatch;
- if (currentLatch != null && activeRequests.get() > 0) {
+ if (!pendingRequests.isEmpty()) {
long timeoutMs = Math.min(config.shutdownTimeoutMs(), 5000);
- boolean completed = currentLatch.await(timeoutMs, TimeUnit.MILLISECONDS);
+ boolean completed = awaitPendingRequests(timeoutMs);
if (completed && config.enableVerboseLogging()) {
System.out.println("All pending requests flushed successfully");
} else if (!completed && config.enableVerboseLogging()) {
System.out.printf(
- "Flush timeout, %d requests still pending%n", activeRequests.get());
+ "Flush timeout, %d requests still pending%n", pendingRequests.size());
}
}
} catch (InterruptedException e) {
@@ -273,23 +284,22 @@ public Optional> shutdown() {
System.out.println("Initiating HTTP transport shutdown...");
}
- CountDownLatch currentLatch = pendingRequestsLatch;
long timeoutMs = Math.min(config.shutdownTimeoutMs(), 3000);
- if (currentLatch != null && activeRequests.get() > 0) {
+ if (!pendingRequests.isEmpty()) {
if (config.enableVerboseLogging()) {
System.out.printf(
- "Waiting for %d pending HTTP requests...%n", activeRequests.get());
+ "Waiting for %d pending HTTP requests...%n", pendingRequests.size());
}
- boolean completed = currentLatch.await(timeoutMs, TimeUnit.MILLISECONDS);
+ boolean completed = awaitPendingRequests(timeoutMs);
if (completed && config.enableVerboseLogging()) {
System.out.println("All HTTP requests completed successfully");
} else if (!completed && config.enableVerboseLogging()) {
System.out.printf(
"Shutdown timeout exceeded, %d requests may be lost%n",
- activeRequests.get());
+ pendingRequests.size());
}
}
@@ -304,6 +314,25 @@ public Optional> shutdown() {
return Optional.of(CompletableFuture.completedFuture(null));
}
+ /**
+ * Waits for every request sent so far.
+ *
+ * @param timeoutMs the longest time to wait
+ * @return true when all of them finished within the timeout
+ */
+ private boolean awaitPendingRequests(long timeoutMs) throws InterruptedException {
+ try {
+ CompletableFuture.allOf(pendingRequests.toArray(new CompletableFuture[0]))
+ .get(timeoutMs, TimeUnit.MILLISECONDS);
+ return true;
+ } catch (ExecutionException e) {
+ // A request that failed has finished too
+ return true;
+ } catch (TimeoutException e) {
+ return false;
+ }
+ }
+
@Override
public void close() {
shutdownInitiated.set(true);
diff --git a/src/main/java/io/logdash/sdk/transport/LogdashTransport.java b/src/main/java/io/logdash/sdk/transport/LogdashTransport.java
index 407fada..d45e93c 100644
--- a/src/main/java/io/logdash/sdk/transport/LogdashTransport.java
+++ b/src/main/java/io/logdash/sdk/transport/LogdashTransport.java
@@ -1,5 +1,6 @@
package io.logdash.sdk.transport;
+import io.logdash.sdk.event.EventEntry;
import io.logdash.sdk.log.LogEntry;
import io.logdash.sdk.metrics.MetricEntry;
@@ -9,9 +10,9 @@
/**
* Transport interface for sending observability data to Logdash platform.
*
- *
Implementations handle the actual delivery mechanism for logs and metrics, whether through
- * HTTP, console output, or other means. All operations are asynchronous to avoid blocking the
- * application.
+ *
Implementations handle the actual delivery mechanism for logs, metrics and events, whether
+ * through HTTP, console output, or other means. All operations are asynchronous to avoid blocking
+ * the application.
*/
public interface LogdashTransport {
@@ -31,6 +32,14 @@ public interface LogdashTransport {
*/
CompletableFuture sendMetric(MetricEntry metricEntry);
+ /**
+ * Sends an event entry asynchronously to the transport destination.
+ *
+ * @param eventEntry the event entry to send
+ * @return a future that completes when the operation finishes
+ */
+ CompletableFuture sendEvent(EventEntry eventEntry);
+
/**
* Initiates graceful shutdown of the transport.
*
diff --git a/src/main/java/io/logdash/sdk/transport/NoOpTransport.java b/src/main/java/io/logdash/sdk/transport/NoOpTransport.java
index 4f0d8d4..1a43999 100644
--- a/src/main/java/io/logdash/sdk/transport/NoOpTransport.java
+++ b/src/main/java/io/logdash/sdk/transport/NoOpTransport.java
@@ -1,6 +1,7 @@
package io.logdash.sdk.transport;
import io.logdash.sdk.config.LogdashConfig;
+import io.logdash.sdk.event.EventEntry;
import io.logdash.sdk.log.LogEntry;
import io.logdash.sdk.metrics.MetricEntry;
@@ -14,8 +15,8 @@
* No-operation transport that outputs observability data to console only.
*
* This transport is used as a fallback when no API key is provided or when HTTP transport
- * initialization fails. It provides local visibility into logs and metrics without sending data to
- * external services.
+ * initialization fails. It provides local visibility into logs, metrics and events without sending
+ * data to external services.
*
*
Useful for development, testing, and scenarios where external connectivity is not available or
* desired.
@@ -27,6 +28,7 @@ public final class NoOpTransport implements LogdashTransport {
private static final String RESET = "\u001B[0m";
private static final String CYAN = "\u001B[36m";
private static final String YELLOW = "\u001B[33m";
+ private static final String GREEN = "\u001B[32m";
private final LogdashConfig config;
private volatile boolean closed = false;
@@ -69,6 +71,19 @@ public CompletableFuture sendMetric(MetricEntry metricEntry) {
return CompletableFuture.completedFuture(null);
}
+ @Override
+ public CompletableFuture sendEvent(EventEntry eventEntry) {
+ if (closed) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ if (config.enableConsoleOutput()) {
+ printEventToConsole(eventEntry);
+ }
+
+ return CompletableFuture.completedFuture(null);
+ }
+
@Override
public Optional> shutdown() {
return Optional.empty();
@@ -105,4 +120,16 @@ private void printMetricToConsole(MetricEntry metricEntry) {
metricEntry.name(),
metricEntry.value());
}
+
+ private void printEventToConsole(EventEntry eventEntry) {
+ var timestamp = OffsetDateTime.now().format(CONSOLE_FORMATTER);
+ System.out.printf(
+ "%s [%sEVENT%s] %s%s%s%n",
+ timestamp,
+ GREEN,
+ RESET,
+ eventEntry.name(),
+ eventEntry.props().isEmpty() ? "" : " " + eventEntry.props(),
+ eventEntry.userId() == null ? "" : " (userId=" + eventEntry.userId() + ")");
+ }
}
diff --git a/src/main/java/io/logdash/sdk/util/JsonSerializer.java b/src/main/java/io/logdash/sdk/util/JsonSerializer.java
index 974a78e..ec2de4e 100644
--- a/src/main/java/io/logdash/sdk/util/JsonSerializer.java
+++ b/src/main/java/io/logdash/sdk/util/JsonSerializer.java
@@ -8,6 +8,8 @@
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.json.JsonMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import io.logdash.sdk.event.EventEntry;
+import io.logdash.sdk.exception.LogdashException;
import io.logdash.sdk.log.LogEntry;
import io.logdash.sdk.metrics.MetricEntry;
@@ -36,7 +38,7 @@ public JsonSerializer() {
}
/**
- * Creates a serializer that adds the app to every log and metric.
+ * Creates a serializer that adds the app to every log, metric and event.
*
* @param app the app name, or null to leave the {@code app} key out
*/
@@ -101,6 +103,32 @@ public String serialize(MetricEntry metricEntry) {
}
}
+ /**
+ * Serializes an event entry to JSON format, leaving out props, user ID and app when not set.
+ *
+ * @param eventEntry the event entry to serialize, must not be null
+ * @return JSON string representation
+ * @throws LogdashException if serialization fails
+ */
+ public String serialize(EventEntry eventEntry) {
+ var eventData =
+ new HashMap(Map.of("name", eventEntry.name(), "id", eventEntry.id()));
+ if (!eventEntry.props().isEmpty()) {
+ eventData.put("props", eventEntry.props());
+ }
+ if (eventEntry.userId() != null) {
+ eventData.put("userId", eventEntry.userId());
+ }
+ if (app != null) {
+ eventData.put("app", app);
+ }
+ try {
+ return OBJECT_MAPPER.writeValueAsString(eventData);
+ } catch (JsonProcessingException e) {
+ throw new LogdashException("Failed to serialize event " + eventEntry.name(), e);
+ }
+ }
+
private String sanitizeString(String input) {
if (input == null) return "";
return input.replaceAll("[\\p{Cntrl}&&[^\t\n\r]]", "");
diff --git a/src/test/java/io/logdash/sdk/LogdashTrackTest.java b/src/test/java/io/logdash/sdk/LogdashTrackTest.java
new file mode 100644
index 0000000..c06c55c
--- /dev/null
+++ b/src/test/java/io/logdash/sdk/LogdashTrackTest.java
@@ -0,0 +1,263 @@
+package io.logdash.sdk;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.List;
+import java.util.Map;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.*;
+import static com.github.tomakehurst.wiremock.stubbing.Scenario.STARTED;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LogdashTrackTest {
+
+ private static final String UUID_V4 =
+ "[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}";
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private WireMockServer wireMockServer;
+ private PrintStream originalOut;
+ private PrintStream originalErr;
+ private ByteArrayOutputStream capturedOut;
+ private ByteArrayOutputStream capturedErr;
+
+ private static String stripAnsiCodes(String input) {
+ return input.replaceAll("\\u001B\\[[;\\d]*m", "");
+ }
+
+ @BeforeEach
+ void setUp() {
+ wireMockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort());
+ wireMockServer.start();
+ wireMockServer.stubFor(post(urlEqualTo("/events")).willReturn(aResponse().withStatus(202)));
+
+ originalOut = System.out;
+ originalErr = System.err;
+ capturedOut = new ByteArrayOutputStream();
+ capturedErr = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(capturedOut));
+ System.setErr(new PrintStream(capturedErr));
+ }
+
+ @AfterEach
+ void tearDown() {
+ System.setOut(originalOut);
+ System.setErr(originalErr);
+ wireMockServer.stop();
+ }
+
+ private Logdash.Builder logdashBuilder() {
+ return Logdash.builder()
+ .apiKey("test-api-key")
+ .baseUrl("http://localhost:" + wireMockServer.port())
+ .enableConsoleOutput(false)
+ .retryDelayMs(50L);
+ }
+
+ private List sentEventIds() {
+ return wireMockServer.findAll(postRequestedFor(urlEqualTo("/events"))).stream()
+ .map(request -> readJson(request.getBodyAsString()).get("id").asText())
+ .toList();
+ }
+
+ private static JsonNode readJson(String json) {
+ try {
+ return MAPPER.readTree(json);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Test
+ void should_send_event_to_events_endpoint() {
+ // Act
+ try (var logdash = logdashBuilder().build()) {
+ logdash.track("agent_run", Map.of("agent", "support", "steps", 4, "success", true), "42");
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(
+ 1,
+ postRequestedFor(urlEqualTo("/events"))
+ .withHeader("project-api-key", equalTo("test-api-key"))
+ .withHeader("Content-Type", equalTo("application/json"))
+ .withRequestBody(
+ equalToJson(
+ """
+ {
+ "name": "agent_run",
+ "props": {"agent": "support", "steps": 4, "success": true},
+ "userId": "42",
+ "id": "${json-unit.regex}%s"
+ }
+ """
+ .formatted(UUID_V4))));
+ }
+
+ @Test
+ void should_leave_props_user_id_and_app_out_when_unset() {
+ // Act
+ try (var logdash = logdashBuilder().build()) {
+ logdash.track("signup");
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(
+ 1,
+ postRequestedFor(urlEqualTo("/events"))
+ .withRequestBody(
+ equalToJson(
+ "{\"name\": \"signup\", \"id\": \"${json-unit.regex}%s\"}"
+ .formatted(UUID_V4))));
+ }
+
+ @Test
+ void should_send_app_with_events_when_set() {
+ // Act
+ try (var logdash = logdashBuilder().app("desktop").build()) {
+ logdash.track("agent_run", Map.of("steps", 4));
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(
+ 1,
+ postRequestedFor(urlEqualTo("/events"))
+ .withRequestBody(matchingJsonPath("$.app", equalTo("desktop")))
+ .withRequestBody(matchingJsonPath("$.props.steps", equalTo("4"))));
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(strings = {"Agent_Run", "agent-run", "1st_run", "pageview", "browser_error"})
+ void should_drop_event_with_invalid_name_with_warning(String name) {
+ // Act
+ try (var logdash = logdashBuilder().enableVerboseLogging(true).build()) {
+ logdash.track(name, Map.of("steps", 4), "42");
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(0, postRequestedFor(urlEqualTo("/events")));
+ assertThat(capturedErr.toString()).contains("Warning: Event dropped: Event name");
+ }
+
+ @Test
+ void should_drop_event_with_email_as_user_id_with_warning() {
+ // Act
+ try (var logdash = logdashBuilder().enableVerboseLogging(true).build()) {
+ logdash.track("signup", Map.of(), "ada@example.com");
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(0, postRequestedFor(urlEqualTo("/events")));
+ assertThat(capturedErr.toString())
+ .contains("Warning: Event dropped: User ID of event signup must be an internal ID")
+ .doesNotContain("ada@example.com");
+ }
+
+ @Test
+ void should_drop_event_instead_of_throwing_when_a_prop_fails() {
+ // Arrange
+ var failingValue =
+ new Object() {
+ @Override
+ public String toString() {
+ throw new IllegalStateException("toString failed");
+ }
+ };
+
+ // Act
+ try (var logdash = logdashBuilder().enableVerboseLogging(true).build()) {
+ logdash.track("agent_run", Map.of("agent", failingValue));
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(0, postRequestedFor(urlEqualTo("/events")));
+ assertThat(capturedErr.toString()).contains("Warning: Event dropped: toString failed");
+ }
+
+ @Test
+ void should_send_tracked_events_on_flush() {
+ try (var logdash = logdashBuilder().build()) {
+ // Act
+ for (int step = 0; step < 5; step++) {
+ logdash.track("agent_step", Map.of("step", step));
+ }
+ logdash.flush();
+
+ // Assert, before close sends what is left
+ wireMockServer.verify(5, postRequestedFor(urlEqualTo("/events")));
+ assertThat(sentEventIds()).doesNotHaveDuplicates().allMatch(id -> id.matches(UUID_V4));
+ }
+ }
+
+ @Test
+ void should_send_tracked_events_on_close() {
+ // Arrange
+ var logdash = logdashBuilder().build();
+ logdash.track("agent_run");
+
+ // Act
+ logdash.close();
+
+ // Assert
+ wireMockServer.verify(1, postRequestedFor(urlEqualTo("/events")));
+ }
+
+ @Test
+ void should_reuse_event_id_when_retrying() {
+ // Arrange
+ wireMockServer.stubFor(
+ post(urlEqualTo("/events"))
+ .inScenario("retry")
+ .whenScenarioStateIs(STARTED)
+ .willReturn(aResponse().withStatus(503))
+ .willSetStateTo("retried"));
+ wireMockServer.stubFor(
+ post(urlEqualTo("/events"))
+ .inScenario("retry")
+ .whenScenarioStateIs("retried")
+ .willReturn(aResponse().withStatus(202)));
+
+ // Act
+ try (var logdash = logdashBuilder().build()) {
+ logdash.track("agent_run");
+ logdash.flush();
+ }
+
+ // Assert
+ var ids = sentEventIds();
+ assertThat(ids).hasSize(2).allMatch(id -> id.matches(UUID_V4));
+ assertThat(ids.get(1)).isEqualTo(ids.get(0));
+ }
+
+ @Test
+ void should_print_event_to_console_without_api_key() {
+ // Act
+ try (var logdash =
+ Logdash.builder().baseUrl("http://localhost:" + wireMockServer.port()).build()) {
+ logdash.track("agent_run", Map.of("agent", "support"), "42");
+ }
+
+ // Assert
+ assertThat(stripAnsiCodes(capturedOut.toString()))
+ .contains("[EVENT] agent_run {agent=support} (userId=42)");
+ wireMockServer.verify(0, anyRequestedFor(anyUrl()));
+ }
+}
diff --git a/src/test/java/io/logdash/sdk/event/EventEntryTest.java b/src/test/java/io/logdash/sdk/event/EventEntryTest.java
new file mode 100644
index 0000000..e732c83
--- /dev/null
+++ b/src/test/java/io/logdash/sdk/event/EventEntryTest.java
@@ -0,0 +1,121 @@
+package io.logdash.sdk.event;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.*;
+
+class EventEntryTest {
+
+ private enum Plan {
+ PRO
+ }
+
+ @Test
+ void should_create_event_entry_with_new_uuid_v4() {
+ // Act
+ var eventEntry = new EventEntry("agent_run", Map.of("steps", 4), "42");
+
+ // Assert
+ assertThat(eventEntry.name()).isEqualTo("agent_run");
+ assertThat(eventEntry.props()).containsExactly(entry("steps", 4));
+ assertThat(eventEntry.userId()).isEqualTo("42");
+ assertThat(eventEntry.id().version()).isEqualTo(4);
+ assertThat(eventEntry.id()).isNotEqualTo(new EventEntry("agent_run", null, null).id());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"a", "agent_run", "step2_done"})
+ void should_accept_valid_names(String name) {
+ assertThat(new EventEntry(name, null, null).name()).isEqualTo(name);
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(strings = {"Agent_Run", "agent-run", "agent run", "1st_run", "_run", "agent_run\n"})
+ void should_throw_exception_for_invalid_name(String name) {
+ assertThatThrownBy(() -> new EventEntry(name, null, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Event name must match ^[a-z][a-z0-9_]{0,63}$");
+ }
+
+ @Test
+ void should_limit_name_to_64_characters() {
+ assertThatNoException().isThrownBy(() -> new EventEntry("a".repeat(64), null, null));
+ assertThatThrownBy(() -> new EventEntry("a".repeat(65), null, null))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"pageview", "pageleave", "browser_error"})
+ void should_throw_exception_for_website_tracker_name(String name) {
+ assertThatThrownBy(() -> new EventEntry(name, null, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("belongs to the website tracker");
+ }
+
+ @Test
+ void should_throw_exception_for_email_as_user_id_without_repeating_it() {
+ assertThatThrownBy(() -> new EventEntry("signup", null, "ada@example.com"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("never an email address")
+ .hasMessageNotContaining("ada@example.com");
+ }
+
+ @Test
+ void should_limit_user_id_to_256_characters() {
+ assertThatNoException().isThrownBy(() -> new EventEntry("signup", null, "1".repeat(256)));
+ assertThatThrownBy(() -> new EventEntry("signup", null, "1".repeat(257)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("at most 256 characters");
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(strings = {" ", "\t"})
+ void should_treat_blank_user_id_as_unset(String userId) {
+ assertThat(new EventEntry("signup", null, userId).userId()).isNull();
+ }
+
+ @Test
+ void should_keep_props_flat_and_in_order() {
+ // Arrange
+ var props = new LinkedHashMap();
+ props.put("plan", Plan.PRO);
+ props.put("steps", 4);
+ props.put("skipped", null);
+ props.put(null, "no key");
+ props.put("success", true);
+ props.put("model", "opus");
+
+ // Act
+ var eventEntry = new EventEntry("agent_run", props, null);
+
+ // Assert
+ assertThat(eventEntry.props())
+ .containsExactly(
+ entry("plan", "PRO"), entry("steps", 4), entry("success", true), entry("model", "opus"));
+ }
+
+ @Test
+ void should_copy_props_into_an_unmodifiable_map() {
+ // Arrange
+ var props = new HashMap(Map.of("steps", 4));
+ var eventEntry = new EventEntry("agent_run", props, null);
+
+ // Act
+ props.put("model", "opus");
+
+ // Assert
+ assertThat(eventEntry.props()).containsOnlyKeys("steps");
+ assertThatThrownBy(() -> eventEntry.props().put("model", "opus"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThat(new EventEntry("agent_run", null, null).props()).isEmpty();
+ }
+}
diff --git a/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java b/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java
index e49844b..4169929 100644
--- a/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java
+++ b/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java
@@ -589,7 +589,7 @@ void should_demonstrate_concurrency_behavior() throws InterruptedException {
}
@Test
- void should_use_efficient_countdownlatch_for_flush() throws InterruptedException {
+ void should_flush_as_soon_as_pending_requests_finish() throws InterruptedException {
// Arrange
wireMockServer.stubFor(
post(urlEqualTo("/logs")).willReturn(aResponse().withStatus(200).withFixedDelay(300)));
diff --git a/src/test/java/io/logdash/sdk/transport/LogdashTransportTest.java b/src/test/java/io/logdash/sdk/transport/LogdashTransportTest.java
index 6391bce..287d8e3 100644
--- a/src/test/java/io/logdash/sdk/transport/LogdashTransportTest.java
+++ b/src/test/java/io/logdash/sdk/transport/LogdashTransportTest.java
@@ -1,5 +1,6 @@
package io.logdash.sdk.transport;
+import io.logdash.sdk.event.EventEntry;
import io.logdash.sdk.log.LogEntry;
import io.logdash.sdk.log.LogLevel;
import io.logdash.sdk.metrics.MetricEntry;
@@ -56,6 +57,11 @@ public CompletableFuture sendMetric(MetricEntry metricEntry) {
return CompletableFuture.completedFuture(null);
}
+ @Override
+ public CompletableFuture sendEvent(EventEntry eventEntry) {
+ return CompletableFuture.completedFuture(null);
+ }
+
@Override
public void close() {
// Test implementation
From 74ecbc085a39b8fccbb1e9d5859b3dd431b5bca3 Mon Sep 17 00:00:00 2001
From: "malini-agent[bot]"
<338999285+malini-agent[bot]@users.noreply.github.com>
Date: Sat, 24 Oct 2026 16:44:10 +0200
Subject: [PATCH 2/2] fix: never block callers on a full queue or retry after
close
---
.../logdash/sdk/transport/HttpTransport.java | 115 ++++++++----------
.../java/io/logdash/sdk/LogdashTrackTest.java | 97 +++++++++++++++
.../io/logdash/sdk/event/EventEntryTest.java | 10 ++
.../sdk/transport/HttpTransportTest.java | 66 ++++++++++
.../sdk/transport/NoOpTransportTest.java | 34 +++++-
5 files changed, 257 insertions(+), 65 deletions(-)
diff --git a/src/main/java/io/logdash/sdk/transport/HttpTransport.java b/src/main/java/io/logdash/sdk/transport/HttpTransport.java
index 0781086..ed5d0e7 100644
--- a/src/main/java/io/logdash/sdk/transport/HttpTransport.java
+++ b/src/main/java/io/logdash/sdk/transport/HttpTransport.java
@@ -28,6 +28,7 @@
*
*
* - Non-blocking concurrency control using tryAcquire
+ *
- Never blocks the caller: a request that finds the queue full is dropped
*
- Request tracking from submission, so flush and shutdown wait for every request sent
*
- Optimized thread pool sizing
*
- Fast retry logic with exponential backoff
@@ -104,10 +105,6 @@ public CompletableFuture sendMetric(MetricEntry metricEntry) {
@Override
public CompletableFuture sendEvent(EventEntry eventEntry) {
- if (shutdownInitiated.get() || closed.get()) {
- return CompletableFuture.completedFuture(null);
- }
-
try {
String json = serializer.serialize(eventEntry);
return sendWithRetry(EVENTS_ENDPOINT, json, "POST");
@@ -133,7 +130,7 @@ private CompletableFuture sendWithRetry(String endpoint, String json, Stri
final long sequenceId = requestCounter.incrementAndGet();
- CompletableFuture request = CompletableFuture.runAsync(
+ Runnable send =
() -> {
boolean acquired = false;
try {
@@ -156,6 +153,10 @@ private CompletableFuture sendWithRetry(String endpoint, String json, Stri
try {
sendHttpRequest(endpoint, json, method, attempt, sequenceId);
return;
+ } catch (InterruptedException e) {
+ // close() interrupts requests in flight: stop instead of retrying
+ Thread.currentThread().interrupt();
+ return;
} catch (Exception e) {
lastException = e;
if (config.enableVerboseLogging()) {
@@ -192,8 +193,18 @@ private CompletableFuture sendWithRetry(String endpoint, String json, Stri
rateLimiter.release();
}
}
- },
- executorService);
+ };
+
+ CompletableFuture request;
+ try {
+ request = CompletableFuture.runAsync(send, executorService);
+ } catch (RejectedExecutionException e) {
+ // A full queue drops the request, so sending never blocks the caller
+ if (config.enableVerboseLogging()) {
+ System.out.printf("Request #%d rejected because the send queue is full%n", sequenceId);
+ }
+ return CompletableFuture.completedFuture(null);
+ }
// Tracked from submission, not from start, so a flush right after a send waits for it
pendingRequests.add(request);
@@ -249,88 +260,64 @@ private void sendHttpRequest(
/**
* Flushes all pending requests, including the ones still queued.
*
- * @return a future that completes when all pending requests finish
+ * @return a future that completes when all pending requests finish, or at the flush timeout
*/
public CompletableFuture flush() {
- return CompletableFuture.runAsync(
- () -> {
- try {
- if (!pendingRequests.isEmpty()) {
- long timeoutMs = Math.min(config.shutdownTimeoutMs(), 5000);
- boolean completed = awaitPendingRequests(timeoutMs);
-
- if (completed && config.enableVerboseLogging()) {
+ return awaitPendingRequests(Math.min(config.shutdownTimeoutMs(), 5000))
+ .thenAccept(
+ completed -> {
+ if (!config.enableVerboseLogging()) {
+ return;
+ }
+ if (completed) {
System.out.println("All pending requests flushed successfully");
- } else if (!completed && config.enableVerboseLogging()) {
+ } else {
System.out.printf(
"Flush timeout, %d requests still pending%n", pendingRequests.size());
}
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- },
- executorService);
+ });
}
@Override
public Optional> shutdown() {
if (shutdownInitiated.compareAndSet(false, true)) {
- return Optional.of(
- CompletableFuture.runAsync(
- () -> {
- try {
- if (config.enableVerboseLogging()) {
- System.out.println("Initiating HTTP transport shutdown...");
- }
-
- long timeoutMs = Math.min(config.shutdownTimeoutMs(), 3000);
+ if (config.enableVerboseLogging()) {
+ System.out.println("Initiating HTTP transport shutdown...");
+ System.out.printf("Waiting for %d pending HTTP requests...%n", pendingRequests.size());
+ }
- if (!pendingRequests.isEmpty()) {
- if (config.enableVerboseLogging()) {
- System.out.printf(
- "Waiting for %d pending HTTP requests...%n", pendingRequests.size());
+ return Optional.of(
+ awaitPendingRequests(Math.min(config.shutdownTimeoutMs(), 3000))
+ .thenAccept(
+ completed -> {
+ closed.set(true);
+ if (!config.enableVerboseLogging()) {
+ return;
}
-
- boolean completed = awaitPendingRequests(timeoutMs);
-
- if (completed && config.enableVerboseLogging()) {
+ if (completed) {
System.out.println("All HTTP requests completed successfully");
- } else if (!completed && config.enableVerboseLogging()) {
+ } else {
System.out.printf(
"Shutdown timeout exceeded, %d requests may be lost%n",
pendingRequests.size());
}
- }
-
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- } finally {
- closed.set(true);
- }
- },
- executorService));
+ }));
}
return Optional.of(CompletableFuture.completedFuture(null));
}
/**
- * Waits for every request sent so far.
+ * Waits for every request sent so far, without holding a thread, so a full queue never rejects
+ * the wait.
*
* @param timeoutMs the longest time to wait
- * @return true when all of them finished within the timeout
+ * @return a future of true when all of them finish within the timeout, or false
*/
- private boolean awaitPendingRequests(long timeoutMs) throws InterruptedException {
- try {
- CompletableFuture.allOf(pendingRequests.toArray(new CompletableFuture[0]))
- .get(timeoutMs, TimeUnit.MILLISECONDS);
- return true;
- } catch (ExecutionException e) {
- // A request that failed has finished too
- return true;
- } catch (TimeoutException e) {
- return false;
- }
+ private CompletableFuture awaitPendingRequests(long timeoutMs) {
+ return CompletableFuture.allOf(pendingRequests.toArray(new CompletableFuture[0]))
+ // A request that failed has finished too
+ .handle((result, error) -> true)
+ .completeOnTimeout(false, timeoutMs, TimeUnit.MILLISECONDS);
}
@Override
@@ -377,7 +364,7 @@ private ExecutorService createOptimizedExecutorService() {
t.setDaemon(true);
return t;
},
- new ThreadPoolExecutor.CallerRunsPolicy());
+ new ThreadPoolExecutor.AbortPolicy());
executor.allowCoreThreadTimeOut(true);
return executor;
diff --git a/src/test/java/io/logdash/sdk/LogdashTrackTest.java b/src/test/java/io/logdash/sdk/LogdashTrackTest.java
index c06c55c..0039a15 100644
--- a/src/test/java/io/logdash/sdk/LogdashTrackTest.java
+++ b/src/test/java/io/logdash/sdk/LogdashTrackTest.java
@@ -26,6 +26,37 @@ class LogdashTrackTest {
"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}";
private static final ObjectMapper MAPPER = new ObjectMapper();
+ /**
+ * A number that JSON serialization cannot write, since Jackson writes unknown numbers with
+ * {@code toString()}.
+ */
+ private static final class UnprintableNumber extends Number {
+ @Override
+ public int intValue() {
+ return 4;
+ }
+
+ @Override
+ public long longValue() {
+ return 4;
+ }
+
+ @Override
+ public float floatValue() {
+ return 4;
+ }
+
+ @Override
+ public double doubleValue() {
+ return 4;
+ }
+
+ @Override
+ public String toString() {
+ throw new IllegalStateException("toString failed");
+ }
+ }
+
private WireMockServer wireMockServer;
private PrintStream originalOut;
private PrintStream originalErr;
@@ -220,6 +251,72 @@ void should_send_tracked_events_on_close() {
wireMockServer.verify(1, postRequestedFor(urlEqualTo("/events")));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void should_drop_events_instead_of_blocking_when_the_send_queue_is_full(boolean verbose) {
+ // Arrange: a slow response holds the only request slot, so the queue fills up
+ wireMockServer.stubFor(
+ post(urlEqualTo("/events")).willReturn(aResponse().withStatus(202).withFixedDelay(2000)));
+
+ try (var logdash =
+ logdashBuilder()
+ .maxConcurrentRequests(1)
+ .shutdownTimeoutMs(500L)
+ .enableVerboseLogging(verbose)
+ .build()) {
+ // Act
+ long start = System.nanoTime();
+ for (int i = 0; i < 500; i++) {
+ logdash.track("agent_run");
+ }
+ long elapsedMs = (System.nanoTime() - start) / 1_000_000;
+
+ // Assert
+ assertThat(elapsedMs).isLessThan(1000);
+ }
+ assertThat(capturedOut.toString().contains("rejected because the send queue is full"))
+ .isEqualTo(verbose);
+ }
+
+ @Test
+ void should_drop_invalid_event_silently_without_verbose_logging() {
+ // Act
+ try (var logdash = logdashBuilder().build()) {
+ logdash.track("Agent_Run");
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(0, postRequestedFor(urlEqualTo("/events")));
+ assertThat(capturedErr.toString()).doesNotContain("Event dropped");
+ }
+
+ @Test
+ void should_drop_events_tracked_after_close() {
+ // Arrange
+ var logdash = logdashBuilder().build();
+ logdash.close();
+
+ // Act
+ logdash.track("agent_run");
+
+ // Assert
+ wireMockServer.verify(0, postRequestedFor(urlEqualTo("/events")));
+ }
+
+ @Test
+ void should_drop_event_whose_props_fail_to_serialize() {
+ // Act
+ try (var logdash = logdashBuilder().enableVerboseLogging(true).build()) {
+ logdash.track("agent_run", Map.of("steps", new UnprintableNumber()));
+ logdash.flush();
+ }
+
+ // Assert
+ wireMockServer.verify(0, postRequestedFor(urlEqualTo("/events")));
+ assertThat(capturedErr.toString()).contains("Failed to serialize event entry");
+ }
+
@Test
void should_reuse_event_id_when_retrying() {
// Arrange
diff --git a/src/test/java/io/logdash/sdk/event/EventEntryTest.java b/src/test/java/io/logdash/sdk/event/EventEntryTest.java
index e732c83..1d3d91d 100644
--- a/src/test/java/io/logdash/sdk/event/EventEntryTest.java
+++ b/src/test/java/io/logdash/sdk/event/EventEntryTest.java
@@ -8,6 +8,7 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.UUID;
import static org.assertj.core.api.Assertions.*;
@@ -30,6 +31,15 @@ void should_create_event_entry_with_new_uuid_v4() {
assertThat(eventEntry.id()).isNotEqualTo(new EventEntry("agent_run", null, null).id());
}
+ @Test
+ void should_keep_the_given_id() {
+ // Arrange
+ var id = UUID.randomUUID();
+
+ // Act & Assert
+ assertThat(new EventEntry("agent_run", null, null, id).id()).isEqualTo(id);
+ }
+
@ParameterizedTest
@ValueSource(strings = {"a", "agent_run", "step2_done"})
void should_accept_valid_names(String name) {
diff --git a/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java b/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java
index 4169929..1917b2e 100644
--- a/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java
+++ b/src/test/java/io/logdash/sdk/transport/HttpTransportTest.java
@@ -14,6 +14,8 @@
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
@@ -620,6 +622,70 @@ void should_flush_as_soon_as_pending_requests_finish() throws InterruptedExcepti
wireMockServer.verify(3, postRequestedFor(urlEqualTo("/logs")));
}
+ @Test
+ void should_stop_waiting_for_a_hanging_request_at_the_flush_and_shutdown_timeouts()
+ throws InterruptedException {
+ // Arrange
+ wireMockServer.stubFor(
+ post(urlEqualTo("/logs")).willReturn(aResponse().withStatus(200).withFixedDelay(3000)));
+ var hangingConfig =
+ LogdashConfig.builder()
+ .apiKey("test-api-key")
+ .baseUrl("http://localhost:" + wireMockServer.port())
+ .enableVerboseLogging(true)
+ .requestTimeoutMs(5000L)
+ .shutdownTimeoutMs(300L)
+ .build();
+ var originalOut = System.out;
+ var capturedOut = new ByteArrayOutputStream();
+ System.setOut(new PrintStream(capturedOut));
+
+ try (var hangingTransport = new HttpTransport(hangingConfig)) {
+ hangingTransport.sendLog(new LogEntry("Hanging", LogLevel.INFO, 1L));
+
+ // Act & Assert
+ assertThat(hangingTransport.flush()).succeedsWithin(2, TimeUnit.SECONDS);
+ assertThat(hangingTransport.shutdown().orElseThrow()).succeedsWithin(2, TimeUnit.SECONDS);
+ } finally {
+ System.setOut(originalOut);
+ }
+
+ assertThat(capturedOut.toString())
+ .contains("Flush timeout, 1 requests still pending")
+ .contains("Shutdown timeout exceeded, 1 requests may be lost");
+
+ // close() interrupted the hanging request, which must not be retried
+ Thread.sleep(300);
+ wireMockServer.verify(1, postRequestedFor(urlEqualTo("/logs")));
+ }
+
+ @Test
+ @Timeout(10)
+ void should_drop_requests_waiting_for_a_slot_on_close() throws InterruptedException {
+ // Arrange: the first request holds the only slot, so the second one waits for it
+ wireMockServer.stubFor(
+ post(urlEqualTo("/logs")).willReturn(aResponse().withStatus(200).withFixedDelay(150)));
+ var singleSlotConfig =
+ LogdashConfig.builder()
+ .apiKey("test-api-key")
+ .baseUrl("http://localhost:" + wireMockServer.port())
+ .maxConcurrentRequests(1)
+ .build();
+ var singleSlotTransport = new HttpTransport(singleSlotConfig);
+ singleSlotTransport.sendLog(new LogEntry("Sending", LogLevel.INFO, 1L));
+ while (wireMockServer.findAll(postRequestedFor(urlEqualTo("/logs"))).isEmpty()) {
+ Thread.sleep(5);
+ }
+ var waiting = singleSlotTransport.sendLog(new LogEntry("Waiting", LogLevel.INFO, 2L));
+
+ // Act
+ singleSlotTransport.close();
+
+ // Assert
+ assertThat(waiting).succeedsWithin(1, TimeUnit.SECONDS);
+ wireMockServer.verify(1, postRequestedFor(urlEqualTo("/logs")));
+ }
+
@Test
void should_handle_multiple_shutdown_calls_safely() {
// Arrange
diff --git a/src/test/java/io/logdash/sdk/transport/NoOpTransportTest.java b/src/test/java/io/logdash/sdk/transport/NoOpTransportTest.java
index 56e351b..fd74c8a 100644
--- a/src/test/java/io/logdash/sdk/transport/NoOpTransportTest.java
+++ b/src/test/java/io/logdash/sdk/transport/NoOpTransportTest.java
@@ -1,6 +1,7 @@
package io.logdash.sdk.transport;
import io.logdash.sdk.config.LogdashConfig;
+import io.logdash.sdk.event.EventEntry;
import io.logdash.sdk.log.LogEntry;
import io.logdash.sdk.log.LogLevel;
import io.logdash.sdk.metrics.MetricEntry;
@@ -11,6 +12,7 @@
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -103,19 +105,23 @@ void should_not_print_when_console_output_disabled() throws Exception {
var logEntry = new LogEntry("Test message", LogLevel.INFO, 1L);
var metricEntry = new MetricEntry("test_metric", 42, MetricType.SET);
+ var eventEntry = new EventEntry("agent_run", Map.of("steps", 4), "42");
// Act
var logFuture = silentTransport.sendLog(logEntry);
var metricFuture = silentTransport.sendMetric(metricEntry);
+ var eventFuture = silentTransport.sendEvent(eventEntry);
// Assert
assertThat(logFuture).succeedsWithin(100, TimeUnit.MILLISECONDS);
assertThat(metricFuture).succeedsWithin(100, TimeUnit.MILLISECONDS);
+ assertThat(eventFuture).succeedsWithin(100, TimeUnit.MILLISECONDS);
// Only initialization messages should be present (none in this case)
var output = capturedOutput.toString();
assertThat(output).doesNotContain("[LOG]");
assertThat(output).doesNotContain("[METRIC]");
+ assertThat(output).doesNotContain("[EVENT]");
silentTransport.close();
}
@@ -126,15 +132,41 @@ void should_handle_operations_after_close() throws Exception {
transport.close();
var logEntry = new LogEntry("Test message", LogLevel.INFO, 1L);
var metricEntry = new MetricEntry("test_metric", 42, MetricType.SET);
+ var eventEntry = new EventEntry("agent_run", Map.of("steps", 4), "42");
// Act
var logFuture = transport.sendLog(logEntry);
var metricFuture = transport.sendMetric(metricEntry);
+ var eventFuture = transport.sendEvent(eventEntry);
// Assert
assertThat(logFuture).succeedsWithin(100, TimeUnit.MILLISECONDS);
assertThat(metricFuture).succeedsWithin(100, TimeUnit.MILLISECONDS);
- assertThat(capturedOutput.toString()).contains("transport closed");
+ assertThat(eventFuture).succeedsWithin(100, TimeUnit.MILLISECONDS);
+ assertThat(capturedOutput.toString()).contains("transport closed").doesNotContain("[EVENT]");
+ }
+
+ @Test
+ void should_print_event_to_console_when_enabled() {
+ // Arrange
+ var eventEntry = new EventEntry("agent_run", Map.of("steps", 4), "42");
+
+ // Act
+ var future = transport.sendEvent(eventEntry);
+
+ // Assert
+ assertThat(future).succeedsWithin(100, TimeUnit.MILLISECONDS);
+ assertThat(stripAnsiCodes(capturedOutput.toString()))
+ .contains("[EVENT] agent_run {steps=4} (userId=42)");
+ }
+
+ @Test
+ void should_print_event_without_props_or_user_id() {
+ // Act
+ transport.sendEvent(new EventEntry("signup", null, null));
+
+ // Assert
+ assertThat(stripAnsiCodes(capturedOutput.toString())).containsPattern("\\[EVENT] signup\\R");
}
@Test