Superuser: Created page with "== bitcoin-monitor.py == # bitcoind-monitor.py # # An exporter for Prometheus and Bitcoin Core. # # Copyright 2018 Kevin M. Gallagher # Copyright 2019,2020 Jeff Stein # # Published at https://github.com/jvstein/bitcoin-prometheus-exporter # Licensed under BSD 3-clause (see LICENSE). # # Dependency licenses (retrieved 2020-05-31): # prometheus_client: Apache 2.0 # python-bitcoinlib: LGPLv3 # riprova: MIT import json import logging import time im..."

2023-02-10T15:49:27Z

Created page with "== bitcoin-monitor.py == # bitcoind-monitor.py # # An exporter for Prometheus and Bitcoin Core. # # Copyright 2018 Kevin M. Gallagher # Copyright 2019,2020 Jeff Stein # # Published at https://github.com/jvstein/bitcoin-prometheus-exporter # Licensed under BSD 3-clause (see LICENSE). # # Dependency licenses (retrieved 2020-05-31): # prometheus_client: Apache 2.0 # python-bitcoinlib: LGPLv3 # riprova: MIT import json import logging import time im..."

New page

== bitcoin-monitor.py ==

# bitcoind-monitor.py
#
# An exporter for Prometheus and Bitcoin Core.
#
# Copyright 2018 Kevin M. Gallagher
# Copyright 2019,2020 Jeff Stein
#
# Published at https://github.com/jvstein/bitcoin-prometheus-exporter
# Licensed under BSD 3-clause (see LICENSE).
#
# Dependency licenses (retrieved 2020-05-31):
# prometheus_client: Apache 2.0
# python-bitcoinlib: LGPLv3
# riprova: MIT

import json
import logging
import time
import os
import signal
import sys
import socket

from datetime import datetime
from functools import lru_cache
from typing import Any
from typing import Dict
from typing import List
from typing import Union
from urllib.parse import quote
from wsgiref.simple_server import make_server

import riprova

from bitcoin.rpc import InWarmupError, Proxy, JSONRPCError
from prometheus_client import make_wsgi_app, Gauge, Counter, Info

logger = logging.getLogger("bitcoin-exporter")

# Create Prometheus metrics to track bitcoind stats.
BITCOIN_INFO = Info('bitcoin', 'The chain the daemon runs on')
BITCOIN_BLOCKS = Gauge("bitcoin_blocks", "Block height")
BITCOIN_DIFFICULTY = Gauge("bitcoin_difficulty", "Difficulty")
BITCOIN_PEERS = Gauge("bitcoin_peers", "Number of peers")
BITCOIN_HASHPS_1 = Gauge(
"bitcoin_hashps_1", "Estimated network hash rate per second for the last block"
)
BITCOIN_HASHPS = Gauge(
"bitcoin_hashps", "Estimated network hash rate per second for the last 120 blocks"
)

BITCOIN_WARNINGS = Counter("bitcoin_warnings", "Number of network or blockchain warnings detected")
BITCOIN_UPTIME = Gauge("bitcoin_uptime", "Number of seconds the Bitcoin daemon has been running")

BITCOIN_MEMINFO_USED = Gauge("bitcoin_meminfo_used", "Number of bytes used")
BITCOIN_MEMINFO_FREE = Gauge("bitcoin_meminfo_free", "Number of bytes available")
BITCOIN_MEMINFO_TOTAL = Gauge("bitcoin_meminfo_total", "Number of bytes managed")
BITCOIN_MEMINFO_LOCKED = Gauge("bitcoin_meminfo_locked", "Number of bytes locked")
BITCOIN_MEMINFO_CHUNKS_USED = Gauge("bitcoin_meminfo_chunks_used", "Number of allocated chunks")
BITCOIN_MEMINFO_CHUNKS_FREE = Gauge("bitcoin_meminfo_chunks_free", "Number of unused chunks")

BITCOIN_MEMPOOL_BYTES = Gauge("bitcoin_mempool_bytes", "Size of mempool in bytes")
BITCOIN_MEMPOOL_SIZE = Gauge(
"bitcoin_mempool_size", "Number of unconfirmed transactions in mempool"
)
BITCOIN_MEMPOOL_USAGE = Gauge("bitcoin_mempool_usage", "Total memory usage for the mempool")

