Merge pull request #1051 from SimunKaracic/initial-caffeine-implementation

Initial caffeine implementation

Author: SimunKaracic

build.sbt

@@ -132,6 +132,7 @@ val instrumentationProjects = Seq[ProjectReference](
   `kamon-okhttp`,
   `kamon-tapir`,
   `kamon-redis`,
+  `kamon-caffeine`,
 )
 
 lazy val instrumentation = (project in file("instrumentation"))
@@ -516,6 +517,20 @@ lazy val `kamon-redis` = (project in file("instrumentation/kamon-redis"))
     )
   ).dependsOn(`kamon-core`, `kamon-testkit` % "test")
 
+lazy val `kamon-caffeine` = (project in file("instrumentation/kamon-caffeine"))
+  .disablePlugins(AssemblyPlugin)
+  .enablePlugins(JavaAgent)
+  .settings(instrumentationSettings)
+  .settings(
+    libraryDependencies ++= Seq(
+      kanelaAgent % "provided",
+      "com.github.ben-manes.caffeine" % "caffeine" % "2.8.5" % "provided",
+
+      scalatest % "test",
+      logbackClassic % "test",
+    )
+  ).dependsOn(`kamon-core`, `kamon-testkit` % "test")
+
 /**
  * Reporters
  */
@@ -756,7 +771,8 @@ val `kamon-bundle` = (project in file("bundle/kamon-bundle"))
     `kamon-play` % "shaded",
     `kamon-redis` % "shaded",
     `kamon-okhttp` % "shaded",
-  )
+    `kamon-caffeine` % "shaded",
+)
 
 lazy val `bill-of-materials` = (project in file("bill-of-materials"))
   .enablePlugins(BillOfMaterialsPlugin)

instrumentation/kamon-caffeine/src/main/resources/reference.conf

@@ -0,0 +1,21 @@
+# Metrics are gathered using the KamonStatsCounter, which needs to be added manually
+# e.g. Caffeine.newBuilder()
+#      	.recordStats(() -> new KamonStatsCounter("cache_name"))
+#      	.build();
+
+
+kanela.modules {
+  caffeine {
+    name = "Caffeine instrumentation"
+    description = "Provides tracing and stats for synchronous cache operations"
+
+    instrumentations = [
+      "kamon.instrumentation.caffeine.CaffeineCacheInstrumentation"
+    ]
+
+    within = [
+      "com.github.benmanes.caffeine.cache.LocalCache",
+      "com.github.benmanes.caffeine.cache.LocalManualCache",
+    ]
+  }
+}

instrumentation/kamon-caffeine/src/main/scala/kamon/instrumentation/caffeine/CaffeineCacheInstrumentation.scala

@@ -0,0 +1,50 @@
+package kamon.instrumentation.caffeine
+
+import kamon.Kamon
+import kamon.trace.Span
+import kanela.agent.api.instrumentation.InstrumentationBuilder
+import kanela.agent.libs.net.bytebuddy.asm.Advice
+
+class CaffeineCacheInstrumentation extends InstrumentationBuilder {
+  onType("com.github.benmanes.caffeine.cache.LocalCache")
+    .advise(method("computeIfAbsent"), classOf[SyncCacheAdvice])
+    .advise(method("getIfPresent"), classOf[GetIfPresentAdvice])
+
+  onType("com.github.benmanes.caffeine.cache.LocalManualCache")
+    .advise(method("getAll"), classOf[SyncCacheAdvice])
+    .advise(method("put"), classOf[SyncCacheAdvice])
+    .advise(method("getIfPresent"), classOf[GetIfPresentAdvice])
+    .advise(method("putAll"), classOf[SyncCacheAdvice])
+    .advise(method("getAllPresent"), classOf[SyncCacheAdvice])
+}
+
+class SyncCacheAdvice
+object SyncCacheAdvice {
+  @Advice.OnMethodEnter()
+  def enter(@Advice.Origin("#m") methodName: String) = {
+    Kamon.clientSpanBuilder(s"caffeine.$methodName", "caffeine").start()
+  }
+
+  @Advice.OnMethodExit(suppress = classOf[Throwable])
+  def exit(@Advice.Enter span: Span): Unit = {
+    span.finish()
+  }
+}
+
+class GetIfPresentAdvice
+object GetIfPresentAdvice {
+  @Advice.OnMethodEnter()
+  def enter(@Advice.Origin("#m") methodName: String) = {
+    Kamon.clientSpanBuilder(s"caffeine.$methodName", "caffeine").start()
+  }
+
+  @Advice.OnMethodExit(suppress = classOf[Throwable])
+  def exit(@Advice.Enter span: Span,
+           @Advice.Return ret: Any,
+           @Advice.Argument(0) key: Any): Unit = {
+    if (ret == null) {
+      span.tag("cache.miss", s"No value for key $key")
+    }
+    span.finish()
+  }
+}

instrumentation/kamon-caffeine/src/main/scala/kamon/instrumentation/caffeine/KamonStatsCounter.scala

