eval: add flag to check for expensive queries (#1645)

For some backends, such as event queries, the expensive
query check may not make sense as there may not be a name
field. Allow it to be disabled for specific backends.

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

@@ -65,14 +65,23 @@ private[stream] class StreamContext(
   private val backends = {
     import scala.jdk.CollectionConverters.*
     config.getConfigList("backends").asScala.toList.map { cfg =>
+      val checkForExpensiveQueries = isNotFalse(cfg, "check-for-expensive-queries")
       EurekaBackend(
         cfg.getString("host"),
         cfg.getString("eureka-uri"),
-        cfg.getString("instance-uri")
+        cfg.getString("instance-uri"),
+        checkForExpensiveQueries
       )
     }
   }
 
+  private def isNotFalse(cfg: Config, k: String): Boolean = {
+    if (cfg.hasPath(k))
+      cfg.getBoolean(k)
+    else
+      true
+  }
+
   private val ignoredTagKeys = {
     import scala.jdk.CollectionConverters.*
     config.getStringList("ignored-tag-keys").asScala.toSet
@@ -163,18 +172,18 @@ private[stream] class StreamContext(
         )
       }
 
+      // Check that there is a backend available for it
+      val backend = findBackendForUri(uri)
+
       // Check that expression is parseable and perform basic static analysis of DataExprs to
       // weed out expensive queries up front
       val (exprType, exprs) = interpreter.parseQuery(uri)
-      if (exprType == ExprType.TIME_SERIES) {
+      if (backend.checkForExpensiveQueries && exprType == ExprType.TIME_SERIES) {
         exprs.foreach {
           case e: StyleExpr => validateStyleExpr(e, ds)
         }
       }
 
-      // Check that there is a backend available for it
-      findBackendForUri(uri)
-
       // Everything is ok
       ds
     }
@@ -281,14 +290,19 @@ private[stream] class StreamContext(
 private[stream] object StreamContext {
 
   sealed trait Backend {
+
     def source: Source[ByteString, Future[IOResult]]
+
+    def checkForExpensiveQueries: Boolean
   }
 
   case class FileBackend(file: Path) extends Backend {
 
     def source: Source[ByteString, Future[IOResult]] = {
       FileIO.fromPath(file).via(EvaluationFlows.sseFraming)
     }
+
+    def checkForExpensiveQueries: Boolean = true
   }
 
   case class ResourceBackend(resource: String) extends Backend {
@@ -298,16 +312,25 @@ private[stream] object StreamContext {
         .fromInputStream(() => Streams.resource(resource))
         .via(EvaluationFlows.sseFraming)
     }
+
+    def checkForExpensiveQueries: Boolean = true
   }
 
   case class SyntheticBackend(interpreter: ExprInterpreter, uri: Uri) extends Backend {
 
     def source: Source[ByteString, Future[IOResult]] = {
       SyntheticDataSource(interpreter, uri)
     }
+
+    def checkForExpensiveQueries: Boolean = true
   }
 
-  case class EurekaBackend(host: String, eurekaUri: String, instanceUri: String) extends Backend {
+  case class EurekaBackend(
+    host: String,
+    eurekaUri: String,
+    instanceUri: String,
+    checkForExpensiveQueries: Boolean = true
+  ) extends Backend {
 
     def source: Source[ByteString, Future[IOResult]] = {
       throw new UnsupportedOperationException("only supported for file and classpath URIs")

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

@@ -312,7 +312,7 @@ class EvaluatorSuite extends FunSuite {
   }
 
   test("create processor, missing q parameter") {
-    val uri = "http://test/api/v1/graph"
+    val uri = "synthetic://test/api/v1/graph"
     val ds1 = Evaluator.DataSources.of(ds("one", uri))
     val msg =
       "IllegalArgumentException: missing required parameter 'q'"
@@ -321,7 +321,7 @@ class EvaluatorSuite extends FunSuite {
 
   test("create processor, expression uses :offset") {
     val expr = "name,foo,:eq,:sum,PT168H,:offset"
-    val uri = s"http://test/api/v1/graph?q=$expr"
+    val uri = s"synthetic://test/api/v1/graph?q=$expr"
     val msg = s"IllegalArgumentException: :offset not supported for streaming evaluation [[$expr]]"
     val ds1 = Evaluator.DataSources.of(ds("one", uri))
     testError(ds1, msg)
@@ -330,7 +330,7 @@ class EvaluatorSuite extends FunSuite {
   test("create processor, expression uses style variant of :offset") {
     val expr = "name,foo,:eq,:sum,(,0h,1w,),:offset"
     val badExpr = "name,foo,:eq,:sum,PT168H,:offset"
-    val uri = s"http://test/api/v1/graph?q=$expr"
+    val uri = s"synthetic://test/api/v1/graph?q=$expr"
     val msg =
       s"IllegalArgumentException: :offset not supported for streaming evaluation [[$badExpr]]"
     val ds1 = Evaluator.DataSources.of(ds("one", uri))
@@ -339,7 +339,7 @@ class EvaluatorSuite extends FunSuite {
 
   test("create processor, reject expensive :in queries") {
     val expr = "name,(,1,2,3,4,5,6,),:in,:sum"
-    val uri = s"http://test/api/v1/graph?q=$expr"
+    val uri = s"synthetic://test/api/v1/graph?q=$expr"
     val msg = s"IllegalArgumentException: rejected expensive query [name,(,1,2,3,4,5,6,),:in], " +
       "narrow the scope to a specific app or name"
     val ds1 = Evaluator.DataSources.of(ds("one", uri))
@@ -348,7 +348,7 @@ class EvaluatorSuite extends FunSuite {
 
   test("create processor, reject expensive :re queries") {
     val expr = "name,foo,:re,:sum"
-    val uri = s"http://test/api/v1/graph?q=$expr"
+    val uri = s"synthetic://test/api/v1/graph?q=$expr"
     val msg = s"IllegalArgumentException: rejected expensive query [name,foo,:re], " +
       "narrow the scope to a specific app or name"
     val ds1 = Evaluator.DataSources.of(ds("one", uri))
@@ -358,7 +358,7 @@ class EvaluatorSuite extends FunSuite {
   test("create processor, ignored tag keys are honored") {
     val query = "name,foo,:re,nf.account,12345,:eq,:and"
     val expr = s"$query,:sum"
-    val uri = s"http://test/api/v1/graph?q=$expr"
+    val uri = s"synthetic://test/api/v1/graph?q=$expr"
     val msg = s"IllegalArgumentException: rejected expensive query [$query], " +
       "narrow the scope to a specific app or name"
     val ds1 = Evaluator.DataSources.of(ds("one", uri))