Added ExpensiveEndpoints and test files

zipkin-hadoop/src/main/scala/com/twitter/zipkin/hadoop/ExpensiveEndpoints.scala

@@ -0,0 +1,64 @@
+/*
+* Copyright 2012 Twitter Inc.
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*      http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package com.twitter.zipkin.hadoop
+
+import com.twitter.zipkin.gen.{Constants, SpanServiceName, Annotation}
+import cascading.pipe.joiner.LeftJoin
+import com.twitter.scalding.{Tsv, DefaultDateRangeJob, Job, Args}
+import sources.{PrepTsvSource, Util, PreprocessedSpanSource}
+
+/**
+ * Per service call (i.e. pair of services), finds the average run time (in microseconds) of that service call
+ */
+class ExpensiveEndpoints(args : Args) extends Job(args) with DefaultDateRangeJob {
+
+  val spanInfo = PreprocessedSpanSource()
+    .read
+    .mapTo(0 -> ('id, 'parent_id, 'cService, 'service, 'annotations))
+      { s: SpanServiceName => (s.id, s.parent_id, s.client_service, s.service_name, s.annotations.toList) }
+    .flatMap('annotations -> 'duration) {
+      al : List[Annotation] => {
+        var clientSend : Option[Annotation] = None
+        var clientReceive : Option[Annotation] = None
+        var serverReceive : Option[Annotation] = None
+        var serverSend : Option[Annotation] = None
+        al.foreach( {
+          a : Annotation => {
+            if (a.getHost != null) {
+              if (Constants.CLIENT_SEND.equals(a.value)) clientSend = Some(a)
+              else if (Constants.CLIENT_RECV.equals(a.value)) clientReceive = Some(a)
+              else if (Constants.SERVER_RECV.equals(a.value)) serverReceive = Some(a)
+              else if (Constants.SERVER_SEND.equals(a.value)) serverSend = Some(a)
+            }
+          }
+        })
+        val clientDuration = for (cs <- clientSend; cr <- clientReceive) yield (cr.timestamp - cs.timestamp)
+        val serverDuration = for (sr <- serverReceive; ss <- serverSend) yield (ss.timestamp - sr.timestamp)
+          // to deal with the case where there is no server duration
+        if (clientDuration == None) serverDuration else clientDuration
+      }
+    }
+
+  val idName = PrepTsvSource()
+    .read
+  /* Join with the original on parent ID to get the parent's service name */
+  val spanInfoWithParent = spanInfo
+    .filter('parent_id){ id : Long => id != 0 }
+    .joinWithSmaller('parent_id -> 'id_1, idName)
+    .groupBy('name_1, 'service){ _.average('duration) }
+    .write(Tsv(args("output")))
+}

zipkin-hadoop/src/main/scala/com/twitter/zipkin/hadoop/sources/Util.scala

@@ -121,7 +121,7 @@ object Util {
    *         second element is the span ID of the copy.
    */
   def repeatSpan(span: gen.Span, count: Int, offset : Int, parentOffset : Int): List[(gen.Span, Int)] = {
-    ((0 to count).toSeq map { i: Int => span.deepCopy().setId(i + offset).setParent_id(i + parentOffset) -> (i + offset)}).toList
+    ((0 to count).toSeq map { i: Int => span.deepCopy().setId(i + offset).setParent_id(if (parentOffset == -1) 0 else i + parentOffset) -> (i + offset)}).toList
   }
 
   /**

zipkin-hadoop/src/test/scala/com/twitter/zipkin/hadoop/ExpensiveEndpointsSpec.scala

@@ -0,0 +1,77 @@
+/*
+* Copyright 2012 Twitter Inc.
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*      http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package com.twitter.zipkin.hadoop
+
+import org.specs.Specification
+import com.twitter.zipkin.gen
+import com.twitter.scalding._
+import com.twitter.zipkin.gen
+import gen.AnnotationType
+import scala.collection.JavaConverters._
+import collection.mutable.HashMap
+import com.twitter.scalding.TupleConversions
+import com.twitter.scalding.DateRange
+import com.twitter.scalding.RichDate
+import com.twitter.zipkin.gen.AnnotationType
+import com.twitter.scalding.JobTest
+import com.twitter.scalding.Tsv
+import sources.{PrepTsvSource, PreprocessedSpanSource, Util}
+
+/**
+* Tests that ExpensiveEndpointSpec finds the average run time of each service
+*/
+
+class ExpensiveEndpointsSpec extends Specification with TupleConversions {
+  noDetailedDiffs()
+
+  implicit val dateRange = DateRange(RichDate(123), RichDate(321))
+
+  val endpoint = new gen.Endpoint(123, 666, "service")
+  val endpoint1 = new gen.Endpoint(123, 666, "service1")
+  val endpoint2 = new gen.Endpoint(123, 666, "service2")
+  val span = new gen.SpanServiceName(12345, "methodcall", 666,
+    List(new gen.Annotation(2000, "sr").setHost(endpoint), new gen.Annotation(3000, "ss").setHost(endpoint)).asJava,
+    List[gen.BinaryAnnotation]().asJava, "service", "service")
+  val span1 = new gen.SpanServiceName(123456, "methodcall", 666,
+    List(new gen.Annotation(1000, "cs").setHost(endpoint2), new gen.Annotation(1500, "sr").setHost(endpoint2), new gen.Annotation(4500, "ss").setHost(endpoint2), new gen.Annotation(5000, "cr").setHost(endpoint2)).asJava,
+    List(new gen.BinaryAnnotation("bye", null, AnnotationType.BOOL)).asJava, "service2", "service2")
+
+  val spans = Util.repeatSpan(span, 30, 40, -1) ++ Util.repeatSpan(span1, 30, 100, 40)
+
+  "ExpensiveEndpoints" should {
+    "Return the most common service calls" in {
+      JobTest("com.twitter.zipkin.hadoop.ExpensiveEndpoints").
+        arg("input", "inputFile").
+        arg("output", "outputFile").
+        arg("date", "2012-01-01T01:00").
+        source(PreprocessedSpanSource(), spans).
+        source(PrepTsvSource(), Util.getSpanIDtoNames(spans)).
+        sink[(String, String, Long)](Tsv("outputFile")) {
+        val result = new HashMap[String, Long]()
+        result("service, service2") = 0
+        outputBuffer => outputBuffer foreach { e =>
+          println(e)
+          result(e._1 + ", " + e._2) = e._3
+        }
+//        result("Unknown Service Name") mustEqual 3000
+//        result("service") mustEqual 2000
+//        result("service2") mustEqual 3000
+          result("service, service2") mustEqual 4000
+      }
+    }.run.finish
+  }
+}