Metrics Publisher Support

[RodrasSilva]

Hello.

I'm migrating my application from the Java AWS SDK to the Kotlin AWS SDK.
When I was building the S3Client using the Java SDK, I was able to provide a Metrics publisher where I could build my logic to transform the SDK-provided metrics into my own:

 return S3AsyncClient.builder()
     .overrideConfiguration { cfg ->
          cfg.addMetricPublisher(S3MetricsPublisher(...))
      }  
     ...

internal class S3MetricsPublisher(
    private val s3Metrics: ImageRepositoryMetrics,
    private val bucket: String
) : MetricPublisher {

    override fun publish(metricCollection: MetricCollection) {
        val metrics = metricsToMap(metricCollection)
        metrics[RETRY_COUNT_KEY.key]?.let { s3Metrics.publishRetryCount(bucket, (it as Int).toDouble()) }
        ...
    } 

How am I able to replicate this behavior using the Kotlin SDK and the telemetry provider?

Thanks in advance.

[ianbotsf]

Hi @RodrasSilva, excellent question! The telemetry and metrics interfaces for the AWS SDK for Kotlin are slightly different from the AWS SDK for Java but it still should be possible to implement your S3 publisher adaptor. The key is providing your own implementation of a MetricsProvider which receives calls to create/record metrics and then adapts them to your publishing logic. A possible implementation can be broken down into several parts:

Providers

As an example, here's an implementation of a telemetry provider which has a custom meter provider and nothing else:

class MyTelemetryProvider : TelemetryProvider {
    override val contextManager = ContextManager.None
    override val loggerProvider = LoggerProvider.None
    override val tracerProvider = TracerProvider.None
    override val meterProvider = MyMeterProvider()
}

class MyMeterProvider : MeterProvider {
    override fun getOrCreateMeter(scope: String) = MyMeter(scope)
}

class MyMeter : Meter {
    …
}

Meters

The Meter class supports creating instruments for 7 different types of measurements:

• async up-down counter: used for discrete measurements which go up and down like HTTP connection usage
• monotonic counter: used for discrete measurements which only go up like the number of calls made
• double histogram: used for aggregating non-discrete measurements which are likely to be statistically useful like latency
• long histogram: currently unused
• double gauge: currently unused
• long gauge: currently unused
• up-down counter: currently unused

Your code for a Meter implementation will vary depending on which metrics you want to capture and publish. You may choose to implement handling for every kind of instrument or only for one or two based on your use case. Here is an example of implementing support for capturing double histograms which would include measurements like latency:

class MyMeter : Meter {
    override fun createDoubleHistogram(
        name: String,
        units: String?,
        description: String?,
    ) = MyDoubleHistogram(name, units, description)

    // Override the other Meter members with no-op implementations
    override fun createAsyncUpDownCounter(…) = NoOpAsyncMeasurementHandle
    override fun createUpDownCounter(…) = NoOpUpDownCounter
    override fun createMonotonicCounter(…) = NoOpMonotonicCounter
    override fun createLongHistogram(…) = NoOpLongHistogram
    override fun createLongGauge(…) = NoOpAsyncMeasurementHandle
    override fun createDoubleGauge(…) = NoOpAsyncMeasurementHandle
}

class MyDoubleHistogram(private val name: String, private val units: String?) : DoubleHistogram {
    override fun record(value: Double, attributes: Attributes, context: Context?) {
        // Handle the measurement
    }
}

Asynchronous batching

The record function will be called synchronously for every measurement emitted. Several measurements will be emitted at different times during a given operation invocation so it's highly recommended to publish them asynchronously. For example, you may create a publishing adapter class which uses a BlockingQueue to collect measurements synchronously and a periodic function to drain the queue and publish them:

data class Measurement(val name: String, val value: Double, val attributes: Attributes)

class MyS3Publisher(private val s3: S3Client, private val coroutineScope: CoroutineScope) {
    private val queue = LinkedBlockingQueue<Measurement>()

    init {
        // Launch a background coroutine to publish metrics every 5 seconds
        coroutineScope.launch {
            while (isActive) {
                delay(5.seconds)
                publish()
            }
        }
    }

    fun put(measurement: Measurement) {
        queue.add(measurement)
    }

    private suspend fun publish() {
        val metrics = mutableListOf<Measurement>()
            .also { queue.drainTo(it) }
            .joinToString(separator = "\n")

        s3.putObject {
            bucket = "foo"
            key = "bar"
            body = ByteStream.fromString(metrics)
        }
    }
}

Wiring it all together

Now that we have a publisher that can asynchronously batch requests, we just need to modify the providers and meters to accept an instance:

class MyTelemetryProvider(publisher: MyS3Publisher) : TelemetryProvider {
    override val meterProvider = MyMeterProvider(publisher)
    // other members omitted
}

class MyMeterProvider(private val publisher: MyS3Publisher) : MeterProvider {
    override fun getOrCreateMeter(scope: String) = MyMeter(publisher)
}

class MyMeter(private val publisher: MyS3Publisher) : Meter {
    override fun createDoubleHistogram(name: String, units: String?, description: String?) =
        MyDoubleHistogram(publisher, name, units)
    // other members omitted
}

class MyDoubleHistogram(
    private val publisher: MyS3Publisher,
    private val name: String,
    private val units: String?,
) : DoubleHistogram {
    override fun record(value: Double, attributes: Attributes, context: Context?) {
        publisher.put(Measurement(name, units, value, attributes))
    }
}

Final notes

Hopefully these examples will be enough to get you started. There are a few things to note about this post though:

• The examples given here are trivial implementations. They may not be suitable for your needs based on batching or timing requirements, coroutine management, metric metadata, etc. Please adapt them as appropriate for your use case.
• The telemetry API (including metrics) is still marked "experimental" in our API. You must opt in to using our ExperimentalApi annotation. We hope to stabilize the API soon and remove this requirement. If you have any feedback as part of your implementation, we'd love to hear it in this discussion or a feature request.
• The only telemetry provider included with the SDK currently is the OpenTelemetry provider, which simplifies publishing metrics and other telemetry signals to an OpenTelemetry backend (which itself supports downstream publishing to a variety of destinations). In the future we plan to offer direct integration with CloudWatch (feature request) and X-Ray (feature request).

Let us know if this sufficiently explains metrics publishing support or if there's anything that could be clarified!

Full code from the above examples

This is the full code from the examples above plus no-op implementations for measurements which were unused in this example:

@OptIn(ExperimentalApi::class)
class MyTelemetryProvider(publisher: MyS3Publisher) : TelemetryProvider {
    override val contextManager = ContextManager.None
    override val loggerProvider = LoggerProvider.None
    override val tracerProvider = TracerProvider.None
    override val meterProvider = MyMeterProvider(publisher)
}

class MyMeterProvider(private val publisher: MyS3Publisher) : MeterProvider {
    override fun getOrCreateMeter(scope: String) = MyMeter(publisher)
}

class MyMeter(private val publisher: MyS3Publisher) : Meter {
    override fun createDoubleHistogram(name: String, units: String?, description: String?) =
        MyDoubleHistogram(publisher, name, units)

    override fun createAsyncUpDownCounter(
        name: String,
        callback: LongUpDownCounterCallback,
        units: String?,
        description: String?
    ): AsyncMeasurementHandle = NoOpMeasurementHandle

    override fun createUpDownCounter(name: String, units: String?, description: String?): UpDownCounter =
        NoOpUpDownCounter

    override fun createMonotonicCounter(name: String, units: String?, description: String?): MonotonicCounter =
        NoOpMonotonicCounter

    override fun createLongHistogram(name: String, units: String?, description: String?): LongHistogram =
        NoOpLongHistogram

    override fun createLongGauge(
        name: String,
        callback: LongGaugeCallback,
        units: String?,
        description: String?
    ) = NoOpMeasurementHandle

    override fun createDoubleGauge(
        name: String,
        callback: DoubleGaugeCallback,
        units: String?,
        description: String?
    ) = NoOpMeasurementHandle
}

class MyDoubleHistogram(
    private val publisher: MyS3Publisher,
    private val name: String,
    private val units: String?,
) : DoubleHistogram {
    override fun record(value: Double, attributes: Attributes, context: Context?) {
        publisher.put(Measurement(name, units, value, attributes))
    }
}

object NoOpMeasurementHandle : AsyncMeasurementHandle {
    override fun stop() = Unit
}

object NoOpLongHistogram : LongHistogram {
    override fun record(value: Long, attributes: Attributes, context: Context?) = Unit
}

object NoOpMonotonicCounter : MonotonicCounter {
    override fun add(value: Long, attributes: Attributes, context: Context?) = Unit
}

object NoOpUpDownCounter : UpDownCounter {
    override fun add(value: Long, attributes: Attributes, context: Context?) = Unit
}

data class Measurement(val name: String, val units: String?, val value: Double, val attributes: Attributes)

class MyS3Publisher(private val s3: S3Client, private val coroutineScope: CoroutineScope) {
    private val queue = LinkedBlockingQueue<Measurement>()

    init {
        coroutineScope.launch {
            while (isActive) {
                delay(5.seconds)
                publish()
            }
        }
    }

    fun put(measurement: Measurement) {
        queue.add(measurement)
    }

    private suspend fun publish() {
        val metrics = mutableListOf<Measurement>()
            .also { queue.drainTo(it) }
            .joinToString(separator = "\n")

        s3.putObject {
            bucket = "foo"
            key = "bar"
            body = ByteStream.fromString(metrics)
        }
    }
}

[ianbotsf]

Yes, I definitely see the challenges with adapting your prior use of the Java SDK with the current Kotlin SDK telemetry APIs. Let me take your feedback and use case to some representatives from the other SDKs and attempt get some consensus.

For context, one of the reasons for the new telemetry API and its current experimental status is that we're attempting to coalesce the user experience across all AWS SDKs—not just for telemetry but in many other areas too. Whereas in the past it might've been easier to make a decision to change the metrics capabilities for one SDK, now we're trying to stick a common architecture in more areas.

[RodrasSilva]

Hello again, @ianbotsf. I hope you are well.
Did you happen to get to present our use cases to the other SDKs representatives? Is this something that different clients would want to be able to migrate from? It would be great to keep this logic in the new SDK.
Thanks 😄

[ianbotsf]

Hi @RodrasSilva, I'm working on a POC for supporting a MetricPublisher-like interface for the AWS SDK for Kotlin within the framework of the larger telemetry spec. My hunch so far is that it won't actually be too much work or require too much adjustment to the telemetry spec that SDKs are migrating to. I'll let you know when I have something more concrete.

[RodrasSilva]

Hello @ianbotsf. I am checking to see if you made any progress on your PoC and if you have anything more concrete I can use for testing. Thank you in advance!

[ianbotsf]

Hi @RodrasSilva, just back from an extended vacation. The PoC is still on my todo list and I can let you know when I have a branch ready to testing. Getting more eyes on it will be helpful! 🙂