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

Fix Prometheus metrics handler hanging on slow scrapes (#733)

Browse Source 
* Fix Prometheus metrics handler hanging on slow scrapes
This commit is contained in:
Artem Rootman
2025-10-13 12:17:41 +03:00
committed by GitHub
parent fba0b139c3
commit 8bc76328f6
1 changed files with 44 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/main/kotlin/io/emeraldpay/dshackle/monitoring/MonitoringSetup.kt
View File

@@ -35,6 +35,11 @@ import java.io.IOException
import java.net.InetAddress import java.net.InetAddress
import java.net.InetSocketAddress import java.net.InetSocketAddress
import java.net.ServerSocket import java.net.ServerSocket
import java.nio.charset.StandardCharsets
import java.util.concurrent.Executors
import java.util.concurrent.ThreadFactory
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import java.util.zip.GZIPOutputStream import java.util.zip.GZIPOutputStream
import javax.annotation.PostConstruct import javax.annotation.PostConstruct
@@ -93,6 +98,15 @@ class MonitoringSetup(
// prometheus is a single thread periodic call, no reason to setup anything complex // prometheus is a single thread periodic call, no reason to setup anything complex
Thread { Thread {
var started = false var started = false
val scrapeThreadCounter = AtomicInteger(0)
val scrapeExecutor = Executors.newFixedThreadPool(
4,
ThreadFactory { runnable ->
Thread(runnable, "prometheus-scrape-${scrapeThreadCounter.incrementAndGet()}").apply {
isDaemon = true
}
},
)
while (true) { while (true) {
if (isTcpPortAvailable(monitoringConfig.prometheus.host, monitoringConfig.prometheus.port)) { if (isTcpPortAvailable(monitoringConfig.prometheus.host, monitoringConfig.prometheus.port)) {
started = true started = true
@@ -105,15 +119,38 @@ class MonitoringSetup(
), ),
0, 0,
) )
server.executor = scrapeExecutor
server.createContext(monitoringConfig.prometheus.path) { httpExchange -> server.createContext(monitoringConfig.prometheus.path) { httpExchange ->
val response = prometheusRegistry.scrape() val startTime = System.nanoTime()
httpExchange.responseHeaders.add("Content-Encoding", "gzip") try {
httpExchange.responseHeaders.add("Content-Type", "text/plain") val response = prometheusRegistry.scrape()
httpExchange.sendResponseHeaders(200, 0) val responseBytes = response.toByteArray(StandardCharsets.UTF_8)
httpExchange.responseBody.use { os -> httpExchange.responseHeaders.add("Content-Encoding", "gzip")
GZIPOutputStream(os).use { gzos -> httpExchange.responseHeaders.add("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
gzos.write(response.toByteArray()) httpExchange.sendResponseHeaders(200, 0)
httpExchange.responseBody.use { os ->
GZIPOutputStream(os).use { gzos ->
gzos.write(responseBytes)
}
} }
val durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime)
if (durationMs > 5000) {
log.warn("Prometheus scrape served in {} ms", durationMs)
} else {
log.debug("Prometheus scrape served in {} ms", durationMs)
}
} catch (e: Exception) {
log.error("Failed to serve Prometheus metrics", e)
val messageBytes = "internal error".toByteArray(StandardCharsets.UTF_8)
runCatching {
httpExchange.responseHeaders.add("Content-Type", "text/plain; charset=utf-8")
httpExchange.sendResponseHeaders(500, messageBytes.size.toLong())
httpExchange.responseBody.use { os ->
os.write(messageBytes)
}
}
} finally {
httpExchange.close()
} }
} }
Thread(server::start).start() Thread(server::start).start()