solution: setup custom upstream methods

This commit is contained in:
Igor Artamonov
2019-08-25 20:50:16 -04:00
parent e5c9124705
commit bc83b1b120
33 changed files with 678 additions and 92 deletions

View File

@@ -98,6 +98,68 @@ Dshackle currently supports
- `ws` websocket connection (supposed to be used in addition to `rpc` connection) - `ws` websocket connection (supposed to be used in addition to `rpc` connection)
- `grpc` connects to another Dshackle instance - `grpc` connects to another Dshackle instance
=== Methods
.By default an ethereum upstream supports following JSON RPC methods:
- `eth_gasPrice`
- `eth_call`
- `eth_estimateGas`
- `eth_getBlockTransactionCountByHash`
- `eth_getUncleCountByBlockHash`
- `eth_getBlockByHash`
- `eth_getTransactionByHash`
- `eth_getTransactionByBlockHashAndIndex`
- `eth_getStorageAt`
- `eth_getCode`
- `eth_getUncleByBlockHashAndIndex`
- `eth_getTransactionCount`
- `eth_blockNumber`
- `eth_getBalance`
- `eth_sendRawTransaction`
- `eth_getBlockTransactionCountByNumber`
- `eth_getUncleCountByBlockNumber`
- `eth_getBlockByNumber`
- `eth_getTransactionByBlockNumberAndIndex`
- `eth_getTransactionReceipt`
- `eth_getUncleByBlockNumberAndIndex`
.Plus following methods are answered directly by Dshackle
- `net_version`
- `net_peerCount`
- `net_listening`
- `web3_clientVersion`
- `eth_protocolVersion`
- `eth_syncing`
- `eth_coinbase`
- `eth_mining`
- `eth_hashrate`
- `eth_accounts`
It's possible to enable additional methods that are available on upstream, or disable an existing method. For that purpose
there is `methods` configuration:
[source, yaml]
----
upstreams:
- id: my-node
chain: ethereum
labels:
archive: true
methods:
enabled:
- name: trace_transaction
disabled:
- name: eth_getBlockByNumber
----
Such configuration option allows to execute methods `trace_transaction` and also disables `eth_getBlockByNumber` on that
particular upstream. If a client tries to execute method `trace_transaction` it will be executed on that upstream, or
another upstream will have such method enabled.
Together with label `archive: true` it's possible to specify during execution that a client wants to execute method only
on an archive node.
=== Authentication === Authentication
==== TLS ==== TLS

View File

@@ -52,8 +52,8 @@ Where:
- `items` as a list of independent requests, which may be executed in different nodes in parallels or in different order, with: - `items` as a list of independent requests, which may be executed in different nodes in parallels or in different order, with:
* `method` - a JSON RPC standard name, ex: `eth_getBlockByHash` * `method` - a JSON RPC standard name, ex: `eth_getBlockByHash`
* `payload` - list of parameters for the methods, encoded as JSON string, ex. `["0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331", true]` * `payload` - list of parameters for the methods, encoded as JSON string, ex. `["0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331", true]`
- `Selector` and `AvailabilityEnum` are described in reference, in short they allow to specify which nodes are allowed - `Selector` and `AvailabilityEnum` are described in reference, in short they allow to specify which nodes must be selected
to execute the request. to execute the reques (i.e. "execute only on an archive node")
.NativeCallReplyItem .NativeCallReplyItem
[source,proto] [source,proto]

View File

@@ -47,6 +47,7 @@ class UpstreamsConfig {
} }
companion object { companion object {
@JvmStatic
fun getDefaults(): Options { fun getDefaults(): Options {
val options = Options() val options = Options()
options.minPeers = 1 options.minPeers = 1
@@ -69,6 +70,7 @@ class UpstreamsConfig {
var isEnabled = true var isEnabled = true
var connection: T? = null var connection: T? = null
val labels = Labels() val labels = Labels()
var methods: Methods? = null
} }
open class UpstreamConnection open class UpstreamConnection
@@ -149,4 +151,13 @@ class UpstreamsConfig {
} }
} }
} }
class Methods(
val enabled: Set<Method>,
val disabled: Set<Method>
)
class Method(
val name: String
)
} }

View File

