diff options
Diffstat (limited to 'kamon-jdbc/src/main/scala/kamon')
-rw-r--r-- | kamon-jdbc/src/main/scala/kamon/jdbc/Jdbc.scala | 14 | ||||
-rw-r--r-- | kamon-jdbc/src/main/scala/kamon/jdbc/instrumentation/StatementInstrumentation.scala | 18 |
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 = { |