StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stabilisation of upstreams:

Browse Source 
- reverted removing of configured upstreams in case of unavailability
- small refactoring in grpc upstreams
- fixing of bug of statuses subscription
This commit is contained in:
a10zn8
2023-01-13 19:36:48 +04:00
parent 36e28af874
commit 16f8121a46
7 changed files with 71 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/main/kotlin/io/emeraldpay/dshackle/rpc/SubscribeStatus.kt
View File

@@ -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<BlockchainOuterClass.StatusRequest>): Flux<BlockchainOuterClass.ChainStatus> {
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)
}

31
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -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<Byte, Boolean> = 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<UpstreamsConfig.GrpcConnection>,
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

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
View File

@@ -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

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -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<Capability> = emptySet()
override fun getBlockchainApi(): ReactorBlockchainGrpc.ReactorBlockchainStub {

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
View File

@@ -108,7 +108,7 @@ open class EthereumPosGrpcUpstream(
private var capabilities: Set<Capability> = 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() {

109
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -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<Chain, DefaultUpstream>()
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<Disposable>()
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<UpstreamChangeEvent> {
private fun processDescription(value: DescribeResponse): Flux<UpstreamChangeEvent> {
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<BlockchainType, (chain: Chain, client: JsonRpcGrpcClient) -> 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]!!
}
}

9
src/test/groovy/io/emeraldpay/dshackle/rpc/SubscribeStatusSpec.groovy
View File

@@ -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<BlockchainOuterClass.ChainStatus>() {
@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
}