@@ -102,6 +102,7 @@ class UpstreamsConfigReader {
internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) { internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
upstream.id = getValueAsString(upNode, "id") upstream.id = getValueAsString(upNode, "id")
upstream.options = tryReadOptions(upNode) upstream.options = tryReadOptions(upNode)
upstream.methods = tryReadMethods(upNode)
} }
internal fun readUpstreamGrpc(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>) { internal fun readUpstreamGrpc(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>) {
@@ -139,6 +140,29 @@ class UpstreamsConfigReader {
} }
} }
internal fun tryReadMethods(upNode: MappingNode): UpstreamsConfig.Methods? {
return getMapping(upNode, "methods")?.let { mnode ->
val enabled = getList<MappingNode>(mnode, "enabled")?.value?.map { m ->
getValueAsString(m, "name")?.let { name ->
UpstreamsConfig.Method(
name = name
)
}
}?.filterNotNull()?.toSet() ?: emptySet()
val disabled = getList<MappingNode>(mnode, "disabled")?.value?.map { m ->
getValueAsString(m, "name")?.let { name ->
UpstreamsConfig.Method(
name = name
)
}
}?.filterNotNull()?.toSet() ?: emptySet()
UpstreamsConfig.Methods(
enabled, disabled
)
}
}
internal fun readOptions(values: MappingNode): UpstreamsConfig.Options { internal fun readOptions(values: MappingNode): UpstreamsConfig.Options {
val options = UpstreamsConfig.Options() val options = UpstreamsConfig.Options()
getValueAsInt(values, "min-peers")?.let { getValueAsInt(values, "min-peers")?.let {

View File

@@ -36,7 +36,7 @@ class Describe(
upstreams.getAvailable().forEach { chain -> upstreams.getAvailable().forEach { chain ->
upstreams.getUpstream(chain)?.let { chainUpstreams -> upstreams.getUpstream(chain)?.let { chainUpstreams ->
val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll()) val status = subscribeStatus.chainStatus(chain, chainUpstreams.getAll())
val targets = chainUpstreams.getSupportedTargets() val targets = chainUpstreams.getMethods().getSupportedMethods()
val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder() val chainDescription = BlockchainOuterClass.DescribeChain.newBuilder()
.setChain(Common.ChainRef.forNumber(chain.id)) .setChain(Common.ChainRef.forNumber(chain.id))
.addAllSupportedMethods(targets) .addAllSupportedMethods(targets)

View File

@@ -95,11 +95,16 @@ class NativeCall(
} }
fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<RawCallDetails>> { fun prepareCall(request: BlockchainOuterClass.NativeCallRequest, upstream: AggregatedUpstream): Flux<CallContext<RawCallDetails>> {
val matcher = Selector.convertToMatcher(request.selector)
return request.itemsList.toFlux().map { return request.itemsList.toFlux().map {
val method = it.method val method = it.method
val params = it.payload.toStringUtf8() val params = it.payload.toStringUtf8()
val callQuorum = upstream.targets?.getQuorumFor(method) ?: AlwaysQuorum()
val matcher = Selector.Builder()
.forMethod(method)
.forLabels(Selector.convertToMatcher(request.selector))
.build()
val callQuorum = upstream.getMethods().getQuorumFor(method) ?: AlwaysQuorum()
callQuorum.init(upstream.getHead()) callQuorum.init(upstream.getHead())
CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params)) CallContext(it.id, upstream, matcher, callQuorum, RawCallDetails(method, params))
@@ -154,7 +159,11 @@ class NativeCall(
return req as List<Any> return req as List<Any>
} }
open class CallContext<T>(val id: Int, val upstream: AggregatedUpstream, val matcher: Selector.Matcher, val callQuorum: CallQuorum, val payload: T) { open class CallContext<T>(val id: Int,
val upstream: AggregatedUpstream,
val matcher: Selector.Matcher,
val callQuorum: CallQuorum,
val payload: T) {
fun <X> withPayload(payload: X): CallContext<X> { fun <X> withPayload(payload: X): CallContext<X> {
return CallContext(id, upstream, matcher, callQuorum, payload) return CallContext(id, upstream, matcher, callQuorum, payload)
} }

View File

@@ -0,0 +1,57 @@
/**
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.quorum.CallQuorum
import java.util.*
import kotlin.collections.HashSet
class AggregatedCallMethods(
private val delegates: Collection<CallMethods>
): CallMethods {
private val allMethods: Set<String>
init {
val buf = HashSet<String>()
delegates.map { it.getSupportedMethods().forEach { m -> buf.add(m) } }
allMethods = Collections.unmodifiableSet(buf)
}
override fun getQuorumFor(method: String): CallQuorum {
return delegates.find {
it.isAllowed(method)
}?.getQuorumFor(method) ?: throw IllegalStateException("No quorum for $method")
}
override fun isAllowed(method: String): Boolean {
return delegates.any { it.isAllowed(method) }
}
override fun getSupportedMethods(): Set<String> {
return allMethods
}
override fun isHardcoded(method: String): Boolean {
return delegates.any { it.isAllowed(method) && it.isHardcoded(method) }
}
override fun executeHardcoded(method: String): Any {
return delegates.find {
it.isAllowed(method) && it.isHardcoded(method)
}?.executeHardcoded(method) ?: throw IllegalStateException("No hardcoded for $method")
}
}

View File

@@ -37,7 +37,6 @@ import java.util.function.Predicate
import kotlin.concurrent.withLock import kotlin.concurrent.withLock
abstract class AggregatedUpstream( abstract class AggregatedUpstream(
val targets: CallMethods,
val objectMapper: ObjectMapper val objectMapper: ObjectMapper
): Upstream, Lifecycle { ): Upstream, Lifecycle {
@@ -48,11 +47,20 @@ abstract class AggregatedUpstream(
) )
var cache: CachingEthereumApi = CachingEthereumApi.empty() var cache: CachingEthereumApi = CachingEthereumApi.empty()
private val reconfigLock = ReentrantLock() private val reconfigLock = ReentrantLock()
private var callMethods: CallMethods = DirectCallMethods()
abstract fun getAll(): List<Upstream> abstract fun getAll(): List<Upstream>
abstract fun addUpstream(upstream: Upstream) abstract fun addUpstream(upstream: Upstream)
abstract fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi> abstract fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi>
fun reconfigure() {
reconfigLock.withLock {
getAll().map { it.getMethods() }.let {
callMethods = AggregatedCallMethods(it)
}
}
}
override fun observeStatus(): Flux<UpstreamAvailability> { override fun observeStatus(): Flux<UpstreamAvailability> {
val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } } val upstreamsFluxes = getAll().map { up -> up.observeStatus().map { UpstreamStatus(up, it) } }
return Flux.merge(upstreamsFluxes) return Flux.merge(upstreamsFluxes)
@@ -60,16 +68,8 @@ abstract class AggregatedUpstream(
.map { it.status } .map { it.status }
} }
override fun getSupportedTargets(): Set<String> { override fun isAvailable(): Boolean {
val list = HashSet<String>() return getAll().any { it.isAvailable() }
getAll().forEach {
list.addAll(it.getSupportedTargets())
}
return list
}
override fun isAvailable(matcher: Selector.Matcher): Boolean {
return getAll().any { it.isAvailable(matcher) }
} }
override fun getStatus(): UpstreamAvailability { override fun getStatus(): UpstreamAvailability {
@@ -82,6 +82,10 @@ abstract class AggregatedUpstream(
return UpstreamsConfig.Options() return UpstreamsConfig.Options()
} }
override fun getMethods(): CallMethods {
return callMethods
}
class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now()) class UpstreamStatus(val upstream: Upstream, val status: UpstreamAvailability, val ts: Instant = Instant.now())
class FilterBestAvailability(): Predicate<UpstreamStatus> { class FilterBestAvailability(): Predicate<UpstreamStatus> {

View File

@@ -22,5 +22,5 @@ interface CallMethods {
fun isAllowed(method: String): Boolean fun isAllowed(method: String): Boolean
fun getSupportedMethods(): Set<String> fun getSupportedMethods(): Set<String>
fun isHardcoded(method: String): Boolean fun isHardcoded(method: String): Boolean
fun hardcoded(method: String): Any fun executeHardcoded(method: String): Any
} }

View File

@@ -16,6 +16,7 @@
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHeadMerge import io.emeraldpay.dshackle.upstream.ethereum.EthereumHeadMerge
@@ -29,9 +30,8 @@ import java.time.Duration
open class ChainUpstreams ( open class ChainUpstreams (
val chain: Chain, val chain: Chain,
private val upstreams: MutableList<Upstream>, private val upstreams: MutableList<Upstream>,
targets: CallMethods,
objectMapper: ObjectMapper objectMapper: ObjectMapper
) : AggregatedUpstream(targets, objectMapper), Lifecycle { ) : AggregatedUpstream(objectMapper), Lifecycle {
private val log = LoggerFactory.getLogger(ChainUpstreams::class.java) private val log = LoggerFactory.getLogger(ChainUpstreams::class.java)
private var seq = 0 private var seq = 0
@@ -99,6 +99,7 @@ open class ChainUpstreams (
override fun addUpstream(upstream: Upstream) { override fun addUpstream(upstream: Upstream) {
upstreams.add(upstream) upstreams.add(upstream)
head = updateHead() head = updateHead()
reconfigure()
} }
override fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi> { override fun getApis(matcher: Selector.Matcher): Iterator<DirectEthereumApi> {
@@ -124,6 +125,10 @@ open class ChainUpstreams (
return 0 return 0
} }
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return upstreams.flatMap { it.getLabels() }
}
fun printStatus() { fun printStatus() {
var height: Long? = null var height: Long? = null
try { try {

View File

@@ -76,7 +76,7 @@ open class ConfiguredUpstreams(
} }
val options = (up.options ?: UpstreamsConfig.Options()) val options = (up.options ?: UpstreamsConfig.Options())
.merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults()) .merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults())
buildEthereumUpstream(up.connection as UpstreamsConfig.EthereumConnection, chain, options, up.labels) buildEthereumUpstream(up as UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>, chain, options)
} }
} }
} }
@@ -119,18 +119,27 @@ open class ConfiguredUpstreams(
return defaultOptions return defaultOptions
} }
private fun buildEthereumUpstream(up: UpstreamsConfig.EthereumConnection, private fun buildEthereumUpstream(config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
chain: Chain, chain: Chain,
options: UpstreamsConfig.Options, options: UpstreamsConfig.Options
labels: UpstreamsConfig.Labels) { ) {
val conn = config.connection!!
var rpcApi: DirectEthereumApi? = null var rpcApi: DirectEthereumApi? = null
val urls = ArrayList<URI>() val urls = ArrayList<URI>()
up.rpc?.let { endpoint -> val methods = if (config.methods != null) {
ManagedCallMethods(getDefaultMethods(chain),
config.methods!!.enabled.map { it.name }.toSet(),
config.methods!!.disabled.map { it.name }.toSet()
)
} else {
getDefaultMethods(chain)
}
conn.rpc?.let { endpoint ->
val rpcTransport = DefaultRpcTransport(endpoint.url) val rpcTransport = DefaultRpcTransport(endpoint.url)
up.rpc?.basicAuth?.let { auth -> conn.rpc?.basicAuth?.let { auth ->
rpcTransport.setBasicAuth(auth.username, auth.password) rpcTransport.setBasicAuth(auth.username, auth.password)
} }
up.rpc?.tls?.let { tls -> conn.rpc?.tls?.let { tls ->
tls.ca?.let { ca -> tls.ca?.let { ca ->
File(ca).inputStream().use { cert -> rpcTransport.setTrustedCertificate(cert) } File(ca).inputStream().use { cert -> rpcTransport.setTrustedCertificate(cert) }
} }
@@ -139,12 +148,12 @@ open class ConfiguredUpstreams(
rpcApi = DirectEthereumApi( rpcApi = DirectEthereumApi(
rpcClient, rpcClient,
objectMapper, objectMapper,
targetFor(chain) methods
) )
urls.add(endpoint.url) urls.add(endpoint.url)
} }
if (rpcApi != null) { if (rpcApi != null) {
val wsApi: EthereumWs? = up.ws?.let { endpoint -> val wsApi: EthereumWs? = conn.ws?.let { endpoint ->
val wsApi = EthereumWs( val wsApi = EthereumWs(
endpoint.url, endpoint.url,
endpoint.origin ?: URI("http://localhost"), endpoint.origin ?: URI("http://localhost"),
@@ -159,7 +168,9 @@ open class ConfiguredUpstreams(
} }
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}") log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options, NodeDetailsList.NodeDetails(1, labels), targetFor(chain)) val ethereumUpstream = EthereumUpstream(chain, rpcApi!!, wsApi, options,
NodeDetailsList.NodeDetails(1, config.labels),
methods)
ethereumUpstream.start() ethereumUpstream.start()
addUpstream(chain, ethereumUpstream) addUpstream(chain, ethereumUpstream)
} }
@@ -171,8 +182,7 @@ open class ConfiguredUpstreams(
endpoint.host!!, endpoint.host!!,
endpoint.port ?: 443, endpoint.port ?: 443,
objectMapper, objectMapper,
up.auth, up.auth
this
) )
log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}") log.info("Using ALL CHAINS (gRPC) upstream, at ${endpoint.host}:${endpoint.port}")
ds.start() ds.start()
@@ -189,7 +199,7 @@ open class ConfiguredUpstreams(
override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams { override fun addUpstream(chain: Chain, up: Upstream): ChainUpstreams {
val current = chainMapping[chain] val current = chainMapping[chain]
if (current == null) { if (current == null) {
val created = ChainUpstreams(chain, ArrayList<Upstream>(), targetFor(chain), objectMapper) val created = ChainUpstreams(chain, ArrayList<Upstream>(), objectMapper)
created.addUpstream(up) created.addUpstream(up)
created.start() created.start()
chainMapping[chain] = created chainMapping[chain] = created
@@ -217,7 +227,7 @@ open class ConfiguredUpstreams(
) )
} }
override fun targetFor(chain: Chain): CallMethods { override fun getDefaultMethods(chain: Chain): CallMethods {
var current = callTargets[chain] var current = callTargets[chain]
if (current == null) { if (current == null) {
current = QuorumBasedMethods(objectMapper, chain) current = QuorumBasedMethods(objectMapper, chain)

View File

@@ -17,26 +17,30 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.quorum.AlwaysQuorum import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum import io.emeraldpay.dshackle.quorum.CallQuorum
import java.util.*
class DirectCallMethods : CallMethods { class DirectCallMethods(private val methods: Set<String>) : CallMethods {
constructor(): this(emptySet())
constructor(methods: Collection<String>): this(methods.toSet())
override fun getQuorumFor(method: String): CallQuorum { override fun getQuorumFor(method: String): CallQuorum {
return AlwaysQuorum() return AlwaysQuorum()
} }
override fun isAllowed(method: String): Boolean { override fun isAllowed(method: String): Boolean {
return true return methods.contains(method)
} }
override fun getSupportedMethods(): Set<String> { override fun getSupportedMethods(): Set<String> {
return emptySet() return methods
} }
override fun isHardcoded(method: String): Boolean { override fun isHardcoded(method: String): Boolean {
return false return false
} }
override fun hardcoded(method: String): Any { override fun executeHardcoded(method: String): Any {
return "unsupported" return "unsupported"
} }
} }

View File

@@ -37,7 +37,7 @@ class FilteringApiIterator(
return false return false
} }
val upstream = upstreams[pos++ % upstreams.size] val upstream = upstreams[pos++ % upstreams.size]
if (upstream.isAvailable(matcher)) { if (upstream.isAvailable() && matcher.matches(upstream)) {
nextUpstream = upstream nextUpstream = upstream
} }
} }

View File

@@ -0,0 +1,55 @@
/**
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.CallQuorum
import java.util.*
class ManagedCallMethods(
private val delegate: CallMethods,
private val enabled: Set<String>,
private val disabled: Set<String>
): CallMethods {
private val allAllowed: Set<String> = Collections.unmodifiableSet(
enabled + delegate.getSupportedMethods() - disabled
)
override fun getQuorumFor(method: String): CallQuorum {
return if (enabled.contains(method)) {
AlwaysQuorum()
} else {
delegate.getQuorumFor(method)
}
}
override fun isAllowed(method: String): Boolean {
return allAllowed.contains(method)
}
override fun getSupportedMethods(): Set<String> {
return allAllowed
}
override fun isHardcoded(method: String): Boolean {
return delegate.isHardcoded(method)
}
override fun executeHardcoded(method: String): Any {
return delegate.executeHardcoded(method)
}
}

View File

@@ -101,7 +101,7 @@ class QuorumBasedMethods(
return hardcodedMethods.contains(method) return hardcodedMethods.contains(method)
} }
override fun hardcoded(method: String): Any { override fun executeHardcoded(method: String): Any {
if ("net_version" == method) { if ("net_version" == method) {
if (Chain.ETHEREUM == chain) { if (Chain.ETHEREUM == chain) {
return "1" return "1"

View File

@@ -19,6 +19,7 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.config.UpstreamsConfig import io.emeraldpay.dshackle.config.UpstreamsConfig
import org.apache.commons.lang3.StringUtils import org.apache.commons.lang3.StringUtils
import java.util.* import java.util.*
import kotlin.collections.ArrayList
class Selector { class Selector {
@@ -27,9 +28,9 @@ class Selector {
val empty = EmptyMatcher() val empty = EmptyMatcher()
@JvmStatic @JvmStatic
fun convertToMatcher(req: BlockchainOuterClass.Selector?): Matcher { fun convertToMatcher(req: BlockchainOuterClass.Selector?): LabelSelectorMatcher {
return when { return when {
req == null -> EmptyMatcher() req == null -> AnyLabelMatcher()
req.hasLabelSelector() -> req.labelSelector.let { selector -> req.hasLabelSelector() -> req.labelSelector.let { selector ->
if (StringUtils.isNotEmpty(selector.name)) { if (StringUtils.isNotEmpty(selector.name)) {
val values = selector.valueList val values = selector.valueList
@@ -41,24 +42,87 @@ class Selector {
LabelMatcher(selector.name, selector.valueList) LabelMatcher(selector.name, selector.valueList)
} }
} else { } else {
EmptyMatcher() AnyLabelMatcher()
} }
} }
req.hasAndSelector() -> AndMatcher(Collections.unmodifiableCollection(req.andSelector.selectorsList.map { convertToMatcher(it) })) req.hasAndSelector() -> AndMatcher(Collections.unmodifiableCollection(req.andSelector.selectorsList.map { convertToMatcher(it) }))
req.hasOrSelector() -> OrMatcher(Collections.unmodifiableCollection(req.orSelector.selectorsList.map { convertToMatcher(it) })) req.hasOrSelector() -> OrMatcher(Collections.unmodifiableCollection(req.orSelector.selectorsList.map { convertToMatcher(it) }))
req.hasNotSelector() -> NotMatcher(convertToMatcher(req.notSelector.selector)) req.hasNotSelector() -> NotMatcher(convertToMatcher(req.notSelector.selector))
req.hasExistsSelector() -> ExistsMatcher(req.existsSelector.name) req.hasExistsSelector() -> ExistsMatcher(req.existsSelector.name)
else -> EmptyMatcher() else -> AnyLabelMatcher()
} }
} }
@JvmStatic
fun extractLabels(matcher: Matcher): LabelSelectorMatcher? {
if (matcher is LabelSelectorMatcher) {
return matcher
}
if (matcher is MultiMatcher) {
return matcher.getLabelMatcher()
}
return null
}
}
class Builder {
private val matchers = ArrayList<Matcher>()
fun forMethod(name: String): Builder {
matchers.add(MethodMatcher(name))
return this
}
fun forLabels(matcher: LabelSelectorMatcher): Builder {
matchers.add(matcher)
return this
}
fun build(): Matcher {
return MultiMatcher(matchers)
}
} }
interface Matcher { interface Matcher {
fun matches(labels: UpstreamsConfig.Labels): Boolean fun matches(up: Upstream): Boolean
fun asProto(): BlockchainOuterClass.Selector? }
class MultiMatcher(
private val matchers: Collection<Matcher>
): Matcher {
override fun matches(up: Upstream): Boolean {
return matchers.all { it.matches(up) }
}
fun getLabelMatcher(): LabelSelectorMatcher? {
return matchers.find { it is LabelSelectorMatcher } as LabelSelectorMatcher?
}
}
class MethodMatcher(
val method: String
): Matcher {
override fun matches(up: Upstream): Boolean {
return up.getMethods().isAllowed(method)
}
}
abstract class LabelSelectorMatcher: Matcher {
override fun matches(up: Upstream): Boolean {
return up.getLabels().any(this::matches)
}
abstract fun matches(labels: UpstreamsConfig.Labels): Boolean
abstract fun asProto(): BlockchainOuterClass.Selector?
} }
class EmptyMatcher: Matcher { class EmptyMatcher: Matcher {
override fun matches(up: Upstream): Boolean {
return true
}
}
class AnyLabelMatcher: LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean { override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return true return true
} }
@@ -66,9 +130,13 @@ class Selector {
override fun asProto(): BlockchainOuterClass.Selector? { override fun asProto(): BlockchainOuterClass.Selector? {
return null return null
} }
override fun matches(up: Upstream): Boolean {
return true
}
} }
class LabelMatcher(val name: String, val values: Collection<String>): Matcher { class LabelMatcher(val name: String, val values: Collection<String>): LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean { override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return labels.get(name)?.let { return labels.get(name)?.let {
labelValue -> values.any { it == labelValue } labelValue -> values.any { it == labelValue }
@@ -84,7 +152,7 @@ class Selector {
} }
} }
class OrMatcher(val matchers: Collection<Matcher>): Matcher { class OrMatcher(val matchers: Collection<LabelSelectorMatcher>): LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean { override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return matchers.any { matcher -> matcher.matches(labels) } return matchers.any { matcher -> matcher.matches(labels) }
} }
@@ -98,7 +166,7 @@ class Selector {
} }
} }
class AndMatcher(val matchers: Collection<Matcher>): Matcher { class AndMatcher(val matchers: Collection<LabelSelectorMatcher>): LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean { override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return matchers.all { matcher -> matcher.matches(labels) } return matchers.all { matcher -> matcher.matches(labels) }
} }
@@ -112,7 +180,7 @@ class Selector {
} }
} }
class NotMatcher(val matcher: Matcher): Matcher { class NotMatcher(val matcher: LabelSelectorMatcher): LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean { override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return !matcher.matches(labels) return !matcher.matches(labels)
} }
@@ -126,7 +194,7 @@ class Selector {
} }
} }
class ExistsMatcher(val name: String): Matcher { class ExistsMatcher(val name: String): LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean { override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return labels.containsKey(name) return labels.containsKey(name)
} }

View File

@@ -21,14 +21,14 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
interface Upstream { interface Upstream {
fun isAvailable(matcher: Selector.Matcher): Boolean fun isAvailable(): Boolean
fun getStatus(): UpstreamAvailability fun getStatus(): UpstreamAvailability
fun observeStatus(): Flux<UpstreamAvailability> fun observeStatus(): Flux<UpstreamAvailability>
fun getHead(): EthereumHead fun getHead(): EthereumHead
fun getApi(matcher: Selector.Matcher): DirectEthereumApi fun getApi(matcher: Selector.Matcher): DirectEthereumApi
// fun getCache(): CachingEthereumApi
fun getOptions(): UpstreamsConfig.Options fun getOptions(): UpstreamsConfig.Options
fun getSupportedTargets(): Set<String>
fun setLag(lag: Long) fun setLag(lag: Long)
fun getLag(): Long fun getLag(): Long
fun getLabels(): Collection<UpstreamsConfig.Labels>
fun getMethods(): CallMethods
} }

View File

@@ -23,6 +23,6 @@ interface Upstreams {
fun getUpstream(chain: Chain): AggregatedUpstream? fun getUpstream(chain: Chain): AggregatedUpstream?
fun getAvailable(): List<Chain> fun getAvailable(): List<Chain>
fun observeChains(): Flux<Chain> fun observeChains(): Flux<Chain>
fun targetFor(chain: Chain): CallMethods fun getDefaultMethods(chain: Chain): CallMethods
fun isAvailable(chain: Chain): Boolean fun isAvailable(chain: Chain): Boolean
} }

View File

@@ -36,7 +36,7 @@ open class DirectEthereumApi(
override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> { override fun execute(id: Int, method: String, params: List<Any>): Mono<ByteArray> {
val result: Mono<out Any> = when { val result: Mono<out Any> = when {
targets.isHardcoded(method) -> Mono.just(method).map { targets.hardcoded(it) } targets.isHardcoded(method) -> Mono.just(method).map { targets.executeHardcoded(it) }
targets.isAllowed(method) -> callUpstream(method, params) targets.isAllowed(method) -> callUpstream(method, params)
else -> Mono.error(RpcException(-32601, "Method not allowed or not found")) else -> Mono.error(RpcException(-32601, "Method not allowed or not found"))
} }

View File

@@ -35,9 +35,6 @@ open class EthereumUpstream(
UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()), UpstreamsConfig.Options.getDefaults(), NodeDetailsList.NodeDetails(1, UpstreamsConfig.Labels()),
DirectCallMethods()) DirectCallMethods())
override fun getSupportedTargets(): Set<String> {
return targets.getSupportedMethods()
}
private val log = LoggerFactory.getLogger(EthereumUpstream::class.java) private val log = LoggerFactory.getLogger(EthereumUpstream::class.java)
@@ -92,8 +89,8 @@ open class EthereumUpstream(
} }
} }
override fun isAvailable(matcher: Selector.Matcher): Boolean { override fun isAvailable(): Boolean {
return getStatus() == UpstreamAvailability.OK && matcher.matches(node.labels) return getStatus() == UpstreamAvailability.OK
} }
override fun getHead(): EthereumHead { override fun getHead(): EthereumHead {
@@ -112,4 +109,12 @@ open class EthereumUpstream(
return options return options
} }
override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return listOf(node.labels)
}
override fun getMethods(): CallMethods {
return targets
}
} }

View File

@@ -36,7 +36,7 @@ import java.util.function.Function
class EthereumGrpcTransport( class EthereumGrpcTransport(
private val chainRef: Common.ChainRef, private val chainRef: Common.ChainRef,
private val selector: BlockchainOuterClass.Selector?, private val labelSelector: BlockchainOuterClass.Selector?,
private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val objectMapper: ObjectMapper private val objectMapper: ObjectMapper
): RpcTransport { ): RpcTransport {
@@ -47,13 +47,13 @@ class EthereumGrpcTransport(
chain: Chain, chain: Chain,
client: ReactorBlockchainGrpc.ReactorBlockchainStub, client: ReactorBlockchainGrpc.ReactorBlockchainStub,
objectMapper: ObjectMapper objectMapper: ObjectMapper
) : this(Common.ChainRef.forNumber(chain.id), Selector.EmptyMatcher().asProto(), client, objectMapper) ) : this(Common.ChainRef.forNumber(chain.id), null, client, objectMapper)
fun withMatcher(matcher: Selector.Matcher): EthereumGrpcTransport { fun withLabels(matcher: Selector.LabelSelectorMatcher?): EthereumGrpcTransport {
if (matcher is Selector.EmptyMatcher && selector == null) { if ((matcher == null || matcher is Selector.AnyLabelMatcher) && labelSelector == null) {
return this return this
} }
return EthereumGrpcTransport(chainRef, matcher.asProto(), client, objectMapper) return EthereumGrpcTransport(chainRef, matcher?.asProto(), client, objectMapper)
} }
override fun close() { override fun close() {
@@ -114,8 +114,8 @@ class EthereumGrpcTransport(
override fun execute(items: List<Batch.BatchItem<out Any, out Any>>): CompletableFuture<BatchStatus> { override fun execute(items: List<Batch.BatchItem<out Any, out Any>>): CompletableFuture<BatchStatus> {
val req = BlockchainOuterClass.NativeCallRequest.newBuilder() val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(chainRef) .setChain(chainRef)
if (selector != null) { if (labelSelector != null) {
req.setSelector(selector) req.setSelector(labelSelector)
} }
val mapping = prepareMapping(items, req) val mapping = prepareMapping(items, req)
return client.nativeCall(req.build()) return client.nativeCall(req.build())

View File

@@ -42,14 +42,15 @@ import java.time.Duration
import java.util.* import java.util.*
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
import java.util.function.Function import java.util.function.Function
import kotlin.collections.ArrayList
open class GrpcUpstream( open class GrpcUpstream(
private val chain: Chain, private val chain: Chain,
private val client: ReactorBlockchainGrpc.ReactorBlockchainStub, private val client: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val objectMapper: ObjectMapper, private val objectMapper: ObjectMapper
private val targets: CallMethods
): DefaultUpstream(), Lifecycle { ): DefaultUpstream(), Lifecycle {
private var allLabels: Collection<UpstreamsConfig.Labels> = ArrayList<UpstreamsConfig.Labels>()
private val log = LoggerFactory.getLogger(GrpcUpstream::class.java) private val log = LoggerFactory.getLogger(GrpcUpstream::class.java)
private val options = UpstreamsConfig.Options.getDefaults() private val options = UpstreamsConfig.Options.getDefaults()
@@ -57,13 +58,13 @@ open class GrpcUpstream(
private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create() private val streamBlocks: TopicProcessor<BlockJson<TransactionId>> = TopicProcessor.create()
private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList()) private val nodes = AtomicReference<NodeDetailsList>(NodeDetailsList())
private val head = Head(this) private val head = Head(this)
private val supportedMethods = HashSet<String>() private var targets: CallMethods = DirectCallMethods()
private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper) private val grpcTransport = EthereumGrpcTransport(chain, client, objectMapper)
private var headSubscription: Disposable? = null private var headSubscription: Disposable? = null
open fun createApi(matcher: Selector.Matcher): DirectEthereumApi { open fun createApi(matcher: Selector.Matcher): DirectEthereumApi {
val rpcClient = DefaultRpcClient(grpcTransport.withMatcher(matcher)) val rpcClient = DefaultRpcClient(grpcTransport.withLabels(Selector.extractLabels(matcher)))
return DirectEthereumApi(rpcClient, objectMapper, targets).let { return DirectEthereumApi(rpcClient, objectMapper, targets).let {
it.upstream = this it.upstream = this
it it
@@ -130,19 +131,24 @@ open class GrpcUpstream(
} }
fun init(conf: BlockchainOuterClass.DescribeChain) { fun init(conf: BlockchainOuterClass.DescribeChain) {
supportedMethods.addAll(conf.supportedMethodsList) targets = DirectCallMethods(conf.supportedMethodsList.toSet())
val nodes = NodeDetailsList() val nodes = NodeDetailsList()
val allLabels = ArrayList<UpstreamsConfig.Labels>()
conf.nodesList.forEach { node -> conf.nodesList.forEach { node ->
val node = NodeDetailsList.NodeDetails(node.quorum, val node = NodeDetailsList.NodeDetails(node.quorum,
node.labelsList.let { provided -> node.labelsList.let { provided ->
val labels = UpstreamsConfig.Labels() val labels = UpstreamsConfig.Labels()
provided.forEach { labels.put(it.name, it.value) } provided.forEach {
labels[it.name] = it.value
}
allLabels.add(labels)
labels labels
} }
) )
nodes.add(node) nodes.add(node)
} }
this.nodes.set(nodes) this.nodes.set(nodes)
this.allLabels = Collections.unmodifiableCollection(allLabels)
conf.status?.let { status -> onStatus(status) } conf.status?.let { status -> onStatus(status) }
} }
@@ -160,13 +166,17 @@ open class GrpcUpstream(
// ------------------------------------------------------------------------------------------ // ------------------------------------------------------------------------------------------
override fun getSupportedTargets(): Set<String> { override fun getLabels(): Collection<UpstreamsConfig.Labels> {
return supportedMethods return allLabels
} }
override fun isAvailable(matcher: Selector.Matcher): Boolean { override fun getMethods(): CallMethods {
return targets
}
override fun isAvailable(): Boolean {
return getStatus() == UpstreamAvailability.OK && headBlock.get() != null && nodes.get().getNodes().any { return getStatus() == UpstreamAvailability.OK && headBlock.get() != null && nodes.get().getNodes().any {
it.quorum > 0 && matcher.matches(it.labels) it.quorum > 0
} }
} }

View File

@@ -39,8 +39,7 @@ class GrpcUpstreams(
private val host: String, private val host: String,
private val port: Int, private val port: Int,
private val objectMapper: ObjectMapper, private val objectMapper: ObjectMapper,
private val auth: UpstreamsConfig.TlsAuth? = null, private val auth: UpstreamsConfig.TlsAuth? = null
private val upstreams: Upstreams
) { ) {
private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java) private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
@@ -111,7 +110,7 @@ class GrpcUpstreams(
lock.withLock { lock.withLock {
val current = known[chain] val current = known[chain]
return if (current == null) { return if (current == null) {
val created = GrpcUpstream(chain, client!!, objectMapper, upstreams.targetFor(chain)) val created = GrpcUpstream(chain, client!!, objectMapper)
known[chain] = created known[chain] = created
created.start() created.start()
created created

View File

@@ -174,4 +174,24 @@ class UpstreamsConfigReaderSpec extends Specification {
} }
} }
} }
def "Parse config with methods"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-methods.yaml")
when:
def act = reader.read(config)
then:
act != null
with(act.upstreams.get(0)) {
methods != null
with(methods) {
enabled.size() == 1
enabled.first().name == "parity_trace"
disabled.size() == 2
disabled.toList()[0].name == "eth_getBlockByNumber"
disabled.toList()[1].name == "admin_shutdown"
}
}
}
} }

View File

@@ -15,6 +15,10 @@
*/ */
package io.emeraldpay.dshackle.test package io.emeraldpay.dshackle.test
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.CallMethods
import io.emeraldpay.dshackle.upstream.NodeDetailsList
import io.emeraldpay.dshackle.upstream.QuorumBasedMethods
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead import io.emeraldpay.dshackle.upstream.ethereum.EthereumHead
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
@@ -29,7 +33,13 @@ class EthereumUpstreamMock extends EthereumUpstream {
EthereumHeadMock ethereumHeadMock = new EthereumHeadMock() EthereumHeadMock ethereumHeadMock = new EthereumHeadMock()
EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api) { EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api) {
super(chain, api) this(chain, api, new QuorumBasedMethods(TestingCommons.objectMapper(), chain))
}
EthereumUpstreamMock(@NotNull Chain chain, @NotNull DirectEthereumApi api, CallMethods methods) {
super(chain, api, null,
UpstreamsConfig.Options.getDefaults(), new NodeDetailsList.NodeDetails(1, new UpstreamsConfig.Labels()),
methods)
setLag(0) setLag(0)
setStatus(UpstreamAvailability.OK) setStatus(UpstreamAvailability.OK)
} }

View File

@@ -20,9 +20,12 @@ import com.fasterxml.jackson.databind.DeserializationFeature
import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.databind.module.SimpleModule import com.fasterxml.jackson.databind.module.SimpleModule
import io.emeraldpay.dshackle.upstream.AggregatedUpstream import io.emeraldpay.dshackle.upstream.AggregatedUpstream
import io.emeraldpay.dshackle.upstream.CallMethods
import io.emeraldpay.dshackle.upstream.ChainUpstreams import io.emeraldpay.dshackle.upstream.ChainUpstreams
import io.emeraldpay.dshackle.upstream.DirectCallMethods import io.emeraldpay.dshackle.upstream.DirectCallMethods
import io.emeraldpay.dshackle.upstream.QuorumBasedMethods
import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi import io.emeraldpay.dshackle.upstream.ethereum.DirectEthereumApi
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.grpc.Chain import io.emeraldpay.grpc.Chain
import io.infinitape.etherjar.rpc.JacksonRpcConverter import io.infinitape.etherjar.rpc.JacksonRpcConverter
import io.infinitape.etherjar.rpc.RpcClient import io.infinitape.etherjar.rpc.RpcClient
@@ -57,6 +60,10 @@ class TestingCommons {
} }
static AggregatedUpstream aggregatedUpstream(DirectEthereumApi api) { static AggregatedUpstream aggregatedUpstream(DirectEthereumApi api) {
return new ChainUpstreams(Chain.ETHEREUM, [upstream(api)], new DirectCallMethods(), objectMapper()) return aggregatedUpstream(upstream(api))
}
static AggregatedUpstream aggregatedUpstream(EthereumUpstream up) {
return new ChainUpstreams(Chain.ETHEREUM, [up], objectMapper())
} }
} }

View File

@@ -41,7 +41,7 @@ class UpstreamsMock implements Upstreams {
@Override @Override
AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) { AggregatedUpstream addUpstream(@NotNull Chain chain, @NotNull Upstream up) {
if (!upstreams.containsKey(chain)) { if (!upstreams.containsKey(chain)) {
upstreams[chain] = new ChainUpstreams(chain, [up], targetFor(chain), TestingCommons.objectMapper()) upstreams[chain] = new ChainUpstreams(chain, [up], TestingCommons.objectMapper())
} else { } else {
upstreams[chain].addUpstream(up) upstreams[chain].addUpstream(up)
} }
@@ -64,7 +64,7 @@ class UpstreamsMock implements Upstreams {
} }
@Override @Override
QuorumBasedMethods targetFor(@NotNull Chain chain) { QuorumBasedMethods getDefaultMethods(@NotNull Chain chain) {
if (target[chain] == null) { if (target[chain] == null) {
QuorumBasedMethods targets = new QuorumBasedMethods(TestingCommons.objectMapper(), chain) QuorumBasedMethods targets = new QuorumBasedMethods(TestingCommons.objectMapper(), chain)
target[chain] = targets target[chain] = targets

View File

@@ -0,0 +1,125 @@
/**
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import spock.lang.Specification
class AggregatedCallMethodsSpec extends Specification {
def "Returns quorum from delegate that owns it"() {
setup:
def quorum = new AlwaysQuorum()
def delegate1 = Mock(CallMethods) {
_ * getSupportedMethods() >> ["eth_no_test", "foo_bar"]
1 * isAllowed("eth_test") >> false
}
def delegate2 = Mock(CallMethods) {
_ * getSupportedMethods() >> ["eth_test", "foo_bar"]
1 * isAllowed("eth_test") >> true
1 * getQuorumFor("eth_test") >> quorum
}
def aggregate = new AggregatedCallMethods([delegate1, delegate2])
when:
def act = aggregate.getQuorumFor("eth_test")
then:
act == quorum
}
def "Allowed if any allowed"() {
setup:
def delegate1 = new DirectCallMethods(["eth_no_test", "foo_bar"] as Set)
def delegate2 = new DirectCallMethods(["eth_test", "foo_bar"] as Set)
def aggregate = new AggregatedCallMethods([delegate1, delegate2])
when:
def act = aggregate.isAllowed("eth_test")
then:
act
when:
act = aggregate.isAllowed("eth_no_test")
then:
act
when:
act = aggregate.isAllowed("foo_bar")
then:
act
when:
act = aggregate.isAllowed("nothing")
then:
!act
}
def "Supported has all methods"() {
setup:
def delegate1 = new DirectCallMethods(["eth_no_test", "foo_bar"] as Set)
def delegate2 = new DirectCallMethods(["eth_test", "foo_bar"] as Set)
def aggregate = new AggregatedCallMethods([delegate1, delegate2])
when:
def act = aggregate.getSupportedMethods()
then:
act.sort() == ["eth_test", "eth_no_test", "foo_bar"].sort()
}
def "Hardcoded if any hardcoded"() {
setup:
def delegate1 = Mock(CallMethods) {
_ * getSupportedMethods() >> ["eth_no_test", "foo_bar"]
1 * isAllowed("eth_test") >> false
1 * isAllowed("eth_no_test") >> true
1 * isHardcoded("eth_no_test") >> false
}
def delegate2 = Mock(CallMethods) {
_ * getSupportedMethods() >> ["eth_test", "foo_bar"]
1 * isAllowed("eth_test") >> true
1 * isAllowed("eth_no_test") >> false
1 * isHardcoded("eth_test") >> true
}
def aggregate = new AggregatedCallMethods([delegate1, delegate2])
when:
def act = aggregate.isHardcoded("eth_test")
then:
act
when:
act = aggregate.isHardcoded("eth_no_test")
then:
!act
}
def "Execute hardcoded on delegate that owns it"() {
setup:
def delegate1 = Mock(CallMethods) {
_ * getSupportedMethods() >> ["eth_no_test", "foo_bar"]
1 * isAllowed("eth_test") >> false
}
def delegate2 = Mock(CallMethods) {
_ * getSupportedMethods() >> ["eth_test", "foo_bar"]
1 * isAllowed("eth_test") >> true
1 * isHardcoded("eth_test") >> true
1 * executeHardcoded("eth_test") >> "hello"
}
def aggregate = new AggregatedCallMethods([delegate1, delegate2])
when:
def act = aggregate.executeHardcoded("eth_test")
then:
act == "hello"
}
}

View File

@@ -0,0 +1,61 @@
/**
* Copyright (c) 2019 ETCDEV GmbH
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import spock.lang.Specification
class ManagedCallMethodsSpec extends Specification {
def "Gets quorum for enabled method"() {
setup:
def managed = new ManagedCallMethods(
new DirectCallMethods(),
["eth_test"] as Set,
[] as Set
)
when:
def act = managed.getQuorumFor("eth_test")
then:
act instanceof AlwaysQuorum
}
def "Allowed contacts all enabled + delegate"() {
setup:
def managed = new ManagedCallMethods(
new DirectCallMethods(["eth_test2"] as Set),
["eth_test"] as Set,
[] as Set
)
when:
def act = managed.getSupportedMethods()
then:
act.sort() == ["eth_test", "eth_test2"].sort()
}
def "Disabled removed from delegate"() {
setup:
def managed = new ManagedCallMethods(
new DirectCallMethods(["eth_test2", "foo_bar"] as Set),
["eth_test"] as Set,
["foo_bar"] as Set
)
when:
def act = managed.getSupportedMethods()
then:
act.sort() == ["eth_test", "eth_test2"].sort()
}
}

View File

@@ -35,7 +35,7 @@ class SelectorSpec extends Specification {
when: when:
def act = Selector.convertToMatcher(null) def act = Selector.convertToMatcher(null)
then: then:
act.class == Selector.EmptyMatcher act.class == Selector.AnyLabelMatcher
} }
def "Convert LABEL match"() { def "Convert LABEL match"() {

View File

@@ -20,8 +20,10 @@ import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.api.proto.ReactorBlockchainGrpc import io.emeraldpay.api.proto.ReactorBlockchainGrpc
import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.test.EthereumApiMock import io.emeraldpay.dshackle.test.EthereumApiMock
import io.emeraldpay.dshackle.test.EthereumUpstreamMock
import io.emeraldpay.dshackle.test.MockServer import io.emeraldpay.dshackle.test.MockServer
import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.DirectCallMethods
import io.emeraldpay.dshackle.upstream.QuorumBasedMethods import io.emeraldpay.dshackle.upstream.QuorumBasedMethods
import io.emeraldpay.dshackle.upstream.Upstreams import io.emeraldpay.dshackle.upstream.Upstreams
import io.emeraldpay.dshackle.upstream.grpc.EthereumGrpcTransport import io.emeraldpay.dshackle.upstream.grpc.EthereumGrpcTransport
@@ -45,7 +47,9 @@ class EthereumGrpcTransportSpec extends Specification {
def callData = [:] def callData = [:]
def otherSideUpstreams = Mock(Upstreams) def otherSideUpstreams = Mock(Upstreams)
def otherSideAggr = TestingCommons.aggregatedUpstream(otherSideApi) def otherSideAggr = TestingCommons.aggregatedUpstream(
new EthereumUpstreamMock(Chain.ETHEREUM, otherSideApi, new DirectCallMethods(["eth_test"]))
)
def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
otherSideApi.upstream = otherSideAggr otherSideApi.upstream = otherSideAggr
@@ -88,7 +92,9 @@ class EthereumGrpcTransportSpec extends Specification {
def callData = [:] def callData = [:]
def otherSideUpstreams = Mock(Upstreams) def otherSideUpstreams = Mock(Upstreams)
def otherSideAggr = TestingCommons.aggregatedUpstream(otherSideApi) def otherSideAggr = TestingCommons.aggregatedUpstream(
new EthereumUpstreamMock(Chain.ETHEREUM, otherSideApi, new DirectCallMethods(["eth_test", "eth_test2"]))
)
def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper) def otherSideNativeCall = new NativeCall(otherSideUpstreams, objectMapper)
otherSideApi.upstream = otherSideAggr otherSideApi.upstream = otherSideAggr

View File

@@ -39,7 +39,6 @@ class GrpcUpstreamSpec extends Specification {
MockServer mockServer = new MockServer() MockServer mockServer = new MockServer()
ObjectMapper objectMapper = TestingCommons.objectMapper() ObjectMapper objectMapper = TestingCommons.objectMapper()
def ethereumTargets = new QuorumBasedMethods(objectMapper, Chain.ETHEREUM)
def "Subscribe to head"() { def "Subscribe to head"() {
setup: setup:
@@ -71,8 +70,11 @@ class GrpcUpstreamSpec extends Specification {
) )
} }
}) })
def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) def upstream = new GrpcUpstream(chain, client, objectMapper)
upstream.setLag(0) upstream.setLag(0)
upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
.addAllSupportedMethods(["eth_getBlockByHash"])
.build())
when: when:
upstream.start() upstream.start()
def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1)) def h = upstream.head.getFlux().next().block(Duration.ofSeconds(1))
@@ -127,8 +129,11 @@ class GrpcUpstreamSpec extends Specification {
finished.complete(true) finished.complete(true)
} }
}) })
def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) def upstream = new GrpcUpstream(chain, client, objectMapper)
upstream.setLag(0) upstream.setLag(0)
upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
.addAllSupportedMethods(["eth_getBlockByHash"])
.build())
when: when:
upstream.start() upstream.start()
finished.get() finished.get()
@@ -184,8 +189,11 @@ class GrpcUpstreamSpec extends Specification {
finished.complete(true) finished.complete(true)
} }
}) })
def upstream = new GrpcUpstream(chain, client, objectMapper, ethereumTargets) def upstream = new GrpcUpstream(chain, client, objectMapper)
upstream.setLag(0) upstream.setLag(0)
upstream.init(BlockchainOuterClass.DescribeChain.newBuilder()
.addAllSupportedMethods(["eth_getBlockByHash"])
.build())
when: when:
upstream.start() upstream.start()
finished.get() finished.get()

View File

@@ -0,0 +1,26 @@
version: v1
defaultOptions:
- chains:
- ethereum
options:
min-peers: 3
upstreams:
- id: local
chain: ethereum
options:
min-peers: 7
methods:
enabled:
- name: "parity_trace"
disabled:
- name: "eth_getBlockByNumber"
- name: "admin_shutdown"
connection:
ethereum:
rpc:
url: "http://localhost:8545"
ws:
url: "ws://localhost:8546"
origin: "http://localhost"