BITCOIN_LATEST_BLOCK_HEIGHT = Gauge(
"bitcoin_latest_block_height", "Height or index of latest block"
)
BITCOIN_LATEST_BLOCK_SIZE = Gauge("bitcoin_latest_block_size", "Size of latest block in bytes")
BITCOIN_LATEST_BLOCK_TXS = Gauge(
"bitcoin_latest_block_txs", "Number of transactions in latest block"
)

BITCOIN_TXCOUNT = Gauge("bitcoin_txcount", "Number of TX since the genesis block")

BITCOIN_NUM_CHAINTIPS = Gauge("bitcoin_num_chaintips", "Number of known blockchain branches")

BITCOIN_TOTAL_BYTES_RECV = Gauge("bitcoin_total_bytes_recv", "Total bytes received")
BITCOIN_TOTAL_BYTES_SENT = Gauge("bitcoin_total_bytes_sent", "Total bytes sent")

BITCOIN_LATEST_BLOCK_INPUTS = Gauge(
"bitcoin_latest_block_inputs", "Number of inputs in transactions of latest block"
)
BITCOIN_LATEST_BLOCK_OUTPUTS = Gauge(
"bitcoin_latest_block_outputs", "Number of outputs in transactions of latest block"
)
BITCOIN_LATEST_BLOCK_VALUE = Gauge(
"bitcoin_latest_block_value", "Bitcoin value of all transactions in the latest block"
)

BITCOIN_BAN_CREATED = Gauge(
"bitcoin_ban_created", "Time the ban was created", labelnames=["address", "reason"]
)
BITCOIN_BANNED_UNTIL = Gauge(
"bitcoin_banned_until", "Time the ban expires", labelnames=["address", "reason"]
)

BITCOIN_SERVER_VERSION = Gauge("bitcoin_server_version", "The server version")
BITCOIN_PROTOCOL_VERSION = Gauge("bitcoin_protocol_version", "The protocol version of the server")

BITCOIN_SIZE_ON_DISK = Gauge("bitcoin_size_on_disk", "Estimated size of the block and undo files")

BITCOIN_VERIFICATION_PROGRESS = Gauge(
"bitcoin_verification_progress", "Estimate of verification progress [0..1]"
)

BITCOIN_RPC_ACTIVE = Gauge("bitcoin_rpc_active", "Number of RPC calls being processed")

BITCOIN_DSPROOF_COUNT = Gauge("bitcoin_dsproof_count", "Number of Double Spend Proofs")

EXPORTER_ERRORS = Counter(
"bitcoin_exporter_errors", "Number of errors encountered by the exporter", labelnames=["type"]
)
PROCESS_TIME = Counter(
"bitcoin_exporter_process_time", "Time spent processing metrics from bitcoin node"
)

BITCOIN_RPC_SCHEME = os.environ.get("BITCOIN_RPC_SCHEME", "http")
BITCOIN_RPC_HOST = os.environ.get("BITCOIN_RPC_HOST", "localhost")
BITCOIN_RPC_PORT = os.environ.get("BITCOIN_RPC_PORT", "8332")
BITCOIN_RPC_USER = os.environ.get("BITCOIN_RPC_USER")
BITCOIN_RPC_PASSWORD = os.environ.get("BITCOIN_RPC_PASSWORD")
BITCOIN_CONF_PATH = os.environ.get("BITCOIN_CONF_PATH")
METRICS_ADDR = os.environ.get("METRICS_ADDR", "") # empty = any address
METRICS_PORT = int(os.environ.get("METRICS_PORT", "8334"))
RETRIES = int(os.environ.get("RETRIES", 5))
TIMEOUT = int(os.environ.get("TIMEOUT", 30))
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")

RETRY_EXCEPTIONS = (InWarmupError, ConnectionError, socket.timeout)

RpcResult = Union[Dict[str, Any], List[Any], str, int, float, bool, None]

def on_retry(err: Exception, next_try: float) -> None:
err_type = type(err)
exception_name = err_type.__module__ + "." + err_type.__name__
EXPORTER_ERRORS.labels(**{"type": exception_name}).inc()
logger.error("Retry after exception %s: %s", exception_name, err)

def error_evaluator(e: Exception) -> bool:
return isinstance(e, RETRY_EXCEPTIONS)

@lru_cache(maxsize=1)
def rpc_client_factory():
# Configuration is done in this order of precedence:
# - Explicit config file.
# - BITCOIN_RPC_USER and BITCOIN_RPC_PASSWORD environment variables.
# - Default bitcoin config file (as handled by Proxy.__init__).
use_conf = (
(BITCOIN_CONF_PATH is not None)
or (BITCOIN_RPC_USER is None)
or (BITCOIN_RPC_PASSWORD is None)
)

