Changing the running task count to gauge
@@ -28,6 +28,7 @@ import org.apache.airavata.helix.task.api.TaskHelper;
import org.apache.airavata.helix.task.api.annotation.TaskOutPort;
import org.apache.airavata.helix.task.api.annotation.TaskParam;
import org.apache.airavata.patform.monitoring.CountMonitor;
+import org.apache.airavata.patform.monitoring.GaugeMonitor;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
@@ -50,7 +51,7 @@ public abstract class AbstractTask extends UserContentStore implements Task {
private final static Logger logger = LoggerFactory.getLogger(AbstractTask.class);
private final static CountMonitor taskInitCounter = new CountMonitor("task_init_count");
- private final static CountMonitor taskRunCounter = new CountMonitor("task_run_count");
+ private final static GaugeMonitor taskRunGauge = new GaugeMonitor("task_run_gauge");
private final static CountMonitor taskCancelCounter = new CountMonitor("task_cancel_count");
private final static CountMonitor taskFailCounter = new CountMonitor("task_fail_count");
private final static CountMonitor taskCompleteCounter = new CountMonitor("task_complete_count");
@@ -93,7 +94,7 @@ public abstract class AbstractTask extends UserContentStore implements Task {
@Override
public final TaskResult run() {
try {
- taskRunCounter.inc();
+ taskRunGauge.inc();
boolean isThisNextJob = getUserContent(WORKFLOW_STARTED, Scope.WORKFLOW) == null ||
this.callbackContext.getJobConfig().getJobId()
.equals(this.callbackContext.getJobConfig().getWorkflow() + "_" + getUserContent(NEXT_JOB, Scope.WORKFLOW));
@@ -111,6 +112,7 @@ public abstract class AbstractTask extends UserContentStore implements Task {
@Override
public final void cancel() {
try {
+ taskRunGauge.dec();
taskCancelCounter.inc();
logger.info("Cancelling task " + taskId);
onCancel();
@@ -128,6 +130,7 @@ public abstract class AbstractTask extends UserContentStore implements Task {
public abstract void onCancel();
protected TaskResult onSuccess(String message) {
+ taskRunGauge.dec();
taskCompleteCounter.inc();
String successMessage = "Task " + getTaskId() + " completed." + (message != null ? " Message : " + message : "");
logger.info(successMessage);
@@ -135,6 +138,7 @@ public abstract class AbstractTask extends UserContentStore implements Task {
}
protected TaskResult onFail(String reason, boolean fatal) {
+ taskRunGauge.dec();
taskFailCounter.inc();
return new TaskResult(fatal ? TaskResult.Status.FATAL_FAILED : TaskResult.Status.FAILED, reason);
}
