mempool_monitor.py

#!/usr/bin/env python3
# Copyright (c) 2022 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.

""" Example logging Bitcoin Core mempool events using the mempool:added,
    mempool:removed, mempool:replaced, and mempool:rejected tracepoints. """

import curses
import sys
from datetime import datetime, timezone

from bcc import BPF, USDT

# BCC: The C program to be compiled to an eBPF program (by BCC) and loaded into
# a sandboxed Linux kernel VM.
PROGRAM = """
# include <uapi/linux/ptrace.h>

// The longest rejection reason is 118 chars and is generated in case of SCRIPT_ERR_EVAL_FALSE by
// strprintf("mandatory-script-verify-flag-failed (%s)", ScriptErrorString(check.GetScriptError()))
#define MAX_REJECT_REASON_LENGTH        118
// The longest string returned by RemovalReasonToString() is 'sizelimit'
#define MAX_REMOVAL_REASON_LENGTH       9
#define HASH_LENGTH                     32

struct added_event
{
  u8    hash[HASH_LENGTH];
  s32   vsize;
  s64   fee;
};

struct removed_event
{
  u8    hash[HASH_LENGTH];
  char  reason[MAX_REMOVAL_REASON_LENGTH];
  s32   vsize;
  s64   fee;
  u64   entry_time;
};

struct rejected_event
{
  u8    hash[HASH_LENGTH];
  char  reason[MAX_REJECT_REASON_LENGTH];
};

struct replaced_event
{
  u8    replaced_hash[HASH_LENGTH];
  s32   replaced_vsize;
  s64   replaced_fee;
  u64   replaced_entry_time;
  u8    replacement_hash[HASH_LENGTH];
  s32   replacement_vsize;
  s64   replacement_fee;
};

// BPF perf buffer to push the data to user space.
BPF_PERF_OUTPUT(added_events);
BPF_PERF_OUTPUT(removed_events);
BPF_PERF_OUTPUT(rejected_events);
BPF_PERF_OUTPUT(replaced_events);

int trace_added(struct pt_regs *ctx) {
  struct added_event added = {};

  bpf_usdt_readarg_p(1, ctx, &added.hash, HASH_LENGTH);
  bpf_usdt_readarg(2, ctx, &added.vsize);
  bpf_usdt_readarg(3, ctx, &added.fee);

  added_events.perf_submit(ctx, &added, sizeof(added));
  return 0;
}

int trace_removed(struct pt_regs *ctx) {
  struct removed_event removed = {};

  bpf_usdt_readarg_p(1, ctx, &removed.hash, HASH_LENGTH);
  bpf_usdt_readarg_p(2, ctx, &removed.reason, MAX_REMOVAL_REASON_LENGTH);
  bpf_usdt_readarg(3, ctx, &removed.vsize);
  bpf_usdt_readarg(4, ctx, &removed.fee);
  bpf_usdt_readarg(5, ctx, &removed.entry_time);

  removed_events.perf_submit(ctx, &removed, sizeof(removed));
  return 0;
}

int trace_rejected(struct pt_regs *ctx) {
  struct rejected_event rejected = {};

  bpf_usdt_readarg_p(1, ctx, &rejected.hash, HASH_LENGTH);
  bpf_usdt_readarg_p(2, ctx, &rejected.reason, MAX_REJECT_REASON_LENGTH);

  rejected_events.perf_submit(ctx, &rejected, sizeof(rejected));
  return 0;
}

int trace_replaced(struct pt_regs *ctx) {
  struct replaced_event replaced = {};

  bpf_usdt_readarg_p(1, ctx, &replaced.replaced_hash, HASH_LENGTH);
  bpf_usdt_readarg(2, ctx, &replaced.replaced_vsize);
  bpf_usdt_readarg(3, ctx, &replaced.replaced_fee);
  bpf_usdt_readarg(4, ctx, &replaced.replaced_entry_time);
  bpf_usdt_readarg_p(5, ctx, &replaced.replacement_hash, HASH_LENGTH);
  bpf_usdt_readarg(6, ctx, &replaced.replacement_vsize);
  bpf_usdt_readarg(7, ctx, &replaced.replacement_fee);

  replaced_events.perf_submit(ctx, &replaced, sizeof(replaced));
  return 0;
}
"""


