Merge pull request #977 from pnerg/otel-trace-exporter

Added support for opentelemetry trace exporter

build.sbt

@@ -476,6 +476,7 @@ lazy val reporters = (project in file("reporters"))
     `kamon-influxdb`,
     `kamon-jaeger`,
     `kamon-newrelic`,
+    `kamon-opentelemetry`,
     `kamon-prometheus`,
     `kamon-statsd`,
     `kamon-zipkin`,
@@ -591,6 +592,16 @@ lazy val `kamon-newrelic` = (project in file("reporters/kamon-newrelic"))
     )
   ).dependsOn(`kamon-core`)
 
+lazy val `kamon-opentelemetry` = (project in file("reporters/kamon-opentelemetry"))
+  .disablePlugins(AssemblyPlugin)
+  .settings(
+    libraryDependencies ++= Seq(
+      "io.opentelemetry" % "opentelemetry-proto" % "0.17.1",
+      "io.grpc" % "grpc-netty" % "1.36.0",
+      scalatest % "test",
+      logbackClassic % "test"
+    )
+  ).dependsOn(`kamon-core`, `kamon-testkit` % "test")
 
 lazy val `kamon-prometheus` = (project in file("reporters/kamon-prometheus"))
   .disablePlugins(AssemblyPlugin)

reporters/kamon-opentelemetry/README.md

