Moving averages for ingestion row stats #5748

jon-wei · 2018-05-07T20:33:15Z

This PR adds optional moving average reports to the "rowStats" worker chat handler endpoint, for IndexTask, KafkaIndexTask, and AppenderatorRealtimeIndexTask.

Moving average collection can be configured via the "movingAverageWindowSizeMillis" and "movingAverageNumWindows" properties in the tuning configs for these tasks. By default these are 1000 and 60 (record the current stats every 1s, and store 60 of these records).

For Kafka indexing, the supervisor also provides an endpoint at /druid/indexer/v1/supervisor/{id}/stats that will collect stats from each of its indexing tasks and present them as a unified report, as a convenience.

drcrallen · 2018-05-08T01:29:20Z

If these are becoming more common, would it make sense to have a Dropwizard compatible metrics registry interface so we don't have to code our own on-heap metrics collectors?

jon-wei · 2018-05-08T01:34:08Z

@drcrallen I've not used Dropwizard before, do you have an example implementation in mind?

drcrallen · 2018-05-08T01:42:08Z

http://metrics.dropwizard.io/4.0.0/manual/core.html

Has a bunch of examples. Such a metrics interface is common among a few other projects.

https://beam.apache.org/documentation/sdks/javadoc/2.4.0/org/apache/beam/sdk/metrics/Metrics.html

https://spark.apache.org/docs/latest/monitoring.html#metrics

jon-wei · 2018-05-08T01:50:42Z

Thanks, Dropwizard looks useful, I'll check it out

b-slim · 2018-05-08T03:05:13Z

@drcrallen i believe drop-wizard still uses onheap space, but i agree maybe it is a good idea to use a common library to do something like moving average or histograms

drcrallen · 2018-05-08T15:48:33Z

@b-slim agreed, that's my main concern. That we would need to maintain a metrics collection library (histograms / moving averages / rate calculators) when there are already good and widely used ones out there.

jon-wei · 2018-05-09T03:13:05Z

@drcrallen Changed this to use Dropwizard metrics Meter objects to collect row stats (instead of FireDepartmentMetrics)

Dropwizard has a dependency on slf4j 1.7.25 (https://github.com/dropwizard/metrics/blob/4.1-development/pom.xml), Druid uses 1.7.12, but there shouldn't be any compatibility issues between those versions (https://www.slf4j.org/compatibility.html).

b-slim · 2018-05-10T17:26:36Z

server/src/main/java/io/druid/indexing/overlord/supervisor/Supervisor.java

@@ -37,6 +38,11 @@

  SupervisorReport getStatus();

+  default Map<String, Object> getStats()
+  {
+    return null;


why not an empty map?

b-slim · 2018-05-10T17:28:59Z

indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java

@@ -919,12 +920,14 @@ private TaskStatus generateAndPublishSegments(
    ) {
      driver.startJob();

+      buildSegmentsMeters = new RowIngestionMeters();


b-slim · 2018-05-10T17:33:00Z

...nsions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java

@@ -618,6 +617,7 @@ public void run()

      Set<Integer> assignment = assignPartitionsAndSeekToNext(consumer, topic);

+      rowIngestionMeters = new RowIngestionMeters();


b-slim · 2018-05-10T17:33:58Z

...nsions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java

@@ -965,6 +964,7 @@ private TaskStatus runInternalLegacy(final TaskToolbox toolbox) throws Exception
        )
    );

+    rowIngestionMeters = new RowIngestionMeters();


oh i see you are reusing it here ...

can we just reset the counters ?

those are different code paths (one is a "legacy" mode), so only one of them would get called in a task

I don't make this final and set this here because each Meter object has a "startTime" field that's set when the object is created, so I want the startTime to be as close to the actual ingestion work

b-slim · 2018-05-10T17:35:33Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

@@ -254,6 +254,46 @@ public DateTime getStartTime(final String id)
    }
  }

