This is an automated email from the ASF dual-hosted git repository.
xvrl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 995d99d add ingest/notices/queueSize metric to give visibility into
supervisor notices queue size (#11417)
995d99d is described below
commit 995d99d9e44145fcb89b2fd6060f4d38f6c39e21
Author: Harini Rajendran <hrajend...@confluent.io>
AuthorDate: Fri Jul 30 09:59:26 2021 -0500
add ingest/notices/queueSize metric to give visibility into supervisor
notices queue size (#11417)
---
docs/operations/metrics.md | 2 +
.../supervisor/SeekableStreamSupervisor.java | 88 ++++++++++-
.../SeekableStreamSupervisorStateTest.java | 170 +++++++++++++++++----
website/.spelling | 1 +
4 files changed, 228 insertions(+), 33 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index d4ca7ad..0ed0ff4 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -181,6 +181,8 @@ These metrics are only available if the
RealtimeMetricsMonitor is included in th
|`ingest/handoff/count`|Number of handoffs that happened.|dataSource, taskId,
taskType.|Varies. Generally greater than 0 once every segment granular period
if cluster operating normally|
|`ingest/sink/count`|Number of sinks not handoffed.|dataSource, taskId,
taskType.|1~3|
|`ingest/events/messageGap`|Time gap between the data time in event and
current system time.|dataSource, taskId, taskType.|Greater than 0, depends on
the time carried in event |
+|`ingest/notices/queueSize`|Number of pending notices to be processed by the
coordinator|dataSource.|Typically 0 and occasionally in lower single digits.
Should not be a very high number. |
+|`ingest/notices/time`|Milliseconds taken to process a notice by the
supervisor|dataSource, noticeType.| < 1s. |
Note: If the JVM does not support CPU time measurement for the current thread,
ingest/merge/cpu and ingest/persists/cpu will be 0.
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index d5e7257..ec2a526 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -274,6 +274,13 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
*/
private interface Notice
{
+ /**
+ * Returns a descriptive label for this notice type. Used for metrics
emission and logging.
+ *
+ * @return task type label
+ */
+ String getType();
+
void handle() throws ExecutionException, InterruptedException,
TimeoutException;
}
@@ -312,6 +319,8 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
private class RunNotice implements Notice
{
+ private static final String TYPE = "run_notice";
+
@Override
public void handle()
{
@@ -323,12 +332,19 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
runInternal();
}
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
}
// change taskCount without resubmitting.
private class DynamicAllocationTasksNotice implements Notice
{
Callable<Integer> scaleAction;
+ private static final String TYPE = "dynamic_allocation_tasks_notice";
DynamicAllocationTasksNotice(Callable<Integer> scaleAction)
{
@@ -382,6 +398,12 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
}
}
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
}
/**
@@ -458,6 +480,8 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
private class ShutdownNotice implements Notice
{
+ private static final String TYPE = "shutdown_notice";
+
@Override
public void handle() throws InterruptedException, ExecutionException,
TimeoutException
{
@@ -468,11 +492,18 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
stopLock.notifyAll();
}
}
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
}
private class ResetNotice implements Notice
{
final DataSourceMetadata dataSourceMetadata;
+ private static final String TYPE = "reset_notice";
ResetNotice(DataSourceMetadata dataSourceMetadata)
{
@@ -484,12 +515,19 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
{
resetInternal(dataSourceMetadata);
}
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
}
protected class CheckpointNotice implements Notice
{
private final int taskGroupId;
private final SeekableStreamDataSourceMetadata<PartitionIdType,
SequenceOffsetType> checkpointMetadata;
+ private static final String TYPE = "checkpoint_notice";
CheckpointNotice(
int taskGroupId,
@@ -560,6 +598,12 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
return true;
}
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
}
// Map<{group id}, {actively reading task group}>; see documentation for
TaskGroup class
@@ -908,7 +952,9 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
notice.handle();
Instant handleNoticeEndTime = Instant.now();
Duration timeElapsed =
Duration.between(handleNoticeStartTime, handleNoticeEndTime);
- log.debug("Handled notice [%s] from notices queue in [%d]
ms, current notices queue size [%d]", notice.getClass().getName(),
timeElapsed.toMillis(), getNoticesQueueSize());
+ String noticeType = notice.getType();
+ log.debug("Handled notice [%s] from notices queue in [%d]
ms, current notices queue size [%d] for datasource [%s]", noticeType,
timeElapsed.toMillis(), getNoticesQueueSize(), dataSource);
+ emitNoticeProcessTime(noticeType, timeElapsed.toMillis());
}
catch (Throwable e) {
stateManager.recordThrowableEvent(e);
@@ -3588,6 +3634,7 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
/**
* default implementation, schedules periodic fetch of latest offsets and
{@link #emitLag} reporting for Kafka and Kinesis
+ * and periodic reporting of {@Link #emitNoticesQueueSize} for various data
sources.
*/
protected void scheduleReporting(ScheduledExecutorService reportingExec)
{
@@ -3610,6 +3657,12 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
spec.getMonitorSchedulerConfig().getEmitterPeriod().getMillis(),
TimeUnit.MILLISECONDS
);
+ reportingExec.scheduleAtFixedRate(
+ this::emitNoticesQueueSize,
+ ioConfig.getStartDelay().getMillis() +
INITIAL_EMIT_LAG_METRIC_DELAY_MILLIS, // wait for tasks to start up
+ spec.getMonitorSchedulerConfig().getEmitterPeriod().getMillis(),
+ TimeUnit.MILLISECONDS
+ );
}
/**
@@ -3658,6 +3711,39 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
&&
makeSequenceNumber(earliestOffset).compareTo(makeSequenceNumber(offsetFromMetadata))
<= 0;
}
+ protected void emitNoticeProcessTime(String noticeType, long timeInMillis)
+ {
+ try {
+ emitter.emit(
+ ServiceMetricEvent.builder()
+ .setDimension("noticeType", noticeType)
+ .setDimension("dataSource", dataSource)
+ .build("ingest/notices/time", timeInMillis)
+ );
+ }
+ catch (Exception e) {
+ log.warn(e, "Unable to emit notices process time");
+ }
+ }
+
+ protected void emitNoticesQueueSize()
+ {
+ if (spec.isSuspended()) {
+ // don't emit metrics if supervisor is suspended
+ return;
+ }
+ try {
+ emitter.emit(
+ ServiceMetricEvent.builder()
+ .setDimension("dataSource", dataSource)
+ .build("ingest/notices/queueSize", getNoticesQueueSize())
+ );
+ }
+ catch (Exception e) {
+ log.warn(e, "Unable to emit notices queue size");
+ }
+ }
+
protected void emitLag()
{
if (spec.isSuspended() || !stateManager.isSteadyState()) {
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
index 91dd10a..5f5d47f 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
@@ -88,6 +88,8 @@ import javax.annotation.Nullable;
import java.io.File;
import java.math.BigInteger;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -97,6 +99,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
public class SeekableStreamSupervisorStateTest extends EasyMockSupport
{
@@ -625,7 +628,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
expectEmitterSupervisor(false);
- CountDownLatch latch = new CountDownLatch(1);
+ CountDownLatch latch = new CountDownLatch(2);
TestEmittingTestSeekableStreamSupervisor supervisor = new
TestEmittingTestSeekableStreamSupervisor(
latch,
ImmutableMap.of("1", 100L, "2", 250L, "3", 500L),
@@ -643,19 +646,23 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
latch.await();
- Assert.assertEquals(6, emitter.getEvents().size());
- Assert.assertEquals("ingest/test/lag",
emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(850L, emitter.getEvents().get(0).toMap().get("value"));
- Assert.assertEquals("ingest/test/maxLag",
emitter.getEvents().get(1).toMap().get("metric"));
- Assert.assertEquals(500L, emitter.getEvents().get(1).toMap().get("value"));
- Assert.assertEquals("ingest/test/avgLag",
emitter.getEvents().get(2).toMap().get("metric"));
- Assert.assertEquals(283L, emitter.getEvents().get(2).toMap().get("value"));
- Assert.assertEquals("ingest/test/lag/time",
emitter.getEvents().get(3).toMap().get("metric"));
- Assert.assertEquals(45000L,
emitter.getEvents().get(3).toMap().get("value"));
- Assert.assertEquals("ingest/test/maxLag/time",
emitter.getEvents().get(4).toMap().get("metric"));
- Assert.assertEquals(20000L,
emitter.getEvents().get(4).toMap().get("value"));
- Assert.assertEquals("ingest/test/avgLag/time",
emitter.getEvents().get(5).toMap().get("metric"));
- Assert.assertEquals(15000L,
emitter.getEvents().get(5).toMap().get("value"));
+ List<Event> events = emitter.getEvents();
+ List<String> whitelist = Arrays.asList("ingest/test/lag",
"ingest/test/maxLag",
+ "ingest/test/avgLag", "ingest/test/lag/time",
"ingest/test/maxLag/time", "ingest/test/avgLag/time");
+ events = filterMetrics(events, whitelist);
+ Assert.assertEquals(6, events.size());
+ Assert.assertEquals("ingest/test/lag",
events.get(0).toMap().get("metric"));
+ Assert.assertEquals(850L, events.get(0).toMap().get("value"));
+ Assert.assertEquals("ingest/test/maxLag",
events.get(1).toMap().get("metric"));
+ Assert.assertEquals(500L, events.get(1).toMap().get("value"));
+ Assert.assertEquals("ingest/test/avgLag",
events.get(2).toMap().get("metric"));
+ Assert.assertEquals(283L, events.get(2).toMap().get("value"));
+ Assert.assertEquals("ingest/test/lag/time",
events.get(3).toMap().get("metric"));
+ Assert.assertEquals(45000L, events.get(3).toMap().get("value"));
+ Assert.assertEquals("ingest/test/maxLag/time",
events.get(4).toMap().get("metric"));
+ Assert.assertEquals(20000L, events.get(4).toMap().get("value"));
+ Assert.assertEquals("ingest/test/avgLag/time",
events.get(5).toMap().get("metric"));
+ Assert.assertEquals(15000L, events.get(5).toMap().get("value"));
verifyAll();
}
@@ -664,7 +671,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
expectEmitterSupervisor(false);
- CountDownLatch latch = new CountDownLatch(1);
+ CountDownLatch latch = new CountDownLatch(2);
TestEmittingTestSeekableStreamSupervisor supervisor = new
TestEmittingTestSeekableStreamSupervisor(
latch,
ImmutableMap.of("1", 100L, "2", 250L, "3", 500L),
@@ -682,13 +689,16 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
latch.await();
- Assert.assertEquals(3, emitter.getEvents().size());
- Assert.assertEquals("ingest/test/lag",
emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(850L, emitter.getEvents().get(0).toMap().get("value"));
- Assert.assertEquals("ingest/test/maxLag",
emitter.getEvents().get(1).toMap().get("metric"));
- Assert.assertEquals(500L, emitter.getEvents().get(1).toMap().get("value"));
- Assert.assertEquals("ingest/test/avgLag",
emitter.getEvents().get(2).toMap().get("metric"));
- Assert.assertEquals(283L, emitter.getEvents().get(2).toMap().get("value"));
+ List<Event> events = emitter.getEvents();
+ List<String> whitelist = Arrays.asList("ingest/test/lag",
"ingest/test/maxLag", "ingest/test/avgLag");
+ events = filterMetrics(events, whitelist);
+ Assert.assertEquals(3, events.size());
+ Assert.assertEquals("ingest/test/lag",
events.get(0).toMap().get("metric"));
+ Assert.assertEquals(850L, events.get(0).toMap().get("value"));
+ Assert.assertEquals("ingest/test/maxLag",
events.get(1).toMap().get("metric"));
+ Assert.assertEquals(500L, events.get(1).toMap().get("value"));
+ Assert.assertEquals("ingest/test/avgLag",
events.get(2).toMap().get("metric"));
+ Assert.assertEquals(283L, events.get(2).toMap().get("value"));
verifyAll();
}
@@ -697,7 +707,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
expectEmitterSupervisor(false);
- CountDownLatch latch = new CountDownLatch(1);
+ CountDownLatch latch = new CountDownLatch(2);
TestEmittingTestSeekableStreamSupervisor supervisor = new
TestEmittingTestSeekableStreamSupervisor(
latch,
null,
@@ -715,13 +725,78 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
latch.await();
- Assert.assertEquals(3, emitter.getEvents().size());
- Assert.assertEquals("ingest/test/lag/time",
emitter.getEvents().get(0).toMap().get("metric"));
- Assert.assertEquals(45000L,
emitter.getEvents().get(0).toMap().get("value"));
- Assert.assertEquals("ingest/test/maxLag/time",
emitter.getEvents().get(1).toMap().get("metric"));
- Assert.assertEquals(20000L,
emitter.getEvents().get(1).toMap().get("value"));
- Assert.assertEquals("ingest/test/avgLag/time",
emitter.getEvents().get(2).toMap().get("metric"));
- Assert.assertEquals(15000L,
emitter.getEvents().get(2).toMap().get("value"));
+ List<Event> events = emitter.getEvents();
+ List<String> whitelist = Arrays.asList("ingest/test/lag/time",
"ingest/test/maxLag/time", "ingest/test/avgLag/time");
+ events = filterMetrics(events, whitelist);
+ Assert.assertEquals(3, events.size());
+ Assert.assertEquals("ingest/test/lag/time",
events.get(0).toMap().get("metric"));
+ Assert.assertEquals(45000L, events.get(0).toMap().get("value"));
+ Assert.assertEquals("ingest/test/maxLag/time",
events.get(1).toMap().get("metric"));
+ Assert.assertEquals(20000L, events.get(1).toMap().get("value"));
+ Assert.assertEquals("ingest/test/avgLag/time",
events.get(2).toMap().get("metric"));
+ Assert.assertEquals(15000L, events.get(2).toMap().get("value"));
+ verifyAll();
+ }
+
+ @Test
+ public void testEmitNoticesQueueSize() throws Exception
+ {
+ expectEmitterSupervisor(false);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ TestEmittingTestSeekableStreamSupervisor supervisor = new
TestEmittingTestSeekableStreamSupervisor(
+ latch,
+ null,
+ null
+ );
+
+
+ supervisor.start();
+
+ Assert.assertTrue(supervisor.stateManager.isHealthy());
+ Assert.assertEquals(BasicState.PENDING,
supervisor.stateManager.getSupervisorState());
+ Assert.assertEquals(BasicState.PENDING,
supervisor.stateManager.getSupervisorState().getBasicState());
+ Assert.assertTrue(supervisor.stateManager.getExceptionEvents().isEmpty());
+ Assert.assertFalse(supervisor.stateManager.isAtLeastOneSuccessfulRun());
+
+
+ latch.await();
+ List<Event> events = emitter.getEvents();
+ List<String> whitelist =
Collections.singletonList("ingest/notices/queueSize");
+ events = filterMetrics(events, whitelist);
+ Assert.assertEquals(1, events.size());
+ Assert.assertEquals("ingest/notices/queueSize",
events.get(0).toMap().get("metric"));
+ Assert.assertEquals(0, events.get(0).toMap().get("value"));
+ Assert.assertEquals("testDS", events.get(0).toMap().get("dataSource"));
+ verifyAll();
+ }
+
+ @Test
+ public void testEmitNoticesTime() throws Exception
+ {
+ expectEmitterSupervisor(false);
+ CountDownLatch latch = new CountDownLatch(2);
+ TestEmittingTestSeekableStreamSupervisor supervisor = new
TestEmittingTestSeekableStreamSupervisor(
+ latch,
+ null,
+ null
+ );
+ supervisor.start();
+ supervisor.emitNoticesTime();
+ Assert.assertTrue(supervisor.stateManager.isHealthy());
+ Assert.assertEquals(BasicState.PENDING,
supervisor.stateManager.getSupervisorState());
+ Assert.assertEquals(BasicState.PENDING,
supervisor.stateManager.getSupervisorState().getBasicState());
+ Assert.assertTrue(supervisor.stateManager.getExceptionEvents().isEmpty());
+ Assert.assertFalse(supervisor.stateManager.isAtLeastOneSuccessfulRun());
+ latch.await();
+ List<Event> events = emitter.getEvents();
+ List<String> whitelist = Collections.singletonList("ingest/notices/time");
+ events = filterMetrics(events, whitelist);
+ Assert.assertEquals(1, events.size());
+ Assert.assertEquals("ingest/notices/time",
events.get(0).toMap().get("metric"));
+ Assert.assertEquals(500L, events.get(0).toMap().get("value"));
+ Assert.assertEquals("testDS", events.get(0).toMap().get("dataSource"));
+ Assert.assertEquals("dummyNoticeType",
events.get(0).toMap().get("noticeType"));
verifyAll();
}
@@ -730,7 +805,7 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
{
expectEmitterSupervisor(true);
- CountDownLatch latch = new CountDownLatch(1);
+ CountDownLatch latch = new CountDownLatch(2);
TestEmittingTestSeekableStreamSupervisor supervisor = new
TestEmittingTestSeekableStreamSupervisor(
latch,
ImmutableMap.of("1", 100L, "2", 250L, "3", 500L),
@@ -748,10 +823,22 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
latch.await();
- Assert.assertEquals(0, emitter.getEvents().size());
+ List<Event> events = emitter.getEvents();
+ List<String> whitelist = Arrays.asList("ingest/test/lag",
"ingest/test/maxLag",
+ "ingest/test/avgLag", "ingest/test/lag/time",
"ingest/test/maxLag/time", "ingest/test/avgLag/time");
+ events = filterMetrics(events, whitelist);
+ Assert.assertEquals(0, events.size());
verifyAll();
}
+ private List<Event> filterMetrics(List<Event> events, List<String> whitelist)
+ {
+ List<Event> result = events.stream()
+ .filter(e -> whitelist.contains(e.toMap().get("metric").toString()))
+ .collect(Collectors.toList());
+ return result;
+ }
+
private void expectEmitterSupervisor(boolean suspended) throws
EntryExistsException
{
spec = createMock(SeekableStreamSupervisorSpec.class);
@@ -1250,6 +1337,19 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
}
@Override
+ protected void emitNoticesQueueSize()
+ {
+ super.emitNoticesQueueSize();
+ latch.countDown();
+ }
+
+ public void emitNoticesTime()
+ {
+ super.emitNoticeProcessTime("dummyNoticeType", 500);
+ latch.countDown();
+ }
+
+ @Override
public LagStats computeLagStats()
{
return null;
@@ -1265,6 +1365,12 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
spec.getMonitorSchedulerConfig().getEmitterPeriod().getMillis(),
TimeUnit.MILLISECONDS
);
+ reportingExec.scheduleAtFixedRate(
+ this::emitNoticesQueueSize,
+ ioConfig.getStartDelay().getMillis(),
+ spec.getMonitorSchedulerConfig().getEmitterPeriod().getMillis(),
+ TimeUnit.MILLISECONDS
+ );
}
}
diff --git a/website/.spelling b/website/.spelling
index d4b568c..0e0ab00 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -1276,6 +1276,7 @@ nativeQueryIds
netAddress
netHwaddr
netName
+noticeType
numComplexMetrics
numDimensions
numMetrics
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
