diff --git a/doc/changes/changelog.md b/doc/changes/changelog.md index 9e62d07..39f13d8 100644 --- a/doc/changes/changelog.md +++ b/doc/changes/changelog.md @@ -1,3 +1,4 @@ # Changes +* [0.2.0](changes_0.2.0.md) * [0.1.0](changes_0.1.0.md) diff --git a/doc/changes/changes_0.2.0.md b/doc/changes/changes_0.2.0.md new file mode 100644 index 0000000..56af214 --- /dev/null +++ b/doc/changes/changes_0.2.0.md @@ -0,0 +1,10 @@ +# Telemetry Java 0.2.0, released 2026-??-?? + +Code name: + +## Summary + +## Features + +* ISSUE_NUMBER: description + diff --git a/pk_generated_parent.pom b/pk_generated_parent.pom index 228b85e..e433067 100644 --- a/pk_generated_parent.pom +++ b/pk_generated_parent.pom @@ -3,7 +3,7 @@ 4.0.0 com.exasol telemetry-java-generated-parent - 0.1.0 + 0.2.0 pom UTF-8 diff --git a/pom.xml b/pom.xml index 7ca6712..17dc7d7 100644 --- a/pom.xml +++ b/pom.xml @@ -1,11 +1,9 @@ - + 4.0.0 com.exasol telemetry-java - 0.1.0 + 0.2.0 telemetry-java Minimal, zero-dependency telemetry library for Java applications. https://github.com/exasol/telemetry-java/ @@ -114,7 +112,7 @@ telemetry-java-generated-parent com.exasol - 0.1.0 + 0.2.0 pk_generated_parent.pom diff --git a/src/main/java/com/exasol/telemetry/AsyncTelemetryClient.java b/src/main/java/com/exasol/telemetry/AsyncTelemetryClient.java new file mode 100644 index 0000000..38fdcc7 --- /dev/null +++ b/src/main/java/com/exasol/telemetry/AsyncTelemetryClient.java @@ -0,0 +1,171 @@ +package com.exasol.telemetry; + +import static java.util.Objects.requireNonNull; + +import java.time.*; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.*; +import java.util.logging.Logger; + +final class AsyncTelemetryClient implements TelemetryClient { + private static final Logger LOGGER = Logger.getLogger(AsyncTelemetryClient.class.getName()); + + private final TelemetryConfig config; + private final BlockingQueue queue; + private final HttpTransport transport; + private final Thread senderThread; + private final Clock clock; + private final CountDownLatch terminated = new CountDownLatch(1); + private volatile boolean closed; + + AsyncTelemetryClient(final TelemetryConfig config) { + this(config, Clock.systemUTC()); + } + + AsyncTelemetryClient(final TelemetryConfig config, final Clock clock) { + this.config = requireNonNull(config, "config"); + this.clock = requireNonNull(clock, "clock"); + this.queue = new ArrayBlockingQueue<>(config.getQueueCapacity()); + this.transport = new HttpTransport(config); + this.senderThread = new Thread(this::runSender, "telemetry-java-sender"); + this.senderThread.setDaemon(true); + this.senderThread.start(); + } + + @Override + public void track(final String feature) { + if (closed || feature == null) { + return; + } + final TelemetryEvent event = new TelemetryEvent(feature, clock.instant()); + enqueue(event); + } + + @SuppressWarnings("java:S899") // Intentionally ignore return value of offer() to avoid blocking the caller. + private void enqueue(final TelemetryEvent event) { + queue.offer(event); + } + + private void runSender() { + try { + while (!closed || !queue.isEmpty()) { + final TelemetryEvent event = queue.poll(100, TimeUnit.MILLISECONDS); + if (event != null) { + sendWithRetry(drainBatch(event)); + } + } + } catch (final InterruptedException ignored) { + Thread.currentThread().interrupt(); + } finally { + terminated.countDown(); + } + } + + private List drainBatch(final TelemetryEvent firstEvent) { + final List batch = new ArrayList<>(); + batch.add(firstEvent); + queue.drainTo(batch); + return batch; + } + + // [impl~telemetry-client-send-with-retry~1->req~async-delivery~1] + private void sendWithRetry(final List events) { + final Instant start = clock.instant(); + final Message message = Message.fromEvents(config.getProjectTag(), config.getProductVersion(), start, events); + final Instant deadline = start.plus(config.getRetryTimeout()); + Duration delay = config.getInitialRetryDelay(); + + while (true) { + if (Thread.currentThread().isInterrupted()) { + return; + } + try { + transport.send(message); + LOGGER.fine(() -> "Telemetry sent to the server with " + events.size() + " event(s)."); + return; + } catch (final Exception exception) { + LOGGER.fine(() -> "Telemetry sending failed for " + events.size() + " event(s): " + + rootCauseMessage(exception)); + if (Thread.currentThread().isInterrupted()) { + return; + } + final Instant now = clock.instant(); + if (!now.isBefore(deadline)) { + return; + } + final Duration remaining = Duration.between(now, deadline); + sleep(min(delay, remaining)); + delay = min(delay.multipliedBy(2), config.getMaxRetryDelay()); + } + } + } + + private static Duration min(final Duration left, final Duration right) { + return left.compareTo(right) <= 0 ? left : right; + } + + private static String rootCauseMessage(final Throwable throwable) { + Throwable cause = throwable; + while (cause != null) { + if (cause instanceof HttpException) { + final HttpException httpException = (HttpException) cause; + return "server status " + httpException.getStatusCode() + " (" + httpException.getServerStatus() + ")"; + } + if (cause.getCause() == null) { + final String message = cause.getMessage(); + if (message == null || message.isBlank()) { + return cause.getClass().getSimpleName(); + } + } + cause = cause.getCause(); + } + return ""; + } + + private void sleep(final Duration duration) { + try { + Thread.sleep(Math.max(1, duration.toMillis())); + } catch (final InterruptedException ignored) { + Thread.currentThread().interrupt(); + } + } + + // Visible for testing + boolean awaitStopped(final Duration timeout) throws InterruptedException { + return terminated.await(timeout.toMillis(), TimeUnit.MILLISECONDS); + } + + // Visible for testing + boolean isRunning() { + return senderThread.isAlive(); + } + + @Override + public void close() { + if (closed) { + return; + } + closed = true; + awaitSenderStop(); + LOGGER.fine("Telemetry is stopped."); + } + + private void awaitSenderStop() { + final long timeoutNanos = config.getRetryTimeout().toNanos(); + final long deadlineNanos = System.nanoTime() + timeoutNanos; + try { + while (senderThread.isAlive()) { + final long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + senderThread.interrupt(); + senderThread.join(); + return; + } + TimeUnit.NANOSECONDS.timedJoin(senderThread, remainingNanos); + } + } catch (final InterruptedException ignored) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/src/main/java/com/exasol/telemetry/NoOpTelemetryClient.java b/src/main/java/com/exasol/telemetry/NoOpTelemetryClient.java new file mode 100644 index 0000000..363d4e8 --- /dev/null +++ b/src/main/java/com/exasol/telemetry/NoOpTelemetryClient.java @@ -0,0 +1,17 @@ +package com.exasol.telemetry; + +final class NoOpTelemetryClient implements TelemetryClient { + NoOpTelemetryClient() { + // Intentionally empty. + } + + @Override + public void track(final String feature) { + // Intentionally does nothing. + } + + @Override + public void close() { + // Intentionally does nothing. + } +} diff --git a/src/main/java/com/exasol/telemetry/TelemetryClient.java b/src/main/java/com/exasol/telemetry/TelemetryClient.java index 692e0f6..ee23365 100644 --- a/src/main/java/com/exasol/telemetry/TelemetryClient.java +++ b/src/main/java/com/exasol/telemetry/TelemetryClient.java @@ -1,11 +1,5 @@ package com.exasol.telemetry; -import static java.util.Objects.requireNonNull; - -import java.time.*; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.*; import java.util.logging.Logger; /** @@ -13,39 +7,7 @@ * Create a client by building a {@link TelemetryConfig} with {@link TelemetryConfig#builder(String, String)} and passing it * to {@link #create(TelemetryConfig)}. */ -public final class TelemetryClient implements AutoCloseable { - private static final Logger LOGGER = Logger.getLogger(TelemetryClient.class.getName()); - - private final TelemetryConfig config; - private final BlockingQueue queue; - private final HttpTransport transport; - private final Thread senderThread; - private final Clock clock; - private final boolean trackingEnabled; - private final CountDownLatch terminated = new CountDownLatch(1); - private volatile boolean closed; - - private TelemetryClient(final TelemetryConfig config) { - this(config, Clock.systemUTC()); - } - - TelemetryClient(final TelemetryConfig config, final Clock clock) { - this.config = requireNonNull(config, "config"); - this.clock = requireNonNull(clock, "clock"); - this.queue = new ArrayBlockingQueue<>(config.getQueueCapacity()); - this.transport = new HttpTransport(config); - this.trackingEnabled = !config.isTrackingDisabled(); - this.senderThread = new Thread(this::runSender, "telemetry-java-sender"); - this.senderThread.setDaemon(true); - if (trackingEnabled) { - this.senderThread.start(); - logEnabled(); - } else { - terminated.countDown(); - logDisabled(); - } - } - +public interface TelemetryClient extends AutoCloseable { /** * Create a telemetry client for the provided configuration. * @@ -53,8 +15,13 @@ private TelemetryClient(final TelemetryConfig config) { * @return telemetry client */ // [impl~telemetry-client-create~1->req~tracking-api~1] - public static TelemetryClient create(final TelemetryConfig config) { - return new TelemetryClient(config); + static TelemetryClient create(final TelemetryConfig config) { + if (config.isTrackingDisabled()) { + logDisabled(config); + return new NoOpTelemetryClient(); + } + logEnabled(config); + return new AsyncTelemetryClient(config); } /** @@ -63,161 +30,31 @@ public static TelemetryClient create(final TelemetryConfig config) { * @param feature feature name provided by the caller */ // [impl~telemetry-client-track~1->req~tracking-api~1] - public void track(final String feature) { - if (!trackingEnabled || closed) { - return; - } - - if (feature == null) { - return; - } - - final TelemetryEvent event = new TelemetryEvent(feature, clock.instant()); - enqueue(event); - } - - @SuppressWarnings("java:S899") // Return value of offer ignored, this is fire-and-forget - private void enqueue(final TelemetryEvent event) { - queue.offer(event); - } - - private void runSender() { - try { - while (!closed || !queue.isEmpty()) { - final TelemetryEvent event = queue.poll(100, TimeUnit.MILLISECONDS); - if (event != null) { - sendWithRetry(drainBatch(event)); - } - } - } catch (final InterruptedException ignored) { - Thread.currentThread().interrupt(); - } finally { - terminated.countDown(); - } - } - - private List drainBatch(final TelemetryEvent firstEvent) { - final List batch = new ArrayList<>(); - batch.add(firstEvent); - queue.drainTo(batch); - return batch; - } - - // [impl~telemetry-client-send-with-retry~1->req~async-delivery~1] - private void sendWithRetry(final List events) { - final Instant start = clock.instant(); - final Message message = Message.fromEvents(config.getProjectTag(), config.getProductVersion(), start, events); - final Instant deadline = start.plus(config.getRetryTimeout()); - Duration delay = config.getInitialRetryDelay(); - - while (true) { - if (Thread.currentThread().isInterrupted()) { - return; - } - try { - transport.send(message); - LOGGER.fine(() -> "Telemetry sent to the server with " + events.size() + " event(s)."); - return; - } catch (final Exception exception) { - LOGGER.fine(() -> "Telemetry sending failed for " + events.size() + " event(s): " - + rootCauseMessage(exception)); - if (Thread.currentThread().isInterrupted()) { - return; - } - final Instant now = clock.instant(); - if (!now.isBefore(deadline)) { - return; - } - final Duration remaining = Duration.between(now, deadline); - sleep(min(delay, remaining)); - delay = min(delay.multipliedBy(2), config.getMaxRetryDelay()); - } - } - } - - private static Duration min(final Duration left, final Duration right) { - return left.compareTo(right) <= 0 ? left : right; - } - - private static String rootCauseMessage(final Throwable throwable) { - Throwable cause = throwable; - while (cause != null) { - if (cause instanceof HttpException) { - final HttpException httpException = (HttpException) cause; - return "server status " + httpException.getStatusCode() + " (" + httpException.getServerStatus() + ")"; - } - if (cause.getCause() == null) { - final String message = cause.getMessage(); - if (message == null || message.isBlank()) { - return cause.getClass().getSimpleName(); - } - } - cause = cause.getCause(); - } - return ""; - } - - private void sleep(final Duration duration) { - try { - Thread.sleep(Math.max(1, duration.toMillis())); - } catch (final InterruptedException ignored) { - Thread.currentThread().interrupt(); - } - } - - boolean awaitStopped(final Duration timeout) throws InterruptedException { - return terminated.await(timeout.toMillis(), TimeUnit.MILLISECONDS); - } - - boolean isRunning() { - return senderThread.isAlive(); - } + void track(String feature); /** - * Stop the sender thread and wait for any queued events to be flushed before returning. + * Stop the client, send all remaining queued telemetry messages and release any resources that it owns. */ @Override // [impl~telemetry-client-close~1->req~shutdown-flush~1] - public void close() { - if (closed) { - return; - } - closed = true; - if (trackingEnabled) { - awaitSenderStop(); - } - LOGGER.fine("Telemetry is stopped."); - } + void close(); - private void awaitSenderStop() { - final long timeoutNanos = config.getRetryTimeout().toNanos(); - final long deadlineNanos = System.nanoTime() + timeoutNanos; - try { - while (senderThread.isAlive()) { - final long remainingNanos = deadlineNanos - System.nanoTime(); - if (remainingNanos <= 0) { - senderThread.interrupt(); - senderThread.join(); - return; - } - TimeUnit.NANOSECONDS.timedJoin(senderThread, remainingNanos); - } - } catch (final InterruptedException ignored) { - Thread.currentThread().interrupt(); - } - } - - private void logEnabled() { - LOGGER.info(() -> "Telemetry is enabled. Set " + TelemetryConfig.DISABLED_ENV + " to any non-empty value to disable telemetry. " + private static void logEnabled(final TelemetryConfig config) { + logger().info(() -> "Telemetry is enabled. Set " + TelemetryConfig.DISABLED_ENV + " to any non-empty value to disable telemetry. " + TelemetryConfig.DISABLED_ENV + "=" + formatEnvValue(config.getDisabledEnvValue()) + ", " + TelemetryConfig.CI_ENV + "=" + formatEnvValue(config.getCiEnvValue()) + "."); } - private void logDisabled() { - LOGGER.info(() -> "Telemetry is disabled via " + config.getDisableMechanism() + "=" + private static void logDisabled(final TelemetryConfig config) { + logger().info(() -> "Telemetry is disabled via " + config.getDisableMechanism() + "=" + formatEnvValue(config.getDisableMechanismValue()) + "."); } + // Create logger in private method to avoid a public static field in the interface. + private static Logger logger() { + return Logger.getLogger(TelemetryClient.class.getName()); + } + private static String formatEnvValue(final String value) { if (value == null) { return ""; diff --git a/src/test/java/com/exasol/telemetry/ShutdownFlushIT.java b/src/test/java/com/exasol/telemetry/ShutdownFlushIT.java index f5b14d7..2ab671c 100644 --- a/src/test/java/com/exasol/telemetry/ShutdownFlushIT.java +++ b/src/test/java/com/exasol/telemetry/ShutdownFlushIT.java @@ -36,9 +36,9 @@ void flushesPendingEventsOnClose() throws Exception { // [itest~shutdown-flush-stops-background-thread~1->req~shutdown-flush~1] @Test void stopsBackgroundThreadsAfterClose() throws Exception { - final TelemetryClient client; + final AsyncTelemetryClient client; try (RecordingHttpServer server = RecordingHttpServer.createSuccessServer()) { - client = TelemetryClient.create(server.configBuilder("shop-ui", PRODUCT_VERSION).build()); + client = new AsyncTelemetryClient(server.configBuilder("shop-ui", PRODUCT_VERSION).build()); client.track("checkout-started"); client.close(); } diff --git a/src/test/java/com/exasol/telemetry/StatusLoggingIT.java b/src/test/java/com/exasol/telemetry/StatusLoggingIT.java index 4d65b36..303af7a 100644 --- a/src/test/java/com/exasol/telemetry/StatusLoggingIT.java +++ b/src/test/java/com/exasol/telemetry/StatusLoggingIT.java @@ -17,18 +17,16 @@ class StatusLoggingIT { private static final String VERSION = "1.2.3"; private static final String FEATURE = "myFeature"; - @SuppressWarnings("java:S3416") // Using captured logger name by intention - private static final Logger CAPTURED_LOGGER = Logger.getLogger(TelemetryClient.class.getName()); - @Test void logsWhenTelemetryIsEnabled() throws Exception { - try (LogCapture capture = new LogCapture(CAPTURED_LOGGER); + try (LogCapture capture = new LogCapture(); RecordingHttpServer server = RecordingHttpServer.createSuccessServer(); TelemetryClient client = TelemetryClient.create(server.configBuilder(PROJECT_TAG, VERSION).build())) { final LogRecord enabledRecord = capture.await(logRecord -> logRecord.getLevel() == Level.INFO && logRecord.getMessage().contains("Telemetry is enabled"), Duration.ofSeconds(1)); - assertThat(client.isRunning(), is(true)); + assertThat(client, instanceOf(AsyncTelemetryClient.class)); + assertThat(((AsyncTelemetryClient) client).isRunning(), is(true)); assertThat(enabledRecord.getMessage(), containsString("Set EXASOL_TELEMETRY_DISABLE to any non-empty value to disable telemetry.")); assertThat(enabledRecord.getMessage(), containsString("EXASOL_TELEMETRY_DISABLE=")); assertThat(enabledRecord.getMessage(), containsString("CI=")); @@ -37,7 +35,7 @@ void logsWhenTelemetryIsEnabled() throws Exception { @Test void logsWhenTelemetryIsDisabledWithMechanism() throws Exception { - try (LogCapture capture = new LogCapture(CAPTURED_LOGGER); + try (LogCapture capture = new LogCapture(); RecordingHttpServer server = RecordingHttpServer.createSuccessServer(); TelemetryClient client = TelemetryClient.create(server.configBuilder(PROJECT_TAG, VERSION) .environment(new MapEnvironment(Map.of(TelemetryConfig.DISABLED_ENV, "disabled"))) @@ -49,7 +47,7 @@ void logsWhenTelemetryIsDisabledWithMechanism() throws Exception { assertThat(envRecord.getMessage(), containsString("EXASOL_TELEMETRY_DISABLE='disabled'")); } - try (LogCapture capture = new LogCapture(CAPTURED_LOGGER); + try (LogCapture capture = new LogCapture(); RecordingHttpServer server = RecordingHttpServer.createSuccessServer(); TelemetryClient client = TelemetryClient.create(server.configBuilder(PROJECT_TAG, VERSION) .environment(new MapEnvironment(Map.of(TelemetryConfig.CI_ENV, "github-actions"))) @@ -64,7 +62,7 @@ void logsWhenTelemetryIsDisabledWithMechanism() throws Exception { @Test void logsSentMessageCount() throws Exception { - try (LogCapture capture = new LogCapture(CAPTURED_LOGGER); + try (LogCapture capture = new LogCapture(); RecordingHttpServer server = RecordingHttpServer.createSuccessServer()) { final TelemetryClient client = TelemetryClient.create(server.configBuilder(PROJECT_TAG, VERSION) .build()); @@ -83,7 +81,7 @@ void logsSentMessageCount() throws Exception { @Test void logsWhenTelemetrySendingFails() throws Exception { - try (LogCapture capture = new LogCapture(CAPTURED_LOGGER); + try (LogCapture capture = new LogCapture(); RecordingHttpServer server = RecordingHttpServer.createFlakyServer(1)) { final TelemetryClient client = TelemetryClient.create(server.configBuilder(PROJECT_TAG, VERSION) .retryTimeout(Duration.ofMillis(500)) @@ -107,7 +105,7 @@ void logsWhenTelemetrySendingFails() throws Exception { @Test void logsWhenTelemetryStops() throws Exception { - try (LogCapture capture = new LogCapture(CAPTURED_LOGGER); + try (LogCapture capture = new LogCapture(); RecordingHttpServer server = RecordingHttpServer.createSuccessServer()) { final TelemetryClient client = TelemetryClient.create(server.configBuilder(PROJECT_TAG, VERSION).build()); try { @@ -127,8 +125,8 @@ private static final class LogCapture extends Handler implements AutoCloseable { private final Level originalLevel; private final boolean originalUseParentHandlers; - private LogCapture(final Logger logger) { - this.logger = logger; + private LogCapture() { + this.logger = Logger.getLogger("com.exasol.telemetry"); this.originalLevel = logger.getLevel(); this.originalUseParentHandlers = logger.getUseParentHandlers(); logger.setLevel(Level.ALL); diff --git a/src/test/java/com/exasol/telemetry/TelemetryClientTest.java b/src/test/java/com/exasol/telemetry/TelemetryClientTest.java index 358d21f..c0b29c3 100644 --- a/src/test/java/com/exasol/telemetry/TelemetryClientTest.java +++ b/src/test/java/com/exasol/telemetry/TelemetryClientTest.java @@ -1,11 +1,10 @@ package com.exasol.telemetry; import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import java.net.URI; -import java.time.Duration; import java.util.Map; import org.junit.jupiter.api.Test; @@ -13,7 +12,7 @@ class TelemetryClientTest { // [utest~telemetry-client-disabled-tracking~1->req~tracking-controls~1] @Test - void doesNotRunSenderWhenTrackingIsDisabled() throws Exception { + void doesNotRunSenderWhenTrackingIsDisabled() { final TelemetryConfig config = TelemetryConfig.builder("project", "1.2.3").endpoint(URI.create("https://example.com")) .environment(new MapEnvironment(Map.of(TelemetryConfig.DISABLED_ENV, "true"))) .build(); @@ -21,8 +20,7 @@ void doesNotRunSenderWhenTrackingIsDisabled() throws Exception { final TelemetryClient client = TelemetryClient.create(config); try { client.track("feature"); - assertThat(client.awaitStopped(Duration.ofMillis(10)), is(true)); - assertThat(client.isRunning(), is(false)); + assertThat(client, instanceOf(NoOpTelemetryClient.class)); } finally { client.close(); } diff --git a/src/test/java/com/exasol/telemetry/TrackingApiIT.java b/src/test/java/com/exasol/telemetry/TrackingApiIT.java index f8bca83..b903254 100644 --- a/src/test/java/com/exasol/telemetry/TrackingApiIT.java +++ b/src/test/java/com/exasol/telemetry/TrackingApiIT.java @@ -74,11 +74,10 @@ void makesDisabledTrackingNoOpWithoutTelemetryOverhead() throws Exception { .environment(new MapEnvironment(Map.of(TelemetryConfig.DISABLED_ENV, "disabled"))) .build(); - try (TelemetryClient client = new TelemetryClient(config, new FailingClock())) { + try (TelemetryClient client = TelemetryClient.create(config)) { client.track(FEATURE); - assertThat(client.awaitStopped(Duration.ofMillis(10)), is(true)); - assertThat(client.isRunning(), is(false)); + assertThat(client, instanceOf(NoOpTelemetryClient.class)); assertThat(server.awaitRequests(1, Duration.ofMillis(150)), empty()); } } @@ -110,21 +109,4 @@ void ignoresNullFeatureNames() throws Exception { assertThat(server.awaitRequests(1, Duration.ofMillis(150)), empty()); } } - - private static final class FailingClock extends Clock { - @Override - public ZoneId getZone() { - return ZoneOffset.UTC; - } - - @Override - public Clock withZone(final ZoneId zone) { - return this; - } - - @Override - public Instant instant() { - throw new AssertionError("disabled tracking should not capture a timestamp"); - } - } }