diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java index cdf767f1db..d80696d139 100644 --- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java +++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java @@ -1032,7 +1032,7 @@ public void dagComplete() { long tgVersionForZk; writeLock.lock(); try { - dagRunning = false; + setDagRunning(false); dagStats = new StatsPerDag(); int pendingCount = 0; for (Entry> entry : pendingTasks.entrySet()) { @@ -1124,7 +1124,7 @@ public void allocateTask(Object task, Resource capability, String[] hosts, Strin if (!dagRunning && metrics != null && id != null) { metrics.setDagId(id.getTaskID().getVertexID().getDAGId().toString()); } - dagRunning = true; + setDagRunning(true); dagStats.registerTaskRequest(hosts, racks); } finally { writeLock.unlock(); @@ -1148,7 +1148,7 @@ public void allocateTask(Object task, Resource capability, ContainerId container if (!dagRunning && metrics != null && id != null) { metrics.setDagId(id.getTaskID().getVertexID().getDAGId().toString()); } - dagRunning = true; + setDagRunning(true); dagStats.registerTaskRequest(null, null); } finally { writeLock.unlock(); @@ -2297,6 +2297,13 @@ private void maybeAddToDelayedTaskQueue(TaskInfo taskInfo) { } } + private void setDagRunning(boolean running) { + dagRunning = running; + if (metrics != null) { + metrics.setDagRunning(running ? 1 : 0); + } + } + // ------ Inner classes defined after this point ------ @VisibleForTesting diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerInfo.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerInfo.java index 0750dc6aa7..cfe9aa5a95 100644 --- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerInfo.java +++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerInfo.java @@ -27,6 +27,7 @@ */ public enum LlapTaskSchedulerInfo implements MetricsInfo { SchedulerMetrics("Llap task scheduler related metrics"), + SchedulerDagRunning("Whether a DAG is currently running"), SchedulerClusterNodeCount("Number of nodes in the cluster"), SchedulerExecutorsPerInstance("Total number of executor threads per node"), SchedulerMemoryPerInstance("Total memory for executors per node in bytes"), diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerMetrics.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerMetrics.java index c6b5cc1770..dfa38f7bab 100644 --- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerMetrics.java +++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/metrics/LlapTaskSchedulerMetrics.java @@ -15,19 +15,7 @@ */ package org.apache.hadoop.hive.llap.tezplugins.metrics; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerClusterNodeCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerCompletedDagCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerCpuCoresPerInstance; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerDisabledNodeCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerExecutorsPerInstance; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerMemoryPerInstance; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerMetrics; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerPendingPreemptionTaskCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerPendingTaskCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerPreemptedTaskCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerRunningTaskCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerSchedulableTaskCount; -import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.SchedulerSuccessfulTaskCount; +import static org.apache.hadoop.hive.llap.tezplugins.metrics.LlapTaskSchedulerInfo.*; import static org.apache.hadoop.metrics2.impl.MsInfo.ProcessName; import static org.apache.hadoop.metrics2.impl.MsInfo.SessionId; @@ -92,6 +80,8 @@ MutableCounterInt wmSpeculativeCount; @Metric MutableCounterInt wmGuaranteedCount; + @Metric + MutableGaugeInt dagRunning; private LlapTaskSchedulerMetrics(String displayName, JvmMetrics jm, String sessionId) { this.name = displayName; @@ -254,6 +244,10 @@ public void setWmUnusedGuaranteed(int unusedGuaranteed) { wmUnusedGuaranteedCount.set(unusedGuaranteed); } + public void setDagRunning(int value) { + dagRunning.set(value); + } + public void resetWmMetrics() { wmTotalGuaranteedCount.set(0); wmUnusedGuaranteedCount.set(0); @@ -265,6 +259,7 @@ public void resetWmMetrics() { private void getTaskSchedulerStats(MetricsRecordBuilder rb) { rb.addGauge(SchedulerClusterNodeCount, clusterNodeCount.value()) + .addCounter(SchedulerDagRunning, dagRunning.value()) .addGauge(SchedulerExecutorsPerInstance, numExecutors.value()) .addGauge(SchedulerMemoryPerInstance, memoryPerInstance.value()) .addGauge(SchedulerCpuCoresPerInstance, cpuCoresPerInstance.value())