diff options
Diffstat (limited to 'kamon-core/src/test/scala/kamon/instrumentation')
5 files changed, 441 insertions, 0 deletions
diff --git a/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala new file mode 100644 index 00000000..0f682500 --- /dev/null +++ b/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala @@ -0,0 +1,87 @@ +/* =================================================== + * Copyright © 2013 the kamon project <http://kamon.io/> + * + * 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 kamon.instrumentation.akka + +import akka.actor.{ Actor, ActorSystem, Props } +import akka.pattern.{ ask, pipe } +import akka.routing.RoundRobinPool +import akka.testkit.{ ImplicitSender, TestKit } +import akka.util.Timeout +import kamon.trace.TraceRecorder +import org.scalatest.WordSpecLike + +import scala.concurrent.duration._ + +class ActorCellInstrumentationSpec extends TestKit(ActorSystem("actor-cell-instrumentation-spec")) with WordSpecLike + with ImplicitSender { + + implicit val executionContext = system.dispatcher + + "the message passing instrumentation" should { + "propagate the TraceContext using bang" in new EchoActorFixture { + val testTraceContext = TraceRecorder.withNewTraceContext("bang-reply") { + ctxEchoActor ! "test" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext using tell" in new EchoActorFixture { + val testTraceContext = TraceRecorder.withNewTraceContext("tell-reply") { + ctxEchoActor.tell("test", testActor) + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext using ask" in new EchoActorFixture { + implicit val timeout = Timeout(1 seconds) + val testTraceContext = TraceRecorder.withNewTraceContext("ask-reply") { + // The pipe pattern use Futures internally, so FutureTracing test should cover the underpinnings of it. + (ctxEchoActor ? "test") pipeTo (testActor) + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext to actors behind a router" in new RoutedEchoActorFixture { + val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + ctxEchoActor ! "test" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + } + + trait EchoActorFixture { + val ctxEchoActor = system.actorOf(Props[TraceContextEcho]) + } + + trait RoutedEchoActorFixture extends EchoActorFixture { + override val ctxEchoActor = system.actorOf(Props[TraceContextEcho].withRouter(RoundRobinPool(nrOfInstances = 1))) + } +} + +class TraceContextEcho extends Actor { + def receive = { + case msg: String ⇒ sender ! TraceRecorder.currentContext + } +} + diff --git a/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala new file mode 100644 index 00000000..3dab44bc --- /dev/null +++ b/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala @@ -0,0 +1,52 @@ +/* =================================================== + * Copyright © 2013 the kamon project <http://kamon.io/> + * + * 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 kamon.instrumentation.akka + +import akka.actor.{ Actor, ActorLogging, ActorSystem, Props } +import akka.event.Logging.LogEvent +import akka.testkit.TestKit +import kamon.trace.{ TraceContextAware, TraceRecorder } +import org.scalatest.{ Inspectors, Matchers, WordSpecLike } + +class ActorLoggingInstrumentationSpec extends TestKit(ActorSystem("actor-logging-instrumentation-spec")) with WordSpecLike + with Matchers with Inspectors { + + "the ActorLogging instrumentation" should { + "attach the TraceContext (if available) to log events" in { + val loggerActor = system.actorOf(Props[LoggerActor]) + system.eventStream.subscribe(testActor, classOf[LogEvent]) + + val testTraceContext = TraceRecorder.withNewTraceContext("logging") { + loggerActor ! "info" + TraceRecorder.currentContext + } + + fishForMessage() { + case event: LogEvent if event.message.toString contains "TraceContext =>" ⇒ + val ctxInEvent = event.asInstanceOf[TraceContextAware].traceContext + ctxInEvent === testTraceContext + + case event: LogEvent ⇒ false + } + } + } +} + +class LoggerActor extends Actor with ActorLogging { + def receive = { + case "info" ⇒ log.info("TraceContext => {}", TraceRecorder.currentContext) + } +} diff --git a/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala new file mode 100644 index 00000000..47867c55 --- /dev/null +++ b/kamon-core/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala @@ -0,0 +1,172 @@ +package kamon.instrumentation.akka + +import akka.actor.SupervisorStrategy.{ Escalate, Restart, Resume, Stop } +import akka.actor._ +import akka.testkit.{ ImplicitSender, TestKit } +import kamon.trace.TraceRecorder +import org.scalatest.WordSpecLike + +import scala.concurrent.duration._ +import scala.util.control.NonFatal + +class ActorSystemMessageInstrumentationSpec extends TestKit(ActorSystem("actor-system-message-instrumentation-spec")) + with WordSpecLike with ImplicitSender { + + implicit val executionContext = system.dispatcher + + "the system message passing instrumentation" should { + "keep the TraceContext while processing the Create message in top level actors" in { + val testTraceContext = TraceRecorder.withNewTraceContext("creating-top-level-actor") { + system.actorOf(Props(new Actor { + testActor ! TraceRecorder.currentContext + def receive: Actor.Receive = { case any ⇒ } + })) + + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "keep the TraceContext while processing the Create message in non top level actors" in { + val testTraceContext = TraceRecorder.withNewTraceContext("creating-non-top-level-actor") { + system.actorOf(Props(new Actor { + def receive: Actor.Receive = { + case any ⇒ + context.actorOf(Props(new Actor { + testActor ! TraceRecorder.currentContext + def receive: Actor.Receive = { case any ⇒ } + })) + } + })) ! "any" + + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "keep the TraceContext in the supervision cycle" when { + "the actor is resumed" in { + val supervisor = supervisorWithDirective(Resume) + + val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-resume") { + supervisor ! "fail" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) // From the parent executing the supervision strategy + + // Ensure we didn't tie the actor with the context + supervisor ! "context" + expectMsg(None) + } + + "the actor is restarted" in { + val supervisor = supervisorWithDirective(Restart, sendPreRestart = true, sendPostRestart = true) + + val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-restart") { + supervisor ! "fail" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) // From the parent executing the supervision strategy + expectMsg(testTraceContext) // From the preRestart hook + expectMsg(testTraceContext) // From the postRestart hook + + // Ensure we didn't tie the actor with the context + supervisor ! "context" + expectMsg(None) + } + + "the actor is stopped" in { + val supervisor = supervisorWithDirective(Stop, sendPostStop = true) + + val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-stop") { + supervisor ! "fail" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) // From the parent executing the supervision strategy + expectMsg(testTraceContext) // From the postStop hook + expectNoMsg(1 second) + } + + "the failure is escalated" in { + val supervisor = supervisorWithDirective(Escalate, sendPostStop = true) + + val testTraceContext = TraceRecorder.withNewTraceContext("fail-and-escalate") { + supervisor ! "fail" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) // From the parent executing the supervision strategy + expectMsg(testTraceContext) // From the grandparent executing the supervision strategy + expectMsg(testTraceContext) // From the postStop hook in the child + expectMsg(testTraceContext) // From the postStop hook in the parent + expectNoMsg(1 second) + } + } + } + + def supervisorWithDirective(directive: SupervisorStrategy.Directive, sendPreRestart: Boolean = false, sendPostRestart: Boolean = false, + sendPostStop: Boolean = false, sendPreStart: Boolean = false): ActorRef = { + class GrandParent extends Actor { + val child = context.actorOf(Props(new Parent)) + + override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() { + case NonFatal(throwable) ⇒ testActor ! TraceRecorder.currentContext; Stop + } + + def receive = { + case any ⇒ child forward any + } + } + + class Parent extends Actor { + val child = context.actorOf(Props(new Child)) + + override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() { + case NonFatal(throwable) ⇒ testActor ! TraceRecorder.currentContext; directive + } + + def receive: Actor.Receive = { + case any ⇒ child forward any + } + + override def postStop(): Unit = { + if (sendPostStop) testActor ! TraceRecorder.currentContext + super.postStop() + } + } + + class Child extends Actor { + def receive = { + case "fail" ⇒ 1 / 0 + case "context" ⇒ sender ! TraceRecorder.currentContext + } + + override def preRestart(reason: Throwable, message: Option[Any]): Unit = { + if (sendPreRestart) testActor ! TraceRecorder.currentContext + super.preRestart(reason, message) + } + + override def postRestart(reason: Throwable): Unit = { + if (sendPostRestart) testActor ! TraceRecorder.currentContext + super.postRestart(reason) + } + + override def postStop(): Unit = { + if (sendPostStop) testActor ! TraceRecorder.currentContext + super.postStop() + } + + override def preStart(): Unit = { + if (sendPreStart) testActor ! TraceRecorder.currentContext + super.preStart() + } + } + + system.actorOf(Props(new GrandParent)) + } +} diff --git a/kamon-core/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala new file mode 100644 index 00000000..d914ffe8 --- /dev/null +++ b/kamon-core/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala @@ -0,0 +1,67 @@ +/* + * ========================================================================================= + * Copyright © 2013 the kamon project <http://kamon.io/> + * + * 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 kamon.instrumentation.akka + +import akka.actor.{ Actor, ActorSystem, Props } +import akka.event.Logging.Warning +import akka.pattern.ask +import akka.testkit.TestKitBase +import akka.util.Timeout +import com.typesafe.config.ConfigFactory +import kamon.trace.{ TraceContextAware, TraceRecorder } +import org.scalatest.{ Matchers, WordSpecLike } + +import scala.concurrent.duration._ + +class AskPatternInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers { + implicit lazy val system: ActorSystem = ActorSystem("ask-pattern-tracing-spec", ConfigFactory.parseString( + """ + |kamon { + | trace { + | ask-pattern-tracing = on + | } + |} + """.stripMargin)) + + "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]) + + val testTraceContext = TraceRecorder.withNewTraceContext("ask-timeout-warning") { + noReply ? "hello" + TraceRecorder.currentContext + } + + val warn = expectMsgPF() { + case warn: Warning if warn.message.toString.contains("Timeout triggered for ask pattern") ⇒ warn + } + val capturedCtx = warn.asInstanceOf[TraceContextAware].traceContext + + capturedCtx should be('defined) + capturedCtx should equal(testTraceContext) + } + } +} + +class NoReply extends Actor { + def receive = { + case any ⇒ + } +} diff --git a/kamon-core/src/test/scala/kamon/instrumentation/scala/FutureInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/scala/FutureInstrumentationSpec.scala new file mode 100644 index 00000000..31afd3ff --- /dev/null +++ b/kamon-core/src/test/scala/kamon/instrumentation/scala/FutureInstrumentationSpec.scala @@ -0,0 +1,63 @@ +/* =================================================== + * Copyright © 2013 the kamon project <http://kamon.io/> + * + * 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 kamon.instrumentation.scala + +import akka.actor.ActorSystem +import akka.testkit.TestKit +import kamon.trace.TraceRecorder +import org.scalatest.concurrent.{ PatienceConfiguration, ScalaFutures } +import org.scalatest.{ Matchers, OptionValues, WordSpecLike } + +import scala.concurrent.Future + +class FutureInstrumentationSpec extends TestKit(ActorSystem("future-instrumentation-spec")) with WordSpecLike with Matchers + with ScalaFutures with PatienceConfiguration with OptionValues { + + implicit val execContext = system.dispatcher + + "a Future created with FutureTracing" should { + "capture the TraceContext available when created" which { + "must be available when executing the future's body" in { + + val (future, testTraceContext) = TraceRecorder.withNewTraceContext("future-body") { + val future = Future(TraceRecorder.currentContext) + + (future, TraceRecorder.currentContext) + } + + whenReady(future)(ctxInFuture ⇒ + ctxInFuture should equal(testTraceContext)) + } + + "must be available when executing callbacks on the future" in { + + val (future, testTraceContext) = TraceRecorder.withNewTraceContext("future-body") { + val future = Future("Hello Kamon!") + // The TraceContext is expected to be available during all intermediate processing. + .map(_.length) + .flatMap(len ⇒ Future(len.toString)) + .map(s ⇒ TraceRecorder.currentContext) + + (future, TraceRecorder.currentContext) + } + + whenReady(future)(ctxInFuture ⇒ + ctxInFuture should equal(testTraceContext)) + } + } + } +} + |