diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/Sketch.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/Sketch.java index ab9de8b4..7c61f007 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/Sketch.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/Sketch.java @@ -13,11 +13,10 @@ * Reusable DDSketch builder. Consumes a batch of observations and populates sum, min, max, count * and distribution bins accordingly. * - *

This implementation maintains at most 4096 bins with 64-bit counters. Number of bins is a hard - * limit and is enforced by the intake. + *

This implementation maintains at most 4096 bins with 64-bit counters. * - *

Prioritizes accuracy of higher key bins (higher percentiles) over lower ones when number of - * bins exceeds the limit. + *

Prioritizes accuracy of higher key bins (higher percentiles) over lower ones when the number + * of bins exceeds the limit. */ public class Sketch { static final double gamma = 130.0 / 128; @@ -59,6 +58,12 @@ static short key(double value) { /** Receives (key, count) pairs from {@link #bins(BinConsumer)}. */ public interface BinConsumer { + /** + * Process one sketch bin. + * + * @param key a value that specifies the range of observations counted in this bin. + * @param count number of observations in the bin. + */ void consumeBin(short key, long count); } @@ -69,7 +74,11 @@ public int size() { return size; } - /** Feeds each populated bin to {@code consumer} in order. */ + /** + * Feeds each populated bin to {@code consumer} in order. + * + * @param consumer a consumer to feed sketch bins to. + */ public void bins(BinConsumer consumer) { int idx = head; for (int i = 0; i < size; i++) { @@ -114,11 +123,13 @@ public long count() { /** * Builds the sketch from the given values. * - * @param observations the observations to include in the sketch + * @param observations the observations to include in the sketch. * @param sampleRate the sampling rate used to collect {@code observations}, in {@code (0, 1]}. * Each observation is weighted by {@code 1 / sampleRate} when accumulating counts and sums. * Rates below ~1.08e-19 saturate the per-observation weight; bin counts and the total * {@code count} field saturate at {@link Long#MAX_VALUE} on overflow. + * @throws IllegalArgumentException if {@code sampleRate} is {@code NaN}, not positive, or + * greater than 1. */ public void build(long[] observations, double sampleRate) { validateSampleRate(sampleRate); @@ -136,11 +147,13 @@ public void build(long[] observations, double sampleRate) { /** * Builds the sketch from the given values. * - * @param observations the observations to include in the sketch + * @param observations the observations to include in the sketch. * @param sampleRate the sampling rate used to collect {@code observations}, in {@code (0, 1]}. * Each observation is weighted by {@code 1 / sampleRate} when accumulating counts and sums. * Rates below ~1.08e-19 saturate the per-observation weight; bin counts and the total * {@code count} field saturate at {@link Long#MAX_VALUE} on overflow. + * @throws IllegalArgumentException if {@code sampleRate} is {@code NaN}, not positive, or + * greater than 1. */ public void build(double[] observations, double sampleRate) { validateSampleRate(sampleRate); diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/DirectHttpClient.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/DirectHttpClient.java index e0b8b5d3..7bb83bc5 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/DirectHttpClient.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/DirectHttpClient.java @@ -11,6 +11,7 @@ import com.datadoghq.dogstatsd.http.serializer.PayloadBuilder; import com.datadoghq.dogstatsd.http.serializer.PayloadConsumer; import java.net.URI; +import java.nio.BufferOverflowException; import java.util.List; import java.util.Objects; @@ -35,6 +36,7 @@ public class DirectHttpClient { * * @param forwarder the forwarder used to send payloads, required. * @return a new builder. + * @throws NullPointerException if {@code forwarder} is null. */ public static Builder builder(final Forwarder forwarder) { return new Builder(forwarder); @@ -126,6 +128,7 @@ public DirectHttpClient build() { * @param value the gauge value. * @param ts the timestamp of the point in seconds since Unix epoch. * @param tags the tags to attach to the point. + * @throws BufferOverflowException if the encoded metric exceeds the maximum payload size. */ public void gauge(String name, double value, long ts, List tags) { seriesBuilder @@ -139,12 +142,14 @@ public void gauge(String name, double value, long ts, List tags) { /** * Records a count point. * - *

For compatibility with aggregated dogstatsd counts, assumes aggregation interval of 10s. + *

For compatibility with aggregated dogstatsd counts, assumes an aggregation interval of + * 10s. * * @param name the metric name, to which the client prefix is prepended. * @param value the count accumulated over the interval starting at {@code ts}. * @param ts the timestamp of the point in seconds since Unix epoch. * @param tags the tags to attach to the point. + * @throws BufferOverflowException if the encoded metric exceeds the maximum payload size. */ public void count(String name, double value, long ts, List tags) { seriesBuilder @@ -163,6 +168,9 @@ public void count(String name, double value, long ts, List tags) { * @param sampleRate the sampling rate used to collect {@code values}, in {@code (0, 1]}. * @param ts the timestamp of the point in seconds since Unix epoch. * @param tags the tags to attach to the point. + * @throws IllegalArgumentException if {@code sampleRate} is {@code NaN}, not positive, or + * greater than 1. + * @throws BufferOverflowException if the encoded metric exceeds the maximum payload size. */ public void distribution( String name, double[] values, double sampleRate, long ts, List tags) { @@ -174,7 +182,11 @@ private String prefixed(final String name) { return prefix.isEmpty() ? name : prefix + name; } - /** Completes any in-progress payloads and submits them to the forwarder. */ + /** + * Completes any in-progress payloads and submits them to the forwarder. + * + * @throws BufferOverflowException if an encoded metric exceeds the maximum payload size. + */ public void flush() { seriesBuilder.close(); sketchesBuilder.close(); diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java index 60568e72..134f20fe 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java @@ -31,6 +31,12 @@ public static Builder builder() { return new Builder(); } + /** + * Returns the base URI that per-payload URIs are resolved against. Always ends with {@code /} + * so that it acts as a prefix for {@link URI#resolve}. + * + * @return the base URI, never null. + */ public URI baseUri() { return baseUri; } @@ -55,9 +61,11 @@ public String externalData() { } /** - * Returns new instance with default settings. + * Returns a new instance with default settings. * - * @return new default instance. + * @return a new default instance. + * @throws IllegalStateException if {@code DD_DOGSTATSD_HTTP_URL} is not defined. + * @throws IllegalArgumentException if the base URI is malformed. */ public static ForwarderContext defaults() { return builder().build(); @@ -112,7 +120,7 @@ public Builder originDetectionEnabled(final boolean val) { * Sets the base URI the series and sketches endpoints are resolved against. Defaults to the * value of the {@code DD_DOGSTATSD_HTTP_URL} environment variable. * - * @param val the base URI, or null to use the default. + * @param uri the base URI, or null to use the default. * @return this builder. */ public Builder baseUri(final String uri) { @@ -120,6 +128,12 @@ public Builder baseUri(final String uri) { return this; } + /** + * Use the supplied map instead of OS environment variables. + * + * @param val the environment map. + * @return this builder. + */ public Builder environment(final Map val) { env = new EnvMap(val); return this; @@ -134,7 +148,9 @@ Builder cgroupReader(final CgroupReader val) { * Builds the context, running detection for any value not set explicitly. * * @return a new context. - * @throws URISyntaxException if baseUri value is not a valid URI. + * @throws IllegalStateException if no base URI was set with {@link #baseUri} and {@code + * DD_DOGSTATSD_HTTP_URL} is not defined. + * @throws IllegalArgumentException if the base URI is malformed. */ public ForwarderContext build() { String local = localData; diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Buffer.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Buffer.java index dbc6f08b..039f2e84 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Buffer.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Buffer.java @@ -41,7 +41,7 @@ void clear() { size = 0; } - /** Return true if buf is null or empty */ + /** Return true if buf is null or empty. */ static boolean isEmpty(Buffer buf) { return buf == null || buf.size == 0; } diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Metric.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Metric.java index 913ec5b9..ac28a0e1 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Metric.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/Metric.java @@ -42,6 +42,7 @@ public T setTags(List tags) { * @param resources List of even length, containing zero or more (type, name) pairs, or null for * no resources. * @return This. + * @throws IllegalArgumentException if {@code resources} has an odd number of elements. */ public T setResources(List resources) { if (resources != null && resources.size() % 2 != 0) { @@ -100,7 +101,13 @@ void encodeDependentFields() { pb.encodeOrigin(origin); } - /** Finish this timeseries and add it to the payload. */ + /** + * Finish this timeseries and add it to the payload. + * + * @throws java.nio.BufferOverflowException if the encoded metric exceeds the maximum payload + * size. The in-progress metric is discarded, so no {@link PayloadBuilder#resetMetric()} is + * needed before encoding further metrics. + */ public void close() { pb.endMetric(); } diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadBuilder.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadBuilder.java index a21983f7..f4012b11 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadBuilder.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadBuilder.java @@ -135,7 +135,7 @@ public void encode(Origin o) { Metric metricInProgress; /** - * Create new PayloadBuilder. + * Create a new PayloadBuilder. * * @param consumer Is given payloads one by one as they are finished. */ @@ -145,12 +145,13 @@ public PayloadBuilder(PayloadConsumer consumer) { } /** - * Begin encoding new count metric. + * Begin encoding a new count metric. * *

Only one metric can be encoded at a time. * * @param name Name of the metric. - * @return Builder instance. + * @return New builder instance. + * @throws BufferOverflowException if finishing the previous metric overflows the payload. */ public ScalarMetric count(String name) { ScalarMetric m = new ScalarMetric(this, 1, name); @@ -159,12 +160,13 @@ public ScalarMetric count(String name) { } /** - * Begin encoding new rate metric. + * Begin encoding a new rate metric. * *

Only one metric can be encoded at a time. * * @param name Name of the metric. * @return New builder instance. + * @throws BufferOverflowException if finishing the previous metric overflows the payload. */ public ScalarMetric rate(String name) { ScalarMetric m = new ScalarMetric(this, 2, name); @@ -173,12 +175,13 @@ public ScalarMetric rate(String name) { } /** - * Begin encoding new gauge metric. + * Begin encoding a new gauge metric. * *

Only one metric can be encoded at a time. * * @param name Name of the metric. * @return New builder instance. + * @throws BufferOverflowException if finishing the previous metric overflows the payload. */ public ScalarMetric gauge(String name) { ScalarMetric m = new ScalarMetric(this, 3, name); @@ -187,12 +190,13 @@ public ScalarMetric gauge(String name) { } /** - * Begin encoding new sketch metric. + * Begin encoding a new sketch metric. * *

Only one metric can be encoded at a time. * * @param name Name of the metric. * @return New builder instance. + * @throws BufferOverflowException if finishing the previous metric overflows the payload. */ public SketchMetric sketch(String name) { SketchMetric m = new SketchMetric(this, 4, name); @@ -305,7 +309,11 @@ void flushPayload() { timestampsDelta.clear(); } - /** Finish any pending data. */ + /** + * Finish any pending data. + * + * @throws BufferOverflowException if finishing the in-progress metric overflows the payload. + */ public void close() { endMetric(); flushPayload(); diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadConsumer.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadConsumer.java index 594afb03..64106aad 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadConsumer.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/PayloadConsumer.java @@ -10,7 +10,7 @@ /** Consumes payloads from the PayloadBuilder. */ public interface PayloadConsumer { /** - * Called when payload builder finishes another payload. + * Called when the payload builder finishes another payload. * * @param payload Completed payload. */ diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/ScalarMetric.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/ScalarMetric.java index 2433b0ca..201bcba2 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/ScalarMetric.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/ScalarMetric.java @@ -19,7 +19,7 @@ protected ScalarMetric self() { } /** - * Add new data point to the timeseries. + * Add a new data point to the timeseries. * * @param timestamp Timestamp of the point in seconds since Unix epoch. * @param value Metric value at timestamp. diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/SketchMetric.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/SketchMetric.java index 3ace6018..e74afbf9 100644 --- a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/SketchMetric.java +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/serializer/SketchMetric.java @@ -53,6 +53,9 @@ protected SketchMetric self() { * @param timestamp Timestamp of the point in seconds since Unix epoch. * @param sketch Sketch supplying the summary statistics and bin distribution. * @return This. + * @throws BufferOverflowException if the sketch's bin data alone would exceed the maximum + * payload size. The metric is no longer valid; call {@link PayloadBuilder#resetMetric()} + * before encoding any further metrics. */ public SketchMetric addPoint(long timestamp, Sketch sketch) { // Skip doing the work if just the bin data would exceed payload size limit. diff --git a/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java b/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java index c763e98a..8afd3208 100644 --- a/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java +++ b/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java @@ -76,6 +76,8 @@ public static Builder builder() { /** * Captures a snapshot of the forwarder's telemetry counters and queue state, clearing delta * counters so subsequent snapshots report activity since this call. + * + * @return a telemetry snapshot. */ public Telemetry.Snapshot snapshot() { return telemetry.snapshot(queue); @@ -102,14 +104,17 @@ public void run() { /** * Enqueues a payload for delivery to the given endpoint. * - *

If the queue is full, behaviour is determined by the {@link WhenFull} policy set with + *

If the queue is full, behavior is determined by the {@link WhenFull} policy set with * {@link Builder#whenFull}. * - * @param url the remote HTTP endpoint to POST the payload to - * @param payload the raw bytes to deliver + * @param url the remote HTTP endpoint to POST the payload to. + * @param payload the raw bytes to deliver. * @throws InterruptedException if the calling thread is interrupted while waiting for space - * ({@link WhenFull#BLOCK} mode only) - * @throws IllegalStateException if the forwarder has been closed via {@link #close(Duration)} + * ({@link WhenFull#BLOCK} mode only). + * @throws IllegalStateException if the forwarder has been closed via {@link #close(Duration)}. + * @throws IllegalArgumentException if {@code payload} is larger than the queue's {@link + * Builder#maxRequestsBytes} limit, in which case it can never be delivered. + * @throws NullPointerException if {@code url} or {@code payload} is null. */ public void send(URI url, byte[] payload) throws InterruptedException { Objects.requireNonNull(url, "url"); @@ -213,8 +218,8 @@ void backoff() throws InterruptedException { * @param timeout maximum time to wait for the backlog to drain. {@code null} means wait * forever. * @return {@code true} if the queue drained cleanly with no unsent payloads remaining; {@code - * false} if the timeout elapsed with data still queued - * @throws InterruptedException if the calling thread is interrupted while waiting + * false} if the timeout elapsed with data still queued. + * @throws InterruptedException if the calling thread is interrupted while waiting. */ public boolean close(Duration timeout) throws InterruptedException { queue.close(); @@ -254,6 +259,7 @@ private Builder() {} * * @param val the maximum number of buffered bytes; must be positive. * @return this builder. + * @throws IllegalArgumentException if {@code val} is not positive. */ public Builder maxRequestsBytes(final long val) { if (val <= 0) { @@ -268,6 +274,7 @@ public Builder maxRequestsBytes(final long val) { * * @param val the maximum number of attempts; must be at least 1. * @return this builder. + * @throws IllegalArgumentException if {@code val} is less than 1. */ public Builder maxTries(final long val) { if (val < 1) { @@ -282,6 +289,7 @@ public Builder maxTries(final long val) { * * @param val the action to take. * @return this builder. + * @throws NullPointerException if {@code val} is null. */ public Builder whenFull(final WhenFull val) { whenFull = Objects.requireNonNull(val, "whenFull"); @@ -293,6 +301,8 @@ public Builder whenFull(final WhenFull val) { * * @param val the connect timeout; must be positive. * @return this builder. + * @throws NullPointerException if {@code val} is null. + * @throws IllegalArgumentException if {@code val} is not positive. */ public Builder connectTimeout(final Duration val) { Objects.requireNonNull(val, "connectTimeout"); @@ -310,6 +320,7 @@ public Builder connectTimeout(final Duration val) { * @param val the request timeout, or {@code null} to disable it; must be positive when * non-null. * @return this builder. + * @throws IllegalArgumentException if {@code val} is non-null and not positive. */ public Builder requestTimeout(final Duration val) { if (val != null && (val.isNegative() || val.isZero())) { @@ -326,6 +337,9 @@ public Builder requestTimeout(final Duration val) { * * @param context the context to take the values from. * @return this builder. + * @throws NullPointerException if {@code context} is null. + * @throws IllegalArgumentException if the context's local or external data contains + * characters that are not valid in an HTTP header value. */ public Builder context(final ForwarderContext context) { contextSet = true; @@ -341,6 +355,8 @@ public Builder context(final ForwarderContext context) { * started yet. * * @return a new forwarder. + * @throws IllegalStateException if no context was set with {@link #context} and the default + * one cannot be built because {@code DD_DOGSTATSD_HTTP_URL} is not defined. */ public Forwarder build() { if (!contextSet) { diff --git a/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Telemetry.java b/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Telemetry.java index 574cdfa8..820298e2 100644 --- a/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Telemetry.java +++ b/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Telemetry.java @@ -18,7 +18,7 @@ public class Telemetry { /** HTTP status code used to record transport-level (no-response) errors. */ public static final int TRANSPORT_ERROR_CODE = 0; - /** Point-in-time view of cumulative counters and queue state. */ + /** Queue state at snapshot time, plus counters for the interval leading up to it. */ public static final class Snapshot { /** * Wall-clock time (Unix epoch milliseconds) at the start of the interval covered by this @@ -27,17 +27,36 @@ public static final class Snapshot { */ public long intervalStartMillis; + /** Number of payloads added to the queue in this interval. */ public long enqueuedPayloads; + + /** Number of payloads successfully delivered in this interval. */ public long deliveredPayloads; + + /** Total size in bytes of payloads added to the queue in this interval. */ public long enqueuedBytes; + + /** Total size in bytes of payloads successfully delivered in this interval. */ public long deliveredBytes; + + /** Number of payloads currently in the queue. */ public long queuePayloads; + + /** Total size in bytes of payloads currently in the queue. */ public long queueBytes; + + /** Maximum number of bytes the queue is allowed to store. */ public long queueMaxBytes; + + /** Number of payloads dropped in this interval. */ public long droppedPayloads; + + /** Total size in bytes of payloads dropped in this interval. */ public long droppedBytes; - /** Nanos elapsed since the oldest queued item was enqueued; {@code 0} if queue is empty. */ + /** + * Nanos elapsed since the oldest queued item was enqueued; {@code 0} if the queue is empty. + */ public long oldestEnqueuedAgeNanos; /** Nanos elapsed since the last successful submission; {@code 0} if none yet. */ @@ -46,7 +65,7 @@ public static final class Snapshot { /** Totals keyed by HTTP code. */ public Map byCode = new HashMap<>(); - /** Default metric name prefix used when none is supplied to {@link Snapshot#encode}. */ + /** Default metric name prefix used when none is supplied to {@link Snapshot#encodeTo}. */ static final String DEFAULT_PREFIX = "datadog.dogstatsd_http.client"; Snapshot(long intervalStartMillis) { @@ -115,7 +134,10 @@ public void encodeTo(String prefix, Encoder enc) { /** Per-code totals within a snapshot's window. */ public static final class CodeCounters { + /** Number of payloads. */ public long payloads; + + /** Total size in bytes of payloads. */ public long bytes; } } @@ -128,7 +150,7 @@ public static final class CodeCounters { private long lastSuccessNanos; private boolean everDelivered; - public Telemetry() { + Telemetry() { this(Clock.systemUTC(), System::nanoTime); } @@ -167,11 +189,7 @@ synchronized void onDrop(long payloads, long bytes) { current.droppedBytes += bytes; } - /** - * Captures a snapshot using the supplied queue stats, then swaps in a fresh accumulator so - * subsequent snapshots report deltas since this call. - */ - public synchronized Snapshot snapshot(BoundedQueue q) { + synchronized Snapshot snapshot(BoundedQueue q) { long now = nanos.getAsLong(); Snapshot s = current; current = new Snapshot(clock.millis());