From 8c52a9cd3460531c286f14aa0b0f8aa23a27f916 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Mon, 12 Dec 2022 11:41:40 +0300 Subject: [PATCH 1/6] remove unavailable upstream from multistream` --- .../dshackle/startup/ConfiguredUpstreams.kt | 19 +++- .../dshackle/upstream/Multistream.kt | 92 +++++++++++-------- .../upstream/ethereum/DefaultEthereumHead.kt | 2 +- 3 files changed, 74 insertions(+), 39 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index c9fb7fce..befc698a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -48,6 +48,9 @@ import org.springframework.boot.ApplicationArguments import org.springframework.boot.ApplicationRunner import org.springframework.context.ApplicationEventPublisher import org.springframework.stereotype.Component +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers import java.net.URI import java.util.concurrent.atomic.AtomicInteger import java.util.function.Function @@ -110,8 +113,20 @@ open class ConfiguredUpstreams( } } upstream?.let { - val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED) - eventPublisher.publishEvent(event) + Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus()) + .distinctUntilChanged() + .subscribeOn(Schedulers.boundedElastic()) + .subscribe { status -> + when (status) { + UpstreamAvailability.UNAVAILABLE -> UpstreamChangeEvent.ChangeType.REMOVED + else -> UpstreamChangeEvent.ChangeType.REVALIDATED + }.let { eventType -> + if (eventType == UpstreamChangeEvent.ChangeType.REMOVED) { + log.warn("Remove upstream ${upstream.getId()} due to $it") + } + eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType)) + } + } } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 6d7d64df..3defc2ba 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -43,6 +43,9 @@ import java.util.* import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock import java.util.function.Predicate +import kotlin.collections.ArrayList +import kotlin.collections.HashMap +import kotlin.collections.HashSet import kotlin.concurrent.withLock /** @@ -64,6 +67,7 @@ abstract class Multistream( private var cacheSubscription: Disposable? = null private val reconfigLock = ReentrantLock() + private val eventLock = ReentrantLock() private var callMethods: CallMethods? = null private var callMethodsFactory: Factory = Factory { return@Factory callMethods ?: throw FunctorException("Not initialized yet") @@ -72,6 +76,7 @@ abstract class Multistream( protected var lagObserver: HeadLagObserver? = null private var subscription: Disposable? = null private var capabilities: Set = emptySet() + private val removed: MutableMap = HashMap() init { UpstreamAvailability.values().forEach { status -> @@ -119,19 +124,32 @@ abstract class Multistream( /** * Add an upstream */ - fun addUpstream(upstream: Upstream) { - upstreams.add(upstream) - onUpstreamsUpdated() - setHead(updateHead()) - monitorUpstream(upstream) - } - - fun removeUpstream(id: String) { - if (upstreams.removeIf { it.getId() == id }) { - onUpstreamsUpdated() - setHead(updateHead()) + fun addUpstream(upstream: Upstream): Boolean = + upstreams.none { + it.getId() == upstream.getId() + }.also { + if (it) { + upstreams.add(upstream) + removed.remove(upstream.getId()) + onUpstreamsUpdated() + setHead(updateHead()) + monitorUpstream(upstream) + } } - } + + fun removeUpstream(id: String): Boolean = + upstreams.removeIf { up -> + up.takeIf { up.getId() == id } + ?.also { removed[id] = up } + ?.let { true } + ?: false + }.also { + if (it) { + onUpstreamsUpdated() + setHead(updateHead()) + } + } + /** * Get a source for direct APIs @@ -301,23 +319,21 @@ abstract class Multistream( } catch (e: Exception) { log.warn("Head processing error: ${e.javaClass} ${e.message}") } - val statuses = upstreams.map { it.getStatus() } + val statuses = upstreams.asSequence().plus(removed.values).map { it.getStatus() } .groupBy { it } .map { "${it.key.name}/${it.value.size}" } .joinToString(",") - val lag = upstreams - .map { - // by default, when no lag is available it uses Long.MAX_VALUE, and it doesn't make sense to print - // status with such value. use NA (as Not Available) instead - val value = it.getLag() - if (value == Long.MAX_VALUE) { - "NA" - } else { - value.toString() - } + val lag = upstreams.plus(removed.values).joinToString(", ") { + // by default, when no lag is available it uses Long.MAX_VALUE, and it doesn't make sense to print + // status with such value. use NA (as Not Available) instead + val value = it.getLag() + if (value == Long.MAX_VALUE) { + "NA" + } else { + value.toString() } - .joinToString(", ") - val weak = upstreams + } + val weak = upstreams.plus(removed.values) .filter { it.getStatus() != UpstreamAvailability.OK } .joinToString(", ") { it.getId() } @@ -333,18 +349,22 @@ abstract class Multistream( fun onUpstreamChange(event: UpstreamChangeEvent) { val chain = event.chain if (this.chain == chain) { - if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { - removeUpstream(event.upstream.getId()) - log.error("Upstream ${event.upstream.getId()} with chain $chain has been removed") - } else { - if (event.upstream is CachesEnabled) { - event.upstream.setCaches(caches) + eventLock.withLock { + if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { + removeUpstream(event.upstream.getId()).takeIf { it }?.let { + log.error("Upstream ${event.upstream.getId()} with chain $chain has been removed") + } + } else { + if (event.upstream is CachesEnabled) { + event.upstream.setCaches(caches) + } + addUpstream(event.upstream).takeIf { it }?.let { + if (!started) { + start() + } + log.error("Upstream ${event.upstream.getId()} with chain $chain has been added") + } } - addUpstream(event.upstream) - if (!started) { - start() - } - log.error("Upstream ${event.upstream.getId()} with chain $chain has been added") } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt index b47b064e..02fc4068 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DefaultEthereumHead.kt @@ -61,7 +61,7 @@ open class DefaultEthereumHead( BlockContainer.fromEthereumJson(it.getResult(), upstreamId) } .onErrorResume { err -> - log.debug("Failed to fetch latest block: ${err.message}") + log.error("Failed to fetch latest block: ${err.message}") Mono.empty() } } From dd147558bfb72ef12fadcd3203ca184f72dd9c2f Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 12:02:12 +0300 Subject: [PATCH 2/6] added metric --- .../dshackle/upstream/AbstractHead.kt | 29 +++++++++++++------ 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 7c7bb5b9..84bd3bef 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -15,8 +15,14 @@ */ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Gauge +import io.micrometer.core.instrument.Meter +import io.micrometer.core.instrument.Metrics +import io.micrometer.core.instrument.Tag import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux @@ -28,6 +34,7 @@ import reactor.core.scheduler.Schedulers import reactor.kotlin.core.publisher.toMono import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.locks.ReentrantLock abstract class AbstractHead @JvmOverloads constructor( @@ -48,19 +55,23 @@ abstract class AbstractHead @JvmOverloads constructor( private var lastHeadUpdated = 0L private val lock = ReentrantLock() + init { + val state = AtomicBoolean(false) + Gauge.builder("stuck_head", state) { + if (it.get()) 1.0 else 0.0 + } + .tag("upstream", upstreamId) + .tag("class", this.javaClass.simpleName) + .register(Metrics.globalRegistry) + Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( { val delay = System.currentTimeMillis() - lastHeadUpdated - if (delay > awaitHeadTimeoutMs) { - log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass} - restart") - if (lock.tryLock()) { - try { - start() - } finally { - lock.unlock() - } - } + val delayed = delay > awaitHeadTimeoutMs + state.set(delayed) + if (delayed) { + log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass}") } }, 300, 30, TimeUnit.SECONDS ) From f2c5142b1d5c2e3b1fec9342d5dfcc9627b01703 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 12:15:06 +0300 Subject: [PATCH 3/6] added metrics + fix code style --- .../emeraldpay/dshackle/upstream/AbstractHead.kt | 14 ++++---------- .../io/emeraldpay/dshackle/upstream/Multistream.kt | 6 +----- 2 files changed, 5 insertions(+), 15 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 84bd3bef..2e6aa454 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -15,14 +15,10 @@ */ package io.emeraldpay.dshackle.upstream -import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice -import io.micrometer.core.instrument.Counter import io.micrometer.core.instrument.Gauge -import io.micrometer.core.instrument.Meter import io.micrometer.core.instrument.Metrics -import io.micrometer.core.instrument.Tag import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux @@ -55,16 +51,14 @@ abstract class AbstractHead @JvmOverloads constructor( private var lastHeadUpdated = 0L private val lock = ReentrantLock() - init { val state = AtomicBoolean(false) Gauge.builder("stuck_head", state) { if (it.get()) 1.0 else 0.0 - } - .tag("upstream", upstreamId) - .tag("class", this.javaClass.simpleName) - .register(Metrics.globalRegistry) - + }.tag("upstream", upstreamId).tag("class", this.javaClass.simpleName).register(Metrics.globalRegistry) + Gauge.builder("current_head", forkChoice) { + it.getHead()?.height?.toDouble() ?: 0.0 + }.tag("upstream", upstreamId).tag("class", this.javaClass.simpleName).register(Metrics.globalRegistry) Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( { val delay = System.currentTimeMillis() - lastHeadUpdated diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 3defc2ba..217bbcb0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -39,13 +39,10 @@ import reactor.core.publisher.Flux import reactor.core.publisher.Mono import java.time.Duration import java.time.Instant -import java.util.* +import java.util.Locale import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.locks.ReentrantLock import java.util.function.Predicate -import kotlin.collections.ArrayList -import kotlin.collections.HashMap -import kotlin.collections.HashSet import kotlin.concurrent.withLock /** @@ -150,7 +147,6 @@ abstract class Multistream( } } - /** * Get a source for direct APIs */ From ae27c755dff9514b84a585b78cd4ce9be54a7430 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 12:52:31 +0300 Subject: [PATCH 4/6] added metrics instance id to log --- src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 217bbcb0..5af3a74e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -333,7 +333,8 @@ abstract class Multistream( .filter { it.getStatus() != UpstreamAvailability.OK } .joinToString(", ") { it.getId() } - log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak]") + val instance = System.identityHashCode(this).toString(16) + log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=[$statuses], lag=[$lag], weak=[$weak] ($instance)") } fun test(event: UpstreamChangeEvent): Boolean { From 68d94f1295c35df78edb2145cb5bdb00da1eedd2 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 12:54:01 +0300 Subject: [PATCH 5/6] better log --- .../kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 5af3a74e..676102ca 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -349,7 +349,7 @@ abstract class Multistream( eventLock.withLock { if (event.type == UpstreamChangeEvent.ChangeType.REMOVED) { removeUpstream(event.upstream.getId()).takeIf { it }?.let { - log.error("Upstream ${event.upstream.getId()} with chain $chain has been removed") + log.warn("Upstream ${event.upstream.getId()} with chain $chain has been removed") } } else { if (event.upstream is CachesEnabled) { @@ -359,7 +359,7 @@ abstract class Multistream( if (!started) { start() } - log.error("Upstream ${event.upstream.getId()} with chain $chain has been added") + log.info("Upstream ${event.upstream.getId()} with chain $chain has been added") } } } From e86a976305ba24488c4a94a875dafe48e0da59b4 Mon Sep 17 00:00:00 2001 From: Maksim Fomenkov Date: Tue, 13 Dec 2022 14:50:00 +0300 Subject: [PATCH 6/6] fix review comments --- .../emeraldpay/dshackle/startup/ConfiguredUpstreams.kt | 10 +++------- .../io/emeraldpay/dshackle/upstream/Multistream.kt | 9 +++++---- 2 files changed, 8 insertions(+), 11 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index befc698a..4dbc5870 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -72,6 +72,7 @@ open class ConfiguredUpstreams( override fun run(args: ApplicationArguments) { log.debug("Starting upstreams") val defaultOptions = buildDefaultOptions(config) + val observerScheduler = Schedulers.newParallel("status-observer", 3) config.upstreams.forEach { up -> if (!up.isEnabled) { log.debug("Upstream ${up.id} is disabled") @@ -106,23 +107,18 @@ open class ConfiguredUpstreams( options ) } - - else -> { - log.error("Chain is unsupported: ${up.chain}") - return@forEach - } } upstream?.let { Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus()) .distinctUntilChanged() - .subscribeOn(Schedulers.boundedElastic()) + .subscribeOn(observerScheduler) .subscribe { status -> when (status) { UpstreamAvailability.UNAVAILABLE -> UpstreamChangeEvent.ChangeType.REMOVED else -> UpstreamChangeEvent.ChangeType.REVALIDATED }.let { eventType -> if (eventType == UpstreamChangeEvent.ChangeType.REMOVED) { - log.warn("Remove upstream ${upstream.getId()} due to $it") + log.warn("Remove upstream ${it::class.java.simpleName}:${upstream.getId()} due to $status") } eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType)) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 676102ca..3f53fe5f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -136,10 +136,11 @@ abstract class Multistream( fun removeUpstream(id: String): Boolean = upstreams.removeIf { up -> - up.takeIf { up.getId() == id } - ?.also { removed[id] = up } - ?.let { true } - ?: false + (up.getId() == id).also { + if (it) { + removed[id] = up + } + } }.also { if (it) { onUpstreamsUpdated()