Add worker status and duration metrics in live and task reports #15180

gargvishesh · 2023-10-17T09:02:19Z

Description

Add worker status and duration metrics in live and task reports for tracking. Format:

  "workers": {
    "0": [
      {
        "workerId": "query-df8d9f94-58ba-4d99-8d01-7eb42d547a7e-worker0_0",
        "state": "SUCCESS",
        "durationMs": 2561
      }
    ],
    "1": [
      {
        "workerId": "query-df8d9f94-58ba-4d99-8d01-7eb42d547a7e-worker1_0",
        "state": "FAILED",
        "durationMs": 500
      },
      {
        "workerId": "query-df8d9f94-58ba-4d99-8d01-7eb42d547a7e-worker1_1",
        "state": "RUNNING",
        "durationMs": -1
      }
    ]
  }

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java

cryptoe

Thanks for the PR @gargvishesh. Left an initial review.
Should be good to go once the comments are addressed.

cryptoe · 2023-10-18T14:37:01Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

+  {
+    final Map<Integer, List<WorkerStats>> workerStats = new TreeMap<>();
+
+    for (Map.Entry<String, TaskTracker> taskEntry : taskTrackers.entrySet()) {


private final Map<String, TaskTracker> taskTrackers = new LinkedHashMap<>();
was only called from the main worker loop until now.
WIth this change, the taskTrackers can be called from the jetty thread or the main controller impl thread.

So we should either make TaskTracker thread safe

Good catch! Now wrapped in Collections.synchronizedMap(). Believe this won't have any significant impact on performance.

Lets use a concurrentHashMap ?

Returns a synchronized (thread-safe) map backed by the specified map. In order to guarantee serial access, it is critical that all access to the backing map is accomplished through the returned map. It is imperative that the user manually synchronize on the returned map when traversing any of its collection views via Iterator, Spliterator or Stream: Map m = Collections.synchronizedMap(new HashMap()); ... Set s = m.keySet(); // Needn't be in synchronized block ... synchronized (m) { // Synchronizing on m, not s! Iterator i = s.iterator(); // Must be in synchronized block while (i.hasNext()) foo(i.next()); }

Since its every easy to use s.iteratator() no ?

cryptoe · 2023-10-18T14:38:13Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

@@ -348,6 +351,68 @@ public boolean isTaskLatest(String taskId)
    }
  }

+  public static class WorkerStats
+  {
+    String workerId;


Lets mark these field final ?

Can't - because of the default constructor.

cryptoe · 2023-10-18T14:39:25Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

+    /**
+     * For JSON deserialization only
+     */
+    public WorkerStats()


We actually want serialization only rite ? so this method can go no ?

Many tests in SQLMSQStatementResourceTypeTest fail as they require deserialization -- such as testReplaceAll, testWithDurableStorage, and testResultFormatWithParamInSelect

cryptoe · 2023-10-18T14:41:18Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

+      this.duration = duration;
+    }
+
+    @JsonProperty()


nit is the () required ?

Removed. Thanks!

cryptoe · 2023-10-18T14:53:03Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

+
+      TaskTracker taskTracker = taskEntry.getValue();
+
+      long duration = (taskTracker.status.getDuration() == -1


I think it should be okay to remove the -1 duration check and always report taskTracker.status.getDuration().
We always rely on the overlord system to gives us the task duration without changing anything.
wdyt ?

The reason was that duration is always -1 for RUNNING workers (since it gets updated only upon completion), so publishing their duration wouldn't ever be useful.

MSQ tracks the start time as the time it requested to launch the job. I currently do not know if duration counter in the overlord is started as soon as the overlord gets the request or when ever that task goes into running state.

To test it what you could do is

Make changes to the MSQWorkerTaskLauncher to log the report every time liveReports is called.

Start a cluster with lets say 4 task slots.

Use 2 slots for other ingestions.

Schedule an MSQ job with 4 slots. (The MSQ job will wait and the worker task launcher will set the start time for each task ).

Kill the job running on the other 2 slots.

Check if the taskDuration in the report is going backward.

Did the check both from a run and in code and found it to be correct: the start time in MSQ is recorded when the task is submitted whereas in overlord is upon start of the run.

Currently TaskStatus doesn't have any field to record a startTime, so for MSQ to get a worker's startTime from the Overlord, this field needs to be added and persisted in the database upon the worker task's start.

There is no such issue with reporting of query's duration periodically in live reports since it's a single hop from (from overlord to indexer) and the query's start time is maintained inside the controller -- so the duration can be calculated on-the-fly.

Since we are more interested in timings of finished workers, I think for now it is fine to just report the duration as -1 for worker tasks instead of adding a new field in TaskStatus which is used at multiple places. But do let me know if you think we should take the route of adding a new field.

Thanks for digging in. Reporting -1 for running tasks SGTM.

cryptoe · 2023-10-18T14:57:41Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

+    }
+
+    @JsonProperty()
+    public TaskState getState()


We should also document these properties in docs/api-reference/sql-ingestion-api.md

Added these properties.

…rlord's number

cryptoe

PR is almost there.

cryptoe · 2023-10-23T13:44:59Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

