Update health chains (#740)

This commit is contained in:
KirillPamPam
2025-11-05 13:06:05 +04:00
committed by GitHub
parent 35fa52a01e
commit 63d61699b2
13 changed files with 372 additions and 154 deletions

View File

@@ -16,8 +16,13 @@
package io.emeraldpay.dshackle.config
import io.emeraldpay.dshackle.Chain
import java.util.concurrent.atomic.AtomicReference
class HealthConfig {
class HealthConfig() {
constructor(newChains: Map<Chain, ChainConfig>) : this() {
updateChains(newChains)
}
companion object {
fun default(): HealthConfig {
@@ -28,14 +33,28 @@ class HealthConfig {
var port: Int = 8082
var host: String = "127.0.0.1"
var path: String = "/health"
val chains = HashMap<Chain, ChainConfig>()
private val chains = AtomicReference<Map<Chain, ChainConfig>>(HashMap<Chain, ChainConfig>())
fun isEnabled(): Boolean {
return chains.isNotEmpty()
return chains.get().isNotEmpty()
}
fun loadChains(): Map<Chain, ChainConfig> = chains.get()
fun configs(): Collection<ChainConfig> {
return chains.values
return chains.get().values
}
fun containsChain(chain: Chain): Boolean {
return chains.get().containsKey(chain)
}
fun config(chain: Chain): ChainConfig? {
return chains.get()[chain]
}
fun updateChains(newChains: Map<Chain, ChainConfig>) {
chains.set(newChains)
}
data class ChainConfig(

View File

@@ -54,6 +54,7 @@ class HealthConfigReader : YamlConfigReader<HealthConfig>() {
if (input == null) {
return
}
val configs = HashMap<Chain, HealthConfig.ChainConfig>()
input.value.forEach { conf ->
val chain = getValueAsString(conf, "chain")
?.let { Global.chainById(it) }
@@ -64,11 +65,12 @@ class HealthConfigReader : YamlConfigReader<HealthConfig>() {
if (chain == Chain.UNSPECIFIED) {
log.warn("Using UNSPECIFIED blockchain for Health Check. Always fails")
}
if (healthConfig.chains.containsKey(chain)) {
if (healthConfig.containsChain(chain)) {
log.warn("Duplicate Health Check config for $chain. Replace previous with new")
}
val minAvailable = getValueAsInt(conf, "min-available") ?: 1
healthConfig.chains[chain] = HealthConfig.ChainConfig(chain, minAvailable.coerceAtLeast(0))
configs[chain] = HealthConfig.ChainConfig(chain, minAvailable.coerceAtLeast(0))
}
healthConfig.updateChains(configs)
}
}

View File

@@ -0,0 +1,25 @@
package io.emeraldpay.dshackle.config.reload
import org.springframework.stereotype.Component
@Component
class HealthReloadConfigProcessor(
private val reloadConfigService: ReloadConfigService,
) : ReloadConfigProcessor {
override fun reload(): Boolean {
val currentCfg = reloadConfigService.currentHealthConfig()
val newCfg = reloadConfigService.readHealthConfig()
if (currentCfg.configs().toSet() == newCfg.configs().toSet()) {
return false
}
reloadConfigService.updateHealthChains(newCfg.loadChains())
return true
}
override fun configType(): String {
return "health config"
}
}

View File

@@ -0,0 +1,131 @@
package io.emeraldpay.dshackle.config.reload
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global.Companion.chainById
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.foundation.ChainOptions
import org.springframework.stereotype.Component
import java.util.stream.Collectors
interface ReloadConfigProcessor {
fun reload(): Boolean
fun configType(): String
}
@Component
class UpstreamConfigReloadConfigProcessor(
private val reloadConfigService: ReloadConfigService,
private val reloadConfigUpstreamService: ReloadConfigUpstreamService,
) : ReloadConfigProcessor {
override fun reload(): Boolean {
val newUpstreamsConfig = reloadConfigService.readUpstreamsConfig()
val currentUpstreamsConfig = reloadConfigService.currentUpstreamsConfig()
if (newUpstreamsConfig == currentUpstreamsConfig) {
return false
}
val chainsToReload = analyzeDefaultOptions(
currentUpstreamsConfig.defaultOptions,
newUpstreamsConfig.defaultOptions,
)
val upstreamsAnalyzeData = analyzeUpstreams(
currentUpstreamsConfig.upstreams,
newUpstreamsConfig.upstreams,
)
val upstreamsToRemove = upstreamsAnalyzeData.removed
.filterNot { chainsToReload.contains(it.second) }
.toSet()
val upstreamsToAdd = upstreamsAnalyzeData.added
reloadConfigService.updateUpstreamsConfig(newUpstreamsConfig)
reloadConfigUpstreamService.reloadUpstreams(chainsToReload, upstreamsToRemove, upstreamsToAdd, newUpstreamsConfig)
return true
}
override fun configType(): String {
return "upstream config"
}
private fun analyzeUpstreams(
currentUpstreams: List<UpstreamsConfig.Upstream<*>>,
newUpstreams: List<UpstreamsConfig.Upstream<*>>,
): UpstreamAnalyzeData {
if (currentUpstreams == newUpstreams) {
return UpstreamAnalyzeData()
}
val reloaded = mutableSetOf<Pair<String, Chain>>()
val removed = mutableSetOf<Pair<String, Chain>>()
val currentUpstreamsMap = currentUpstreams.associateBy { it.id!! to chainById(it.chain) }
val newUpstreamsMap = newUpstreams.associateBy { it.id!! to chainById(it.chain) }
currentUpstreamsMap.forEach {
val newUpstream = newUpstreamsMap[it.key]
if (newUpstream == null) {
removed.add(it.key)
} else if (newUpstream != it.value) {
reloaded.add(it.key)
}
}
val added = newUpstreamsMap
.minus(currentUpstreamsMap.keys)
.mapTo(mutableSetOf()) { it.key }
.plus(reloaded)
return UpstreamAnalyzeData(added, removed.plus(reloaded))
}
private fun analyzeDefaultOptions(
currentDefaultOptions: List<ChainOptions.DefaultOptions>,
newDefaultOptions: List<ChainOptions.DefaultOptions>,
): Set<Chain> {
val chainsToReload = mutableSetOf<Chain>()
val currentOptions = getChainOptions(currentDefaultOptions)
val newOptions = getChainOptions(newDefaultOptions)
if (currentOptions == newOptions) {
return emptySet()
}
val removed = mutableSetOf<Chain>()
currentOptions.forEach {
val newChainOption = newOptions[it.key]
if (newChainOption == null) {
removed.add(chainById(it.key))
} else if (newChainOption != it.value) {
chainsToReload.add(chainById(it.key))
}
}
val added = newOptions.minus(currentOptions.keys).map { chainById(it.key) }
return chainsToReload.plus(added).plus(removed)
}
private fun getChainOptions(
defaultOptions: List<ChainOptions.DefaultOptions>,
): Map<String, List<ChainOptions.PartialOptions>> {
return defaultOptions.stream()
.flatMap { options -> options.chains?.stream()?.map { it to options.options } }
.collect(
Collectors.groupingBy(
{ it.first },
Collectors.mapping(
{ it.second },
Collectors.toUnmodifiableList(),
),
),
)
}
private data class UpstreamAnalyzeData(
val added: Set<Pair<String, Chain>> = emptySet(),
val removed: Set<Pair<String, Chain>> = emptySet(),
)
}

View File

@@ -1,7 +1,10 @@
package io.emeraldpay.dshackle.config.reload
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Config
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.config.HealthConfig
import io.emeraldpay.dshackle.config.HealthConfigReader
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.config.UpstreamsConfigReader
@@ -17,12 +20,21 @@ class ReloadConfigService(
) {
private val optionsReader = ChainOptionsReader()
private val upstreamsConfigReader = UpstreamsConfigReader(fileResolver, optionsReader)
private val healthConfigReader = HealthConfigReader()
fun readUpstreamsConfig() = upstreamsConfigReader.read(config.getConfigPath().inputStream())!!
fun currentUpstreamsConfig() = mainConfig.initialConfig!!
fun currentHealthConfig() = mainConfig.health
fun readHealthConfig() = healthConfigReader.read(config.getConfigPath().inputStream())!!
fun updateUpstreamsConfig(newConfig: UpstreamsConfig) {
mainConfig.upstreams = newConfig
}
fun updateHealthChains(newChains: Map<Chain, HealthConfig.ChainConfig>) {
mainConfig.health.updateChains(newChains)
}
}

View File

@@ -1,20 +1,14 @@
package io.emeraldpay.dshackle.config.reload
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Global.Companion.chainById
import io.emeraldpay.dshackle.config.UpstreamsConfig
import io.emeraldpay.dshackle.foundation.ChainOptions
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component
import sun.misc.Signal
import sun.misc.SignalHandler
import java.util.concurrent.locks.ReentrantLock
import java.util.stream.Collectors
@Component
class ReloadConfigSetup(
private val reloadConfigService: ReloadConfigService,
private val reloadConfigUpstreamService: ReloadConfigUpstreamService,
private val processors: List<ReloadConfigProcessor>,
) : SignalHandler {
companion object {
@@ -43,124 +37,24 @@ class ReloadConfigSetup(
try {
log.info("Reloading config...")
if (reloadConfig()) {
log.info("Config is reloaded")
} else {
log.info("There is nothing to reload, config is the same")
if (processors.isEmpty()) {
log.warn("No reload config processors")
return
}
processors.forEach {
if (it.reload()) {
log.info("{} is reloaded", it.configType())
} else {
log.info("There is nothing to reload, {} is the same", it.configType())
}
}
} finally {
log.info("Reloading config has been completed")
reloadLock.unlock()
}
} else {
log.warn("Reloading is in progress")
}
}
private fun reloadConfig(): Boolean {
val newUpstreamsConfig = reloadConfigService.readUpstreamsConfig()
val currentUpstreamsConfig = reloadConfigService.currentUpstreamsConfig()
if (newUpstreamsConfig == currentUpstreamsConfig) {
return false
}
val chainsToReload = analyzeDefaultOptions(
currentUpstreamsConfig.defaultOptions,
newUpstreamsConfig.defaultOptions,
)
val upstreamsAnalyzeData = analyzeUpstreams(
currentUpstreamsConfig.upstreams,
newUpstreamsConfig.upstreams,
)
val upstreamsToRemove = upstreamsAnalyzeData.removed
.filterNot { chainsToReload.contains(it.second) }
.toSet()
val upstreamsToAdd = upstreamsAnalyzeData.added
reloadConfigService.updateUpstreamsConfig(newUpstreamsConfig)
reloadConfigUpstreamService.reloadUpstreams(chainsToReload, upstreamsToRemove, upstreamsToAdd, newUpstreamsConfig)
return true
}
private fun analyzeUpstreams(
currentUpstreams: List<UpstreamsConfig.Upstream<*>>,
newUpstreams: List<UpstreamsConfig.Upstream<*>>,
): UpstreamAnalyzeData {
if (currentUpstreams == newUpstreams) {
return UpstreamAnalyzeData()
}
val reloaded = mutableSetOf<Pair<String, Chain>>()
val removed = mutableSetOf<Pair<String, Chain>>()
val currentUpstreamsMap = currentUpstreams.associateBy { it.id!! to chainById(it.chain) }
val newUpstreamsMap = newUpstreams.associateBy { it.id!! to chainById(it.chain) }
currentUpstreamsMap.forEach {
val newUpstream = newUpstreamsMap[it.key]
if (newUpstream == null) {
removed.add(it.key)
} else if (newUpstream != it.value) {
reloaded.add(it.key)
}
}
val added = newUpstreamsMap
.minus(currentUpstreamsMap.keys)
.mapTo(mutableSetOf()) { it.key }
.plus(reloaded)
return UpstreamAnalyzeData(added, removed.plus(reloaded))
}
private fun analyzeDefaultOptions(
currentDefaultOptions: List<ChainOptions.DefaultOptions>,
newDefaultOptions: List<ChainOptions.DefaultOptions>,
): Set<Chain> {
val chainsToReload = mutableSetOf<Chain>()
val currentOptions = getChainOptions(currentDefaultOptions)
val newOptions = getChainOptions(newDefaultOptions)
if (currentOptions == newOptions) {
return emptySet()
}
val removed = mutableSetOf<Chain>()
currentOptions.forEach {
val newChainOption = newOptions[it.key]
if (newChainOption == null) {
removed.add(chainById(it.key))
} else if (newChainOption != it.value) {
chainsToReload.add(chainById(it.key))
}
}
val added = newOptions.minus(currentOptions.keys).map { chainById(it.key) }
return chainsToReload.plus(added).plus(removed)
}
private fun getChainOptions(
defaultOptions: List<ChainOptions.DefaultOptions>,
): Map<String, List<ChainOptions.PartialOptions>> {
return defaultOptions.stream()
.flatMap { options -> options.chains?.stream()?.map { it to options.options } }
.collect(
Collectors.groupingBy(
{ it.first },
Collectors.mapping(
{ it.second },
Collectors.toUnmodifiableList(),
),
),
)
}
private data class UpstreamAnalyzeData(
val added: Set<Pair<String, Chain>> = emptySet(),
val removed: Set<Pair<String, Chain>> = emptySet(),
)
}

View File

@@ -102,7 +102,7 @@ class HealthCheckSetup(
val details = chains.flatMap { chain ->
var chainUnavailable = false
val up = multistreamHolder.getUpstream(chain)
val required = healthConfig.chains[chain]
val required = healthConfig.config(chain)
if (!up.isAvailable()) {
if (required != null) {
anyUnavailable = true

View File

@@ -26,11 +26,11 @@ class HealthCheckSetupSpec extends Specification {
def "OK when meets availability - 1"() {
setup:
def config = new HealthConfig().tap {
it.chains[Chain.ETHEREUM__MAINNET] = new HealthConfig.ChainConfig(
Chain.ETHEREUM__MAINNET, 1
)
}
def config = new HealthConfig(
Collections.singletonMap(
Chain.ETHEREUM__MAINNET, new HealthConfig.ChainConfig(Chain.ETHEREUM__MAINNET, 1)
)
)
def up1 = Mock(Upstream)
def ethereumUpstreams = Mock(Multistream)
def multistream = Mock(MultistreamHolder)
@@ -50,11 +50,11 @@ class HealthCheckSetupSpec extends Specification {
def "OK when meets availability - 1 - bitcoin"() {
setup:
def config = new HealthConfig().tap {
it.chains[Chain.BITCOIN__MAINNET] = new HealthConfig.ChainConfig(
Chain.BITCOIN__MAINNET, 1
)
}
def config = new HealthConfig(
Collections.singletonMap(
Chain.BITCOIN__MAINNET, new HealthConfig.ChainConfig(Chain.BITCOIN__MAINNET, 1)
)
)
def up1 = Mock(Upstream)
def bitcoinUpstreams = Mock(Multistream)
def multistream = Mock(MultistreamHolder)
@@ -74,11 +74,11 @@ class HealthCheckSetupSpec extends Specification {
def "OK when meets availability - 2/3"() {
setup:
def config = new HealthConfig().tap {
it.chains[Chain.ETHEREUM__MAINNET] = new HealthConfig.ChainConfig(
Chain.ETHEREUM__MAINNET, 2
)
}
def config = new HealthConfig(
Collections.singletonMap(
Chain.ETHEREUM__MAINNET, new HealthConfig.ChainConfig(Chain.ETHEREUM__MAINNET, 1)
)
)
def up1 = Mock(Upstream)
def up2 = Mock(Upstream)
def up3 = Mock(Upstream)
@@ -102,11 +102,11 @@ class HealthCheckSetupSpec extends Specification {
def "OK when doesn't meet availability - 2/3"() {
setup:
def config = new HealthConfig().tap {
it.chains[Chain.ETHEREUM__MAINNET] = new HealthConfig.ChainConfig(
Chain.ETHEREUM__MAINNET, 2
)
}
def config = new HealthConfig(
Collections.singletonMap(
Chain.ETHEREUM__MAINNET, new HealthConfig.ChainConfig(Chain.ETHEREUM__MAINNET, 2)
)
)
def up1 = Mock(Upstream)
def up2 = Mock(Upstream)
def up3 = Mock(Upstream)
@@ -130,11 +130,11 @@ class HealthCheckSetupSpec extends Specification {
def "OK when meets availability - 2/3 - detailed"() {
setup:
def config = new HealthConfig().tap {
it.chains[Chain.ETHEREUM__MAINNET] = new HealthConfig.ChainConfig(
Chain.ETHEREUM__MAINNET, 2
)
}
def config = new HealthConfig(
Collections.singletonMap(
Chain.ETHEREUM__MAINNET, new HealthConfig.ChainConfig(Chain.ETHEREUM__MAINNET, 1)
)
)
def up1 = Mock(Upstream)
def up2 = Mock(Upstream)
def up3 = Mock(Upstream)

View File

@@ -0,0 +1,93 @@
package io.emeraldpay.dshackle.config.reload
import io.emeraldpay.dshackle.Chain
import io.emeraldpay.dshackle.Config
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.config.HealthConfig
import io.emeraldpay.dshackle.config.HealthConfigReader
import io.emeraldpay.dshackle.config.MainConfig
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test
import org.mockito.kotlin.mock
import org.mockito.kotlin.whenever
import org.springframework.util.ResourceUtils
import java.io.File
class HealthReloadConfigProcessorTest {
private val fileResolver = FileResolver(File(""))
private val mainConfig = MainConfig()
private val config = mock<Config>()
private val reloadConfigService = ReloadConfigService(config, fileResolver, mainConfig)
private val processor = HealthReloadConfigProcessor(reloadConfigService)
private val healthConfigReader = HealthConfigReader()
@BeforeEach
fun setupTests() {
mainConfig.health = HealthConfig.default()
}
@Test
fun `get health processor config type`() {
assertThat(processor.configType()).isEqualTo("health config")
}
@Test
fun `cant reload config if they are equal`() {
val newConfigFile = ResourceUtils.getFile("classpath:configs/health-initial.yaml")
whenever(config.getConfigPath()).thenReturn(newConfigFile)
val initialConfigIs = ResourceUtils.getFile("classpath:configs/health-initial.yaml").inputStream()
val initialConfig = healthConfigReader.read(initialConfigIs)!!
mainConfig.health = initialConfig
val result = processor.reload()
assertThat(result).isFalse
}
@Test
fun `cant reload config if they everything is different but not blockchain list`() {
val newConfigFile = ResourceUtils.getFile("classpath:configs/health-changed-params.yaml")
whenever(config.getConfigPath()).thenReturn(newConfigFile)
val initialConfigIs = ResourceUtils.getFile("classpath:configs/health-initial.yaml").inputStream()
val initialConfig = healthConfigReader.read(initialConfigIs)!!
mainConfig.health = initialConfig
val result = processor.reload()
assertThat(result).isFalse
}
@Test
fun `reload health config`() {
val newConfigFile = ResourceUtils.getFile("classpath:configs/health-changed.yaml")
whenever(config.getConfigPath()).thenReturn(newConfigFile)
val initialConfigIs = ResourceUtils.getFile("classpath:configs/health-initial.yaml").inputStream()
val initialConfig = healthConfigReader.read(initialConfigIs)!!
mainConfig.health = initialConfig
assertThat(mainConfig.health.configs().toSet()).isEqualTo(
setOf(
HealthConfig.ChainConfig(Chain.BSC__MAINNET, 0),
),
)
val reloaded = processor.reload()
val newChains = mainConfig.health.configs()
assertThat(reloaded).isTrue
assertThat(newChains.toSet()).isEqualTo(
setOf(
HealthConfig.ChainConfig(Chain.ARBITRUM__MAINNET, 5),
HealthConfig.ChainConfig(Chain.OPTIMISM__MAINNET, 1),
HealthConfig.ChainConfig(Chain.BSC__MAINNET, 1),
),
)
}
}

View File

@@ -53,6 +53,20 @@ class ReloadConfigTest {
mainConfig.upstreams = null
}
@Test
fun `reload config and use all processors`() {
val firstProcessor = mock<ReloadConfigProcessor>()
val secondProcessor = mock<ReloadConfigProcessor>()
val reloadConfig = ReloadConfigSetup(listOf(firstProcessor, secondProcessor))
reloadConfig.handle(Signal("HUP"))
verify(firstProcessor).reload()
verify(firstProcessor).configType()
verify(secondProcessor).reload()
verify(secondProcessor).configType()
}
@Test
fun `reload upstreams changes`() {
val up1 = upstream("local1")
@@ -76,7 +90,8 @@ class ReloadConfigTest {
currentMultistreamHolder,
configuredUpstreams,
)
val reloadConfig = ReloadConfigSetup(reloadConfigService, reloadConfigUpstreamService)
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(reloadConfigService, reloadConfigUpstreamService)
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
val initialConfigIs = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml").inputStream()
val initialConfig = upstreamsConfigReader.read(initialConfigIs)!!
@@ -127,7 +142,8 @@ class ReloadConfigTest {
currentMultistreamHolder,
configuredUpstreams,
)
val reloadConfig = ReloadConfigSetup(reloadConfigService, reloadConfigUpstreamService)
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(reloadConfigService, reloadConfigUpstreamService)
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
val initialConfigIs = ResourceUtils.getFile("classpath:configs/upstreams-initial.yaml").inputStream()
val initialConfig = upstreamsConfigReader.read(initialConfigIs)!!
val newConfig = upstreamsConfigReader.read(newConfigFile.inputStream())!!
@@ -157,7 +173,8 @@ class ReloadConfigTest {
val reloadConfigUpstreamService = mock<ReloadConfigUpstreamService>()
val reloadConfig = ReloadConfigSetup(reloadConfigService, reloadConfigUpstreamService)
val upstreamCfgReloadProcessor = UpstreamConfigReloadConfigProcessor(reloadConfigService, reloadConfigUpstreamService)
val reloadConfig = ReloadConfigSetup(listOf(upstreamCfgReloadProcessor))
whenever(config.getConfigPath()).thenReturn(initialConfigFile)

View File

@@ -0,0 +1,7 @@
health:
port: 8081
host: 1.0.0.0
path: /healtha
blockchains:
- chain: bsc
min-available: 0

View File

@@ -0,0 +1,11 @@
health:
port: 8082
host: 0.0.0.0
path: /health
blockchains:
- chain: arbitrum
min-available: 5
- chain: bsc
min-available: 1
- chain: optimism
min-available: 1

View File

@@ -0,0 +1,7 @@
health:
port: 8082
host: 0.0.0.0
path: /health
blockchains:
- chain: bsc
min-available: 0