OkHttp 지표 수집과 장애 격리

BulkheadInterceptor.kt

import io.github.resilience4j.bulkhead.Bulkhead
import okhttp3.Interceptor
import okhttp3.Response

class BulkheadInterceptor(private val bulkhead: Bulkhead) : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response =
        bulkhead.executeCallable { chain.proceed(chain.request()) }
}

CircuitBreakerInterceptor.kt

import io.github.resilience4j.circuitbreaker.CircuitBreaker
import okhttp3.Interceptor
import okhttp3.Response

class CircuitBreakerInterceptor(private val circuitBreaker: CircuitBreaker) : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response =
        circuitBreaker.executeCallable { chain.proceed(chain.request()) }
}

OkHttpMetrics.kt

import okhttp3.OkHttpClient

interface OkHttpMetrics {
    fun add(builder: OkHttpClient.Builder, name: String): OkHttpClient = builder.build()

    fun add(client: OkHttpClient, name: String) = Unit

    fun remove(client: OkHttpClient) = Unit
}

PrometheusOkHttpMetrics.kt

import io.prometheus.client.Collector
import io.prometheus.client.Counter
import io.prometheus.client.GaugeMetricFamily
import io.prometheus.client.Histogram
import okhttp3.Call
import okhttp3.EventListener
import okhttp3.OkHttpClient
import okhttp3.Protocol
import java.io.IOException
import java.net.InetAddress
import java.net.InetSocketAddress
import java.net.Proxy

class PrometheusOkHttpMetrics : Collector(), OkHttpMetrics {

    private val children: MutableMap<OkHttpClient, String> = mutableMapOf()

    private val callStartedTotal = Counter.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("call_started_total")
        .labelNames("name")
        .help("Total number of calls started")
        .create()

    private val callHandledTotal = Counter.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("call_handled_total")
        .labelNames("name", "error")
        .help("Total number of calls handled")
        .create()

    private val callHandlingSeconds = Histogram.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("call_handling_seconds")
        .labelNames("name")
        .help("Histogram of call handling latency")
        .create()

    private val requestStartedTotal = Counter.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("request_started_total")
        .labelNames("name")
        .help("Total number of requests started")
        .create()

    private val requestHandledTotal = Counter.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("request_handled_total")
        .labelNames("name", "http_status", "error")
        .help("Total number of requests handled")
        .create()

    private val requestHandlingSeconds = Histogram.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("request_handling_seconds")
        .labelNames("name")
        .help("Histogram of request handling latency")
        .create()

    private val connectHandledTotal = Counter.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("connect_handled_total")
        .labelNames("name", "server", "error")
        .help("Total number of connects handled")
        .create()

    private val connectHandlingSeconds = Histogram.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("connect_handling_seconds")
        .labelNames("name", "server")
        .help("Histogram of connect handling latency")
        .exponentialBuckets(0.001, 2.0, 12)
        .create()

    private val dnsHandlingSeconds = Histogram.build()
        .namespace("okhttp")
        .subsystem("client")
        .name("dns_handling_seconds")
        .labelNames("name", "server")
        .help("Histogram of DNS handling latency")
        .exponentialBuckets(0.001, 2.0, 12)
        .create()

