Compaction Strategy Cassandra Ops & Automation

Monitoring Streaming Progress & ETA During a Cassandra Bootstrap

A bootstrap that streams half a terabyte can run for hours, and the difference between “it is working” and “it is wedged” is invisible from nodetool status alone — the node reads UJ either way. To babysit a join intelligently you need per-stream percent-complete and a moving ETA, so you can tell a healthy slow transfer from a stalled one and decide whether to raise throughput or intervene. This guide builds a Python 3.10+ monitor that derives both from Cassandra’s own streaming telemetry, and it pairs with the bootstrapping new nodes safely procedure it observes. It assumes nodetool access to the joining node running Cassandra 4.0, 4.1, or 5.0, and optionally the JMX port if you prefer structured MBeans over text parsing. The monitor is read-only and bounded — it polls on a fixed interval and exits when the join completes or a hard deadline passes, which makes it a natural wait loop to drop into an automated bootstrap orchestrator.

Pre-conditions

The monitor only makes sense against a node that is actively streaming. Confirm it is in the joining state and receiving data before you start watching.

# 1. The target node must be in JOINING mode with active receive sessions.
nodetool netstats | grep -E "Mode|Receiving"

Expected output: Mode: JOINING followed by one or more Receiving N files, ... bytes total lines. If Mode: NORMAL, the join already finished; if there are no Receiving lines, streaming has not begun yet.

# 2. Confirm nodetool responds quickly — the monitor polls it repeatedly.
time nodetool netstats > /dev/null

Expected output: the command returns in well under a second. If it is slow, JMX pressure will make your poll interval unreliable; widen the interval accordingly.

# 3. Record the current streaming throughput cap so you can interpret the ETA.
nodetool getstreamthroughput

Expected output: a value such as 200 (Mb/s on 4.0; unit-typed on 4.1+). An ETA that implies a rate far below this cap points to a source-side or network bottleneck rather than throttling.

Implementation

The monitor parses nodetool netstats on a bounded interval, aggregates received-versus-total bytes across every peer session, and derives an ETA from the byte-rate observed between polls. It reports per-peer percentages and an overall figure, and it flags a stall when bytes stop advancing. Parsing netstats keeps it dependency-free; the same numbers are available from the StreamManager JMX MBean if you would rather query structured data.

#!/usr/bin/env python3
# requirements: Python 3.10+, nodetool reachable for the joining node
"""Live streaming-progress and ETA monitor for a Cassandra 4.x/5.x bootstrap.

Parses `nodetool netstats`, aggregates received/total bytes per peer, and
derives a moving-average ETA. Read-only; polls on a bounded interval until the
join completes, stalls past a grace window, or a hard deadline elapses.
"""
from __future__ import annotations

import logging
import re
import subprocess
import time
from collections import deque
from dataclasses import dataclass, field

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("stream-monitor")

POLL_INTERVAL_S = 30
HARD_DEADLINE_S = 6 * 60 * 60      # give up watching after six hours
STALL_GRACE_S = 10 * 60           # no byte progress for this long => stalled
RATE_WINDOW = 10                  # samples in the moving-average rate window

# "Receiving 5 files, 12884901888 bytes total. Already received 3 files, 7730941132 bytes total"
_RECV_RE = re.compile(
    r"Receiving\s+\d+\s+files,\s+(\d+)\s+bytes total.*?"
    r"Already received\s+\d+\s+files.*?(\d+)\s+bytes total",
    re.DOTALL,
)
_PEER_RE = re.compile(r"/(\d{1,3}(?:\.\d{1,3}){3})")


@dataclass
class Progress:
    received: int = 0
    total: int = 0
    per_peer: dict[str, tuple[int, int]] = field(default_factory=dict)  # addr -> (recv, total)

    @property
    def pct(self) -> float:
        return 100.0 * self.received / self.total if self.total else 0.0


def run_netstats() -> str | None:
    try:
        r = subprocess.run(["nodetool", "netstats"], capture_output=True,
                           text=True, timeout=20, check=False)
        return r.stdout if r.returncode == 0 else None
    except subprocess.TimeoutExpired:
        log.warning("nodetool netstats timed out; will retry next poll.")
        return None


def parse_progress(text: str) -> Progress:
    """Aggregate received/total bytes across all peer receive sessions."""
    prog = Progress()
    # Split into per-peer blocks so a peer address associates with its counts.
    blocks = re.split(r"(?=/\d{1,3}(?:\.\d{1,3}){3})", text)
    for block in blocks:
        peer = _PEER_RE.search(block)
        if not peer:
            continue
        recv_total = 0
        recv_done = 0
        for total_s, done_s in _RECV_RE.findall(block):
            recv_total += int(total_s)
            recv_done += int(done_s)
        if recv_total:
            prog.per_peer[peer.group(1)] = (recv_done, recv_total)
            prog.received += recv_done
            prog.total += recv_total
    return prog


def is_complete(text: str) -> bool:
    return "Mode: NORMAL" in text and "Receiving" not in text


def format_eta(seconds: float) -> str:
    if seconds <= 0 or seconds != seconds:  # non-positive or NaN
        return "unknown"
    m, s = divmod(int(seconds), 60)
    h, m = divmod(m, 60)
    return f"{h}h{m:02d}m" if h else f"{m}m{s:02d}s"


