[GitHub] vvraskin closed pull request #4165: Update to Kamon 1.1.3 from 0.6 series

[GitBox <gi...@apache.org>]

vvraskin closed pull request #4165: Update to Kamon 1.1.3 from 0.6 series
URL: https://github.com/apache/incubator-openwhisk/pull/4165
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/common/scala/build.gradle b/common/scala/build.gradle
index d7b849741c..0a040b7576 100644
--- a/common/scala/build.gradle
+++ b/common/scala/build.gradle
@@ -64,8 +64,8 @@ dependencies {
     compile 'com.github.ben-manes.caffeine:caffeine:2.6.2'
     compile 'com.google.code.findbugs:jsr305:3.0.2'
     compile 'io.fabric8:kubernetes-client:4.0.3'
-    compile 'io.kamon:kamon-core_2.12:0.6.7'
-    compile 'io.kamon:kamon-statsd_2.12:0.6.7'
+    compile 'io.kamon:kamon-core_2.12:1.1.3'
+    compile 'io.kamon:kamon-statsd_2.12:1.0.0'
     //for mesos
     compile 'com.adobe.api.platform.runtime:mesos-actor:0.0.17'
 
diff --git a/common/scala/src/main/resources/application.conf b/common/scala/src/main/resources/application.conf
index 40c98a70c2..40049f8a9f 100644
--- a/common/scala/src/main/resources/application.conf
+++ b/common/scala/src/main/resources/application.conf
@@ -19,7 +19,11 @@ akka.http {
 
 #kamon related configuration
 kamon {
-
+    environment {
+      # Identifier for this service. For keeping it backward compatible setting to natch previous
+      # statsd name
+      service = "openwhisk-statsd"
+    }
     metric {
         tick-interval = 1 second
     }
@@ -40,17 +44,12 @@ kamon {
             dispatcher =  [ "*" ]
         }
 
-        simple-metric-key-generator {
-            # Application prefix for all metrics pushed to StatsD. The default namespacing scheme for metrics follows
-            # this pattern:
-            #    application.host.entity.entity-name.metric-name
-            application = "openwhisk-statsd"
-        }
+        metric-key-generator = org.apache.openwhisk.common.WhiskStatsDMetricKeyGenerator
     }
 
-    modules {
-        kamon-statsd.auto-start = yes
-    }
+    reporters = [
+        "kamon.statsd.StatsDReporter"
+    ]
 }
 
 whisk {
diff --git a/common/scala/src/main/scala/org/apache/openwhisk/common/Logging.scala b/common/scala/src/main/scala/org/apache/openwhisk/common/Logging.scala
index 3bc3597e69..b7cff4e3d2 100644
--- a/common/scala/src/main/scala/org/apache/openwhisk/common/Logging.scala
+++ b/common/scala/src/main/scala/org/apache/openwhisk/common/Logging.scala
@@ -20,9 +20,11 @@ package org.apache.openwhisk.common
 import java.io.PrintStream
 import java.time.{Clock, Instant, ZoneId}
 import java.time.format.DateTimeFormatter
+
 import akka.event.Logging._
 import akka.event.LoggingAdapter
 import kamon.Kamon
+import kamon.statsd.{MetricKeyGenerator, SimpleMetricKeyGenerator}
 import org.apache.openwhisk.core.entity.ControllerInstanceId
 
 trait Logging {
@@ -210,17 +212,15 @@ object LogMarkerToken {
 }
 
 object MetricEmitter {
-
-  val metrics = Kamon.metrics
-
   def emitCounterMetric(token: LogMarkerToken, times: Long = 1): Unit = {
     if (TransactionId.metricsKamon) {
       if (TransactionId.metricsKamonTags) {
-        metrics
-          .counter(token.toString, token.tags)
+        Kamon
+          .counter(createName(token.toString, "counter"))
+          .refine(token.tags)
           .increment(times)
       } else {
-        metrics.counter(token.toStringWithSubAction).increment(times)
+        Kamon.counter(createName(token.toStringWithSubAction, "counter")).increment(times)
       }
     }
   }
@@ -228,14 +228,40 @@ object MetricEmitter {
   def emitHistogramMetric(token: LogMarkerToken, value: Long): Unit = {
     if (TransactionId.metricsKamon) {
       if (TransactionId.metricsKamonTags) {
-        metrics
-          .histogram(token.toString, token.tags)
+        Kamon
+          .histogram(createName(token.toString, "histogram"))
+          .refine(token.tags)
           .record(value)
       } else {
-        metrics.histogram(token.toStringWithSubAction).record(value)
+        Kamon.histogram(createName(token.toStringWithSubAction, "histogram")).record(value)
       }
     }
   }
+
+  /**
+   * Kamon 1.0 onwards does not include the metric type in the metric name which cause issue
+   * for us as we use same metric name for counter and histogram. So to be backward compatible we
+   * need to prefix the name with type
+   */
+  private def createName(name: String, metricType: String) = {
+    s"$metricType.$name"
+  }
+}
+
+/**
+ * Name generator to make names compatible to pre Kamon 1.0 logic. Statsd reporter "normalizes"
+ * the key name by replacing all `.` with `_`. Pre 1.0 the metric category was added by Statsd
+ * reporter itself. However now we pass it explicitly. So to retain the pre 1.0 name we need to replace
+ * normalized name with one having category followed by `.` instead of `_`
+ */
+class WhiskStatsDMetricKeyGenerator(config: com.typesafe.config.Config) extends MetricKeyGenerator {
+  val simpleGen = new SimpleMetricKeyGenerator(config)
+  override def generateKey(name: String, tags: Map[String, String]): String = {
+    val key = simpleGen.generateKey(name, tags)
+    if (key.contains(".counter_")) key.replace(".counter_", ".counter.")
+    else if (key.contains(".histogram_")) key.replace(".histogram_", ".histogram.")
+    else key
+  }
 }
 
 object LoggingMarkers {
diff --git a/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala b/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala
index 1cb38ee478..5136a66b40 100644
--- a/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala
+++ b/core/controller/src/main/scala/org/apache/openwhisk/core/controller/Controller.scala
@@ -42,8 +42,9 @@ import org.apache.openwhisk.core.loadBalancer.{InvokerState, LoadBalancerProvide
 import org.apache.openwhisk.http.{BasicHttpService, BasicRasService}
 import org.apache.openwhisk.spi.SpiLoader
 
+import scala.concurrent.ExecutionContext.Implicits
 import scala.concurrent.duration.DurationInt
-import scala.concurrent.{Await, Future}
+import scala.concurrent.Await
 import scala.util.{Failure, Success}
 
 /**
@@ -206,15 +207,14 @@ object Controller {
       "runtimes" -> runtimes.toJson)
 
   def main(args: Array[String]): Unit = {
-    Kamon.start()
+    Kamon.loadReportersFromConfig()
     implicit val actorSystem = ActorSystem("controller-actor-system")
     implicit val logger = new AkkaLogging(akka.event.Logging.getLogger(actorSystem, this))
 
     // Prepare Kamon shutdown
     CoordinatedShutdown(actorSystem).addTask(CoordinatedShutdown.PhaseActorSystemTerminate, "shutdownKamon") { () =>
       logger.info(this, s"Shutting down Kamon with coordinated shutdown")
-      Kamon.shutdown()
-      Future.successful(Done)
+      Kamon.stopAllReporters().map(_ => Done)(Implicits.global)
     }
 
     // extract configuration data from the environment
diff --git a/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala b/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala
index 14ca32b63b..2f35afbf81 100644
--- a/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala
+++ b/core/invoker/src/main/scala/org/apache/openwhisk/core/invoker/Invoker.scala
@@ -37,7 +37,7 @@ import org.apache.openwhisk.spi.SpiLoader
 import org.apache.openwhisk.utils.ExecutionContextFactory
 
 import scala.concurrent.duration._
-import scala.concurrent.{Await, Future}
+import scala.concurrent.Await
 import scala.util.{Failure, Try}
 
 case class CmdLineArgs(uniqueName: Option[String] = None, id: Option[Int] = None, displayedName: Option[String] = None)
@@ -59,13 +59,12 @@ object Invoker {
   def initKamon(instance: Int): Unit = {
     // Replace the hostname of the invoker to the assigned id of the invoker.
     val newKamonConfig = Kamon.config
-      .withValue(
-        "kamon.statsd.simple-metric-key-generator.hostname-override",
-        ConfigValueFactory.fromAnyRef(s"invoker$instance"))
-    Kamon.start(newKamonConfig)
+      .withValue("kamon.environment.host", ConfigValueFactory.fromAnyRef(s"invoker$instance"))
+    Kamon.reconfigure(newKamonConfig)
   }
 
   def main(args: Array[String]): Unit = {
+    Kamon.loadReportersFromConfig()
     implicit val ec = ExecutionContextFactory.makeCachedThreadPoolExecutionContext()
     implicit val actorSystem: ActorSystem =
       ActorSystem(name = "invoker-actor-system", defaultExecutionContext = Some(ec))
@@ -75,8 +74,7 @@ object Invoker {
     // Prepare Kamon shutdown
     CoordinatedShutdown(actorSystem).addTask(CoordinatedShutdown.PhaseActorSystemTerminate, "shutdownKamon") { () =>
       logger.info(this, s"Shutting down Kamon with coordinated shutdown")
-      Kamon.shutdown()
-      Future.successful(Done)
+      Kamon.stopAllReporters().map(_ => Done)
     }
 
     // load values for the required properties from the environment


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services