Feature/instrument process node

.sbtopts

@@ -0,0 +1,4 @@
+-J-XX:MaxMetaspaceSize=2048m
+-J-Xmx4096m
+-J-XX:+CMSClassUnloadingEnabled
+-J-XX:+UseConcMarkSweepGC

.travis.yml

@@ -1,12 +1,12 @@
 language: scala
+sbt_args: -Djava.io.tmpdir=$TRAVIS_BUILD_DIR/tmp
 script:
-  - sbt test
+  - sbt +test
 scala:
-   - 2.12.8
+   - 2.12.9
 jdk:
-   - oraclejdk8
+   - openjdk8
 before_script:
   - mkdir $TRAVIS_BUILD_DIR/tmp
-  - export SBT_OPTS="-Djava.io.tmpdir=$TRAVIS_BUILD_DIR/tmp"
 sudo: false
 

README.md

@@ -14,19 +14,19 @@ the requests you are doing are behaving.
 
 ### Getting Started
 
-Kamon Cassandra module is currently available for Scala 2.11 and 2.12.
+Kamon Kafka module is currently available for Scala 2.11 and 2.12.
 
 Supported releases and dependencies are shown below.
 
 | kamon-kafka  | status | jdk  | scala            
 |:------:|:------:|:----:|------------------
-|  1.0.0 | stable | 1.8+ | 2.11, 2.12  
+|  2.0.0 | snapshot | 1.8+ | 2.11, 2.12  
 
 To get started with SBT, simply add the following to your `build.sbt`
 file:
 
 ```scala
-libraryDependencies += "io.kamon" %% "kamon-kafka" % "1.0.0"
+libraryDependencies += "io.kamon" %% "kamon-kafka" % "2.0.0-SNAPSHOT"
 ```
 
 

build.sbt

@@ -13,12 +13,13 @@
  * =========================================================================================
  */
 
-val kamonCore           = "io.kamon"            %% "kamon-core"                       % "1.1.3"
-val kamonTestkit        = "io.kamon"            %% "kamon-testkit"                    % "1.1.3"
-val scalaExtension      = "io.kamon"            %% "kanela-scala-extension"           % "0.0.10"
+val kamonCore           = "io.kamon"            %% "kamon-core"                       % "2.0.0"
+val kamonTestkit        = "io.kamon"            %% "kamon-testkit"                    % "2.0.0"
+val scalaExtension      = "io.kamon"            %% "kamon-instrumentation-common"     % "2.0.0"
 
 val kafkaClient         = "org.apache.kafka"    % "kafka-clients"	                    % "0.11.0.0"
 val kafkaStreams        = "org.apache.kafka"    % "kafka-streams"	                    % "2.0.0"
+val kafkaStreamsScala   = "org.apache.kafka"    %% "kafka-streams-scala"	            % "2.0.0"
 
 val kafkaTest           = "net.manub"           %% "scalatest-embedded-kafka"         % "2.0.0"
 val kafkaStreamTest     = "net.manub"           %% "scalatest-embedded-kafka-streams" % "2.0.0"
@@ -34,9 +35,9 @@ lazy val kafkaClients = (project in file("kamon-kafka-clients"))
   .enablePlugins(JavaAgent)
   .settings(bintrayPackage := "kamon-kafka")
   .settings(name := "kamon-kafka-clients")
