diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index 89b12488..eb08161b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt @@ -31,6 +31,8 @@ class Global { companion object { + var metricsExtended = false + @JvmStatic val objectMapper: ObjectMapper = createObjectMapper() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfig.kt index be76c87b..d232f571 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfig.kt @@ -30,6 +30,7 @@ class MonitoringConfig( } var enableJvm: Boolean = true + var enableExtended: Boolean = false data class PrometheusConfig( val enabled: Boolean, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfigReader.kt index 09947852..b8f4e319 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/MonitoringConfigReader.kt @@ -44,7 +44,9 @@ class MonitoringConfigReader: YamlConfigReader(), ConfigReader } val prometheus = readPrometheus(getMapping(input, "prometheus")) return MonitoringConfig(enabled, prometheus).also { conf -> - getValueAsBool(input, "enableJVM")?.let { conf.enableJvm = it } + getValueAsBool(input, "JVM")?.let { conf.enableJvm = it } + getValueAsBool(input, "jvm")?.let { conf.enableJvm = it } + getValueAsBool(input, "extended")?.let { conf.enableExtended = it } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt index eb05a97d..3b8fd1f7 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt @@ -16,7 +16,7 @@ package io.emeraldpay.dshackle.monitoring import com.sun.net.httpserver.HttpServer -import io.emeraldpay.dshackle.config.MainConfig +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.MonitoringConfig import io.micrometer.core.instrument.Meter import io.micrometer.core.instrument.Metrics @@ -32,9 +32,7 @@ import io.micrometer.prometheus.PrometheusConfig import io.micrometer.prometheus.PrometheusMeterRegistry import org.springframework.beans.factory.annotation.Autowired -import org.springframework.beans.factory.annotation.Value import java.io.IOException -import java.io.OutputStream import java.net.InetSocketAddress import javax.annotation.PostConstruct @@ -69,6 +67,9 @@ class MonitoringSetup( ProcessorMetrics().bindTo(Metrics.globalRegistry) JvmThreadMetrics().bindTo(Metrics.globalRegistry) } + if (monitoringConfig.enableExtended) { + Global.metricsExtended = true + } if (monitoringConfig.prometheus.enabled) { // use standard JVM server with a single thread blocking processing diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt index 446e4380..371bc376 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt @@ -16,11 +16,20 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.grpc.Chain +import io.micrometer.core.instrument.DistributionSummary +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag import org.reactivestreams.Subscriber import reactor.core.publisher.EmitterProcessor import reactor.core.publisher.Flux import java.time.Duration +import java.util.* +import java.util.concurrent.locks.Lock +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock import kotlin.math.max import kotlin.math.min import kotlin.math.pow @@ -28,9 +37,10 @@ import kotlin.math.roundToLong import kotlin.random.Random class FilteredApis( + val chain: Chain, private val allUpstreams: List, private val matcher: Selector.Matcher, - pos: Int, + private val pos: Int, /** * Limit of retries */ @@ -42,6 +52,8 @@ class FilteredApis( private const val DEFAULT_DELAY_STEP = 100 private const val MAX_WAIT_MILLIS = 5000L + private const val metricsCode = "select" + @JvmStatic fun startFrom(upstreams: List, pos: Int): List { return if (upstreams.size <= 1 || pos == 0) { @@ -51,14 +63,19 @@ class FilteredApis( upstreams.subList(safePosition, upstreams.size) + upstreams.subList(0, safePosition) } } + + private val metrics = EnumMap(Chain::class.java) + private val metricsSetup: Lock = ReentrantLock() } - constructor(allUpstreams: List, + constructor(chain: Chain, + allUpstreams: List, matcher: Selector.Matcher, - pos: Int) : this(allUpstreams, matcher, pos, 10, 7) + pos: Int) : this(chain, allUpstreams, matcher, pos, 10, 7) - constructor(allUpstreams: List, - matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10) + constructor(chain: Chain, + allUpstreams: List, + matcher: Selector.Matcher) : this(chain, allUpstreams, matcher, 0, 10, 10) private val delay: Int private val standardUpstreams: List @@ -87,6 +104,30 @@ class FilteredApis( .plus(standardUpstreams) .plus(fallbackUpstreams) + if (Global.metricsExtended) { + getMetrics(chain).let { monitoring -> + monitoring.countStd.record(standardUpstreams.size.toDouble()) + monitoring.countFallback.record(fallbackUpstreams.size.toDouble()) + } + } + } + + private fun getMetrics(chain: Chain): Monitoring { + val existing = metrics[chain] + return if (existing == null) { + metricsSetup.withLock { + val existingDoubleCheck = metrics[chain] + if (existingDoubleCheck != null) { + existingDoubleCheck + } else { + val created = Monitoring(chain) + metrics[chain] = created + created + } + } + } else { + existing + } } fun waitDuration(rawn: Long): Duration { @@ -110,9 +151,16 @@ class FilteredApis( .delaySubscription(waitDuration(r + 1)) }.let { Flux.concat(it) } - Flux.concat(first, retries) - .filter(Upstream::isAvailable) - .filter(matcher::matches) + var result = Flux.concat(first, retries) + + if (Global.metricsExtended) { + var count = 0 + result = result + .doOnNext { count++ } + .doFinally { metrics[chain]?.tried?.record(count.toDouble()) } + } + + result.filter { up -> up.isAvailable() && matcher.matches(up) } .zipWith(control) .map { it.t1 } .subscribe(subscriber) @@ -132,4 +180,19 @@ class FilteredApis( override fun toString(): String { return "Filter API: ${allUpstreams.size} upstreams with $matcher" } + + class Monitoring(chain: Chain) { + val countStd: DistributionSummary = DistributionSummary.builder("$metricsCode.exist") + .description("Count of available upstreams to select") + .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "std"))) + .register(Metrics.globalRegistry) + val countFallback: DistributionSummary = DistributionSummary.builder("$metricsCode.exist") + .description("Count of available fallback upstreams to select") + .tags(listOf(Tag.of("chain", chain.chainCode), Tag.of("role", "fallback"))) + .register(Metrics.globalRegistry) + val tried: DistributionSummary = DistributionSummary.builder("$metricsCode.tried") + .description("How many upstreams were checked") + .tags(listOf(Tag.of("chain", chain.chainCode))) + .register(Metrics.globalRegistry) + } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 0dc33572..5fe53e10 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -69,18 +69,25 @@ abstract class Multistream( init { UpstreamAvailability.values().forEach { status -> Metrics.gauge("$metrics.availability", - listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.toLowerCase())) - , this) { + listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.toLowerCase())), this) { upstreams.count { it.getStatus() == status }.toDouble() } } + Metrics.gauge("$metrics.connected", + listOf(Tag.of("chain", chain.chainCode)), this) { + upstreams.size.toDouble() + } + upstreams.forEach { up -> - Metrics.gauge("$metrics.lag", - listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", up.getId())) - , this) { - up.getLag().toDouble() - } + monitorUpstream(up) + } + } + + private fun monitorUpstream(upstream: Upstream) { + Metrics.gauge("$metrics.lag", + listOf(Tag.of("chain", chain.chainCode), Tag.of("upstream", upstream.getId())), upstream) { + it.getLag().toDouble() } } @@ -102,6 +109,7 @@ abstract class Multistream( upstreams.add(upstream) onUpstreamsUpdated() setHead(updateHead()) + monitorUpstream(upstream) } fun removeUpstream(id: String) { @@ -119,7 +127,7 @@ abstract class Multistream( if (seq >= Int.MAX_VALUE / 2) { seq = 0 } - return FilteredApis(upstreams, matcher, i) + return FilteredApis(chain, upstreams, matcher, i) } /**