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

Switch no response error to null or separate errors (#210)

Browse Source
This commit is contained in:
KirillPamPam
2023-05-22 17:57:32 +04:00
committed by GitHub
parent 72c4f1192b
commit 4f21cb995d
22 changed files with 779 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/kotlin/io/emeraldpay/dshackle/commons/constatnts.kt
View File

@@ -9,6 +9,7 @@ const val SPAN_REQUEST_API_TYPE = "request.api.type"
const val SPAN_REQUEST_UPSTREAM_ID = "request.upstreamId"
const val SPAN_RESPONSE_UPSTREAM_ID = "response.upstreamId"
const val SPAN_REQUEST_ID = "request.id"
const val SPAN_NO_RESPONSE_MESSAGE = "no-response.message"
const val LOCAL_READER = "localReader"
const val REMOTE_QUORUM_RPC_READER = "remoteQuorumRpcReader"

12
src/main/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandler.kt → src/main/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandler.kt
View File

@@ -5,14 +5,14 @@ import brave.handler.SpanHandler
import brave.propagation.TraceContext
import com.fasterxml.jackson.databind.ObjectMapper
import com.github.benmanes.caffeine.cache.Caffeine
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.cloud.sleuth.Span
import java.time.Duration
class ErrorSpanHandler(
class ProviderSpanHandler(
@Qualifier("spanMapper")
private val spanMapper: ObjectMapper,
private val spanExportableList: List<SpanExportable>
) : SpanHandler() {
private val spans = Caffeine
.newBuilder()
@@ -48,7 +48,7 @@ class ErrorSpanHandler(
}
}
return if (spansInfo.hasError) {
return if (spansInfo.exportable) {
spanMapper.writeValueAsString(spansInfo.spans)
} else {
""
@@ -68,13 +68,13 @@ class ErrorSpanHandler(
private fun processSpanInfo(span: MutableSpan, spansInfo: SpansInfo) {
spansInfo.spans.add(span)
if (span.tags().containsKey(SPAN_ERROR)) {
spansInfo.hasError = true
if (spanExportableList.any { it.isExportable(span) }) {
spansInfo.exportable = true
}
}
private data class SpansInfo(
var hasError: Boolean = false,
var exportable: Boolean = false,
val spans: MutableList<MutableSpan> = mutableListOf()
)
}

4
src/main/kotlin/io/emeraldpay/dshackle/config/spans/ServerSpansInterceptor.kt
View File

@@ -9,7 +9,7 @@ import org.springframework.cloud.sleuth.Tracer
class ServerSpansInterceptor(
private val tracer: Tracer,
private val errorSpanHandler: ErrorSpanHandler
private val providerSpanHandler: ProviderSpanHandler
) : ServerInterceptor {
override fun <ReqT : Any?, RespT : Any?> interceptCall(
call: ServerCall<ReqT, RespT>,
@@ -32,7 +32,7 @@ class ServerSpansInterceptor(
tracer.currentSpan()?.let {
val parentId = it.context().parentId()
if (parentId != null) {
val spans = errorSpanHandler.getErrorSpans(it.context().spanId(), it)
val spans = providerSpanHandler.getErrorSpans(it.context().spanId(), it)
if (spans.isNotBlank()) {
headers.put(Metadata.Key.of(SPAN_HEADER, Metadata.ASCII_STRING_MARSHALLER), spans)
}

9
src/main/kotlin/io/emeraldpay/dshackle/config/spans/SpanConfig.kt
View File

@@ -56,13 +56,14 @@ open class SpanConfig {
@Bean
open fun errorSpanHandler(
@Qualifier("spanMapper")
spanMapper: ObjectMapper
): ErrorSpanHandler = ErrorSpanHandler(spanMapper)
spanMapper: ObjectMapper,
spanExportableList: List<SpanExportable>
): ProviderSpanHandler = ProviderSpanHandler(spanMapper, spanExportableList)
@Bean
open fun serverSpansInterceptor(
tracer: org.springframework.cloud.sleuth.Tracer,
errorSpanHandler: ErrorSpanHandler
): ServerInterceptor = ServerSpansInterceptor(tracer, errorSpanHandler)
providerSpanHandler: ProviderSpanHandler
): ServerInterceptor = ServerSpansInterceptor(tracer, providerSpanHandler)
}
}

20
src/main/kotlin/io/emeraldpay/dshackle/config/spans/export.kt Normal file
View File

@@ -0,0 +1,20 @@
package io.emeraldpay.dshackle.config.spans
import brave.handler.MutableSpan
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
import org.springframework.stereotype.Component
interface SpanExportable {
fun isExportable(span: MutableSpan): Boolean
}
@Component
class ErrorSpanExportable : SpanExportable {
override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_ERROR)
}
@Component
class NoResponseSpanExportable : SpanExportable {
override fun isExportable(span: MutableSpan): Boolean = span.tags().containsKey(SPAN_NO_RESPONSE_MESSAGE)
}

9
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumReaderFactory.kt
View File

@@ -20,6 +20,11 @@ import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import org.springframework.cloud.sleuth.Tracer
import java.util.concurrent.atomic.AtomicInteger
interface QuorumReader : Reader<JsonRpcRequest, QuorumRpcReader.Result> {
fun attempts(): AtomicInteger
}
// creates instance of a Quorum based reader
interface QuorumReaderFactory {
@@ -30,10 +35,10 @@ interface QuorumReaderFactory {
}
}
fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): Reader<JsonRpcRequest, QuorumRpcReader.Result>
fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): QuorumReader
class Default : QuorumReaderFactory {
override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): Reader<JsonRpcRequest, QuorumRpcReader.Result> {
override fun create(apis: ApiSource, quorum: CallQuorum, signer: ResponseSigner?, tracer: Tracer): QuorumReader {
return QuorumRpcReader(apis, quorum, signer, tracer)
}
}

38
src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt
View File

@@ -15,10 +15,11 @@
*/
package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.commons.API_READER
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_API_TYPE
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_UPSTREAM_ID
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.reader.SpannedReader
import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.Upstream
@@ -48,7 +49,7 @@ class QuorumRpcReader(
private val quorum: CallQuorum,
private val signer: ResponseSigner?,
private val tracer: Tracer
) : Reader<JsonRpcRequest, QuorumRpcReader.Result> {
) : QuorumReader {
companion object {
private val log = LoggerFactory.getLogger(QuorumRpcReader::class.java)
@@ -56,6 +57,8 @@ class QuorumRpcReader(
constructor(apiControl: ApiSource, quorum: CallQuorum, tracer: Tracer) : this(apiControl, quorum, null, tracer)
override fun attempts(): AtomicInteger = apiControl.attempts()
override fun read(key: JsonRpcRequest): Mono<Result> {
// needs at least one response, so start a request
apiControl.request(1)
@@ -95,7 +98,7 @@ class QuorumRpcReader(
.transform(processResult(defaultResult))
}
fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function<Flux<Upstream>, Mono<CallQuorum>> {
private fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function<Flux<Upstream>, Mono<CallQuorum>> {
val quorumReduce = BiFunction<CallQuorum, Tuple4<ByteArray, Optional<ResponseSigner.Signature>, Upstream, Optional<String>>, CallQuorum> { res, a ->
if (res.record(a.t1, a.t2.orElse(null), a.t3, a.t4.orElse(null))) {
log.trace("Quorum is resolved for method ${key.method}")
@@ -122,7 +125,7 @@ class QuorumRpcReader(
}
}
fun processResult(defaultResult: Mono<Result>): Function<Mono<CallQuorum>, Mono<Result>> {
private fun processResult(defaultResult: Mono<Result>): Function<Mono<CallQuorum>, Mono<Result>> {
return Function { quorumResult ->
quorumResult
.filter { it.isResolved() } // return nothing if not resolved
@@ -134,7 +137,7 @@ class QuorumRpcReader(
}
}
fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple4<ByteArray, Optional<ResponseSigner.Signature>, Upstream, Optional<String>>> {
private fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple4<ByteArray, Optional<ResponseSigner.Signature>, Upstream, Optional<String>>> {
val apiReader = api.getIngressReader()
val spanParams = mapOf(
SPAN_REQUEST_API_TYPE to apiReader.javaClass.name,
@@ -152,7 +155,7 @@ class QuorumRpcReader(
.map { Tuples.of(it.t1, it.t2, api, it.t3) }
}
fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function<Mono<ByteArray>, Mono<Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Optional<String>>>> {
private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function<Mono<ByteArray>, Mono<Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Optional<String>>>> {
return Function { src ->
src.map {
val signature = response.providedSignature
@@ -166,7 +169,7 @@ class QuorumRpcReader(
}
}
fun <T> withErrorResume(api: Upstream, key: JsonRpcRequest): Function<Mono<T>, Mono<T>> {
private fun <T> withErrorResume(api: Upstream, key: JsonRpcRequest): Function<Mono<T>, Mono<T>> {
return Function { src ->
src.onErrorResume { err ->
log.error("Error during call upstream ${api.getId()} with method ${key.method}", err)
@@ -195,22 +198,33 @@ class QuorumRpcReader(
}
}
fun setupDefaultResult(key: JsonRpcRequest): Mono<Result> {
private fun setupDefaultResult(key: JsonRpcRequest): Mono<Result> {
return Mono.just(quorum).flatMap { q ->
if (q.isFailed()) {
val err = q.getError()?.asException(JsonRpcResponse.NumberId(key.id))
?: JsonRpcException(JsonRpcResponse.NumberId(key.id), JsonRpcError(-32603, "Unhandled Upstream error"))
log.warn("Quorum is failed. Method ${key.method}, message ${err.message}")
Mono.error<Result>(err)
Mono.error(err)
} else {
log.warn("Did not get any result from upstream. Method [${key.method}] using [$q]")
Mono.empty<Result>()
noResponse(key.method, q)
}
}
}
fun getValidAttemptsCount(): AtomicInteger =
apiControl.attempts()
private fun noResponse(method: String, q: CallQuorum): Mono<Result> {
return apiControl.upstreamsMatchesResponse()?.run {
tracer.currentSpan()?.tag(SPAN_NO_RESPONSE_MESSAGE, getFullCause())
val cause = getCause(method) ?: return Mono.empty()
if (cause.shouldReturnNull) {
Mono.just(
Result(Global.nullValue, null, 1, emptyList(), null)
)
} else {
Mono.error(RpcException(1, "No response for method $method. Cause - ${cause.cause}"))
}
} ?: Mono.error(RpcException(1, "Quorum [$q] is not resolved [isResolved - ${q.isResolved()}]"))
}
class Result(
val value: ByteArray,

11
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -66,7 +66,6 @@ import reactor.core.publisher.Mono
import reactor.kotlin.core.publisher.toMono
import reactor.util.context.Context
import java.util.EnumMap
import java.util.concurrent.atomic.AtomicInteger
@Service
open class NativeCall(
@@ -392,18 +391,16 @@ open class NativeCall(
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum, signer, tracer)
val counter = if (reader is QuorumRpcReader) {
reader.getValidAttemptsCount()
} else {
AtomicInteger(-1)
}
val counter = reader.attempts()
return SpannedReader(reader, tracer, REMOTE_QUORUM_RPC_READER)
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
.map {
val bytes = ctx.resultDecorator.processResult(it)
validateResult(bytes, "remote", ctx)
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.providedUpstreamId ?: it.resolvers.first().getId(), ctx)
val upId = it.providedUpstreamId
?: if (it.resolvers.isEmpty()) ctx.upstream.getId() else it.resolvers.first().getId()
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx)
}
.onErrorResume { t ->
Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx))

8
src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt
View File

@@ -63,9 +63,11 @@ class TrackBitcoinAddress(
/**
* Criteria for a remote grpc upstream that can provide a balance
*/
private val balanceUpstreamMatcher = Selector.LocalAndMatcher(
Selector.GrpcMatcher(),
Selector.CapabilityMatcher(Capability.BALANCE)
private val balanceUpstreamMatcher = Selector.MultiMatcher(
listOf(
Selector.GrpcMatcher(),
Selector.CapabilityMatcher(Capability.BALANCE)
)
)
@EventListener

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

@@ -29,4 +29,6 @@ interface ApiSource : Publisher<Upstream> {
fun request(tries: Int)
fun attempts(): AtomicInteger
fun upstreamsMatchesResponse(): UpstreamsMatchesResponse?
}

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

@@ -49,6 +49,7 @@ class FilteredApis(
private val retryLimit: Long,
jitter: Int
) : ApiSource {
private val internalMatcher: Selector.Matcher
companion object {
private val log = LoggerFactory.getLogger(FilteredApis::class.java)
@@ -94,6 +95,7 @@ class FilteredApis(
private var started = false
private val control = Sinks.many().unicast().onBackpressureBuffer<Boolean>()
private var upstreamsMatchesResponse: UpstreamsMatchesResponse? = UpstreamsMatchesResponse()
init {
delay = if (jitter > 0) {
@@ -129,6 +131,9 @@ class FilteredApis(
monitoring.countFallback.record(fallbackUpstreams.size.toDouble())
}
}
internalMatcher = Selector.MultiMatcher(
listOf(Selector.AvailabilityMatcher(), matcher)
)
}
private fun getMetrics(chain: Chain): Monitoring {
@@ -181,14 +186,16 @@ class FilteredApis(
}
result.filter { up ->
(up.isAvailable() && matcher.matches(up)).also {
if (it) {
counter.incrementAndGet()
}
}
val matchesResponse = internalMatcher.matchesWithCause(up)
processMatchesResponse(up.getId(), matchesResponse)
matchesResponse.matched()
}
.zipWith(control.asFlux())
.map { it.t1 }
.map {
upstreamsMatchesResponse = null
counter.incrementAndGet()
it.t1
}
.doOnSubscribe {
if (!started) {
// in addition to subscription the FilteredAPI should use request() method to prepare the control flow
@@ -198,6 +205,14 @@ class FilteredApis(
.subscribe(subscriber)
}
private fun processMatchesResponse(upstreamId: String, matchesResponse: MatchesResponse) {
upstreamsMatchesResponse?.run {
if (!matchesResponse.matched()) {
addUpstreamMatchesResponse(upstreamId, matchesResponse)
}
}
}
override fun resolve() {
control.tryEmitComplete()
}
@@ -213,8 +228,10 @@ class FilteredApis(
override fun attempts(): AtomicInteger =
counter
override fun upstreamsMatchesResponse(): UpstreamsMatchesResponse? = upstreamsMatchesResponse
override fun toString(): String {
return "Filter API: ${allUpstreams.size} upstreams with $matcher"
return "Filter API: ${allUpstreams.size} upstreams with $internalMatcher"
}
class Monitoring(chain: Chain) {

83
src/main/kotlin/io/emeraldpay/dshackle/upstream/MatchesResponse.kt Normal file
View File

@@ -0,0 +1,83 @@
package io.emeraldpay.dshackle.upstream
sealed class MatchesResponse {
fun matched(): Boolean {
return when (this) {
is Success -> true
else -> false
}
}
fun getCause(): String? =
when (this) {
is LabelResponse -> "No label `${this.name}` with values ${this.values}"
AvailabilityResponse -> "Upstream is not available"
is CapabilityResponse -> "Upstream does not have capability ${this.capability}"
is ExistsResponse -> "Label ${this.name} does not exist"
GrpcResponse -> "Upstream is not grpc"
is HeightResponse -> "Upstream height ${this.currentHeight} is less than ${this.height}"
is MethodResponse -> "Method ${this.method} is not supported"
is MultiResponse ->
this.allResponses
.filter { it !is Success }
.joinToString("; ") { it.getCause()!! }
is NotMatchedResponse -> "Not matched - ${response.getCause()}"
is SameNodeResponse -> "Upstream does not have hash ${this.upstreamHash}"
else -> null
}
object Success : MatchesResponse()
data class LabelResponse(
val name: String,
val values: Collection<String>
) : MatchesResponse()
data class NotMatchedResponse(
val response: MatchesResponse
) : MatchesResponse()
data class MultiResponse(
private val responses: Set<MatchesResponse>
) : MatchesResponse() {
val allResponses = mutableSetOf<MatchesResponse>()
init {
responses.forEach {
if (it is MultiResponse) {
it.allResponses.forEach { resp ->
allResponses.add(resp)
}
} else {
allResponses.add(it)
}
}
}
}
data class MethodResponse(
val method: String
) : MatchesResponse()
data class ExistsResponse(
val name: String
) : MatchesResponse()
data class CapabilityResponse(
val capability: Capability
) : MatchesResponse()
object GrpcResponse : MatchesResponse()
data class HeightResponse(
val height: Long,
val currentHeight: Long
) : MatchesResponse()
data class SameNodeResponse(
val upstreamHash: Byte
) : MatchesResponse()
object AvailabilityResponse : MatchesResponse()
}

223
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -18,6 +18,14 @@ package io.emeraldpay.dshackle.upstream
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.upstream.MatchesResponse.AvailabilityResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.CapabilityResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.ExistsResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.GrpcResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.HeightResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.NotMatchedResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.SameNodeResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.Success
import org.apache.commons.lang3.StringUtils
import java.util.Collections
@@ -121,17 +129,27 @@ class Selector {
}
}
interface Matcher {
fun matches(up: Upstream): Boolean
abstract class Matcher {
fun matches(up: Upstream): Boolean = matchesWithCause(up).matched()
fun describeInternal(): String
abstract fun matchesWithCause(up: Upstream): MatchesResponse
abstract fun describeInternal(): String
}
data class MultiMatcher(
private val matchers: Collection<Matcher>
) : Matcher {
override fun matches(up: Upstream): Boolean {
return matchers.all { it.matches(up) }
) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse {
val responses = matchers.map { it.matchesWithCause(up) }
return if (responses.all { it is Success }) {
Success
} else {
MatchesResponse.MultiResponse(
responses.filter { it !is Success }.toSet()
)
}
}
fun getMatchers(): Collection<Matcher> {
@@ -158,10 +176,14 @@ class Selector {
data class MethodMatcher(
val method: String
) : Matcher {
override fun matches(up: Upstream): Boolean {
return up.getMethods().isCallable(method)
}
) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse =
if (up.getMethods().isCallable(method)) {
Success
} else {
MatchesResponse.MethodResponse(method)
}
override fun describeInternal(): String {
return "allow method $method"
@@ -172,19 +194,26 @@ class Selector {
}
}
abstract class LabelSelectorMatcher : Matcher {
override fun matches(up: Upstream): Boolean {
return up.getLabels().any(this::matches)
abstract class LabelSelectorMatcher : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse {
val labelsResponses = up.getLabels().map { matchesWithCause(it) }
return if (labelsResponses.any { it is Success }) {
Success
} else {
MatchesResponse.MultiResponse(
labelsResponses.filter { it !is Success }.toSet()
)
}
}
abstract fun matches(labels: UpstreamsConfig.Labels): Boolean
abstract fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse
abstract fun asProto(): BlockchainOuterClass.Selector?
}
class EmptyMatcher : Matcher {
override fun matches(up: Upstream): Boolean {
return true
}
class EmptyMatcher : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse = Success
override fun describeInternal(): String {
return "empty"
@@ -197,18 +226,12 @@ class Selector {
class AnyLabelMatcher : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return true
}
override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse = Success
override fun asProto(): BlockchainOuterClass.Selector? {
return null
}
override fun matches(up: Upstream): Boolean {
return true
}
override fun describeInternal(): String {
return "any label"
}
@@ -218,26 +241,22 @@ class Selector {
}
}
class LocalAndMatcher(vararg val matchers: Matcher) : Matcher {
class LabelMatcher(
val name: String,
val values: Collection<String>
) : LabelSelectorMatcher() {
override fun matches(up: Upstream): Boolean {
return matchers.all { it.matches(up) }
}
override fun describeInternal(): String {
return "local upstream"
}
override fun toString(): String {
return "Matcher: ${describeInternal()}"
}
}
class LabelMatcher(val name: String, val values: Collection<String>) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return labels.get(name)?.let { labelValue ->
override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse {
val response = labels[name]?.let { labelValue ->
values.any { it == labelValue }
} ?: false
return if (response) {
Success
} else {
MatchesResponse.LabelResponse(
name, values
)
}
}
override fun asProto(): BlockchainOuterClass.Selector {
@@ -257,9 +276,19 @@ class Selector {
}
}
class OrMatcher(val matchers: Collection<LabelSelectorMatcher>) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return matchers.any { matcher -> matcher.matches(labels) }
class OrMatcher(
val matchers: Collection<LabelSelectorMatcher>,
) : LabelSelectorMatcher() {
override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse {
val responses = matchers.map { it.matchesWithCause(labels) }
return if (responses.any { it is Success }) {
Success
} else {
MatchesResponse.MultiResponse(
responses.filter { it !is Success }.toSet()
)
}
}
override fun asProto(): BlockchainOuterClass.Selector {
@@ -279,9 +308,19 @@ class Selector {
}
}
class AndMatcher(val matchers: Collection<LabelSelectorMatcher>) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return matchers.all { matcher -> matcher.matches(labels) }
class AndMatcher(
val matchers: Collection<LabelSelectorMatcher>
) : LabelSelectorMatcher() {
override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse {
val responses = matchers.map { it.matchesWithCause(labels) }
return if (responses.all { it is Success }) {
Success
} else {
MatchesResponse.MultiResponse(
responses.filter { it !is Success }.toSet()
)
}
}
override fun asProto(): BlockchainOuterClass.Selector {
@@ -301,9 +340,17 @@ class Selector {
}
}
class NotMatcher(val matcher: LabelSelectorMatcher) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return !matcher.matches(labels)
class NotMatcher(
val matcher: LabelSelectorMatcher
) : LabelSelectorMatcher() {
override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse {
val response = matcher.matchesWithCause(labels)
return if (response !is Success) {
Success
} else {
NotMatchedResponse(response)
}
}
override fun asProto(): BlockchainOuterClass.Selector {
@@ -323,10 +370,16 @@ class Selector {
}
}
class ExistsMatcher(val name: String) : LabelSelectorMatcher() {
override fun matches(labels: UpstreamsConfig.Labels): Boolean {
return labels.containsKey(name)
}
class ExistsMatcher(
val name: String
) : LabelSelectorMatcher() {
override fun matchesWithCause(labels: UpstreamsConfig.Labels): MatchesResponse =
if (labels.containsKey(name)) {
Success
} else {
ExistsResponse(name)
}
override fun asProto(): BlockchainOuterClass.Selector {
return BlockchainOuterClass.Selector.newBuilder().setExistsSelector(
@@ -345,10 +398,14 @@ class Selector {
}
}
class CapabilityMatcher(val capability: Capability) : Matcher {
override fun matches(up: Upstream): Boolean {
return up.getCapabilities().contains(capability)
}
class CapabilityMatcher(val capability: Capability) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse =
if (up.getCapabilities().contains(capability)) {
Success
} else {
CapabilityResponse(capability)
}
override fun describeInternal(): String {
return "provides $capability API"
@@ -359,10 +416,14 @@ class Selector {
}
}
class GrpcMatcher : Matcher {
override fun matches(up: Upstream): Boolean {
return up.isGrpc()
}
class GrpcMatcher : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse =
if (up.isGrpc()) {
Success
} else {
GrpcResponse
}
override fun describeInternal(): String {
return "is gRPC"
@@ -373,9 +434,15 @@ class Selector {
}
}
class HeightMatcher(val height: Long) : Matcher {
override fun matches(up: Upstream): Boolean {
return (up.getHead().getCurrentHeight() ?: 0) >= height
class HeightMatcher(val height: Long) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse {
val currentHeight = up.getHead().getCurrentHeight() ?: 0
return if (currentHeight >= height) {
Success
} else {
HeightResponse(height, currentHeight)
}
}
override fun equals(other: Any?): Boolean {
@@ -400,9 +467,14 @@ class Selector {
}
}
class SameNodeMatcher(private val upstreamHash: Byte) : Matcher {
override fun matches(up: Upstream): Boolean =
up.nodeId() == upstreamHash
class SameNodeMatcher(private val upstreamHash: Byte) : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse =
if (up.nodeId() == upstreamHash) {
Success
} else {
SameNodeResponse(upstreamHash)
}
override fun describeInternal(): String =
"upstream node-id=${upstreamHash.toUByte()}"
@@ -417,4 +489,17 @@ class Selector {
return other.upstreamHash == upstreamHash
}
}
class AvailabilityMatcher : Matcher() {
override fun matchesWithCause(up: Upstream): MatchesResponse =
if (up.isAvailable()) {
Success
} else {
AvailabilityResponse
}
override fun describeInternal(): String = "availability"
override fun toString(): String = "Matcher: ${describeInternal()}"
}
}

99
src/main/kotlin/io/emeraldpay/dshackle/upstream/UpstreamsMatchesResponse.kt Normal file
View File

@@ -0,0 +1,99 @@
package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.upstream.MatchesResponse.HeightResponse
import io.emeraldpay.dshackle.upstream.MatchesResponse.MultiResponse
import kotlin.math.min
class UpstreamsMatchesResponse {
companion object {
private val possibleNullReturnMethods = listOf(
"eth_getTransactionByHash",
"eth_getTransactionReceipt",
"eth_getBlockByHash",
"eth_getBlockByNumber",
"eth_getTransactionByBlockHashAndIndex",
"eth_getTransactionByBlockNumberAndIndex",
"eth_getUncleByBlockHashAndIndex",
"eth_getUncleByBlockNumberAndIndex"
)
}
private val responses = LinkedHashSet<UpstreamNotMatchedResponse>()
fun addUpstreamMatchesResponse(upstreamId: String, response: MatchesResponse) {
if (!response.matched()) {
responses.add(UpstreamNotMatchedResponse(upstreamId, response))
}
}
fun getFullCause(): String? =
if (responses.isEmpty()) {
null
} else {
responses
.joinToString("; ") { "${it.upstreamId} - ${it.matchesResponse.getCause()}" }
.run {
substring(0, min(200, this.length))
}
}
fun getCause(method: String): NotMatchesCause? {
if (responses.isEmpty()) {
return null
}
val commonMatchesResponse = hasCommonMatchesResponse()
return if (commonMatchesResponse is HeightResponse && possibleNullReturnMethods.contains(method)) {
NotMatchesCause(true)
} else if (commonMatchesResponse != null && commonMatchesResponse !is HeightResponse) {
NotMatchesCause(false, commonMatchesResponse.getCause())
} else {
null
}
}
private fun hasCommonMatchesResponse(): MatchesResponse? {
val matchesResponses = responses.map { it.matchesResponses }
val commonResponses = mutableListOf<MatchesResponse>()
for (matchesResponse in matchesResponses[0]) {
if (matchesResponses.all { it.any { resp -> matchesResponse.javaClass == resp.javaClass } }) {
commonResponses.add(matchesResponse)
}
}
if (commonResponses.isEmpty()) {
return null
}
val heightResponse = commonResponses.find { it is HeightResponse }
return heightResponse ?: commonResponses[0]
}
data class NotMatchesCause(
val shouldReturnNull: Boolean,
val cause: String? = null
)
private class UpstreamNotMatchedResponse(
val upstreamId: String,
val matchesResponse: MatchesResponse
) {
val matchesResponses: Set<MatchesResponse> = when (matchesResponse) {
is MultiResponse -> matchesResponse.allResponses
else -> setOf(matchesResponse)
}
override fun equals(other: Any?): Boolean {
if (this === other) return true
if (javaClass != other?.javaClass) return false
other as UpstreamNotMatchedResponse
if (upstreamId != other.upstreamId) return false
return true
}
override fun hashCode(): Int {
return upstreamId.hashCode()
}
}
}

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

@@ -17,9 +17,11 @@ class RemoteUnspentReader(
private val log = LoggerFactory.getLogger(RemoteUnspentReader::class.java)
}
private val selector = Selector.LocalAndMatcher(
Selector.GrpcMatcher(),
Selector.CapabilityMatcher(Capability.BALANCE)
private val selector = Selector.MultiMatcher(
listOf(
Selector.GrpcMatcher(),
Selector.CapabilityMatcher(Capability.BALANCE)
)
)
override fun read(key: Address): Mono<List<SimpleUnspent>> {

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

@@ -289,4 +289,33 @@ class QuorumRpcReaderSpec extends Specification {
.verify(Duration.ofSeconds(1))
}
def "Error if no upstreams"() {
setup:
def api = Stub(Reader)
def up = Mock(Upstream) {
_ * getId() >> "id1"
_ * isAvailable() >> false
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * getIngressReader() >> api
}
def apis = new FilteredApis(
Chain.ETHEREUM,
[up], Selector.empty
)
def reader = new QuorumRpcReader(apis, new AlwaysQuorum(), Stub(Tracer))
when:
def act = reader.read(new JsonRpcRequest("eth_test", []))
.map {
new String(it.value)
}
then:
StepVerifier.create(act)
.expectErrorMatches { t ->
t instanceof RpcException && t.rpcMessage == "No response for method eth_test. Cause - Upstream is not available" && t.error.code == 1
}
.verify(Duration.ofSeconds(4))
}
}

13
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -25,6 +25,7 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.CacheConfig
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
import io.emeraldpay.dshackle.quorum.QuorumReader
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
import io.emeraldpay.dshackle.reader.Reader
@@ -52,6 +53,7 @@ import spock.lang.Ignore
import spock.lang.Specification
import java.time.Duration
import java.util.concurrent.atomic.AtomicInteger
class NativeCallSpec extends Specification {
@@ -131,7 +133,7 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList(ups), null))
}
}
@@ -152,7 +154,8 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * attempts() >> new AtomicInteger(1)
1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.empty()
}
}
@@ -176,7 +179,7 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_test", [], 10)) >> Mono.error(
new JsonRpcException(JsonRpcResponse.Id.from(12), new JsonRpcError(-32123, "Foo Bar", "Foo Bar Baz"), true)
)
@@ -613,7 +616,7 @@ class NativeCallSpec extends Specification {
}
def nativeCall = nativeCall(multistreamHolder)
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null))
}
}
@@ -648,7 +651,7 @@ class NativeCallSpec extends Specification {
}
def nativeCall = nativeCall(multistreamHolder)
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null))
}
}

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

@@ -33,6 +33,8 @@ import spock.lang.Specification
import java.time.Duration
import static java.util.List.of
class FilteredApisSpec extends Specification {
def ethereumTargets = new DefaultEthereumMethods(Chain.ETHEREUM)
@@ -194,6 +196,29 @@ class FilteredApisSpec extends Specification {
.expectNext(ups[2], ups[3], ups[4], ups[5], ups[0], ups[1])
.expectComplete()
.verify(Duration.ofSeconds(1))
act.attempts().get() == 6
}
def "FilteredApis is requested 3 times"() {
setup:
def apis = (0..5).collect {
new EthereumApiStub(it)
}
def ups = apis.collect {
TestingCommons.upstream(it)
}
when:
def act = new FilteredApis(Chain.ETHEREUM, ups, Selector.empty, 2, 1, 0)
act.request(3)
then:
StepVerifier.create(act)
.expectNext(ups[2], ups[3], ups[4])
.then {
act.resolve()
}
.expectComplete()
.verify(Duration.ofSeconds(1))
act.attempts().get() == 3
}
def "Start with offset - 5 items"() {
@@ -330,4 +355,207 @@ class FilteredApisSpec extends Specification {
.expectComplete()
.verify(Duration.ofSeconds(1))
}
def "No upstreams if they all are unavailable"() {
setup:
List<Upstream> ups = [
Mock(Upstream) {
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> false
_ * getId() >> "id1"
_ * getStatus() >> UpstreamAvailability.SYNCING
},
Mock(Upstream) {
_ * getId() >> "id2"
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> false
_ * getStatus() >> UpstreamAvailability.SYNCING
}
]
when:
def act = new FilteredApis(Chain.ETHEREUM, ups, Selector.empty)
act.request(1)
then:
StepVerifier.create(act)
.expectNextCount(0)
.expectComplete()
.verify(Duration.ofSeconds(5))
act.upstreamsMatchesResponse() != null
act.upstreamsMatchesResponse().getFullCause() == "id1 - Upstream is not available; id2 - Upstream is not available"
act.upstreamsMatchesResponse().getCause("").cause == "Upstream is not available"
}
def "No upstreams if they all are not matched"() {
setup:
List<Upstream> ups = [
Mock(Upstream) {
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> true
_ * getId() >> "id1"
_ * getStatus() >> UpstreamAvailability.OK
_ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive")))
},
Mock(Upstream) {
_ * getId() >> "id2"
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> true
_ * getStatus() >> UpstreamAvailability.OK
_ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive")))
}
]
when:
def act = new FilteredApis(Chain.ETHEREUM, ups, new Selector.LabelMatcher("node", of("test")))
act.request(1)
then:
StepVerifier.create(act)
.expectNextCount(0)
.expectComplete()
.verify(Duration.ofSeconds(5))
act.upstreamsMatchesResponse() != null
act.upstreamsMatchesResponse().getFullCause() == "id1 - No label `node` with values [test]; id2 - No label `node` with values [test]"
act.upstreamsMatchesResponse().getCause("").cause == "No label `node` with values [test]"
}
def "No upstreams if they all are not matched by first matcher"() {
setup:
List<Upstream> ups = [
Mock(Upstream) {
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> false
_ * getId() >> "id1"
_ * getStatus() >> UpstreamAvailability.OK
_ * getHead() >> Mock(Head) {
_ * getCurrentHeight() >> 100000
}
_ * getLabels() >> of(
UpstreamsConfig.Labels.fromMap(
Map.of("node", "archive", "type", "super")
)
)
},
Mock(Upstream) {
_ * getId() >> "id2"
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> false
_ * getHead() >> Mock(Head) {
_ * getCurrentHeight() >> 100000
}
_ * getStatus() >> UpstreamAvailability.OK
_ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive")))
}
]
when:
def act = new FilteredApis(
Chain.ETHEREUM, ups,
new Selector.MultiMatcher(
of(
new Selector.HeightMatcher(100000000),
)
)
)
act.request(1)
then:
StepVerifier.create(act)
.expectNextCount(0)
.expectComplete()
.verify(Duration.ofSeconds(5))
act.upstreamsMatchesResponse() != null
act.upstreamsMatchesResponse().getFullCause() == "id1 - Upstream is not available; Upstream height 100000 is less than 100000000; id2 - Upstream is not available; Upstream height 100000 is less than 100000000"
act.upstreamsMatchesResponse().getCause("eth_getTransactionByHash").cause == null
act.upstreamsMatchesResponse().getCause("eth_getTransactionByHash").shouldReturnNull
}
def "No upstreams if they all are not matched and return null cause"() {
setup:
List<Upstream> ups = [
Mock(Upstream) {
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> false
_ * getId() >> "id1"
_ * getStatus() >> UpstreamAvailability.OK
_ * getHead() >> Mock(Head) {
_ * getCurrentHeight() >> 100000
}
_ * getLabels() >> of(
UpstreamsConfig.Labels.fromMap(
Map.of("node", "archive", "type", "super")
)
)
},
Mock(Upstream) {
_ * getId() >> "id2"
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> false
_ * getHead() >> Mock(Head) {
_ * getCurrentHeight() >> 100000
}
_ * getStatus() >> UpstreamAvailability.OK
_ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive")))
}
]
when:
def act = new FilteredApis(
Chain.ETHEREUM, ups,
new Selector.MultiMatcher(
of(
new Selector.HeightMatcher(100000000),
)
)
)
act.request(1)
then:
StepVerifier.create(act)
.expectNextCount(0)
.expectComplete()
.verify(Duration.ofSeconds(5))
act.upstreamsMatchesResponse() != null
act.upstreamsMatchesResponse().getFullCause() == "id1 - Upstream is not available; Upstream height 100000 is less than 100000000; id2 - Upstream is not available; Upstream height 100000 is less than 100000000"
act.upstreamsMatchesResponse().getCause("other") == null
}
def "Second upstream if first is not matched"() {
setup:
def up = Mock(Upstream) {
_ * getId() >> "id2"
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> true
_ * getHead() >> Mock(Head) {
_ * getCurrentHeight() >> 100000001
}
_ * getStatus() >> UpstreamAvailability.OK
_ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "test")))
}
List<Upstream> ups = [
Mock(Upstream) {
_ * getRole() >> UpstreamsConfig.UpstreamRole.PRIMARY
_ * isAvailable() >> true
_ * getId() >> "id1"
_ * getStatus() >> UpstreamAvailability.OK
_ * getHead() >> Mock(Head) {
_ * getCurrentHeight() >> 100000
}
_ * getLabels() >> of(UpstreamsConfig.Labels.fromMap(Map.of("node", "archive")))
}, up
]
when:
def act = new FilteredApis(
Chain.ETHEREUM, ups,
new Selector.MultiMatcher(
of(
new Selector.HeightMatcher(100000000),
new Selector.LabelMatcher("node", of("test"))
)
)
)
act.request(1)
then:
StepVerifier.create(act)
.expectNext(up)
.then {
act.resolve()
}
.expectComplete()
.verify(Duration.ofSeconds(5))
act.upstreamsMatchesResponse() == null
}
}

30
src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy
View File

@@ -178,9 +178,12 @@ class SelectorSpec extends Specification {
def "LABEL matches single label"() {
setup:
def matcher = new Selector.LabelMatcher("test", ["foo"])
def up = Mock(Upstream) {
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)]
}
expect:
matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
matcher.matches(up)
where:
maps << [
@@ -193,9 +196,12 @@ class SelectorSpec extends Specification {
def "LABEL matches one label two values"() {
setup:
def matcher = new Selector.LabelMatcher("test", ["foo", "bar"])
def up = Mock(Upstream) {
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)]
}
expect:
matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
matcher.matches(up)
where:
maps << [
@@ -212,9 +218,12 @@ class SelectorSpec extends Specification {
new Selector.LabelMatcher("test", ["foo", "bar"])
]
)
def up = Mock(Upstream) {
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)]
}
expect:
matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
matcher.matches(up)
where:
maps << [
@@ -232,9 +241,12 @@ class SelectorSpec extends Specification {
new Selector.LabelMatcher("test2", ["baz"])
]
)
def up = Mock(Upstream) {
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)]
}
expect:
matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
matcher.matches(up)
where:
maps << [
@@ -252,9 +264,12 @@ class SelectorSpec extends Specification {
new Selector.LabelMatcher("test2", ["baz"])
]
)
def up = Mock(Upstream) {
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)]
}
expect:
matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
matcher.matches(up)
where:
maps << [
@@ -277,9 +292,12 @@ class SelectorSpec extends Specification {
)
]
)
def up = Mock(Upstream) {
1 * getLabels() >> [UpstreamsConfig.Labels.fromMap(maps)]
}
expect:
matcher.matches(UpstreamsConfig.Labels.fromMap(maps))
matcher.matches(up)
where:
maps << [

30
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy
View File

@@ -5,9 +5,9 @@ import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.cache.Caches
import io.emeraldpay.dshackle.cache.CurrentBlockCache
import io.emeraldpay.dshackle.data.DefaultContainer
import io.emeraldpay.dshackle.quorum.QuorumReader
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.test.TestingCommons
import io.emeraldpay.dshackle.upstream.ApiSource
import io.emeraldpay.dshackle.upstream.Head
@@ -56,7 +56,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _,) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null)
@@ -86,7 +86,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null
@@ -124,7 +124,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null
@@ -160,7 +160,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null
@@ -196,7 +196,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList<Byte>(), null
@@ -233,7 +233,7 @@ class EthereumDirectReaderSpec extends Specification {
up, caches, new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList<Byte>(), null
@@ -261,7 +261,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null
@@ -292,7 +292,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null
@@ -324,7 +324,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null
@@ -365,11 +365,11 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) {
2 * create(_, _, _, _) >> Mock(Reader) {
2 * create(_, _, _, _) >> Mock(QuorumReader) {
2 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >>>
[Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
}
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> result
}
}
@@ -408,11 +408,11 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
ethereumDirectReader.quorumReaderFactory = Mock(QuorumReaderFactory) {
2 * create(_, _, _, _) >> Mock(Reader) {
2 * create(_, _, _, _) >> Mock(QuorumReader) {
2 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >>>
[Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]
}
1 * create(_, _, _, _) >> Mock(Reader) {
1 * create(_, _, _, _) >> Mock(QuorumReader) {
1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> result
}
}
@@ -442,7 +442,7 @@ class EthereumDirectReaderSpec extends Specification {
up, Caches.default(), new CurrentBlockCache(), calls, TestingCommons.tracerMock()
)
reader.quorumReaderFactory = Mock(QuorumReaderFactory) {
4 * create(_, _, _, _) >> Mock(Reader) {
4 * create(_, _, _, _) >> Mock(QuorumReader) {
4 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >>>
[Mono.error(new RuntimeException()), Mono.error(new RuntimeException()),
Mono.error(new RuntimeException()), Mono.error(new RuntimeException())]

6
src/test/kotlin/io/emeraldpay/dshackle/config/spans/CollectSpanConfigTest.kt
View File

@@ -37,7 +37,7 @@ class CollectSpanConfigTest {
appCtx.getBean(SpanConfig::class.java)
}
assertThrows(NoSuchBeanDefinitionException::class.java) {
appCtx.getBean(ErrorSpanHandler::class.java)
appCtx.getBean(ProviderSpanHandler::class.java)
}
assertThrows(NoSuchBeanDefinitionException::class.java) {
appCtx.getBean(ServerSpansInterceptor::class.java)
@@ -72,7 +72,7 @@ class CollectSpanConfigTest {
appCtx.getBean(SpanConfig::class.java)
}
assertDoesNotThrow {
appCtx.getBean(ErrorSpanHandler::class.java)
appCtx.getBean(ProviderSpanHandler::class.java)
}
assertDoesNotThrow {
appCtx.getBean(ServerSpansInterceptor::class.java)
@@ -113,7 +113,7 @@ class CollectSpanConfigTest {
appCtx.getBean(SpanConfig::class.java)
}
assertThrows(NoSuchBeanDefinitionException::class.java) {
appCtx.getBean(ErrorSpanHandler::class.java)
appCtx.getBean(ProviderSpanHandler::class.java)
}
assertThrows(NoSuchBeanDefinitionException::class.java) {
appCtx.getBean(ServerSpansInterceptor::class.java)

36
src/test/kotlin/io/emeraldpay/dshackle/config/spans/ErrorSpanHandlerTest.kt → src/test/kotlin/io/emeraldpay/dshackle/config/spans/ProviderSpanHandlerTest.kt
View File

@@ -5,16 +5,20 @@ import brave.handler.SpanHandler
import brave.propagation.TraceContext
import com.fasterxml.jackson.module.kotlin.readValue
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import io.emeraldpay.dshackle.commons.SPAN_NO_RESPONSE_MESSAGE
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.MethodSource
import org.mockito.Mockito
import org.mockito.Mockito.`when`
import org.springframework.cloud.sleuth.Span
import org.springframework.cloud.sleuth.brave.bridge.BraveTraceContext
class ErrorSpanHandlerTest {
class ProviderSpanHandlerTest {
private val mapper = SpanConfig().spanMapper()
private val spanExportableList = listOf(ErrorSpanExportable(), NoResponseSpanExportable())
private val ctx = TraceContext.newBuilder()
.traceId(1223324)
.spanId(234235)
@@ -48,13 +52,11 @@ class ErrorSpanHandlerTest {
assertEquals("", result)
}
@Test
fun `span with length of traceId greater than 20 and with parentId is collected`() {
val spanId = "f7e83f2b69ec684d"
@ParameterizedTest
@MethodSource("spans")
fun `span with length of traceId greater than 20 and with parentId is collected`(span: MutableSpan) {
val currentSpan = Mockito.mock(Span::class.java)
val handler = spanHandler()
val span = span("6666632728347823749827349723985", spanId)
.apply { parentId("f7e83f2b69ec682d") }
`when`(currentSpan.context()).thenReturn(BraveTraceContext(ctx))
@@ -92,5 +94,25 @@ class ErrorSpanHandlerTest {
tag(SPAN_ERROR, "true")
}
private fun spanHandler() = ErrorSpanHandler(mapper)
companion object {
@JvmStatic
fun spans() = listOf(
MutableSpan()
.apply {
traceId("6666632728347823749827349723985")
id("f7e83f2b69ec682d")
tag(SPAN_ERROR, "true")
parentId("f7e83f2b69ec682d")
},
MutableSpan()
.apply {
traceId("6666632728347823749827349723985")
id("f7e83f2b69ec111d")
parentId("f7e83f2b69ec682d")
tag(SPAN_NO_RESPONSE_MESSAGE, "noResp")
}
)
}
private fun spanHandler() = ProviderSpanHandler(mapper, spanExportableList)
}