Add custom timeInTopic metric to timeseries-aggregator (#172)

* Add timeInTopic metric to timeseries-aggregator

* Fix spelling error

* Make explicit method call

* Remove extraneous mean

* Correct units in the histogram update

timeseries-aggregator/build/docker/jmxtrans-agent.xml

@@ -74,6 +74,8 @@
                resultAlias="metricpoint.ser.success.#attribute#"/>
         <query objectName="metrics:name=metricpoints.disordered" attributes="OneMinuteRate,MeanRate"
                resultAlias="metricpoint.disordered.#attribute#"/>
+        <query objectName="metrics:name=metricpoints.timeInTopic" attributes="95thPercentile,99thPercentile,999thPercentile,Count,Max,Median,Mean"
+               resultAlias="metricpoint.timeInTopic.#attribute#"/>
         <query objectName="metrics:name=metricpoints.compute.failure" attributes="OneMinuteRate,MeanRate"
                resultAlias="metricpoint.compute.failure.#attribute#"/>
         <query objectName="metrics:name=metricprocessor.invalid" attributes="OneMinuteRate,MeanRate"

timeseries-aggregator/src/main/scala/com/expedia/www/haystack/trends/aggregation/WindowedMetric.scala

@@ -18,7 +18,9 @@
 
 package com.expedia.www.haystack.trends.aggregation
 
-import com.codahale.metrics.Meter
+import java.util.concurrent.TimeUnit
+
+import com.codahale.metrics.{Histogram, Meter}
 import com.expedia.metrics.MetricData
 import com.expedia.www.haystack.commons.entities.Interval.Interval
 import com.expedia.www.haystack.commons.metrics.MetricsSupport
@@ -38,6 +40,7 @@ import scala.util.Try
 class WindowedMetric private(var windowedMetricsMap: mutable.TreeMap[TimeWindow, Metric], metricFactory: MetricFactory, numberOfWatermarkedWindows: Int, interval: Interval) extends MetricsSupport {
 
   private val disorderedMetricPointMeter: Meter = metricRegistry.meter("metricpoints.disordered")
+  private val timeInTopicMetricPointHistogram: Histogram = metricRegistry.histogram("metricpoints.timeInTopic")
   private var computedMetrics = List[(Long, Metric)]()
   private val LOGGER = LoggerFactory.getLogger(this.getClass)
 
@@ -52,6 +55,8 @@ class WindowedMetric private(var windowedMetricsMap: mutable.TreeMap[TimeWindow,
     * @param incomingMetricData - incoming metric data
     */
   def compute(incomingMetricData: MetricData): Unit = {
+    timeInTopicMetricPointHistogram.update(TimeUnit.SECONDS.toMillis(incomingMetricData.getTimestamp()) - System.currentTimeMillis())
+
     val incomingMetricPointTimeWindow = TimeWindow.apply(incomingMetricData.getTimestamp, interval)
 
     val matchedWindowedMetric = windowedMetricsMap.get(incomingMetricPointTimeWindow)