tag atlas.eval.datapoints counters with upstream host (#1899)

The counters emitted by TimeGrouped and AggregatorSettings previously
had no dimension identifying the upstream backend, so dropped and
buffered counts from all hosts folded together. Thread the groupBy-host
key from createStreamsFlow through createProcessorFlow into TimeGrouped
so each counter now carries a host tag, and default AggregatorSettings
host to "_" for non-host-scoped callers.

Author: brharrington

atlas-eval/src/main/scala/com/netflix/atlas/eval/model/AggrDatapoint.scala

@@ -97,19 +97,30 @@ object AggrDatapoint {
     *     Limit for the number of intermediate datapoints.
     * @param registry
     *     Registry used for reporting metrics related to the aggregation behavior.
+    * @param host
+    *     Upstream host identifier used to tag metrics so dropped counts can be
+    *     attributed to a specific backend. Defaults to `"_"` for non-host-scoped
+    *     callers.
     */
   case class AggregatorSettings(
     maxInputDatapoints: Int,
     maxIntermediateDatapoints: Int,
-    registry: Registry
+    registry: Registry,
+    host: String = "_"
   ) {
 
     /**
       * Counter for tracking number of datapoints that are dropped due to exceeding the
       * configured limits.
       */
     val droppedCounter: Counter =
-      registry.counter("atlas.eval.datapoints", "id", "dropped-datapoints-limit-exceeded")
+      registry.counter(
+        "atlas.eval.datapoints",
+        "id",
+        "dropped-datapoints-limit-exceeded",
+        "host",
+        host
+      )
   }
 
   /**

atlas-eval/src/main/scala/com/netflix/atlas/eval/stream/EvaluatorImpl.scala

@@ -202,7 +202,8 @@ private[stream] abstract class EvaluatorImpl(
   }
 
   protected def createPublisherImpl(uri: Uri): Publisher[JsonSupport] = {
-    val ds = DataSources.of(new DataSource("_", uri.toString()))
+    val dataSource = new DataSource("_", uri.toString())
+    val ds = DataSources.of(dataSource)
 
     val source = uri.scheme match {
       case s if s.startsWith("http") =>
@@ -217,7 +218,7 @@ private[stream] abstract class EvaluatorImpl(
 
     val (logSrc, context) = createStreamContextSource
     source
-      .via(createProcessorFlow(context))
+      .via(createProcessorFlow(context, getHost(dataSource)))
       .via(new OnUpstreamFinish[MessageEnvelope](context.dsLogger.close()))
       .merge(logSrc, eagerComplete = false)
       .map(_.message)
@@ -242,8 +243,13 @@ private[stream] abstract class EvaluatorImpl(
       .via(new FillRemovedKeysWith[String, DataSources](_ => DataSources.empty()))
       .flatMapMerge(Int.MaxValue, dssMap => Source(dssMap.toList))
       .groupBy(Int.MaxValue, _._1, true) // groupBy host
-      .map(_._2) // keep only DataSources
-      .via(createProcessorFlow(context))
+      .flatMapPrefix(1) { prefix =>
+        val host = prefix.headOption.map(_._1).getOrElse("_")
+        Flow[(String, DataSources)]
+          .prepend(Source(prefix.toList))
+          .map(_._2) // keep only DataSources
+          .via(createProcessorFlow(context, host))
+      }
       .mergeSubstreams
       .via(new OnUpstreamFinish[MessageEnvelope](context.dsLogger.close()))
       .merge(logSrc, eagerComplete = false)
@@ -263,7 +269,8 @@ private[stream] abstract class EvaluatorImpl(
   }
 
   private[stream] def createProcessorFlow(
-    context: StreamContext
+    context: StreamContext,
+    host: String
   ): Flow[DataSources, MessageEnvelope, NotUsed] = {
 
     val g = GraphDSL.create() { implicit builder =>
@@ -283,7 +290,7 @@ private[stream] abstract class EvaluatorImpl(
           Source(t.groupByStep)
         }
         .groupBy(Int.MaxValue, _.step, allowClosedSubstreamRecreation = true)
-        .via(new TimeGrouped(context))
+        .via(new TimeGrouped(context, host))
         .mergeSubstreams
         .via(context.monitorFlow("11_GroupedDatapoints"))
 

atlas-eval/src/main/scala/com/netflix/atlas/eval/stream/TimeGrouped.scala

@@ -37,9 +37,13 @@ import com.netflix.atlas.eval.model.TimeGroupsTuple
   *
   * @param context
   *     Shared context for the evaluation stream.
+  * @param host
+  *     Upstream host identifier used to tag metrics so activity can be attributed to a
+  *     specific backend.
   */
 private[stream] class TimeGrouped(
-  context: StreamContext
+  context: StreamContext,
+  host: String
 ) extends GraphStage[FlowShape[DatapointsTuple, TimeGroupsTuple]] {
 
   type AggrMap = java.util.HashMap[DataExpr, AggrDatapoint.Aggregator]
@@ -59,7 +63,8 @@ private[stream] class TimeGrouped(
   private val aggrSettings = AggrDatapoint.AggregatorSettings(
     maxInputDatapointsPerExpression,
     maxIntermediateDatapointsPerExpression,
-    context.registry
+    context.registry,
+    host
   )
 
   private val in = Inlet[DatapointsTuple]("TimeGrouped.in")
@@ -69,9 +74,9 @@ private[stream] class TimeGrouped(
 
   private val metricName = "atlas.eval.datapoints"
   private val registry = context.registry
-  private val droppedOld = registry.counter(metricName, "id", "dropped-old")
-  private val droppedFuture = registry.counter(metricName, "id", "dropped-future")
-  private val buffered = registry.counter(metricName, "id", "buffered")
+  private val droppedOld = registry.counter(metricName, "id", "dropped-old", "host", host)
+  private val droppedFuture = registry.counter(metricName, "id", "dropped-future", "host", host)
+  private val buffered = registry.counter(metricName, "id", "buffered", "host", host)
   private val clock = registry.clock()
 
   override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {

atlas-eval/src/test/scala/com/netflix/atlas/eval/stream/TimeGroupedSuite.scala

@@ -40,6 +40,8 @@ class TimeGroupedSuite extends FunSuite {
 
   private val context = TestContext.createContext(materializer, registry = registry)
 
+  private val host = "test"
+
   private val step = 10
 
   private def result(future: Future[List[TimeGroup]]): List[TimeGroup] = {
@@ -57,7 +59,7 @@ class TimeGroupedSuite extends FunSuite {
   private def run(data: List[AggrDatapoint]): List[TimeGroup] = {
     val future = Source
       .single(DatapointsTuple(data))
-      .via(new TimeGrouped(context))
+      .via(new TimeGrouped(context, host))
       .flatMapConcat(t => Source(t.groups))
       .runFold(List.empty[TimeGroup])((acc, g) => g :: acc)
     result(future)
@@ -121,7 +123,7 @@ class TimeGroupedSuite extends FunSuite {
   }
 
   private def count(id: String): Long = {
-    registry.counter("atlas.eval.datapoints", "id", id).count()
+    registry.counter("atlas.eval.datapoints", "id", id, "host", host).count()
   }
 
   private def counts: (Long, Long) = {