Merge pull request #85 from p2p-org/fix_daemon_heads

fix daemon heads + head metrics

src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt

@@ -28,20 +28,26 @@ import reactor.core.publisher.Sinks.EmitResult.FAIL_ZERO_SUBSCRIBER
 import reactor.core.publisher.Sinks.EmitResult.OK
 import reactor.core.scheduler.Schedulers
 import reactor.kotlin.core.publisher.toMono
+import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.Executors
+import java.util.concurrent.Future
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.locks.ReentrantLock
 
 abstract class AbstractHead @JvmOverloads constructor(
     private val forkChoice: ForkChoice,
     private val blockValidator: BlockValidator = BlockValidator.ALWAYS_VALID,
-    awaitHeadTimeoutMs: Long = 60_000,
+    private val awaitHeadTimeoutMs: Long = 60_000,
     private val upstreamId: String = ""
 ) : Head {
 
     companion object {
         private val log = LoggerFactory.getLogger(AbstractHead::class.java)
+        private val instances: MutableMap<String, AtomicInteger> = ConcurrentHashMap()
+        private val running: MutableMap<String, AtomicInteger> = ConcurrentHashMap()
+        private val executor = Executors.newSingleThreadScheduledExecutor()
     }
 
     private var stream = Sinks.many().multicast().directBestEffort<BlockContainer>()
@@ -50,25 +56,27 @@ abstract class AbstractHead @JvmOverloads constructor(
     private var stopping = false
     private var lastHeadUpdated = 0L
     private val lock = ReentrantLock()
+    private var future: Future<*>? = null
+    private val delayed = AtomicBoolean(false)
 
     init {
-        val state = AtomicBoolean(false)
+
-        Gauge.builder("stuck_head", state) {
+        val className = this.javaClass.simpleName
+
+        Gauge.builder("stuck_head", delayed) {
             if (it.get()) 1.0 else 0.0
-        }.tag("upstream", upstreamId).tag("class", this.javaClass.simpleName).register(Metrics.globalRegistry)
+        }.tag("upstream", upstreamId).tag("class", className).register(Metrics.globalRegistry)
         Gauge.builder("current_head", forkChoice) {
             it.getHead()?.height?.toDouble() ?: 0.0
-        }.tag("upstream", upstreamId).tag("class", this.javaClass.simpleName).register(Metrics.globalRegistry)
+        }.tag("upstream", upstreamId).tag("class", className).register(Metrics.globalRegistry)
-        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
+
-            {
+        instances.computeIfAbsent(className) {
-                val delay = System.currentTimeMillis() - lastHeadUpdated
+            AtomicInteger(0).also { toHeadCountMetric(it, "allocated") }
-                val delayed = delay > awaitHeadTimeoutMs
+        }.incrementAndGet()
-                state.set(delayed)
+    }
-                if (delayed) {
+
-                    log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass}")
+    protected fun finalize() {
-                }
+        instances[this.javaClass.simpleName]?.decrementAndGet()
-            }, 300, 30, TimeUnit.SECONDS
-        )
     }
 
     fun follow(source: Flux<BlockContainer>): Disposable {
@@ -154,9 +162,37 @@ abstract class AbstractHead @JvmOverloads constructor(
 
     override fun stop() {
         stopping = true
+        log.debug("Stop ${this.javaClass.simpleName} $upstreamId")
+        future?.let {
+            running[this.javaClass.simpleName]?.decrementAndGet()
+            it.cancel(true)
+        }
+        future = null
     }
 
     override fun start() {
         stopping = false
+        log.debug("Start ${this.javaClass.simpleName} $upstreamId")
+        if (future == null) {
+            future = executor.scheduleAtFixedRate(
+                {
+                    val delay = System.currentTimeMillis() - lastHeadUpdated
+                    delayed.set(delay > awaitHeadTimeoutMs)
+                    if (delayed.get()) {
+                        log.warn("No head updates $upstreamId for $delay ms @ ${this.javaClass.simpleName}")
+                    }
+                }, 180, 30, TimeUnit.SECONDS
+            )
+
+            running.computeIfAbsent(this.javaClass.simpleName) {
+                AtomicInteger(0).also { toHeadCountMetric(it, "running") }
+            }.incrementAndGet()
+        }
+    }
+
+    private fun toHeadCountMetric(counter: AtomicInteger, status: String) {
+        Gauge.builder("head_count", counter) {
+            it.get().toDouble()
+        }.tag("class", this.javaClass.simpleName).tag("status", status).register(Metrics.globalRegistry)
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt

@@ -112,33 +112,33 @@ open class EthereumPosMultiStream(
     }
 
     override fun updateHead(): Head {
-        head?.let {
+        this.head?.takeIf { it is Lifecycle }?.apply { stop() }
-            if (it is Lifecycle) {
-                it.stop()
-            }
-        }
         lagObserver?.stop()
         lagObserver = null
-        val head = if (upstreams.size == 1) {
+
-            val upstream = upstreams.first()
+        return when (upstreams.size) {
-            upstream.setLag(0)
+            0 -> EmptyHead()
-            upstream.getHead().apply {
+            1 -> upstreams.first().let {
-                if (this is Lifecycle) {
+                it.setLag(0)
-                    this.start()
+                it.getHead().apply {
+                    if (this is Lifecycle) {
+                        start()
+                    }
                 }
             }
-        } else {
+
-            val heads = upstreams.map { it.getHead() }
+            else -> upstreams.map { it.getHead() }.let { heads ->
-            val newHead = MergedHead(heads, PriorityForkChoice(), "ETH Pos Multistream").apply {
+                MergedHead(heads, PriorityForkChoice(), "ETH Pos Multistream").apply {
-                this.start()
+                    start()
+                }.also {
+                    this.lagObserver = EthereumPosHeadLagObserver(it, ArrayList(upstreams)).apply {
+                        start()
+                    }
+                }
             }
-            val lagObserver = EthereumPosHeadLagObserver(newHead, ArrayList(upstreams))
+        }.apply {
-            this.lagObserver = lagObserver
+            onHeadUpdated(this)
-            lagObserver.start()
-            newHead
         }
-        onHeadUpdated(head)
-        return head
     }
 
     override fun getLabels(): Collection<UpstreamsConfig.Labels> {