+  {
+    final Map<Integer, List<WorkerStats>> workerStats = new TreeMap<>();
+
+    for (Map.Entry<String, TaskTracker> taskEntry : taskTrackers.entrySet()) {


Lets use a concurrentHashMap ?

Returns a synchronized (thread-safe) map backed by the specified map. In order to guarantee serial access, it is critical that all access to the backing map is accomplished through the returned map. It is imperative that the user manually synchronize on the returned map when traversing any of its collection views via Iterator, Spliterator or Stream: Map m = Collections.synchronizedMap(new HashMap()); ... Set s = m.keySet(); // Needn't be in synchronized block ... synchronized (m) { // Synchronizing on m, not s! Iterator i = s.iterator(); // Must be in synchronized block while (i.hasNext()) foo(i.next()); }

Since its every easy to use s.iteratator() no ?

cryptoe · 2023-10-23T13:45:39Z

...ore/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java

@@ -111,7 +115,7 @@ private enum State
  // Mutable state accessible only to the main loop. LinkedHashMap since order of key set matters. Tasks are added


Lets update the comments as well.

cryptoe

Changes LGTM.
Thanks @gargvishesh .

…he#15180) Add worker status and duration metrics in live and task reports for tracking.

gargvishesh added 2 commits October 17, 2023 14:29

Add workers status and duration in task report.

613023c

Update field names

3a23d03

github-actions bot added Area - Batch Ingestion Area - MSQ For multi stage queries - https://github.com/apache/druid/issues/12262 labels Oct 17, 2023

Update class name

4804c2a

github-advanced-security bot found potential problems Oct 17, 2023

View reviewed changes

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java Fixed Show fixed Hide fixed

gargvishesh changed the title ~~Add worker status and duration metrics in task report~~ Add worker status and duration metrics in live and task reports Oct 17, 2023

gargvishesh added 2 commits October 17, 2023 16:38

Adding default constructor for JSON and other misc changes

d6dc8d9

Change iterator from keyset to entryset

50a0eb7

gargvishesh marked this pull request as ready for review October 18, 2023 08:33

cryptoe reviewed Oct 18, 2023

View reviewed changes

Address review comments

2a8024b

github-actions bot added the Area - Documentation label Oct 19, 2023

gargvishesh added 2 commits October 19, 2023 23:18

Remove auto indent from non-modified lines in sql-ingestion-api.md

ea490fd

Remove on-the-fly calculation of worker task duration and rely on ove…

23497f3

…rlord's number

cryptoe reviewed Oct 23, 2023

View reviewed changes

gargvishesh added 2 commits October 27, 2023 19:29

Change taskTrackers map to ConcurrentSkipListMap and update comments

e98c4f6

Add workerId comparator in taskTrackers Map

7381596

cryptoe approved these changes Oct 30, 2023

View reviewed changes

cryptoe merged commit 039b055 into apache:master Oct 30, 2023
82 checks passed

CaseyPan pushed a commit to CaseyPan/druid that referenced this pull request Nov 17, 2023

Add worker status and duration metrics in live and task reports (apac…

ca58ce5

…he#15180) Add worker status and duration metrics in live and task reports for tracking.

LakshSingla added this to the 29.0.0 milestone Jan 29, 2024

LakshSingla mentioned this pull request Feb 13, 2024

[DRAFT] 29.0.0 release notes #15896

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add worker status and duration metrics in live and task reports #15180

Add worker status and duration metrics in live and task reports #15180

gargvishesh commented Oct 17, 2023 •

edited

Loading

cryptoe left a comment

cryptoe Oct 18, 2023

gargvishesh Oct 19, 2023

cryptoe Oct 23, 2023

cryptoe Oct 18, 2023

gargvishesh Oct 19, 2023

cryptoe Oct 18, 2023

gargvishesh Oct 19, 2023

cryptoe Oct 18, 2023

gargvishesh Oct 19, 2023

cryptoe Oct 18, 2023

gargvishesh Oct 19, 2023

cryptoe Oct 20, 2023

gargvishesh Oct 20, 2023 •

edited by cryptoe

Loading

cryptoe Oct 23, 2023

cryptoe Oct 18, 2023

gargvishesh Oct 19, 2023

cryptoe left a comment

cryptoe Oct 23, 2023

cryptoe Oct 23, 2023

cryptoe left a comment


		TaskTracker taskTracker = taskEntry.getValue();

		long duration = (taskTracker.status.getDuration() == -1

		@@ -111,7 +115,7 @@ private enum State
		// Mutable state accessible only to the main loop. LinkedHashMap since order of key set matters. Tasks are added

Add worker status and duration metrics in live and task reports #15180

Add worker status and duration metrics in live and task reports #15180

Conversation

gargvishesh commented Oct 17, 2023 • edited Loading

Description

cryptoe 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

Choose a reason for hiding this comment

gargvishesh Oct 20, 2023 • edited by cryptoe 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

cryptoe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cryptoe left a comment

Choose a reason for hiding this comment

gargvishesh commented Oct 17, 2023 •

edited

Loading

gargvishesh Oct 20, 2023 •

edited by cryptoe

Loading