You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by le...@apache.org on 2018/10/15 20:30:35 UTC
[incubator-druid] branch master updated: Emit
emitter/buffers/allocated/delta and emitter/buffers/failed/delta (#6425)
This is an automated email from the ASF dual-hosted git repository.
leventov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new b57d5df Emit emitter/buffers/allocated/delta and emitter/buffers/failed/delta (#6425)
b57d5df is described below
commit b57d5df07479174f9a8bbbab7b131040457ce6c6
Author: Roman Leventov <le...@gmail.com>
AuthorDate: Mon Oct 15 17:30:29 2018 -0300
Emit emitter/buffers/allocated/delta and emitter/buffers/failed/delta (#6425)
@gianm gauge `emitter/buffers/failed` and `emitter/buffers/totalAllocated` metrics proved to be very inconvenient in practice: they are not additive across a fleet of nodes, are randomly reset when a JVM restarts (that may be completely independent from metrics emitting issues). Regarding [problem identification](https://github.com/apache/incubator-druid/pull/4973#issuecomment-338119125), `emitter/buffers/dropped` serves this well.
Currently in this PR I added `/delta` vs. `/total` metrics, but if there are no objections I would remove `/total` and so align `emitter/buffers/allocated` and `/failed` with `/dropped` and `/events/emitted` which are already delta-only.
---
.../java/util/emitter/core/HttpPostEmitter.java | 4 +-
.../java/util/metrics/HttpPostEmitterMonitor.java | 53 +++++++++++++++++-----
.../emitter/core/HttpPostEmitterStressTest.java | 4 +-
3 files changed, 45 insertions(+), 16 deletions(-)
diff --git a/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java b/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java
index 8f4951e..61dc66b 100644
--- a/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java
+++ b/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java
@@ -882,12 +882,12 @@ public class HttpPostEmitter implements Flushable, Closeable, Emitter
return approximateBuffersToReuseCount.get();
}
- public int getFailedBuffers()
+ public int getTotalFailedBuffers()
{
return emittingThread.approximateFailedBuffersCount.get();
}
- public int getDroppedBuffers()
+ public int getTotalDroppedBuffers()
{
return droppedBuffers.get();
}
diff --git a/core/src/main/java/org/apache/druid/java/util/metrics/HttpPostEmitterMonitor.java b/core/src/main/java/org/apache/druid/java/util/metrics/HttpPostEmitterMonitor.java
index a8fd802..e925fab 100644
--- a/core/src/main/java/org/apache/druid/java/util/metrics/HttpPostEmitterMonitor.java
+++ b/core/src/main/java/org/apache/druid/java/util/metrics/HttpPostEmitterMonitor.java
@@ -31,7 +31,9 @@ public class HttpPostEmitterMonitor extends FeedDefiningMonitor
private final ImmutableMap<String, String> extraDimensions;
private final ServiceMetricEvent.Builder builder;
private long lastTotalEmittedEvents = 0;
- private int lastDroppedBuffers = 0;
+ private int lastTotalDroppedBuffers = 0;
+ private int lastTotalAllocatedBuffers = 0;
+ private int lastTotalFailedBuffers = 0;
public HttpPostEmitterMonitor(
String feed,
@@ -48,15 +50,10 @@ public class HttpPostEmitterMonitor extends FeedDefiningMonitor
@Override
public boolean doMonitor(ServiceEmitter emitter)
{
- long newTotalEmittedEvents = httpPostEmitter.getTotalEmittedEvents();
- long totalEmittedEventsDiff = newTotalEmittedEvents - lastTotalEmittedEvents;
- emitter.emit(builder.build("emitter/events/emitted", totalEmittedEventsDiff));
- lastTotalEmittedEvents = newTotalEmittedEvents;
-
- int newDroppedBuffers = httpPostEmitter.getDroppedBuffers();
- int droppedBuffersDiff = newDroppedBuffers - lastDroppedBuffers;
- emitter.emit(builder.build("emitter/buffers/dropped", droppedBuffersDiff));
- lastDroppedBuffers = newDroppedBuffers;
+ emitEmittedEvents(emitter);
+ emitDroppedBuffers(emitter);
+ emitAllocatedBuffers(emitter);
+ emitFailedBuffers(emitter);
emitTimeCounterMetrics(emitter, httpPostEmitter.getBatchFillingTimeCounter(), "emitter/batchFilling/");
emitTimeCounterMetrics(emitter, httpPostEmitter.getSuccessfulSendingTimeCounter(), "emitter/successfulSending/");
@@ -64,14 +61,46 @@ public class HttpPostEmitterMonitor extends FeedDefiningMonitor
emitter.emit(builder.build("emitter/events/emitQueue", httpPostEmitter.getEventsToEmit()));
emitter.emit(builder.build("emitter/events/large/emitQueue", httpPostEmitter.getLargeEventsToEmit()));
- emitter.emit(builder.build("emitter/buffers/totalAllocated", httpPostEmitter.getTotalAllocatedBuffers()));
+
emitter.emit(builder.build("emitter/buffers/emitQueue", httpPostEmitter.getBuffersToEmit()));
- emitter.emit(builder.build("emitter/buffers/failed", httpPostEmitter.getFailedBuffers()));
+
emitter.emit(builder.build("emitter/buffers/reuseQueue", httpPostEmitter.getBuffersToReuse()));
return true;
}
+ private void emitEmittedEvents(ServiceEmitter emitter)
+ {
+ long newTotalEmittedEvents = httpPostEmitter.getTotalEmittedEvents();
+ long emittedEventsDelta = newTotalEmittedEvents - lastTotalEmittedEvents;
+ emitter.emit(builder.build("emitter/events/emitted/delta", emittedEventsDelta));
+ lastTotalEmittedEvents = newTotalEmittedEvents;
+ }
+
+ private void emitDroppedBuffers(ServiceEmitter emitter)
+ {
+ int newTotalDroppedBuffers = httpPostEmitter.getTotalDroppedBuffers();
+ int droppedBuffersDelta = newTotalDroppedBuffers - lastTotalDroppedBuffers;
+ emitter.emit(builder.build("emitter/buffers/dropped/delta", droppedBuffersDelta));
+ lastTotalDroppedBuffers = newTotalDroppedBuffers;
+ }
+
+ private void emitAllocatedBuffers(ServiceEmitter emitter)
+ {
+ int newTotalAllocatedBuffers = httpPostEmitter.getTotalAllocatedBuffers();
+ int allocatedBuffersDelta = newTotalAllocatedBuffers - lastTotalAllocatedBuffers;
+ emitter.emit(builder.build("emitter/buffers/allocated/delta", allocatedBuffersDelta));
+ lastTotalAllocatedBuffers = newTotalAllocatedBuffers;
+ }
+
+ private void emitFailedBuffers(ServiceEmitter emitter)
+ {
+ int newTotalFailedBuffers = httpPostEmitter.getTotalFailedBuffers();
+ int failedBuffersDelta = newTotalFailedBuffers - lastTotalFailedBuffers;
+ emitter.emit(builder.build("emitter/buffers/failed/delta", failedBuffersDelta));
+ lastTotalFailedBuffers = newTotalFailedBuffers;
+ }
+
private void emitTimeCounterMetrics(ServiceEmitter emitter, ConcurrentTimeCounter timeCounter, String metricNameBase)
{
long timeSumAndCount = timeCounter.getTimeSumAndCountAndReset();
diff --git a/core/src/test/java/org/apache/druid/java/util/emitter/core/HttpPostEmitterStressTest.java b/core/src/test/java/org/apache/druid/java/util/emitter/core/HttpPostEmitterStressTest.java
index 08e5d9d..94c9d5b 100644
--- a/core/src/test/java/org/apache/druid/java/util/emitter/core/HttpPostEmitterStressTest.java
+++ b/core/src/test/java/org/apache/druid/java/util/emitter/core/HttpPostEmitterStressTest.java
@@ -227,7 +227,7 @@ public class HttpPostEmitterStressTest
emitter.emit(smallEvent);
- Assert.assertTrue(emitter.getFailedBuffers() <= 10);
+ Assert.assertTrue(emitter.getTotalFailedBuffers() <= 10);
Assert.assertTrue(emitter.getBuffersToEmit() <= 12);
}
threadsCompleted.countDown();
@@ -241,7 +241,7 @@ public class HttpPostEmitterStressTest
emitter.emit(bigEvent);
- Assert.assertTrue(emitter.getFailedBuffers() <= 10);
+ Assert.assertTrue(emitter.getTotalFailedBuffers() <= 10);
Assert.assertTrue(emitter.getBuffersToEmit() <= 12);
}
threadsCompleted.countDown();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org