Fix metrics for streaming jobs

marmaray/src/main/java/com/uber/marmaray/common/actions/JobDagActions.java

@@ -62,7 +62,7 @@ public final class JobDagActions {
     private static final String ACTION_TARGET = "action_target";
 
     @Getter
-    private final Queue<IJobDagAction> actions;
+    private Queue<IJobDagAction> actions;
     private final Reporters reporters;
 
     @Getter
@@ -78,6 +78,10 @@ public JobDagActions(@NonNull final Reporters reporters, @NotEmpty final String
         this.reporters = reporters;
     }
 
+    public void reset() {
+        this.actions = new ConcurrentLinkedDeque<>();
+    }
+
     /**
      * Add an action to the container
      * @param action IAction to hold

marmaray/src/main/java/com/uber/marmaray/common/actions/ReporterAction.java

@@ -36,8 +36,8 @@ public class ReporterAction implements IJobDagAction {
     @Getter
     private final Reporters reporters;
     @Getter
-    private final JobMetrics jobMetrics;
-    private final DataFeedMetrics dataFeedMetrics;
+    private JobMetrics jobMetrics;
+    private DataFeedMetrics dataFeedMetrics;
 
     @Override
     public boolean execute(final boolean success) {

marmaray/src/main/java/com/uber/marmaray/common/job/Dag.java

@@ -26,4 +26,5 @@ public Dag(@NonNull final String jobName, @NonNull final String dataFeedName) {
 
     public abstract IStatus execute();
 
+    public abstract void clean();
 }

marmaray/src/main/java/com/uber/marmaray/common/job/JobDag.java

@@ -211,6 +211,14 @@ public IStatus execute() {
         return status;
     }
 
+    @Override
+    public void clean() {
+        this.sinkDag.clean();
+        this.dataFeedMetrics.clean();
+        this.jobMetrics.clean();
+        this.postJobDagActions.reset();
+    }
+
     private void reportStatus(final boolean successful) {
         final long statusValue =
                 successful ? DataFeedMetricNames.RESULT_SUCCESS : DataFeedMetricNames.RESULT_FAILURE;

marmaray/src/main/java/com/uber/marmaray/common/job/JobManager.java

@@ -115,7 +115,7 @@ private JobManager(@NonNull final Configuration conf, @NotEmpty final String app
         this.jobMetrics = new JobMetrics(appName);
         this.reporters = reporters;
         this.sparkFactory = sparkFactory;
-        this.reporters.addReporter(new ConsoleReporter());
+//        this.reporters.addReporter(new ConsoleReporter());
         this.jobLockManager = new JobLockManager(conf, frequency, shouldLockFrequency);
         this.postJobManagerActions = new JobDagActions(this.reporters, "jobManager");
         this.appId = sparkFactory.getSparkContext().sc().applicationId();
@@ -251,14 +251,15 @@ public void run(@NonNull final StreamingConfiguration streamingConf) {
                 }
             } finally {
                 this.postJobManagerActions.execute(isSuccess.get());
-                shutdown(!isSuccess.get());
+                resetManager(!isSuccess.get());
                 this.reporters.getReporters().forEach(IReporter::finish);
             }
         } while (streamingConf.isEnabled);
+        this.shutdown(false);
     }
 
     /**
-     * Add {@link JobDag} to be executed on {@link #run()}
+     * Add {@link JobDag} to be executed on
      * @param jobDag JobDag to be added
      */
     public void addJobDag(@NonNull final Dag jobDag) {
@@ -270,7 +271,7 @@ public void addJobDag(@NonNull final Dag jobDag) {
     }
 
     /**
-     * Remove {@link JobDag} to be executed on {@link #run()}
+     * Remove {@link JobDag} to be executed on
      * @param jobDag JobDag to be added
      */
     public void removeJobDag(@NonNull final Dag jobDag) {
@@ -282,7 +283,7 @@ public void removeJobDag(@NonNull final Dag jobDag) {
     }
 
     /**
-     * Add collection of {@link JobDag} to be executed on {@link #run()}
+     * Add collection of {@link JobDag} to be executed on
      * @param jobDags collection of JobDags to be added
      */
     public void addJobDags(@NonNull final Collection<? extends JobDag> jobDags) {
@@ -315,19 +316,29 @@ public static void reset() {
         }
     }
 
-    private void shutdown(final boolean forceShutdown) {
-        ThreadPoolService.shutdown(forceShutdown);
+    private void resetManager(final boolean forceReset) {
+        ThreadPoolService.shutdown(forceReset);
         if (this.isJobManagerMetadataEnabled()) {
             this.jobDags.forEach(jobDag -> this.getTracker().set(jobDag.getDataFeedName(),
-                jobDag.getJobManagerMetadata()));
+                    jobDag.getJobManagerMetadata()));
             try {
                 this.getTracker().writeJobManagerMetadata();
             } catch (MetadataException e) {
                 log.error("Unable to save metadata: {}", e.getMessage());
             }
         }
-//        this.sparkFactory.stop();
+
+        // Cleanup metrics
+        this.jobDags.forEach(dag -> dag.clean());
+        this.postJobManagerActions.reset();
+        this.jobLockManager.reset();
+        this.jobMetrics.clean();
+    }
+
+    public void shutdown(final boolean forceShutdown) {
+        this.resetManager(forceShutdown);
         this.jobLockManager.stop();
+        this.sparkFactory.stop();
     }
 
     private static void setSparkStageName(@NonNull final JavaSparkContext jsc, @NotEmpty final String dataFeedName) {
@@ -350,9 +361,9 @@ private final class JobLockManager {
         private final String jobFrequency;
 
         @NonNull
-        private final TimerMetric managerTimerMetric;
+        private TimerMetric managerTimerMetric;
         @NonNull
-        private final HashMap<String, TimerMetric> dagTimerMetricMap;
+        private HashMap<String, TimerMetric> dagTimerMetricMap;
 
         private JobLockManager(@NonNull final Configuration conf, @NotEmpty final String frequency,
                 final boolean shouldLockFrequency) {
@@ -396,8 +407,18 @@ private boolean lockDag(@NotEmpty final String jobDagName, @NotEmpty final Strin
         private void stop() {
             log.info("Closing the LockManager in the JobManager.");
             this.lockManager.close();
+            this.reset();
+        }
+
+        private void reset() {
+            managerTimerMetric.stop();
             reporters.report(managerTimerMetric);
+            this.managerTimerMetric = new TimerMetric(JobMetricNames.JOB_MANAGER_LOCK_TIME_MS,
+                    ImmutableMap.of(JOB_FREQUENCY_TAG, jobFrequency,
+                            JOB_NAME_TAG, appName));
+
             dagTimerMetricMap.forEach((dagName, timerMetric) -> reporters.report(timerMetric));
+            this.dagTimerMetricMap = new HashMap<>();
         }
     }
 }

marmaray/src/main/java/com/uber/marmaray/common/job/JobSubDag.java

@@ -127,6 +127,16 @@ public void setDataFeedMetrics(@NonNull final DataFeedMetrics dataFeedMetrics) {
     protected void commitNode() {
     }
 
+    public void clean() {
+        if (this.dataFeedMetrics.isPresent()) {
+            this.dataFeedMetrics.get().clean();
+        }
+
+        if (this.jobMetrics.isPresent()) {
+            this.jobMetrics.get().clean();
+        }
+    }
+
     /**
      * Called to retrieve payload for child dag.
      *

marmaray/src/main/java/com/uber/marmaray/common/metrics/DataFeedMetrics.java

@@ -44,10 +44,10 @@ public class DataFeedMetrics implements Reportable, Serializable {
     private static final String JOB_TAG = "job";
 
     @Getter
-    private final Set<Metric> metricSet = Sets.newHashSet();
+    private Set<Metric> metricSet = Sets.newHashSet();
 
     @Getter
-    private final Set<Metric> failureMetricSet = Sets.newHashSet();
+    private Set<Metric> failureMetricSet = Sets.newHashSet();
 
     private final Map<String, String> baseTags;
 
@@ -128,4 +128,9 @@ public void gaugeAll(final IReporter reporter) {
         guageNonFailureMetric(reporter);
         gauageFailureMetric(reporter);
     }
+
+    public void clean() {
+        this.failureMetricSet = Sets.newHashSet();
+        this.metricSet = Sets.newHashSet();
+    }
 }

marmaray/src/main/java/com/uber/marmaray/common/metrics/JobMetrics.java

@@ -35,7 +35,7 @@
 public class JobMetrics implements Reportable {
     private static final String METRIC_PREFIX = "job-";
     private static final String JOB_TAG = "job";
-    private final Set<Metric> metricSet = Sets.newHashSet();
+    private Set<Metric> metricSet = Sets.newHashSet();
     private final Map<String, String> baseTags;
     @NotEmpty
     private final String jobName;
@@ -72,4 +72,8 @@ private static String getMetricName(@NonNull final JobMetricType metricType) {
     public void gaugeAll(@NonNull final IReporter reporter) {
         metricSet.forEach(reporter::gauge);
     }
+
+    public void clean() {
+        this.metricSet = Sets.newHashSet();
+    }
 }

marmaray/src/main/java/com/uber/marmaray/common/reporters/ElasticsearchReporter.java

@@ -28,7 +28,7 @@
 public class ElasticsearchReporter implements IReporter<Metric> {
     final private String serverURL;
     final private int serverPort;
-    final private BulkRequest request;
+    private BulkRequest request;
     final private String indexName;
 
     public ElasticsearchReporter(@NonNull final String serverURL, @NonNull final int serverPort,
@@ -68,6 +68,7 @@ public void finish() {
                     }
                 }
             }
+            this.request = new BulkRequest();
             client.close();
         } catch (IOException e) {
             e.printStackTrace();

marmaray/src/main/java/com/uber/marmaray/examples/job/KafkaToHoodieJob.java

@@ -13,6 +13,7 @@
 import com.uber.marmaray.common.metadata.IMetadataManager;
 import com.uber.marmaray.common.metrics.*;
 import com.uber.marmaray.common.reporters.ConsoleReporter;
+import com.uber.marmaray.common.reporters.ElasticsearchReporter;
 import com.uber.marmaray.common.reporters.Reporters;
 import com.uber.marmaray.common.schema.kafka.KafkaSchemaJSONServiceReader;
 import com.uber.marmaray.common.sinks.hoodie.HoodieSink;
@@ -75,10 +76,10 @@ private void run(final String[] args) throws IOException {
         final StreamingConfiguration streamingConf = new StreamingConfiguration(conf);
 
         final Reporters reporters = new Reporters();
-        reporters.addReporter(new ConsoleReporter());
+        reporters.addReporter(new ElasticsearchReporter("0.0.0.0", 9200, "marmaray_metrics"));
 
         final Map<String, String> metricTags = Collections.emptyMap();
-        final DataFeedMetrics dataFeedMetrics = new DataFeedMetrics("KafkaToHoodieJob", metricTags);
+        final DataFeedMetrics dataFeedMetrics = new DataFeedMetrics("catalog_name_job", metricTags);
 
         log.info("Initializing configurations for job");
         final TimerMetric confInitMetric = new TimerMetric(DataFeedMetricNames.INIT_CONFIG_LATENCY_MS,
@@ -90,7 +91,7 @@ private void run(final String[] args) throws IOException {
         try {
             hadoopConf = new HadoopConfiguration(conf);
             kafkaSourceConf = new KafkaSourceConfiguration(conf);
-            hoodieConf = new HoodieConfiguration(conf, "test_hoodie");
+            hoodieConf = new HoodieConfiguration(conf, "catalog_name");
         } catch (final Exception e) {
             final LongMetric configError = new LongMetric(DataFeedMetricNames.DISPERSAL_CONFIGURATION_INIT_ERRORS, 1);
             configError.addTags(metricTags);
@@ -151,8 +152,8 @@ private void run(final String[] args) throws IOException {
 
             log.info("Initializing job dag");
             final JobDag jobDag = new JobDag(kafkaSource, hoodieSink, metadataManager, workUnitCalculator,
-                    "test", "test", new JobMetrics("marmaray"), dataFeedMetrics,
-                    reporters);
+                    "catalog_name", "catalog_name", new JobMetrics("catalog_name"),
+                    dataFeedMetrics, reporters);
 
             jobManager.addJobDag(jobDag);
 
@@ -182,8 +183,6 @@ private void run(final String[] args) throws IOException {
             jobLatencyMetric.stop();
             reporters.report(jobLatencyMetric);
             reporters.finish();
-        } finally {
-            sparkFactory.stop();
         }
     }