eval: require precise scoping for hi-res streams (Netflix#1624)

For hi-res streams, require more precise scoping that allows
us to more efficiently match the data and run it only where
needed. This would ideally be applied everywhere, but for
backwards compatibility the 1m step is opted out for now.

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

@@ -27,7 +27,6 @@ import com.netflix.atlas.eval.stream.Evaluator.DataSources
 import com.netflix.atlas.eval.util.HostRewriter
 import com.typesafe.config.Config
 
-import java.time.Duration
 import scala.util.Success
 
 private[stream] class ExprInterpreter(config: Config) {
@@ -36,17 +35,9 @@ private[stream] class ExprInterpreter(config: Config) {
 
   private val hostRewriter = new HostRewriter(config.getConfig("atlas.eval.host-rewrite"))
 
-  private val maxStep = config.getDuration("atlas.eval.stream.limits.max-step")
-
   def eval(uri: Uri): GraphConfig = {
     val graphCfg = grapher.toGraphConfig(uri)
 
-    // Check step size is within bounds
-    if (graphCfg.stepSize > maxStep.toMillis) {
-      val step = Duration.ofMillis(graphCfg.stepSize)
-      throw new IllegalArgumentException(s"max allowed step size exceeded ($step > $maxStep)")
-    }
-
     // Check that data expressions are supported. The streaming path doesn't support
     // time shifts, filters, and integral. The filters and integral are excluded because
     // they can be confusing as the time window for evaluation is not bounded.

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

@@ -84,6 +84,9 @@ private[stream] class StreamContext(
   def maxIntermediateDatapointsPerExpression: Int =
     config.getInt("limits.max-intermediate-datapoints")
 
+  // Maximum allowed step size for a stream
+  private val maxStep = config.getDuration("limits.max-step")
+
   val interpreter = new ExprInterpreter(rootConfig)
 
   def findBackendForUri(uri: Uri): Backend = {
@@ -150,11 +153,25 @@ private[stream] class StreamContext(
     Try {
       val uri = Uri(ds.uri)
 
+      // Check step size is within bounds
+      if (ds.step.toMillis > maxStep.toMillis) {
+        throw new IllegalArgumentException(
+          s"max allowed step size exceeded (${ds.step} > $maxStep)"
+        )
+      }
+
       // Check that expression is parseable and perform basic static analysis of DataExprs to
       // weed out expensive queries up front
       val results = interpreter.eval(uri).exprs
       results.foreach(_.expr.dataExprs.foreach(validateDataExpr))
 
+      // For hi-res streams, require more precise scoping that allows us to more efficiently
+      // match the data and run it only where needed. This would ideally be applied everywhere,
+      // but for backwards compatiblity the 1m step is opted out for now.
+      if (ds.step.toMillis < 60_000) {
+        results.foreach(_.expr.dataExprs.foreach(expr => restrictsNameAndApp(expr.query)))
+      }
+
       // Check that there is a backend available for it
       findBackendForUri(uri)
 
@@ -178,6 +195,27 @@ private[stream] class StreamContext(
     }
   }
 
+  private def restrictsNameAndApp(query: Query): Unit = {
+    val dnf = Query.dnfList(query)
+    if (!dnf.forall(isRestricted)) {
+      throw new IllegalArgumentException(
+        s"rejected expensive query [$query], hi-res streams " +
+          "must restrict name and nf.app with :eq or :in"
+      )
+    }
+  }
+
+  private def isRestricted(query: Query): Boolean = {
+    isRestricted(query, Set("nf.app", "nf.cluster", "nf.asg")) && isRestricted(query, Set("name"))
+  }
+
+  private def isRestricted(query: Query, keys: Set[String]): Boolean = query match {
+    case Query.And(q1, q2) => isRestricted(q1, keys) || isRestricted(q2, keys)
+    case Query.Equal(k, _) => keys.contains(k)
+    case Query.In(k, _)    => keys.contains(k)
+    case _                 => false
+  }
+
   /**
     * Emit an error to the sources where the number of input
     * or intermediate datapoints exceed for an expression.