@@ -0,0 +1,47 @@
+package kamon.instrumentation.caffeine
+
+import com.github.benmanes.caffeine.cache.RemovalCause
+import com.github.benmanes.caffeine.cache.stats.{CacheStats, StatsCounter}
+import kamon.Kamon
+
+class KamonStatsCounter(name: String) extends StatsCounter {
+  val hitsCounter = Kamon.counter(s"cache.${name}.hits").withoutTags()
+  val missesCounter = Kamon.counter(s"cache.${name}.misses").withoutTags()
+  val evictionCount = Kamon.counter(s"cache.${name}.evictions").withoutTags()
+  val loadSuccessTime = Kamon.timer(s"cache.${name}.load-time.success").withoutTags()
+  val loadFailureTime = Kamon.timer(s"cache.${name}.load-time.failure").withoutTags()
+  val evictionWeight = Kamon.counter(s"cache.${name}.eviction.weight")
+  val evictionWeightInstruments = RemovalCause.values()
+    .map(cause => cause -> evictionWeight.withTag("eviction.cause", cause.name()))
+    .toMap
+
+  override def recordHits(count: Int): Unit = hitsCounter.increment(count)
+
+  override def recordMisses(count: Int): Unit = missesCounter.increment(count)
+
+  override def recordLoadSuccess(loadTime: Long): Unit = loadSuccessTime.record(loadTime)
+
+  override def recordLoadFailure(loadTime: Long): Unit = loadFailureTime.record(loadTime)
+
+
+  override def recordEviction(): Unit = {
+    evictionCount.increment()
+  }
+
+  override def recordEviction(weight: Int): Unit = {
+    evictionCount.increment()
+    evictionWeight.withoutTags().increment(weight)
+  }
+
+  override def recordEviction(weight: Int, cause: RemovalCause): Unit = {
+    evictionCount.increment()
+    evictionWeightInstruments.get(cause).map(_.increment(weight))
+  }
+
+  /**
+    * Overrides the snapshot method and returns stubbed CacheStats.
+    * When using KamonStatsCounter, it is assumed that you are using a
+    * reporter, and are not going to be printing or logging the stats.
+    */
+  override def snapshot() = new CacheStats(0, 0, 0, 0, 0, 0, 0)
+}

instrumentation/kamon-caffeine/src/test/scala/kamon/instrumentation/caffeine/CaffeineAsyncCacheSpec.scala

@@ -0,0 +1,46 @@
+package kamon.instrumentation.caffeine
+
+import com.github.benmanes.caffeine.cache.{AsyncCache, Caffeine}
+import kamon.testkit.TestSpanReporter
+import org.scalatest.concurrent.Eventually.eventually
+import org.scalatest.concurrent.Waiters.timeout
+import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpec}
+
+import java.util
+import java.util.concurrent.CompletableFuture
+import scala.collection.JavaConverters._
+import scala.concurrent.duration.DurationInt
+
+class CaffeineAsyncCacheSpec
+  extends WordSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with TestSpanReporter {
+  "Caffeine instrumentation for async caches" should {
+    val cache: AsyncCache[String, String] = Caffeine.newBuilder()
+      .buildAsync[String, String]()
+
+    "not create a span when using put" in {
+      cache.put("a", CompletableFuture.completedFuture("key"))
+      eventually(timeout(2.seconds)) {
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+
+    "not create a span when using get" in {
+      cache.get("a", new java.util.function.Function[String, String] {
+        override def apply(a: String): String = "value"
+      })
+      eventually(timeout(2.seconds)) {
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+
+    "not create a span when using getIfPresent" in {
+      cache.getIfPresent("not_exists")
+      eventually(timeout(2.seconds)) {
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+  }
+}

instrumentation/kamon-caffeine/src/test/scala/kamon/instrumentation/caffeine/CaffeineSyncCacheSpec.scala

@@ -0,0 +1,72 @@
+package kamon.instrumentation.caffeine
+
+import com.github.benmanes.caffeine.cache.{Cache, Caffeine}
+import kamon.testkit.TestSpanReporter
+import org.scalatest.OptionValues.convertOptionToValuable
+import org.scalatest.concurrent.Eventually.eventually
+import org.scalatest.concurrent.Waiters.timeout
+import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpec}
+
+import scala.collection.JavaConverters._
+import scala.concurrent.duration.DurationInt
+
+class CaffeineSyncCacheSpec
+  extends WordSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with TestSpanReporter {
+  "Caffeine instrumentation for sync caches" should {
+    val cache: Cache[String, String] = Caffeine.newBuilder()
+      .build[String, String]()
+
+    "create a span when putting a value" in {
+      cache.put("a", "key")
+      eventually(timeout(2.seconds)) {
+        val span = testSpanReporter().nextSpan().value
+        span.operationName shouldBe "caffeine.put"
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+
+    "create a span when accessing an existing key" in {
+      cache.get("a", new java.util.function.Function[String, String] {
+        override def apply(a: String): String = "value"
+      })
+      eventually(timeout(2.seconds)) {
+        val span = testSpanReporter().nextSpan().value
+        span.operationName shouldBe "caffeine.computeIfAbsent"
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+
+    "create only one span when using putAll" in {
+      val map = Map("b" -> "value", "c" -> "value").asJava
+      cache.putAll(map)
+      eventually(timeout(2.seconds)) {
+        val span = testSpanReporter().nextSpan().value
+        span.operationName shouldBe "caffeine.putAll"
+        testSpanReporter().spans() shouldBe empty
+        testSpanReporter().clear()
+      }
+    }
+
+    "create a tagged span when accessing a key that does not exist" in {
+      cache.getIfPresent("not_exists")
+      eventually(timeout(2.seconds)) {
+        val span = testSpanReporter().nextSpan().value
+        span.operationName shouldBe "caffeine.getIfPresent"
+        span.tags.all().foreach(_.key shouldBe "cache.miss")
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+
+    "create a span when using getAllPresent" in {
+      cache.getAllPresent(Seq("a", "b").asJava)
+      eventually(timeout(2.seconds)) {
+        val span = testSpanReporter().nextSpan().value
+        span.operationName shouldBe "caffeine.getAllPresent"
+        testSpanReporter().spans() shouldBe empty
+      }
+    }
+  }
+}