From 379604d4425e9cfed89e2913c627c501d886071d Mon Sep 17 00:00:00 2001
From: Rishabh Jain <rishabhjainnsit@gmail.com>
Date: Thu, 26 Sep 2019 19:11:13 -0400
Subject: [PATCH] STORM-3186: Customizable configuration for metric reporting
 interval

---
 .../src/main/java/org/apache/storm/DaemonConfig.java       | 6 ++++++
 .../metrics/reporters/ConsolePreparableReporter.java       | 7 ++++++-
 .../daemon/metrics/reporters/CsvPreparableReporter.java    | 7 ++++++-
 3 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
index 28076f4edfb..210d632e505 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -74,6 +74,12 @@ public class DaemonConfig implements Validated {
     @IsString
     public static final String STORM_DAEMON_METRICS_REPORTER_PLUGIN_DOMAIN = "storm.daemon.metrics.reporter.plugin.domain";
 
+    /**
+     * We report the metrics with this interval period.
+     */
+    @IsString
+    public static final String STORM_DAEMON_METRICS_REPORTER_INTERVAL_SECS = "storm.daemon.metrics.reporter.interval.secs";
+
     /**
      * A specify csv reporter directory for CvsPreparableReporter daemon metrics reporter.
      */
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java
index 97470ee4b6b..758b87598f3 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/ConsolePreparableReporter.java
@@ -17,13 +17,17 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+
+import org.apache.storm.DaemonConfig;
 import org.apache.storm.daemon.metrics.ClientMetricsUtils;
+import org.apache.storm.utils.ObjectReader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class ConsolePreparableReporter implements PreparableReporter {
     private static final Logger LOG = LoggerFactory.getLogger(ConsolePreparableReporter.class);
     ConsoleReporter reporter = null;
+    Integer reportingIntervalPeriod = null;
 
     @Override
     public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf) {
@@ -46,13 +50,14 @@ public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf
             builder.convertDurationsTo(durationUnit);
         }
         reporter = builder.build();
+        reportingIntervalPeriod = ObjectReader.getInt(topoConf.get(DaemonConfig.STORM_DAEMON_METRICS_REPORTER_INTERVAL_SECS), 10);
     }
 
     @Override
     public void start() {
         if (reporter != null) {
             LOG.debug("Starting...");
-            reporter.start(10, TimeUnit.SECONDS);
+            reporter.start(reportingIntervalPeriod, TimeUnit.SECONDS);
         } else {
             throw new IllegalStateException("Attempt to start without preparing " + getClass().getSimpleName());
         }
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java
index 51b91120b1f..322ed8867af 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/metrics/reporters/CsvPreparableReporter.java
@@ -18,14 +18,18 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+
+import org.apache.storm.DaemonConfig;
 import org.apache.storm.daemon.metrics.ClientMetricsUtils;
 import org.apache.storm.daemon.metrics.MetricsUtils;
+import org.apache.storm.utils.ObjectReader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class CsvPreparableReporter implements PreparableReporter {
     private static final Logger LOG = LoggerFactory.getLogger(CsvPreparableReporter.class);
     CsvReporter reporter = null;
+    Integer reportingIntervalPeriod = null;
 
     @Override
     public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf) {
@@ -49,13 +53,14 @@ public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf
 
         File csvMetricsDir = MetricsUtils.getCsvLogDir(topoConf);
         reporter = builder.build(csvMetricsDir);
+        reportingIntervalPeriod = ObjectReader.getInt(topoConf.get(DaemonConfig.STORM_DAEMON_METRICS_REPORTER_INTERVAL_SECS), 10);
     }
 
     @Override
     public void start() {
         if (reporter != null) {
             LOG.debug("Starting...");
-            reporter.start(10, TimeUnit.SECONDS);
+            reporter.start(reportingIntervalPeriod, TimeUnit.SECONDS);
         } else {
             throw new IllegalStateException("Attempt to start without preparing " + getClass().getSimpleName());
         }