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

solution: fallback upstreams

Browse Source 
fix: #15
This commit is contained in:
Igor Artamonov
2020-07-10 23:36:35 -04:00
parent 242c479893
commit 3df1c0a572
19 changed files with 281 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -74,6 +74,7 @@ class UpstreamsConfig {
var connection: T? = null
val labels = Labels()
var methods: Methods? = null
var role: UpstreamRole = UpstreamRole.STANDARD
@Suppress("unchecked")
fun <Z : UpstreamConnection> cast(type: Class<Z>): Upstream<Z> {
@@ -84,6 +85,11 @@ class UpstreamsConfig {
}
}
enum class UpstreamRole {
STANDARD,
FALLBACK
}
open class UpstreamConnection
open class RpcConnection : UpstreamConnection() {

10
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -23,6 +23,7 @@ import org.yaml.snakeyaml.nodes.MappingNode
import org.yaml.snakeyaml.nodes.ScalarNode
import reactor.util.function.Tuples
import java.io.InputStream
import java.lang.IllegalArgumentException
import java.net.URI
import java.time.Duration
@@ -185,6 +186,15 @@ class UpstreamsConfigReader(
internal fun readUpstreamStandard(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
upstream.chain = getValueAsString(upNode, "chain")
getValueAsString(upNode, "role")?.let {
val name = it.trim()
try {
val role = UpstreamsConfig.UpstreamRole.valueOf(name.toUpperCase())
upstream.role = role
} catch (e: IllegalArgumentException) {
log.warn("Unsupported role `$name` for upstream ${upstream.id}")
}
}
if (hasAny(upNode, "labels")) {
getMapping(upNode, "labels")?.let { labels ->
labels.value.stream()

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

@@ -145,7 +145,8 @@ open class ConfiguredUpstreams(
val methods = buildMethods(config, chain)
val upstream = BitcoinUpstream(config.id
?: "bitcoin-${seq.getAndIncrement()}", chain, directApi,
options, QuorumForLabels.QuorumItem(1, config.labels),
options, config.role,
QuorumForLabels.QuorumItem(1, config.labels),
methods)
upstream.start()
@@ -183,7 +184,8 @@ open class ConfiguredUpstreams(
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val ethereumUpstream = EthereumUpstream(
config.id!!,
chain, directApi, wsFactoryApi, options,
chain, directApi, wsFactoryApi,
options, config.role,
QuorumForLabels.QuorumItem(1, config.labels),
methods
)

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

@@ -27,10 +27,12 @@ abstract class DefaultUpstream(
defaultLag: Long,
defaultAvail: UpstreamAvailability,
private val options: UpstreamsConfig.Options,
private val role: UpstreamsConfig.UpstreamRole,
private val targets: CallMethods?
) : Upstream {
constructor(id: String, options: UpstreamsConfig.Options, targets: CallMethods?) : this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, targets)
constructor(id: String, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets)
private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
private val statusStream: TopicProcessor<UpstreamAvailability> = TopicProcessor.create()
@@ -85,6 +87,10 @@ abstract class DefaultUpstream(
return options
}
override fun getRole(): UpstreamsConfig.UpstreamRole {
return role;
}
override fun getMethods(): CallMethods {
return targets ?: throw IllegalStateException("Methods are not set")
}

52
src/main/kotlin/io/emeraldpay/dshackle/upstream/FilteredApis.kt
View File

@@ -16,9 +16,7 @@
*/
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.config.UpstreamsConfig
import org.reactivestreams.Subscriber
import reactor.core.publisher.EmitterProcessor
import reactor.core.publisher.Flux
@@ -33,13 +31,26 @@ class FilteredApis(
allUpstreams: List<Upstream>,
private val matcher: Selector.Matcher,
pos: Int,
private val repeatLimit: Long,
/**
* Limit of retries
*/
private val retryLimit: Long,
jitter: Int
) : ApiSource {
companion object {
private const val DEFAULT_DELAY_STEP = 100
private const val MAX_WAIT_MILLIS = 5000L
@JvmStatic
fun <T> startFrom(upstreams: List<T>, pos: Int): List<T> {
return if (upstreams.size <= 1 || pos == 0) {
upstreams
} else {
val safePosition = pos % upstreams.size
upstreams.subList(safePosition, upstreams.size) + upstreams.subList(0, safePosition)
}
}
}
constructor(allUpstreams: List<Upstream>,
@@ -50,7 +61,8 @@ class FilteredApis(
matcher: Selector.Matcher) : this(allUpstreams, matcher, 0, 10, 10)
private val delay: Int
private val upstreams: List<Upstream>
private val standardUpstreams: List<Upstream>
private val standardWithFallback: List<Upstream>
private val control = EmitterProcessor.create<Boolean>(32, false)
@@ -61,12 +73,20 @@ class FilteredApis(
DEFAULT_DELAY_STEP
}
upstreams = if (allUpstreams.size == 1 || pos == 0 || allUpstreams.isEmpty()) {
allUpstreams
} else {
val safePosition = pos % allUpstreams.size
allUpstreams.subList(safePosition, allUpstreams.size) + allUpstreams.subList(0, safePosition)
standardUpstreams = allUpstreams.filter {
it.getRole() == UpstreamsConfig.UpstreamRole.STANDARD
}.let {
startFrom(it, pos)
}
val fallbackUpstreams = allUpstreams.filter {
it.getRole() == UpstreamsConfig.UpstreamRole.FALLBACK
}.let {
startFrom(it, pos)
}
standardWithFallback = emptyList<Upstream>()
.plus(standardUpstreams)
.plus(fallbackUpstreams)
}
fun waitDuration(rawn: Long): Duration {
@@ -79,9 +99,15 @@ class FilteredApis(
}
override fun subscribe(subscriber: Subscriber<in Upstream>) {
val first = Flux.fromIterable(upstreams)
val retries = (1 until repeatLimit).map { r ->
Flux.fromIterable(upstreams).delaySubscription(waitDuration(r))
// initially try only standard upstreams
val first = Flux.fromIterable(standardUpstreams)
// if all failed, try both standard and fallback upstreams, repeating in cycle
val retries = (0 until (retryLimit - 1)).map { r ->
Flux.fromIterable(standardWithFallback)
// add delay to let upstream to restore if it's a temp failure
// but delay only start of the check, not between upstreams
// i.e. if all upstreams failed -> wait -> check all without waiting in between
.delaySubscription(waitDuration(r + 1))
}.let { Flux.concat(it) }
Flux.concat(first, retries)

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -145,10 +145,16 @@ abstract class Multistream(
else upstreams.map { it.getStatus() }.min()!!
}
//TODO options for multistream are useless
override fun getOptions(): UpstreamsConfig.Options {
return UpstreamsConfig.Options()
}
//TODO roles for multistream are useless
override fun getRole(): UpstreamsConfig.UpstreamRole {
return UpstreamsConfig.UpstreamRole.STANDARD
}
override fun getMethods(): CallMethods {
return callMethods ?: throw IllegalStateException("Methods are not initialized yet")
}

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/Upstream.kt
View File

@@ -31,6 +31,7 @@ interface Upstream {
fun getHead(): Head
fun getApi(): Reader<JsonRpcRequest, JsonRpcResponse>
fun getOptions(): UpstreamsConfig.Options
fun getRole(): UpstreamsConfig.UpstreamRole
fun setLag(lag: Long)
fun getLag(): Long
fun getLabels(): Collection<UpstreamsConfig.Labels>

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

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

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

@@ -40,13 +40,16 @@ open class EthereumUpstream(
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, targets), Upstream, CachesEnabled, Lifecycle {
) : 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(), QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels()),
DirectCallMethods())
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)

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

@@ -60,6 +60,7 @@ open class EthereumGrpcUpstream(
) : DefaultUpstream(
"$parentId/${chain.chainCode}",
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.STANDARD,
null
), Lifecycle {

36
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

@@ -202,4 +202,40 @@ class UpstreamsConfigReaderSpec extends Specification {
where:
id << ["test", "test_test", "test-test", "test123", "test1test", "foo_bar_12"]
}
def "Parse config without fallback role"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-basic.yaml")
when:
def act = reader.read(config)
then:
act != null
act.upstreams.size() == 2
act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD
act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.STANDARD
}
def "Parse config with fallback role"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-roles.yaml")
when:
def act = reader.read(config)
then:
act != null
act.upstreams.size() == 2
act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD
act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.FALLBACK
}
def "Parse config with invalid role"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-roles-invalid.yaml")
when:
def act = reader.read(config)
then:
act != null
act.upstreams.size() == 2
act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.STANDARD
act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.STANDARD
}
}

7
src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy
View File

@@ -15,6 +15,7 @@
*/
package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.FilteredApis
@@ -34,6 +35,7 @@ class QuorumRpcReaderSpec extends Specification {
setup:
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
1 * getApi() >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_test", [])) >> Mono.just(JsonRpcResponse.ok("1"))
}
@@ -60,6 +62,7 @@ class QuorumRpcReaderSpec extends Specification {
setup:
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(1, "test")),
@@ -89,6 +92,7 @@ class QuorumRpcReaderSpec extends Specification {
setup:
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),
@@ -119,6 +123,7 @@ class QuorumRpcReaderSpec extends Specification {
setup:
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(1, "test")),
@@ -148,6 +153,7 @@ class QuorumRpcReaderSpec extends Specification {
setup:
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
3 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),
@@ -178,6 +184,7 @@ class QuorumRpcReaderSpec extends Specification {
setup:
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
3 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),

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

@@ -61,7 +61,9 @@ class EthereumUpstreamMock extends EthereumUpstream {
EthereumUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
super(id, chain, api, null,
UpstreamsConfig.Options.getDefaults(), new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()),
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.STANDARD,
new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()),
methods)
setLag(0)
setStatus(UpstreamAvailability.OK)
@@ -85,4 +87,9 @@ class EthereumUpstreamMock extends EthereumUpstream {
Head getHead() {
return ethereumHeadMock
}
@Override
String toString() {
return "Upstream mock ${getId()}"
}
}

4
src/test/groovy/io/emeraldpay/dshackle/test/TestingCommons.groovy
View File

@@ -43,6 +43,10 @@ class TestingCommons {
return new EthereumApiMock()
}
static EthereumUpstreamMock upstream(String id, Reader<JsonRpcRequest, JsonRpcResponse> api) {
return new EthereumUpstreamMock(id, Chain.ETHEREUM, api)
}
static EthereumUpstreamMock upstream(Reader<JsonRpcRequest, JsonRpcResponse> api) {
return new EthereumUpstreamMock(Chain.ETHEREUM, api)
}

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

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsFactory
import io.emeraldpay.grpc.Chain
import reactor.core.publisher.Flux
import reactor.test.StepVerifier
import spock.lang.Retry
import spock.lang.Specification
@@ -50,6 +51,7 @@ class FilteredApisSpec extends Specification {
TestingCommons.api().tap { it.id = "${i++}" },
(EthereumWsFactory) null,
new UpstreamsConfig.Options(),
UpstreamsConfig.UpstreamRole.STANDARD,
new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)),
ethereumTargets
)
@@ -175,7 +177,52 @@ class FilteredApisSpec extends Specification {
then:
StepVerifier.create(act)
.expectNext(ups[2], ups[3], ups[4], ups[5], ups[0], ups[1])
.expectComplete()
.verify(Duration.ofSeconds(1))
.expectComplete()
.verify(Duration.ofSeconds(1))
}
def "Start with offset"() {
expect:
FilteredApis.startFrom([0, 1, 2, 3, 4], pos) == exp
where:
pos | exp
0 | [0, 1, 2, 3, 4]
1 | [1, 2, 3, 4, 0]
2 | [2, 3, 4, 0, 1]
3 | [3, 4, 0, 1, 2]
4 | [4, 0, 1, 2, 3]
5 | [0, 1, 2, 3, 4]
6 | [1, 2, 3, 4, 0]
}
def "Starts with standard"() {
setup:
List<Upstream> standard = (0..1).collect {
TestingCommons.upstream(
it.toString(),
new EthereumApiStub(it)
)
}
def fallback = [
Mock(Upstream) {
_ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
_ * isAvailable() >> true
}
]
when:
def act = new FilteredApis([] + fallback + standard,
Selector.empty, 0, 3, 0)
act.request(10)
then:
StepVerifier.create(act)
.expectNext(standard[0], standard[1]).as("Initial requests")
.expectNext(standard[0], standard[1]).as("Retry with standard")
.expectNext(fallback[0]).as("Retry with fallback")
.expectNext(standard[0], standard[1]).as("Second retry with standard")
.expectNext(fallback[0]).as("Second retry with fallback")
.expectComplete()
.verify(Duration.ofSeconds(1))
}
}

31
src/test/resources/upstreams-roles-invalid.yaml Normal file
View File

@@ -0,0 +1,31 @@
version: v1
defaults:
- chains:
- ethereum
options:
min-peers: 3
upstreams:
- id: local
chain: ethereum
connection:
ethereum:
rpc:
url: "http://localhost:8545"
ws:
url: "ws://localhost:8546"
origin: "http://localhost"
basic-auth:
username: 9c199ad8f281f20154fc258fe41a6814
password: 258fe4149c199ad8f2811a68f20154fc
- id: infura
chain: ethereum
role: fallbackZ
connection:
ethereum:
rpc:
url: "https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2"
basic-auth:
username: 4fc258fe41a68149c199ad8f281f2015
password: 1a68f20154fc258fe4149c199ad8f281

31
src/test/resources/upstreams-roles.yaml Normal file
View File

@@ -0,0 +1,31 @@
version: v1
defaults:
- chains:
- ethereum
options:
min-peers: 3
upstreams:
- id: local
chain: ethereum
connection:
ethereum:
rpc:
url: "http://localhost:8545"
ws:
url: "ws://localhost:8546"
origin: "http://localhost"
basic-auth:
username: 9c199ad8f281f20154fc258fe41a6814
password: 258fe4149c199ad8f2811a68f20154fc
- id: infura
chain: ethereum
role: fallback
connection:
ethereum:
rpc:
url: "https://mainnet.infura.io/v3/fa28c968191849c1aff541ad1d8511f2"
basic-auth:
username: 4fc258fe41a68149c199ad8f281f2015
password: 1a68f20154fc258fe4149c199ad8f281