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

added metric

Browse Source
This commit is contained in:
Maksim Fomenkov
2022-12-13 12:02:12 +03:00
parent 8c52a9cd34
commit dd147558bf
1 changed files with 20 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt
View File

@@ -15,8 +15,14 @@
*/ */
package io.emeraldpay.dshackle.upstream package io.emeraldpay.dshackle.upstream
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockContainer
import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
import io.micrometer.core.instrument.Counter
import io.micrometer.core.instrument.Gauge
import io.micrometer.core.instrument.Meter
import io.micrometer.core.instrument.Metrics
import io.micrometer.core.instrument.Tag
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import reactor.core.Disposable import reactor.core.Disposable
import reactor.core.publisher.Flux import reactor.core.publisher.Flux
@@ -28,6 +34,7 @@ import reactor.core.scheduler.Schedulers
import reactor.kotlin.core.publisher.toMono import reactor.kotlin.core.publisher.toMono
import java.util.concurrent.Executors import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantLock import java.util.concurrent.locks.ReentrantLock
abstract class AbstractHead @JvmOverloads constructor( abstract class AbstractHead @JvmOverloads constructor(
@@ -48,19 +55,23 @@ abstract class AbstractHead @JvmOverloads constructor(
private var lastHeadUpdated = 0L private var lastHeadUpdated = 0L
private val lock = ReentrantLock() private val lock = ReentrantLock()
init { init {
val state = AtomicBoolean(false)
Gauge.builder("stuck_head", state) {
if (it.get()) 1.0 else 0.0
}
.tag("upstream", upstreamId)
.tag("class", this.javaClass.simpleName)
.register(Metrics.globalRegistry)
Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
{ {
val delay = System.currentTimeMillis() - lastHeadUpdated val delay = System.currentTimeMillis() - lastHeadUpdated
if (delay > awaitHeadTimeoutMs) { val delayed = delay > awaitHeadTimeoutMs
log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass} - restart") state.set(delayed)
if (lock.tryLock()) { if (delayed) {
try { log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass}")
start()
} finally {
lock.unlock()
}
}
} }
}, 300, 30, TimeUnit.SECONDS }, 300, 30, TimeUnit.SECONDS
) )