Add shuffle metrics for parallel indexing #10359

jihoonson · 2020-09-04T23:32:33Z

Description

Part of #10352. This PR adds these metrics for middleManagers. These metrics have the supervisorTaskId as their dimension.

ingest/shuffle/bytes: Number of bytes shuffled per emissionPeriod.
ingest/shuffle/requests: Number of shuffle requests per emissionPeriod.

I haven't updated document yet, will add them with missing shuffle configurations together in a follow-up PR.

This PR has:

been self-reviewed.
- using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
added documentation for new or modified features or behaviors.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added or updated version, license, or notice information in licenses.yaml
added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
added integration tests.
been tested in a test Druid cluster.

abhishekagarwal87 · 2020-09-07T10:31:35Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMetrics.java

+        .computeIfAbsent(supervisorTaskId, k -> new PerDatasourceShuffleMetrics()).accumulate(fileLength);
+  }
+
+  public Map<String, PerDatasourceShuffleMetrics> snapshot()


should this be renamed to snapshotAndReset or may be just reset ?

public Map<String, PerDatasourceShuffleMetrics> reset() { return Collections.unmodifiableMap(datasourceMetrics.getAndSet(new ConcurrentHashMap<>())); }

Sounds good. Changed to snapshotAndReset() since it sounds more intuitive to me.

abhishekagarwal87 · 2020-09-07T10:51:34Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMetrics.java

