Add job id to log and add running worker failure metrics (#790)

Co-authored-by: ggao <ggao@netflix.com>

Author: hellolittlej

mantis-runtime-executor/src/main/java/io/mantisrx/server/worker/RunningWorker.java

@@ -18,6 +18,9 @@
 
 import static io.mantisrx.server.core.utils.StatusConstants.STATUS_MESSAGE_FORMAT;
 
+import com.netflix.spectator.api.Tag;
+import io.mantisrx.common.metrics.Metrics;
+import io.mantisrx.common.metrics.MetricsRegistry;
 import io.mantisrx.runtime.Context;
 import io.mantisrx.runtime.Job;
 import io.mantisrx.runtime.MantisJobState;
@@ -28,12 +31,14 @@
 import io.mantisrx.server.core.JobSchedulingInfo;
 import io.mantisrx.server.core.Status;
 import io.mantisrx.server.core.Status.TYPE;
-import java.util.Iterator;
+
+import java.util.*;
 import java.util.concurrent.CountDownLatch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import rx.Observable;
 import rx.Observer;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.subjects.PublishSubject;
@@ -43,6 +48,8 @@
 public class RunningWorker {
 
     private static final Logger logger = LoggerFactory.getLogger(RunningWorker.class);
+    private static final String workerFailureMetricName = "workerFailure";
+    private static final String workerMonitorMetricId = "runningWorkerMonitor";
     private final int totalStagesNet;
     private Action0 onTerminateCallback;
     private Action0 onCompleteCallback;
@@ -150,7 +157,19 @@ public void signalCompleted() {
     public void signalFailed(Throwable t) {
         logger.info("JobId: " + jobId + ", stage: " + stageNum + " workerIndex: " + workerIndex + " workerNumber: " + workerNum + ","
                 + " signaling failed");
-        logger.error("Worker failure detected, shutting down job", t);
+        logger.error("Worker failure detected, shutting down job: {}", jobId, t);
+        // Send failure metrics when data emission failed
+        if (t instanceof OnErrorThrowable) {
+            Metrics jobFailureMetrics = new Metrics.Builder()
+                .id(workerMonitorMetricId, Tag.of("jobId", this.jobId),
+                    Tag.of("workerIndex", String.valueOf(this.workerIndex)),
+                    Tag.of("stageNum", String.valueOf(this.stageNum)))
+                .addCounter(workerFailureMetricName)
+                .build();
+
+            MetricsRegistry.getInstance().registerAndGet(jobFailureMetrics).getCounter(workerFailureMetricName).increment();
+        }
+
         jobStatus.onNext(new Status(jobId, stageNum, workerIndex, workerNum,
                 TYPE.INFO, String.format(STATUS_MESSAGE_FORMAT, stageNum, workerIndex, workerNum, "failed. error: " + t.getMessage()),
                 MantisJobState.Failed));