simple test for ActorLogging instrumentation

5 files changed, 243 insertions, 0 deletions

diff --git a/kamon-trace/src/test/scala/kamon/trace/instrumentation/ActorLoggingSpec.scala b/kamon-trace/src/test/scala/kamon/trace/instrumentation/ActorLoggingSpec.scala
new file mode 100644
index 00000000..896faf69
--- /dev/null
+++ b/kamon-trace/src/test/scala/kamon/trace/instrumentation/ActorLoggingSpec.scala
@@ -0,0 +1,34 @@
+package kamon.trace.instrumentation
+
+import akka.testkit.TestKit
+import org.scalatest.{Inspectors, Matchers, WordSpecLike}
+import akka.actor.{Props, ActorLogging, Actor, ActorSystem}
+import akka.event.Logging.{LogEvent}
+import kamon.trace.{ContextAware, TraceContext, Trace}
+
+class ActorLoggingSpec extends TestKit(ActorSystem("actor-logging-spec")) with WordSpecLike with Matchers with Inspectors {
+
+  "the ActorLogging instrumentation" should {
+    "attach the TraceContext (if available) to log events" in {
+      val testTraceContext = Some(TraceContext(Actor.noSender, 1))
+      val loggerActor = system.actorOf(Props[LoggerActor])
+      system.eventStream.subscribe(testActor, classOf[LogEvent])
+
+      Trace.withContext(testTraceContext) {
+        loggerActor ! "info"
+      }
+
+      expectMsgPF() {
+        case event: LogEvent =>
+          val ctxInEvent = event.asInstanceOf[ContextAware].traceContext
+          ctxInEvent should equal(testTraceContext)
+      }
+    }
+  }
+}
+
+class LoggerActor extends Actor with ActorLogging {
+  def receive = {
+    case "info" => log.info("TraceContext => {}", Trace.context())
+  }
+}
diff --git a/kamon-trace/src/test/scala/kamon/trace/instrumentation/ActorMessagePassingTracingSpec.scala b/kamon-trace/src/test/scala/kamon/trace/instrumentation/ActorMessagePassingTracingSpec.scala
new file mode 100644
index 00000000..441f3e47
--- /dev/null
+++ b/kamon-trace/src/test/scala/kamon/trace/instrumentation/ActorMessagePassingTracingSpec.scala
@@ -0,0 +1,73 @@
+package kamon.trace.instrumentation
+
+import org.scalatest.{WordSpecLike, Matchers}
+import akka.actor.{ActorRef, Actor, Props, ActorSystem}
+
+import akka.testkit.{ImplicitSender, TestKit}
+import kamon.trace.Trace
+import akka.pattern.{pipe, ask}
+import akka.util.Timeout
+import scala.concurrent.duration._
+import scala.concurrent.{Await, Future}
+import akka.routing.RoundRobinRouter
+import kamon.trace.TraceContext
+
+
+class ActorMessagePassingTracingSpec extends TestKit(ActorSystem("actor-message-passing-tracing-spec")) with WordSpecLike with ImplicitSender {
+  implicit val executionContext = system.dispatcher
+
+  "the message passing instrumentation" should {
+      "propagate the TraceContext using bang" in new TraceContextEchoFixture {
+        Trace.withContext(testTraceContext) {
+          ctxEchoActor ! "test"
+        }
+
+        expectMsg(testTraceContext)
+      }
+
+      "propagate the TraceContext using tell" in new TraceContextEchoFixture {
+        Trace.withContext(testTraceContext) {
+          ctxEchoActor.tell("test", testActor)
+        }
+
+        expectMsg(testTraceContext)
+      }
+
+      "propagate the TraceContext using ask" in new TraceContextEchoFixture {
+        implicit val timeout = Timeout(1 seconds)
+        Trace.withContext(testTraceContext) {
+          // The pipe pattern use Futures internally, so FutureTracing test should cover the underpinnings of it.
+          (ctxEchoActor ? "test") pipeTo(testActor)
+        }
+
+        expectMsg(testTraceContext)
+      }
+
+      "propagate the TraceContext to actors behind a router" in new RoutedTraceContextEchoFixture {
+        Trace.withContext(testTraceContext) {
+          ctxEchoActor ! "test"
+        }
+
+        expectMsg(testTraceContext)
+      }
+  }
+
+  trait TraceContextEchoFixture {
+    val testTraceContext = Some(Trace.newTraceContext(""))
+    val ctxEchoActor = system.actorOf(Props[TraceContextEcho])
+  }
+
+  trait RoutedTraceContextEchoFixture extends TraceContextEchoFixture {
+    override val ctxEchoActor = system.actorOf(Props[TraceContextEcho].withRouter(RoundRobinRouter(nrOfInstances = 1)))
+  }
+}
+
+class TraceContextEcho extends Actor {
+  def receive = {
+    case msg: String ⇒ sender ! Trace.context()
+  }
+}
+
+
+
+
diff --git a/kamon-trace/src/test/scala/kamon/trace/instrumentation/AskPatternTracingSpec.scala b/kamon-trace/src/test/scala/kamon/trace/instrumentation/AskPatternTracingSpec.scala
new file mode 100644
index 00000000..cb18fe86
--- /dev/null
+++ b/kamon-trace/src/test/scala/kamon/trace/instrumentation/AskPatternTracingSpec.scala
@@ -0,0 +1,43 @@
+package kamon.trace.instrumentation
+
+import akka.testkit.TestKit
+import akka.actor.{Props, Actor, ActorSystem}
+import org.scalatest.{Matchers, WordSpecLike}
+import akka.event.Logging.Warning
+import scala.concurrent.duration._
+import akka.pattern.ask
+import akka.util.Timeout
+import kamon.trace.{Trace, ContextAware}
+import org.scalatest.OptionValues._
+
+
+class AskPatternTracingSpec extends TestKit(ActorSystem("ask-pattern-tracing-spec")) with WordSpecLike with Matchers {
+
+  "the AskPatternTracing" should {
+    "log a warning with a stack trace and TraceContext taken from the moment the ask was triggered" in {
+      implicit val ec = system.dispatcher
+      implicit val timeout = Timeout(10 milliseconds)
+      val noReply = system.actorOf(Props[NoReply])
+      system.eventStream.subscribe(testActor, classOf[Warning])
+
+      within(500 milliseconds) {
+        val initialCtx = Trace.start("ask-test")
+        noReply ? "hello"
+
+        val warn = expectMsgPF() {
+          case warn: Warning if warn.message.toString.contains("Timeout triggered for ask pattern") => warn
+        }
+        val capturedCtx = warn.asInstanceOf[ContextAware].traceContext
+
+        capturedCtx should be('defined)
+        capturedCtx.value should equal (initialCtx)
+      }
+    }
+  }
+}
+
+class NoReply extends Actor {
+  def receive = {
+    case any =>
+  }
+}
diff --git a/kamon-trace/src/test/scala/kamon/trace/instrumentation/FutureTracingSpec.scala b/kamon-trace/src/test/scala/kamon/trace/instrumentation/FutureTracingSpec.scala
new file mode 100644
index 00000000..9ba98381
--- /dev/null
+++ b/kamon-trace/src/test/scala/kamon/trace/instrumentation/FutureTracingSpec.scala
@@ -0,0 +1,56 @@
+package kamon.trace.instrumentation
+
+import scala.concurrent.{ExecutionContext, Await, Promise, Future}
+import org.scalatest.{Matchers, OptionValues, WordSpec}
+import org.scalatest.concurrent.{ScalaFutures, PatienceConfiguration}
+import java.util.UUID
+import scala.util.{Random, Success}
+import scala.concurrent.duration._
+import java.util.concurrent.TimeUnit
+import akka.actor.{Actor, ActorSystem}
+import kamon.trace.{Trace, TraceContext}
+
+
+class FutureTracingSpec extends WordSpec with Matchers with ScalaFutures with PatienceConfiguration with OptionValues {
+
+  implicit val execContext = ExecutionContext.Implicits.global
+
+  "a Future created with FutureTracing" should {
+    "capture the TraceContext available when created" which {
+      "must be available when executing the future's body" in new TraceContextFixture {
+        var future: Future[Option[TraceContext]] = _
+
+        Trace.withContext(testTraceContext) {
+          future = Future(Trace.context)
+        }
+
+        whenReady(future)( ctxInFuture =>
+          ctxInFuture should equal(testTraceContext)
+        )
+      }
+
+      "must be available when executing callbacks on the future" in new TraceContextFixture {
+        var future: Future[Option[TraceContext]] = _
+
+        Trace.withContext(testTraceContext) {
+          future = Future("Hello Kamon!")
+            // The TraceContext is expected to be available during all intermediate processing.
+            .map (_.length)
+            .flatMap(len => Future(len.toString))
+            .map (s => Trace.context())
+        }
+
+        whenReady(future)( ctxInFuture =>
+          ctxInFuture should equal(testTraceContext)
+        )
+      }
+    }
+  }
+
+  trait TraceContextFixture {
+    val random = new Random(System.nanoTime)
+    val testTraceContext = Some(TraceContext(Actor.noSender, random.nextInt))
+  }
+}
+
+
diff --git a/kamon-trace/src/test/scala/kamon/trace/instrumentation/TraceAggregatorSpec.scala b/kamon-trace/src/test/scala/kamon/trace/instrumentation/TraceAggregatorSpec.scala
new file mode 100644
index 00000000..81f92b6e
--- /dev/null
+++ b/kamon-trace/src/test/scala/kamon/trace/instrumentation/TraceAggregatorSpec.scala
@@ -0,0 +1,37 @@
+package kamon.trace.instrumentation
+
+import org.scalatest.{WordSpecLike, WordSpec}
+import akka.testkit.{TestKitBase, TestKit}
+import akka.actor.ActorSystem
+import scala.concurrent.duration._
+import kamon.trace.UowTracing.{Finish, Rename, Start}
+import kamon.trace.{UowTrace, UowTraceAggregator}
+
+class TraceAggregatorSpec extends TestKit(ActorSystem("TraceAggregatorSpec")) with WordSpecLike {
+
+  "a TraceAggregator" should {
+    "send a UowTrace message out after receiving a Finish message" in new AggregatorFixture {
+      within(1 second) {
+        aggregator ! Start(1, "/accounts")
+        aggregator ! Finish(1)
+
+        //expectMsg(UowTrace("UNKNOWN", Seq(Start(1, "/accounts"), Finish(1))))
+      }
+    }
+
+    "change the uow name after receiving a Rename message" in new AggregatorFixture {
+      within(1 second) {
+        aggregator ! Start(1, "/accounts")
+        aggregator ! Rename(1, "test-uow")
+        aggregator ! Finish(1)
+
+        //expectMsg(UowTrace("test-uow", Seq(Start(1, "/accounts"), Finish(1))))
+      }
+    }
+  }
+
+
+  trait AggregatorFixture {
+    val aggregator = system.actorOf(UowTraceAggregator.props(testActor, 10 seconds))
+  }
+}