Merge branch 'master' into master-play-route-name-generator

Author: Yaroslav Derman

.github/workflows/ci.yml

@@ -11,6 +11,6 @@ jobs:
       - run: git fetch --depth=1 origin '+refs/tags/*:refs/tags/*'
       - uses: olafurpg/setup-scala@v10
         with:
-          java-version: [email protected]242
+          java-version: [email protected]252
       - name: Test
         run: csbt +test

build.sbt

@@ -392,6 +392,12 @@ lazy val `kamon-akka` = (project in file("instrumentation/kamon-akka"))
   )
 
 
+def akkaHttpVersion(scalaVersion: String) = scalaVersion match {
+  case "2.11" => "10.1.12"
+  case _      => "10.2.3"
+}
+
+
 lazy val `kamon-akka-http` = (project in file("instrumentation/kamon-akka-http"))
   .enablePlugins(JavaAgent)
   .disablePlugins(AssemblyPlugin)
@@ -401,18 +407,18 @@ lazy val `kamon-akka-http` = (project in file("instrumentation/kamon-akka-http")
     javaAgents += "org.mortbay.jetty.alpn" % "jetty-alpn-agent" % "2.0.10" % "test",
     libraryDependencies ++= Seq(
       kanelaAgent % "provided",
-      "com.typesafe.akka" %% "akka-http"            % "10.1.12" % "provided",
-      "com.typesafe.akka" %% "akka-http2-support"   % "10.1.12" % "provided",
-      "com.typesafe.akka" %% "akka-stream"          % "2.5.31" % "provided",
+      "com.typesafe.akka" %% "akka-http"            % akkaHttpVersion(scalaBinaryVersion.value) % "provided",
+      "com.typesafe.akka" %% "akka-http2-support"   % akkaHttpVersion(scalaBinaryVersion.value)  % "provided",
+      "com.typesafe.akka" %% "akka-stream"          % "2.5.32" % "provided",
 
       scalatest % "test",
       slf4jApi % "test",
       slf4jnop % "test",
       okHttp % "test",
-      "com.typesafe.akka" %% "akka-http-testkit"    % "10.1.12" % "test",
+      "com.typesafe.akka" %% "akka-http-testkit"    % akkaHttpVersion(scalaBinaryVersion.value) % "test",
       "de.heikoseeberger" %% "akka-http-json4s"     % "1.27.0" % "test",
       "org.json4s"        %% "json4s-native"        % "3.6.7" % "test",
-    )
+    ),
   )).dependsOn(`kamon-akka`, `kamon-testkit` % "test")
 
 

core/kamon-core-tests/src/test/scala/kamon/KamonLifecycleSpec.scala

@@ -61,5 +61,5 @@ object KamonWithTemporaryReporter extends App {
   Kamon.registerModule("dummy span reporter", new DummySpanReporter())
 
   Thread.sleep(5000)
-  Kamon.stopModules()
+  Kamon.stop()
 }

core/kamon-core/src/main/scala/kamon/Init.scala

@@ -20,11 +20,13 @@ import com.typesafe.config.Config
 import kamon.status.InstrumentationStatus
 import org.slf4j.LoggerFactory
 
+import scala.concurrent.Future
+
 /**
   * Provides APIs for handling common initialization tasks like starting modules, attaching instrumentation and
   * reconfiguring Kamon.
   */
