diff options
Diffstat (limited to 'kamon-akka/src/test/scala/kamon/instrumentation/akka')
4 files changed, 506 insertions, 0 deletions
diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala new file mode 100644 index 00000000..06a232bd --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorCellInstrumentationSpec.scala @@ -0,0 +1,125 @@ +/* =================================================== + * 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._ +import akka.testkit.{ TestKitBase, ImplicitSender, TestKit } +import akka.util.Timeout +import com.typesafe.config.ConfigFactory +import kamon.trace.TraceRecorder +import org.scalatest.{ BeforeAndAfterAll, WordSpecLike } + +import scala.concurrent.duration._ + +class ActorCellInstrumentationSpec extends TestKitBase with WordSpecLike with ImplicitSender with BeforeAndAfterAll { + implicit lazy val system: ActorSystem = ActorSystem("actor-cell-instrumentation-spec") + 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 simple router" in new EchoSimpleRouterFixture { + val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + router.route("test", testActor) + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext to actors behind a pool router" in new EchoPoolRouterFixture { + val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + pool ! "test" + TraceRecorder.currentContext + } + + expectMsg(testTraceContext) + } + + "propagate the TraceContext to actors behind a group router" in new EchoGroupRouterFixture { + val testTraceContext = TraceRecorder.withNewTraceContext("router-reply") { + group ! "test" + TraceRecorder.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 ! TraceRecorder.currentContext + } +} + diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala new file mode 100644 index 00000000..598e9327 --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorLoggingInstrumentationSpec.scala @@ -0,0 +1,71 @@ +/* =================================================== + * 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.TestKitBase +import com.typesafe.config.ConfigFactory +import kamon.trace.TraceLocal.AvailableToMdc +import kamon.trace.logging.MdcKeysSupport +import kamon.trace.{ TraceContextAware, TraceLocal, TraceRecorder } +import org.scalatest.{ BeforeAndAfterAll, Inspectors, Matchers, WordSpecLike } +import org.slf4j.MDC + +class ActorLoggingInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with Inspectors with MdcKeysSupport with BeforeAndAfterAll { + implicit lazy val system: ActorSystem = ActorSystem("actor-logging-instrumentation-spec", + ConfigFactory.parseString("""akka.loggers = ["akka.event.slf4j.Slf4jLogger"]""")) + + "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 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" + TraceRecorder.withNewTraceContext("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 = {})", TraceRecorder.currentContext.name, TraceRecorder.currentContext.token) + } +} diff --git a/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala new file mode 100644 index 00000000..0e9025af --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/instrumentation/akka/ActorSystemMessageInstrumentationSpec.scala @@ -0,0 +1,176 @@ +package kamon.instrumentation.akka + +import akka.actor.SupervisorStrategy.{ Escalate, Restart, Resume, Stop } +import akka.actor._ +import akka.testkit.{ TestKitBase, ImplicitSender } +import com.typesafe.config.ConfigFactory +import kamon.trace.{ EmptyTraceContext, TraceRecorder } +import org.scalatest.WordSpecLike + +import scala.concurrent.duration._ +import scala.util.control.NonFatal + +class ActorSystemMessageInstrumentationSpec extends TestKitBase with WordSpecLike with ImplicitSender { + implicit lazy val system: ActorSystem = ActorSystem("actor-system-message-instrumentation-spec", ConfigFactory.parseString( + """ + |akka.loglevel = OFF + """.stripMargin)) + + 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(EmptyTraceContext) + } + + "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(EmptyTraceContext) + } + + "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" ⇒ throw new ArithmeticException("Division by zero.") + 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-akka/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala b/kamon-akka/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala new file mode 100644 index 00000000..5c9905ba --- /dev/null +++ b/kamon-akka/src/test/scala/kamon/instrumentation/akka/AskPatternInstrumentationSpec.scala @@ -0,0 +1,134 @@ +/* + * ========================================================================================= + * 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, TestKitBase } +import akka.util.Timeout +import com.typesafe.config.ConfigFactory +import kamon.Kamon +import kamon.akka.Akka +import kamon.trace.{ TraceContext, TraceContextAware, TraceRecorder } +import org.scalatest.{ BeforeAndAfterAll, Matchers, WordSpecLike } + +import scala.concurrent.duration._ + +class AskPatternInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with BeforeAndAfterAll { + implicit lazy val system: ActorSystem = ActorSystem("ask-pattern-tracing-spec", + ConfigFactory.parseString("""akka.loggers = ["akka.event.slf4j.Slf4jLogger"]""")) + + implicit 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() { + TraceRecorder.withNewTraceContext("ask-timeout-warning") { + noReplyActorRef ? "hello" + TraceRecorder.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)) { + TraceRecorder.withNewTraceContext("ask-timeout-warning") { + noReplyActorRef ? "hello" + TraceRecorder.currentContext + } + } + } + } + + "configured in off mode" should { + "should not log any warning messages" in new NoReplyFixture { + setAskPatternTimeoutWarningMode("off") + + expectTimeoutWarning(expectWarning = false) { + TraceRecorder.withNewTraceContext("ask-timeout-warning") { + noReplyActorRef ? "hello" + TraceRecorder.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 ⇒ + } +} |