Skip to content

Commit 1b8eaa0

Browse files
committed
add opentelemetry metrics exporter
1 parent cebcc5a commit 1b8eaa0

11 files changed

+600
-155
lines changed

build.sbt

+4-2
Original file line numberDiff line numberDiff line change
@@ -812,8 +812,10 @@ lazy val `kamon-opentelemetry` = (project in file("reporters/kamon-opentelemetry
812812
.settings(
813813
crossScalaVersions += `scala_3_version`,
814814
libraryDependencies ++= Seq(
815-
"io.opentelemetry" % "opentelemetry-exporter-otlp-http-trace" % "1.13.0",
816-
"io.opentelemetry" % "opentelemetry-exporter-otlp-trace" % "1.13.0",
815+
"io.opentelemetry" % "opentelemetry-exporter-otlp-http-trace" % "1.14.0",
816+
"io.opentelemetry" % "opentelemetry-exporter-otlp-trace" % "1.14.0",
817+
"io.opentelemetry" % "opentelemetry-exporter-otlp-http-metrics" % "1.14.0",
818+
"io.opentelemetry" % "opentelemetry-exporter-otlp-metrics" % "1.14.0",
817819
// Compile-time dependency required in scala 3
818820
"com.google.auto.value" % "auto-value-annotations" % "1.9" % "compile",
819821

project/Build.scala

+1-1
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ object BaseProject extends AutoPlugin {
5959
val `scala_2.11_version` = "2.11.12"
6060
val `scala_2.12_version` = "2.12.15"
6161
val `scala_2.13_version` = "2.13.8"
62-
val scala_3_version = "3.2.0"
62+
val scala_3_version = "3.3.0"
6363

6464
// This installs the GPG signing key from the
6565
setupGpg()

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

+24-1
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,23 @@ kamon.otel {
2727
attributes = ""
2828
attributes = ${?OTEL_RESOURCE_ATTRIBUTES}
2929

30+
metrics {
31+
endpoint = ${kamon.otel.endpoint}
32+
full-endpoint = ${?OTEL_EXPORTER_OTLP_METRICS_ENDPOINT}
33+
34+
compression = ${kamon.otel.compression}
35+
compression = ${?OTEL_EXPORTER_OTLP_METRICS_COMPRESSION}
36+
37+
headers = ${kamon.otel.headers}
38+
headers = ${?OTEL_EXPORTER_OTLP_METRICS_HEADERS}
39+
40+
timeout = ${kamon.otel.timeout}
41+
timeout = ${?OTEL_EXPORTER_OTLP_METRICS_TIMEOUT}
42+
43+
protocol = ${kamon.otel.protocol}
44+
protocol = ${?OTEL_EXPORTER_OTLP_METRICS_PROTOCOL}
45+
}
46+
3047
trace {
3148
endpoint = ${kamon.otel.endpoint}
3249
full-endpoint = ${?OTEL_EXPORTER_OTLP_TRACES_ENDPOINT}
@@ -64,7 +81,13 @@ kamon.modules {
6481
otel-trace-reporter {
6582
enabled = true
6683
name = "OpenTelemetry Trace Reporter"
67-
description = "Sends trace data to a OpenTelemetry server via gRPC"
84+
description = "Sends trace data to a OpenTelemetry server via gRPC/REST+json"
6885
factory = "kamon.otel.OpenTelemetryTraceReporter$Factory"
6986
}
87+
otel-metrics-reporter {
88+
enabled = true
89+
name = "OpenTelemetry Metrics Reporter"
90+
description = "Sends metrics data to a OpenTelemetry server via gRPC/REST+json"
91+
factory = "kamon.otel.OpenTelemetryMetricsReporter$Factory"
92+
}
7093
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
/*
2+
* Copyright 2013-2021 The Kamon Project <https://kamon.io>
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package kamon.otel
17+
18+
import io.opentelemetry.sdk.common.InstrumentationScopeInfo
19+
import io.opentelemetry.sdk.metrics.data._
20+
import io.opentelemetry.sdk.metrics.internal.data._
21+
import io.opentelemetry.sdk.resources.Resource
22+
import kamon.metric.Instrument.Snapshot
23+
import kamon.metric.{Distribution, MeasurementUnit, MetricSnapshot, PeriodSnapshot}
24+
import kamon.tag.Lookups
25+
import kamon.trace.Span.TagKeys
26+
27+
import java.lang.{Double => JDouble, Long => JLong}
28+
import java.time.Instant
29+
import java.util.{Collection => JCollection}
30+
import scala.collection.JavaConverters._
31+
32+
class WithResourceMetricsConverter(resource: Resource, kamonVersion: String, from: Instant, to: Instant) {
33+
private val fromNs = from.toEpochMilli * 1000000
34+
private val toNs = to.toEpochMilli * 1000000
35+
36+
private def instrumentationScopeInfo(snapshot: MetricSnapshot[_, _]): InstrumentationScopeInfo = {
37+
// logic for looking up the component doesn't really seem to make sense - to be compliant we should probably be grouping the metrics by component before calling this
38+
InstrumentationScopeInfo.create(snapshot.instruments.headOption.flatMap(_.tags.get(Lookups.option(TagKeys.Component))) getOrElse "kamon-instrumentation", kamonVersion, null)
39+
}
40+
41+
private def toString(unit: MeasurementUnit): String = unit.magnitude.name
42+
43+
def toGaugeDatum(g: Snapshot[Double]): DoublePointData = ImmutableDoublePointData.create(fromNs, toNs, SpanConverter.toAttributes(g.tags), g.value)
44+
45+
def toGaugeData(g: Seq[Snapshot[Double]]): GaugeData[DoublePointData] = ImmutableGaugeData.create(g.map(toGaugeDatum).asJava)
46+
47+
def convertGauge(gauge: MetricSnapshot.Values[Double]): MetricData =
48+
ImmutableMetricData.createDoubleGauge(
49+
resource,
50+
instrumentationScopeInfo(gauge),
51+
gauge.name,
52+
gauge.description,
53+
toString(gauge.settings.unit),
54+
toGaugeData(gauge.instruments))
55+
56+
def toHistogramDatum(s: Snapshot[Distribution]): HistogramPointData =
57+
ImmutableHistogramPointData.create(
58+
fromNs,
59+
toNs,
60+
SpanConverter.toAttributes(s.tags),
61+
JDouble valueOf s.value.sum.toDouble,
62+
JDouble valueOf s.value.min.toDouble,
63+
JDouble valueOf s.value.max.toDouble,
64+
s.value.buckets.map(JDouble valueOf _.value.toDouble).asJava,
65+
s.value.buckets.map(JLong valueOf _.frequency).asJava
66+
)
67+
68+
def toHistogramData(any: Seq[Snapshot[Distribution]]): HistogramData =
69+
ImmutableHistogramData.create(AggregationTemporality.CUMULATIVE, any.map(toHistogramDatum).asJava)
70+
71+
def convertHistogram(histogram: MetricSnapshot.Distributions): MetricData =
72+
ImmutableMetricData.createDoubleHistogram(
73+
resource,
74+
instrumentationScopeInfo(histogram),
75+
histogram.name,
76+
histogram.description,
77+
toString(histogram.settings.unit),
78+
toHistogramData(histogram.instruments))
79+
80+
def toCounterDatum(g: Snapshot[Long]): LongPointData =
81+
ImmutableLongPointData.create(fromNs, toNs, SpanConverter.toAttributes(g.tags), g.value)
82+
83+
def toCounterData(g: Seq[Snapshot[Long]]): SumData[LongPointData] =
84+
ImmutableSumData.create(false, AggregationTemporality.CUMULATIVE, g.map(toCounterDatum).asJava)
85+
86+
def convertCounter(counter: MetricSnapshot.Values[Long]): MetricData =
87+
ImmutableMetricData.createLongSum(
88+
resource,
89+
instrumentationScopeInfo(counter),
90+
counter.name,
91+
counter.description,
92+
toString(counter.settings.unit),
93+
toCounterData(counter.instruments))
94+
95+
}
96+
97+
/**
98+
* Converts Kamon metrics to OpenTelemetry [[MetricData]]s
99+
*/
100+
private[otel] object MetricsConverter {
101+
def convert(resource: Resource, kamonVersion: String)(metrics: PeriodSnapshot): JCollection[MetricData] = {
102+
val converter = new WithResourceMetricsConverter(resource, kamonVersion, metrics.from, metrics.to)
103+
val gauges = metrics.gauges.map(converter.convertGauge)
104+
val histograms = (metrics.histograms ++ metrics.timers ++ metrics.rangeSamplers).map(converter.convertHistogram)
105+
val counters = metrics.counters.map(converter.convertCounter)
106+
107+
(gauges ++ histograms ++ counters).asJava
108+
}
109+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
package kamon.otel
2+
3+
import com.typesafe.config.Config
4+
import io.opentelemetry.sdk.resources.Resource
5+
import kamon.Kamon
6+
import kamon.status.Status
7+
import kamon.tag.Tag
8+
import org.slf4j.LoggerFactory
9+
10+
import java.net.URL
11+
import java.time.Duration
12+
13+
case class OpenTelemetryConfiguration(protocol: String, endpoint: String, compressionEnabled: Boolean, headers: Seq[(String, String)], timeout: Duration)
14+
15+
object OpenTelemetryConfiguration {
16+
private val logger = LoggerFactory.getLogger(classOf[OpenTelemetryConfiguration])
17+
18+
object Component extends Enumeration {
19+
val Trace, Metrics = Value
20+
type Component = Value
21+
}
22+
23+
import Component._
24+
25+
/**
26+
* Builds an otel configuration object using the provided typesafe configuration.
27+
*
28+
* @param config
29+
* @return
30+
*/
31+
def apply(config: Config, component: Component): OpenTelemetryConfiguration = {
32+
val name = component.toString.toLowerCase
33+
val otelExporterConfig = config.getConfig(s"kamon.otel.$name")
34+
val endpoint = otelExporterConfig.getString("endpoint")
35+
val fullEndpoint = if (otelExporterConfig.hasPath("full-endpoint")) Some(otelExporterConfig.getString("full-endpoint")) else None
36+
val compression = otelExporterConfig.getString("compression") match {
37+
case "gzip" => true
38+
case x =>
39+
if (x != "") logger.warn(s"unrecognised compression $x. Defaulting to no compression")
40+
false
41+
}
42+
val protocol = otelExporterConfig.getString("protocol") match {
43+
case "http/protobuf" => "http/protobuf"
44+
case "grpc" => "grpc"
45+
case x =>
46+
logger.warn(s"Unrecognised opentelemetry schema type $x. Defaulting to grpc")
47+
"grpc"
48+
}
49+
val headers = otelExporterConfig.getString("headers").split(',').filter(_.nonEmpty).map(_.split("=", 2)).map {
50+
case Array(k) => k -> ""
51+
case Array(k, v) => k -> v
52+
}.toSeq
53+
val timeout = otelExporterConfig.getDuration("timeout")
54+
// See https://opentelemetry.io/docs/reference/specification/protocol/exporter/#endpoint-urls-for-otlphttp
55+
val httpSuffix = component match {
56+
case Trace => "traces"
57+
case Metrics => "metrics"
58+
}
59+
val url = (protocol, fullEndpoint) match {
60+
case ("http/protobuf", Some(full)) =>
61+
val parsed = new URL(full)
62+
if (parsed.getPath.isEmpty) full :+ '/' else full
63+
// Seems to be some dispute as to whether the / should technically be added in the case that the base path doesn't
64+
// include it. Adding because it's probably what's desired most of the time, and can always be overridden by full-endpoint
65+
case ("http/protobuf", None) => if (endpoint.endsWith("/")) s"${endpoint}v1/$httpSuffix" else s"$endpoint/v1/$httpSuffix"
66+
case (_, Some(full)) => full
67+
case (_, None) => endpoint
68+
}
69+
70+
logger.info(s"Configured endpoint for OpenTelemetry $name reporting [$url] using $protocol protocol")
71+
72+
OpenTelemetryConfiguration(protocol, url, compression, headers, timeout)
73+
}
74+
75+
private val kamonSettings: Status.Settings = Kamon.status().settings()
76+
/**
77+
* Builds the resource information added as resource labels to the exported metrics/traces
78+
*
79+
* @return
80+
*/
81+
def buildResource(attributes: Map[String, String]): Resource = {
82+
val env = Kamon.environment
83+
val builder = Resource.builder()
84+
.put("host.name", kamonSettings.environment.host)
85+
.put("service.instance.id", kamonSettings.environment.instance)
86+
.put("service.name", env.service)
87+
.put("telemetry.sdk.name", "kamon")
88+
.put("telemetry.sdk.language", "scala")
89+
.put("telemetry.sdk.version", kamonSettings.version)
90+
91+
attributes.foreach { case (k, v) => builder.put(k, v) }
92+
//add all kamon.environment.tags as KeyValues to the Resource object
93+
env.tags.iterator().foreach {
94+
case t: Tag.String => builder.put(t.key, t.value)
95+
case t: Tag.Boolean => builder.put(t.key, t.value)
96+
case t: Tag.Long => builder.put(t.key, t.value)
97+
}
98+
99+
builder.build()
100+
}
101+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2013-2021 The Kamon Project <https://kamon.io>
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package kamon.otel
17+
18+
import com.typesafe.config.Config
19+
import io.opentelemetry.sdk.metrics.data.MetricData
20+
import io.opentelemetry.sdk.resources.Resource
21+
import kamon.Kamon
22+
import kamon.metric.PeriodSnapshot
23+
import kamon.module.{MetricReporter, Module, ModuleFactory}
24+
import kamon.status.Status
25+
import kamon.tag.Tag
26+
import org.slf4j.LoggerFactory
27+
28+
import java.net.URLDecoder
29+
import java.util
30+
import java.util.{Collection => JCollection}
31+
import scala.concurrent.ExecutionContext
32+
import scala.util.{Failure, Success, Try}
33+
34+
object OpenTelemetryMetricsReporter {
35+
private val logger = LoggerFactory.getLogger(classOf[OpenTelemetryMetricsReporter])
36+
private val kamonSettings: Status.Settings = Kamon.status().settings()
37+
38+
class Factory extends ModuleFactory {
39+
override def create(settings: ModuleFactory.Settings): Module = {
40+
logger.info("Creating OpenTelemetry Metrics Reporter")
41+
42+
val module = new OpenTelemetryMetricsReporter(OtlpMetricsService.apply)(settings.executionContext)
43+
module.reconfigure(settings.config)
44+
module
45+
}
46+
}
47+
}
48+
49+
import kamon.otel.OpenTelemetryMetricsReporter._
50+
51+
/**
52+
* Converts internal Kamon metrics to OpenTelemetry format and sends to a configured OpenTelemetry endpoint using gRPC or REST.
53+
*/
54+
class OpenTelemetryMetricsReporter(metricsServiceFactory: Config => MetricsService)(implicit ec: ExecutionContext) extends MetricReporter {
55+
private var metricsService: Option[MetricsService] = None
56+
private var metricsConverterFunc: PeriodSnapshot => JCollection[MetricData] = (_ => new util.ArrayList[MetricData](0))
57+
58+
def isEmpty(snapshot: PeriodSnapshot): Boolean =
59+
snapshot.gauges.isEmpty && snapshot.timers.isEmpty && snapshot.counters.isEmpty && snapshot.histograms.isEmpty && snapshot.rangeSamplers.isEmpty
60+
61+
override def reportPeriodSnapshot(snapshot: PeriodSnapshot): Unit = {
62+
if (!isEmpty(snapshot)) {
63+
metricsService.foreach(ts => ts.exportMetrics(metricsConverterFunc(snapshot)).onComplete {
64+
case Success(_) => logger.debug("Successfully exported metrics")
65+
66+
//TODO is there result for which a retry is relevant? Perhaps a glitch in the receiving service
67+
//Keeping logs to debug as the underlying exporter will log if it fails to export metrics, and the failure isn't surfaced in the response anyway
68+
case Failure(t) => logger.debug("Failed to export metrics", t)
69+
})
70+
}
71+
}
72+
73+
override def reconfigure(newConfig: Config): Unit = {
74+
logger.info("Reconfigure OpenTelemetry Metrics Reporter")
75+
76+
//pre-generate the function for converting Kamon metrics to proto metrics
77+
val attributes: Map[String, String] =
78+
newConfig.getString("kamon.otel.attributes").split(',').filter(_ contains '=').map(_.trim.split("=", 2)).map {
79+
case Array(k, v) =>
80+
val decoded = Try(URLDecoder.decode(v.trim, "UTF-8"))
81+
decoded.failed.foreach(t => throw new IllegalArgumentException(s"value for attribute ${k.trim} is not a url-encoded string", t))
82+
k.trim -> decoded.get
83+
}.toMap
84+
val resource: Resource = OpenTelemetryConfiguration.buildResource(attributes)
85+
this.metricsConverterFunc = MetricsConverter.convert(resource, kamonSettings.version)
86+
87+
this.metricsService = Option(metricsServiceFactory.apply(newConfig))
88+
}
89+
90+
override def stop(): Unit = {
91+
logger.info("Stopping OpenTelemetry Metrics Reporter")
92+
this.metricsService.foreach(_.close())
93+
this.metricsService = None
94+
}
95+
96+
}

0 commit comments

Comments
 (0)