Merge pull request #174 from ExpediaDotCom/traceId-krastogi

Adding metric to track missing trace result from cassandra

Author: vsen

reader/build/docker/jmxtrans-agent.xml

@@ -5,6 +5,7 @@
         <query objectName="metrics:name=cassandra.read.time" attributes="99thPercentile,50thPercentile,OneMinuteRate" resultAlias="cassandra.read.time.#attribute#"/>
         <query objectName="metrics:name=cassandra.read.failures" attributes="OneMinuteRate" resultAlias="cassandra.read.failures.#attribute#"/>
         <query objectName="metrics:name=search.trace.rejected" attributes="OneMinuteRate" resultAlias="trace-search.rejected.#attribute#"/>
+        <query objectName="metrics:name=cassandra.traces.failures" attributes="OneMinuteRate" resultAlias="cassandra.traces.failures.#attribute#"/>
 
         <!-- reader endpoint metrics -->
         <query objectName="metrics:name=TraceReader.getFieldNames" attributes="50thPercentile,99thPercentile,OneMinuteRate" resultAlias="trace.reader.getFieldNames.#attribute#"/>

reader/src/main/scala/com/expedia/www/haystack/trace/reader/metrics/AppMetricNames.scala

@@ -24,6 +24,7 @@ object AppMetricNames {
 
   val CASSANDRA_READ_TIME = "cassandra.read.time"
   val CASSANDRA_READ_FAILURES = "cassandra.read.failures"
+  val CASSANDRA_TRACES_FAILURE = "cassandra.traces.failures"
 
   val SEARCH_TRACE_REJECTED = "search.trace.rejected"
   val COUNT_BUCKET_REJECTED = "count.bucket.rejected"

reader/src/main/scala/com/expedia/www/haystack/trace/reader/stores/readers/cassandra/CassandraReadRawTracesResultListener.scala

@@ -36,9 +36,11 @@ object CassandraReadRawTracesResultListener {
 }
 
 class CassandraReadRawTracesResultListener(asyncResult: ResultSetFuture,
+                                           promise: Promise[Seq[Trace]],
                                            timer: Timer.Context,
                                            failure: Meter,
-                                           promise: Promise[Seq[Trace]]) extends Runnable {
+                                           tracesFailure: Meter,
+                                           traceIdCount: Int) extends Runnable {
   override def run(): Unit = {
     timer.close()
 
@@ -47,6 +49,7 @@ class CassandraReadRawTracesResultListener(asyncResult: ResultSetFuture,
       .flatMap(tryDeserialize)
     match {
       case Success(traces) =>
+        tracesFailure.mark(traceIdCount - traces.length)
         promise.success(traces)
       case Failure(ex) =>
         if (fatalError(ex)) {
@@ -56,6 +59,7 @@ class CassandraReadRawTracesResultListener(asyncResult: ResultSetFuture,
           LOGGER.error("Failed in reading the record from cassandra", ex)
         }
         failure.mark()
+        tracesFailure.mark(traceIdCount)
         promise.failure(ex)
     }
   }

reader/src/main/scala/com/expedia/www/haystack/trace/reader/stores/readers/cassandra/CassandraTraceReader.scala

@@ -30,6 +30,7 @@ class CassandraTraceReader(cassandra: CassandraSession, config: CassandraConfigu
 
   private val readTimer = metricRegistry.timer(AppMetricNames.CASSANDRA_READ_TIME)
   private val readFailures = metricRegistry.meter(AppMetricNames.CASSANDRA_READ_FAILURES)
+  private val tracesFailures = metricRegistry.meter(AppMetricNames.CASSANDRA_TRACES_FAILURE)
 
   def readTrace(traceId: String): Future[Trace] = {
     val timer = readTimer.time()
@@ -56,7 +57,7 @@ class CassandraTraceReader(cassandra: CassandraSession, config: CassandraConfigu
     try {
       val statement = cassandra.newSelectRawTracesBoundStatement(traceIds)
       val asyncResult = cassandra.executeAsync(statement)
-      asyncResult.addListener(new CassandraReadRawTracesResultListener(asyncResult, timer, readFailures, promise), dispatcher)
+      asyncResult.addListener(new CassandraReadRawTracesResultListener(asyncResult, promise, timer, readFailures, tracesFailures, traceIds.size), dispatcher)
       promise.future
     } catch {
       case ex: Exception =>

reader/src/test/scala/com/expedia/www/haystack/trace/reader/unit/stores/readers/cassandra/CassandraReadRawTracesResultListenerSpec.scala

@@ -24,7 +24,7 @@ import com.expedia.open.tracing.Span
 import com.expedia.open.tracing.api.Trace
 import com.expedia.open.tracing.buffer.SpanBuffer
 import com.expedia.www.haystack.trace.commons.clients.cassandra.CassandraTableSchema
-import com.expedia.www.haystack.trace.reader.stores.readers.cassandra.{CassandraReadRawTracesResultListener, CassandraReadTraceResultListener}
+import com.expedia.www.haystack.trace.reader.stores.readers.cassandra.CassandraReadRawTracesResultListener
 import com.expedia.www.haystack.trace.reader.unit.BaseUnitTestSpec
 import io.grpc.{Status, StatusException}
 import org.easymock.EasyMock
@@ -40,6 +40,7 @@ class CassandraReadRawTracesResultListenerSpec extends BaseUnitTestSpec {
       val resultSet = mock[ResultSet]
       val promise = mock[Promise[Seq[Trace]]]
       val failureMeter = mock[Meter]
+      val tracesFailures = mock[Meter]
       val timer = mock[Timer.Context]
 
       val mockSpanBufferRow_1 = mock[Row]
@@ -56,8 +57,10 @@ class CassandraReadRawTracesResultListenerSpec extends BaseUnitTestSpec {
 
 
       val capturedTraces = EasyMock.newCapture[Seq[Trace]]()
+      val capturedMeter = EasyMock.newCapture[Int]()
       expecting {
         timer.close()
+        tracesFailures.mark(EasyMock.capture(capturedMeter))
         mockReadResult.get().andReturn(resultSet)
         resultSet.all().andReturn(List(mockSpanBufferRow_1, mockSpanBufferRow_2, mockSpanBufferRow_3).asJava)
         mockSpanBufferRow_1.getBytes(CassandraTableSchema.SPANS_COLUMN_NAME).andReturn(ByteBuffer.wrap(spanBuffer_1.toByteArray))
@@ -66,14 +69,16 @@ class CassandraReadRawTracesResultListenerSpec extends BaseUnitTestSpec {
         promise.success(EasyMock.capture(capturedTraces)).andReturn(promise)
       }
 
-      whenExecuting(mockReadResult, promise, failureMeter, timer, resultSet, mockSpanBufferRow_1, mockSpanBufferRow_2, mockSpanBufferRow_3) {
-        val listener = new CassandraReadRawTracesResultListener(mockReadResult, timer, failureMeter, promise)
+      whenExecuting(mockReadResult, promise, tracesFailures, failureMeter, timer, resultSet, mockSpanBufferRow_1, mockSpanBufferRow_2, mockSpanBufferRow_3) {
+        val listener = new CassandraReadRawTracesResultListener(mockReadResult, promise, timer, failureMeter, tracesFailures, 2)
         listener.run()
         val traceIdSpansMap: Map[String, Set[String]] = capturedTraces.getValue.map(capturedTrace =>
           capturedTrace.getTraceId -> capturedTrace.getChildSpansList.asScala.map(_.getSpanId).toSet).toMap
 
         traceIdSpansMap("TRACE_ID1") shouldEqual Set("SPAN_ID_1", "SPAN_ID_2")
         traceIdSpansMap("TRACE_ID3") shouldEqual Set("SPAN_ID_3")
+
+        capturedMeter.getValue shouldEqual 0
       }
     }
 
@@ -82,6 +87,7 @@ class CassandraReadRawTracesResultListenerSpec extends BaseUnitTestSpec {
       val resultSet = mock[ResultSet]
       val promise = mock[Promise[Seq[Trace]]]
       val failureMeter = mock[Meter]
+      val tracesFailures = mock[Meter]
       val timer = mock[Timer.Context]
 
       val mockSpanBufferRow_1 = mock[Row]
@@ -90,42 +96,49 @@ class CassandraReadRawTracesResultListenerSpec extends BaseUnitTestSpec {
       val span_1 = Span.newBuilder().setTraceId("TRACE_ID").setSpanId("SPAN_ID_1")
       val spanBuffer_1 = SpanBuffer.newBuilder().setTraceId("TRACE_ID").addChildSpans(span_1).build()
 
+      val capturedMeter = EasyMock.newCapture[Int]()
       expecting {
         timer.close()
         failureMeter.mark()
+        tracesFailures.mark(EasyMock.capture(capturedMeter))
         mockReadResult.get().andReturn(resultSet)
         resultSet.all().andReturn(List(mockSpanBufferRow_1, mockSpanBufferRow_2).asJava)
         mockSpanBufferRow_1.getBytes(CassandraTableSchema.SPANS_COLUMN_NAME).andReturn(ByteBuffer.wrap(spanBuffer_1.toByteArray))
         mockSpanBufferRow_2.getBytes(CassandraTableSchema.SPANS_COLUMN_NAME).andReturn(ByteBuffer.wrap("illegal bytes".getBytes))
         promise.failure(EasyMock.anyObject()).andReturn(promise)
       }
 
-      whenExecuting(mockReadResult, promise, failureMeter, timer, resultSet, mockSpanBufferRow_1, mockSpanBufferRow_2) {
-        val listener = new CassandraReadRawTracesResultListener(mockReadResult, timer, failureMeter, promise)
+      whenExecuting(mockReadResult, promise, failureMeter, tracesFailures, timer, resultSet, mockSpanBufferRow_1, mockSpanBufferRow_2) {
+        val listener = new CassandraReadRawTracesResultListener(mockReadResult, promise, timer, failureMeter, tracesFailures, 1)
         listener.run()
+        capturedMeter.getValue shouldEqual 1
       }
     }
 
     it("should return an exception for empty traceId") {
       val mockReadResult = mock[ResultSetFuture]
       val resultSet = mock[ResultSet]
-      val promise = mock[Promise[Trace]]
+      val promise = mock[Promise[Seq[Trace]]]
       val failureMeter = mock[Meter]
+      val tracesFailures = mock[Meter]
       val timer = mock[Timer.Context]
 
       val capturedException = EasyMock.newCapture[StatusException]()
+      val capturedMeter = EasyMock.newCapture[Int]()
       expecting {
         timer.close()
         failureMeter.mark()
+        tracesFailures.mark(EasyMock.capture(capturedMeter))
         mockReadResult.get().andReturn(resultSet)
         resultSet.all().andReturn(List[Row]().asJava)
         promise.failure(EasyMock.capture(capturedException)).andReturn(promise)
       }
 
-      whenExecuting(mockReadResult, promise, failureMeter, timer, resultSet) {
-        val listener = new CassandraReadTraceResultListener(mockReadResult, timer, failureMeter, promise)
+      whenExecuting(mockReadResult, promise, failureMeter, tracesFailures, timer, resultSet) {
+        val listener = new CassandraReadRawTracesResultListener(mockReadResult, promise, timer, failureMeter, tracesFailures, 0)
         listener.run()
         capturedException.getValue.getStatus.getCode shouldEqual Status.NOT_FOUND.getCode
+        capturedMeter.getValue shouldEqual 0
       }
     }
   }