aboutsummaryrefslogtreecommitdiff
path: root/kamon-core
diff options
context:
space:
mode:
authorIvan Topolnjak <ivantopo@gmail.com>2013-08-26 22:25:08 -0300
committerIvan Topolnjak <ivantopo@gmail.com>2013-08-26 22:25:08 -0300
commit72a16d4396f5f6dc745c2376d4e50475d4714004 (patch)
tree21213af0c9baae0e84bbe1326a8e8f9d494ca8e4 /kamon-core
parentd8a8a0cc0efb79697605efed1efbaf99b98921dd (diff)
parent30ee4d88346448066c5ae0f12b683343b678577f (diff)
downloadKamon-72a16d4396f5f6dc745c2376d4e50475d4714004.tar.gz
Kamon-72a16d4396f5f6dc745c2376d4e50475d4714004.tar.bz2
Kamon-72a16d4396f5f6dc745c2376d4e50475d4714004.zip
Merge branch 'master' of github.com:dpsoft/Kamon
Diffstat (limited to 'kamon-core')
-rw-r--r--kamon-core/src/main/main.iml12
-rw-r--r--kamon-core/src/main/resources/META-INF/aop.xml3
-rw-r--r--kamon-core/src/main/resources/application.conf27
-rw-r--r--kamon-core/src/main/scala/kamon/Kamon.scala48
-rw-r--r--kamon-core/src/main/scala/kamon/TraceContext.scala21
-rw-r--r--kamon-core/src/main/scala/kamon/TraceContextSwap.scala4
-rw-r--r--kamon-core/src/main/scala/kamon/executor/eventbus.scala16
-rw-r--r--kamon-core/src/main/scala/kamon/instrumentation/ActorRefTellInstrumentation.scala60
-rw-r--r--kamon-core/src/main/scala/kamon/instrumentation/MessageQueueMetrics.scala4
-rw-r--r--kamon-core/src/main/scala/kamon/instrumentation/RunnableInstrumentation.scala4
-rw-r--r--kamon-core/src/main/scala/kamon/metric/Metrics.scala6
-rw-r--r--kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala42
-rw-r--r--kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala14
-rw-r--r--kamon-core/src/test/test.iml12
14 files changed, 157 insertions, 116 deletions
diff --git a/kamon-core/src/main/main.iml b/kamon-core/src/main/main.iml
new file mode 100644
index 00000000..702d93f1
--- /dev/null
+++ b/kamon-core/src/main/main.iml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+ <component name="NewModuleRootManager" inherit-compiler-output="true">
+ <exclude-output />
+ <content url="file://$MODULE_DIR$">
+ <sourceFolder url="file://$MODULE_DIR$/scala" isTestSource="false" />
+ </content>
+ <orderEntry type="inheritedJdk" />
+ <orderEntry type="sourceFolder" forTests="false" />
+ </component>
+</module>
+
diff --git a/kamon-core/src/main/resources/META-INF/aop.xml b/kamon-core/src/main/resources/META-INF/aop.xml
index e6d61fa1..0f427611 100644
--- a/kamon-core/src/main/resources/META-INF/aop.xml
+++ b/kamon-core/src/main/resources/META-INF/aop.xml
@@ -9,6 +9,7 @@
<aspect name="kamon.instrumentation.ActorRefTellInstrumentation"/>
<aspect name="kamon.instrumentation.ActorCellInvokeInstrumentation"/>
+ <aspect name="kamon.instrumentation.UnregisteredActorRefInstrumentation"/>
<aspect name="kamon.instrumentation.RunnableInstrumentation" />
<aspect name="kamon.instrumentation.MessageQueueInstrumentation" />
@@ -18,7 +19,7 @@
<!-- <aspect name="kamon.instrumentation.ExecutorServiceFactoryProviderInstrumentation"/>
<aspect name="kamon.instrumentation.NamedExecutorServiceFactoryDelegateInstrumentation"/>-->
<aspect name="kamon.instrumentation.ActorSystemInstrumentation"/>
- <aspect name ="kamon.instrumentation.ForkJoinPoolInstrumentation"/>
+ <!--<aspect name ="kamon.instrumentation.ForkJoinPoolInstrumentation"/>-->
diff --git a/kamon-core/src/main/resources/application.conf b/kamon-core/src/main/resources/application.conf
index 370acae9..2f8d8d87 100644
--- a/kamon-core/src/main/resources/application.conf
+++ b/kamon-core/src/main/resources/application.conf
@@ -1,4 +1,9 @@
akka {
+ loglevel = DEBUG
+ stdout-loglevel = DEBUG
+
+ extensions = ["kamon.dashboard.DashboardExtension"]
+
actor {
default-dispatcher {
fork-join-executor {
@@ -16,28 +21,12 @@ akka {
throughput = 100
}
+ debug {
+ unhandled = on
+ }
}
}
-# Dispatcher is the name of the event-based dispatcher
-#type = Dispatcher
-
-# What kind of ExecutionService to use
-#executor = "kamon.executor.InstrumentedExecutorServiceConfigurator"
-
-# Min number of threads to cap factor-based parallelism number to
-#parallelism-min = 2
-
-# Parallelism (threads) ... ceil(available processors * factor)
-#parallelism-factor = 2.0
-
-# Max number of threads to cap factor-based parallelism number to
-#parallelism-max = 10
-
-# Throughput defines the maximum number of messages to be
-# processed per actor before the thread jumps to the next actor.
-# Set to 1 for as fair as possible.
-#throughput = 100
diff --git a/kamon-core/src/main/scala/kamon/Kamon.scala b/kamon-core/src/main/scala/kamon/Kamon.scala
index c3080909..298f43eb 100644
--- a/kamon-core/src/main/scala/kamon/Kamon.scala
+++ b/kamon-core/src/main/scala/kamon/Kamon.scala
@@ -1,42 +1,17 @@
package kamon
import akka.actor.{Actor, Props, ActorSystem}
-import scala.collection.JavaConverters._
-import java.util.concurrent.ConcurrentHashMap
-import kamon.metric.{HistogramSnapshot, Histogram, Atomic, ActorSystemMetrics}
-import scala.concurrent.duration.{FiniteDuration, Duration}
+import kamon.metric.{HistogramSnapshot, ActorSystemMetrics}
+import scala.concurrent.duration.FiniteDuration
import com.newrelic.api.agent.NewRelic
+import scala.collection.concurrent.TrieMap
object Kamon {
-
- val ctx = new ThreadLocal[Option[TraceContext]] {
- override def initialValue() = None
- }
-
implicit lazy val actorSystem = ActorSystem("kamon")
-
- def context() = ctx.get()
- def clear = ctx.remove()
- def set(traceContext: TraceContext) = ctx.set(Some(traceContext))
-
- def start = set(newTraceContext)
- def stop = ctx.get match {
- case Some(context) => context.close
- case None =>
- }
-
- def newTraceContext(): TraceContext = TraceContext()
-
-
- val publisher = actorSystem.actorOf(Props[TransactionPublisher])
-
- def publish(tx: FullTransaction) = publisher ! tx
-
-
-
object Metric {
- val actorSystems = new ConcurrentHashMap[String, ActorSystemMetrics] asScala
+
+ val actorSystems = TrieMap.empty[String, ActorSystemMetrics]
def actorSystemNames: List[String] = actorSystems.keys.toList
def registerActorSystem(name: String) = actorSystems.getOrElseUpdate(name, ActorSystemMetrics(name))
@@ -44,21 +19,12 @@ object Kamon {
def actorSystem(name: String): Option[ActorSystemMetrics] = actorSystems.get(name)
}
-
-
val metricManager = actorSystem.actorOf(Props[MetricManager], "metric-manager")
val newrelicReporter = actorSystem.actorOf(Props[NewrelicReporterActor], "newrelic-reporter")
}
-
-
-
-
-
-
-
object Tracer {
val ctx = new ThreadLocal[Option[TraceContext]] {
override def initialValue() = None
@@ -68,13 +34,13 @@ object Tracer {
def clear = ctx.remove()
def set(traceContext: TraceContext) = ctx.set(Some(traceContext))
- def start = ??? //set(newTraceContext)
+ def start = set(newTraceContext)
def stop = ctx.get match {
case Some(context) => context.close
case None =>
}
- //def newTraceContext(): TraceContext = TraceContext()
+ def newTraceContext(): TraceContext = TraceContext()(Kamon.actorSystem)
}
diff --git a/kamon-core/src/main/scala/kamon/TraceContext.scala b/kamon-core/src/main/scala/kamon/TraceContext.scala
index 6b32550f..62d7f57e 100644
--- a/kamon-core/src/main/scala/kamon/TraceContext.scala
+++ b/kamon-core/src/main/scala/kamon/TraceContext.scala
@@ -1,31 +1,34 @@
package kamon
import java.util.UUID
-import akka.actor.{ActorSystem, ActorPath}
+import akka.actor._
import akka.agent.Agent
import java.util.concurrent.TimeUnit
import scala.util.{Failure, Success}
import akka.util.Timeout
-case class TraceContext(id: UUID, private val entries: Agent[List[TraceEntry]], userContext: Option[Any] = None) {
+case class TraceContext(id: UUID, entries: ActorRef, userContext: Option[Any] = None) {
implicit val timeout = Timeout(30, TimeUnit.SECONDS)
implicit val as = Kamon.actorSystem.dispatcher
- def append(entry: TraceEntry) = entries send (entry :: _)
- def close = entries.future.onComplete({
- case Success(list) => Kamon.publish(FullTransaction(id, list))
- case Failure(t) => println("WTF!")
- })
+ def append(entry: TraceEntry) = entries ! entry
+ def close = entries ! "Close" // TODO type this thing!.
}
object TraceContext {
- implicit val as2 = Kamon.actorSystem.dispatcher
- def apply()(implicit actorSystem: ActorSystem) = new TraceContext(UUID.randomUUID(), Agent[List[TraceEntry]](Nil))
+ def apply()(implicit system: ActorSystem) = new TraceContext(UUID.randomUUID(), system.actorOf(Props[TraceAccumulator])) // TODO: Move to a kamon specific supervisor, like /user/kamon/tracer
}
+class TraceAccumulator extends Actor {
+ def receive = {
+ case a => println("Trace Accumulated: "+a)
+ }
+}
+
+
trait TraceEntry
case class CodeBlockExecutionTime(name: String, begin: Long, end: Long) extends TraceEntry
diff --git a/kamon-core/src/main/scala/kamon/TraceContextSwap.scala b/kamon-core/src/main/scala/kamon/TraceContextSwap.scala
index 68ee808b..24661445 100644
--- a/kamon-core/src/main/scala/kamon/TraceContextSwap.scala
+++ b/kamon-core/src/main/scala/kamon/TraceContextSwap.scala
@@ -10,9 +10,9 @@ trait TraceContextSwap {
def withContext[A](ctx: Option[TraceContext], primary: => A, fallback: => A): A = {
ctx match {
case Some(context) => {
- Kamon.set(context)
+ Tracer.set(context)
val bodyResult = primary
- Kamon.clear
+ Tracer.clear
bodyResult
}
diff --git a/kamon-core/src/main/scala/kamon/executor/eventbus.scala b/kamon-core/src/main/scala/kamon/executor/eventbus.scala
index 599f2a7a..a1c099d4 100644
--- a/kamon-core/src/main/scala/kamon/executor/eventbus.scala
+++ b/kamon-core/src/main/scala/kamon/executor/eventbus.scala
@@ -5,7 +5,7 @@ import akka.event.LookupClassification
import akka.actor._
import java.util.concurrent.TimeUnit
-import kamon.{CodeBlockExecutionTime, Kamon, TraceContext}
+import kamon.{Tracer, CodeBlockExecutionTime, Kamon, TraceContext}
import akka.util.Timeout
import scala.util.{Random, Success, Failure}
import scala.concurrent.Future
@@ -36,7 +36,7 @@ case class Pong()
class PingActor extends Actor with ActorLogging {
- val pong = context.actorOf(Props[PongActor])
+ val pong = context.actorOf(Props[PongActor], "Pong")
val random = new Random()
def receive = {
case Pong() => {
@@ -66,14 +66,14 @@ object TryAkka extends App{
}
}))
- Kamon.start
+ Tracer.start
for(i <- 1 to 4) {
- val ping = system.actorOf(Props[PingActor])
+ val ping = system.actorOf(Props[PingActor], "Ping" + i)
ping ! Pong()
}
- def threadPrintln(body: String) = println(s"[${Thread.currentThread().getName}] - [${Kamon.context}] : $body")
+ def threadPrintln(body: String) = println(s"[${Thread.currentThread().getName}] - [${Tracer.context}] : $body")
/*
val newRelicReporter = new NewRelicReporter(registry)
@@ -86,13 +86,13 @@ object TryAkka extends App{
- Kamon.start
+ //Tracer.start
- Kamon.context.get.append(CodeBlockExecutionTime("some-block", System.nanoTime(), System.nanoTime()))
+ Tracer.context.get.append(CodeBlockExecutionTime("some-block", System.nanoTime(), System.nanoTime()))
threadPrintln("Before doing it")
val f = Future { threadPrintln("This is happening inside the future body") }
- Kamon.stop
+ Tracer.stop
//Thread.sleep(3000)
diff --git a/kamon-core/src/main/scala/kamon/instrumentation/ActorRefTellInstrumentation.scala b/kamon-core/src/main/scala/kamon/instrumentation/ActorRefTellInstrumentation.scala
index 82915ce9..7d3e36ca 100644
--- a/kamon-core/src/main/scala/kamon/instrumentation/ActorRefTellInstrumentation.scala
+++ b/kamon-core/src/main/scala/kamon/instrumentation/ActorRefTellInstrumentation.scala
@@ -3,12 +3,10 @@ package kamon.instrumentation
import org.aspectj.lang.annotation._
import org.aspectj.lang.ProceedingJoinPoint
import akka.actor.{Props, ActorSystem, ActorRef}
-import kamon.{Kamon, TraceContext}
+import kamon.{Tracer, TraceContext}
import akka.dispatch.{MessageDispatcher, Envelope}
-import com.codahale.metrics.{Timer, ExponentiallyDecayingReservoir, Histogram}
+import com.codahale.metrics.Timer
import kamon.metric.{MetricDirectory, Metrics}
-import com.codahale.metrics
-import kamon.instrumentation.TraceableMessage
import scala.Some
case class TraceableMessage(traceContext: Option[TraceContext], message: Any, timer: Timer.Context)
@@ -18,16 +16,19 @@ case class TraceableMessage(traceContext: Option[TraceContext], message: Any, ti
class ActorRefTellInstrumentation {
import ProceedingJoinPointPimp._
- @Pointcut("execution(* akka.actor.ScalaActorRef+.$bang(..)) && target(actor) && args(message, sender)")
+ val t2 = Metrics.registry.timer("some" + "LATENCY")
+
+ @Pointcut("execution(* akka.actor.ScalaActorRef+.$bang(..)) && !within(akka.event.Logging.StandardOutLogger) && !within(akka.pattern.PromiseActorRef) && !within(akka.actor.DeadLetterActorRef) && target(actor) && args(message, sender)")
def sendingMessageToActorRef(actor: ActorRef, message: Any, sender: ActorRef) = {}
@Around("sendingMessageToActorRef(actor, message, sender)")
def around(pjp: ProceedingJoinPoint, actor: ActorRef, message: Any, sender: ActorRef): Unit = {
- val actorName = MetricDirectory.nameForActor(actor)
- val t = Metrics.registry.timer(actorName + "LATENCY")
- //println(s"About to proceed with: $actor $message $sender ${Kamon.context}")
- pjp.proceedWithTarget(actor, TraceableMessage(Kamon.context, message, t.time()), sender)
+ //val actorName = MetricDirectory.nameForActor(actor)
+ //val t = Metrics.registry.timer(actorName + "LATENCY")
+ //println(s"Wrapped message from [$sender] to [$actor] with content: [$message]")
+ pjp.proceedWithTarget(actor, TraceableMessage(Tracer.context, message, t2.time()), sender)
+
}
}
@@ -48,6 +49,7 @@ class ActorCellInvokeInstrumentation {
val actorName = MetricDirectory.nameForActor(ref)
val histogramName = MetricDirectory.nameForMailbox(system.name, actorName)
+ //println("=====> Created ActorCell for: "+ref.toString())
/** TODO: Find a better way to filter the things we don't want to measure. */
//if(system.name != "kamon" && actorName.startsWith("/user")) {
processingTimeTimer = Metrics.registry.timer(histogramName + "/PROCESSINGTIME")
@@ -56,17 +58,17 @@ class ActorCellInvokeInstrumentation {
}
- @Pointcut("execution(* akka.actor.ActorCell.invoke(*)) && args(envelope)")
+ @Pointcut("(execution(* akka.actor.ActorCell.invoke(*)) || execution(* akka.routing.RoutedActorCell.sendMessage(*))) && args(envelope)")
def invokingActorBehaviourAtActorCell(envelope: Envelope) = {}
@Around("invokingActorBehaviourAtActorCell(envelope)")
def around(pjp: ProceedingJoinPoint, envelope: Envelope): Unit = {
import ProceedingJoinPointPimp._
- println("ENVELOPE --------------------->"+envelope)
+ //println("ENVELOPE --------------------->"+envelope)
envelope match {
case Envelope(TraceableMessage(ctx, msg, timer), sender) => {
- timer.stop()
+ //timer.stop()
val originalEnvelope = envelope.copy(message = msg)
@@ -74,10 +76,10 @@ class ActorCellInvokeInstrumentation {
val pt = processingTimeTimer.time()
ctx match {
case Some(c) => {
- Kamon.set(c)
- println(s"ENVELOPE ORIGINAL: [$c]---------------->"+originalEnvelope)
+ Tracer.set(c)
+ //println(s"ENVELOPE ORIGINAL: [$c]---------------->"+originalEnvelope)
pjp.proceedWith(originalEnvelope)
- Kamon.clear
+ Tracer.clear
}
case None => pjp.proceedWith(originalEnvelope)
}
@@ -87,3 +89,31 @@ class ActorCellInvokeInstrumentation {
}
}
}
+
+
+@Aspect
+class UnregisteredActorRefInstrumentation {
+ @Pointcut("execution(* akka.spray.UnregisteredActorRefBase+.handle(..)) && args(message, sender)")
+ def sprayResponderHandle(message: Any, sender: ActorRef) = {}
+
+ @Around("sprayResponderHandle(message, sender)")
+ def sprayInvokeAround(pjp: ProceedingJoinPoint, message: Any, sender: ActorRef): Unit = {
+ import ProceedingJoinPointPimp._
+ println("Handling unregistered actor ref message: "+message)
+ message match {
+ case TraceableMessage(ctx, msg, timer) => {
+ timer.stop()
+
+ ctx match {
+ case Some(c) => {
+ Tracer.set(c)
+ pjp.proceedWith(msg.asInstanceOf[AnyRef]) // TODO: define if we should use Any or AnyRef and unify with the rest of the instrumentation.
+ Tracer.clear
+ }
+ case None => pjp.proceedWith(msg.asInstanceOf[AnyRef])
+ }
+ }
+ case _ => pjp.proceed
+ }
+ }
+}
diff --git a/kamon-core/src/main/scala/kamon/instrumentation/MessageQueueMetrics.scala b/kamon-core/src/main/scala/kamon/instrumentation/MessageQueueMetrics.scala
index c21502ac..6a1e291f 100644
--- a/kamon-core/src/main/scala/kamon/instrumentation/MessageQueueMetrics.scala
+++ b/kamon-core/src/main/scala/kamon/instrumentation/MessageQueueMetrics.scala
@@ -48,12 +48,12 @@ class MonitoredMessageQueue(val delegate: MessageQueue, val queueSizeHistogram:
def enqueue(receiver: ActorRef, handle: Envelope) = {
delegate.enqueue(receiver, handle)
- queueSizeHistogram.update(numberOfMessages)
+ //queueSizeHistogram.update(numberOfMessages)
}
def dequeue(): Envelope = {
val envelope = delegate.dequeue()
- queueSizeHistogram.update(numberOfMessages)
+ //queueSizeHistogram.update(numberOfMessages)
envelope
}
diff --git a/kamon-core/src/main/scala/kamon/instrumentation/RunnableInstrumentation.scala b/kamon-core/src/main/scala/kamon/instrumentation/RunnableInstrumentation.scala
index e75a638f..30041321 100644
--- a/kamon-core/src/main/scala/kamon/instrumentation/RunnableInstrumentation.scala
+++ b/kamon-core/src/main/scala/kamon/instrumentation/RunnableInstrumentation.scala
@@ -1,7 +1,7 @@
package kamon.instrumentation
import org.aspectj.lang.annotation._
-import kamon.{Kamon, TraceContext}
+import kamon.{Tracer, TraceContext}
import org.aspectj.lang.ProceedingJoinPoint
import scala.Some
@@ -37,7 +37,7 @@ class RunnableInstrumentation {
* Aspect members
*/
- private val traceContext = Kamon.context
+ private val traceContext = Tracer.context
/**
diff --git a/kamon-core/src/main/scala/kamon/metric/Metrics.scala b/kamon-core/src/main/scala/kamon/metric/Metrics.scala
index cdc0a334..edf532ae 100644
--- a/kamon-core/src/main/scala/kamon/metric/Metrics.scala
+++ b/kamon-core/src/main/scala/kamon/metric/Metrics.scala
@@ -1,9 +1,10 @@
package kamon.metric
-import java.util.concurrent.{ConcurrentHashMap, ConcurrentSkipListSet, TimeUnit}
+import java.util.concurrent.TimeUnit
import akka.actor.ActorRef
import com.codahale.metrics
import com.codahale.metrics.{MetricFilter, Metric, ConsoleReporter, MetricRegistry}
+import scala.collection.concurrent.TrieMap
object Metrics {
@@ -85,8 +86,7 @@ trait HistogramSnapshot {
case class ActorSystemMetrics(actorSystemName: String) {
- import scala.collection.JavaConverters._
- val dispatchers = new ConcurrentHashMap[String, DispatcherMetricCollector] asScala
+ val dispatchers = TrieMap.empty[String, DispatcherMetricCollector]
private[this] def createDispatcherCollector: DispatcherMetricCollector = DispatcherMetricCollector(CodahaleHistogram(), CodahaleHistogram(), CodahaleHistogram())
diff --git a/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala b/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala
index 0026d953..ccc7740b 100644
--- a/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala
+++ b/kamon-core/src/test/scala/akka/instrumentation/ActorInstrumentationSpec.scala
@@ -1,13 +1,18 @@
package akka.instrumentation
import org.scalatest.{WordSpecLike, Matchers}
-import akka.actor.{Actor, Props, ActorSystem}
+import akka.actor.{ActorRef, Actor, Props, ActorSystem}
import akka.testkit.{ImplicitSender, TestKit}
-import kamon.{TraceContext, Kamon}
+import kamon.{TraceContext, Tracer}
+import akka.pattern.{pipe, ask}
+import akka.util.Timeout
+import scala.concurrent.duration._
+import akka.routing.RoundRobinRouter
class ActorInstrumentationSpec extends TestKit(ActorSystem("ActorInstrumentationSpec")) with WordSpecLike with Matchers with ImplicitSender {
+ implicit val executionContext = system.dispatcher
"an instrumented actor ref" when {
"used inside the context of a transaction" should {
@@ -17,28 +22,51 @@ class ActorInstrumentationSpec extends TestKit(ActorSystem("ActorInstrumentation
expectMsg(Some(testTraceContext))
}
- "propagate the trace context using tell" in {
+ "propagate the trace context using tell" in new TraceContextEchoFixture {
+ echo.tell("test", testActor)
+ expectMsg(Some(testTraceContext))
+ }
+
+ "propagate the trace context using ask" in new TraceContextEchoFixture {
+ implicit val timeout = Timeout(1 seconds)
+ (echo ? "test") pipeTo(testActor)
+
+ expectMsg(Some(testTraceContext))
}
- "propagate the trace context using ask" in {
+ "propagate the trace context to actors behind a rounter" in new RoutedTraceContextEchoFixture {
+ val contexts: Seq[Option[TraceContext]] = for(_ <- 1 to 10) yield Some(tellWithNewContext(echo, "test"))
+ expectMsgAllOf(contexts: _*)
}
}
}
trait TraceContextEchoFixture {
- val testTraceContext = Kamon.newTraceContext()
+ val testTraceContext = Tracer.newTraceContext()
val echo = system.actorOf(Props[TraceContextEcho])
- Kamon.set(testTraceContext)
+ Tracer.set(testTraceContext)
+ }
+
+ trait RoutedTraceContextEchoFixture extends TraceContextEchoFixture {
+ override val echo = system.actorOf(Props[TraceContextEcho].withRouter(RoundRobinRouter(nrOfInstances = 10)))
+
+ def tellWithNewContext(target: ActorRef, message: Any): TraceContext = {
+ val context = Tracer.newTraceContext()
+ Tracer.set(context)
+
+ target ! message
+ context
+ }
}
}
class TraceContextEcho extends Actor {
def receive = {
- case msg ⇒ sender ! Kamon.context()
+ case msg: String ⇒ sender ! Tracer.context()
}
}
diff --git a/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala b/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
index de65aaca..fe89695b 100644
--- a/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
+++ b/kamon-core/src/test/scala/kamon/instrumentation/RunnableInstrumentationSpec.scala
@@ -3,7 +3,7 @@ package kamon.instrumentation
import scala.concurrent.{Await, Promise, Future}
import org.scalatest.{Matchers, OptionValues, WordSpec}
import org.scalatest.concurrent.{ScalaFutures, PatienceConfiguration}
-import kamon.{Kamon, TraceContext}
+import kamon.{Tracer, Kamon, TraceContext}
import java.util.UUID
import scala.util.Success
import scala.concurrent.duration._
@@ -27,7 +27,7 @@ class RunnableInstrumentationSpec extends WordSpec with Matchers with ScalaFutur
val onCompleteContext = Promise[TraceContext]()
futureWithContext.onComplete({
- case _ => onCompleteContext.complete(Success(Kamon.context.get))
+ case _ => onCompleteContext.complete(Success(Tracer.context.get))
})
whenReady(onCompleteContext.future) { result =>
@@ -49,7 +49,7 @@ class RunnableInstrumentationSpec extends WordSpec with Matchers with ScalaFutur
val onCompleteContext = Promise[Option[TraceContext]]()
futureWithoutContext.onComplete({
- case _ => onCompleteContext.complete(Success(Kamon.context))
+ case _ => onCompleteContext.complete(Success(Tracer.context))
})
whenReady(onCompleteContext.future) { result =>
@@ -68,14 +68,14 @@ class RunnableInstrumentationSpec extends WordSpec with Matchers with ScalaFutur
class FutureWithContextFixture {
val testContext = TraceContext()
- Kamon.set(testContext)
+ Tracer.set(testContext)
- val futureWithContext = Future { Kamon.context}
+ val futureWithContext = Future { Tracer.context}
}
trait FutureWithoutContextFixture {
- Kamon.clear // Make sure no TraceContext is available
- val futureWithoutContext = Future { Kamon.context }
+ Tracer.clear // Make sure no TraceContext is available
+ val futureWithoutContext = Future { Tracer.context }
}
}
diff --git a/kamon-core/src/test/test.iml b/kamon-core/src/test/test.iml
new file mode 100644
index 00000000..90381726
--- /dev/null
+++ b/kamon-core/src/test/test.iml
@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+ <component name="NewModuleRootManager" inherit-compiler-output="true">
+ <exclude-output />
+ <content url="file://$MODULE_DIR$">
+ <sourceFolder url="file://$MODULE_DIR$/scala" isTestSource="true" />
+ </content>
+ <orderEntry type="inheritedJdk" />
+ <orderEntry type="sourceFolder" forTests="false" />
+ </component>
+</module>
+