add topk/bottomk that aggregates others (Netflix#1248)

Fixes Netflix#1224.

Adds variants of topk/bottomk that in addition to computing
the highest priority values will also return an aggregate
time series that includes all of the other time series that
were not high enough priority. This can be useful to see
what proportion of the overall volume is represented by
the highest priority time series. The operators have the
same signature as `:topk` and `:bottomk`, but the
aggregation to use is specified in the operator name.
There are four aggregates supported:

- min
- max
- sum
- avg

Count was not included as an aggregate because it could be
ambiguous and does not seem useful.

As part of this work there was some refactoring of the
existing aggregation operations to make the count case
more consistent.

Author: brharrington

atlas-core/src/main/scala/com/netflix/atlas/core/model/DataExpr.scala

@@ -126,29 +126,28 @@ object DataExpr {
     }
   }
 
-  sealed trait AggregateFunction extends DataExpr with BinaryOp {
+  sealed trait AggregateFunction extends DataExpr {
 
     def labelString: String
 
     def withConsolidation(f: ConsolidationFunction): AggregateFunction
 
+    def aggregator(start: Long, end: Long): TimeSeries.Aggregator
+
     override def eval(context: EvalContext, data: List[TimeSeries]): ResultSet = {
       val filtered = data.filter(t => query.matches(t.tags))
       val aggr =
         if (filtered.isEmpty) TimeSeries.noData(query, context.step)
         else {
           val tags = commonTags(filtered.head.tags)
-          val t = TimeSeries.aggregate(filtered.iterator, context.start, context.end, this)
+          val aggr = aggregator(context.start, context.end)
+          filtered.foreach(aggr.update)
+          val t = aggr.result()
           TimeSeries(tags, TimeSeries.toLabel(tags), t.data)
         }
       val rs = consolidate(context.step, List(aggr))
       ResultSet(this, rs, context.state)
     }
-
-    // Make sure we get toString locally rather than from BinaryOp
-    override def toString: String = {
-      if (offset.isZero) exprString else s"$exprString,$offset,:offset"
-    }
   }
 
   case class Sum(
@@ -164,13 +163,15 @@ object DataExpr {
 
     override def withOffset(d: Duration): Sum = copy(offset = d)
 
-    def labelString: String = s"sum(${query.labelString})"
+    override def labelString: String = s"sum(${query.labelString})"
 
     override def exprString: String = {
       if (cf == ConsolidationFunction.Avg) s"$query,:sum" else s"$query,:sum,$cf"
     }
 
-    def apply(v1: Double, v2: Double): Double = Math.addNaN(v1, v2)
+    override def aggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.addNaN)
+    }
   }
 
   case class Count(
@@ -179,35 +180,22 @@ object DataExpr {
     offset: Duration = Duration.ZERO
   ) extends AggregateFunction {
 
-    override def eval(context: EvalContext, data: List[TimeSeries]): ResultSet = {
-      val filtered = data.filter(t => query.matches(t.tags)).map { t =>
-        TimeSeries(t.tags, t.label, t.data.mapValues(v => if (v.isNaN) Double.NaN else 1.0))
-      }
-      val aggr =
-        if (filtered.isEmpty) TimeSeries.noData(query, context.step)
-        else {
-          val tags = commonTags(filtered.head.tags)
-          val t = TimeSeries.aggregate(filtered.iterator, context.start, context.end, this)
-          TimeSeries(tags, TimeSeries.toLabel(tags), t.data)
-        }
-      val rs = consolidate(context.step, List(aggr))
-      ResultSet(this, rs, context.state)
-    }
-
     override def withConsolidation(f: ConsolidationFunction): AggregateFunction = f match {
       case v: SumOrAvgCf => copy(cf = v)
       case v             => Consolidation(copy(query = query, offset = offset), v)
     }
 
     override def withOffset(d: Duration): Count = copy(offset = d)
 
-    def labelString: String = s"count(${query.labelString})"
+    override def labelString: String = s"count(${query.labelString})"
 
     override def exprString: String = {
       if (cf == ConsolidationFunction.Avg) s"$query,:count" else s"$query,:count,$cf"
     }
 
-    def apply(v1: Double, v2: Double): Double = Math.addNaN(v1, v2)
+    override def aggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.CountAggregator(start, end)
+    }
   }
 
   case class Min(query: Query, offset: Duration = Duration.ZERO) extends AggregateFunction {
@@ -220,11 +208,13 @@ object DataExpr {
 
     override def withOffset(d: Duration): Min = copy(offset = d)
 
-    def labelString: String = s"min(${query.labelString})"
+    override def labelString: String = s"min(${query.labelString})"
 
     override def exprString: String = s"$query,:min"
 
-    def apply(v1: Double, v2: Double): Double = Math.minNaN(v1, v2)
+    override def aggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.minNaN)
+    }
   }
 
   case class Max(query: Query, offset: Duration = Duration.ZERO) extends AggregateFunction {
@@ -237,11 +227,13 @@ object DataExpr {
 
     override def withOffset(d: Duration): Max = copy(offset = d)
 
-    def labelString: String = s"max(${query.labelString})"
+    override def labelString: String = s"max(${query.labelString})"
 
     override def exprString: String = s"$query,:max"
 
-    def apply(v1: Double, v2: Double): Double = Math.maxNaN(v1, v2)
+    override def aggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.maxNaN)
+    }
   }
 
   case class Consolidation(af: AggregateFunction, cf: ConsolidationFunction)
