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

Support gRPC upstream labels

Browse Source
This commit is contained in:
Maxksim Fomenkov
2022-08-17 13:45:14 +03:00
parent 936049f6e5
commit b9866fe19a
14 changed files with 166 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,8 +20,6 @@ import io.emeraldpay.dshackle.FileResolver
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import org.yaml.snakeyaml.nodes.MappingNode
import org.yaml.snakeyaml.nodes.ScalarNode
import reactor.util.function.Tuples
import java.io.InputStream
import java.net.URI
import java.time.Duration
@@ -83,7 +81,7 @@ class UpstreamsConfigReader(
}
}
getList<MappingNode>(input, "upstreams")?.value?.forEachIndexed { _, upNode ->
getList<MappingNode>(input, "upstreams")?.value?.forEach { upNode ->
val connNode = getMapping(upNode, "connection")
if (hasAny(connNode, "ethereum")) {
readUpstream(config, upNode) {
@@ -211,7 +209,11 @@ class UpstreamsConfigReader(
return connection
}
private fun <T : UpstreamsConfig.UpstreamConnection> readUpstream(config: UpstreamsConfig, upNode: MappingNode, connFactory: () -> T) {
private fun <T : UpstreamsConfig.UpstreamConnection> readUpstream(
config: UpstreamsConfig,
upNode: MappingNode,
connFactory: () -> T
) {
val upstream = UpstreamsConfig.Upstream<T>()
readUpstreamCommon(upNode, upstream)
readUpstreamStandard(upNode, upstream)
@@ -241,18 +243,21 @@ class UpstreamsConfigReader(
getValueAsBool(upNode, "enabled")?.let {
upstream.isEnabled = it
}
if (hasAny(upNode, "labels")) {
getMapping(upNode, "labels")?.let { labels ->
labels.value.stream()
.map { it.keyNode.valueAsString() to it.valueNode.valueAsString() }
.filter { StringUtils.isNotBlank(it.first) && StringUtils.isNotBlank(it.second) }
.forEach {
upstream.labels[it.first!!.trim()] = it.second!!.trim()
}
}
}
}
internal fun readUpstreamGrpc(
upNode: MappingNode,
) {
// Dshackle gRPC connection dispatches requests to different upstreams, which may
// be on different blockchains, and each may have different set of labels.
// So the labels and chains assigned to the gRPC connection make no sense.
if (hasAny(upNode, "labels")) {
// Actual labels from underlying upstreams are handled by GrpcUpstreamStatus
log.warn("Labels should be not applied to gRPC upstream")
}
if (hasAny(upNode, "chain")) {
log.warn("Chain should be not applied to gRPC upstream")
}
@@ -272,18 +277,6 @@ class UpstreamsConfigReader(
log.warn("Unsupported role `$name` for upstream ${upstream.id}")
}
}
if (hasAny(upNode, "labels")) {
getMapping(upNode, "labels")?.let { labels ->
labels.value.stream()
.filter { n -> n.keyNode is ScalarNode && n.valueNode is ScalarNode }
.map { n -> Tuples.of((n.keyNode as ScalarNode).value, (n.valueNode as ScalarNode).value) }
.map { kv -> Tuples.of(kv.t1.trim(), kv.t2.trim()) }
.filter { kv -> StringUtils.isNotEmpty(kv.t1) && StringUtils.isNotEmpty(kv.t2) }
.forEach { kv ->
upstream.labels[kv.t1] = kv.t2
}
}
}
}
internal fun tryReadOptions(upNode: MappingNode): UpstreamsConfig.Options? {

36
src/main/kotlin/io/emeraldpay/dshackle/config/YamlConfigReader.kt
View File

@@ -37,18 +37,12 @@ abstract class YamlConfigReader {
return asMappingNode(yaml.compose(InputStreamReader(input)))
}
protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean {
if (mappingNode == null) {
return false
}
return mappingNode.value
.stream()
.filter { n -> n.keyNode is ScalarNode }
.filter { n ->
val sn = n.keyNode as ScalarNode
key == sn.value
}.count() > 0
}
protected fun hasAny(mappingNode: MappingNode?, key: String): Boolean =
mappingNode?.let { node ->
node.value
.stream()
.anyMatch { it.keyNode.valueAsString() == key }
} ?: false
@Suppress("UNCHECKED_CAST")
private fun <T> getValue(mappingNode: MappingNode?, key: String, type: Class<T>): T? {
@@ -57,21 +51,15 @@ abstract class YamlConfigReader {
}
return mappingNode.value
.stream()
.filter { n -> n.keyNode is ScalarNode && type.isAssignableFrom(n.valueNode.javaClass) }
.filter { n ->
val sn = n.keyNode as ScalarNode
key == sn.value
}
.filter { type.isAssignableFrom(it.valueNode.javaClass) }
.filter { it.keyNode.valueAsString() == key }
.map { n -> n.valueNode as T }
.findFirst().let {
if (it.isPresent) {
it.get()
} else {
null
}
}
.findFirst()
.orElse(null)
}
fun Node.valueAsString(): String? = if (this is ScalarNode) this.value else null
protected fun getMapping(mappingNode: MappingNode?, key: String): MappingNode? {
return getValue(mappingNode, key, MappingNode::class.java)
}

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

@@ -256,7 +256,8 @@ open class ConfiguredUpstreams(
endpoint.port,
endpoint.auth,
fileResolver,
endpoint.upstreamRating
endpoint.upstreamRating,
config.labels
).apply {
timeout = options.timeout
}

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt
View File

@@ -11,12 +11,7 @@ import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
import io.emeraldpay.grpc.Chain
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import io.micrometer.core.instrument.Timer
class EthereumWsConnector(
wsFactory: EthereumWsFactory,

5
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/BitcoinGrpcUpstream.kt
View File

@@ -50,7 +50,8 @@ class BitcoinGrpcUpstream(
role: UpstreamsConfig.UpstreamRole,
chain: Chain,
val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val client: JsonRpcGrpcClient
private val client: JsonRpcGrpcClient,
overrideLabels: UpstreamsConfig.Labels?
) : BitcoinUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
chain,
@@ -95,7 +96,7 @@ class BitcoinGrpcUpstream(
}
}
}
private val upstreamStatus = GrpcUpstreamStatus()
private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
var timeout = Defaults.timeout
private var capabilities: Set<Capability> = emptySet()

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

@@ -49,7 +49,8 @@ open class EthereumGrpcUpstream(
role: UpstreamsConfig.UpstreamRole,
private val chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val client: JsonRpcGrpcClient
private val client: JsonRpcGrpcClient,
overrideLabels: UpstreamsConfig.Labels?
) : EthereumUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
UpstreamsConfig.Options.getDefaults(),
@@ -93,7 +94,7 @@ open class EthereumGrpcUpstream(
}
private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
private val upstreamStatus = GrpcUpstreamStatus()
private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, MostWorkForkChoice())
private var capabilities: Set<Capability> = emptySet()

11
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt
View File

@@ -24,7 +24,11 @@ import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.data.BlockId
import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.startup.QuorumForLabels
import io.emeraldpay.dshackle.upstream.*
import io.emeraldpay.dshackle.upstream.Capability
import io.emeraldpay.dshackle.upstream.Head
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.UpstreamAvailability
import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream
import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice
@@ -50,7 +54,8 @@ open class EthereumPosGrpcUpstream(
private val chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
client: JsonRpcGrpcClient,
nodeRating: Int
nodeRating: Int,
overrideLabels: UpstreamsConfig.Labels?
) : EthereumPosUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
UpstreamsConfig.Options.getDefaults(),
@@ -94,7 +99,7 @@ open class EthereumPosGrpcUpstream(
}
private val log = LoggerFactory.getLogger(EthereumGrpcUpstream::class.java)
private val upstreamStatus = GrpcUpstreamStatus()
private val upstreamStatus = GrpcUpstreamStatus(overrideLabels)
private val grpcHead = GrpcHead(chain, this, remote, blockConverter, reloadBlock, NoChoiceWithPriorityForkChoice(nodeRating))
private var capabilities: Set<Capability> = emptySet()

22
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatus.kt
View File

@@ -24,14 +24,15 @@ import org.slf4j.LoggerFactory
import java.util.Collections
import java.util.concurrent.atomic.AtomicReference
class GrpcUpstreamStatus {
class GrpcUpstreamStatus(
private val overrideLabels: UpstreamsConfig.Labels?
) {
companion object {
private val log = LoggerFactory.getLogger(GrpcUpstreamStatus::class.java)
}
private val allLabels: AtomicReference<Collection<UpstreamsConfig.Labels>> = AtomicReference(emptyList())
private val nodes = AtomicReference<QuorumForLabels>(QuorumForLabels())
private val nodes = AtomicReference(QuorumForLabels())
private var targets: CallMethods? = null
fun update(conf: BlockchainOuterClass.DescribeChain) {
@@ -39,18 +40,17 @@ class GrpcUpstreamStatus {
val updateNodes = QuorumForLabels()
conf.nodesList.forEach { remoteNode ->
val labels = UpstreamsConfig.Labels()
remoteNode.labelsList.forEach {
labels[it.name] = it.value
}
overrideLabels?.let { labels.putAll(it) }
val node = QuorumForLabels.QuorumItem(
remoteNode.quorum,
remoteNode.labelsList.let { provided ->
val labels = UpstreamsConfig.Labels()
provided.forEach {
labels[it.name] = it.value
}
updateLabels.add(labels)
labels
}
labels
)
updateNodes.add(node)
updateLabels.add(labels)
}
this.nodes.set(updateNodes)

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -57,7 +57,8 @@ class GrpcUpstreams(
private val port: Int,
private val auth: AuthConfig.ClientTlsAuth? = null,
private val fileResolver: FileResolver,
private val nodeRating: Int
private val nodeRating: Int,
private val labels: UpstreamsConfig.Labels
) {
private val log = LoggerFactory.getLogger(GrpcUpstreams::class.java)
@@ -205,7 +206,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient)
val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient, labels)
created.timeout = this.timeout
known[chain] = created
created.start()
@@ -221,7 +222,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating)
val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating, labels)
created.timeout = this.timeout
known[chain] = created
created.start()
@@ -237,7 +238,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient)
val created = BitcoinGrpcUpstream(id, role, chain, client!!, rpcClient, labels)
created.timeout = this.timeout
known[chain] = created
created.start()

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

@@ -233,7 +233,8 @@ class UpstreamsConfigReaderSpec extends Specification {
act.upstreams.size() == 2
with(act.upstreams.get(0)) {
connection instanceof UpstreamsConfig.GrpcConnection
labels.isEmpty()
labels.size() == 1
labels["provider"] == "some_service"
}
with(act.upstreams.get(1)) {
!labels.isEmpty()

6
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy
View File

@@ -81,7 +81,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics))
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -139,7 +139,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics))
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -201,7 +201,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
finished.complete(true)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics))
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))