+  public Map<String, Object> getMovingAverages(final String id)


can you add nullable annotation

I made this return an empty map instead for those cases

b-slim · 2018-05-10T17:36:23Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

+      return null;
+    }
+    catch (IOException e) {
+      throw Throwables.propagate(e);


can we start moving away form propagates it is going to be removed form guava.

Changed this and other Throwables.propagate calls in this file

drcrallen · 2018-05-12T00:09:55Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

@@ -216,7 +215,7 @@ public boolean resume(final String id)
    }
    catch (IOException | InterruptedException e) {
      log.error("Exception [%s] while pausing Task [%s]", e.getMessage(), id);
-      throw Throwables.propagate(e);
+      throw new RuntimeException(e);


it is odd to log and throw. Such a thing usually leads to double logging. Would it make more sense just to put the log message in the RuntimeException?

Removed the log and added the message to the RuntimeException

drcrallen · 2018-05-12T00:10:33Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

@@ -622,7 +663,7 @@ private FullResponseHolder submitRequest(
            Thread.sleep(sleepTime);
          }
          catch (InterruptedException e2) {
-            Throwables.propagate(e2);
+            throw new RuntimeException(e2);


This needs to set Thread.currentThread().interrupt()

Added the interrupt call

drcrallen · 2018-05-12T00:11:45Z

indexing-service/src/main/java/io/druid/indexing/common/stats/RowIngestionMeters.java

+
+import java.util.Map;
+
+public class RowIngestionMeters


without increasing the scope too much, can this be a bind-able interface with a DropwizardRowIngestionMeter as the default bound thing? this would allow extensions to specify their own if desired.

Essentially, this class would be renamed to DropwizardRowIngestionMeter and RowIngestionMeters would be an interface with the methods you need

Sure, I'll look into making this extensible

The RowIngestionMeters are now created by an injected RowIngestionMetersFactory that can be replaced by extensions

drcrallen · 2018-05-12T00:13:26Z

indexing-service/src/main/java/io/druid/indexing/common/stats/RowIngestionMeters.java

+    return totals;
+  }
+
+  public Map<String, Object> getMovingAverages()


these don't have strong guarantees. is there any way the method invocations can have something that has guarantees around it?

What guarantees do you think this method should provide?

drcrallen · 2018-05-12T00:14:42Z

...service/src/main/java/io/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTask.java

@@ -143,7 +143,7 @@ private static String makeTaskId(RealtimeAppenderatorIngestionSpec spec)
  private volatile FireDepartmentMetrics metrics = null;

  @JsonIgnore
-  private TaskMetricsGetter metricsGetter;
+  private RowIngestionMeters rowIngestionMeters;


if this is injectable this could simply be @JacksonInject

Alternatively, it might make more sense to have a Factory object as the thing that is injected, and create the meters objects as needed in the classes.

I went with injecting a factory for the meters

can this be final and created in the constructor?

drcrallen · 2018-05-17T16:12:21Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

-          Throwables.propagateIfInstanceOf(e.getCause(), IOException.class);
-          Throwables.propagateIfInstanceOf(e.getCause(), ChannelException.class);
-          throw Throwables.propagate(e);
+          if (e.getCause() instanceof IOException || e.getCause() instanceof ChannelException) {


Can this be more explicit? things that catch Exception are dangerous.

For example, catching catch(IOException ioe){throw ioe;} first should handle the IOE cases. Then if it is the httpClient.go call that could be the issue, things that would need the IOException unwrapped can be handled explicitly. Then catch(Exception re){throw new RuntimeException(e, "Unhandled exception");} is probably good enough to handle all the other items that might crop up. Also make sure nothing here can throw InterruptedException or add a special handler for it please.

Things that catch(Exception e) can cause issues :-(

Added an explicit handler for IOException and ChannelException, added a handler for InterruptedException

drcrallen · 2018-05-17T16:13:24Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

@@ -622,7 +665,8 @@ private FullResponseHolder submitRequest(
            Thread.sleep(sleepTime);
          }
          catch (InterruptedException e2) {
-            Throwables.propagate(e2);
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e2);


suggest e.addSuppressed(e2) and throwing the original exception

Changed to throw original exception with addSuppressed

drcrallen · 2018-05-17T16:13:48Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

@@ -633,7 +677,7 @@ private FullResponseHolder submitRequest(
      }
      catch (Exception e) {
        log.warn(e, "Exception while sending request");
-        throw e;
+        throw new RuntimeException(e);


doh... another Exception catch

removed the RuntimeException wrapper here

drcrallen · 2018-05-17T16:14:16Z

...kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+    try {
+      return getCurrentTotalStats();
+    }
+    catch (Exception e) {


does this need to catch all exceptions?

InterruptedException is not handled properly here

Added handler for InterruptedException and another handler for ExecutionException and TimeoutException, removed general Exception catch

drcrallen · 2018-05-17T16:15:41Z

...kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+      for (String taskId : group.taskIds()) {
+        futures.add(
+            Futures.transform(
+                taskClient.getMovingAveragesAsync(taskId),


is this really so compute intensive that parallel even helps?

or is this based on network IO?

Either way please add a javadoc on the method

this is network IO, the supervisor is retrieving stats from each of the tasks it manages

added javadoc

drcrallen · 2018-05-17T16:19:59Z

...kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+        allStats.put(String.valueOf(groupId), groupStats);
+      }
+
+      final Map<String, Object> eGroupStats = groupStats;


this is concurrently modified from who knows how many threads. Can the Futures.allAsList result be used instead? something like a Stream of Pair<String, Map<String, Object>> with a Collectors.toMap should work

changed this to return a result object from each future, map is built only after all results are retrieved

drcrallen · 2018-05-17T16:21:38Z

indexing-service/src/main/java/io/druid/indexing/common/stats/DropwizardRowIngestionMeters.java

+    fifteenMinute.put(UNPARSEABLE, unparseable.getFifteenMinuteRate());
+    fifteenMinute.put(THROWN_AWAY, thrownAway.getFifteenMinuteRate());
+
+    movingAverages.put("1m", oneMinute);


can these be static final strings somewhere?

Made these static final strings

drcrallen · 2018-05-17T16:24:47Z

indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java

        );
      }
    }

    returnMap.put("totals", totalsMap);
+    returnMap.put("averages", averagesMap);


movingAverages?

renamed these to movingAverages

drcrallen · 2018-05-17T16:25:14Z

indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java

@@ -496,16 +510,16 @@ public TaskStatus run(final TaskToolbox toolbox) throws Exception
  private Map<String, Object> getTaskCompletionRowStats()
  {
    Map<String, Object> metrics = Maps.newHashMap();
-    if (determinePartitionsMetricsGetter != null) {
+    if (determinePartitionsMeters != null) {


suggest moving both of these to final objects instead of having them nullable

moved to constructor

b-slim · 2018-05-22T12:47:39Z

LGTM

b-slim · 2018-05-22T12:47:49Z

👍

jon-wei · 2018-05-23T19:59:35Z

@drcrallen Did you have any more comments on this?

jon-wei · 2018-05-31T01:57:11Z

@b-slim @drcrallen The patch had a bug where metrics emission was broken because the RealtimeMetricsMonitor was not hooked up to the new ingestion meters, I added a commit that fixes that. Can you take another look at this PR?

jihoonson · 2018-06-01T20:59:20Z

...nsions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java

@@ -511,8 +513,8 @@ private TaskStatus runInternal(final TaskToolbox toolbox) throws Exception
        null
    );
    fireDepartmentMetrics = fireDepartmentForMetrics.getMetrics();
-    metricsGetter = new FireDepartmentMetricsTaskMetricsGetter(fireDepartmentMetrics);
-    toolbox.getMonitorScheduler().addMonitor(TaskRealtimeMetricsMonitorBuilder.build(this, fireDepartmentForMetrics));
+    toolbox.getMonitorScheduler()


FireDepartmentMetricsTaskMetricsGetter is no longer used in any place. Please remove it.

Removed the class

jihoonson · 2018-06-01T21:13:22Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

+          true
+      );
+      return response.getContent() == null || response.getContent().isEmpty()
+             ? ImmutableMap.of()


nit: Collections.emptyMap() can be better to avoid unnecessary map creation.

Changed this to Collections.emptyMap()

jihoonson · 2018-06-01T21:13:42Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

+             : jsonMapper.readValue(response.getContent(), JacksonUtils.TYPE_REFERENCE_MAP_STRING_OBJECT);
+    }
+    catch (NoTaskLocationException e) {
+      return ImmutableMap.of();


Changed this to Collections.emptyMap()

jihoonson · 2018-06-01T21:18:39Z

...-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

+    }
+  }
+
+  public ListenableFuture<Map<String, Object>> getMovingAveragesAsync(final String id)


I think it's better to convert all ListenableFuture to CompletableFuture in another PR if needed.

jihoonson · 2018-06-01T21:20:18Z

...kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+    }
+    catch (InterruptedException ie) {
+      Thread.currentThread().interrupt();
+      log.error("getStats() interrupted.");


Please add ie to the log.

Add ie to the log

jihoonson · 2018-06-01T21:35:31Z

...kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+    }
+
+    List<StatsFromTaskResult> results = Futures.successfulAsList(futures).get(futureTimeoutInSeconds, TimeUnit.SECONDS);
+    for (StatsFromTaskResult result : results) {


The list of results is in the same order as the input list, and if any of the provided futures fails or is canceled, its corresponding position will contain null (which is indistinguishable from the future having a successful value of null).

result can be null if we should use Futures.successfulAsList().

Added a null check for result, an error is logged if result is null

jihoonson · 2018-06-01T21:37:50Z

...kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

+
+    List<StatsFromTaskResult> results = Futures.successfulAsList(futures).get(futureTimeoutInSeconds, TimeUnit.SECONDS);
+    for (StatsFromTaskResult result : results) {
+      Map<String, Object> groupMap = (Map<String, Object>) allStats.putIfAbsent(result.getGroupId(), Maps.newHashMap());


Map<String, Object> groupMap = (Map<String, Object>) allStats.putIfAbsent(result.getGroupId(), Maps.newHashMap()); if (groupMap == null) { groupMap = (Map<String, Object>) allStats.get(result.getGroupId()); }

Does this mean Map<String, Object> groupMap = (Map<String, Object>) allStats.computeIfAbsent(result.getGroupId(), new HashMap<>());?

Changed this to use computeIfAbsent

jihoonson · 2018-06-01T21:41:44Z

...e/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaIndexTaskClientTest.java

@@ -181,7 +180,7 @@ public void testInternalServerError()
    verifyAll();
  }

-  @Test(expected = IAE.class)
+  @Test(expected = RuntimeException.class)


I think RuntimeException is too broad, and this check might be false positive. Would you improve this test by adding ExpectedException and checking the cause and the error message of the RuntimeException?

Changed this and other tests here to use ExpectedException and also check the exception messages

jihoonson · 2018-06-01T21:46:09Z

indexing-service/src/main/java/io/druid/indexing/common/stats/RowIngestionMeters.java

+ * FireDepartmentMetrics are not currently supported, so we continue to use FireDepartmentMetrics alongside
+ * RowIngestionMeters to avoid unnecessary overhead from maintaining these moving averages.
+ */
+public interface RowIngestionMeters


Would you explain why this is an interface? Is it for smoothly migrating to some other libraries instead of Dropwizard in the future?

It's an interface so that extensions can provide custom implementations, see #5748 (comment)

Does it make sense to make this as an ExtensionPoint?

Added @ExtensionPoint

jihoonson · 2018-06-01T21:48:10Z

indexing-service/src/main/java/io/druid/indexing/common/stats/TaskRealtimeMetricsMonitor.java

+import java.util.Map;
+
+/**
+ * Replaces the old RealtimeMetricsMonitor for indexing tasks that use a single FireDepartment, with changes to


Would you add a comment to RealtimeMetricsMonitor as well that RealtimeMetricsMonitor is an old one and TaskRealtimeMetricsMonitor is recommended to use instead?

Added a comment to RealtimeMetricsMonitor

jon-wei · 2018-06-02T02:03:18Z

@jihoonson thanks for the review!

jihoonson

LGTM. @jon-wei thanks!

* Moving averages for ingestion row stats * PR comments * Make RowIngestionMeters extensible * test and checkstyle fixes * More PR comments * Fix metrics * Add some comments * PR comments * Comments

jon-wei added Feature Area - Batch Ingestion labels May 7, 2018

jon-wei force-pushed the task_moving_averages2 branch from 5f85da5 to 2e02ce1 Compare May 8, 2018 00:34

jon-wei force-pushed the task_moving_averages2 branch 4 times, most recently from 3d7039b to 93a0f3f Compare May 9, 2018 03:09

jon-wei force-pushed the task_moving_averages2 branch from 93a0f3f to a5e3913 Compare May 9, 2018 03:58

Moving averages for ingestion row stats

78dd24e

jon-wei force-pushed the task_moving_averages2 branch from a5e3913 to 78dd24e Compare May 9, 2018 21:39

b-slim reviewed May 10, 2018

View reviewed changes

PR comments

c895750

drcrallen reviewed May 12, 2018

View reviewed changes

Make RowIngestionMeters extensible

bd63beb

drcrallen reviewed May 17, 2018

View reviewed changes

jon-wei force-pushed the task_moving_averages2 branch 2 times, most recently from c279755 to 3d26d5a Compare May 18, 2018 00:27

More PR comments

77b183b

jon-wei force-pushed the task_moving_averages2 branch from 3d26d5a to 77b183b Compare May 18, 2018 00:38

jon-wei force-pushed the task_moving_averages2 branch from a61f381 to 5e3cd6d Compare May 29, 2018 20:15

Fix metrics

ad71384

jon-wei force-pushed the task_moving_averages2 branch from 5e3cd6d to ad71384 Compare May 30, 2018 00:58

Add some comments

2d954d7

jihoonson reviewed Jun 1, 2018

View reviewed changes

PR comments

2fe8122

Comments

ded9ba9

jihoonson approved these changes Jun 4, 2018

View reviewed changes

jon-wei merged commit 684b5d1 into apache:master Jun 5, 2018

dclim added this to the 0.13.0 milestone Oct 8, 2018

dclim mentioned this pull request Oct 10, 2018

Druid 0.13.0-incubating release notes #6442

Closed

		@@ -618,6 +617,7 @@ public void run()

		Set<Integer> assignment = assignPartitionsAndSeekToNext(consumer, topic);

		rowIngestionMeters = new RowIngestionMeters();

Moving averages for ingestion row stats #5748

Moving averages for ingestion row stats #5748

Conversation

jon-wei commented May 7, 2018

drcrallen commented May 8, 2018

jon-wei commented May 8, 2018

drcrallen commented May 8, 2018

jon-wei commented May 8, 2018

b-slim commented May 8, 2018

drcrallen commented May 8, 2018

jon-wei commented May 9, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen May 12, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen May 17, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jon-wei May 18, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

b-slim commented May 22, 2018

b-slim commented May 22, 2018

jon-wei commented May 23, 2018

jon-wei commented May 31, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drcrallen May 12, 2018 •

edited

Loading

drcrallen May 17, 2018 •

edited

Loading

jon-wei May 18, 2018 •

edited

Loading