@@ -263,11 +255,13 @@ object DataExpr {
       Consolidation(af.withOffset(d).asInstanceOf[AggregateFunction], cf)
     }
 
-    def labelString: String = af.labelString
+    override def labelString: String = af.labelString
 
     override def exprString: String = s"$af,$cf"
 
-    def apply(v1: Double, v2: Double): Double = af(v1, v2)
+    override def aggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      af.aggregator(start, end)
+    }
   }
 
   case class GroupBy(af: AggregateFunction, keys: List[String]) extends DataExpr {

atlas-core/src/main/scala/com/netflix/atlas/core/model/FilterExpr.scala

@@ -149,6 +149,11 @@ object FilterExpr {
     /** Comparator that determines the priority order. */
     def comparator: Comparator[TimeSeriesSummary]
 
+    /** Aggregation to use for time series that are not one of the K highest priority. */
+    def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      TimeSeries.NoopAggregator
+    }
+
     /** Grouped expression to select the input from. */
     def expr: TimeSeriesExpr
 
@@ -172,12 +177,23 @@ object FilterExpr {
 
     def eval(context: EvalContext, data: Map[DataExpr, List[TimeSeries]]): ResultSet = {
       val buffer = new BoundedPriorityBuffer[TimeSeriesSummary](k, comparator)
+      val aggregator = othersAggregator(context.start, context.end)
       val rs = expr.eval(context, data)
       rs.data.foreach { t =>
         val v = SummaryStats(t, context.start, context.end).get(stat)
-        buffer.add(TimeSeriesSummary(t, v))
+        val other = buffer.add(TimeSeriesSummary(t, v))
+        if (other != null) {
+          aggregator.update(other.timeSeries)
+        }
+      }
+      val newData = aggregator match {
+        case aggr if aggr.isEmpty =>
+          buffer.toList.map(_.timeSeries)
+        case _ =>
+          val others = aggregator.result()
+          val otherTags = others.tags ++ finalGrouping.map(k => k -> "--others--").toMap
+          others.withTags(otherTags) :: buffer.toList.map(_.timeSeries)
       }
-      val newData = buffer.toList.map(_.timeSeries)
       ResultSet(this, newData, rs.state)
     }
   }
@@ -188,12 +204,96 @@ object FilterExpr {
     override def comparator: Comparator[TimeSeriesSummary] = StatComparator.reversed()
   }
 
+  case class TopKOthersMin(expr: TimeSeriesExpr, stat: String, k: Int) extends PriorityFilterExpr {
+    override def opName: String = "topk-others-min"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator.reversed()
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.minNaN)
+    }
+  }
+
+  case class TopKOthersMax(expr: TimeSeriesExpr, stat: String, k: Int) extends PriorityFilterExpr {
+    override def opName: String = "topk-others-max"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator.reversed()
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.maxNaN)
+    }
+  }
+
+  case class TopKOthersSum(expr: TimeSeriesExpr, stat: String, k: Int) extends PriorityFilterExpr {
+    override def opName: String = "topk-others-sum"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator.reversed()
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.addNaN)
+    }
+  }
+
+  case class TopKOthersAvg(expr: TimeSeriesExpr, stat: String, k: Int) extends PriorityFilterExpr {
+    override def opName: String = "topk-others-avg"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator.reversed()
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.AvgAggregator(start, end)
+    }
+  }
+
   case class BottomK(expr: TimeSeriesExpr, stat: String, k: Int) extends PriorityFilterExpr {
     override def opName: String = "bottomk"
 
     override def comparator: Comparator[TimeSeriesSummary] = StatComparator
   }
 
