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

add detailed upstream availability metrics (#132)

Browse Source 
* add detailed upstream availability metrics

* fix review issues
This commit is contained in:
Vadim Vlasov
2023-02-09 13:28:44 +08:00
committed by GitHub
parent 9948b6a781
commit fc36e63ea6
1 changed files with 24 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -40,7 +40,6 @@ import reactor.core.publisher.Mono
import reactor.core.publisher.Sinks import reactor.core.publisher.Sinks
import java.time.Duration import java.time.Duration
import java.time.Instant import java.time.Instant
import java.util.Locale
import java.util.concurrent.atomic.AtomicReference import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantLock
import java.util.function.Predicate import java.util.function.Predicate
@@ -74,7 +73,7 @@ abstract class Multistream(
private var subscription: Disposable? = null private var subscription: Disposable? = null
private var capabilities: Set<Capability> = emptySet() private var capabilities: Set<Capability> = emptySet()
private val removed: MutableMap<String, Upstream> = HashMap() private val removed: MutableMap<String, Upstream> = HashMap()
private val meters: MutableMap<String, Meter.Id> = HashMap() private val meters: MutableMap<String, List<Meter.Id>> = HashMap()
private val addedUpstreams = Sinks.many() private val addedUpstreams = Sinks.many()
.multicast() .multicast()
.directBestEffort<Upstream>() .directBestEffort<Upstream>()
@@ -86,7 +85,7 @@ abstract class Multistream(
UpstreamAvailability.values().forEach { status -> UpstreamAvailability.values().forEach { status ->
Metrics.gauge( Metrics.gauge(
"$metrics.availability", "$metrics.availability",
listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.lowercase(Locale.getDefault()))), listOf(Tag.of("chain", chain.chainCode), Tag.of("status", status.name.lowercase())),
this this
) { ) {
upstreams.count { it.getStatus() == status }.toDouble() upstreams.count { it.getStatus() == status }.toDouble()
@@ -105,18 +104,30 @@ abstract class Multistream(
} }
} }
private fun monitorUpstream(upstream: Upstream) { private fun removeUpstreamMeters(upstreamId: String) {
val id = upstream.getId() meters[upstreamId]?.forEach {
// remove gouge for given upstream if exists - otherwise metric will stuck with prev upstream instance
meters[id]?.let {
Metrics.globalRegistry.remove(it) Metrics.globalRegistry.remove(it)
} }
}
meters[id] = Gauge.builder("$metrics.lag", upstream) { it.getLag().toDouble() } private fun monitorUpstream(upstream: Upstream) {
.tag("chain", chain.chainCode) val upstreamId = upstream.getId()
.tag("upstream", id)
.register(Metrics.globalRegistry) // otherwise metric will stuck with prev upstream instance
.id removeUpstreamMeters(upstreamId)
meters[upstreamId] = listOf(
Gauge.builder("$metrics.lag", upstream) { it.getLag().toDouble() }
.tag("chain", chain.chainCode)
.tag("upstream", upstreamId)
.register(Metrics.globalRegistry)
.id,
Gauge.builder("$metrics.availability.status", upstream) { it.getStatus().grpcId.toDouble() }
.tag("chain", chain.chainCode)
.tag("upstream", upstreamId)
.register(Metrics.globalRegistry)
.id,
)
} }
open fun init() { open fun init() {
@@ -157,6 +168,7 @@ abstract class Multistream(
if (it) { if (it) {
removeHead(id) removeHead(id)
onUpstreamsUpdated() onUpstreamsUpdated()
removeUpstreamMeters(id)
} }
} }