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

problem: doesn't add single dshackle-based upstream

Browse Source
This commit is contained in:
Igor Artamonov
2020-08-20 21:01:15 -04:00
parent 9a22b0a020
commit d9fd0619b3
11 changed files with 178 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/kotlin/io/emeraldpay/dshackle/rpc/Describe.kt
View File

@@ -21,8 +21,6 @@ import io.emeraldpay.api.proto.Common
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.grpc.EthereumGrpcUpstream
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service
import reactor.core.publisher.Mono
@@ -47,12 +45,8 @@ class Describe(
chainUpstreams.getAll().let { ups ->
ups.forEach { up ->
val nodes = QuorumForLabels()
if (up is EthereumUpstream) {
nodes.add(up.node)
} else if (up is BitcoinUpstream) {
nodes.add(up.node)
} else if (up is EthereumGrpcUpstream) {
nodes.add(up.getNodes())
if (up is DefaultUpstream) {
nodes.add(up.getQuorumByLabel())
}
nodes.getAll().forEach { node ->
val nodeDetails = BlockchainOuterClass.NodeDetails.newBuilder()

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

@@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.upstream.CurrentMultistreamHolder
import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinUpstream
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
import io.emeraldpay.dshackle.upstream.grpc.GrpcUpstreams
@@ -182,7 +183,7 @@ open class ConfiguredUpstreams(
}
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val ethereumUpstream = EthereumUpstream(
val ethereumUpstream = EthereumRpcUpstream(
config.id!!,
chain, directApi, wsFactoryApi,
options, config.role,

9
src/main/kotlin/io/emeraldpay/dshackle/startup/QuorumForLabels.kt
View File

@@ -26,11 +26,15 @@ import kotlin.concurrent.write
/**
* Summary details over few upstream nodes. Provides aggregate quorum for nodes with particular label
*/
class QuorumForLabels {
class QuorumForLabels() {
private val lock = ReentrantReadWriteLock()
private val nodes = ArrayList<QuorumItem>()
constructor(node: QuorumItem) : this() {
add(node)
}
fun add(node: QuorumItem) {
lock.read {
val existing = nodes.find { it.labels == node.labels }
@@ -59,6 +63,9 @@ class QuorumForLabels {
*/
class QuorumItem(val quorum: Int, val labels: UpstreamsConfig.Labels) {
companion object {
fun empty(): QuorumItem {
return QuorumItem(0, UpstreamsConfig.Labels())
}
}
}

41
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -50,8 +50,6 @@ class CurrentMultistreamHolder(
private val log = LoggerFactory.getLogger(CurrentMultistreamHolder::class.java)
private val objectMapper: ObjectMapper = Global.objectMapper
private val chainMapping = ConcurrentHashMap<Chain, Multistream>()
private val chainsBus = TopicProcessor.create<Chain>()
private val callTargets = HashMap<Chain, CallMethods>()
@@ -59,27 +57,32 @@ class CurrentMultistreamHolder(
fun update(change: UpstreamChange) {
updateLock.withLock {
log.debug("Upstream update: ${change.type} ${change.chain} via ${change.upstream.getId()}")
val chain = change.chain
when (BlockchainType.fromBlockchain(chain)) {
BlockchainType.ETHEREUM -> {
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain] as Multistream?
val factory = Callable {
EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
try {
when (BlockchainType.fromBlockchain(chain)) {
BlockchainType.ETHEREUM -> {
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain] as Multistream?
val factory = Callable {
EthereumMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
}
processUpdate(change, up, current, factory)
}
processUpdate(change, up, current, factory)
}
BlockchainType.BITCOIN -> {
val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping[chain] as Multistream?
val factory = Callable {
BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
BlockchainType.BITCOIN -> {
val up = change.upstream.cast(BitcoinUpstream::class.java)
val current = chainMapping[chain] as Multistream?
val factory = Callable {
BitcoinMultistream(chain, ArrayList(), cachesFactory.getCaches(chain)) as Multistream
}
processUpdate(change, up, current, factory)
}
else -> {
log.error("Update for unsupported chain: $chain")
}
processUpdate(change, up, current, factory)
}
else -> {
log.error("Update for unsupported chain: $chain")
}
} catch (e: Throwable) {
log.error("Failed to update upstream", e)
}
}
}

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

@@ -17,6 +17,7 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import reactor.core.publisher.Flux
import reactor.extra.processor.TopicProcessor
@@ -28,11 +29,15 @@ abstract class DefaultUpstream(
defaultAvail: UpstreamAvailability,
private val options: UpstreamsConfig.Options,
private val role: UpstreamsConfig.UpstreamRole,
private val targets: CallMethods?
private val targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?
) : Upstream {
constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets)
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, QuorumForLabels.QuorumItem.empty())
constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, node: QuorumForLabels.QuorumItem?) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
@@ -95,5 +100,12 @@ abstract class DefaultUpstream(
return targets ?: throw IllegalStateException("Methods are not set")
}
private val quorumByLabel = node?.let { QuorumForLabels(it) }
?: QuorumForLabels(QuorumForLabels.QuorumItem.empty())
open fun getQuorumByLabel(): QuorumForLabels {
return quorumByLabel
}
class Status(val lag: Long, val avail: UpstreamAvailability, val status: UpstreamAvailability)
}

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
View File

@@ -35,9 +35,9 @@ open class BitcoinUpstream(
private val directApi: Reader<JsonRpcRequest, JsonRpcResponse>,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
val node: QuorumForLabels.QuorumItem,
node: QuorumForLabels.QuorumItem,
callMethods: CallMethods
) : DefaultUpstream(id, options, role, callMethods), Lifecycle {
) : DefaultUpstream(id, options, role, callMethods, node), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinUpstream::class.java)

117
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt Normal file
View File

@@ -0,0 +1,117 @@
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import java.time.Duration
open class EthereumRpcUpstream(
id: String,
val chain: Chain,
private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
private val ethereumWsFactory: EthereumWsFactory? = null,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
private val node: QuorumForLabels.QuorumItem,
targets: CallMethods
) : EthereumUpstream(id, options, role, targets, node), Upstream, CachesEnabled, Lifecycle {
constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>) :
this(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD,
QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
DirectCallMethods())
private val log = LoggerFactory.getLogger(EthereumRpcUpstream::class.java)
private val head: Head = this.createHead()
private var validatorSubscription: Disposable? = null
override fun setCaches(caches: Caches) {
if (head is CachesEnabled) {
head.setCaches(caches)
}
}
override fun start() {
log.info("Configured for ${chain.chainName}")
if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
log.warn("Disable validation for upstream ${this.getId()}")
this.setLag(0)
this.setStatus(UpstreamAvailability.OK)
} else {
log.debug("Start validation for upstream ${this.getId()}")
val validator = EthereumUpstreamValidator(this, getOptions())
validatorSubscription = validator.start()
.subscribe(this::setStatus)
}
}
override fun isRunning(): Boolean {
return true
}
override fun stop() {
validatorSubscription?.dispose()
validatorSubscription = null
if (head is Lifecycle) {
head.stop()
}
}
open fun createHead(): Head {
return if (ethereumWsFactory != null) {
val ws = ethereumWsFactory.create(this).apply {
connect()
}
val wsHead = EthereumWsHead(ws).apply {
start()
}
// receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead = EthereumRpcHead(getApi(), Duration.ofSeconds(60)).apply {
start()
}
MergedHead(listOf(rpcHead, wsHead)).apply {
start()
}
} else {
log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC")
EthereumRpcHead(getApi()).apply {
start()
}
}
}
override fun getHead(): Head {
return head
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
return directReader
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return listOf(node.labels)
}
@Suppress("unchecked")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return this as T
}
}

113
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
View File

@@ -16,120 +16,15 @@
*/
package io.emeraldpay.dshackle.upstream.ethereum
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CachesEnabled
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.grpc.Chain
import org.slf4j.LoggerFactory
import org.springframework.context.Lifecycle
import reactor.core.Disposable
import reactor.core.publisher.Mono
import java.time.Duration
open class EthereumUpstream(
abstract class EthereumUpstream(
id: String,
val chain: Chain,
private val directReader: Reader<JsonRpcRequest, JsonRpcResponse>,
private val ethereumWsFactory: EthereumWsFactory? = null,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
val node: QuorumForLabels.QuorumItem,
targets: CallMethods
) : DefaultUpstream(id, options, role, targets), Upstream, CachesEnabled, Lifecycle {
constructor(id: String, chain: Chain, api: Reader<JsonRpcRequest, JsonRpcResponse>) :
this(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), UpstreamsConfig.UpstreamRole.STANDARD,
QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
DirectCallMethods())
private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
private val head: Head = this.createHead()
private var validatorSubscription: Disposable? = null
override fun setCaches(caches: Caches) {
if (head is CachesEnabled) {
head.setCaches(caches)
}
}
override fun start() {
log.info("Configured for ${chain.chainName}")
if (getOptions().disableValidation != null && getOptions().disableValidation!!) {
log.warn("Disable validation for upstream ${this.getId()}")
this.setLag(0)
this.setStatus(UpstreamAvailability.OK)
} else {
log.debug("Start validation for upstream ${this.getId()}")
val validator = EthereumUpstreamValidator(this, getOptions())
validatorSubscription = validator.start()
.subscribe(this::setStatus)
}
}
override fun isRunning(): Boolean {
return true
}
override fun stop() {
validatorSubscription?.dispose()
validatorSubscription = null
if (head is Lifecycle) {
head.stop()
}
}
open fun createHead(): Head {
return if (ethereumWsFactory != null) {
val ws = ethereumWsFactory.create(this).apply {
connect()
}
val wsHead = EthereumWsHead(ws).apply {
start()
}
// receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
val rpcHead = EthereumRpcHead(getApi(), Duration.ofSeconds(60)).apply {
start()
}
MergedHead(listOf(rpcHead, wsHead)).apply {
start()
}
} else {
log.warn("Setting up upstream ${this.getId()} with RPC-only access, less effective than WS+RPC")
EthereumRpcHead(getApi()).apply {
start()
}
}
}
override fun getHead(): Head {
return head
}
override fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse> {
return directReader
}
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return listOf(node.labels)
}
@Suppress("unchecked")
override fun <T : Upstream> cast(selfType: Class<T>): T {
if (!selfType.isAssignableFrom(this.javaClass)) {
throw ClassCastException("Cannot cast ${this.javaClass} to $selfType")
}
return this as T
}
}
targets: CallMethods?,
node: QuorumForLabels.QuorumItem?
) : DefaultUpstream(id, options, role, targets, node)

7
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt
View File

@@ -16,7 +16,6 @@
*/
package io.emeraldpay.dshackle.upstream.grpc
import com.fasterxml.jackson.databind.ObjectMapper
import com.salesforce.reactorgrpc.GrpcRetry
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.Common
@@ -57,11 +56,11 @@ open class EthereumGrpcUpstream(
private val chain: Chain,
private val blockchainStub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val client: JsonRpcGrpcClient
) : DefaultUpstream(
) : EthereumUpstream(
"$parentId/${chain.chainCode}",
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.STANDARD,
null
null, null
), Lifecycle {
private var allLabels: Collection<UpstreamsConfig.Labels> = ArrayList<UpstreamsConfig.Labels>()
@@ -176,7 +175,7 @@ open class EthereumGrpcUpstream(
)
}
fun getNodes(): QuorumForLabels {
override fun getQuorumByLabel(): QuorumForLabels {
return nodes.get()
}

3
src/test/groovy/io/emeraldpay/dshackle/test/EthereumUpstreamMock.groovy
View File

@@ -26,6 +26,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
@@ -35,7 +36,7 @@ import io.emeraldpay.grpc.Chain
import org.jetbrains.annotations.NotNull
import org.reactivestreams.Publisher
class EthereumUpstreamMock extends EthereumUpstream {
class EthereumUpstreamMock extends EthereumRpcUpstream {
EthereumHeadMock ethereumHeadMock = new EthereumHeadMock()

3
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.test.EthereumApiStub
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
import io.emeraldpay.grpc.Chain
@@ -45,7 +46,7 @@ class FilteredApisSpec extends Specification {
[test: "foo"],
[test: "baz"]
].collect {
new EthereumUpstream(
new EthereumRpcUpstream(
"test",
Chain.ETHEREUM,
TestingCommons.api().tap { it.id = "${i++}" },