426 lines
18 KiB
Kotlin
426 lines
18 KiB
Kotlin
/**
|
|
* Copyright (c) 2020 EmeraldPay, Inc
|
|
* 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.config
|
|
|
|
import io.emeraldpay.dshackle.FileResolver
|
|
import io.emeraldpay.dshackle.config.UpstreamsConfig.ManualBoundSetting
|
|
import io.emeraldpay.dshackle.foundation.ChainOptions
|
|
import io.emeraldpay.dshackle.foundation.ChainOptionsReader
|
|
import io.emeraldpay.dshackle.foundation.YamlConfigReader
|
|
import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
|
|
import io.emeraldpay.dshackle.upstream.lowerbound.ManualLowerBoundType
|
|
import org.apache.commons.lang3.StringUtils
|
|
import org.slf4j.LoggerFactory
|
|
import org.yaml.snakeyaml.nodes.MappingNode
|
|
import org.yaml.snakeyaml.nodes.Tag
|
|
import java.io.InputStream
|
|
import java.net.URI
|
|
import java.util.Locale
|
|
|
|
const val DEFAULT_MAX_CONNECTIONS = 1500
|
|
const val DEFAULT_QUEUE_SIZE = 1000
|
|
|
|
class UpstreamsConfigReader(
|
|
private val fileResolver: FileResolver,
|
|
private val optionsReader: ChainOptionsReader,
|
|
) : YamlConfigReader<UpstreamsConfig>() {
|
|
|
|
private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
|
|
private val authConfigReader = AuthConfigReader()
|
|
private val knownNodeIds: MutableSet<Int> = HashSet()
|
|
|
|
override fun read(input: MappingNode?): UpstreamsConfig? {
|
|
return getMapping(input, "cluster")?.let {
|
|
readInternal(it)
|
|
}
|
|
}
|
|
|
|
fun readInternal(input: InputStream): UpstreamsConfig {
|
|
val configNode = readNode(input)
|
|
return readInternal(configNode)
|
|
}
|
|
private fun readInternal(input: MappingNode?): UpstreamsConfig {
|
|
val config = UpstreamsConfig()
|
|
knownNodeIds.clear()
|
|
|
|
getList<MappingNode>(input, "defaults")?.value?.forEach { opts ->
|
|
val defaultOptions = ChainOptions.DefaultOptions()
|
|
config.defaultOptions.add(defaultOptions)
|
|
defaultOptions.chains = getListOfString(opts, "chains")
|
|
getMapping(opts, "options")?.let { values ->
|
|
defaultOptions.options = optionsReader.readOptions(values)
|
|
}
|
|
}
|
|
|
|
config.upstreams = ArrayList()
|
|
|
|
getValueAsString(input, "include")?.let { path ->
|
|
fileResolver.resolve(path).let { file ->
|
|
if (file.exists() && file.isFile && file.canRead()) {
|
|
readInternal(file.inputStream()).let {
|
|
it.upstreams.forEach { upstream -> config.upstreams.add(upstream) }
|
|
}
|
|
} else {
|
|
log.warn("Failed to read config from $path")
|
|
}
|
|
}
|
|
}
|
|
|
|
getListOfString(input, "include")?.forEach { path ->
|
|
fileResolver.resolve(path).let { file ->
|
|
if (file.exists() && file.isFile && file.canRead()) {
|
|
readInternal(file.inputStream()).let {
|
|
it.upstreams.forEach { upstream -> config.upstreams.add(upstream) }
|
|
}
|
|
} else {
|
|
log.warn("Failed to read config from $path")
|
|
}
|
|
}
|
|
}
|
|
|
|
getList<MappingNode>(input, "upstreams")?.value?.forEach { upNode ->
|
|
val connNode = getMapping(upNode, "connection")
|
|
if (hasAny(connNode, "generic")) {
|
|
readUpstream(config, upNode) {
|
|
readRpcConnection(getMapping(connNode, "generic")!!)
|
|
}
|
|
} else if (hasAny(connNode, "ethereum")) {
|
|
readUpstream(config, upNode) {
|
|
readRpcConnection(getMapping(connNode, "ethereum")!!)
|
|
}
|
|
} else if (hasAny(connNode, "bitcoin")) {
|
|
readUpstream(config, upNode) {
|
|
readBitcoinConnection(getMapping(connNode, "bitcoin")!!)
|
|
}
|
|
} else if (hasAny(connNode, "ethereum-pos")) {
|
|
readUpstream(config, upNode) {
|
|
readEthereumPosConnection(getMapping(connNode, "ethereum-pos")!!)
|
|
}
|
|
} else if (hasAny(connNode, "grpc")) {
|
|
val connConfigNode = getMapping(connNode, "grpc")!!
|
|
val upstream = UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>()
|
|
readUpstreamCommon(upNode, upstream)
|
|
readUpstreamGrpc(upNode)
|
|
if (isValid(upstream)) {
|
|
config.upstreams.add(upstream)
|
|
val connection = UpstreamsConfig.GrpcConnection()
|
|
upstream.connection = connection
|
|
getValueAsInt(connConfigNode, "upstream-rating")?.let {
|
|
connection.upstreamRating = it
|
|
}
|
|
getValueAsString(connConfigNode, "host")?.let {
|
|
connection.host = it
|
|
}
|
|
getValueAsInt(connConfigNode, "port")?.let {
|
|
connection.port = it
|
|
}
|
|
connection.auth = authConfigReader.readClientTls(connConfigNode)
|
|
connection.tokenAuth = authConfigReader.readTokenAuth(connConfigNode)
|
|
} else {
|
|
log.error("Upstream at #0 has invalid configuration")
|
|
}
|
|
}
|
|
}
|
|
|
|
return config
|
|
}
|
|
|
|
private fun readBitcoinConnection(connConfigNode: MappingNode): UpstreamsConfig.BitcoinConnection {
|
|
val connection = UpstreamsConfig.BitcoinConnection()
|
|
.apply { rpc = readRpcConfig(connConfigNode) }
|
|
|
|
getMapping(connConfigNode, "esplora")?.let { node ->
|
|
getValueAsString(node, "url")?.let { url ->
|
|
val http = UpstreamsConfig.HttpEndpoint(URI(url), DEFAULT_MAX_CONNECTIONS, DEFAULT_QUEUE_SIZE)
|
|
http.basicAuth = authConfigReader.readClientBasicAuth(node)
|
|
http.bearerAuth = readBearerAuth(node, http.basicAuth, url)
|
|
http.tls = authConfigReader.readClientTls(node)
|
|
connection.esplora = http
|
|
}
|
|
}
|
|
getMapping(connConfigNode, "zeromq")?.let { node ->
|
|
getValueAsString(node, "address")?.let { address ->
|
|
val zmqConfig: Pair<String, Int>? = try {
|
|
if (address.contains(":")) {
|
|
address.split(":").let {
|
|
Pair(it[0], it[1].toInt())
|
|
}
|
|
} else {
|
|
Pair("127.0.0.1", address.toInt())
|
|
}
|
|
} catch (t: Throwable) {
|
|
log.warn("Invalid config for ZeroMQ: $address. Expected to be in format HOST:PORT")
|
|
null
|
|
}
|
|
zmqConfig?.let {
|
|
connection.zeroMq = UpstreamsConfig.BitcoinZeroMq(it.first, it.second)
|
|
}
|
|
}
|
|
}
|
|
return connection
|
|
}
|
|
|
|
private fun readBearerAuth(
|
|
node: MappingNode?,
|
|
basicAuth: AuthConfig.ClientBasicAuth?,
|
|
url: String,
|
|
): AuthConfig.ClientBearerAuth? {
|
|
val bearerAuth = authConfigReader.readClientBearerAuth(node)
|
|
if (bearerAuth != null && basicAuth != null) {
|
|
log.warn("Both basic-auth and bearer-auth are configured for $url, basic-auth is used")
|
|
return null
|
|
}
|
|
return bearerAuth
|
|
}
|
|
|
|
private fun readRpcConfig(connConfigNode: MappingNode): UpstreamsConfig.HttpEndpoint? {
|
|
return getMapping(connConfigNode, "rpc")?.let { node ->
|
|
val maxConnections = getValueAsInt(node, "max-connections") ?: DEFAULT_MAX_CONNECTIONS
|
|
val queueSize = getValueAsInt(node, "queue-size") ?: DEFAULT_QUEUE_SIZE
|
|
|
|
getValueAsString(node, "url")?.let { url ->
|
|
val http = UpstreamsConfig.HttpEndpoint(URI(url), maxConnections, queueSize)
|
|
http.basicAuth = authConfigReader.readClientBasicAuth(node)
|
|
http.bearerAuth = readBearerAuth(node, http.basicAuth, url)
|
|
http.tls = authConfigReader.readClientTls(node)
|
|
http
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun readEthereumPosConnection(connConfigNode: MappingNode): UpstreamsConfig.EthereumPosConnection {
|
|
val connection = UpstreamsConfig.EthereumPosConnection()
|
|
getMapping(connConfigNode, "execution")?.let {
|
|
connection.execution = readRpcConnection(it)
|
|
}
|
|
getValueAsInt(connConfigNode, "upstream-rating")?.let {
|
|
connection.upstreamRating = it
|
|
}
|
|
return connection
|
|
}
|
|
|
|
private fun readRpcConnection(connConfigNode: MappingNode): UpstreamsConfig.RpcConnection {
|
|
val connection = UpstreamsConfig.RpcConnection()
|
|
.apply { rpc = readRpcConfig(connConfigNode) }
|
|
|
|
getValueAsString(connConfigNode, "connector-mode")?.let {
|
|
connection.connectorMode = it
|
|
}
|
|
|
|
getList<MappingNode>(connConfigNode, "additional")
|
|
?.value
|
|
?.mapNotNull {
|
|
connection.addEndpoint(readRpcConnection(it))
|
|
}
|
|
|
|
getValueAsString(connConfigNode, "tag")?.let {
|
|
connection.tag = it
|
|
}
|
|
|
|
getMapping(connConfigNode, "ws")?.let { node ->
|
|
getValueAsString(node, "url")?.let { url ->
|
|
val ws = UpstreamsConfig.WsEndpoint(URI(url))
|
|
connection.ws = ws
|
|
getValueAsString(node, "origin")?.let { origin ->
|
|
ws.origin = URI(origin)
|
|
}
|
|
ws.basicAuth = authConfigReader.readClientBasicAuth(node)
|
|
ws.bearerAuth = readBearerAuth(node, ws.basicAuth, url)
|
|
|
|
getValueAsBytes(node, "frameSize")?.let {
|
|
if (it < 65_535) {
|
|
throw IllegalStateException("frameSize cannot be less than 64Kb")
|
|
}
|
|
ws.frameSize = it
|
|
}
|
|
getValueAsBytes(node, "msgSize")?.let {
|
|
if (it < 65_535) {
|
|
throw IllegalStateException("msgSize cannot be less than 64Kb")
|
|
}
|
|
ws.msgSize = it
|
|
}
|
|
getValueAsInt(node, "connections")?.let {
|
|
if (it < 1 || it > 1024) {
|
|
throw IllegalStateException("connection limit should be in 1..1024")
|
|
}
|
|
ws.connections = it
|
|
}
|
|
}
|
|
}
|
|
return connection
|
|
}
|
|
|
|
private fun <T : UpstreamsConfig.UpstreamConnection> readUpstream(
|
|
config: UpstreamsConfig,
|
|
upNode: MappingNode,
|
|
connFactory: (String?) -> T,
|
|
) {
|
|
val upstream = UpstreamsConfig.Upstream<T>()
|
|
readUpstreamCommon(upNode, upstream)
|
|
readUpstreamStandard(upNode, upstream)
|
|
if (isValid(upstream)) {
|
|
config.upstreams.add(upstream)
|
|
upstream.connection = connFactory(upstream.chain)
|
|
} else {
|
|
log.error("Upstream at #0 has invalid configuration")
|
|
}
|
|
}
|
|
|
|
fun isValid(upstream: UpstreamsConfig.Upstream<*>): Boolean {
|
|
val id = upstream.id
|
|
// In general, we just check that id is suitable for urls and references,
|
|
// Besides that, if Response Signatures are enabled (CurrentResponseSigner) then it's critical that the id cannot have `/` symbol
|
|
if (id == null || id.length < 3 || !id.matches(Regex("[a-zA-Z][a-zA-Z0-9_-]+[a-zA-Z0-9]"))) {
|
|
log.warn("Invalid id: $id")
|
|
return false
|
|
}
|
|
return upstream.nodeId?.let {
|
|
if (it !in 1..65535) {
|
|
log.warn("Invalid node-id: $it. Must be in range [1, 65535].")
|
|
false
|
|
} else if (!knownNodeIds.add(it)) {
|
|
log.warn("Duplicated node-id: $it. Must be in unique.")
|
|
false
|
|
} else {
|
|
true
|
|
}
|
|
} ?: true
|
|
}
|
|
|
|
private fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
|
|
upstream.id = getValueAsString(upNode, "id")
|
|
upstream.nodeId = getValueAsInt(upNode, "node-id")
|
|
upstream.options = optionsReader.read(upNode)
|
|
upstream.methods = tryReadMethods(upNode)
|
|
upstream.methodGroups = tryReadMethodGroups(upNode)
|
|
upstream.additionalSettings = readAdditionalSettings(upNode)
|
|
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()
|
|
}
|
|
}
|
|
}
|
|
if (hasAny(upNode, "custom-headers")) {
|
|
getMapping(upNode, "custom-headers")?.let { headers ->
|
|
val headersMap = headers.value
|
|
.map { it.keyNode.valueAsString() to it.valueNode.valueAsString() }
|
|
.filter { StringUtils.isNotBlank(it.first) && StringUtils.isNotBlank(it.second) }
|
|
.associate { it.first!!.trim() to it.second!!.trim() }
|
|
upstream.customHeaders = headersMap
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun readAdditionalSettings(upNode: MappingNode): UpstreamsConfig.AdditionalSettings? {
|
|
return getMapping(upNode, "additional-settings")
|
|
?.let { settings ->
|
|
val manualLowerBounds = getMapping(settings, "manual-lower-bounds")
|
|
?.let { bounds ->
|
|
bounds.value.asSequence()
|
|
.filter {
|
|
StringUtils.isNotBlank(it.keyNode.valueAsString()) && it.valueNode.tag == Tag.MAP
|
|
}
|
|
.map {
|
|
val setting = readManualBoundSetting(it.valueNode as MappingNode)
|
|
|
|
if (setting != null) {
|
|
LowerBoundType.byName(it.keyNode.valueAsString()!!) to setting
|
|
} else {
|
|
null
|
|
}
|
|
}
|
|
.filterNotNull()
|
|
.associate { it.first to it.second }
|
|
} ?: emptyMap()
|
|
UpstreamsConfig.AdditionalSettings(manualLowerBounds)
|
|
}
|
|
}
|
|
|
|
private fun readManualBoundSetting(node: MappingNode): ManualBoundSetting? {
|
|
val type = getValueAsString(node, "type")
|
|
?.let { ManualLowerBoundType.byName(it) } ?: return null
|
|
val value = getValueAsLong(node, "value") ?: return null
|
|
return ManualBoundSetting(type, value)
|
|
}
|
|
|
|
private fun readUpstreamGrpc(
|
|
upNode: MappingNode,
|
|
) {
|
|
if (hasAny(upNode, "chain")) {
|
|
log.warn("Chain should be not applied to gRPC upstream")
|
|
}
|
|
}
|
|
|
|
private fun readUpstreamStandard(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
|
|
upstream.chain = getValueAsString(upNode, "chain")
|
|
getValueAsString(upNode, "role")?.let {
|
|
val name = it.trim().let {
|
|
// `standard` was initial role, now split into `primary` and `secondary`
|
|
if (it == "standard") "primary" else it
|
|
}
|
|
try {
|
|
val role = UpstreamsConfig.UpstreamRole.valueOf(name.uppercase(Locale.getDefault()))
|
|
upstream.role = role
|
|
} catch (e: IllegalArgumentException) {
|
|
log.warn("Unsupported role `$name` for upstream ${upstream.id}")
|
|
}
|
|
}
|
|
}
|
|
|
|
private fun tryReadMethods(upNode: MappingNode): UpstreamsConfig.Methods? {
|
|
return getMapping(upNode, "methods")?.let { mnode ->
|
|
val enabled = getList<MappingNode>(mnode, "enabled")?.value?.mapNotNull { m ->
|
|
getValueAsString(m, "name")?.let { name ->
|
|
UpstreamsConfig.Method(
|
|
name = name,
|
|
quorum = getValueAsString(m, "quorum"),
|
|
static = getValueAsString(m, "static"),
|
|
)
|
|
}
|
|
}?.toSet() ?: emptySet()
|
|
val disabled = getList<MappingNode>(mnode, "disabled")?.value?.mapNotNull { m ->
|
|
getValueAsString(m, "name")?.let { name ->
|
|
UpstreamsConfig.Method(
|
|
name = name,
|
|
)
|
|
}
|
|
}?.toSet() ?: emptySet()
|
|
|
|
UpstreamsConfig.Methods(
|
|
enabled,
|
|
disabled,
|
|
)
|
|
}
|
|
}
|
|
|
|
private fun tryReadMethodGroups(upNode: MappingNode): UpstreamsConfig.MethodGroups? {
|
|
return getMapping(upNode, "method-groups")?.let {
|
|
UpstreamsConfig.MethodGroups(
|
|
enabled = getListOfString(it, "enabled")?.toSet() ?: emptySet(),
|
|
disabled = getListOfString(it, "disabled")?.toSet() ?: emptySet(),
|
|
)
|
|
}
|
|
}
|
|
}
|