if use_conf:
logger.info("Using config file: %s", BITCOIN_CONF_PATH or "<default>")
return lambda: Proxy(btc_conf_file=BITCOIN_CONF_PATH, timeout=TIMEOUT)
else:
host = BITCOIN_RPC_HOST
host = "{}:{}@{}".format(quote(BITCOIN_RPC_USER), quote(BITCOIN_RPC_PASSWORD), host)
if BITCOIN_RPC_PORT:
host = "{}:{}".format(host, BITCOIN_RPC_PORT)
service_url = "{}://{}".format(BITCOIN_RPC_SCHEME, host)
logger.info("Using environment configuration")
return lambda: Proxy(service_url=service_url, timeout=TIMEOUT)

def rpc_client():
return rpc_client_factory()()

@riprova.retry(
timeout=TIMEOUT,
backoff=riprova.ExponentialBackOff(),
on_retry=on_retry,
error_evaluator=error_evaluator,
)
def bitcoinrpc(*args) -> RpcResult:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("RPC call: " + " ".join(str(a) for a in args))

result = rpc_client().call(*args)

logger.debug("Result: %s", result)
return result

@lru_cache(maxsize=1)
def getblockstats(block_hash: str):
try:
block = bitcoinrpc("getblockstats", block_hash, ["total_size", "txs", "height", "ins", "outs", "total_out"])
except Exception:
logger.exception("Failed to retrieve block " + block_hash + " statistics from bitcoind.")
return None
return block

def refresh_metrics() -> None:
uptime = int(bitcoinrpc("uptime"))
meminfo = bitcoinrpc("getmemoryinfo", "stats")["locked"]
blockchaininfo = bitcoinrpc("getblockchaininfo")
networkinfo = bitcoinrpc("getnetworkinfo")
chaintips = len(bitcoinrpc("getchaintips"))
mempool = bitcoinrpc("getmempoolinfo")
nettotals = bitcoinrpc("getnettotals")
rpcinfo = bitcoinrpc("getrpcinfo")
dsprooflist = bitcoinrpc("getdsprooflist")
txstats = bitcoinrpc("getchaintxstats")
latest_blockstats = getblockstats(str(blockchaininfo["bestblockhash"]))
hashps_120 = float(bitcoinrpc("getnetworkhashps", 120)) # 120 is the default
hashps_1 = float(bitcoinrpc("getnetworkhashps", 1))

banned = bitcoinrpc("listbanned")

BITCOIN_INFO.info({"chain": blockchaininfo["chain"], "coin": "BCH"})
BITCOIN_UPTIME.set(uptime)
BITCOIN_BLOCKS.set(blockchaininfo["blocks"])
BITCOIN_PEERS.set(networkinfo["connections"])
BITCOIN_DIFFICULTY.set(blockchaininfo["difficulty"])
BITCOIN_HASHPS.set(hashps_120)
BITCOIN_HASHPS_1.set(hashps_1)
BITCOIN_SERVER_VERSION.set(networkinfo["version"])
BITCOIN_PROTOCOL_VERSION.set(networkinfo["protocolversion"])
BITCOIN_SIZE_ON_DISK.set(blockchaininfo["size_on_disk"])
BITCOIN_VERIFICATION_PROGRESS.set(blockchaininfo["verificationprogress"])

for ban in banned:
BITCOIN_BAN_CREATED.labels(address=ban["address"], reason=ban.get("ban_reason", "manually added")).set(
ban["ban_created"]
)
BITCOIN_BANNED_UNTIL.labels(address=ban["address"], reason=ban.get("ban_reason", "manually added")).set(
ban["banned_until"]
)

if networkinfo["warnings"]:
BITCOIN_WARNINGS.inc()

BITCOIN_TXCOUNT.set(txstats["txcount"])

BITCOIN_NUM_CHAINTIPS.set(chaintips)

BITCOIN_MEMINFO_USED.set(meminfo["used"])
BITCOIN_MEMINFO_FREE.set(meminfo["free"])
BITCOIN_MEMINFO_TOTAL.set(meminfo["total"])
BITCOIN_MEMINFO_LOCKED.set(meminfo["locked"])
BITCOIN_MEMINFO_CHUNKS_USED.set(meminfo["chunks_used"])
BITCOIN_MEMINFO_CHUNKS_FREE.set(meminfo["chunks_free"])

