1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
/* =========================================================================================
* Copyright © 2013-2014 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.play.instrumentation
import kamon.Kamon
import kamon.play.{ Play, PlayExtension }
import kamon.trace.TraceLocal.{ HttpContextKey, HttpContext }
import kamon.trace._
import org.aspectj.lang.ProceedingJoinPoint
import org.aspectj.lang.annotation._
import play.api.Routes
import play.api.mvc.Results._
import play.api.mvc._
import play.libs.Akka
@Aspect
class RequestInstrumentation {
import RequestInstrumentation._
@DeclareMixin("play.api.mvc.RequestHeader+")
def mixinContextAwareNewRequest: TraceContextAware = TraceContextAware.default
@After("call(* play.api.GlobalSettings.onStart(*)) && args(application)")
def afterApplicationStart(application: play.api.Application): Unit = {
Kamon(Play)
}
@Before("call(* play.api.GlobalSettings.onRouteRequest(..)) && args(requestHeader)")
def beforeRouteRequest(requestHeader: RequestHeader): Unit = {
import Kamon.tracer
val playExtension = Kamon(Play)
val defaultTraceName = playExtension.generateTraceName(requestHeader)
val token = if (playExtension.includeTraceToken) {
requestHeader.headers.toSimpleMap.find(_._1 == playExtension.traceTokenHeaderName).map(_._2)
} else None
val newContext = token.map(t ⇒ tracer.newContext(defaultTraceName, t)).getOrElse(tracer.newContext(defaultTraceName))
TraceContext.setCurrentContext(newContext)
}
@Around("call(* play.api.GlobalSettings.doFilter(*)) && args(next)")
def aroundDoFilter(pjp: ProceedingJoinPoint, next: EssentialAction): Any = {
val essentialAction = (requestHeader: RequestHeader) ⇒ {
val playExtension = Kamon(Play)
val executor = playExtension.defaultDispatcher
def onResult(result: Result): Result = {
TraceContext.map { ctx ⇒
ctx.finish()
recordHttpServerMetrics(result.header, ctx.name)
if (playExtension.includeTraceToken) result.withHeaders(playExtension.traceTokenHeaderName -> ctx.token)
else result
} getOrElse result
}
//store in TraceLocal useful data to diagnose errors
storeDiagnosticData(requestHeader)
//override the current trace name
normaliseTraceName(requestHeader).map(TraceContext.currentContext.rename)
// Invoke the action
next(requestHeader).map(onResult)(executor)
}
pjp.proceed(Array(EssentialAction(essentialAction)))
}
@Before("call(* play.api.GlobalSettings.onError(..)) && args(request, ex)")
def beforeOnError(request: TraceContextAware, ex: Throwable): Unit = {
TraceContext.map { ctx ⇒
recordHttpServerMetrics(InternalServerError.header, ctx.name)
}
}
def recordHttpServerMetrics(header: ResponseHeader, traceName: String): Unit =
Kamon(Play).httpServerMetrics.recordResponse(traceName, header.status.toString)
def storeDiagnosticData(request: RequestHeader): Unit = {
val agent = request.headers.get(UserAgent).getOrElse(Unknown)
val forwarded = request.headers.get(XForwardedFor).getOrElse(Unknown)
TraceLocal.store(HttpContextKey)(HttpContext(agent, request.uri, forwarded))
}
}
object RequestInstrumentation {
import kamon.util.TriemapAtomicGetOrElseUpdate.Syntax
import java.util.Locale
import scala.collection.concurrent.TrieMap
val UserAgent = "User-Agent"
val XForwardedFor = "X-Forwarded-For"
val Unknown = "unknown"
private val cache = TrieMap.empty[String, String]
def normaliseTraceName(requestHeader: RequestHeader): Option[String] = requestHeader.tags.get(Routes.ROUTE_VERB).map({ verb ⇒
val path = requestHeader.tags(Routes.ROUTE_PATTERN)
cache.atomicGetOrElseUpdate(s"$verb$path", {
val traceName = {
// Convert paths of form GET /foo/bar/$paramname<regexp>/blah to foo.bar.paramname.blah.get
val p = path.replaceAll("""\$([^<]+)<[^>]+>""", "$1").replace('/', '.').dropWhile(_ == '.')
val normalisedPath = {
if (p.lastOption.filter(_ != '.').isDefined) s"$p."
else p
}
s"$normalisedPath${verb.toLowerCase(Locale.ENGLISH)}"
}
traceName
})
})
}
|
