-
Notifications
You must be signed in to change notification settings - Fork 276
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improve observability of the workload launcher/monitor. (#10398)
- Loading branch information
Showing
17 changed files
with
412 additions
and
135 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,39 +1,44 @@ | ||
| plugins { | ||
| id("io.airbyte.gradle.jvm.lib") | ||
| id("io.airbyte.gradle.publish") | ||
| kotlin("jvm") | ||
| kotlin("kapt") | ||
| } | ||
|
|
||
| dependencies { | ||
| implementation( project(":airbyte-commons")) | ||
| implementation( project(":airbyte-config:config-models")) | ||
| implementation( project(":airbyte-db:jooq")) | ||
| implementation( project(":airbyte-db:db-lib")) | ||
| kapt(libs.bundles.micronaut.annotation.processor) | ||
|
|
||
| implementation( libs.guava) | ||
| implementation( libs.google.cloud.storage) | ||
| implementation(project(":airbyte-commons")) | ||
| implementation(project(":airbyte-config:config-models")) | ||
| implementation(project(":airbyte-db:jooq")) | ||
| implementation(project(":airbyte-db:db-lib")) | ||
|
|
||
| implementation(libs.guava) | ||
| implementation(libs.google.cloud.storage) | ||
| compileOnly(libs.lombok) | ||
| annotationProcessor( libs.lombok) | ||
| annotationProcessor(libs.lombok) | ||
|
|
||
| implementation( libs.otel.semconv) | ||
| implementation( libs.otel.sdk) | ||
| implementation( libs.otel.sdk.testing) | ||
| implementation( libs.micrometer.statsd) | ||
| implementation( platform(libs.otel.bom)) | ||
| implementation(libs.otel.semconv) | ||
| implementation(libs.otel.sdk) | ||
| implementation(libs.otel.sdk.testing) | ||
| implementation(libs.micrometer.statsd) | ||
| implementation(platform(libs.otel.bom)) | ||
| implementation(("io.opentelemetry:opentelemetry-api")) | ||
| implementation(("io.opentelemetry:opentelemetry-sdk")) | ||
| implementation(("io.opentelemetry:opentelemetry-exporter-otlp")) | ||
|
|
||
| implementation( libs.java.dogstatsd.client) | ||
| implementation( libs.bundles.datadog) | ||
| implementation(libs.java.dogstatsd.client) | ||
| implementation(libs.bundles.datadog) | ||
|
|
||
| testImplementation( project(":airbyte-config:config-persistence")) | ||
| testImplementation( project(":airbyte-test-utils")) | ||
| testImplementation( libs.platform.testcontainers.postgresql) | ||
| testImplementation(project(":airbyte-config:config-persistence")) | ||
| testImplementation(project(":airbyte-test-utils")) | ||
| testImplementation(libs.platform.testcontainers.postgresql) | ||
| testRuntimeOnly(libs.junit.jupiter.engine) | ||
| testImplementation( libs.bundles.junit) | ||
| testImplementation( libs.assertj.core) | ||
| testImplementation(libs.bundles.junit) | ||
| testImplementation(libs.assertj.core) | ||
| testImplementation(libs.mockk) | ||
| testImplementation((variantOf(libs.opentracing.util.test) { classifier("tests") })) | ||
|
|
||
| testImplementation( libs.junit.pioneer) | ||
| testImplementation(libs.junit.pioneer) | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
30 changes: 30 additions & 0 deletions
30
airbyte-metrics/metrics-lib/src/main/kotlin/io/airbyte/metrics/annotations/Instrument.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package io.airbyte.metrics.annotations | ||
|
|
||
| import io.micronaut.aop.Around | ||
|
|
||
| annotation class Tag(val key: String, val value: String) | ||
|
|
||
| /** | ||
| * Set this annotation to a function to instrument metric emissions. | ||
| * | ||
| * As we are using [io.airbyte.metrics.lib.MetricsRegistry] to define the metrics, values provided for [start], [end] and [duration] need to be | ||
| * valid values from a [io.airbyte.metrics.lib.MetricsRegistry]. | ||
| * | ||
| * For the [end] and [duration] metric, a `status` tag with values `ok` or `error` will be automatically added. Status being `error` if an | ||
| * exception is thrown, `ok` otherwise. | ||
| * | ||
| * @property start if not empty defines the name of metric to emit on start of the method. | ||
| * @property end if not empty defines the name metric to emit at the end of the method. | ||
| * @property duration if not empty defines the name metric to the duration of the method. | ||
| * @property tags defines a list of custom tags to be added to each metrics. | ||
| */ | ||
| @MustBeDocumented | ||
| @Retention(AnnotationRetention.RUNTIME) | ||
| @Target(AnnotationTarget.FUNCTION) | ||
| @Around | ||
| annotation class Instrument( | ||
| val start: String = "", | ||
| val end: String = "", | ||
| val duration: String = "", | ||
| val tags: Array<Tag> = [], | ||
| ) |
76 changes: 76 additions & 0 deletions
76
.../metrics-lib/src/main/kotlin/io/airbyte/metrics/interceptors/InstrumentInterceptorBase.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package io.airbyte.metrics.interceptors | ||
|
|
||
| import io.airbyte.metrics.annotations.Instrument | ||
| import io.airbyte.metrics.annotations.Tag | ||
| import io.airbyte.metrics.lib.MetricAttribute | ||
| import io.micronaut.aop.MethodInterceptor | ||
| import io.micronaut.aop.MethodInvocationContext | ||
| import io.micronaut.core.annotation.AnnotationValue | ||
| import kotlin.time.Duration | ||
| import kotlin.time.TimeSource | ||
|
|
||
| abstract class InstrumentInterceptorBase : MethodInterceptor<Any, Any> { | ||
| companion object { | ||
| const val START = "start" | ||
| const val END = "end" | ||
| const val DURATION = "duration" | ||
| const val TAGS = "tags" | ||
|
|
||
| const val SUCCESS_STATUS = "ok" | ||
| const val FAILURE_STATUS = "error" | ||
| } | ||
|
|
||
| abstract fun emitStartMetric( | ||
| startMetricName: String, | ||
| tags: Array<MetricAttribute>, | ||
| ) | ||
|
|
||
| abstract fun emitEndMetric( | ||
| endMetricName: String, | ||
| success: Boolean, | ||
| tags: Array<MetricAttribute>, | ||
| ) | ||
|
|
||
| abstract fun emitDurationMetric( | ||
| durationMetricName: String, | ||
| duration: Duration, | ||
| success: Boolean, | ||
| tags: Array<MetricAttribute>, | ||
| ) | ||
|
|
||
| override fun intercept(context: MethodInvocationContext<Any, Any>): Any? { | ||
| val annotationValue = context.getAnnotation(Instrument::class.java) | ||
| return if (annotationValue != null) { | ||
| doIntercept(annotationValue, context) | ||
| } else { | ||
| context.proceed() | ||
| } | ||
| } | ||
|
|
||
| private fun doIntercept( | ||
| annotationValue: AnnotationValue<Instrument>, | ||
| context: MethodInvocationContext<Any, Any>, | ||
| ): Any? { | ||
| val tags = readTags(annotationValue) | ||
|
|
||
| annotationValue.stringValue(START).ifPresent { emitStartMetric(it, tags) } | ||
|
|
||
| var success = true | ||
| val startTime = TimeSource.Monotonic.markNow() | ||
| try { | ||
| return context.proceed() | ||
| } catch (e: Throwable) { | ||
| success = false | ||
| throw e | ||
| } finally { | ||
| annotationValue.stringValue(END).ifPresent { emitEndMetric(it, success, tags) } | ||
| annotationValue.stringValue(DURATION).ifPresent { emitDurationMetric(it, startTime.elapsedNow(), success, tags) } | ||
| } | ||
| } | ||
|
|
||
| private fun readTags(annotationValue: AnnotationValue<Instrument>): Array<MetricAttribute> { | ||
| return annotationValue.getAnnotations<Tag>(TAGS) | ||
| .map { MetricAttribute(it.stringValue("key").orElse(""), it.stringValue("value").orElse("")) } | ||
| .toTypedArray() | ||
| } | ||
| } |
49 changes: 49 additions & 0 deletions
49
...-lib/src/main/kotlin/io/airbyte/metrics/interceptors/MetricClientInstrumentInterceptor.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| package io.airbyte.metrics.interceptors | ||
|
|
||
| import io.airbyte.metrics.annotations.Instrument | ||
| import io.airbyte.metrics.lib.MetricAttribute | ||
| import io.airbyte.metrics.lib.MetricClient | ||
| import io.airbyte.metrics.lib.MetricTags | ||
| import io.airbyte.metrics.lib.OssMetricsRegistry | ||
| import io.micronaut.aop.InterceptorBean | ||
| import jakarta.inject.Singleton | ||
| import kotlin.time.Duration | ||
| import kotlin.time.DurationUnit | ||
|
|
||
| @Singleton | ||
| @InterceptorBean(Instrument::class) | ||
| class MetricClientInstrumentInterceptor(private val metricClient: MetricClient) : InstrumentInterceptorBase() { | ||
| override fun emitStartMetric( | ||
| startMetricName: String, | ||
| tags: Array<MetricAttribute>, | ||
| ) { | ||
| metricClient.count(OssMetricsRegistry.valueOf(startMetricName), 1, *tags) | ||
| } | ||
|
|
||
| override fun emitEndMetric( | ||
| endMetricName: String, | ||
| success: Boolean, | ||
| tags: Array<MetricAttribute>, | ||
| ) { | ||
| metricClient.count( | ||
| OssMetricsRegistry.valueOf(endMetricName), | ||
| 1, | ||
| MetricAttribute(MetricTags.STATUS, if (success) SUCCESS_STATUS else FAILURE_STATUS), | ||
| *tags, | ||
| ) | ||
| } | ||
|
|
||
| override fun emitDurationMetric( | ||
| durationMetricName: String, | ||
| duration: Duration, | ||
| success: Boolean, | ||
| tags: Array<MetricAttribute>, | ||
| ) { | ||
| metricClient.distribution( | ||
| OssMetricsRegistry.valueOf(durationMetricName), | ||
| duration.toDouble(DurationUnit.MILLISECONDS), | ||
| MetricAttribute(MetricTags.STATUS, if (success) SUCCESS_STATUS else FAILURE_STATUS), | ||
| *tags, | ||
| ) | ||
| } | ||
| } |
Oops, something went wrong.