一个超级简单的见证人监控报警装置

oflyhigh

Published: 27 Nov 2025 › Updated: 27 Nov 2025 一个超级简单的见证人监控报警装置

一个超级简单的见证人监控报警装置

对于一个见证人来讲，最害怕三件事情是什么？答案是丢块、丢块还是丢块！

(图源：pixabay)

如果一个见证人节点长时间、持续性地大量丢块，没有出块奖励是小事情，每次丢块都会使我们的链延迟三秒。

虽然下个见证人会打包链延迟期间的交易，但是经常轮到自己出块却不干活，终归会让人觉得这个见证人技术能力稍微欠缺了一点点，可能会导致别人大量地撤票，见证人排名下滑。

但是就好比前些天CloudFlare出现全球性大故障，见证人节点或者放机房、或者放家里总会有可能遇到电力故障、网络故障、主机故障等问题，偶发的丢块在所难免，如何在丢块发生时，及时妥当地处理才是我们要解决的问题。

最优雅的方案呢，就是有至少两台以上见证人节点，运行监控脚本，当主节点丢块时，马上通过更新见证人参数中的出块公钥，将出块节点切换到备份节点上去。

另外一种方案，假设我们只有一个节点，运行监控脚本，当主节点丢块时，马上通过更新见证人参数中的出块公钥，使见证人临时离线，然后排查解决好故障后，再重新上线。这样可以避免持续不断地丢块。

对于TOP20节点，或者排名靠前的节点，因为出块的频率很高（TOP20几乎每分钟轮到一次出块机会），所以应该选择上述两种方案。但对于排名靠后的节点，也可以使用简单的监控报警装置，及时发现丢块，然后手工处理就好。

下边是一个简单的监控报警脚本，供参考：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import requests
import time
import logging
import RPi.GPIO as GPIO

# ----------------------
# 配置区域
# ----------------------
RPC_URL = "https://api.hive.blog"
WITNESS = "oflyhigh"
PRE_MISSED = 40

BUZZ_PIN = 11
BUZZ_FREQ = 6000
BEEP_DURATION = 180     # 秒
RETRY = 3               # 网络重试次数
TIMEOUT = 5             # 请求超时秒数

# ----------------------
# 日志
# ----------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

# ----------------------
# 蜂鸣器控制
# ----------------------
GPIO.setmode(GPIO.BOARD)
GPIO.setup(BUZZ_PIN, GPIO.OUT)
Buzz = GPIO.PWM(BUZZ_PIN, BUZZ_FREQ)


def beep(seconds):
    """蜂鸣提醒（避免阻塞主线程太久，可分段 beep）"""
    logging.warning(f"🚨 Witness missed block! Beeping for {seconds} seconds.")
    Buzz.start(50)

    end = time.time() + seconds
    while time.time() < end:
        Buzz.ChangeFrequency(BUZZ_FREQ)
        time.sleep(0.5)

    Buzz.stop()


# ----------------------
# RPC 调用
# ----------------------
def fetch_witness_missed(session, witness):
    """查询 witness 并返回 missed 值"""
    payload = {
        "jsonrpc": "2.0",
        "method": "condenser_api.get_witness_by_account",
        "params": [witness],
        "id": 1
    }

    headers = {
        "accept": "application/json",
        "content-type": "application/json"
    }

    for attempt in range(1, RETRY + 1):
        try:
            resp = session.post(RPC_URL, json=payload, headers=headers, timeout=TIMEOUT)

            if resp.status_code != 200:
                logging.error(f"RPC status {resp.status_code}: {resp.text}")
                continue

            data = resp.json()

            if "result" not in data or data["result"] is None:
                logging.error(f"Invalid RPC response: {data}")
                continue

            missed = data["result"].get("total_missed", None)
            if missed is None:
                logging.error("RPC result missing 'total_missed'")
                continue

            return missed

        except Exception as e:
            logging.error(f"RPC request failed (attempt {attempt}/{RETRY}): {e}")
            time.sleep(1)

    return None  # 多次失败返回 None


# ----------------------
# 主流程
# ----------------------
def main():
    session = requests.Session()
    logging.info(f"Checking witness '{WITNESS}'...")

    missed = fetch_witness_missed(session, WITNESS)
    if missed is None:
        logging.error("❌ Failed to get witness info.")
        return

    logging.info(f"Current missed: {missed}, Previous: {PRE_MISSED}")

    if missed > PRE_MISSED:
        beep(BEEP_DURATION)
    else:
        logging.info("👍 No new missed blocks.")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        logging.info("Program interrupted by user.")
    finally:
        GPIO.cleanup()
        logging.info("GPIO cleanup complete.")