def main(bitcoind_path):
    bitcoind_with_usdts = USDT(path=str(bitcoind_path))

    # attaching the trace functions defined in the BPF program
    # to the tracepoints
    bitcoind_with_usdts.enable_probe(probe="mempool:added", fn_name="trace_added")
    bitcoind_with_usdts.enable_probe(probe="mempool:removed", fn_name="trace_removed")
    bitcoind_with_usdts.enable_probe(probe="mempool:replaced", fn_name="trace_replaced")
    bitcoind_with_usdts.enable_probe(probe="mempool:rejected", fn_name="trace_rejected")
    bpf = BPF(text=PROGRAM, usdt_contexts=[bitcoind_with_usdts])

    events = []

    def get_timestamp():
        return datetime.now(timezone.utc)

    def handle_added(_, data, size):
        event = bpf["added_events"].event(data)
        events.append((get_timestamp(), "added", event))

    def handle_removed(_, data, size):
        event = bpf["removed_events"].event(data)
        events.append((get_timestamp(), "removed", event))

    def handle_rejected(_, data, size):
        event = bpf["rejected_events"].event(data)
        events.append((get_timestamp(), "rejected", event))

    def handle_replaced(_, data, size):
        event = bpf["replaced_events"].event(data)
        events.append((get_timestamp(), "replaced", event))

    bpf["added_events"].open_perf_buffer(handle_added)
    # By default, open_perf_buffer uses eight pages for a buffer, making for a total
    # buffer size of 32k on most machines. In practice, this size is insufficient:
    # Each `mempool:removed` event takes up 57 bytes in the buffer (32 bytes for txid,
    # 9 bytes for removal reason, and 8 bytes each for vsize and fee). Full blocks
    # contain around 2k transactions, requiring a buffer size of around 114kB. To cover
    # this amount, 32 4k pages are required.
    bpf["removed_events"].open_perf_buffer(handle_removed, page_cnt=32)
    bpf["rejected_events"].open_perf_buffer(handle_rejected)
    bpf["replaced_events"].open_perf_buffer(handle_replaced)

    curses.wrapper(loop, bpf, events)


def loop(screen, bpf, events):
    dashboard = Dashboard(screen)
    while True:
        try:
            bpf.perf_buffer_poll(timeout=50)
            dashboard.render(events)
        except KeyboardInterrupt:
            exit()


