From ded42ce71f17a8c554958b10707dc47dc88af913 Mon Sep 17 00:00:00 2001
From: Mohit Palriwal <mpalriwal@netflix.com>
Date: Tue, 29 Oct 2024 12:03:12 -0700
Subject: [PATCH] Fix tagger for nf.asg to cluster/app conversion

---
 .../atlas/cloudwatch/NetflixTagger.scala      | 20 ++++++-----
 .../src/test/resources/application.conf       |  1 +
 .../src/test/resources/test-rules.conf        | 16 +++++++++
 .../atlas/cloudwatch/CWMPProcessSuite.scala   | 33 +++++++++++++++++++
 4 files changed, 61 insertions(+), 9 deletions(-)

diff --git a/atlas-cloudwatch/src/main/scala/com/netflix/atlas/cloudwatch/NetflixTagger.scala b/atlas-cloudwatch/src/main/scala/com/netflix/atlas/cloudwatch/NetflixTagger.scala
index c1c0d885..9e661035 100644
--- a/atlas-cloudwatch/src/main/scala/com/netflix/atlas/cloudwatch/NetflixTagger.scala
+++ b/atlas-cloudwatch/src/main/scala/com/netflix/atlas/cloudwatch/NetflixTagger.scala
@@ -35,23 +35,25 @@ class NetflixTagger(config: Config) extends DefaultTagger(config) {
 
   override def apply(dimensions: List[Dimension]): Map[String, String] = {
     val baseTags = super.apply(dimensions)
-    val extractedTags = keys.flatMap { k =>
+    val resultTags = scala.collection.mutable.Map[String, String]()
+
+    keys.foreach { k =>
       baseTags.get(k) match {
         case Some(v) =>
           val name = Names.parseName(v)
-          List(
+          resultTags ++= List(
             opt("nf.app", name.getApp),
             opt("nf.cluster", name.getCluster),
             opt("nf.stack", name.getStack)
-          )
+          ).flatten
         case None =>
-          // If baseTags.get(k) is empty, add default values
-          List(
-            Some("nf.app"     -> "cloudwatch"),
-            Some("nf.cluster" -> "cloudwatch")
-          )
+          // Only add default values if the keys are not already set
+          if (!resultTags.contains("nf.app")) resultTags("nf.app") = "cloudwatch"
+          if (!resultTags.contains("nf.cluster")) resultTags("nf.cluster") = "cloudwatch"
       }
     }
-    extractedTags.flatten.toMap ++ baseTags
+
+    // Merge resultTags with baseTags, giving priority to resultTags
+    resultTags.toMap ++ baseTags
   }
 }
diff --git a/atlas-cloudwatch/src/test/resources/application.conf b/atlas-cloudwatch/src/test/resources/application.conf
index cf781013..0d9a10c9 100644
--- a/atlas-cloudwatch/src/test/resources/application.conf
+++ b/atlas-cloudwatch/src/test/resources/application.conf
@@ -18,6 +18,7 @@ atlas {
     categories = ${?atlas.cloudwatch.categories} [
       "ut1",
       "ut5",
+      "ut-ec2",
       "ut-asg",
       "ut-timeout",
       "ut-offset",
diff --git a/atlas-cloudwatch/src/test/resources/test-rules.conf b/atlas-cloudwatch/src/test/resources/test-rules.conf
index 098460c3..a0d694aa 100644
--- a/atlas-cloudwatch/src/test/resources/test-rules.conf
+++ b/atlas-cloudwatch/src/test/resources/test-rules.conf
@@ -61,6 +61,22 @@ atlas {
       ]
     }
 
+    ut-ec2 = {
+      namespace = "AWS/UT1"
+      period = 5m
+
+      dimensions = [
+        "AutoScalingGroupName"
+      ]
+
+      metrics = [
+        {
+          name = "EBSReadBytes"
+          alias = "aws.ec2.ebs.ebsReadBytes"
+          conversion = "sum,rate"
+        }
+      ]
+    }
     ut-asg = {
       namespace = "AWS/UT1"
       period = 1m
diff --git a/atlas-cloudwatch/src/test/scala/com/netflix/atlas/cloudwatch/CWMPProcessSuite.scala b/atlas-cloudwatch/src/test/scala/com/netflix/atlas/cloudwatch/CWMPProcessSuite.scala
index ce0e7746..f429731e 100644
--- a/atlas-cloudwatch/src/test/scala/com/netflix/atlas/cloudwatch/CWMPProcessSuite.scala
+++ b/atlas-cloudwatch/src/test/scala/com/netflix/atlas/cloudwatch/CWMPProcessSuite.scala
@@ -327,6 +327,39 @@ class CWMPProcessSuite extends BaseCloudWatchMetricsProcessorSuite {
     assertCounters(1)
   }
 
+  test("processDatapoints ec2 bytes") {
+    val dp = makeFirehoseMetric(
+      "AWS/UT1",
+      "EBSReadBytes",
+      List(
+        Dimension.builder().name("AutoScalingGroupName").value("someApp-someStack-v001").build()
+      ),
+      Array(1, 1, 1, 1),
+      "Bytes",
+      ts(-5.minute)
+    )
+    processor.processDatapoints(List(dp), ts(-5.minute))
+
+    assertPublished(
+      List(
+        com.netflix.atlas.core.model.Datapoint(
+          Map(
+            "name"         -> "aws.ec2.ebs.ebsReadBytes",
+            "nf.region"    -> "us-west-2",
+            "atlas.dstype" -> "rate",
+            "nf.stack"     -> "someStack",
+            "nf.app"       -> "someApp",
+            "nf.asg"       -> "someApp-someStack-v001",
+            "nf.cluster"   -> "someApp-someStack"
+          ),
+          ts,
+          0.0033333333333333335
+        )
+      )
+    )
+    assertCounters(1)
+  }
+
   test("processDatapoints matched percent") {
     processor.processDatapoints(
       List(