From 64e0ac58fda1e085dabcd6a19c38e7a53710f164 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 26 Jul 2019 22:29:19 -0400 Subject: [PATCH] solution: stream updates to blockchain status --- .../dshackle/rpc/SubscribeStatus.kt | 22 ++++++++ .../dshackle/upstream/AggregatedUpstreams.kt | 55 ++++++++++++++++++- .../dshackle/upstream/ChainUpstreams.kt | 7 ++- .../upstream/EthereumGrpcTransport.kt | 2 + .../dshackle/upstream/EthereumUpstream.kt | 10 ++++ .../dshackle/upstream/GrpcUpstream.kt | 17 ++++-- .../emeraldpay/dshackle/upstream/Upstream.kt | 2 + .../upstream/EthereumGrpcTransportSpec.groovy | 4 +- .../upstream/UpstreamAvailabilitySpec.groovy | 36 ++++++++++++ 9 files changed, 143 insertions(+), 12 deletions(-) create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/UpstreamAvailabilitySpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt index df1cef75..db7ab69a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt @@ -1,10 +1,14 @@ package io.emeraldpay.dshackle.rpc import io.emeraldpay.api.proto.BlockchainOuterClass +import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.Upstreams +import io.grpc.StatusRuntimeException import io.grpc.stub.StreamObserver import org.springframework.beans.factory.annotation.Autowired import org.springframework.stereotype.Service +import reactor.core.Disposable @Service class SubscribeStatus( @@ -13,6 +17,24 @@ class SubscribeStatus( fun subscribeStatus(request: BlockchainOuterClass.StatusRequest, responseObserver: StreamObserver) { upstreams.getAvailable().forEach { chain -> + var d: Disposable? = null + d = upstreams.getUpstream(chain)?.observeStatus()?.subscribe { availability -> + val chainStatus = BlockchainOuterClass.ChainStatus.newBuilder() + .setChain(Common.ChainRef.forNumber(chain.id)) + .setAvailable(availability == UpstreamAvailability.OK) + .setQuorum(0) + if (availability == UpstreamAvailability.OK) { + upstreams.getUpstream(chain)?.getOptions()?.let { opts -> + chainStatus.setQuorum(opts.quorum) + } + } + try { + responseObserver.onNext(chainStatus.build()) + } catch (e: StatusRuntimeException) { + // gRPC channel was closed + d?.dispose() + } + } } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt index fa1d6cef..d462b1d0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AggregatedUpstreams.kt @@ -1,12 +1,44 @@ package io.emeraldpay.dshackle.upstream -abstract class AggregatedUpstreams { +import io.emeraldpay.dshackle.config.UpstreamsConfig +import reactor.core.publisher.Flux +import java.time.Duration +import java.time.Instant +import java.util.concurrent.atomic.AtomicReference +import java.util.function.Predicate + +abstract class AggregatedUpstreams: Upstream { abstract fun getAll(): List abstract fun addUpstream(upstream: Upstream) abstract fun getApis(quorum: Int): Iterator - abstract fun getApi(): EthereumApi - abstract fun getHead(): EthereumHead + + override fun observeStatus(): Flux { + val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } } + return Flux.merge(upstreamsFluxes) + .filter(FilterBestAvailability()) + .map { it.status } + } + + override fun isAvailable(): Boolean { + return getAll().any { it.isAvailable() } + } + + override fun getStatus(): UpstreamAvailability { + val upstreams = getAll() + return if (upstreams.isEmpty()) UpstreamAvailability.UNAVAILABLE + else upstreams.map { it.getStatus() }.min()!! + } + + override fun getOptions(): UpstreamsConfig.Options { + val options = UpstreamsConfig.Options() + options.quorum = getAll().filter { + it.getStatus() == UpstreamAvailability.OK + }.sumBy { + it.getOptions().quorum + } + return options + } class SingleApi( private val quorumApi: QuorumApi @@ -47,6 +79,23 @@ abstract class AggregatedUpstreams { } throw IllegalStateException("No upstream API available") } + } + class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) + + class FilterBestAvailability(): Predicate { + private val lastRef = AtomicReference() + + override fun test(t: UpstreamStatus): Boolean { + val last = lastRef.get() + val changed = last == null + || t.status > last.status + || (last.upstream == t.upstream && t.status != last.status) + || last.ts.isBefore(Instant.now() - Duration.ofSeconds(60)) + if (changed) { + lastRef.set(t) + } + return changed + } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt index 824d74f2..24338354 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ChainUpstreams.kt @@ -1,5 +1,6 @@ package io.emeraldpay.dshackle.upstream +import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain import org.slf4j.LoggerFactory import java.io.Closeable @@ -57,9 +58,9 @@ class ChainUpstreams ( } fun printStatus() { - var height: Long = -1 + var height: Long? = null try { - height = head!!.getHead().block(Duration.ofSeconds(1))?.number ?: -1 + height = head!!.getHead().block(Duration.ofSeconds(1))?.number } catch (e: IllegalStateException) { //timout } catch (e: Exception) { @@ -70,7 +71,7 @@ class ChainUpstreams ( .map { "${it.key.name}/${it.value.size}" } .joinToString(",") - log.info("State of ${chain.chainCode}: height=$height, status=$statuses") + log.info("State of ${chain.chainCode}: height=${height ?: '?'}, status=$statuses") } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt index 109bc8f8..89844ce9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumGrpcTransport.kt @@ -16,6 +16,7 @@ import reactor.core.publisher.Mono import reactor.core.publisher.toFlux import reactor.util.function.Tuple3 import reactor.util.function.Tuples +import java.time.Duration import java.util.concurrent.CompletableFuture import java.util.function.Function @@ -97,6 +98,7 @@ class EthereumGrpcTransport( bi.onError(RpcException(-32603, "RPC response not received")) } } + .timeout(Duration.ofSeconds(15)) .toFuture() } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt index aca88ed3..15bc4fcc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EthereumUpstream.kt @@ -2,7 +2,11 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.BlockJson import org.slf4j.LoggerFactory +import reactor.core.publisher.Flux +import reactor.core.publisher.TopicProcessor import java.util.concurrent.atomic.AtomicReference class EthereumUpstream( @@ -24,6 +28,7 @@ class EthereumUpstream( private val validator = UpstreamValidator(this, options) private val status = AtomicReference(UpstreamAvailability.UNAVAILABLE) + private val statusStream: TopicProcessor = TopicProcessor.create() init { log.info("Configured for ${chain.chainName}") @@ -31,6 +36,7 @@ class EthereumUpstream( validator.start() .subscribe { status.set(it) + statusStream.onNext(it) } } @@ -42,6 +48,10 @@ class EthereumUpstream( return status.get() } + override fun observeStatus(): Flux { + return Flux.from(statusStream) + } + override fun getHead(): EthereumHead { return head } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt index 625d2493..844e9381 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/GrpcUpstream.kt @@ -38,6 +38,7 @@ open class GrpcUpstream( private var status = AtomicReference(UpstreamAvailability.UNAVAILABLE) private val head = Head(this) private val api: EthereumApi + private val statusStream: TopicProcessor = TopicProcessor.create() init { val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper) @@ -52,7 +53,7 @@ open class GrpcUpstream( .toMono() val retry: Function, Flux> = Function { - status.set(UpstreamAvailability.UNAVAILABLE) + setStatus(UpstreamAvailability.UNAVAILABLE) client.subscribeHead(chainRef) } @@ -80,14 +81,14 @@ open class GrpcUpstream( log.debug("New block ${block.number} on ${chain}") headBlock.set(block) streamBlocks.onNext(block) - status.set(UpstreamAvailability.OK) + setStatus(UpstreamAvailability.OK) } } fun init(conf: BlockchainOuterClass.DescribeChain) { val available = conf.available val quorum = conf.quorum - status.set( + setStatus( if (available && quorum > 0) UpstreamAvailability.OK else UpstreamAvailability.UNAVAILABLE ) } @@ -95,11 +96,15 @@ open class GrpcUpstream( fun onStatus(value: BlockchainOuterClass.ChainStatus) { val available = value.available val quorum = value.quorum - status.set( + setStatus( if (available && quorum > 0) UpstreamAvailability.OK else UpstreamAvailability.UNAVAILABLE ) } + private fun setStatus(value: UpstreamAvailability) { + status.set(value) + statusStream.onNext(value) + } // ------------------------------------------------------------------------------------------ @@ -111,6 +116,10 @@ open class GrpcUpstream( return status.get() } + override fun observeStatus(): Flux { + return Flux.from(statusStream) + } + override fun getHead(): EthereumHead { return head } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt index b84b2e1c..77a65180 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt @@ -1,10 +1,12 @@ package io.emeraldpay.dshackle.upstream import io.emeraldpay.dshackle.config.UpstreamsConfig +import reactor.core.publisher.Flux interface Upstream { fun isAvailable(): Boolean fun getStatus(): UpstreamAvailability + fun observeStatus(): Flux fun getHead(): EthereumHead fun getApi(): EthereumApi fun getOptions(): UpstreamsConfig.Options diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy index db51452c..35a729b2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/EthereumGrpcTransportSpec.groovy @@ -43,7 +43,7 @@ class EthereumGrpcTransportSpec extends Specification { def status = transport.execute(batch.items).get() then: - 1 * otherSideUpstreams.ethereumUpstream(Chain.ETHEREUM) >> otherSideAggr + 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr 1 * otherSideAggr.api >> otherSideApi status.failed == 0 status.succeed == 1 @@ -87,7 +87,7 @@ class EthereumGrpcTransportSpec extends Specification { def status = transport.execute(batch.items).get() then: - 1 * otherSideUpstreams.ethereumUpstream(Chain.ETHEREUM) >> otherSideAggr + 1 * otherSideUpstreams.getUpstream(Chain.ETHEREUM) >> otherSideAggr 1 * otherSideAggr.api >> otherSideApi status.failed == 0 status.succeed == 2 diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/UpstreamAvailabilitySpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/UpstreamAvailabilitySpec.groovy new file mode 100644 index 00000000..38c9af3e --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/UpstreamAvailabilitySpec.groovy @@ -0,0 +1,36 @@ +package io.emeraldpay.dshackle.upstream + +import spock.lang.Specification + +class UpstreamAvailabilitySpec extends Specification { + + def "Defined order"() { + expect: + UpstreamAvailability.OK.compareTo(UpstreamAvailability.IMMATURE) < 0 + UpstreamAvailability.IMMATURE.compareTo(UpstreamAvailability.SYNCING) < 0 + UpstreamAvailability.SYNCING.compareTo(UpstreamAvailability.LAGGING) < 0 + UpstreamAvailability.LAGGING.compareTo(UpstreamAvailability.UNAVAILABLE) < 0 + + UpstreamAvailability.OK.compareTo(UpstreamAvailability.UNAVAILABLE) < 0 + UpstreamAvailability.UNAVAILABLE.compareTo(UpstreamAvailability.OK) > 0 + } + + def "Sort array"() { + setup: + def items = [UpstreamAvailability.OK, UpstreamAvailability.SYNCING, UpstreamAvailability.IMMATURE, UpstreamAvailability.OK] + when: + Collections.sort(items) + then: + items == [UpstreamAvailability.OK, UpstreamAvailability.OK, UpstreamAvailability.IMMATURE, UpstreamAvailability.SYNCING] + } + + def "First is most available"() { + expect: + [UpstreamAvailability.UNAVAILABLE, UpstreamAvailability.OK].toSorted().first() == UpstreamAvailability.OK + [UpstreamAvailability.OK, UpstreamAvailability.UNAVAILABLE].toSorted().first() == UpstreamAvailability.OK + [UpstreamAvailability.OK, UpstreamAvailability.LAGGING].toSorted().first() == UpstreamAvailability.OK + [UpstreamAvailability.LAGGING, UpstreamAvailability.UNAVAILABLE].toSorted().first() == UpstreamAvailability.LAGGING + [UpstreamAvailability.OK, UpstreamAvailability.SYNCING].toSorted().first() == UpstreamAvailability.OK + [UpstreamAvailability.SYNCING, UpstreamAvailability.UNAVAILABLE].toSorted().first() == UpstreamAvailability.SYNCING + } +}