@@ -0,0 +1,10 @@
+# Kamon OpenTelemetry Exporter
+The exporter currently only provides a OpenTelemetry (OTLP) exporter for Kamon spans (metrics to be supported)
+
+The reporter relies on the [opentelemetry-proto](https://github.com/open-telemetry/opentelemetry-proto) library for the gRPC communication with an OpenTelemetry (OTLP) service.
+
+## Trace Exporter
+Converts internal finished Kamon spans to OTEL proto format and exports them the to configured endpoint.
+
+## Metrics Exporter
+To be implemented

reporters/kamon-opentelemetry/src/main/resources/reference.conf

@@ -0,0 +1,43 @@
+# ======================================== #
+# kamon-otlp reference configuration       #
+# ======================================== #
+
+kamon.otel.trace {
+  # Hostname and port where the OTLP Server is running
+  host = "localhost"
+  port = 4317
+
+  # Decides whether to use HTTP or HTTPS when connecting to the OTel server
+  protocol = "http"
+
+  # default to support the ENV:s as described at
+  # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/exporter.md
+  endpoint = ${kamon.otel.trace.protocol}"://"${kamon.otel.trace.host}":"${kamon.otel.trace.port}
+  endpoint = ${?OTEL_EXPORTER_OTLP_ENDPOINT}
+  endpoint = ${?OTEL_EXPORTER_OTLP_TRACES_ENDPOINT}
+
+  # Enable or disable including tags from kamon.environment.tags as resource labels in the exported trace
+  # Any keys containing a hyphen (-) will be converted to a dot (.) as that is the standard.
+  # e.g. service-version becomes service.version
+  # reference: https://github.com/kamon-io/Kamon/blob/master/core/kamon-core/src/main/resources/reference.conf#L23
+  include-environment-tags = no
+
+  # Arbitrary key-value pairs that further identify the environment where this service instance is running.
+  # These are added as KeyValue labels to the Resource part of the exported traces
+  # Requires 'include-environment-tags' to be set to 'yes'
+  #
+  # kamon.environment.tags {
+  #   service-version = "x.x.x"
+  #   service-namespace = "ns"
+  #   service-instance.id = "xxx-yyy"
+  # }
+}
+
+kamon.modules {
+  otel-trace-reporter {
+    enabled = true
+    name = "OpenTelemetry Trace Reporter"
+    description = "Sends trace data to a OpenTelemetry server via gRPC"
+    factory = "kamon.otel.OpenTelemetryTraceReporter$Factory"
+  }
+}

reporters/kamon-opentelemetry/src/main/scala/kamon/otel/OpenTelemetryTraceReporter.scala

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2013-2021 The Kamon Project <https://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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kamon.otel
+
+import com.typesafe.config.Config
+import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest
+import io.opentelemetry.proto.common.v1.InstrumentationLibrary
+import io.opentelemetry.proto.resource.v1.Resource
+import io.opentelemetry.proto.trace.v1.ResourceSpans
+import kamon.Kamon
+import kamon.module.{Module, ModuleFactory, SpanReporter}
+import kamon.trace.Span
+import kamon.otel.SpanConverter._
+import org.slf4j.LoggerFactory
+
+import java.util.Collections
+import scala.concurrent.ExecutionContext
+import scala.util.{Failure, Success}
+
+object OpenTelemetryTraceReporter {
+  private val logger = LoggerFactory.getLogger(classOf[OpenTelemetryTraceReporter])
+  private val kamonVersion = Kamon.status().settings().version
+
+  class Factory extends ModuleFactory {
+    override def create(settings: ModuleFactory.Settings): Module = {
+      logger.info("Creating OpenTelemetry Trace Reporter")
+
+      val module = new OpenTelemetryTraceReporter(GrpcTraceService(_))(settings.executionContext)
+      module.reconfigure(settings.config)
+      module
+    }
+  }
+}
+
+import kamon.otel.OpenTelemetryTraceReporter._
+/**
+ * Converts internal finished Kamon spans to OpenTelemetry format and sends to a configured OpenTelemetry endpoint using gRPC.
+ */
+class OpenTelemetryTraceReporter(traceServiceFactory:Config=>TraceService)(implicit ec:ExecutionContext) extends SpanReporter {
+  private var traceService:Option[TraceService] = None
+  private var spanConverterFunc:Seq[Span.Finished]=>ResourceSpans = (_ => ResourceSpans.newBuilder().build())
+
+  override def reportSpans(spans: Seq[Span.Finished]): Unit = {
+    if(!spans.isEmpty) {
+      val resources = Collections.singletonList(spanConverterFunc(spans)) //all spans should belong to the same single resource
+      val exportTraceServiceRequest = ExportTraceServiceRequest.newBuilder()
+        .addAllResourceSpans(resources)
+        .build()
+
+      traceService.foreach (
+        _.export(exportTraceServiceRequest).onComplete {
+          case Success(_) => logger.debug("Successfully exported traces")
+
+          //TODO is there result for which a retry is relevant? Perhaps a glitch in the receiving service
+          //Keeping logs to debug as other exporters (e.g.Zipkin) won't log anything if it fails to export traces
+          case Failure(t) => logger.debug("Failed to export traces", t)
+        }
+      )
+    }
+  }
+
+  override def reconfigure(newConfig: Config): Unit = {
+    logger.info("Reconfigure OpenTelemetry Trace Reporter")
+
+    //pre-generate the function for converting Kamon span to proto span
+    val instrumentationLibrary:InstrumentationLibrary = InstrumentationLibrary.newBuilder().setName("kamon").setVersion(kamonVersion).build()
+    val resource:Resource = buildResource(newConfig.getBoolean("kamon.otel.trace.include-environment-tags"))
+    this.spanConverterFunc = SpanConverter.toProtoResourceSpan(resource, instrumentationLibrary)
+
+    this.traceService = Option(traceServiceFactory.apply(newConfig))
+  }
+
+  override def stop(): Unit = {
+    logger.info("Stopping OpenTelemetry Trace Reporter")
+    this.traceService.foreach(_.close())
+    this.traceService = None
+  }
+
+  /**
+   * Builds the resource information added as resource labels to the exported traces
+   * @param includeEnvTags
+   * @return
+   */
+  private def buildResource(includeEnvTags:Boolean):Resource = {
+    val env = Kamon.environment
+    val builder = Resource.newBuilder()
+      .addAttributes(stringKeyValue("service.name", env.service))
+      .addAttributes(stringKeyValue("telemetry.sdk.name", "kamon"))
+      .addAttributes(stringKeyValue("telemetry.sdk.language", "scala"))
+      .addAttributes(stringKeyValue("telemetry.sdk.version", kamonVersion))
+
+    //add all kamon.environment.tags as KeyValues to the Resource object
+    if(includeEnvTags) {
+      env.tags.iterator().map(toProtoKeyValue).foreach(builder.addAttributes)
+    }
+
+    builder.build()
+  }
+}