BITCOIN_MEMPOOL_BYTES.set(mempool["bytes"])
BITCOIN_MEMPOOL_SIZE.set(mempool["size"])
BITCOIN_MEMPOOL_USAGE.set(mempool["usage"])

BITCOIN_TOTAL_BYTES_RECV.set(nettotals["totalbytesrecv"])
BITCOIN_TOTAL_BYTES_SENT.set(nettotals["totalbytessent"])

if latest_blockstats is not None:
BITCOIN_LATEST_BLOCK_SIZE.set(latest_blockstats["total_size"])
BITCOIN_LATEST_BLOCK_TXS.set(latest_blockstats["txs"])
BITCOIN_LATEST_BLOCK_HEIGHT.set(latest_blockstats["height"])
BITCOIN_LATEST_BLOCK_INPUTS.set(latest_blockstats["ins"])
BITCOIN_LATEST_BLOCK_OUTPUTS.set(latest_blockstats["outs"])
BITCOIN_LATEST_BLOCK_VALUE.set(latest_blockstats["total_out"])

# Subtract one because we don't want to count the "getrpcinfo" call itself
BITCOIN_RPC_ACTIVE.set(len(rpcinfo["active_commands"]) - 1)

BITCOIN_DSPROOF_COUNT.set(len(dsprooflist))

def sigterm_handler(signal, frame) -> None:
logger.critical("Received SIGTERM. Exiting.")
sys.exit(0)

def exception_count(e: Exception) -> None:
err_type = type(e)
exception_name = err_type.__module__ + "." + err_type.__name__
EXPORTER_ERRORS.labels(**{"type": exception_name}).inc()

def main():
# Set up logging to look similar to bitcoin logs (UTC).
logging.basicConfig(
format="%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%dT%H:%M:%SZ"
)
logging.Formatter.converter = time.gmtime
logger.setLevel(LOG_LEVEL)

# Handle SIGTERM gracefully.
signal.signal(signal.SIGTERM, sigterm_handler)

app = make_wsgi_app()

last_refresh = None

def refresh_app(*args, **kwargs):
nonlocal last_refresh
process_start = datetime.now()

if not last_refresh or (process_start - last_refresh).total_seconds() > 5: # Limit updates to every 5 seconds
# Allow riprova.MaxRetriesExceeded and unknown exceptions to crash the process.
try:
refresh_metrics()
except riprova.exceptions.RetryError as e:
logger.error("Refresh failed during retry. Cause: " + str(e))
exception_count(e)
except JSONRPCError as e:
logger.debug("Bitcoin RPC error refresh", exc_info=True)
exception_count(e)
except json.decoder.JSONDecodeError as e:
logger.error("RPC call did not return JSON. Bad credentials? " + str(e))
sys.exit(1)

duration = datetime.now() - process_start
PROCESS_TIME.inc(duration.total_seconds())
logger.info("Refresh took %s seconds", duration)
last_refresh = process_start

return app(*args, **kwargs)

httpd = make_server(METRICS_ADDR, METRICS_PORT, refresh_app)
httpd.serve_forever()

if __name__ == "__main__":
main()

== fulcrum-monitor.py ==

#!/usr/bin/env python3
# Copyright (c) 2021 Axel Gembe <derago@gmail.com>
# Copyright 2018 Kevin M. Gallagher
# Copyright 2019,2020 Jeff Stein
#
# Based on https://github.com/jvstein/bitcoin-prometheus-exporter
# Published at https://github.com/EchterAgo/fulcrum-prometheus-exporter
# Licensed under BSD 3-clause (see LICENSE).

import json
import logging
import time
import os
import signal
import sys
import urllib

from datetime import datetime
from wsgiref.simple_server import make_server

from prometheus_client import make_wsgi_app, Gauge, Counter, Info

logger = logging.getLogger("fulcrum-exporter")

# Create Prometheus metrics to track Fulcrum stats.
BITCOIND_EXTANT_REQUEST_CONTEXTS = Gauge("fulcrum_bitcoind_extant_request_contexts", "Extant bitcoind request contexts")
BITCOIND_REQUEST_CONTEXT_TABLE_SIZE = Gauge(
"fulcrum_bitcoind_request_context_table_size", "bitcoind request context table size")
BITCOIND_REQUEST_TIMEOUT_COUNT = Gauge("fulcrum_bitcoind_request_timeout_count", "bitcoind request timeout count")
BITCOIND_REQUEST_ZOMBIE_COUNT = Gauge("fulcrum_bitcoind_request_zombie_count", "bitcoind request zombie count")
BITCOIND_RPCCLIENT_COUNT = Gauge("fulcrum_bitcoind_rpcclient_count", "Number of bitcoind RPC clients in existence")

