allegro-internal/flex-roadmap#819 reactor metrics

Author: nastassia-dailidava

envoy-control-core/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/utils/Metrics.kt

@@ -6,7 +6,7 @@ import io.micrometer.core.instrument.noop.NoopTimer
 
 val noopTimer = NoopTimer(Meter.Id("", Tags.empty(), null, null, Meter.Type.TIMER))
 const val REACTOR_METRIC = "reactor.stats"
-const val REACTOR_TOTAL_METRIC = "reactor.stats.total"
+const val REACTOR_DISCARDED_METRIC = "reactor.discarded"
 const val SERVICES_STATE_METRIC = "services.state"
 const val SERVICES_STATE_ERRORS_METRIC = "services.state.errors.total"
 const val SNAPSHOT_METRIC = "snapshot"

envoy-control-core/src/main/kotlin/pl/allegro/tech/servicemesh/envoycontrol/utils/ReactorUtils.kt

@@ -1,6 +1,7 @@
 package pl.allegro.tech.servicemesh.envoycontrol.utils
 
 import io.micrometer.core.instrument.MeterRegistry
+import io.micrometer.core.instrument.Tags
 import org.reactivestreams.Subscription
 import org.slf4j.LoggerFactory
 import reactor.core.Disposable
@@ -11,6 +12,7 @@ import reactor.core.scheduler.Scheduler
 import reactor.core.scheduler.Schedulers
 import java.time.Duration
 import java.util.concurrent.TimeUnit
+import kotlin.streams.asSequence
 
 private val logger = LoggerFactory.getLogger("pl.allegro.tech.servicemesh.envoycontrol.utils.ReactorUtils")
 private val defaultScheduler by lazy { Schedulers.newSingle("reactor-utils-scheduler") }
@@ -50,8 +52,7 @@ fun <T> Flux<T>.measureBuffer(
 fun <T> Flux<T>.measureDiscardedItems(name: String, meterRegistry: MeterRegistry): Flux<T> = this
     .doOnDiscard(Any::class.java) {
         meterRegistry.counter(
-            REACTOR_METRIC,
-            METRIC_TYPE_TAG, "discarded-items",
+            REACTOR_DISCARDED_METRIC,
             METRIC_EMITTER_TAG, name
         ).increment()
     }
@@ -110,7 +111,12 @@ private fun measureQueueSubscriptionBuffer(
     name: String,
     meterRegistry: MeterRegistry
 ) {
-    logger.info("subscription $subscription name: $name meterRegistry: $meterRegistry")
+    meterRegistry.gauge(
+        REACTOR_METRIC,
+        Tags.of(METRIC_TYPE_TAG, "buffer-size", METRIC_EMITTER_TAG, name),
+        subscription,
+        queueSubscriptionBufferExtractor
+    )
 }
 
 private fun measureScannableBuffer(
@@ -119,7 +125,49 @@ private fun measureScannableBuffer(
     innerSources: Int,
     meterRegistry: MeterRegistry
 ) {
-  logger.info("scannable $scannable name: $name innerSources: $innerSources meterRegistry: $meterRegistry")
+    val buffered = scannable.scan(Scannable.Attr.BUFFERED)
+    if (buffered == null) {
+        logger.error(
+            "Cannot register metric $REACTOR_METRIC 'with $METRIC_EMITTER_TAG: $name'. Buffer size not available. " +
+                "Use measureBuffer() only on supported reactor operators"
+        )
+        return
+    }
+
+    meterRegistry.gauge(
+        REACTOR_METRIC,
+        Tags.of(METRIC_TYPE_TAG, "buffer-size", METRIC_EMITTER_TAG, name),
+        scannable,
+        scannableBufferExtractor
+    )
+
+    /**
+     * Special case for FlatMap derived operators like merge(). The main buffer attribute doesn't return actual
+     * buffer (that is controlled by `prefetch` parameter) size. Instead it returns simply number of connected sources.
+     *
+     * To access actual buffer size, we need to extract it from inners(). We don't know how many sources will
+     * be available, so it must be stated explicitly as innerSources parameter.
+     */
+    for (i in 0 until innerSources) {
+        meterRegistry.gauge(
+            REACTOR_METRIC,
+            Tags.of(METRIC_TYPE_TAG, "buffer-size", METRIC_EMITTER_TAG, "${(name)}_$i"),
+            scannable,
+            innerBufferExtractor(i)
+        )
+    }
+}
+
+private val scannableBufferExtractor = { s: Scannable -> s.scan(Scannable.Attr.BUFFERED)?.toDouble() ?: -1.0 }
+private fun innerBufferExtractor(index: Int) = { s: Scannable ->
+    s.inners().asSequence()
+        .elementAtOrNull(index)
+        ?.let(scannableBufferExtractor)
+        ?: -1.0
+}
+
+private val queueSubscriptionBufferExtractor = { s: Fuseable.QueueSubscription<*> ->
+    s.size.toDouble()
 }
 
 sealed class ParallelizableScheduler

envoy-control-core/src/test/kotlin/pl/allegro/tech/servicemesh/envoycontrol/utils/ReactorUtilsTest.kt

@@ -3,7 +3,6 @@ package pl.allegro.tech.servicemesh.envoycontrol.utils
 import io.micrometer.core.instrument.Tags
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import org.assertj.core.api.Assertions.assertThat
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.fail
 import org.testcontainers.shaded.org.awaitility.Awaitility
@@ -13,7 +12,6 @@ import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import java.util.function.BiFunction
 
-@Disabled
 class ReactorUtilsTest {
 
     @Test