diff options
7 files changed, 58 insertions, 14 deletions
diff --git a/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala b/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala index 13e506e2..d002c574 100644 --- a/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala +++ b/kamon-core/src/main/scala/akka/instrumentation/ActorMessagePassingTracing.scala @@ -44,8 +44,6 @@ class BehaviourInvokeTracing { cellWithMetrics.metricIdentity = metricIdentity cellWithMetrics.actorMetricsRecorder = metricsExtension.register(metricIdentity, ActorMetrics.Factory) - val executor = Contexts.lookupExecutionContext(Contexts.kamonDefaultDispatcher)(system) - system.scheduler.schedule(0 milliseconds, 100 milliseconds) { cellWithMetrics.actorMetricsRecorder.map { am ⇒ @@ -56,7 +54,7 @@ class BehaviourInvokeTracing { record(max) record(sum) } - }(executor) + }(metricsExtension.defaultDispatcher) } @Pointcut("(execution(* akka.actor.ActorCell.invoke(*)) || execution(* akka.routing.RoutedActorCell.sendMessage(*))) && this(cell) && args(envelope)") diff --git a/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala b/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala index 359540fc..9a08da71 100644 --- a/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala +++ b/kamon-core/src/main/scala/kamon/metrics/MetricsExtension.scala @@ -19,7 +19,7 @@ package kamon.metrics import scala.collection.concurrent.TrieMap import akka.actor._ import com.typesafe.config.Config -import kamon.util.GlobPathFilter +import kamon.util.{ Contexts, GlobPathFilter } import kamon.Kamon import akka.actor import kamon.metrics.Metrics.MetricGroupFilter @@ -76,6 +76,7 @@ class MetricsExtension(system: ExtendedActorSystem) extends Kamon.Extension { allFilters.toMap } + val defaultDispatcher = Contexts.lookupExecutionContext(Contexts.kamonDefaultDispatcher)(system) } object Metrics extends ExtensionId[MetricsExtension] with ExtensionIdProvider { diff --git a/kamon-play/src/main/scala/kamon/play/Play.scala b/kamon-play/src/main/scala/kamon/play/Play.scala index d52de450..d29c5666 100644 --- a/kamon-play/src/main/scala/kamon/play/Play.scala +++ b/kamon-play/src/main/scala/kamon/play/Play.scala @@ -18,6 +18,7 @@ package kamon.play import akka.actor.{ ExtendedActorSystem, Extension, ExtensionIdProvider, ExtensionId } import kamon.Kamon +import kamon.util.Contexts object Play extends ExtensionId[PlayExtension] with ExtensionIdProvider { override def lookup(): ExtensionId[_ <: Extension] = Play @@ -29,6 +30,7 @@ class PlayExtension(private val system: ExtendedActorSystem) extends Kamon.Exten private val config = system.settings.config.getConfig("kamon.play") + val defaultDispatcher = Contexts.lookupExecutionContext(Contexts.kamonDefaultDispatcher)(system) val includeTraceToken: Boolean = config.getBoolean("include-trace-token-header") val traceTokenHeaderName: String = config.getString("trace-token-header-name") } diff --git a/kamon-play/src/main/scala/kamon/play/instrumentation/RequestInstrumentation.scala b/kamon-play/src/main/scala/kamon/play/instrumentation/RequestInstrumentation.scala index e671d097..3a84f741 100644 --- a/kamon-play/src/main/scala/kamon/play/instrumentation/RequestInstrumentation.scala +++ b/kamon-play/src/main/scala/kamon/play/instrumentation/RequestInstrumentation.scala @@ -16,7 +16,6 @@ package kamon.play.instrumentation -import scala.concurrent.ExecutionContext.Implicits.global import kamon.trace.{ TraceRecorder, TraceContextAware } import kamon.Kamon import kamon.play.Play @@ -26,6 +25,7 @@ import akka.actor.ActorSystem import org.aspectj.lang.ProceedingJoinPoint import org.aspectj.lang.annotation._ import scala.Some +import kamon.util.Contexts @Aspect class RequestInstrumentation { @@ -63,9 +63,11 @@ class RequestInstrumentation { } private[this] val kamonRequestFilter = Filter { (nextFilter, requestHeader) ⇒ + processRequest(requestHeader) val incomingContext = TraceRecorder.currentContext + val executor = Kamon(Play)(Akka.system()).defaultDispatcher nextFilter(requestHeader).map { result ⇒ @@ -80,7 +82,7 @@ class RequestInstrumentation { } else result } simpleResult - } + }(executor) } private[this] def processRequest(requestHeader: RequestHeader): Unit = { diff --git a/kamon-play/src/main/scala/kamon/play/instrumentation/WSInstrumentation.scala b/kamon-play/src/main/scala/kamon/play/instrumentation/WSInstrumentation.scala index 421f7be2..0951d2c9 100644 --- a/kamon-play/src/main/scala/kamon/play/instrumentation/WSInstrumentation.scala +++ b/kamon-play/src/main/scala/kamon/play/instrumentation/WSInstrumentation.scala @@ -55,8 +55,7 @@ object WSInstrumentation { def basicRequestAttributes(request: WSRequest): Map[String, String] = { Map[String, String]( "host" -> request.header("host").getOrElse("Unknown"), - "path" -> request.method, - "method" -> request.method) + "path" -> request.method) } } diff --git a/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala b/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala index 4b31fb25..0c3783bb 100644 --- a/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala +++ b/kamon-play/src/test/scala/kamon/play/WSInstrumentationSpec.scala @@ -19,25 +19,67 @@ package kamon.play import play.api.mvc.Action import play.api.mvc.Results.Ok import play.api.libs.ws.WS -import org.scalatestplus.play.{ OneServerPerSuite, PlaySpec } +import org.scalatestplus.play.OneServerPerSuite import play.api.test._ import play.api.test.Helpers._ +import akka.actor.ActorSystem +import akka.testkit.{ TestKitBase, TestProbe } -class WSInstrumentationSpec extends PlaySpec with OneServerPerSuite { +import com.typesafe.config.ConfigFactory +import org.scalatest.{ Matchers, WordSpecLike } +import kamon.Kamon +import kamon.metrics.{ TraceMetrics, Metrics } +import kamon.metrics.Subscriptions.TickMetricSnapshot +import kamon.metrics.TraceMetrics.ElapsedTime + +class WSInstrumentationSpec extends TestKitBase with WordSpecLike with Matchers with OneServerPerSuite { System.setProperty("config.file", "./kamon-play/src/test/resources/conf/application.conf") + import scala.collection.immutable.StringLike._ + implicit lazy val system: ActorSystem = ActorSystem("play-ws-instrumentation-spec", ConfigFactory.parseString( + """ + |akka { + | loglevel = ERROR + |} + | + |kamon { + | metrics { + | tick-interval = 2 seconds + | + | filters = [ + | { + | trace { + | includes = [ "*" ] + | excludes = [] + | } + | } + | ] + | } + |} + """.stripMargin)) + implicit override lazy val app = FakeApplication(withRoutes = { case ("GET", "/async") ⇒ Action { Ok("ok") } }) "the WS instrumentation" should { "respond to the Async Action and complete the WS request" in { - val response = await(WS.url("http://localhost:19001/async").get()) - response.status mustBe (OK) + val metricListener = TestProbe() + Kamon(Metrics)(system).subscribe(TraceMetrics, "*", metricListener.ref, permanently = true) + metricListener.expectMsgType[TickMetricSnapshot] + + val response = await(WS.url("http://localhost:19001/async").get()) + response.status should be(OK) - //Thread.sleep(2000) //wait to complete the future + // val tickSnapshot = metricListener.expectMsgType[TickMetricSnapshot] + // val traceMetrics = tickSnapshot.metrics.find { case (k, v) ⇒ k.name.contains("async") } map (_._2.metrics) + // traceMetrics should not be empty + // + // traceMetrics map { metrics ⇒ + // metrics(ElapsedTime).numberOfMeasurements should be(1L) + // } } } }
\ No newline at end of file diff --git a/project/Projects.scala b/project/Projects.scala index 73f8f237..b3925110 100644 --- a/project/Projects.scala +++ b/project/Projects.scala @@ -74,7 +74,7 @@ object Projects extends Build { .settings(basicSettings: _*) .settings(formatSettings: _*) .settings(aspectJSettings: _*) - .settings(libraryDependencies ++= compile(playTest, aspectJ) ++ test(playTest, slf4Api)) + .settings(libraryDependencies ++= compile(playTest, aspectJ) ++ test(playTest, akkaTestKit, slf4Api)) .dependsOn(kamonCore) lazy val kamonStatsd = Project("kamon-statsd", file("kamon-statsd")) |