def monitor() -> int:
    deadline = time.monotonic() + HARD_DEADLINE_S
    rates: deque[float] = deque(maxlen=RATE_WINDOW)
    last: tuple[float, int] | None = None      # (timestamp, received_bytes)
    last_progress_ts = time.monotonic()
    last_received = 0

    while time.monotonic() < deadline:
        text = run_netstats()
        if text is None:
            time.sleep(POLL_INTERVAL_S)
            continue
        if is_complete(text):
            log.info("Streaming complete — node has left JOINING mode.")
            return 0

        prog = parse_progress(text)
        now = time.monotonic()

        # Moving-average byte rate over the last RATE_WINDOW deltas.
        if last is not None:
            prev_ts, prev_recv = last
            elapsed = now - prev_ts
            if elapsed > 0 and prog.received >= prev_recv:
                rates.append((prog.received - prev_recv) / elapsed)
        last = (now, prog.received)

        avg_rate = sum(rates) / len(rates) if rates else 0.0
        remaining = max(prog.total - prog.received, 0)
        eta = remaining / avg_rate if avg_rate > 0 else float("nan")

        # Stall detection: has the received byte count advanced at all?
        if prog.received > last_received:
            last_received = prog.received
            last_progress_ts = now
        elif now - last_progress_ts > STALL_GRACE_S:
            log.error("STALLED: no byte progress for %s. Check for StreamException.",
                      format_eta(now - last_progress_ts))
            return 3

        log.info("Overall %.1f%% (%.1f/%.1f GiB) | %.1f MiB/s | ETA %s",
                 prog.pct, prog.received / 1024**3, prog.total / 1024**3,
                 avg_rate / 1024**2, format_eta(eta))
        for addr, (recv, total) in sorted(prog.per_peer.items()):
            log.info("   %-15s %5.1f%% (%.1f/%.1f GiB)",
                     addr, 100.0 * recv / total, recv / 1024**3, total / 1024**3)

        time.sleep(POLL_INTERVAL_S)

    log.error("Hard deadline reached; join still incomplete.")
    return 2


if __name__ == "__main__":
    raise SystemExit(monitor())

Two design choices keep the ETA honest. First, the rate is a moving average over the last several polls rather than an instantaneous delta, so a single slow interval does not swing the estimate wildly. Second, stall detection is independent of the ETA: even while an ETA is being computed, if the received byte count has not advanced for the grace window, the monitor declares a stall and exits non-zero — the signal that streaming has broken rather than merely slowed.

The netstats text format is stable across 4.0, 4.1, and 5.0, which is why the regexes above hold, but the same figures are exposed structurally through JMX under the org.apache.cassandra.net:type=StreamManager MBean. Querying the MBean returns per-session progress objects with explicit received and total byte counters, avoiding the fragility of text parsing if a future release reflows the netstats output. Whichever source you read, resist polling faster than every 15 to 30 seconds: each nodetool invocation opens a JMX connection and competes with the very streaming work you are measuring, so an aggressive interval both skews the rate calculation and steals cycles from the join. The bounded loop here caps total watch time at the hard deadline, then exits so a forgotten monitor never lingers against a node that finished hours ago. Because the whole tool is read-only, it is safe to run several instances — one per concurrent observer — without any risk to the joining node’s data.

Streaming progress bar with received bytes, remaining bytes, and derived ETA A horizontal progress bar divided into a received segment covering about sixty-five percent and a remaining segment covering the rest; a callout below the boundary shows the ETA derived from the moving-average byte rate applied to the remaining bytes. Bootstrap streaming — received vs remaining received 65% remaining 35% 0% 100% ETA ≈ remaining ÷ rate moving-average bytes/s
Progress is the sum of received bytes over total bytes across all peers; the ETA divides the remaining bytes by the moving-average transfer rate.

Verification steps

Run the monitor alongside the join and confirm its numbers track reality. Cross-check the overall percentage against a raw netstats snapshot:

nodetool netstats | grep -E "Already received"

The sum of the received/total byte pairs in that output should match the monitor’s Overall line within a poll interval. When the monitor logs Streaming complete, confirm the node has genuinely left joining mode:

nodetool netstats | grep Mode          # expect "Mode: NORMAL"
nodetool status | grep <joining-addr>  # expect UN

If you want a second opinion on the rate, sample the streaming MBean directly and compare its byte counters against the parsed values:

# StreamManager exposes per-session progress under this JMX domain.
nodetool sjk mxdump -q "org.apache.cassandra.net:type=StreamManager" 2>/dev/null || true

Troubleshooting

  • Stalled stream frozen at X%. The monitor’s percentage stops advancing and, after the grace window, it logs STALLED: no byte progress. Root cause: a source replica became unresponsive or its disk saturated, leaving one session parked mid-transfer while the others finished. Fix: inspect nodetool netstats to find which peer’s session is stuck, check that peer’s health and I/O, and if the session is dead, resume the join with nodetool bootstrap resume on the joining node; a resume retries only the incomplete ranges.
  • StreamException / broken pipe mid-transfer. The received count regresses or streaming drops entirely, and system.log on the joining node shows StreamException: Stream failed or a Broken pipe. Root cause: a transient network fault or a source node restart severed the socket. Fix: because a broken session will not self-heal, run nodetool bootstrap resume to reopen streams for the failed ranges, and if failures recur, lower streaming_connections_per_host to reduce per-peer socket pressure before retrying.
  • Throughput throttled too low, ETA keeps growing. The monitor reports a healthy but tiny rate — well below the value from nodetool getstreamthroughput — and the ETA drifts upward each poll. Root cause: the stream throughput cap on the source nodes is set conservatively, or a prior remediation left it low. Fix: raise the cap live on the source nodes with nodetool setstreamthroughput <higher-value> and watch the monitor’s MiB/s figure climb; keep the cap within the source nodes’ disk and NIC headroom so you do not starve client traffic.
Back to Bootstrapping New Nodes Safely