Adjustments in availability determination:
- now isAvailable return true in case upstreams are OK or LAGGING - All upstreams in Filtered API now sorted by status
This commit is contained in:
@@ -65,7 +65,7 @@ abstract class DefaultUpstream(
|
|||||||
}
|
}
|
||||||
|
|
||||||
override fun isAvailable(): Boolean {
|
override fun isAvailable(): Boolean {
|
||||||
return getStatus() == UpstreamAvailability.OK
|
return getStatus() == UpstreamAvailability.OK || getStatus() == UpstreamAvailability.LAGGING
|
||||||
}
|
}
|
||||||
|
|
||||||
fun onStatus(value: BlockchainOuterClass.ChainStatus) {
|
fun onStatus(value: BlockchainOuterClass.ChainStatus) {
|
||||||
|
|||||||
@@ -160,11 +160,11 @@ class FilteredApis(
|
|||||||
|
|
||||||
override fun subscribe(subscriber: Subscriber<in Upstream>) {
|
override fun subscribe(subscriber: Subscriber<in Upstream>) {
|
||||||
// initially try only standard upstreams
|
// initially try only standard upstreams
|
||||||
val first = Flux.fromIterable(primaryUpstreams)
|
val first = Flux.fromIterable(primaryUpstreams.sortedBy { it.getStatus().grpcId })
|
||||||
val second = Flux.fromIterable(secondaryUpstreams)
|
val second = Flux.fromIterable(secondaryUpstreams.sortedBy { it.getStatus().grpcId })
|
||||||
// if all failed, try both standard and fallback upstreams, repeating in cycle
|
// if all failed, try both standard and fallback upstreams, repeating in cycle
|
||||||
val retries = (0 until (retryLimit - 1)).map { r ->
|
val retries = (0 until (retryLimit - 1)).map { r ->
|
||||||
Flux.fromIterable(standardWithFallback)
|
Flux.fromIterable(standardWithFallback.sortedBy { it.getStatus().grpcId })
|
||||||
// add a delay to let upstream to restore if it's a temp failure
|
// add a delay to let upstream to restore if it's a temp failure
|
||||||
// but delay only start of the check, not between upstreams
|
// but delay only start of the check, not between upstreams
|
||||||
// i.e. if all upstreams failed -> wait -> check all without waiting in between
|
// i.e. if all upstreams failed -> wait -> check all without waiting in between
|
||||||
|
|||||||
@@ -227,6 +227,7 @@ class FilteredApisSpec extends Specification {
|
|||||||
Mock(Upstream) {
|
Mock(Upstream) {
|
||||||
_ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
|
_ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
|
||||||
_ * isAvailable() >> true
|
_ * isAvailable() >> true
|
||||||
|
_ * getStatus() >> UpstreamAvailability.OK
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
when:
|
when:
|
||||||
@@ -259,12 +260,14 @@ class FilteredApisSpec extends Specification {
|
|||||||
Mock([name: "fallback"], Upstream) {
|
Mock([name: "fallback"], Upstream) {
|
||||||
_ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
|
_ * getRole() >> UpstreamsConfig.UpstreamRole.FALLBACK
|
||||||
_ * isAvailable() >> true
|
_ * isAvailable() >> true
|
||||||
|
_ * getStatus() >> UpstreamAvailability.OK
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
List<Upstream> secondary = [
|
List<Upstream> secondary = [
|
||||||
Mock([name: "secondary"], Upstream) {
|
Mock([name: "secondary"], Upstream) {
|
||||||
_ * getRole() >> UpstreamsConfig.UpstreamRole.SECONDARY
|
_ * getRole() >> UpstreamsConfig.UpstreamRole.SECONDARY
|
||||||
_ * isAvailable() >> true
|
_ * isAvailable() >> true
|
||||||
|
_ * getStatus() >> UpstreamAvailability.OK
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
when:
|
when:
|
||||||
@@ -287,4 +290,35 @@ class FilteredApisSpec extends Specification {
|
|||||||
.expectComplete()
|
.expectComplete()
|
||||||
.verify(Duration.ofSeconds(1))
|
.verify(Duration.ofSeconds(1))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def "Use LAGGING after OK"() {
|
||||||
|
setup:
|
||||||
|
List<Upstream> lagging = [
|
||||||
|
Mock([name: "lagging"], Upstream) {
|
||||||
|
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
|
||||||
|
_ * isAvailable() >> true
|
||||||
|
_ * getStatus() >> UpstreamAvailability.LAGGING
|
||||||
|
}
|
||||||
|
]
|
||||||
|
List<Upstream> ok = [
|
||||||
|
Mock([name: "ok"], Upstream) {
|
||||||
|
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
|
||||||
|
_ * isAvailable() >> true
|
||||||
|
_ * getStatus() >> UpstreamAvailability.OK
|
||||||
|
}
|
||||||
|
]
|
||||||
|
when:
|
||||||
|
def act = new FilteredApis(Chain.ETHEREUM,
|
||||||
|
[] + lagging + ok,
|
||||||
|
Selector.empty, 0, 2, 0)
|
||||||
|
act.request(4)
|
||||||
|
then:
|
||||||
|
StepVerifier.create(act)
|
||||||
|
.expectNext(ok[0]).as("Initial requests with ok")
|
||||||
|
.expectNext(lagging[0]).as("Initial requests with lagging")
|
||||||
|
.expectNext(ok[0]).as("retry requests with ok")
|
||||||
|
.expectNext(lagging[0]).as("retry requests with lagging")
|
||||||
|
.expectComplete()
|
||||||
|
.verify(Duration.ofSeconds(1))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user