+  case class BottomKOthersMin(expr: TimeSeriesExpr, stat: String, k: Int)
+      extends PriorityFilterExpr {
+    override def opName: String = "bottomk-others-min"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.minNaN)
+    }
+  }
+
+  case class BottomKOthersMax(expr: TimeSeriesExpr, stat: String, k: Int)
+      extends PriorityFilterExpr {
+    override def opName: String = "bottomk-others-max"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.maxNaN)
+    }
+  }
+
+  case class BottomKOthersSum(expr: TimeSeriesExpr, stat: String, k: Int)
+      extends PriorityFilterExpr {
+    override def opName: String = "bottomk-others-sum"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.SimpleAggregator(start, end, Math.addNaN)
+    }
+  }
+
+  case class BottomKOthersAvg(expr: TimeSeriesExpr, stat: String, k: Int)
+      extends PriorityFilterExpr {
+    override def opName: String = "bottomk-others-avg"
+
+    override def comparator: Comparator[TimeSeriesSummary] = StatComparator
+
+    override def othersAggregator(start: Long, end: Long): TimeSeries.Aggregator = {
+      new TimeSeries.AvgAggregator(start, end)
+    }
+  }
+
   /**
     * Caches the statistic value associated with a time series. Used for priority filters to
     * avoid recomputing the summary statistics on each comparison operation.

atlas-core/src/main/scala/com/netflix/atlas/core/model/FilterVocabulary.scala

@@ -42,8 +42,16 @@ object FilterVocabulary extends Vocabulary {
     Macro("stat-max-mf", List("max", ":stat"), List("42")),
     Macro("stat-avg-mf", List("avg", ":stat"), List("42")),
     // Priority operators: https://github.com/Netflix/atlas/issues/1224
-    BottomK,
-    TopK
+    PriorityK("bottomk", FilterExpr.BottomK.apply),
+    PriorityK("bottomk-others-min", FilterExpr.BottomKOthersMin.apply),
+    PriorityK("bottomk-others-max", FilterExpr.BottomKOthersMax.apply),
+    PriorityK("bottomk-others-sum", FilterExpr.BottomKOthersSum.apply),
+    PriorityK("bottomk-others-avg", FilterExpr.BottomKOthersAvg.apply),
+    PriorityK("topk", FilterExpr.TopK.apply),
+    PriorityK("topk-others-min", FilterExpr.TopKOthersMin.apply),
+    PriorityK("topk-others-max", FilterExpr.TopKOthersMax.apply),
+    PriorityK("topk-others-sum", FilterExpr.TopKOthersSum.apply),
+    PriorityK("topk-others-avg", FilterExpr.TopKOthersAvg.apply)
   )
 
   case object Stat extends SimpleWord {
@@ -228,37 +236,8 @@ object FilterVocabulary extends Vocabulary {
       List("name,sps,:eq,:sum,(,nf.cluster,),:by,:stat-max,30e3,:gt")
   }
 
-  case object TopK extends SimpleWord {
-
-    override def name: String = "topk"
-
-    protected def matcher: PartialFunction[List[Any], Boolean] = {
-      case IntType(_) :: (_: String) :: TimeSeriesType(_) :: _ => true
-      case IntType(_) :: (_: String) :: (_: StyleExpr) :: _    => true
-    }
-
-    protected def executor: PartialFunction[List[Any], List[Any]] = {
-      case IntType(k) :: (s: String) :: TimeSeriesType(t) :: stack =>
-        FilterExpr.TopK(t, s, k) :: stack
-      case IntType(k) :: (s: String) :: (t: StyleExpr) :: stack =>
-        t.copy(expr = FilterExpr.TopK(t.expr, s, k)) :: stack
-    }
-
-    override def signature: String = "TimeSeriesExpr stat:String k:Int -- FilterExpr"
-
-    override def summary: String =
-      """
-        |Limit the output to the `K` time series with the highest values for the specified
-        |summary statistic.
-      """.stripMargin.trim
-
-    override def examples: List[String] =
-      List("name,sps,:eq,:sum,(,nf.cluster,),:by,max,5")
-  }
-
-  case object BottomK extends SimpleWord {
-
-    override def name: String = "bottomk"
+  case class PriorityK(name: String, op: (TimeSeriesExpr, String, Int) => FilterExpr)
+      extends SimpleWord {
 
     protected def matcher: PartialFunction[List[Any], Boolean] = {
       case IntType(_) :: (_: String) :: TimeSeriesType(_) :: _ => true
@@ -267,17 +246,17 @@ object FilterVocabulary extends Vocabulary {
 
     protected def executor: PartialFunction[List[Any], List[Any]] = {
       case IntType(k) :: (s: String) :: TimeSeriesType(t) :: stack =>
-        FilterExpr.BottomK(t, s, k) :: stack
+        op(t, s, k) :: stack
       case IntType(k) :: (s: String) :: (t: StyleExpr) :: stack =>
-        t.copy(expr = FilterExpr.BottomK(t.expr, s, k)) :: stack
+        t.copy(expr = op(t.expr, s, k)) :: stack
     }
 
     override def signature: String = "TimeSeriesExpr stat:String k:Int -- FilterExpr"
 
     override def summary: String =
       """
-        |Limit the output to the `K` time series with the lowest values for the specified
-        |summary statistic.
+        |Limit the output to the `K` time series with the highest priority values for the
+        |specified summary statistic.
       """.stripMargin.trim
 
     override def examples: List[String] =