diff options
Diffstat (limited to 'kamon-akka/src/test/scala/kamon/akka')
4 files changed, 517 insertions, 0 deletions
diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala new file mode 100644 index 00000000..8f7ae613 --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorCellInstrumentationSpec.scala @@ -0,0 +1,122 @@ +/* =================================================== + * 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, Props } +import akka.pattern.{ ask, pipe } +import akka.routing._ +import akka.util.Timeout +import kamon.testkit.BaseKamonSpec +import kamon.trace.TraceContext + +import scala.concurrent.duration._ + +class ActorCellInstrumentationSpec extends BaseKamonSpec("actor-cell-instrumentation-spec") { + implicit lazy val executionContext = system.dispatcher + + "the message passing instrumentation" should { + "propagate the TraceContext using bang" in new EchoActorFixture { + val testTraceContext = TraceContext.withContext(newContext("bang-reply")) { + ctxEchoActor ! "test" + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext using tell" in new EchoActorFixture { + val testTraceContext = TraceContext.withContext(newContext("tell-reply")) { + ctxEchoActor.tell("test", testActor) + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext using ask" in new EchoActorFixture { + implicit val timeout = Timeout(1 seconds) + val testTraceContext = TraceContext.withContext(newContext("ask-reply")) { + // The pipe pattern use Futures internally, so FutureTracing test should cover the underpinnings of it. + (ctxEchoActor ? "test") pipeTo (testActor) + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext to actors behind a simple router" in new EchoSimpleRouterFixture { + val testTraceContext = TraceContext.withContext(newContext("router-reply")) { + router.route("test", testActor) + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext to actors behind a pool router" in new EchoPoolRouterFixture { + val testTraceContext = TraceContext.withContext(newContext("router-reply")) { + pool ! "test" + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext to actors behind a group router" in new EchoGroupRouterFixture { + val testTraceContext = TraceContext.withContext(newContext("router-reply")) { + group ! "test" + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + } + + override protected def afterAll(): Unit = shutdown() + + trait EchoActorFixture { + val ctxEchoActor = system.actorOf(Props[TraceContextEcho]) + } + + trait EchoSimpleRouterFixture { + val router = { + val routees = Vector.fill(5) { + val r = system.actorOf(Props[TraceContextEcho]) + ActorRefRoutee(r) + } + Router(RoundRobinRoutingLogic(), routees) + } + } + + trait EchoPoolRouterFixture { + val pool = system.actorOf(RoundRobinPool(nrOfInstances = 5).props(Props[TraceContextEcho]), "pool-router") + } + + trait EchoGroupRouterFixture { + val routees = Vector.fill(5) { + system.actorOf(Props[TraceContextEcho]) + } + + val group = system.actorOf(RoundRobinGroup(routees.map(_.path.toStringWithoutAddress)).props(), "group-router") + } +} + +class TraceContextEcho extends Actor { + def receive = { + case msg: String ⇒ sender ! TraceContext.currentContext + } +} + diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala new file mode 100644 index 00000000..21706af9 --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorLoggingInstrumentationSpec.scala @@ -0,0 +1,76 @@ +/* =================================================== + * 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, Props } +import akka.event.Logging.LogEvent +import com.typesafe.config.ConfigFactory +import kamon.testkit.BaseKamonSpec +import kamon.trace.TraceLocal.AvailableToMdc +import kamon.trace.logging.MdcKeysSupport +import kamon.trace.{ TraceContextAware, TraceLocal, TraceContext } +import org.scalatest.Inspectors +import org.slf4j.MDC + +class ActorLoggingInstrumentationSpec extends BaseKamonSpec("actor-logging-instrumentation-spec") with Inspectors with MdcKeysSupport { + override lazy val config = + ConfigFactory.parseString( + """ + |akka { + | loggers = ["akka.event.slf4j.Slf4jLogger"] + |} + """.stripMargin) + + "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 = TraceContext.withContext(newContext("logging")) { + loggerActor ! "info" + TraceContext.currentContext + } + + fishForMessage() { + case event: LogEvent if event.message.toString startsWith "TraceContext" ⇒ + val ctxInEvent = event.asInstanceOf[TraceContextAware].traceContext + ctxInEvent === testTraceContext + + case event: LogEvent ⇒ false + } + } + + "allow retrieve a value from the MDC when was created a key of type AvailableToMdc" in { + val testString = "Hello World" + TraceContext.withContext(newContext("logging-with-mdc")) { + TraceLocal.store(AvailableToMdc("some-cool-key"))(testString) + + withMdc { + MDC.get("other-key") shouldBe (null) + MDC.get("some-cool-key") should equal(testString) + } + } + } + } + + override protected def afterAll(): Unit = shutdown() +} + +class LoggerActor extends Actor with ActorLogging { + def receive = { + case "info" ⇒ log.info("TraceContext(name = {}, token = {})", TraceContext.currentContext.name, TraceContext.currentContext.token) + } +} diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala new file mode 100644 index 00000000..8c1033ae --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/ActorSystemMessageInstrumentationSpec.scala @@ -0,0 +1,180 @@ +package kamon.instrumentation.akka + +import akka.actor.SupervisorStrategy.{ Escalate, Restart, Resume, Stop } +import akka.actor._ +import akka.testkit.ImplicitSender +import com.typesafe.config.ConfigFactory +import kamon.testkit.BaseKamonSpec +import kamon.trace.{ EmptyTraceContext, TraceContext } +import org.scalatest.WordSpecLike + +import scala.concurrent.duration._ +import scala.util.control.NonFatal + +class ActorSystemMessageInstrumentationSpec extends BaseKamonSpec("actor-system-message-instrumentation-spec") with WordSpecLike with ImplicitSender { + override lazy val config = + ConfigFactory.parseString( + """ + |akka { + | loglevel = OFF + |} + """.stripMargin) + + implicit lazy 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 = TraceContext.withContext(newContext("creating-top-level-actor")) { + system.actorOf(Props(new Actor { + testActor ! TraceContext.currentContext + def receive: Actor.Receive = { case any ⇒ } + })) + + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "keep the TraceContext while processing the Create message in non top level actors" in { + val testTraceContext = TraceContext.withContext(newContext("creating-non-top-level-actor")) { + system.actorOf(Props(new Actor { + def receive: Actor.Receive = { + case any ⇒ + context.actorOf(Props(new Actor { + testActor ! TraceContext.currentContext + def receive: Actor.Receive = { case any ⇒ } + })) + } + })) ! "any" + + TraceContext.currentContext + } + + expectMsg(testTraceContext) + } + + "keep the TraceContext in the supervision cycle" when { + "the actor is resumed" in { + val supervisor = supervisorWithDirective(Resume) + + val testTraceContext = TraceContext.withContext(newContext("fail-and-resume")) { + supervisor ! "fail" + TraceContext.currentContext + } + + expectMsg(testTraceContext) // From the parent executing the supervision strategy + + // Ensure we didn't tie the actor with the context + supervisor ! "context" + expectMsg(EmptyTraceContext) + } + + "the actor is restarted" in { + val supervisor = supervisorWithDirective(Restart, sendPreRestart = true, sendPostRestart = true) + + val testTraceContext = TraceContext.withContext(newContext("fail-and-restart")) { + supervisor ! "fail" + TraceContext.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(EmptyTraceContext) + } + + "the actor is stopped" in { + val supervisor = supervisorWithDirective(Stop, sendPostStop = true) + + val testTraceContext = TraceContext.withContext(newContext("fail-and-stop")) { + supervisor ! "fail" + TraceContext.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 = TraceContext.withContext(newContext("fail-and-escalate")) { + supervisor ! "fail" + TraceContext.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 ! TraceContext.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 ! TraceContext.currentContext; directive + } + + def receive: Actor.Receive = { + case any ⇒ child forward any + } + + override def postStop(): Unit = { + if (sendPostStop) testActor ! TraceContext.currentContext + super.postStop() + } + } + + class Child extends Actor { + def receive = { + case "fail" ⇒ throw new ArithmeticException("Division by zero.") + case "context" ⇒ sender ! TraceContext.currentContext + } + + override def preRestart(reason: Throwable, message: Option[Any]): Unit = { + if (sendPreRestart) testActor ! TraceContext.currentContext + super.preRestart(reason, message) + } + + override def postRestart(reason: Throwable): Unit = { + if (sendPostRestart) testActor ! TraceContext.currentContext + super.postRestart(reason) + } + + override def postStop(): Unit = { + if (sendPostStop) testActor ! TraceContext.currentContext + super.postStop() + } + + override def preStart(): Unit = { + if (sendPreStart) testActor ! TraceContext.currentContext + super.preStart() + } + } + + system.actorOf(Props(new GrandParent)) + } +} diff --git a/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala new file mode 100644 index 00000000..0d63a19e --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/akka/instrumentation/AskPatternInstrumentationSpec.scala @@ -0,0 +1,139 @@ +/* + * ========================================================================================= + * 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 java.util.concurrent.atomic.AtomicInteger + +import akka.actor._ +import akka.event.Logging.Warning +import akka.pattern.ask +import akka.testkit.TestProbe +import akka.util.Timeout +import com.typesafe.config.ConfigFactory +import kamon.Kamon +import kamon.akka.Akka +import kamon.testkit.BaseKamonSpec +import kamon.trace.{ TraceContext, TraceContextAware } + +import scala.concurrent.duration._ + +class AskPatternInstrumentationSpec extends BaseKamonSpec("ask-pattern-tracing-spec") { + override lazy val config = + ConfigFactory.parseString( + """ + |akka { + | loglevel = OFF + |} + """.stripMargin) + + implicit lazy val ec = system.dispatcher + implicit val askTimeout = Timeout(10 millis) + + // TODO: Make this work with ActorSelections + + "the AskPatternInstrumentation" when { + "configured in heavyweight mode" should { + "log a warning with a full stack trace and the TraceContext taken from the moment the ask was triggered for a actor" in new NoReplyFixture { + setAskPatternTimeoutWarningMode("heavyweight") + + expectTimeoutWarning() { + TraceContext.withContext(newContext("ask-timeout-warning")) { + noReplyActorRef ? "hello" + TraceContext.currentContext + } + } + } + } + + "configured in lightweight mode" should { + "log a warning with a short source location description and the TraceContext taken from the moment the ask was triggered for a actor" in new NoReplyFixture { + setAskPatternTimeoutWarningMode("lightweight") + + expectTimeoutWarning(messageSizeLimit = Some(1)) { + TraceContext.withContext(newContext("ask-timeout-warning")) { + noReplyActorRef ? "hello" + TraceContext.currentContext + } + } + } + } + + "configured in off mode" should { + "should not log any warning messages" in new NoReplyFixture { + setAskPatternTimeoutWarningMode("off") + + expectTimeoutWarning(expectWarning = false) { + TraceContext.withContext(newContext("ask-timeout-warning")) { + noReplyActorRef ? "hello" + TraceContext.currentContext + } + } + } + } + } + + override protected def afterAll(): Unit = shutdown() + + def expectTimeoutWarning(messageSizeLimit: Option[Int] = None, expectWarning: Boolean = true)(thunk: ⇒ TraceContext): Unit = { + val listener = warningListener() + val testTraceContext = thunk + + if (expectWarning) { + val warning = listener.fishForMessage() { + case Warning(_, _, msg) if msg.toString.startsWith("Timeout triggered for ask pattern registered at") ⇒ true + case others ⇒ false + }.asInstanceOf[Warning] + + warning.asInstanceOf[TraceContextAware].traceContext should equal(testTraceContext) + messageSizeLimit.map { messageLimit ⇒ + warning.message.toString.lines.size should be(messageLimit) + } + } else { + listener.expectNoMsg() + } + } + + def warningListener(): TestProbe = { + val listener = TestProbe() + system.eventStream.subscribe(listener.ref, classOf[Warning]) + listener + } + + def setAskPatternTimeoutWarningMode(mode: String): Unit = { + val target = Kamon(Akka)(system) + val field = target.getClass.getDeclaredField("askPatternTimeoutWarning") + field.setAccessible(true) + field.set(target, mode) + } + + val fixtureCounter = new AtomicInteger(0) + + trait NoReplyFixture { + def noReplyActorRef: ActorRef = system.actorOf(Props[NoReply], "no-reply-" + fixtureCounter.incrementAndGet()) + + def noReplyActorSelection: ActorSelection = { + val target = noReplyActorRef + system.actorSelection(target.path) + } + } +} + +class NoReply extends Actor { + def receive = { + case any ⇒ + } +} |