-trait Init { self: ModuleLoading with Configuration with CurrentStatus =>
+trait Init { self: ModuleLoading with Configuration with CurrentStatus with Metrics with Tracing =>
   private val _logger = LoggerFactory.getLogger(classOf[Init])
 
   /**
@@ -44,6 +46,12 @@ trait Init { self: ModuleLoading with Configuration with CurrentStatus =>
     self.reconfigure(config)
     self.loadModules()
   }
+  
+  def stop(): Future[Unit] = {
+    self.clearRegistry()
+    self.stopTracer()
+    self.stopModules()
+  }
 
   /**
     * Tries to attach the Kanela instrumentation agent, if the Kamon Bundle dependency is available on the classpath. If

core/kamon-core/src/main/scala/kamon/Metrics.scala

@@ -32,4 +32,7 @@ trait Metrics extends MetricBuilding { self: Configuration with Utilities =>
   protected def registry(): MetricRegistry =
     _metricRegistry
 
+  protected def clearRegistry(): Unit = {
+    _metricRegistry.clear()
+  }
 }

core/kamon-core/src/main/scala/kamon/Tracing.scala

@@ -89,4 +89,7 @@ trait Tracing { self: Configuration with Utilities with ContextStorage =>
   protected def tracer(): Tracer =
     _tracer
 
+  def stopTracer(): Unit = {
+    _tracer.stop()
+  }
 }

core/kamon-core/src/main/scala/kamon/Utilities.scala

@@ -28,7 +28,7 @@ import scala.collection.concurrent.TrieMap
   */
 trait Utilities { self: Configuration =>
   private val _clock = new Clock.Anchored()
-  private val _scheduler = Executors.newScheduledThreadPool(1, numberedThreadFactory("kamon-scheduler", daemon = true))
+  private val _scheduler = newScheduledThreadPool(1, numberedThreadFactory("kamon-scheduler", daemon = true))
   private val _filters = TrieMap.empty[String, Filter]
 
   reconfigureUtilities(self.config())

core/kamon-core/src/main/scala/kamon/metric/MetricRegistry.scala

@@ -27,7 +27,6 @@ import kamon.util.Clock
 import org.slf4j.LoggerFactory
 
 import scala.collection.concurrent.TrieMap
-import scala.reflect.ClassTag
 
 /**
   * Handles creation and snapshotting of metrics. If a metric is created twice, the very same instance will be returned.
@@ -204,4 +203,8 @@ class MetricRegistry(config: Config, scheduler: ScheduledExecutorService, clock:
   /** Returns the current status of all metrics contained in the registry */
   def status(): Status.MetricRegistry =
     Status.MetricRegistry(_metrics.values.map(_.status()).toSeq)
+
+  def clear(): Unit = {
+    _metrics.values.foreach { metric => metric.status().instruments.foreach(i => metric.remove(i.tags)) }
+  }
 }

core/kamon-core/src/main/scala/kamon/module/ModuleRegistry.scala

@@ -18,9 +18,8 @@ package kamon
 package module
 
 import java.time.{Duration, Instant}
-import java.util.concurrent.{CountDownLatch, Executors, ScheduledFuture, TimeUnit}
+import java.util.concurrent.{CountDownLatch, Executors, ForkJoinPool, ScheduledFuture, TimeUnit}
 import java.util.concurrent.atomic.AtomicReference
-
 import com.typesafe.config.Config
 import kamon.module.Module.Registration
 import kamon.status.Status
@@ -39,8 +38,9 @@ import scala.util.control.NonFatal
 class ModuleRegistry(configuration: Configuration, clock: Clock, metricRegistry: MetricRegistry, tracer: Tracer) {
 
   private val _logger = LoggerFactory.getLogger(classOf[ModuleRegistry])
-  private val _metricsTickerExecutor = Executors.newScheduledThreadPool(1, threadFactory("kamon-metrics-ticker", daemon = true))
-  private val _spansTickerExecutor = Executors.newScheduledThreadPool(1, threadFactory("kamon-spans-ticker", daemon = true))
+  private val _moduleRegistryEC: ExecutionContext = ExecutionContext.fromExecutor(newScheduledThreadPool(1, threadFactory("kamon-module-registry", daemon = true)))
+  private val _metricsTickerExecutor = newScheduledThreadPool(1, threadFactory("kamon-metrics-ticker", daemon = true))
+  private val _spansTickerExecutor = newScheduledThreadPool(1, threadFactory("kamon-spans-ticker", daemon = true))
 
   private val _metricsTickerSchedule = new AtomicReference[ScheduledFuture[_]]()
   private val _spansTickerSchedule = new AtomicReference[ScheduledFuture[_]]()
@@ -127,7 +127,7 @@ class ModuleRegistry(configuration: Configuration, clock: Clock, metricRegistry:
     * spans available until the call to stop.
     */
   def stopModules(): Future[Unit] = synchronized {
-    implicit val cleanupExecutor = ExecutionContext.Implicits.global
+    implicit val cleanupExecutor = _moduleRegistryEC
     stopReporterTickers()
 
     var stoppedSignals: List[Future[Unit]] = Nil
@@ -137,11 +137,7 @@ class ModuleRegistry(configuration: Configuration, clock: Clock, metricRegistry:
         true
     }
 
-    val latch = new CountDownLatch(stoppedSignals.size)
-    stoppedSignals.foreach(f => f.onComplete(_ => latch.countDown()))
-
-    // TODO: Completely destroy modules that fail to stop within the 30 second timeout.
-    Future(latch.await(30, TimeUnit.SECONDS))
+    Future.sequence(stoppedSignals).map(_ => ())
   }
 
 
@@ -369,7 +365,7 @@ class ModuleRegistry(configuration: Configuration, clock: Clock, metricRegistry:
     * context. The returned future completes when the module finishes its stop procedure.
     */
   private def stopModule(entry: Entry[Module]): Future[Unit] = synchronized {
-    val cleanupExecutor = ExecutionContext.Implicits.global
+    val cleanupExecutor = _moduleRegistryEC
 
     // Remove the module from all registries
     _registeredModules = _registeredModules - entry.name

core/kamon-core/src/main/scala/kamon/package.scala

@@ -15,14 +15,25 @@
  */
 
 import java.util.concurrent.atomic.AtomicLong
-import java.util.concurrent.{Executors, ThreadFactory}
-
+import java.util.concurrent.{Executors, ScheduledExecutorService, ScheduledThreadPoolExecutor, ThreadFactory, TimeUnit}
 import com.typesafe.config.{Config, ConfigUtil}
 
 import scala.collection.concurrent.TrieMap
 
 package object kamon {
 
+  def newScheduledThreadPool(corePoolSize: Int, threadFactory: ThreadFactory): ScheduledExecutorService = {
+    val tp = new ScheduledThreadPoolExecutor(corePoolSize, threadFactory)
+    val timeout = 5000
+    // Call in this order or it throws!
+    tp.setKeepAliveTime(timeout, TimeUnit.MILLISECONDS)
+    tp.allowCoreThreadTimeOut(true)
+
+    tp.setRemoveOnCancelPolicy(true)
+
+    tp
+  }
+
   /**
     * Creates a thread factory that assigns the specified name to all created Threads.
     */