CONTROLLER_INFO = Info('fulcrum_controller', 'The chain and coin the controller runs on')
CONTROLLER_HEADER_COUNT = Gauge("fulcrum_controller_header_count", "Number of headers the controller knows about")
CONTROLLER_TX_NUM = Gauge("fulcrum_controller_tx_num", "Number of transactions the controller knows about")
CONTROLLER_UTXO_SET_COUNT = Gauge("fulcrum_controller_utxo_set_count", "Number of outputs in the UTXO set")
CONTROLLER_UTXO_SET_SIZE = Gauge("fulcrum_controller_utxo_set_size_mb", "Size of the UTXO set")
CONTROLLER_ZMQ_NOTIFICATION_COUNT = Gauge("fulcrum_controller_zmq_notification_count", "Number of ZMQ notifications received")
CONTROLLER_TASK_COUNT = Gauge("fulcrum_controller_task_count", "Number of controller tasks")

JEMALLOC_STATS_ACTIVE = Gauge("fulcrum_jemalloc_stats_active", "Jemalloc active bytes")
JEMALLOC_STATS_ALLOCATED = Gauge("fulcrum_jemalloc_stats_allocated", "Jemalloc allocated bytes")
JEMALLOC_STATS_MAPPED = Gauge("fulcrum_jemalloc_stats_mapped", "Jemalloc mapped bytes")
JEMALLOC_STATS_METADATA = Gauge("fulcrum_jemalloc_stats_metadata", "Jemalloc metadata bytes")
JEMALLOC_STATS_RESIDENT = Gauge("fulcrum_jemalloc_stats_resident", "Jemalloc resident bytes")
JEMALLOC_STATS_RETAINED = Gauge("fulcrum_jemalloc_stats_retained", "Jemalloc retained bytes")

MEMORY_USAGE_PHYSICAL_KB = Gauge("fulcrum_memory_usage_physical_kb", "Physical memory usage in kilobytes")
MEMORY_USAGE_VIRTUAL_KB = Gauge("fulcrum_memory_usage_virtual_kb", "Virtual memory usage in kilobytes")

JOB_QUEUE_EXTANT_JOBS = Gauge("fulcrum_job_queue_extant_jobs", "Number of jobs in the job queue")
JOB_QUEUE_EXTANT_JOBS_MAX_LIFETIME = Gauge("fulcrum_job_queue_extant_jobs_max_lifetime",
"Maximum number of jobs in the job queue over the process's life time")
JOB_QUEUE_EXTANT_JOBS_LIMIT = Gauge("fulcrum_job_queue_extant_jobs_limit", "Limit for number of jobs in the job queue")
JOB_QUEUE_COUNT_LIFETIME = Gauge("fulcrum_job_queue_count_lifetime",
"Number of jobs processed by the job queue over the process's life time")
JOB_QUEUE_OVERFLOWS_LIFETIME = Gauge("fulcrum_job_queue_overflows_lifetime",
"Number of job queue overflows over the process's life time")
JOB_QUEUE_THREAD_COUNT_MAX = Gauge("fulcrum_job_queue_thread_count_max", "Maximum number of threads for the job queue")

SERVER_MANAGER_PEER_COUNT = Gauge("fulcrum_server_manager_peer_count", "Peer count", labelnames=["peertype"])
SERVER_MANAGER_SERVER_CLIENT_COUNT = Gauge("fulcrum_server_manager_server_client_count",
"Client count by server type", labelnames=["servertype"])
SERVER_MANAGER_CLIENT_COUNT = Gauge("fulcrum_server_manager_client_count", "Client count")
SERVER_MANAGER_CLIENT_COUNT_MAX_LIFETIME = Gauge(
"fulcrum_server_manager_client_count_max_lifetime", "Max client count over the process's life time")
SERVER_MANAGER_TOTAL_LIFETIME_CLIENTS = Gauge(
"fulcrum_server_manager_total_lifetime_clients", "Total number of clients over the process's life time")
SERVER_MANAGER_TRANSACTIONS_SENT_COUNT = Gauge(
"fulcrum_server_manager_transactions_sent_count", "Number of transactions sent using this server")
SERVER_MANAGER_TRANSACTIONS_SENT_SIZE_BYTES = Gauge(
"fulcrum_server_manager_transactions_sent_size_bytes", "Size of the transactions sent using this server in bytes")

