lwc-events: guard against exception from extractValue (#1745)

Ensure that if the extractValue implementation for an event
throws an exception, it will not impact processing other
events. If the extract fails, then it will treat the value
as `null`.

atlas-lwc-events/src/main/scala/com/netflix/atlas/lwc/events/AbstractLwcEventClient.scala

@@ -67,7 +67,7 @@ abstract class AbstractLwcEventClient(clock: Clock) extends LwcEventClient {
             expr.dataExpr,
             clock,
             sub.step,
-            Some(event => expr.projectionKeys.map(event.extractValue)),
+            Some(event => expr.projectionKeys.map(event.extractValueSafe)),
             submit
           )
           EventHandler(

atlas-lwc-events/src/main/scala/com/netflix/atlas/lwc/events/DatapointConverter.scala

@@ -91,8 +91,8 @@ private[events] object DatapointConverter {
       case Some(k) =>
         tags.get("statistic") match {
           case Some("count")          => _ => 1.0
-          case Some("totalOfSquares") => event => squared(event.extractValue(k), event.value)
-          case _                      => event => toDouble(event.extractValue(k), event.value)
+          case Some("totalOfSquares") => event => squared(event.extractValueSafe(k), event.value)
+          case _                      => event => toDouble(event.extractValueSafe(k), event.value)
         }
       case None =>
         event => toDouble(event.value, 1.0)
@@ -339,7 +339,7 @@ private[events] object DatapointConverter {
 
     private def getRawValue(event: LwcEvent): Any = {
       params.tags.get("value") match {
-        case Some(k) => event.extractValue(k)
+        case Some(k) => event.extractValueSafe(k)
         case None    => event.value
       }
     }

atlas-lwc-events/src/main/scala/com/netflix/atlas/lwc/events/DatapointEvent.scala

@@ -59,7 +59,7 @@ case class DatapointEvent(
   @scala.annotation.tailrec
   private def encodeColumns(columns: List[String], gen: JsonGenerator): Unit = {
     if (columns.nonEmpty) {
-      Json.encode(gen, extractValue(columns.head))
+      Json.encode(gen, extractValueSafe(columns.head))
       encodeColumns(columns.tail, gen)
     }
   }

atlas-lwc-events/src/main/scala/com/netflix/atlas/lwc/events/LwcEvent.scala

@@ -17,6 +17,8 @@ package com.netflix.atlas.lwc.events
 
 import com.fasterxml.jackson.core.JsonGenerator
 import com.netflix.atlas.json.Json
+import com.netflix.spectator.api.Spectator
+import org.slf4j.LoggerFactory
 
 import java.io.StringWriter
 import scala.util.Using
@@ -48,7 +50,7 @@ trait LwcEvent {
     * to ensure the two are consistent.
     */
   def tagValue(key: String): String = {
-    extractValue(key) match {
+    extractValueSafe(key) match {
       case v: String  => v
       case e: Enum[_] => e.name()
       case _          => null
@@ -61,6 +63,25 @@ trait LwcEvent {
     */
   def extractValue(key: String): Any
 
+  /**
+    * Internal method for extracting the value that handles exceptions if any. If an exception
+    * is thrown, then the value will be treated as `null`. The underlying exception will be
+    * logged at trace level as it can be quite noisy if it is a common pattern across events.
+    */
+  private[events] final def extractValueSafe(key: String): Any = {
+    try {
+      extractValue(key)
+    } catch {
+      case e: Exception =>
+        LoggerFactory.getLogger(getClass).trace(s"failed to extract value for key: $key", e)
+        Spectator
+          .globalRegistry()
+          .counter("lwc.extractValueFailures", "error", e.getClass.getSimpleName)
+          .increment()
+        null
+    }
+  }
+
   /** Encode the raw event as JSON. */
   def encode(gen: JsonGenerator): Unit = {
     Json.encode(gen, rawEvent)
@@ -78,7 +99,7 @@ trait LwcEvent {
   def encodeAsRow(columns: List[String], gen: JsonGenerator): Unit = {
     gen.writeStartArray()
     columns.foreach { key =>
-      Json.encode(gen, extractValue(key))
+      Json.encode(gen, extractValueSafe(key))
     }
     gen.writeEndArray()
   }

atlas-lwc-events/src/test/scala/com/netflix/atlas/lwc/events/LwcEventDurationSuite.scala

@@ -49,13 +49,13 @@ class LwcEventDurationSuite extends FunSuite {
   }
 
   test("extractValue: exists") {
-    assertEquals(sampleLwcEvent.extractValue("app"), "www")
-    assertEquals(sampleLwcEvent.extractValue("node"), "i-123")
-    assertEquals(sampleLwcEvent.extractValue("duration"), Duration.ofMillis(42))
+    assertEquals(sampleLwcEvent.extractValueSafe("app"), "www")
+    assertEquals(sampleLwcEvent.extractValueSafe("node"), "i-123")
+    assertEquals(sampleLwcEvent.extractValueSafe("duration"), Duration.ofMillis(42))
   }
 
   test("extractValue: missing") {
-    assertEquals(sampleLwcEvent.extractValue("foo"), null)
+    assertEquals(sampleLwcEvent.extractValueSafe("foo"), null)
   }
 
   test("defautl value") {

atlas-lwc-events/src/test/scala/com/netflix/atlas/lwc/events/LwcEventSuite.scala

@@ -48,13 +48,17 @@ class LwcEventSuite extends FunSuite {
   }
 
   test("extractValue: exists") {
-    assertEquals(sampleLwcEvent.extractValue("app"), "www")
-    assertEquals(sampleLwcEvent.extractValue("node"), "i-123")
-    assertEquals(sampleLwcEvent.extractValue("duration"), 42L)
+    assertEquals(sampleLwcEvent.extractValueSafe("app"), "www")
+    assertEquals(sampleLwcEvent.extractValueSafe("node"), "i-123")
+    assertEquals(sampleLwcEvent.extractValueSafe("duration"), 42L)
   }
 
   test("extractValue: missing") {
-    assertEquals(sampleLwcEvent.extractValue("foo"), null)
+    assertEquals(sampleLwcEvent.extractValueSafe("foo"), null)
+  }
+
+  test("extractValue: exception thrown") {
+    assertEquals(sampleLwcEvent.extractValueSafe("npe"), null)
   }
 
   test("toJson: raw event") {
@@ -87,6 +91,7 @@ object LwcEventSuite {
       case "tags"     => span.tags
       case "duration" => span.duration
       case "level"    => Logger.Level.TRACE
+      case "npe"      => throw new NullPointerException()
       case k          => span.tags.getOrElse(k, null)
     }
   }