You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by na...@apache.org on 2015/08/03 21:02:17 UTC
samza git commit: SAMZA-751: Make MetricsSnapshotReporter's interval
to report metrics configurable
Repository: samza
Updated Branches:
refs/heads/master bf094c8e5 -> 9d6747598
SAMZA-751: Make MetricsSnapshotReporter's interval to report metrics configurable
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/9d674759
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/9d674759
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/9d674759
Branch: refs/heads/master
Commit: 9d6747598111434270f9be549947df1a1ab3bc1e
Parents: bf094c8
Author: Shadi A. Noghabi <ab...@illinois.edu>
Authored: Mon Aug 3 11:50:39 2015 -0700
Committer: Navina <na...@gmail.com>
Committed: Mon Aug 3 11:50:39 2015 -0700
----------------------------------------------------------------------
.../versioned/jobs/configuration-table.html | 14 ++++++++++++++
.../scala/org/apache/samza/config/MetricsConfig.scala | 3 +++
.../metrics/reporter/MetricsSnapshotReporter.scala | 8 ++++----
.../reporter/MetricsSnapshotReporterFactory.scala | 6 ++++++
4 files changed, 27 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/9d674759/docs/learn/documentation/versioned/jobs/configuration-table.html
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html
index ea73b40..926fd50 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -1377,6 +1377,20 @@
</ul>
</td>
</tr>
+ <tr>
+ <td class="property" id="metrics-reporter-polling-interval">metrics.reporter.<br><span class="reporter">reporter-name</span>.interval</td>
+ <td class="default"></td>
+ <td class="description">
+ If you have registered the metrics reporter
+ <a href="#metrics-reporter-class" class="property">metrics.reporter.*.class</a>
+ <code>= org.apache.samza.metrics.reporter.MetricsSnapshotReporterFactory</code>,
+ you can use this property to configure how frequently the reporter will report the metrics
+ registered with it. The value for this property should be length of the interval between
+ consecutive metric reporting. This value is in seconds, and should be a positive integer value.
+ This property is optional and set to 60 by default, which means metrics will be reported every
+ 60 seconds.
+ </td>
+ </tr>
</tbody>
</table>
</body>
http://git-wip-us.apache.org/repos/asf/samza/blob/9d674759/samza-core/src/main/scala/org/apache/samza/config/MetricsConfig.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/config/MetricsConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/MetricsConfig.scala
index f7a11c5..c3fd8bf 100644
--- a/samza-core/src/main/scala/org/apache/samza/config/MetricsConfig.scala
+++ b/samza-core/src/main/scala/org/apache/samza/config/MetricsConfig.scala
@@ -25,6 +25,7 @@ object MetricsConfig {
val METRICS_REPORTERS = "metrics.reporters"
val METRICS_REPORTER_FACTORY = "metrics.reporter.%s.class"
val METRICS_SNAPSHOT_REPORTER_STREAM = "metrics.reporter.%s.stream"
+ val METRICS_SNAPSHOT_REPORTER_INTERVAL= "metrics.reporter.%s.interval"
implicit def Config2Metrics(config: Config) = new MetricsConfig(config)
}
@@ -36,6 +37,8 @@ class MetricsConfig(config: Config) extends ScalaMapConfig(config) {
def getMetricsReporterStream(name: String): Option[String] = getOption(MetricsConfig.METRICS_SNAPSHOT_REPORTER_STREAM format name)
+ def getMetricsReporterInterval(name: String): Option[String] = getOption(MetricsConfig.METRICS_SNAPSHOT_REPORTER_INTERVAL format name)
+
/**
* Returns a list of all metrics names from the config file. Useful for
* getting individual metrics.
http://git-wip-us.apache.org/repos/asf/samza/blob/9d674759/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.scala b/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.scala
index b6696f8..fb438a4 100644
--- a/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.scala
+++ b/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporter.scala
@@ -56,6 +56,7 @@ object MetricsSnapshotReporter {
class MetricsSnapshotReporter(
producer: SystemProducer,
out: SystemStream,
+ pollingInterval: Int,
jobName: String,
jobId: String,
containerName: String,
@@ -69,8 +70,8 @@ class MetricsSnapshotReporter(
val resetTime = clock()
var registries = List[(String, ReadableMetricsRegistry)]()
- info("got metrics snapshot reporter properties [job name: %s, job id: %s, containerName: %s, version: %s, samzaVersion: %s, host: %s]"
- format (jobName, jobId, containerName, version, samzaVersion, host))
+ info("got metrics snapshot reporter properties [job name: %s, job id: %s, containerName: %s, version: %s, samzaVersion: %s, host: %s, pollingInterval %s]"
+ format (jobName, jobId, containerName, version, samzaVersion, host, pollingInterval))
def start {
info("Starting producer.")
@@ -79,8 +80,7 @@ class MetricsSnapshotReporter(
info("Starting reporter timer.")
- // TODO could make this configurable.
- executor.scheduleWithFixedDelay(this, 0, 60, TimeUnit.SECONDS)
+ executor.scheduleWithFixedDelay(this, 0, pollingInterval, TimeUnit.SECONDS)
}
def register(source: String, registry: ReadableMetricsRegistry) {
http://git-wip-us.apache.org/repos/asf/samza/blob/9d674759/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala b/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
index dcf0435..6b7f0ba 100644
--- a/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
+++ b/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
@@ -107,9 +107,15 @@ class MetricsSnapshotReporterFactory extends MetricsReporterFactory with Logging
info("Got serde %s." format serde)
+ val pollingInterval: Int = config
+ .getMetricsReporterInterval(name)
+ .getOrElse("60").toInt
+
+ info("Setting polling interval to %d" format pollingInterval)
val reporter = new MetricsSnapshotReporter(
producer,
systemStream,
+ pollingInterval,
jobName,
jobId,
containerName,