STORAGE_DB_SHARED_BLOCK_CACHE = Gauge("fulcrum_storage_db_shared_block_cache",
"Storage shared block cache", labelnames=["type"])
STORAGE_DB_SHARED_WRITE_BUFFER_MANAGER = Gauge("fulcrum_storage_db_shared_write_buffer_manager",
"Storage shared write buffer manager", labelnames=["type"])
STORAGE_DB_STATS_CUR_SIZE_ALL_MEM_TABLES_BYTES = Gauge("fulcrum_storage_db_stats_cur_size_all_mem_tables_bytes",
"Approximate size of active and unflushed immutable memtables in bytes", labelnames=["database"])
STORAGE_DB_STATS_ESTIMATE_TABLE_READERS_MEM_BYTES = Gauge("fulcrum_storage_db_stats_estimate_table_readers_mem_bytes",
"Estimated memory used for reading SST tables, excluding memory used in block cache (e.g., filter and index blocks) in bytes", labelnames=["database"])
STORAGE_CACHES_LRU_SIZE_BYTES = Gauge("fulcrum_storage_caches_lru_size_bytes",
"LRU Cache size in bytes", labelnames=["cachetype"])
STORAGE_CACHES_LRU_ENTRY_COUNT = Gauge("fulcrum_storage_caches_lru_entry_count",
"LRU Cache entry count", labelnames=["cachetype"])
STORAGE_CACHES_LRU_APPROX_HITS = Gauge("fulcrum_storage_caches_lru_approx_hits",
"LRU Cache approximate hits", labelnames=["cachetype"])
STORAGE_CACHES_LRU_APPROX_MISSES = Gauge("fulcrum_storage_caches_lru_approx_misses",
"LRU Cache approximate misses", labelnames=["cachetype"])
STORAGE_CACHES_MERKLEHEADERS = Gauge("fulcrum_storage_caches_merkleheaders", "Merkleheader cache size", labelnames=["type"])
STORAGE_MERGE_CALLS = Gauge("fulcrum_storage_merge_calls", "Merged storage calls")

SUBSMGR_ACTIVE_SUBS_COUNT = Gauge("fulcrum_subsmgr_active_subs_count", "Number of active client subscriptions")
SUBSMGR_UNIQUE_SCRIPTHASH_SUBS = Gauge("fulcrum_subsmgr_unique_subs_count",
"Number of unique scripthashes subscribed (including zombies)")
SUBSMGR_PENDING_NOTIF_COUNT = Gauge("fulcrum_subsmgr_pending_notif_count", "Number of pending notifications")
SUBSMGR_SUBS_BUCKET_COUNT = Gauge("fulcrum_subsmgr_subs_bucket_count", "Number of subscription buckets")
SUBSMGR_SUBS_CACHE_HITS = Gauge("fulcrum_subsmgr_subs_cache_hits", "Number of subscription cache hits")
SUBSMGR_SUBS_CACHE_MISSES = Gauge("fulcrum_subsmgr_subs_cache_misses", "Number of subscription cache misses")
SUBSMGR_SUBS_LOAD_FACTOR = Gauge("fulcrum_subsmgr_subs_load_factor", "Subscription load factor")

EXPORTER_ERRORS = Counter("fulcrum_exporter_errors",
"Number of errors encountered by the exporter", labelnames=["type"])
PROCESS_TIME = Counter("fulcrum_exporter_process_time", "Time spent processing metrics from Fulcrum")

FULCRUM_STATS_URL = os.environ.get("FULCRUM_STATS_URL", "http://127.0.0.1:8080/stats")
METRICS_ADDR = os.environ.get("METRICS_ADDR", "") # empty = any address
METRICS_PORT = int(os.environ.get("METRICS_PORT", "50039"))
RETRIES = int(os.environ.get("RETRIES", 5))
TIMEOUT = int(os.environ.get("TIMEOUT", 30))
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")

def refresh_metrics() -> None:
with urllib.request.urlopen(FULCRUM_STATS_URL) as stats_json:
stats = json.load(stats_json)

daemon = stats["Bitcoin Daemon"]
BITCOIND_EXTANT_REQUEST_CONTEXTS.set(daemon["extant request contexts"])
BITCOIND_REQUEST_CONTEXT_TABLE_SIZE.set(daemon["request context table size"])
BITCOIND_REQUEST_TIMEOUT_COUNT.set(daemon["request timeout count"])
BITCOIND_REQUEST_ZOMBIE_COUNT.set(daemon["request zombie count"])
BITCOIND_RPCCLIENT_COUNT.set(len(daemon["rpc clients"]))