-  .settings(scalaVersion := "2.12.8")
-  .settings(crossScalaVersions := Seq("2.11.12", "2.12.8"))
-  .settings(javaAgents += "io.kamon" % "kanela-agent" % "0.0.15" % "compile;test")
+  .settings(scalaVersion := "2.12.9")
+  .settings(crossScalaVersions := Seq("2.11.12", "2.12.9"))
+  .settings(javaAgents += "io.kamon" % "kanela-agent" % "1.0.1" % "compile;test")
   .settings(resolvers += Resolver.bintrayRepo("kamon-io", "snapshots"))
   .settings(resolvers += Resolver.mavenLocal)
   .settings(
@@ -50,13 +51,13 @@ lazy val kafkaStream = (project in file("kamon-kafka-streams"))
   .dependsOn(kafkaClients % "compile->compile;test->test")
   .settings(bintrayPackage := "kamon-kafka")
   .settings(name := "kamon-kafka-streams")
-  .settings(scalaVersion := "2.12.8")
-  .settings(crossScalaVersions := Seq("2.11.12", "2.12.8"))
-  .settings(javaAgents += "io.kamon"    % "kanela-agent"   % "0.0.15"  % "compile;test")
+  .settings(scalaVersion := "2.12.9")
+  .settings(crossScalaVersions := Seq("2.11.12", "2.12.9"))
+  .settings(javaAgents += "io.kamon"    % "kanela-agent"   % "1.0.1"  % "compile;test")
   .settings(resolvers += Resolver.bintrayRepo("kamon-io", "snapshots"))
   .settings(resolvers += Resolver.mavenLocal)
   .settings(
         libraryDependencies ++=
-          compileScope(kamonCore, kafkaStreams, scalaExtension) ++
+          compileScope(kamonCore, kafkaStreams, kafkaStreamsScala, scalaExtension) ++
             providedScope(lombok) ++
             testScope(kamonTestkit, scalatest, slf4jApi, logbackClassic, kafkaStreamTest))

kamon-kafka-clients/src/main/java/kamon/kafka/client/instrumentation/advisor/Advisors.java

@@ -1,6 +1,5 @@
-/*
- * =========================================================================================
- * Copyright © 2013-2019 the kamon project <http://kamon.io/>
+/* =========================================================================================
+ * Copyright (C) 2013-2019 the kamon project <http://kamon.io/>
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
  * except in compliance with the License. You may obtain a copy of the License at
@@ -28,6 +27,8 @@
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.ProducerRecord;
 
+import java.io.ByteArrayOutputStream;
+
 @Value
 public class Advisors {
     /**
@@ -48,23 +49,26 @@ public static class SendMethodAdvisor {
         public static void onEnter(@Advice.Argument(value = 0, readOnly = false) ProducerRecord record,
                                    @Advice.Argument(value = 1, readOnly = false) Callback callback,
                                    @Advice.Local("scope") Storage.Scope scope) {
-
             val currentContext = Kamon.currentContext();
             val topic = record.topic() == null ? "kafka" : record.topic();
             val partition = record.partition() == null ? "unknown-partition" : record.partition().toString();
             val value = record.key() == null ? "unknown-key" : record.key().toString();
 
-            val span = Kamon.buildSpan("kafka.produce")
-                    .asChildOf(currentContext.get(Span.ContextKey()))
-                    .withMetricTag("span.kind", "producer")
-                    .withTag("kafka.key", value)
-                    .withTag("kafka.partition", partition)
-                    .withTag("kafka.topic", topic)
+            val span = Kamon.producerSpanBuilder("kafka.produce", "kafkaPublisher")
+                    .asChildOf(currentContext.get(Span.Key()))
+                    .tag("span.kind", "producer")
+                    .tag("kafka.key", value)
+                    .tag("kafka.partition", partition)
+                    .tag("kafka.topic", topic)
                     .start();
 
-            val ctx = currentContext.withKey(Span.ContextKey(), span);
+            val ctx = currentContext.withEntry(Span.Key(), span);
+
+            val out = new ByteArrayOutputStream();
+            // ugly, ugly, ugly ... :(
+            Kamon.defaultBinaryPropagation().write(ctx, kamon.context.BinaryPropagation$ByteStreamWriter$.MODULE$.of(out));
 
-            record.headers().add("kamon-context", Kamon.contextCodec().Binary().encode(ctx).array());
+            record.headers().add("kamon-context", out.toByteArray());
 
             scope = Kamon.storeContext(ctx);
             callback = new ProducerCallback(callback, scope);
@@ -75,8 +79,8 @@ public static void onEnter(@Advice.Argument(value = 0, readOnly = false) Produce
         public static void onExit(@Advice.Local("scope") Storage.Scope scope,
                                   @Advice.Thrown final Throwable throwable) {
 
-            val span = scope.context().get(Span.ContextKey());
-            if (throwable != null) span.addError(throwable.getMessage(), throwable);
+            val span = scope.context().get(Span.Key());
+            if (throwable != null) span.fail(throwable.getMessage(), throwable);
             span.finish();
             scope.close();
         }

kamon-kafka-clients/src/main/scala/kamon/kafka/Kafka.scala

@@ -17,17 +17,16 @@
 package kamon.kafka
 
 import com.typesafe.config.Config
-import kamon.{Kamon, OnReconfigureHook}
+import kamon.Kamon
 
 object Kafka {
   @volatile var followStrategy: Boolean = followStrategyFromConfig(Kamon.config())
 
   private def followStrategyFromConfig(config: Config): Boolean =
     Kamon.config.getBoolean("kamon.kafka.follow-strategy")
 
-  Kamon.onReconfigure(new OnReconfigureHook {
-    override def onReconfigure(newConfig: Config): Unit = {
+  Kamon.onReconfigure( (newConfig: Config) => {
       followStrategy = followStrategyFromConfig(newConfig)
     }
-  })
+  )
 }

kamon-kafka-clients/src/main/scala/kamon/kafka/instrumentation/ConsumerInstrumentation.scala

@@ -16,31 +16,32 @@
 
 package kamon.kafka.instrumentation
 
-import java.nio.ByteBuffer
+import java.io.ByteArrayOutputStream
 
 import kamon.Kamon
+import kamon.context.BinaryPropagation.{ByteStreamReader, ByteStreamWriter}
 import kamon.context.Context
 import kamon.kafka.Kafka
 import kamon.kafka.client.instrumentation.advisor.Advisors.PollMethodAdvisor
 import kamon.trace.Span
-import kanela.agent.scala.KanelaInstrumentation
+import kanela.agent.api.instrumentation.InstrumentationBuilder
 import org.apache.kafka.clients.consumer.ConsumerRecords
 
 import scala.collection.mutable
 
-class ConsumerInstrumentation extends KanelaInstrumentation {
+class ConsumerInstrumentation extends InstrumentationBuilder {
 
   /**
     * Instruments org.apache.kafka.clients.consumer.KafkaConsumer::poll(long)
     */
-  forTargetType("org.apache.kafka.clients.consumer.KafkaConsumer") { builder =>
-    builder
-      .withAdvisorFor(method("poll").and(withArgument(0, classOf[Long])), classOf[PollMethodAdvisor])
-      .build()
-  }
+  onType("org.apache.kafka.clients.consumer.KafkaConsumer")
+    .advise(method("poll").and(withArgument(0, classOf[Long])), classOf[PollMethodAdvisor])
 }
 
 object RecordProcessor {
+
+  import scala.collection.JavaConverters._
+
   /**
     * Inject Context into Records
     */
@@ -51,33 +52,40 @@ object RecordProcessor {
 
       val consumerSpansForTopic = new mutable.LinkedHashMap[String, Span]()
 
-      records.partitions().forEach(partition => {
+      records.partitions().asScala.foreach(partition => {
         val topic = partition.topic
 
-        records.records(partition).forEach(record => {
+        records.records(partition).asScala.foreach(record => {
           val header = Option(record.headers.lastHeader("kamon-context"))
 
-          val currentContext = header.map(h => Kamon.contextCodec.Binary.decode(ByteBuffer.wrap(h.value))).getOrElse(Context.Empty)
+          val currentContext = header.map{ h =>
+            Kamon.defaultBinaryPropagation().read(ByteStreamReader.of(h.value()))
+          }.getOrElse(Context.Empty)
 
           val span = consumerSpansForTopic.getOrElseUpdate(topic, {
-            val spanBuilder = Kamon.buildSpan("poll")
-              .withMetricTag("span.kind", "consumer")
-              .withTag("kafka.partition", partition.partition)
-              .withTag("kafka.topic", topic)
-              .withTag("kafka.offset", record.offset)
-              .withFrom(instant)
-
-            if(Kafka.followStrategy) spanBuilder.asChildOf(currentContext.get(Span.ContextKey))
+            val spanBuilder = Kamon.spanBuilder("poll")
+              .tag("span.kind", "consumer")
+              .tag("kafka.partition", partition.partition)
+              .tag("kafka.topic", topic)
+              .tag("kafka.offset", record.offset)
+
+            // Key could be optional ... see tests
+            Option(record.key()).foreach(k => spanBuilder.tag("kafka.key", record.key().toString))
+
+            if(Kafka.followStrategy) spanBuilder.asChildOf(currentContext.get(Span.Key))
             else {
-              val context = currentContext.get(Span.ContextKey).context()
-              spanBuilder.withTag("trace.related.trace_id", context.traceID.string)
-              spanBuilder.withTag("trace.related.span_id", context.spanID.string)
+              val currentSpan = currentContext.get(Span.Key)
+              spanBuilder
+                .tag("trace.related.trace_id", currentSpan.id.string)
+                .tag("trace.related.span_id", currentSpan.trace.id.string)
             }
             spanBuilder.start()
           })
 
-          val ctx = currentContext.withKey(Span.ContextKey, span)
-          record.headers.add("kamon-context", Kamon.contextCodec.Binary.encode(ctx).array())
+          val out = new ByteArrayOutputStream();
+          Kamon.defaultBinaryPropagation().write(currentContext.withEntry(Span.Key, span), ByteStreamWriter.of(out));
+
+          record.headers.add("kamon-context", out.toByteArray)
         })
       })
 

kamon-kafka-clients/src/main/scala/kamon/kafka/instrumentation/ProducerInstrumentation.scala

@@ -19,28 +19,26 @@ package kamon.kafka.instrumentation
 import kamon.context.Storage.Scope
 import kamon.kafka.client.instrumentation.advisor.Advisors.SendMethodAdvisor
 import kamon.trace.Span
-import kanela.agent.scala.KanelaInstrumentation
+import kanela.agent.api.instrumentation.InstrumentationBuilder
 import org.apache.kafka.clients.producer.{Callback, RecordMetadata}
 
-class ProducerInstrumentation extends KanelaInstrumentation {
+class ProducerInstrumentation extends InstrumentationBuilder {
 
   /**
     * Instruments "org.apache.kafka.clients.producer.KafkaProducer::Send()
     */
-  forTargetType("org.apache.kafka.clients.producer.KafkaProducer") { builder =>
-    builder
-      .withAdvisorFor(method("send").and(takesArguments(2)), classOf[SendMethodAdvisor])
-      .build()
-  }
+  onType("org.apache.kafka.clients.producer.KafkaProducer")
+    .advise(method("send").and(takesArguments(2)), classOf[SendMethodAdvisor])
+
 }
 
 /**
   * Producer Callback Wrapper
   */
 final class ProducerCallback(callback: Callback, scope: Scope) extends Callback {
   override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
-    val span = scope.context.get(Span.ContextKey)
-    if(exception != null) span.addError(exception.getMessage, exception)
+    val span = scope.context.get(Span.Key)
+    if(exception != null) span.fail(exception.getMessage, exception)
     try if(callback != null) callback.onCompletion(metadata, exception)
     finally {
       span.finish()