    override fun add(builder: OkHttpClient.Builder, name: String): OkHttpClient {
        return builder
                .addNetworkInterceptor {
                    val requestStartAt = System.nanoTime()
                    requestStartedTotal.labels(name).inc()
                    try {
                        it.proceed(it.request())
                                .also {
                                    requestHandledTotal.labels(name, it.code.toString(), "").inc()
                                }
                    } catch (ioe: IOException) {
                        requestHandledTotal.labels(name, "", ioe.javaClass.simpleName).inc()
                        throw ioe
                    } finally {
                        requestHandlingSeconds.labels(name).observe((System.nanoTime() - requestStartAt) / NANOSECONDS_PER_SECOND)
                    }
                }
                .eventListenerFactory(object : EventListener.Factory {
                    override fun create(call: Call) =
                        object : EventListener() {
                            private var startedAt = 0L
                            private var dnsStartAt = 0L
                            private var connectStartAt = 0L

                            override fun callStart(call: Call) {
                                startedAt = System.nanoTime()
                                callStartedTotal.labels(name).inc()
                            }

                            override fun dnsStart(call: Call, domainName: String) {
                                dnsStartAt = System.nanoTime()
                            }

                            override fun dnsEnd(call: Call, domainName: String, inetAddressList: List<InetAddress>) {
                                dnsHandlingSeconds.labels(name, domainName).observe((System.nanoTime() - dnsStartAt) / NANOSECONDS_PER_SECOND)
                            }

                            override fun connectStart(call: Call, inetSocketAddress: InetSocketAddress, proxy: Proxy) {
                                connectStartAt = System.nanoTime()
                            }

                            override fun connectEnd(call: Call, inetSocketAddress: InetSocketAddress, proxy: Proxy, protocol: Protocol?) {
                                connectHandledTotal.labels(name, inetSocketAddress.hostName, "").inc()
                                connectHandlingSeconds.labels(name, inetSocketAddress.hostName).observe((System.nanoTime() - connectStartAt) / NANOSECONDS_PER_SECOND)
                            }

                            override fun connectFailed(call: Call, inetSocketAddress: InetSocketAddress, proxy: Proxy, protocol: Protocol?, ioe: IOException) {
                                connectHandledTotal.labels(name, inetSocketAddress.hostName, ioe.javaClass.simpleName).inc()
                                connectHandlingSeconds.labels(name, inetSocketAddress.hostName).observe((System.nanoTime() - connectStartAt) / NANOSECONDS_PER_SECOND)
                            }

                            override fun callEnd(call: Call) {
                                callHandledTotal.labels(name, "").inc()
                                callHandlingSeconds.labels(name).observe((System.nanoTime() - startedAt) / NANOSECONDS_PER_SECOND)
                            }

                            override fun callFailed(call: Call, ioe: IOException) {
                                callHandledTotal.labels(name, ioe.javaClass.simpleName).inc()
                                callHandlingSeconds.labels(name).observe((System.nanoTime() - startedAt) / NANOSECONDS_PER_SECOND)
                            }
                        }
                })
                .build()
                .also { add(it, name) }
    }

    override fun add(client: OkHttpClient, name: String) {
        children.put(client, name)
    }

    override fun remove(client: OkHttpClient) {
        children.remove(client)
    }

    override fun collect(): MutableList<MetricFamilySamples> {
        val mfs = mutableListOf<MetricFamilySamples>()

        mfs.addAll(callStartedTotal.collect())
        mfs.addAll(callHandledTotal.collect())
        mfs.addAll(callHandlingSeconds.collect())
        mfs.addAll(requestStartedTotal.collect())
        mfs.addAll(requestHandledTotal.collect())
        mfs.addAll(requestHandlingSeconds.collect())
        mfs.addAll(connectHandledTotal.collect())
        mfs.addAll(connectHandlingSeconds.collect())
        mfs.addAll(dnsHandlingSeconds.collect())

        val labelNames = listOf("name", "state")

        val currentMf = GaugeMetricFamily(
            "okhttp_client_connections_current", "Current connection count of a pool",
            labelNames
        )
        mfs.add(currentMf)

        children.forEach {
            val client = it.key
            val name = it.value
            val connPool = client.connectionPool
            val connCurrent = connPool.connectionCount()
            val connIdle = connPool.idleConnectionCount()
            currentMf.addMetric(listOf(name, "active"), (connCurrent - connIdle).toDouble())
            currentMf.addMetric(listOf(name, "idle"), connIdle.toDouble())
        }

        return mfs
    }
}

RetryInterceptor.kt

import io.github.resilience4j.retry.Retry
import okhttp3.Interceptor
import okhttp3.Response

class RetryInterceptor(private val retry: Retry) : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response =
        retry.executeCallable { chain.proceed(chain.request()) }
}