Add TaskCountStatsMonitor to monitor task count stats #6657
Conversation
| @@ -100,6 +103,11 @@ | |||
|
|
|||
| private static final EmittingLogger log = new EmittingLogger(TaskQueue.class); | |||
|
|
|||
| private final Map<String, AtomicLong> totalSuccessfulTaskCount = new ConcurrentHashMap<>(); | |||
| private final Map<String, AtomicLong> totalFailedTaskCount = new ConcurrentHashMap<>(); | |||
| private Map<String, Long> prevSuccessfulTaskCount = new HashMap<>(); | |||
There was a problem hiding this comment.
How about prevTotalSuccessfulTaskCount? Same for prevFailedTaskCount.
|
|
||
| if (status.isSuccess()) { | ||
| totalSuccessfulTaskCount.computeIfAbsent(task.getDataSource(), k -> new AtomicLong()); | ||
| totalSuccessfulTaskCount.get(task.getDataSource()).incrementAndGet(); |
There was a problem hiding this comment.
computeIfAbsent() returns the proper instance of AtomicLong, so you can just call incrementAndGet() directly on that instance like below. This is better because we can avoid unnecessary get().
totalSuccessfulTaskCount.computeIfAbsent(task.getDataSource(), k -> new AtomicLong())
.incrementAndGet();| totalSuccessfulTaskCount.get(task.getDataSource()).incrementAndGet(); | ||
| } else { | ||
| totalFailedTaskCount.computeIfAbsent(task.getDataSource(), k -> new AtomicLong()); | ||
| totalFailedTaskCount.get(task.getDataSource()).incrementAndGet(); |
| import java.util.concurrent.atomic.AtomicReference; | ||
| import java.util.concurrent.locks.ReentrantLock; | ||
|
|
||
| /** | ||
| * Encapsulates the indexer leadership lifecycle. | ||
| */ | ||
| public class TaskMaster | ||
| public class TaskMaster implements TaskCountStatsProvider |
There was a problem hiding this comment.
Does TaskQueue also need to be TaskCountStatsProvider?
There was a problem hiding this comment.
Hmm, TaskCountStatsProvider is bind to TaskMaster, there is no need to let TaskQueue be TaskCountStatsProvider.
| |`task/success/count`|Number of successful tasks per emission period. This metric is only available if the TaskCountStatsMonitor module is included.|dataSource.|Varies.| | ||
| |`task/failed/count`|Number of failed tasks per emission period. This metric is only available if the TaskCountStatsMonitor module is included.|dataSource.|Varies.| | ||
| |`task/running/count`|Number of current running tasks. This metric is only available if the TaskCountStatsMonitor module is included.|dataSource.|Varies.| | ||
| |`task/pending/count`|Number of current pending tasks. This metric is only available if the TaskCountStatsMonitor module is included.|dataSource.|Varies.| |
There was a problem hiding this comment.
Can we also add a metric for waiting tasks? It will be useful too.
| @@ -586,4 +602,36 @@ private void syncFromStorage() | |||
| return rv; | |||
| } | |||
|
|
|||
| private Map<String, Long> getDeltaValues(Map<String, Long> total, Map<String, Long> prev) | |||
| { | |||
| return total.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() - prev.getOrDefault(e.getKey(), 0L))); | |||
There was a problem hiding this comment.
Please check the line width. The max line is 120 (https://github.com/apache/incubator-druid/blob/master/eclipse_formatting.xml#L96). Same for the below methods.
|
|
||
| import java.util.Map; | ||
|
|
||
| public class TaskCountStatsMonitor extends AbstractMonitor |
There was a problem hiding this comment.
Would you please add a unit test? I think it should verify the values emitted from this monitor.
| @Override | ||
| public Map<String, Long> getSuccessfulTaskCount() | ||
| { | ||
| return null; |
Have tested in real cluster and now running in production.