ctrl = stats["Controller"]
CONTROLLER_INFO.info({"chain": ctrl["Chain"], "coin": ctrl["Coin"]})
CONTROLLER_HEADER_COUNT.set(ctrl["Header count"])
CONTROLLER_TX_NUM.set(ctrl["TxNum"])
CONTROLLER_UTXO_SET_COUNT.set(ctrl["UTXO set"])
CONTROLLER_UTXO_SET_SIZE.set(float(ctrl["UTXO set bytes"].split()[0]))
CONTROLLER_ZMQ_NOTIFICATION_COUNT.set(sum(value["notifications"]
for key, value in ctrl["ZMQ Notifiers (active)"].items()))
CONTROLLER_TASK_COUNT.set(len(ctrl["tasks"]))

if "Jemalloc" in stats:
jas = stats["Jemalloc"]["stats"]
JEMALLOC_STATS_ACTIVE.set(jas["active"])
JEMALLOC_STATS_ALLOCATED.set(jas["allocated"])
JEMALLOC_STATS_MAPPED.set(jas["mapped"])
JEMALLOC_STATS_METADATA.set(jas["metadata"])
JEMALLOC_STATS_RESIDENT.set(jas["resident"])
JEMALLOC_STATS_RETAINED.set(jas["retained"])

MEMORY_USAGE_PHYSICAL_KB.set(stats["Memory Usage"]["physical kB"])
MEMORY_USAGE_VIRTUAL_KB.set(stats["Memory Usage"]["virtual kB"])

jobq = stats["Misc"]["Job Queue (Thread Pool)"]
JOB_QUEUE_EXTANT_JOBS.set(jobq["extant jobs"])
JOB_QUEUE_EXTANT_JOBS_MAX_LIFETIME.set(jobq["extant jobs (max lifetime)"])
JOB_QUEUE_EXTANT_JOBS_LIMIT.set(jobq["extant limit"])
JOB_QUEUE_COUNT_LIFETIME.set(jobq["job count (lifetime)"])
JOB_QUEUE_OVERFLOWS_LIFETIME.set(jobq["job queue overflows (lifetime)"])
JOB_QUEUE_THREAD_COUNT_MAX.set(jobq["thread count (max)"])

srvm = stats["Server Manager"]

for peertype in ["bad", "failed", "peers", "queued"]:
SERVER_MANAGER_PEER_COUNT.labels(peertype).set(len(srvm["PeerMgr"][peertype]))

for servertype in ["AdminSrv", "SslSrv", "TcpSrv", "WsSrv", "WssSrv"]:
servers = [value for key, value in srvm["Servers"].items() if key.startswith(servertype)]
SERVER_MANAGER_SERVER_CLIENT_COUNT.labels(servertype).set(sum(s["numClients"] for s in servers))

SERVER_MANAGER_CLIENT_COUNT.set(srvm["number of clients"])
SERVER_MANAGER_CLIENT_COUNT_MAX_LIFETIME.set(srvm["number of clients (max lifetime)"])
SERVER_MANAGER_TOTAL_LIFETIME_CLIENTS.set(srvm["number of clients (total lifetime connections)"])
SERVER_MANAGER_TRANSACTIONS_SENT_COUNT.set(srvm["transactions sent"])
SERVER_MANAGER_TRANSACTIONS_SENT_SIZE_BYTES.set(srvm["transactions sent (bytes)"])

stor = stats["Storage"]
STORAGE_DB_SHARED_BLOCK_CACHE.labels("capacity").set(stor["DB Shared Block Cache"]["capacity"])
STORAGE_DB_SHARED_BLOCK_CACHE.labels("usage").set(stor["DB Shared Block Cache"]["usage"])
STORAGE_DB_SHARED_WRITE_BUFFER_MANAGER.labels("buffersize").set(
stor["DB Shared Write Buffer Manager"]["buffer size"])
STORAGE_DB_SHARED_WRITE_BUFFER_MANAGER.labels("memoryusage").set(
stor["DB Shared Write Buffer Manager"]["memory usage"])

