atlas-eval: expose data rate per datasource via diagnostic messages #1192

jfz · 2020-07-30T22:27:10Z

Expose raw data rate (from LWC) and final data out put rate for each DataSource.
Raw data rate is collected at TimedGroup for each DataExpr and carry it to down stream, and summarized in FinalExprEval and emit.

jfz · 2020-07-30T22:28:03Z

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

        // Generate the time series and diagnostic output
        val output = recipients.flatMap {
-          case (expr, ids) =>
+          case (styleExpr, ids) =>


Renamed for easier to follow, I was a little confused when first saw expr.expr.dataExprs

jfz · 2020-07-30T22:31:40Z

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

@@ -101,9 +103,11 @@ object AggrDatapoint {
      extends Aggregator {

    private var value = init.value
+    rawDatapointCounter += 1


Need increase here because the constructor take 1 Datapoint

brharrington

Did an initial pass, will take another look tomorrow. Once we have ironed out the message format, can we update the README for the eval project to include details on the messages?

brharrington · 2020-07-31T02:07:33Z

atlas-akka/src/main/scala/com/netflix/atlas/akka/DiagnosticMessage.scala

@@ -43,6 +44,14 @@ object DiagnosticMessage {
    error(s"${t.getClass.getSimpleName}: ${t.getMessage}")
  }

+  def rate(dataType: String, count: Long, timestamp: Long, step: Long): DiagnosticMessage = {


This seems pretty specific to the eval library. Could we just use an info message with the right information encoded?

brharrington · 2020-07-31T02:35:15Z

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

          case (k, vs) =>
            val t = Instant.ofEpochMilli(timestamp)
-            k -> DiagnosticMessage.info(s"$t: ${vs.length} input datapoints for [$k]")
+            k -> DiagnosticMessage.info(s"$t: ${vs.values.length} input datapoints for [$k]")


Not sure this is still needed. I think we can refactor it to be consistent with the new messages or remove if it is redundant.

brharrington · 2020-07-31T02:39:41Z

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

+        val rawDataRateMessages = dataSourceIdToDataExprs.map(kv =>
+          new MessageEnvelope(
+            kv._1,
+            DiagnosticMessage.rate("raw", getNumRawDatapoints(kv._2, group), group.timestamp, step)


Would it make sense to output for each data expr? If it is a complex expression with many parts, then that would allow them to see which sub-expressions are contributing the most to the cost.

brharrington · 2020-07-31T02:46:01Z

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

+
+              // Collect final output data count
+              ids.foreach(id =>
+                finalDataRateMap.get(id) match {


Using RefIntHashMap would be more efficient here. For our current uses it is unlikely we would overflow the int, though we could create a long specialization if needed.

brharrington · 2020-07-31T02:52:42Z

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

+
+        // Final output data rate for each DataSource
+        val finalDataRateMessages = finalDataRateMap.map(kv => {
+          new MessageEnvelope(kv._1, DiagnosticMessage.rate("final", kv._2, group.timestamp, step))


This could already be computed by the user with counting the datapoints. I think we are missing the intermediate size that the user couldn't compute. So we would have:

Input size: number of data points coming in as raw input.

Intermediate size: number of aggregate data points that are used in the eval step. This is not currently reported by this change.

Output size: number of data points for the final result.

The first two could potentially be reported per data expression. Since we expect the user to pay attention to these data points and process them, we might want to make them a standard JSON payload rather than comma separated string so they are easy to consume.

"intermediate size" looks like internal implementation detail, I wonder how is "intermediate size" useful to users, is it actionable without changing raw input size?

As an example consider this expression:

nf.app,www,:eq, name,foo,:eq,:and, (,id,),:by, nf.app,www,:eq, (,id,),:by, 1,:gt, :mul

This is similar to a pattern we saw recently. In this case the second clause that only restricts by app had a high input size and intermediate size. A slight variation would be:

nf.app,www,:eq, name,foo,:eq,:and, (,id,),:by, nf.app,www,:eq, name,bar,:eq,:and, (,id,),:by, 1,:gt, :mul

Lets suppose that the input size is the same for both foo and bar. However, bar has many distinct values for id, but foo only has a few. The output size will only indicate the set that are in common between both. The difference in potential processing cost and memory usage, something we would want to support limits on, would be based on the intermediate size for the group by. Also if there is a concern about the cost, this is the difference that would tell you which clause is leading to higher cost as the input size will be the same.

jfz · 2020-08-01T00:57:20Z

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

@@ -186,19 +182,6 @@ class FinalExprEvalSuite extends AnyFunSuite {
        val ts = env.getMessage.asInstanceOf[TimeSeriesMessage]
        checkValue(ts, expectedValue)
    }


TBD - Will add some validation for the new data rate messages

Looks like this is still pending

added validations

brharrington

Mostly looks good. Would be nice to improve the test coverage for the new messages and update the readme to give an overview.

brharrington · 2020-08-04T02:08:03Z

atlas-akka/src/main/scala/com/netflix/atlas/akka/DiagnosticMessage.scala

@@ -26,6 +26,7 @@ object DiagnosticMessage {
  final val Warning: String = "warn"
  final val Error: String = "error"
  final val Close: String = "close"
+  final val Rate: String = "rate"


Doesn't look like this is used anymore, it can be removed.

brharrington · 2020-08-04T02:18:50Z

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

+    dataExpr: DataExpr,
+    amount: Int
+  ): Unit = {
+    counts.get(id) match {


This could be simplified to:

counts.getOrElseUpdate(id, new RefIntHashMap[DataExpr]).increment(dataExpr, amount)

brharrington · 2020-08-04T02:33:09Z

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

+
+import scala.collection.mutable
+
+class EvalDataRateCollector(timestamp: Long, step: Long) {


It would be nice to have some tests for this class.

Should be covered by FinalExprEvalSuite

brharrington · 2020-08-04T02:50:55Z

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

          .toList

+        // Compute the map from DataSource id to DataExprs
+        dataSourceIdToDataExprs = mutable.Map.empty


This could be refactored to keep it immutable since it is never modified later, something like:

dataSourceIdToDataExprs = recipients .flatMap { case (styleExpr, ids) => ids.map(_ -> styleExpr.expr.dataExprs.toSet) } .foldLeft(Map.empty[String, Set[DataExpr]]) { case (acc, (id, dataExprs)) => acc + acc.get(id).fold(id -> dataExprs) { vs => id -> (dataExprs ++ vs) } }

sounds good, made the change with one difference: made the init value of foldLeft mutable map to avoid creating new Map on every update

brharrington · 2020-08-04T02:53:43Z

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

+          case (id, dataExprSet) => {
+            dataExprSet.foreach(dataExpr => {
+              group.dataExprValues.get(dataExpr).foreach { info =>
+                {


This additional level of braces is unnecessary.

brharrington · 2020-08-04T02:57:46Z

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

@@ -186,19 +182,6 @@ class FinalExprEvalSuite extends AnyFunSuite {
        val ts = env.getMessage.asInstanceOf[TimeSeriesMessage]
        checkValue(ts, expectedValue)
    }


Looks like this is still pending

jfz commented Jul 30, 2020

View reviewed changes

jfz changed the title ~~atlas-eval: expose date rate~~ atlas-eval: expose data rate per datasource via diagnostic message Jul 30, 2020

jfz marked this pull request as ready for review July 30, 2020 23:02

jfz requested a review from brharrington July 30, 2020 23:02

jfz changed the title ~~atlas-eval: expose data rate per datasource via diagnostic message~~ atlas-eval: expose data rate per datasource via diagnostic messages Jul 30, 2020

rename root project to atlas to be more IDE friendly

0c4245c

jfz force-pushed the lwcRate branch from 82b5f42 to 0c4245c Compare July 30, 2020 23:14

Merge remote-tracking branch 'up/master' into lwcRate

935c40e

brharrington reviewed Jul 31, 2020

View reviewed changes

fix unit test

e53c4c9

jfz force-pushed the lwcRate branch from 5015a89 to e74dfdf Compare August 1, 2020 00:54

jfz commented Aug 1, 2020

View reviewed changes

jfz force-pushed the lwcRate branch 4 times, most recently from 627c8c3 to cef982c Compare August 3, 2020 06:28

jfz requested a review from brharrington August 3, 2020 19:00

collect data rate for raw input, intermediate input and final output

daa0c51

jfz force-pushed the lwcRate branch from cef982c to daa0c51 Compare August 4, 2020 02:43

brharrington reviewed Aug 4, 2020

View reviewed changes

jfz force-pushed the lwcRate branch 2 times, most recently from 37f93d2 to 5118510 Compare August 4, 2020 20:22

code review fixes, adding unit test, document data rate messages

51dbdee

jfz force-pushed the lwcRate branch from 5118510 to 51dbdee Compare August 4, 2020 22:56

brharrington approved these changes Aug 5, 2020

View reviewed changes

brharrington merged commit 324434a into Netflix:master Aug 5, 2020

brharrington added this to the 1.7.0 milestone Aug 5, 2020

jfz deleted the lwcRate branch August 5, 2020 16:28

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

atlas-eval: expose data rate per datasource via diagnostic messages #1192

atlas-eval: expose data rate per datasource via diagnostic messages #1192

jfz commented Jul 30, 2020 •

edited

Loading

jfz Jul 30, 2020 •

edited

Loading

jfz Jul 30, 2020

brharrington left a comment

brharrington Jul 31, 2020

brharrington Jul 31, 2020

brharrington Jul 31, 2020

brharrington Jul 31, 2020

brharrington Jul 31, 2020

jfz Jul 31, 2020

brharrington Jul 31, 2020

jfz Aug 1, 2020

brharrington Aug 4, 2020

jfz Aug 4, 2020

brharrington left a comment

brharrington Aug 4, 2020

brharrington Aug 4, 2020

brharrington Aug 4, 2020

jfz Aug 4, 2020

brharrington Aug 4, 2020

jfz Aug 4, 2020

brharrington Aug 4, 2020

brharrington Aug 4, 2020


		import scala.collection.mutable

		class EvalDataRateCollector(timestamp: Long, step: Long) {

atlas-eval: expose data rate per datasource via diagnostic messages #1192

atlas-eval: expose data rate per datasource via diagnostic messages #1192

Conversation

jfz commented Jul 30, 2020 • edited Loading

jfz Jul 30, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brharrington left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brharrington left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jfz commented Jul 30, 2020 •

edited

Loading

jfz Jul 30, 2020 •

edited

Loading