From 0bf5e7745d3874d9282f88623270a0f8beb9b239 Mon Sep 17 00:00:00 2001 From: Hardik Bajaj <58038410+hardikbajaj@users.noreply.github.com> Date: Wed, 17 Apr 2024 18:35:31 +0530 Subject: [PATCH] Add configurable parameters for statsd client (#16283) Statsd client sometimes drops metrics when this queueSize of statsd client with max unprocessed messages is completely full. This causes some high cardinality metrics like per partition lag being droppped. There are multiple parameters of statsdclient that can be initialized and can help increase the load/capacity of client to not to drop metrics more frequently. Properties like queueSize, poolSize, processorWorkers and senderWorkers will now be configurable at runtime --- docs/development/extensions-contrib/statsd.md | 4 ++ .../druid/emitter/statsd/StatsDEmitter.java | 4 ++ .../emitter/statsd/StatsDEmitterConfig.java | 53 ++++++++++++++++++- .../emitter/statsd/StatsDEmitterTest.java | 23 ++++---- 4 files changed, 73 insertions(+), 11 deletions(-) diff --git a/docs/development/extensions-contrib/statsd.md b/docs/development/extensions-contrib/statsd.md index 6a3ce7a2dbc..3e5713f586f 100644 --- a/docs/development/extensions-contrib/statsd.md +++ b/docs/development/extensions-contrib/statsd.md @@ -44,6 +44,10 @@ All the configuration parameters for the StatsD emitter are under `druid.emitter |`druid.emitter.statsd.includeHost`|Flag to include the hostname as part of the metric name.|no|false| |`druid.emitter.statsd.dimensionMapPath`|JSON file defining the StatsD type, and desired dimensions for every Druid metric|no|Default mapping provided. See below.| |`druid.emitter.statsd.blankHolder`|The blank character replacement as StatsD does not support path with blank character|no|"-"| +|`druid.emitter.statsd.queueSize`|Maximum number of unprocessed messages in the message queue.|no|Default value of StatsD Client(4096)| +|`druid.emitter.statsd.poolSize`|Network packet buffer pool size.|no|Default value of StatsD Client(512)| +|`druid.emitter.statsd.processorWorkers`|The number of processor worker threads assembling buffers for submission.|no|Default value of StatsD Client(1)| +|`druid.emitter.statsd.senderWorkers`| The number of sender worker threads submitting buffers to the socket.|no|Default value of StatsD Client(1)| |`druid.emitter.statsd.dogstatsd`|Flag to enable [DogStatsD](https://docs.datadoghq.com/developers/dogstatsd/) support. Causes dimensions to be included as tags, not as a part of the metric name. `convertRange` fields will be ignored.|no|false| |`druid.emitter.statsd.dogstatsdConstantTags`|If `druid.emitter.statsd.dogstatsd` is true, the tags in the JSON list of strings will be sent with every event.|no|[]| |`druid.emitter.statsd.dogstatsdServiceAsTag`|If `druid.emitter.statsd.dogstatsd` and `druid.emitter.statsd.dogstatsdServiceAsTag` are true, druid service (e.g. `druid/broker`, `druid/coordinator`, etc) is reported as a tag (e.g. `druid_service:druid/broker`) instead of being included in metric name (e.g. `druid.broker.query.time`) and `druid` is used as metric prefix (e.g. `druid.query.time`).|no|false| diff --git a/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java b/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java index 84972f81e53..d49133044a5 100644 --- a/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java +++ b/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitter.java @@ -63,6 +63,10 @@ public class StatsDEmitter implements Emitter .hostname(config.getHostname()) .port(config.getPort()) .constantTags(config.isDogstatsd() ? config.getDogstatsdConstantTags().toArray(new String[0]) : EMPTY_ARRAY) + .queueSize(config.getQueueSize()) + .bufferPoolSize(config.getPoolSize()) + .processorWorkers(config.getProcessorWorkers()) + .senderWorkers(config.getSenderWorkers()) .errorHandler(new StatsDClientErrorHandler() { private int exceptionCount = 0; diff --git a/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitterConfig.java b/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitterConfig.java index 64c9ce6a61f..5189545b5d1 100644 --- a/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitterConfig.java +++ b/extensions-contrib/statsd-emitter/src/main/java/org/apache/druid/emitter/statsd/StatsDEmitterConfig.java @@ -22,6 +22,7 @@ package org.apache.druid.emitter.statsd; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.base.Preconditions; +import com.timgroup.statsd.NonBlockingStatsDClient; import javax.annotation.Nullable; import java.util.Collections; @@ -56,6 +57,14 @@ public class StatsDEmitterConfig private final Boolean dogstatsdServiceAsTag; @JsonProperty private final Boolean dogstatsdEvents; + @JsonProperty + private final Integer queueSize; + @JsonProperty + private final Integer poolSize; + @JsonProperty + private final Integer processorWorkers; + @JsonProperty + private final Integer senderWorkers; @JsonCreator public StatsDEmitterConfig( @@ -69,7 +78,11 @@ public class StatsDEmitterConfig @JsonProperty("dogstatsd") @Nullable Boolean dogstatsd, @JsonProperty("dogstatsdConstantTags") @Nullable List dogstatsdConstantTags, @JsonProperty("dogstatsdServiceAsTag") @Nullable Boolean dogstatsdServiceAsTag, - @JsonProperty("dogstatsdEvents") @Nullable Boolean dogstatsdEvents + @JsonProperty("dogstatsdEvents") @Nullable Boolean dogstatsdEvents, + @JsonProperty("queueSize") @Nullable Integer queueSize, + @JsonProperty("poolSize") @Nullable Integer poolSize, + @JsonProperty("processorWorkers") @Nullable Integer processorWorkers, + @JsonProperty("senderWorkers") @Nullable Integer senderWorkers ) { this.hostname = Preconditions.checkNotNull(hostname, "StatsD hostname cannot be null."); @@ -83,6 +96,10 @@ public class StatsDEmitterConfig this.dogstatsdConstantTags = dogstatsdConstantTags != null ? dogstatsdConstantTags : Collections.emptyList(); this.dogstatsdServiceAsTag = dogstatsdServiceAsTag != null ? dogstatsdServiceAsTag : false; this.dogstatsdEvents = dogstatsdEvents != null ? dogstatsdEvents : false; + this.queueSize = queueSize != null ? queueSize : NonBlockingStatsDClient.DEFAULT_QUEUE_SIZE; + this.poolSize = poolSize != null ? poolSize : NonBlockingStatsDClient.DEFAULT_POOL_SIZE; + this.processorWorkers = processorWorkers != null ? processorWorkers : NonBlockingStatsDClient.DEFAULT_PROCESSOR_WORKERS; + this.senderWorkers = senderWorkers != null ? senderWorkers : NonBlockingStatsDClient.DEFAULT_SENDER_WORKERS; } @Override @@ -121,6 +138,18 @@ public class StatsDEmitterConfig if (!Objects.equals(dogstatsdServiceAsTag, that.dogstatsdServiceAsTag)) { return false; } + if (!Objects.equals(queueSize, that.queueSize)) { + return false; + } + if (!Objects.equals(poolSize, that.poolSize)) { + return false; + } + if (!Objects.equals(processorWorkers, that.processorWorkers)) { + return false; + } + if (!Objects.equals(senderWorkers, that.senderWorkers)) { + return false; + } return Objects.equals(dogstatsdConstantTags, that.dogstatsdConstantTags); } @@ -128,7 +157,7 @@ public class StatsDEmitterConfig public int hashCode() { return Objects.hash(hostname, port, prefix, separator, includeHost, dimensionMapPath, - blankHolder, dogstatsd, dogstatsdConstantTags, dogstatsdServiceAsTag); + blankHolder, dogstatsd, dogstatsdConstantTags, dogstatsdServiceAsTag, queueSize, poolSize, processorWorkers, senderWorkers); } @JsonProperty @@ -197,4 +226,24 @@ public class StatsDEmitterConfig { return dogstatsdEvents; } + @JsonProperty + public Integer getQueueSize() + { + return queueSize; + } + @JsonProperty + public Integer getPoolSize() + { + return poolSize; + } + @JsonProperty + public Integer getProcessorWorkers() + { + return processorWorkers; + } + @JsonProperty + public Integer getSenderWorkers() + { + return senderWorkers; + } } diff --git a/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java b/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java index 05a185c8b7e..3f33cef88aa 100644 --- a/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java +++ b/extensions-contrib/statsd-emitter/src/test/java/org/apache/druid/emitter/statsd/StatsDEmitterTest.java @@ -43,7 +43,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null, null, null, null), + new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null, null, null, null, null, null, null, null), new ObjectMapper(), client ); @@ -60,7 +60,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, true, null, null, null), + new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, true, null, null, null, null, null, null, null), new ObjectMapper(), client ); @@ -77,7 +77,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null, null, null, null), + new StatsDEmitterConfig("localhost", 8888, null, null, null, null, null, null, null, null, null, null, null, null, null), new ObjectMapper(), client ); @@ -103,7 +103,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, null, null, null, null), + new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, null, null, null, null, null, null, null, null), new ObjectMapper(), client ); @@ -129,7 +129,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, true, null, null, null), + new StatsDEmitterConfig("localhost", 8888, null, "#", true, null, null, true, null, null, null, null, null, null, null), new ObjectMapper(), client ); @@ -156,7 +156,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, null, null, null, null), + new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, null, null, null, null, null, null, null, null), new ObjectMapper(), client ); @@ -173,7 +173,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, true, null, true, null), + new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, true, null, true, null, null, null, null, null), new ObjectMapper(), client ); @@ -192,7 +192,7 @@ public class StatsDEmitterTest { StatsDClient client = mock(StatsDClient.class); StatsDEmitter emitter = new StatsDEmitter( - new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, true, null, true, true), + new StatsDEmitterConfig("localhost", 8888, null, null, true, null, null, true, null, true, true, null, null, null, null), new ObjectMapper(), client ); @@ -239,7 +239,12 @@ public class StatsDEmitterTest true, ImmutableList.of("tag1", "value1"), true, - true + true, + 5100, + 512, + 1, + 1 + ); try (StatsDEmitter emitter = StatsDEmitter.of(config, new ObjectMapper())) {