aboutsummaryrefslogtreecommitdiff
path: root/kamon-jdbc/src/main/scala
diff options
context:
space:
mode:
Diffstat (limited to 'kamon-jdbc/src/main/scala')
-rw-r--r--kamon-jdbc/src/main/scala/kamon/jdbc/Jdbc.scala14
-rw-r--r--kamon-jdbc/src/main/scala/kamon/jdbc/instrumentation/StatementInstrumentation.scala18
2 files changed, 29 insertions, 3 deletions
diff --git a/kamon-jdbc/src/main/scala/kamon/jdbc/Jdbc.scala b/kamon-jdbc/src/main/scala/kamon/jdbc/Jdbc.scala
index 9de65821..036789b7 100644
--- a/kamon-jdbc/src/main/scala/kamon/jdbc/Jdbc.scala
+++ b/kamon-jdbc/src/main/scala/kamon/jdbc/Jdbc.scala
@@ -23,11 +23,16 @@ import kamon.Kamon
object Jdbc extends ExtensionId[JdbcExtension] with ExtensionIdProvider {
override def lookup(): ExtensionId[_ <: Extension] = Jdbc
override def createExtension(system: ExtendedActorSystem): JdbcExtension = new JdbcExtension(system)
+
+ val SegmentLibraryName = "jdbc"
}
class JdbcExtension(system: ExtendedActorSystem) extends Kamon.Extension {
private val config = system.settings.config.getConfig("kamon.jdbc")
+ private val nameGeneratorFQN = config.getString("name-generator")
+ private val nameGenerator: JdbcNameGenerator = system.dynamicAccess.createInstanceFor[JdbcNameGenerator](nameGeneratorFQN, Nil).get
+
private val slowQueryProcessorClass = config.getString("slow-query-processor")
private val slowQueryProcessor: SlowQueryProcessor = system.dynamicAccess.createInstanceFor[SlowQueryProcessor](slowQueryProcessorClass, Nil).get
@@ -38,6 +43,7 @@ class JdbcExtension(system: ExtendedActorSystem) extends Kamon.Extension {
def processSlowQuery(sql: String, executionTime: Long) = slowQueryProcessor.process(sql, executionTime, slowQueryThreshold)
def processSqlError(sql: String, ex: Throwable) = sqlErrorProcessor.process(sql, ex)
+ def generateJdbcSegmentName(statement: String): String = nameGenerator.generateJdbcSegmentName(statement)
}
trait SlowQueryProcessor {
@@ -48,6 +54,14 @@ trait SqlErrorProcessor {
def process(sql: String, ex: Throwable): Unit
}
+trait JdbcNameGenerator {
+ def generateJdbcSegmentName(statement: String): String
+}
+
+class DefaultJdbcNameGenerator extends JdbcNameGenerator {
+ def generateJdbcSegmentName(statement: String): String = s"Jdbc[$statement]"
+}
+
class DefaultSqlErrorProcessor extends SqlErrorProcessor {
import org.slf4j.LoggerFactory
diff --git a/kamon-jdbc/src/main/scala/kamon/jdbc/instrumentation/StatementInstrumentation.scala b/kamon-jdbc/src/main/scala/kamon/jdbc/instrumentation/StatementInstrumentation.scala
index 41eef716..b2f35c6c 100644
--- a/kamon-jdbc/src/main/scala/kamon/jdbc/instrumentation/StatementInstrumentation.scala
+++ b/kamon-jdbc/src/main/scala/kamon/jdbc/instrumentation/StatementInstrumentation.scala
@@ -23,7 +23,7 @@ import kamon.jdbc.Jdbc
import kamon.jdbc.metric.StatementsMetrics
import kamon.jdbc.metric.StatementsMetricsGroupFactory.GroupRecorder
import kamon.metric.Metrics
-import kamon.trace.TraceRecorder
+import kamon.trace.{ TraceContext, SegmentCategory, TraceRecorder }
import org.aspectj.lang.ProceedingJoinPoint
import org.aspectj.lang.annotation.{ Around, Aspect, Pointcut }
import org.slf4j.LoggerFactory
@@ -51,8 +51,10 @@ class StatementInstrumentation {
implicit val statementRecorder: Option[GroupRecorder] = Kamon(Metrics)(system).register(StatementsMetrics(Statements), StatementsMetrics.Factory)
sql.replaceAll(CommentPattern, Empty) match {
- case SelectStatement(_) ⇒ recordRead(pjp, sql, system)
- case InsertStatement(_) | UpdateStatement(_) | DeleteStatement(_) ⇒ recordWrite(pjp, sql, system)
+ case SelectStatement(_) ⇒ withSegment(ctx, system, Select)(recordRead(pjp, sql, system))
+ case InsertStatement(_) ⇒ withSegment(ctx, system, Insert)(recordWrite(pjp, sql, system))
+ case UpdateStatement(_) ⇒ withSegment(ctx, system, Update)(recordWrite(pjp, sql, system))
+ case DeleteStatement(_) ⇒ withSegment(ctx, system, Delete)(recordWrite(pjp, sql, system))
case anythingElse ⇒
log.debug(s"Unable to parse sql [$sql]")
pjp.proceed()
@@ -65,6 +67,12 @@ class StatementInstrumentation {
try thunk finally timeSpent(System.nanoTime() - start)
}
+ def withSegment[A](ctx: TraceContext, system: ActorSystem, statement: String)(thunk: ⇒ A): A = {
+ val segmentName = Jdbc(system).generateJdbcSegmentName(statement)
+ val segment = ctx.startSegment(segmentName, SegmentCategory.Database, Jdbc.SegmentLibraryName)
+ try thunk finally segment.finish()
+ }
+
def recordRead(pjp: ProceedingJoinPoint, sql: String, system: ActorSystem)(implicit statementRecorder: Option[GroupRecorder]): Any = {
withTimeSpent(pjp.proceedWithErrorHandler(sql, system)) { timeSpent ⇒
statementRecorder.map(stmr ⇒ stmr.reads.record(timeSpent))
@@ -95,6 +103,10 @@ object StatementInstrumentation {
val CommentPattern = "/\\*.*?\\*/" //for now only removes comments of kind / * anything * /
val Empty = ""
val Statements = "jdbc-statements"
+ val Select = "Select"
+ val Insert = "Insert"
+ val Update = "Update"
+ val Delete = "Delete"
implicit class PimpedProceedingJoinPoint(pjp: ProceedingJoinPoint) {
def proceedWithErrorHandler(sql: String, system: ActorSystem)(implicit statementRecorder: Option[GroupRecorder]): Any = {