From 16f8121a461568069235d5eaef56bcb422734599 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Fri, 13 Jan 2023 19:36:48 +0400 Subject: [PATCH] Stabilisation of upstreams: - reverted removing of configured upstreams in case of unavailability - small refactoring in grpc upstreams - fixing of bug of statuses subscription --- .../dshackle/rpc/SubscribeStatus.kt | 23 ++-- .../dshackle/startup/ConfiguredUpstreams.kt | 31 +++-- .../dshackle/upstream/DefaultUpstream.kt | 3 +- .../upstream/grpc/BitcoinGrpcUpstream.kt | 2 +- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 2 +- .../dshackle/upstream/grpc/GrpcUpstreams.kt | 109 +++++++----------- .../dshackle/rpc/SubscribeStatusSpec.groovy | 9 +- 7 files changed, 71 insertions(+), 108 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index 5b75f605..a0bbc122 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -22,33 +22,30 @@ import io.emeraldpay.dshackle.Chain import io.emeraldpay.dshackle.upstream.Multistream import io.emeraldpay.dshackle.upstream.MultistreamHolder import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service import reactor.core.publisher.Flux import reactor.core.publisher.Mono @Service class SubscribeStatus( - @Autowired private val multistreamHolder: MultistreamHolder + private val multistreamHolder: MultistreamHolder ) { fun subscribeStatus(requestMono: Mono): Flux { return requestMono.flatMapMany { req -> - // check status for all requested chains - val all = req.chainsList.map { - val chain = Chain.byId(it.number) - val up = multistreamHolder.getUpstream(chain) - if (up == null) { - // when the chain is not configured return just UNAVAILABLE - Mono.just(chainUnavailable(chain)).flux() - } else { - // when configured subscribe to its updates + if (req.chainsCount == 0) { + Flux.error(RuntimeException("empty chains list")) + } else { + // check status for all requested chains + val all = req.chainsList.map { + val chain = Chain.byId(it.number) + val up = multistreamHolder.getUpstream(chain) up.observeStatus().map { availability -> chainStatus(chain, availability, up) } } + Flux.merge(all) } - Flux.merge(all) } } @@ -74,6 +71,4 @@ class SubscribeStatus( .setQuorum(quorum) .build() } - - class ChainSubscription(val chain: Chain, val up: Multistream, val avail: UpstreamAvailability) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index c84b8637..5f0c3a9a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -47,11 +47,11 @@ 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.Scheduler import reactor.core.scheduler.Schedulers import java.net.URI import java.util.concurrent.Executor +import java.util.concurrent.Executors import java.util.concurrent.atomic.AtomicInteger import java.util.function.Function import kotlin.math.abs @@ -68,13 +68,13 @@ open class ConfiguredUpstreams( private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java) private var seq = AtomicInteger(0) - private val hashes: MutableMap = HashMap() + lateinit var grpcUpstreamsScheduler: Scheduler + 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") @@ -111,20 +111,8 @@ open class ConfiguredUpstreams( } } upstream?.let { - Flux.concat(Mono.just(UpstreamChangeEvent.ChangeType.ADDED), upstream.observeStatus()) - .distinctUntilChanged() - .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 ${it::class.java.simpleName}:${upstream.getId()} due to $status") - } - eventPublisher.publishEvent(UpstreamChangeEvent(chain, upstream, eventType)) - } - } + val event = UpstreamChangeEvent(chain, upstream, UpstreamChangeEvent.ChangeType.ADDED) + eventPublisher.publishEvent(event) } } } @@ -301,6 +289,12 @@ open class ConfiguredUpstreams( config: UpstreamsConfig.Upstream, options: UpstreamsConfig.Options ) { + if (!this::grpcUpstreamsScheduler.isInitialized) { + grpcUpstreamsScheduler = Schedulers.fromExecutorService( + Executors.newFixedThreadPool(2), + "GrpcUpstreamsStatuses" + ) + } val endpoint = config.connection!! val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}") val ds = GrpcUpstreams( @@ -313,6 +307,7 @@ open class ConfiguredUpstreams( fileResolver, endpoint.upstreamRating, config.labels, + grpcUpstreamsScheduler, channelExecutor ).apply { timeout = options.timeout diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index d9fb547d..2bcb861f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -80,7 +80,6 @@ abstract class DefaultUpstream( fun onStatus(value: BlockchainOuterClass.ChainStatus) { val available = value.availability - val quorum = value.quorum setStatus( if (available != null) UpstreamAvailability.fromGrpc(available.number) else UpstreamAvailability.UNAVAILABLE ) @@ -98,7 +97,7 @@ abstract class DefaultUpstream( } } - fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability { + private fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability { if (options.disableValidation == true) { // if we specifically told that this upstream should be _always valid_ then skip // the status calculation and trust the proposed value as is diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt index 27fe66e6..5eedd426 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt @@ -98,7 +98,7 @@ class BitcoinGrpcUpstream( } private val upstreamStatus = GrpcUpstreamStatus(overrideLabels) private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice()) - var timeout = Defaults.timeout + private val timeout = Defaults.timeout private var capabilities: Set = emptySet() override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index 142c3dd1..50abdf1d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -108,7 +108,7 @@ open class EthereumPosGrpcUpstream( private var capabilities: Set = emptySet() private val defaultReader: JsonRpcReader = client.getReader() - var timeout = Defaults.timeout + private val timeout = Defaults.timeout private val ethereumSubscriptions = EthereumDshackleIngressSubscription(chain, remote) override fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt index 78d31d18..27243cb3 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt @@ -16,7 +16,8 @@ */ package io.emeraldpay.dshackle.upstream.grpc -import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.BlockchainOuterClass.* +import io.emeraldpay.api.proto.Common import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.dshackle.BlockchainType import io.emeraldpay.dshackle.Chain @@ -26,6 +27,7 @@ import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics @@ -43,6 +45,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils import org.slf4j.LoggerFactory import reactor.core.Disposable import reactor.core.publisher.Flux +import reactor.core.scheduler.Scheduler import java.io.IOException import java.time.Duration import java.util.concurrent.Executor @@ -60,13 +63,14 @@ class GrpcUpstreams( private val fileResolver: FileResolver, private val nodeRating: Int, private val labels: UpstreamsConfig.Labels, - private val executor: Executor + private val chainStatusScheduler: Scheduler, + private val grpcExecutor: Executor ) { private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) var timeout = Defaults.timeout - private var client: ReactorBlockchainGrpc.ReactorBlockchainStub? = null + lateinit var client: ReactorBlockchainGrpc.ReactorBlockchainStub private val known = HashMap() private val lock = ReentrantLock() @@ -75,7 +79,7 @@ class GrpcUpstreams( // some messages are very large. many of them in megabytes, some even in gigabytes (ex. ETH Traces) .maxInboundMessageSize(Defaults.maxMessageSize) .enableRetry() - .executor(executor) + .executor(grpcExecutor) .maxRetryAttempts(3) if (auth != null && StringUtils.isNotEmpty(auth.ca)) { chanelBuilder @@ -91,9 +95,9 @@ class GrpcUpstreams( val statusSubscription = AtomicReference() - val updates = Flux.interval(Duration.ZERO, Duration.ofSeconds(20)) + return Flux.interval(Duration.ZERO, Duration.ofSeconds(20)) .flatMap { - client.describe(BlockchainOuterClass.DescribeRequest.newBuilder().build()) + client.describe(DescribeRequest.newBuilder().build()) }.onErrorContinue { t, _ -> if (ExceptionUtils.indexOfType(t, IOException::class.java) >= 0) { log.warn("gRPC upstream $host:$port is unavailable. (${t.javaClass}: ${t.message})") @@ -106,7 +110,10 @@ class GrpcUpstreams( }.flatMap { value -> processDescription(value) }.doOnNext { - val subscription = client.subscribeStatus(BlockchainOuterClass.StatusRequest.newBuilder().build()) + val subscription = client.subscribeStatus( + StatusRequest.newBuilder() + .addChains(Common.ChainRef.forNumber(it.chain.id)).build() + ).subscribeOn(chainStatusScheduler) .subscribe { value -> val chain = Chain.byId(value.chain.number) if (chain != Chain.UNSPECIFIED) { @@ -120,11 +127,9 @@ class GrpcUpstreams( }.doOnError { t -> log.error("Failed to process update from gRPC upstream $id", t) } - - return updates } - fun processDescription(value: BlockchainOuterClass.DescribeResponse): Flux { + private fun processDescription(value: DescribeResponse): Flux { log.info("Start processing grpc upstream description for $id with chains ${value.chainsList.map { it.chain.name }}") val current = value.chainsList.filter { Chain.byId(it.chain.number) != Chain.UNSPECIFIED @@ -154,7 +159,7 @@ class GrpcUpstreams( return Flux.fromIterable(removed + added) } - internal fun withTls(auth: AuthConfig.ClientTlsAuth): SslContext { + private fun withTls(auth: AuthConfig.ClientTlsAuth): SslContext { val sslContext = SslContextBuilder.forClient() .clientAuth(ClientAuth.REQUIRE) sslContext.trustManager(fileResolver.resolve(auth.ca!!).inputStream()) @@ -176,13 +181,31 @@ class GrpcUpstreams( return sslContext.build() } - fun getOrCreate(chain: Chain): UpstreamChangeEvent { + private val creators: Map DefaultUpstream> = mapOf( + BlockchainType.EVM_POW to { chain, rpcClient -> + EthereumGrpcUpstream(id, hash, role, chain, client, rpcClient, labels) + }, + BlockchainType.EVM_POS to { chain, rpcClient -> + EthereumPosGrpcUpstream(id, hash, role, chain, client, rpcClient, nodeRating, labels) + }, + BlockchainType.BITCOIN to { chain, rpcClient -> + BitcoinGrpcUpstream(id, role, chain, client, rpcClient, labels) + } + ) + + private fun getOrCreate(chain: Chain): UpstreamChangeEvent { + val metrics = makeMetrics(chain) + val creator = creators.getValue(BlockchainType.from(chain)) + return getOrCreate(chain, metrics, creator) + } + + private fun makeMetrics(chain: Chain): RpcMetrics { val metricsTags = listOf( Tag.of("upstream", id), Tag.of("chain", chain.chainCode) ) - val metrics = RpcMetrics( + return RpcMetrics( Timer.builder("upstream.grpc.conn") .description("Request time through a Dshackle/gRPC connection") .tags(metricsTags) @@ -193,68 +216,24 @@ class GrpcUpstreams( .tags(metricsTags) .register(Metrics.globalRegistry) ) - - val blockchainType = BlockchainType.from(chain) - if (blockchainType == BlockchainType.EVM_POW) { - return getOrCreateEthereum(chain, metrics) - } else if (blockchainType == BlockchainType.BITCOIN) { - return getOrCreateBitcoin(chain, metrics) - } else if (blockchainType == BlockchainType.EVM_POS) { - return getOrCreateEthereumPos(chain, metrics) - } else { - throw IllegalArgumentException("Unsupported blockchain: $chain") - } } - fun getOrCreateEthereum(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent { + private fun getOrCreate( + chain: Chain, + metrics: RpcMetrics, + creator: (chain: Chain, client: JsonRpcGrpcClient) -> DefaultUpstream + ): UpstreamChangeEvent { lock.withLock { val current = known[chain] return if (current == null) { - val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = EthereumGrpcUpstream(id, hash, role, chain, client!!, rpcClient, labels) - created.timeout = this.timeout + val rpcClient = JsonRpcGrpcClient(client, chain, metrics) + val created = creator(chain, rpcClient) known[chain] = created - created.start() + if (created is Lifecycle) created.start() UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED) } else { UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED) } } } - - fun getOrCreateEthereumPos(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent { - lock.withLock { - val current = known[chain] - return if (current == null) { - val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = EthereumPosGrpcUpstream(id, hash, role, chain, client!!, rpcClient, nodeRating, labels) - created.timeout = this.timeout - known[chain] = created - created.start() - UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED) - } else { - UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED) - } - } - } - - fun getOrCreateBitcoin(chain: Chain, metrics: RpcMetrics): UpstreamChangeEvent { - lock.withLock { - val current = known[chain] - return if (current == null) { - val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics) - val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient, labels) - created.timeout = this.timeout - known[chain] = created - created.start() - UpstreamChangeEvent(chain, created, UpstreamChangeEvent.ChangeType.ADDED) - } else { - UpstreamChangeEvent(chain, current, UpstreamChangeEvent.ChangeType.REVALIDATED) - } - } - } - - fun get(chain: Chain): DefaultUpstream { - return known[chain]!! - } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/SubscribeStatusSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/SubscribeStatusSpec.groovy index 082263c1..c8819aed 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/SubscribeStatusSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/SubscribeStatusSpec.groovy @@ -15,7 +15,7 @@ import java.time.Duration class SubscribeStatusSpec extends Specification { - def "gives UNAVAIL if non-configured chain is requested"() { + def "returns requested statuses"() { setup: def ethereumUp = Mock(Upstream) { _ * getStatus() >> UpstreamAvailability.OK @@ -28,17 +28,15 @@ class SubscribeStatusSpec extends Specification { def ups = Mock(MultistreamHolder) { _ * it.getAvailable() >> [Chain.ETHEREUM] 1 * it.getUpstream(Chain.ETHEREUM) >> ethereumUpAll - 1 * it.getUpstream(Chain.BITCOIN) >> null } def ctrl = new SubscribeStatus(ups) when: def req = BlockchainOuterClass.StatusRequest.newBuilder() .addChains(Common.ChainRef.CHAIN_ETHEREUM) - .addChains(Common.ChainRef.CHAIN_BITCOIN) .build() def act = ctrl.subscribeStatus(Mono.just(req)) - .take(2) + .take(1) // sort just for testing .sort(new Comparator() { @Override @@ -48,9 +46,6 @@ class SubscribeStatusSpec extends Specification { }) then: StepVerifier.create(act) - .expectNextMatches { - it.chainValue == Chain.BITCOIN.id && it.availability == BlockchainOuterClass.AvailabilityEnum.AVAIL_UNAVAILABLE - } .expectNextMatches { it.chainValue == Chain.ETHEREUM.id && it.availability == BlockchainOuterClass.AvailabilityEnum.AVAIL_OK }