+  {
+    datasourceMetrics
+        .get()
+        .computeIfAbsent(supervisorTaskId, k -> new PerDatasourceShuffleMetrics()).accumulate(fileLength);


it's still possible to miss an update in reporting because of race condition, right? Since the reference could be reset while the accumulation is happening.

The race condition exists, but it should be fine because the missing update should be included in the next call to snapshotAndReset(). I added javadocs explaining why.

I think this needs to use something like AtomicReference.getAndUpdate so that it isn't racy with the monitor/emitter? Though I'm not sure getAndUpdate or the similar methods are actually appropriate since they are supposed to be side-effect free, so I'm not really sure how exactly to resolve this.

Like, the potentially problematic scenario I'm thinking of is where shuffleRequested is called "before" snapshotAndReset. It seems like once AtomicReference.get has completed, snapshotAndReset can proceed, so now the shuffle monitor has the same concurrent map we are still actively updating, and it is preparing to build the metrics to emit. It seems super unlikely that it would be a problem, but unless I'm missing something it does seem possible.

Ah, you guys are right. Will fix it.

The problem is that any updates on the reference to datasourceMetrics should be synchronized with any updates on the map itself and its values. I could use ConcurrentHashMap.compute() if I didn't have to reset the reference to the map when a snapshot is taken, but I think it's needed since the map can keep growing over time otherwise. I'm not sure if there is any other way than using a big lock. I made this change, let me know if you have a better idea.

the lock should suffice. shuffleRequested doesn't need to be a high throughput call.

clintropolis

overall lgtm

clintropolis · 2020-09-09T06:38:24Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMetrics.java

+  {
+    datasourceMetrics
+        .get()
+        .computeIfAbsent(supervisorTaskId, k -> new PerDatasourceShuffleMetrics()).accumulate(fileLength);


I think this needs to use something like AtomicReference.getAndUpdate so that it isn't racy with the monitor/emitter? Though I'm not sure getAndUpdate or the similar methods are actually appropriate since they are supposed to be side-effect free, so I'm not really sure how exactly to resolve this.

Like, the potentially problematic scenario I'm thinking of is where shuffleRequested is called "before" snapshotAndReset. It seems like once AtomicReference.get has completed, snapshotAndReset can proceed, so now the shuffle monitor has the same concurrent map we are still actively updating, and it is preparing to build the metrics to emit. It seems super unlikely that it would be a problem, but unless I'm missing something it does seem possible.

abhishekagarwal87 · 2020-09-10T08:54:58Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMetrics.java

+
+  /**
+   * This method is called whenever the monitoring thread takes a snapshot of the current metrics. The map inside
+   * AtomicReference will be reset to an empty map after this call. This is to return the snapshot metrics collected


This comment needs an update after the latest changes.

Good catch. Fixed.

suneet-s

Overall, looks very nice! Just one ask about a feature flag. I don't have a strong opinion on the name of the metric, but would love to know your thoughts

suneet-s · 2020-09-21T22:28:30Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMetrics.java

+   */
+  public void shuffleRequested(String supervisorTaskId, long fileLength)
+  {
+    synchronized (lock) {


Since there is a risk of the locking introducing a slow down here because of contention, can we update this to include a feature flag check?

This way, if there are some unforeseen issues with locking, we can disable metric computation and reporting. I think a static feature flag - like a system property would be good enough for this use case.

I don't think this locking would introduce any noticeable slow down, but feature flag sounds good. Now, ShuffleMetrics and ShuffleMonitor will work only when ShuffleMonitor is defined in druid.monitoring.monitors. Added some doc for that too.

I like this approach a lot 🤘

suneet-s · 2020-09-21T22:29:58Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMetrics.java

+   * whenever a snapshot is taken since the map can keep growing over time otherwise. For concurrent access pattern,
+   * see {@link #shuffleRequested} and {@link #snapshotAndReset()}.
+   */
+  @GuardedBy("lock")


Just curious - why did you choose to use the guarded by pattern instead of a ConcurrentMap?

There was some prior discussion about it. It was mainly because not only updating the datasourceMetrics map, but also updating PerDatasourceShuffleMetrics should be synchronized as well. For example, if it was updating PerDatasourceShuffleMetrics when snapshotAndReset() is called, it should guarantee that the updating will be done before snapshotAndReset().

Ah - that makes sense. Thanks for the explanation

suneet-s · 2020-09-21T22:32:39Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMonitor.java

+{
+  private static final String SUPERVISOR_TASK_ID_DIMENSION = "supervisorTaskId";
+  private static final String SHUFFLE_BYTES_KEY = "shuffle/bytes";
+  private static final String SHUFFLE_REQUESTS_KEY = "shuffle/requests";


other ingestion related metrics start with "ingest/" any thoughts on whether these metrics fall under the ingestion metrics category?

I was thinking about where the metrics would live in the docs which is why I was asking this question. I thought maybe it belonged here https://druid.apache.org/docs/latest/operations/metrics.html#ingestion-metrics-realtime-process ?

Good question. The new metrics don't seem to belong to any existing section, so I added a new one. But our current doc doesn't seem organized well (for example, the metrics in the above link are not only for realtime processes, but for all task types as well), maybe we need to tidy up at some point after #10352 is done.

Also, I modified the metrics to start with ingest/ similar to other ingestion metrics.

suneet-s · 2020-10-08T21:28:57Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleModule.java

+  {
+    // ShuffleMonitor cannot be registered dynamically, but can only via the static configuration (MonitorsConfig).
+    // As a result, it is safe to check only one time if it is registered in MonitorScheduler.
+    final Optional<ShuffleMonitor> maybeMonitor = monitorScheduler.findMonitor(ShuffleMonitor.class);


I see that MonitorScheduler has a removeMonitor method, and ShuffleMetrics is provided as a Singleton. Can someone remove the ShuffleMonitor while Druid is running? If they do that how would it impact ShuffleMetrics being reported

Currently, a monitor can be removed when 1) the monitor() method returns false or 2) tasks de-register task-specific monitors such as TaskRealtimeMetricsMonitor which is used in the deprecated Tranquility. So, ShuffleMonitor cannot be removed once a node is started.

In the future, I think we may want to dynamically register and remove monitors (because it's cool). In that case, we probably need to check all monitor implementations we have if they have any issues to do that. We can come back to ShuffleMonitor later to handle the case you mentioned.

suneet-s

LGTM with some asks for unit tests

suneet-s · 2020-10-09T01:27:25Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleModule.java

+  @Override
+  public void configure(Binder binder)
+  {
+    Jerseys.addResource(binder, ShuffleResource.class);


Can you add a ModuleTest that validates the ShuffleResource and Optional<ShuffleMetrics>is injectable a. I think I've written AuthorizerMapperModuleTest that would be a similar example

suneet-s · 2020-10-09T01:36:51Z

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/ShuffleMonitor.java

+        emitter.emit(metricBuilder.build(SHUFFLE_REQUESTS_KEY, perDatasourceShuffleMetrics.getShuffleRequests()));
+      });
+    }
+    return true;


Should we add unit tests for this function?

Oops, I thought I added one already. Added now.

suneet-s · 2020-10-09T15:14:47Z

indexing-service/src/test/java/org/apache/druid/indexing/worker/shuffle/ShuffleModuleTest.java

+  {
+    final ShuffleMonitor shuffleMonitor = new ShuffleMonitor();
+    final MonitorScheduler monitorScheduler = Mockito.mock(MonitorScheduler.class);
+    Mockito.when(monitorScheduler.findMonitor(ArgumentMatchers.eq(ShuffleMonitor.class)))


nit:

Suggested change

Mockito.when(monitorScheduler.findMonitor(ArgumentMatchers.eq(ShuffleMonitor.class)))

Mockito.when(monitorScheduler.findMonitor(ShuffleMonitor.class))

suneet-s

Looks like the analyzeDependencies job is failing

[WARNING] Unused declared dependencies found:
[WARNING]    org.checkerframework:checker-qual:jar:2.5.7:compile

suneet-s · 2020-10-09T15:15:48Z

indexing-service/src/test/java/org/apache/druid/indexing/worker/shuffle/ShuffleModuleTest.java

+    final ShuffleMonitor shuffleMonitor = new ShuffleMonitor();
+    final MonitorScheduler monitorScheduler = Mockito.mock(MonitorScheduler.class);
+    Mockito.when(monitorScheduler.findMonitor(ArgumentMatchers.eq(ShuffleMonitor.class)))
+           .thenReturn(Optional.of(shuffleMonitor));
+    injector = Guice.createInjector(
+        binder -> {
+          binder.bindScope(LazySingleton.class, Scopes.SINGLETON);
+          binder.bind(MonitorScheduler.class).toInstance(monitorScheduler);
+          binder.bind(IntermediaryDataManager.class).toInstance(Mockito.mock(IntermediaryDataManager.class));
+        },
+        shuffleModule
+    );


nit: you can move this into a @Before method

As monitorScheduler behaves differently in tests, I think it's better to make them in each test. I extracted other common codes as a util method.

jihoonson · 2020-10-09T19:08:17Z

Looks like the analyzeDependencies job is failing
[WARNING] Unused declared dependencies found:
[WARNING]    org.checkerframework:checker-qual:jar:2.5.7:compile

Thanks, fixed now.

* Add shuffle metrics for parallel indexing * javadoc and concurrency test * concurrency * fix javadoc * Feature flag * doc * fix doc and add a test * checkstyle * add tests * fix build and address comments

Add shuffle metrics for parallel indexing

7e0fc0a

jihoonson added Area - Batch Ingestion Design Review labels Sep 4, 2020

abhishekagarwal87 reviewed Sep 7, 2020

View reviewed changes

javadoc and concurrency test

b73677f

clintropolis reviewed Sep 9, 2020

View reviewed changes

concurrency

fcd8cb2

jihoonson added Area - Metrics/Event Emitting Release Notes labels Sep 10, 2020

abhishekagarwal87 reviewed Sep 10, 2020

View reviewed changes

abhishekagarwal87 approved these changes Sep 10, 2020

View reviewed changes

fix javadoc

bf23cbc

clintropolis approved these changes Sep 10, 2020

View reviewed changes

suneet-s reviewed Sep 21, 2020

View reviewed changes

jihoonson added 4 commits October 6, 2020 14:48

Feature flag

80b2949

doc

e258dd3

fix doc and add a test

e205c1d

checkstyle

d70529d

suneet-s reviewed Oct 8, 2020

View reviewed changes

suneet-s approved these changes Oct 9, 2020

View reviewed changes

add tests

394b473

suneet-s reviewed Oct 9, 2020

View reviewed changes

suneet-s approved these changes Oct 9, 2020

View reviewed changes

fix build and address comments

23312bb

suneet-s approved these changes Oct 11, 2020

View reviewed changes

suneet-s merged commit ad437dd into apache:master Oct 11, 2020

jihoonson added this to the 0.21.0 milestone Jan 4, 2021

jihoonson mentioned this pull request Jan 13, 2021

[Draft] 0.21.0 Release Notes #10752

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add shuffle metrics for parallel indexing #10359

Add shuffle metrics for parallel indexing #10359

jihoonson commented Sep 4, 2020 •

edited

Loading

abhishekagarwal87 Sep 7, 2020

jihoonson Sep 8, 2020

abhishekagarwal87 Sep 7, 2020

jihoonson Sep 8, 2020

clintropolis Sep 9, 2020

jihoonson Sep 9, 2020

jihoonson Sep 9, 2020

abhishekagarwal87 Sep 10, 2020

clintropolis left a comment

clintropolis Sep 9, 2020

abhishekagarwal87 Sep 10, 2020

jihoonson Sep 10, 2020

suneet-s left a comment

suneet-s Sep 21, 2020

jihoonson Oct 6, 2020

suneet-s Oct 8, 2020

suneet-s Sep 21, 2020

jihoonson Oct 6, 2020

suneet-s Oct 8, 2020

suneet-s Sep 21, 2020

jihoonson Oct 6, 2020

jihoonson Oct 6, 2020

suneet-s Oct 8, 2020

jihoonson Oct 8, 2020

suneet-s left a comment

suneet-s Oct 9, 2020

suneet-s Oct 9, 2020

jihoonson Oct 9, 2020

suneet-s Oct 9, 2020 •

edited

Loading

suneet-s left a comment

suneet-s Oct 9, 2020

jihoonson Oct 9, 2020

jihoonson commented Oct 9, 2020

	Mockito.when(monitorScheduler.findMonitor(ArgumentMatchers.eq(ShuffleMonitor.class)))
	Mockito.when(monitorScheduler.findMonitor(ShuffleMonitor.class))

Add shuffle metrics for parallel indexing #10359

Add shuffle metrics for parallel indexing #10359

Conversation

jihoonson commented Sep 4, 2020 • edited Loading

Description

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

clintropolis left a 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

suneet-s left a 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

suneet-s left a 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

suneet-s Oct 9, 2020 • edited Loading

Choose a reason for hiding this comment

suneet-s left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jihoonson commented Oct 9, 2020

jihoonson commented Sep 4, 2020 •

edited

Loading

suneet-s Oct 9, 2020 •

edited

Loading