96
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreamStatusSpec.groovy
View File

@@ -25,7 +25,7 @@ class GrpcUpstreamStatusSpec extends Specification {
def "Updates with new labels"() {
setup:
def status = new GrpcUpstreamStatus()
def status = new GrpcUpstreamStatus(null)
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()
@@ -86,9 +86,74 @@ class GrpcUpstreamStatusSpec extends Specification {
]
}
def "Updates with new labels override"() {
setup:
def status = new GrpcUpstreamStatus(UpstreamsConfig.Labels.fromMap([fix: "value"]))
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()
.addNodes(
BlockchainOuterClass.NodeDetails.newBuilder()
.setQuorum(1)
.addLabels(
BlockchainOuterClass.Label.newBuilder().setName("test").setValue("foo")
)
)
.build()
)
def act = status.getLabels()
then:
act.toList() == [
UpstreamsConfig.Labels.fromMap([test: "foo", fix: "value"])
]
// replace with new value
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()
.addNodes(
BlockchainOuterClass.NodeDetails.newBuilder()
.setQuorum(1)
.addLabels(
BlockchainOuterClass.Label.newBuilder().setName("test").setValue("bar")
).addLabels(
BlockchainOuterClass.Label.newBuilder().setName("fix").setValue("val")
)
)
.build()
)
act = status.getLabels()
then:
act.toList() == [
UpstreamsConfig.Labels.fromMap([test: "bar", fix: "value"])
]
// more values
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()
.addNodes(
BlockchainOuterClass.NodeDetails.newBuilder()
.setQuorum(1)
.addLabels(
BlockchainOuterClass.Label.newBuilder().setName("test1").setValue("bar")
)
.addLabels(
BlockchainOuterClass.Label.newBuilder().setName("test2").setValue("baz")
)
)
.build()
)
act = status.getLabels()
then:
act.toList() == [
UpstreamsConfig.Labels.fromMap([test1: "bar", test2: "baz", fix: "value"])
]
}
def "Updates with new nodes"() {
setup:
def status = new GrpcUpstreamStatus()
def status = new GrpcUpstreamStatus(null)
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()
@@ -108,9 +173,34 @@ class GrpcUpstreamStatusSpec extends Specification {
}
}
def "Updates with new nodes override labels"() {
setup:
def status = new GrpcUpstreamStatus(UpstreamsConfig.Labels.fromMap([fix: "value"]))
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()
.addNodes(
BlockchainOuterClass.NodeDetails.newBuilder()
.setQuorum(1)
.addLabels(
BlockchainOuterClass.Label.newBuilder().setName("test").setValue("foo")
)
.addLabels(
BlockchainOuterClass.Label.newBuilder().setName("fix").setValue("val")
)
)
.build()
)
def act = status.getNodes()
then:
act == new QuorumForLabels().tap {
it.add(new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap([test: "foo", fix: "value"])))
}
}
def "Updates with methods"() {
setup:
def status = new GrpcUpstreamStatus()
def status = new GrpcUpstreamStatus(null)
when:
status.update(
BlockchainOuterClass.DescribeChain.newBuilder()

2
src/test/resources/upstreams-labels.yaml
View File

@@ -8,6 +8,8 @@ defaultOptions:
upstreams:
- id: remote
labels:
provider: some_service
connection:
grpc:
host: "10.2.0.15"