for database in ["blkinfo", "meta", "scripthash_history", "scripthash_unspent", "undo", "utxoset"]:
STORAGE_DB_STATS_CUR_SIZE_ALL_MEM_TABLES_BYTES.labels(database).set(
stor["DB Stats"][database]["rocksdb.cur-size-all-mem-tables"])
STORAGE_DB_STATS_ESTIMATE_TABLE_READERS_MEM_BYTES.labels(database).set(
stor["DB Stats"][database]["rocksdb.estimate-table-readers-mem"])

STORAGE_CACHES_LRU_SIZE_BYTES.labels("blockheight2txhashes").set(
stor["caches"]["LRU Cache: Block Height -> TxHashes"]["Size bytes"])
STORAGE_CACHES_LRU_ENTRY_COUNT.labels("blockheight2txhashes").set(
stor["caches"]["LRU Cache: Block Height -> TxHashes"]["nBlocks"])
STORAGE_CACHES_LRU_APPROX_HITS.labels("blockheight2txhashes").set(
stor["caches"]["LRU Cache: Block Height -> TxHashes"]["~hits"])
STORAGE_CACHES_LRU_APPROX_MISSES.labels("blockheight2txhashes").set(
stor["caches"]["LRU Cache: Block Height -> TxHashes"]["~misses"])

STORAGE_CACHES_LRU_SIZE_BYTES.labels("txnum2txhash").set(stor["caches"]["LRU Cache: TxNum -> TxHash"]["Size bytes"])
STORAGE_CACHES_LRU_ENTRY_COUNT.labels("txnum2txhash").set(stor["caches"]["LRU Cache: TxNum -> TxHash"]["nItems"])
STORAGE_CACHES_LRU_APPROX_HITS.labels("txnum2txhash").set(stor["caches"]["LRU Cache: TxNum -> TxHash"]["~hits"])
STORAGE_CACHES_LRU_APPROX_MISSES.labels("txnum2txhash").set(stor["caches"]["LRU Cache: TxNum -> TxHash"]["~misses"])

STORAGE_CACHES_MERKLEHEADERS.labels("count").set(stor["caches"]["merkleHeaders_Size"])
STORAGE_CACHES_MERKLEHEADERS.labels("bytes").set(stor["caches"]["merkleHeaders_SizeBytes"])

STORAGE_MERGE_CALLS.set(stor["merge calls"])

subm = stats["SubsMgr"]
SUBSMGR_ACTIVE_SUBS_COUNT.set(subm["Num. active client subscriptions"])
SUBSMGR_UNIQUE_SCRIPTHASH_SUBS.set(subm["Num. unique scripthashes subscribed (including zombies)"])
SUBSMGR_PENDING_NOTIF_COUNT.set(len(subm["pendingNotifications"]))
SUBSMGR_SUBS_BUCKET_COUNT.set(subm["subscriptions bucket count"])
SUBSMGR_SUBS_CACHE_HITS.set(subm["subscriptions cache hits"])
SUBSMGR_SUBS_CACHE_MISSES.set(subm["subscriptions cache misses"])
SUBSMGR_SUBS_LOAD_FACTOR.set(subm["subscriptions load factor"])

def sigterm_handler(signal, frame) -> None:
logger.critical("Received SIGTERM. Exiting.")
sys.exit(0)

def exception_count(e: Exception) -> None:
err_type = type(e)
exception_name = err_type.__module__ + "." + err_type.__name__
EXPORTER_ERRORS.labels(**{"type": exception_name}).inc()

def main():
# Set up logging to look similar to bitcoin logs (UTC).
logging.basicConfig(
format="%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%dT%H:%M:%SZ"
)
logging.Formatter.converter = time.gmtime
logger.setLevel(LOG_LEVEL)

# Handle SIGTERM gracefully.
signal.signal(signal.SIGTERM, sigterm_handler)

app = make_wsgi_app()

last_refresh = None

def refresh_app(*args, **kwargs):
nonlocal last_refresh
process_start = datetime.now()

if not last_refresh or (process_start - last_refresh).total_seconds() > 1: # Limit updates to every 1 seconds
try:
refresh_metrics()
except Exception as e:
logger.debug("Refresh failed", exc_info=True)
exception_count(e)

duration = datetime.now() - process_start
PROCESS_TIME.inc(duration.total_seconds())
logger.info("Refresh took %s seconds", duration)
last_refresh = process_start

return app(*args, **kwargs)

httpd = make_server(METRICS_ADDR, METRICS_PORT, refresh_app)
httpd.serve_forever()

if __name__ == "__main__":
main()

Prometheus files - Revision history