class Dashboard:
    """Visualization of mempool state using ncurses."""

    INFO_WIN_HEIGHT = 2
    EVENT_WIN_HEIGHT = 7

    def __init__(self, screen):
        screen.nodelay(True)
        curses.curs_set(False)
        self._screen = screen
        self._time_started = datetime.now(timezone.utc)
        self._timestamps = {"added": [], "removed": [], "rejected": [], "replaced": []}
        self._event_history = {"added": 0, "removed": 0, "rejected": 0, "replaced": 0}
        self._init_windows()

    def _init_windows(self):
        """Initialize all windows."""
        self._init_info_win()
        self._init_event_count_win()
        self._init_event_rate_win()
        self._init_event_log_win()

    @staticmethod
    def create_win(x, y, height, width, title=None):
        """Helper function to create generic windows and decorate them with box and title if requested."""
        win = curses.newwin(height, width, x, y)
        if title:
            win.box()
            win.addstr(0, 2, title, curses.A_BOLD)
        return win

    def _init_info_win(self):
        """Create and populate the info window."""
        self._info_win = Dashboard.create_win(
            x=0, y=1, height=Dashboard.INFO_WIN_HEIGHT, width=22
        )
        self._info_win.addstr(0, 0, "Mempool Monitor", curses.A_REVERSE)
        self._info_win.addstr(1, 0, "Press CTRL-C to stop.", curses.A_NORMAL)
        self._info_win.refresh()

    def _init_event_count_win(self):
        """Create and populate the event count window."""
        self._event_count_win = Dashboard.create_win(
            x=3, y=1, height=Dashboard.EVENT_WIN_HEIGHT, width=37, title="Event count"
        )
        header = " {:<8} {:>8} {:>7} {:>7} "
        self._event_count_win.addstr(
            1, 1, header.format("Event", "total", "1 min", "10 min"), curses.A_UNDERLINE
        )
        self._event_count_win.refresh()

    def _init_event_rate_win(self):
        """Create and populate the event rate window."""
        self._event_rate_win = Dashboard.create_win(
            x=3, y=40, height=Dashboard.EVENT_WIN_HEIGHT, width=42, title="Event rate"
        )
        header = " {:<8} {:>9} {:>9} {:>9} "
        self._event_rate_win.addstr(
            1, 1, header.format("Event", "total", "1 min", "10 min"), curses.A_UNDERLINE
        )
        self._event_rate_win.refresh()

    def _init_event_log_win(self):
        """Create windows showing event log. This comprises a dummy boxed window and an
        inset window so line breaks don't overwrite box."""
        # dummy boxed window
        num_rows, num_cols = self._screen.getmaxyx()
        space_above = Dashboard.INFO_WIN_HEIGHT + 1 + Dashboard.EVENT_WIN_HEIGHT + 1
        box_win_height = num_rows - space_above
        box_win_width = num_cols - 2
        win_box = Dashboard.create_win(
            x=space_above,
            y=1,
            height=box_win_height,
            width=box_win_width,
            title="Event log",
        )
        # actual logging window
        log_lines = box_win_height - 2  # top and bottom box lines
        log_line_len = box_win_width - 2 - 1  # box lines and left padding
        win = win_box.derwin(log_lines, log_line_len, 1, 2)
        win.idlok(True)
        win.scrollok(True)
        win_box.refresh()
        win.refresh()
        self._event_log_win_box = win_box
        self._event_log_win = win

    def calculate_metrics(self, events):
        """Calculate count and rate metrics."""
        count, rate = {}, {}
        for event_ts, event_type, event_data in events:
            self._timestamps[event_type].append(event_ts)
        for event_type, ts in self._timestamps.items():
            # remove timestamps older than ten minutes but keep track of their
            # count for the 'total' metric
            #
            self._event_history[event_type] += len(
                [t for t in ts if Dashboard.timestamp_age(t) >= 600]
            )
            ts = [t for t in ts if Dashboard.timestamp_age(t) < 600]
            self._timestamps[event_type] = ts
            # count metric
            count_1m = len([t for t in ts if Dashboard.timestamp_age(t) < 60])
            count_10m = len(ts)
            count_total = self._event_history[event_type] + len(ts)
            count[event_type] = (count_total, count_1m, count_10m)
            # rate metric
            runtime = Dashboard.timestamp_age(self._time_started)
            rate_1m = count_1m / min(60, runtime)
            rate_10m = count_10m / min(600, runtime)
            rate_total = count_total / runtime
            rate[event_type] = (rate_total, rate_1m, rate_10m)
        return count, rate

    def _update_event_count(self, count):
        """Update the event count window."""
        w = self._event_count_win
        row_format = " {:<8} {:>6}tx {:>5}tx {:>5}tx "
        for line, metric in enumerate(["added", "removed", "replaced", "rejected"]):
            w.addstr(2 + line, 1, row_format.format(metric, *count[metric]))
        w.refresh()

    def _update_event_rate(self, rate):
        """Update the event rate window."""
        w = self._event_rate_win
        row_format = " {:<8} {:>5.1f}tx/s {:>5.1f}tx/s {:>5.1f}tx/s "
        for line, metric in enumerate(["added", "removed", "replaced", "rejected"]):
            w.addstr(2 + line, 1, row_format.format(metric, *rate[metric]))
        w.refresh()

    def _update_event_log(self, events):
        """Update the event log window."""
        w = self._event_log_win
        for event in events:
            w.addstr(Dashboard.parse_event(event) + "\n")
        w.refresh()

    def render(self, events):
        """Render the dashboard."""
        count, rate = self.calculate_metrics(events)
        self._update_event_count(count)
        self._update_event_rate(rate)
        self._update_event_log(events)
        events.clear()

    @staticmethod
    def parse_event(event):
        """Converts events into human-readable messages"""

        ts_dt, type_, data = event
        ts = ts_dt.strftime("%H:%M:%SZ")
        if type_ == "added":
            return (
                f"{ts} added {bytes(data.hash)[::-1].hex()}"
                f" with feerate {data.fee/data.vsize:.2f} sat/vB"
                f" ({data.fee} sat, {data.vsize} vbytes)"
            )

        if type_ == "removed":
            return (
                f"{ts} removed {bytes(data.hash)[::-1].hex()}"
                f" with feerate {data.fee/data.vsize:.2f} sat/vB"
                f" ({data.fee} sat, {data.vsize} vbytes)"
                f" received {ts_dt.timestamp()-data.entry_time:.1f} seconds ago"
                f": {data.reason.decode('UTF-8')}"
            )

        if type_ == "rejected":
            return (
                f"{ts} rejected {bytes(data.hash)[::-1].hex()}"
                f": {data.reason.decode('UTF-8')}"
            )

        if type_ == "replaced":
            return (
                f"{ts} replaced {bytes(data.replaced_hash)[::-1].hex()}"
                f" with feerate {data.replaced_fee/data.replaced_vsize:.2f} sat/vB"
                f" received {ts_dt.timestamp()-data.replaced_entry_time:.1f} seconds ago"
                f" ({data.replaced_fee} sat, {data.replaced_vsize} vbytes)"
                f" with {bytes(data.replacement_hash)[::-1].hex()}"
                f" with feerate {data.replacement_fee/data.replacement_vsize:.2f} sat/vB"
                f" ({data.replacement_fee} sat, {data.replacement_vsize} vbytes)"
            )

        raise NotImplementedError("Unsupported event type: {type_}")

    @staticmethod
    def timestamp_age(timestamp):
        """Return age of timestamp in seconds."""
        return (datetime.now(timezone.utc) - timestamp).total_seconds()


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("USAGE: ", sys.argv[0], "path/to/bitcoind")
        exit(1)

    path = sys.argv[1]
    main(path)