Atlas Cloud Watch: Adjust the firehose receiver to only publish value…

…s that are (#422)

equal to or greater than the configured offset interval. This avoids reporting stale
data unknowingly and handling timeouts similar to the polling version. This means
the offsets must be tuned per AWS namespace (and sometimes metric).
Also incorporate the debugger.

atlas-cloudwatch/src/main/scala/com/netflix/atlas/cloudwatch/LocalCloudWatchMetricsProcessor.scala

@@ -46,9 +46,10 @@ class LocalCloudWatchMetricsProcessor(
   registry: Registry,
   rules: CloudWatchRules,
   tagger: Tagger,
-  publishRouter: PublishRouter
+  publishRouter: PublishRouter,
+  debugger: CloudWatchDebugger
 )(override implicit val system: ActorSystem)
-    extends CloudWatchMetricsProcessor(config, registry, rules, tagger, publishRouter) {
+    extends CloudWatchMetricsProcessor(config, registry, rules, tagger, publishRouter, debugger) {
 
   //                                               writeTS, expSec, data
   private val cache = new ConcurrentHashMap[Long, (Long, Long, Array[Byte])]

atlas-cloudwatch/src/main/scala/com/netflix/atlas/cloudwatch/RedisClusterCloudWatchMetricsProcessor.scala

@@ -65,9 +65,10 @@ class RedisClusterCloudWatchMetricsProcessor(
   jedis: JedisCluster,
   leaderStatus: LeaderStatus,
   rules: CloudWatchRules,
-  publishRouter: PublishRouter
+  publishRouter: PublishRouter,
+  debugger: CloudWatchDebugger
 )(override implicit val system: ActorSystem)
-    extends CloudWatchMetricsProcessor(config, registry, rules, tagger, publishRouter) {
+    extends CloudWatchMetricsProcessor(config, registry, rules, tagger, publishRouter, debugger) {
 
   private implicit val executionContext = system.dispatchers.lookup("redis-io-dispatcher")
 

atlas-cloudwatch/src/main/scala/com/netflix/atlas/spring/CloudWatchConfiguration.scala

@@ -18,11 +18,12 @@ package com.netflix.atlas.spring
 import akka.actor.ActorSystem
 import com.netflix.atlas.akka.AkkaHttpClient
 import com.netflix.atlas.akka.DefaultAkkaHttpClient
+import com.netflix.atlas.cloudwatch.CloudWatchDebugger
 import com.netflix.atlas.cloudwatch.CloudWatchMetricsProcessor
 import com.netflix.atlas.cloudwatch.CloudWatchRules
-import com.netflix.atlas.cloudwatch.NetflixTagger
 import com.netflix.atlas.cloudwatch.PublishRouter
 import com.netflix.atlas.cloudwatch.RedisClusterCloudWatchMetricsProcessor
+import com.netflix.atlas.cloudwatch.NetflixTagger
 import com.netflix.atlas.cloudwatch.Tagger
 import com.netflix.iep.leader.api.LeaderStatus
 import com.netflix.spectator.api.Registry
@@ -45,6 +46,10 @@ import java.util.Optional
 @Configuration
 class CloudWatchConfiguration extends StrictLogging {
 
+  // REPLACE With the dyn config jar include
+  @Bean
+  def getConfig: Config = ConfigFactory.load()
+
   @Bean
   def cloudWatchRules(config: Config): CloudWatchRules = new CloudWatchRules(config)
 
@@ -74,6 +79,7 @@ class CloudWatchConfiguration extends StrictLogging {
     leaderStatus: LeaderStatus,
     rules: CloudWatchRules,
     publishRouter: PublishRouter,
+    debugger: CloudWatchDebugger,
     system: ActorSystem
   ): CloudWatchMetricsProcessor = {
     val r = registry.orElseGet(() => globalRegistry())
@@ -84,7 +90,8 @@ class CloudWatchConfiguration extends StrictLogging {
       jedis,
       leaderStatus,
       rules,
-      publishRouter
+      publishRouter,
+      debugger
     )(system)
   }
 
@@ -126,4 +133,13 @@ class CloudWatchConfiguration extends StrictLogging {
       poolConfig
     )
   }
+
+  @Bean
+  def debugger(
+    config: Config,
+    registry: Optional[Registry]
+  ): CloudWatchDebugger = {
+    val r = registry.orElseGet(() => globalRegistry())
+    new CloudWatchDebugger(config, r)
+  }
 }

atlas-cloudwatch/src/test/resources/application.conf

@@ -17,6 +17,7 @@ atlas {
     testMode = false
     categories = ${?atlas.cloudwatch.categories} [
       "ut1",
+      "ut5",
       "ut-asg",
       "ut-timeout",
       "ut-offset",

atlas-cloudwatch/src/test/resources/test-rules.conf

@@ -34,6 +34,23 @@ atlas {
       ]
     }
 
+    ut5 = {
+      namespace = "AWS/UT1"
+      period = 5m
+
+      dimensions = [
+        "MyTag"
+      ]
+
+      metrics = [
+        {
+          name = "5Min"
+          alias = "aws.utm.5min"
+          conversion = "sum,rate"
+        }
+      ]
+    }
+
     ut-asg = {
       namespace = "AWS/UT1"
       period = 1m