Observers

Interface

quickrib.observers.observer.Observer

Bases: Protocol

Base class and structural type for everything attached to a RIBTable.

The three update_* hooks run inside the replay loop, once per BGP message, so keep them O(1) and allocation-free; push aggregation into :meth:dump. See the performance notes in AGENTS.md.

RIBTable ignores whatever the hooks return, but they are typed as returning something so an observer can hand a result to a subclass, as :class:~quickrib.observers.update_tagger.UpdateTagsCounter does with the tagger's flags.

Source code in src/quickrib/observers/observer.py

class Observer(Protocol):
    """Base class and structural type for everything attached to a ``RIBTable``.

    The three ``update_*`` hooks run inside the replay loop, once per BGP
    message, so keep them O(1) and allocation-free; push aggregation into
    :meth:`dump`. See the performance notes in AGENTS.md.

    ``RIBTable`` ignores whatever the hooks return, but they are typed as
    returning something so an observer can hand a result to a subclass, as
    :class:`~quickrib.observers.update_tagger.UpdateTagsCounter` does with the
    tagger's flags.
    """

    name: str = "observer"
    """Identifies the observer; :meth:`RIBTable.compare` pairs observers by it."""

    def __init__(self, name: str = "observer") -> None:
        self.name = name

    def update_rib(self, bgpelem: ParsedElement) -> Any:
        """A RIB-dump entry, replayed while the table is being built."""
        return None

    def update_announcement(
        self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData]
    ) -> Any:
        """An announcement.

        ``data`` is the prefix's new node ``.data``; ``old_data`` is a snapshot of
        the previous one, or ``None`` when the peer did not have this prefix.
        Comparing the two is how an observer tells a real change from a
        re-announcement that only moved an attribute it does not care about.
        """
        return None

    def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData]) -> Any:
        """A withdrawal. ``data`` is the withdrawn node ``.data``, or ``None`` if
        the peer did not have the prefix, in which case nothing was counted and
        there is nothing to take back."""
        return None

    def dump(self, ts: datetime) -> Any:
        """Emit the result for the window ending at ``ts``, and return it.

        ``ts`` is the window *boundary* (``start_time + k * dump_res``), not the
        message that happened to cross it. Every elapsed boundary is reported
        even across a gap in the stream, so a series has one entry per window
        however busy the collectors were.
        """
        return None

    def compare(self, other: Observer) -> None:
        """Report how this observer differs from one built on a ground-truth RIB."""
        return None

    def set_rib(self, rib: RIBTable) -> None:
        """Receive a back-reference to the subject, once the RIB is built.

        The escape hatch for observers that genuinely cannot work from the
        notifications alone; treat the reference as read-only.
        """
        return None

    def set_full_feed_peers(self, ff_peers: set[VP]) -> None:
        """Receive the pipeline's full-feed selection, before any replay."""
        return None

name = name class-attribute instance-attribute

Identifies the observer; :meth:RIBTable.compare pairs observers by it.

update_rib(bgpelem)

A RIB-dump entry, replayed while the table is being built.

Source code in src/quickrib/observers/observer.py

def update_rib(self, bgpelem: ParsedElement) -> Any:
    """A RIB-dump entry, replayed while the table is being built."""
    return None

update_announcement(bgpelem, data, old_data)

An announcement.

data is the prefix's new node .data; old_data is a snapshot of the previous one, or None when the peer did not have this prefix. Comparing the two is how an observer tells a real change from a re-announcement that only moved an attribute it does not care about.

Source code in src/quickrib/observers/observer.py

def update_announcement(
    self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData]
) -> Any:
    """An announcement.

    ``data`` is the prefix's new node ``.data``; ``old_data`` is a snapshot of
    the previous one, or ``None`` when the peer did not have this prefix.
    Comparing the two is how an observer tells a real change from a
    re-announcement that only moved an attribute it does not care about.
    """
    return None

update_withdrawal(bgpelem, data)

A withdrawal. data is the withdrawn node .data, or None if the peer did not have the prefix, in which case nothing was counted and there is nothing to take back.

Source code in src/quickrib/observers/observer.py

def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData]) -> Any:
    """A withdrawal. ``data`` is the withdrawn node ``.data``, or ``None`` if
    the peer did not have the prefix, in which case nothing was counted and
    there is nothing to take back."""
    return None

dump(ts)

Emit the result for the window ending at ts, and return it.

ts is the window boundary (start_time + k * dump_res), not the message that happened to cross it. Every elapsed boundary is reported even across a gap in the stream, so a series has one entry per window however busy the collectors were.

Source code in src/quickrib/observers/observer.py

def dump(self, ts: datetime) -> Any:
    """Emit the result for the window ending at ``ts``, and return it.

    ``ts`` is the window *boundary* (``start_time + k * dump_res``), not the
    message that happened to cross it. Every elapsed boundary is reported
    even across a gap in the stream, so a series has one entry per window
    however busy the collectors were.
    """
    return None

compare(other)

Report how this observer differs from one built on a ground-truth RIB.

Source code in src/quickrib/observers/observer.py

def compare(self, other: Observer) -> None:
    """Report how this observer differs from one built on a ground-truth RIB."""
    return None

set_rib(rib)

Receive a back-reference to the subject, once the RIB is built.

The escape hatch for observers that genuinely cannot work from the notifications alone; treat the reference as read-only.

Source code in src/quickrib/observers/observer.py

def set_rib(self, rib: RIBTable) -> None:
    """Receive a back-reference to the subject, once the RIB is built.

    The escape hatch for observers that genuinely cannot work from the
    notifications alone; treat the reference as read-only.
    """
    return None

set_full_feed_peers(ff_peers)

Receive the pipeline's full-feed selection, before any replay.

Source code in src/quickrib/observers/observer.py

def set_full_feed_peers(self, ff_peers: set[VP]) -> None:
    """Receive the pipeline's full-feed selection, before any replay."""
    return None

AS Hegemony

quickrib.observers.hegemony.HegemonyObserver

Bases: Observer

Incremental edge AS-hegemony.

For every vantage point (collector, peer ASN, peer IP) the observer keeps

• edge_paths_count[vp][edge]: the summed IP-space weight of the prefixes routed by that VP whose AS path contains edge;
• num_total_paths[vp]: the VP's total routed IP space, which normalises its betweenness centrality.

Both are updated in update_rib / update_announcement / update_withdrawal at a cost of O(path length) (plus O(#more-specifics) per update in deaggregate mode). :meth:dump turns the current counts into a :class:~quickrib.observers._gdecomp.Graph of per-edge hegemony (per-VP betweenness, then :func:trim_mean across VPs).

Parameters:

Name	Type	Description	Default
alpha	float	Trimmed-mean cut fraction passed to :func:trim_mean.	0.2
deaggregate	bool	False (default): weight each prefix by its plain block size (2**(32-plen)). This is the paper's ips_count variant: cheap, exactly reversible, but a prefix and its more-specifics both count their full block, so overlapping space is counted several times. True: weight each prefix by its exclusive address space, maintained incrementally with a per-VP radix trie: announcing a more-specific transfers space away from its covering prefix (and from every edge on that prefix's path), withdrawing it gives the space back. Matches the intent of the deaggregation-aware :func:radix_prefix_size used by :func:edge_hegemony.	False
min_hegemony	float	Edges whose trimmed-mean hegemony is at or below this are dropped from the dumped graph. Zero by default: only weightless edges go. Opt in to a real cut only if a smaller graph is worth what it costs, which is more than it looks. See the note below.	0.0

Relation to :func:edge_hegemony

deaggregate=True reproduces :func:edge_hegemony exactly (up to floating-point accumulation order): same per-VP betweenness weighted by deaggregation-aware prefix size, same normalisation by the VP's covered address count, same :func:trim_mean across VPs.

deaggregate=False skips deaggregation: a prefix and each of its more-specifics both count their full block, so num_total_paths[vp] sums overlapping space. Cheaper and exactly reversible; absolute magnitudes differ from :func:edge_hegemony while rankings stay close.

Why min_hegemony defaults to zero

A non-zero cut decides graph membership by comparing a float to a threshold, and consumers downstream are not continuous in membership: the MAD detector pads the edge vector to a power of two, so one edge entering or leaving can re-bin every wavelet coefficient. With a cut at 1e-15 an edge sitting on the threshold flipped on a last-bit difference, which made mathematically-equivalent changes look like regressions (see AGENTS.md, "A MAD score is not reproducible to the last bit").

At zero, membership is decided by sign rather than by comparison with a small number: an edge is in the graph exactly when its hegemony is strictly positive. A sum of non-negative terms is positive or it is not, whatever order it is accumulated in, so no rounding can add or remove an edge.

Zero is also the only cut at which the incremental and post-hoc paths agree on membership. An edge carrying no weight is absent from edge_paths_count, which never stores a zero, but present in the post-hoc walk. In deaggregate mode this happens whenever a prefix is entirely covered by its own more-specifics, leaving it an effective size of zero. Dropping it on both sides is what keeps deaggregate=True equal to :func:edge_hegemony.

Note

This observer never needs a back-reference to the RIBTable: it is warm-started automatically because QuickRIB.build_rib replays every RIB entry through update_rib before the first dump, so set_rib is a no-op even though the pipeline does call it.

Source code in src/quickrib/observers/hegemony.py

class HegemonyObserver(Observer):
    """Incremental edge AS-hegemony.

    For every vantage point (collector, peer ASN, peer IP) the observer keeps

    * ``edge_paths_count[vp][edge]``: the summed IP-space weight of the
      prefixes routed by that VP whose AS path contains ``edge``;
    * ``num_total_paths[vp]``: the VP's total routed IP space, which normalises
      its betweenness centrality.

    Both are updated in ``update_rib`` / ``update_announcement`` /
    ``update_withdrawal`` at a cost of O(path length) (plus O(#more-specifics)
    per update in ``deaggregate`` mode). :meth:`dump` turns the current counts
    into a :class:`~quickrib.observers._gdecomp.Graph` of per-edge hegemony
    (per-VP betweenness, then :func:`trim_mean` across VPs).

    Parameters
    ----------
    alpha:
        Trimmed-mean cut fraction passed to :func:`trim_mean`.
    deaggregate:
        ``False`` (default): weight each prefix by its plain block size
        (``2**(32-plen)``). This is the paper's ``ips_count`` variant: cheap,
        exactly reversible, but a prefix and its more-specifics both count their
        full block, so overlapping space is counted several times.

        ``True``: weight each prefix by its *exclusive* address space,
        maintained incrementally with a per-VP radix trie: announcing a
        more-specific transfers space away from its covering prefix (and from
        every edge on that prefix's path), withdrawing it gives the space back.
        Matches the intent of the deaggregation-aware :func:`radix_prefix_size`
        used by :func:`edge_hegemony`.
    min_hegemony:
        Edges whose trimmed-mean hegemony is at or below this are dropped from
        the dumped graph. **Zero by default: only weightless edges go.** Opt in
        to a real cut only if a smaller graph is worth what it costs, which is
        more than it looks. See the note below.

    Relation to :func:`edge_hegemony`
    --------------------------------
    ``deaggregate=True`` reproduces :func:`edge_hegemony` exactly (up to
    floating-point accumulation order): same per-VP betweenness weighted by
    deaggregation-aware prefix size, same normalisation by the VP's covered
    address count, same :func:`trim_mean` across VPs.

    ``deaggregate=False`` skips deaggregation: a prefix and each of its
    more-specifics both count their full block, so ``num_total_paths[vp]`` sums
    overlapping space. Cheaper and exactly reversible; absolute magnitudes
    differ from :func:`edge_hegemony` while rankings stay close.

    Why ``min_hegemony`` defaults to zero
    -------------------------------------
    A non-zero cut decides graph *membership* by comparing a float to a
    threshold, and consumers downstream are not continuous in membership: the
    MAD detector pads the edge vector to a power of two, so one edge entering or
    leaving can re-bin every wavelet coefficient. With a cut at 1e-15 an edge
    sitting on the threshold flipped on a last-bit difference, which made
    mathematically-equivalent changes look like regressions (see AGENTS.md, "A
    MAD score is not reproducible to the last bit").

    At zero, membership is decided by *sign* rather than by comparison with a
    small number: an edge is in the graph exactly when its hegemony is strictly
    positive. A sum of non-negative terms is positive or it is not, whatever
    order it is accumulated in, so no rounding can add or remove an edge.

    Zero is also the only cut at which the incremental and post-hoc paths agree
    on membership. An edge carrying no weight is absent from
    ``edge_paths_count``, which never stores a zero, but present in the post-hoc
    walk. In ``deaggregate`` mode this happens whenever a prefix is entirely
    covered by its own more-specifics, leaving it an effective size of zero. Dropping it on both sides is what
    keeps ``deaggregate=True`` equal to :func:`edge_hegemony`.

    Note
    ----
    This observer never needs a back-reference to the ``RIBTable``: it is
    warm-started automatically because ``QuickRIB.build_rib`` replays every RIB
    entry through ``update_rib`` before the first ``dump``, so ``set_rib`` is a
    no-op even though the pipeline does call it.
    """

    def __init__(
        self,
        name: str = "hegemony",
        alpha: float = 0.2,
        deaggregate: bool = False,
        min_hegemony: float = 0.0,
    ) -> None:
        self.name = name
        self.alpha = alpha
        self.deaggregate = deaggregate
        self.min_hegemony = min_hegemony

        # vp -> edge -> summed weight (effective space in deaggregate mode)
        self.edge_paths_count: dict[VP, dict[tuple[str, str], float]] = defaultdict(dict)
        # vp -> summed weight over all the VP's prefixes
        self.num_total_paths: dict[VP, float] = defaultdict(float)

        # deaggregate mode only: vp -> radix trie, node .data = {"path", "overlap"}
        # where overlap == summed block size of the prefix's direct more-specifics.
        self._tries: dict[VP, radix.Radix] = defaultdict(radix.Radix)

        # dump() memoisation so several readers (MAD variants) share one instance.
        self._last_dump_ts: Optional[datetime] = None
        self._last_graph: Optional[Graph] = None

    # ------------------------------------------------------------------ helpers

    def _edge_add(self, vp: VP, path: list[str], weight: float) -> None:
        ec = self.edge_paths_count[vp]
        for edge in _canon_edges(path):
            new = ec.get(edge, 0) + weight
            if new == 0:
                ec.pop(edge, None)
            else:
                ec[edge] = new

    def _plain_add(self, vp: VP, prefix: str, path: list[str]) -> None:
        w = block_size(prefix)
        self._edge_add(vp, path, w)
        self.num_total_paths[vp] += w

    def _plain_remove(self, vp: VP, prefix: str, path: list[str]) -> None:
        w = block_size(prefix)
        self._edge_add(vp, path, -w)
        self.num_total_paths[vp] -= w

    def _deagg_insert(self, vp: VP, prefix: str, path: list[str]) -> None:
        trie = self._tries[vp]
        if trie.search_exact(prefix) is not None:
            # already present (duplicate RIB entry): treat as a path move
            self._deagg_move(vp, prefix, path)
            return
        rnode = trie.add(prefix)
        nominal = block_size(prefix)
        # every direct child of the new node was, before insertion, a direct
        # child of its parent (nothing can sit between parent and new node).
        overlap_p = sum(block_size(c.prefix) for c in direct_children(trie, rnode))
        rnode.data["path"] = path
        rnode.data["overlap"] = overlap_p
        eff_p = nominal - overlap_p

        self._edge_add(vp, path, eff_p)
        self.num_total_paths[vp] += eff_p

        parent = rnode.parent
        if parent is not None:
            # parent gains the new node as a direct child (+nominal) and loses
            # the children that reparented to it (-overlap_p).
            delta_overlap = nominal - overlap_p
            parent.data["overlap"] += delta_overlap
            self._edge_add(vp, parent.data["path"], -delta_overlap)
            self.num_total_paths[vp] -= delta_overlap

    def _deagg_move(self, vp: VP, prefix: str, new_path: list[str]) -> None:
        trie = self._tries[vp]
        rnode = trie.search_exact(prefix)
        if rnode is None:
            self._deagg_insert(vp, prefix, new_path)
            return
        eff_p = block_size(prefix) - rnode.data["overlap"]
        self._edge_add(vp, rnode.data["path"], -eff_p)
        self._edge_add(vp, new_path, eff_p)
        rnode.data["path"] = new_path

    def _deagg_withdraw(self, vp: VP, prefix: str) -> None:
        trie = self._tries[vp]
        rnode = trie.search_exact(prefix)
        if rnode is None:
            return
        nominal = block_size(prefix)
        overlap_p = rnode.data["overlap"]
        eff_p = nominal - overlap_p

        self._edge_add(vp, rnode.data["path"], -eff_p)
        self.num_total_paths[vp] -= eff_p

        parent = rnode.parent
        if parent is not None:
            # children reparent to parent: parent.overlap += sum(nominal children)
            # - nominal(prefix) == overlap_p - nominal == -eff_p
            delta_overlap = overlap_p - nominal
            parent.data["overlap"] += delta_overlap
            self._edge_add(vp, parent.data["path"], -delta_overlap)
            self.num_total_paths[vp] -= delta_overlap

        trie.delete(prefix)

    # ---------------------------------------------------------------- observer


    def update_rib(self, bgpelem) -> None:
        vp = (bgpelem.collector, bgpelem.peer_asn, bgpelem.peer_address)
        path = bgpelem.fields["as-path"]
        if self.deaggregate:
            self._deagg_insert(vp, bgpelem.fields["prefix"], path)
        else:
            self._plain_add(vp, bgpelem.fields["prefix"], path)

    def update_announcement(self, bgpelem, data: RIBNodeData, old_data: Optional[RIBNodeData]) -> None:
        vp = (bgpelem.collector, bgpelem.peer_asn, bgpelem.peer_address)
        prefix = bgpelem.fields["prefix"]
        new_path = data["as-path"]
        if old_data is not None:
            old_path = old_data["as-path"]
            if old_path == new_path:
                return  # duplicate announcement, nothing changes
            if self.deaggregate:
                self._deagg_move(vp, prefix, new_path)
            else:
                self._plain_remove(vp, prefix, old_path)
                self._plain_add(vp, prefix, new_path)
        else:
            if self.deaggregate:
                self._deagg_insert(vp, prefix, new_path)
            else:
                self._plain_add(vp, prefix, new_path)

    def update_withdrawal(self, bgpelem, data: Optional[RIBNodeData]) -> None:
        if data is None:
            return  # prefix was not in the RIB
        vp = (bgpelem.collector, bgpelem.peer_asn, bgpelem.peer_address)
        if self.deaggregate:
            self._deagg_withdraw(vp, bgpelem.fields["prefix"])
        else:
            self._plain_remove(vp, bgpelem.fields["prefix"], data["as-path"])

    def dump(self, ts: datetime) -> Graph:
        """Return the current per-edge hegemony as a :class:`Graph` (bidirectional)."""
        if ts == self._last_dump_ts and self._last_graph is not None:
            return self._last_graph

        per_vp_scores: dict[tuple[str, str], list[float]] = defaultdict(list)
        for vp, ec in self.edge_paths_count.items():
            norm = self.num_total_paths.get(vp, 0.0)
            if norm <= 0:
                continue
            for edge, score in betweenness(ec, norm).items():
                per_vp_scores[edge].append(score)

        hegemony = _trim_across_vps(per_vp_scores, len(self.edge_paths_count), self.alpha)
        graph = _hegemony_graph(hegemony, self.min_hegemony)

        self._last_dump_ts, self._last_graph = ts, graph
        return graph

    def compare(self, other) -> None:  # pragma: no cover
        pass

    def set_rib(self, rib) -> None:  # pragma: no cover - not called by the pipeline
        pass

dump(ts)

Return the current per-edge hegemony as a :class:Graph (bidirectional).

Source code in src/quickrib/observers/hegemony.py

def dump(self, ts: datetime) -> Graph:
    """Return the current per-edge hegemony as a :class:`Graph` (bidirectional)."""
    if ts == self._last_dump_ts and self._last_graph is not None:
        return self._last_graph

    per_vp_scores: dict[tuple[str, str], list[float]] = defaultdict(list)
    for vp, ec in self.edge_paths_count.items():
        norm = self.num_total_paths.get(vp, 0.0)
        if norm <= 0:
            continue
        for edge, score in betweenness(ec, norm).items():
            per_vp_scores[edge].append(score)

    hegemony = _trim_across_vps(per_vp_scores, len(self.edge_paths_count), self.alpha)
    graph = _hegemony_graph(hegemony, self.min_hegemony)

    self._last_dump_ts, self._last_graph = ts, graph
    return graph

quickrib.observers.hegemony.RibHegemonyObserver

Bases: Observer

Post-hoc edge hegemony as an observer.

A thin wrapper around :func:edge_hegemony: :meth:dump recomputes the entire edge-hegemony graph from the current RIB (mirroring what examples/hegemony.py does by hand). update_* are no-ops: the observer holds no incremental state, it just needs a reference to the RIBTable. QuickRIB.build_rib hands it one through :meth:set_rib once the table is built; passing rib= to the constructor is only needed outside the pipeline.

Slower than the incremental :class:HegemonyObserver (a full RIB walk per dump) but the two agree when the latter runs with deaggregate=True.

Source code in src/quickrib/observers/hegemony.py

class RibHegemonyObserver(Observer):
    """Post-hoc edge hegemony as an observer.

    A thin wrapper around :func:`edge_hegemony`: :meth:`dump` recomputes the
    entire edge-hegemony graph from the current RIB (mirroring what
    ``examples/hegemony.py`` does by hand). ``update_*`` are no-ops: the
    observer holds no incremental state, it just needs a reference to the
    ``RIBTable``. ``QuickRIB.build_rib`` hands it one through
    :meth:`set_rib` once the table is built; passing ``rib=`` to the constructor
    is only needed outside the pipeline.

    Slower than the incremental :class:`HegemonyObserver` (a full RIB walk per
    dump) but the two agree when the latter runs with ``deaggregate=True``.
    """

    def __init__(
        self,
        rib=None,
        name: str = "hegemony",
        alpha: float = 0.2,
        min_hegemony: float = 0.0,
    ) -> None:
        self.rib = rib
        self.name = name
        self.alpha = alpha
        self.min_hegemony = min_hegemony
        self.last_edge_hegemony: dict[tuple[str, str], float] = {}
        self._last_dump_ts: Optional[datetime] = None
        self._last_graph: Optional[Graph] = None


    def update_rib(self, bgpelem) -> None:
        pass

    def update_announcement(self, bgpelem, data, old_data) -> None:
        pass

    def update_withdrawal(self, bgpelem, data) -> None:
        pass

    def set_rib(self, rib) -> None:
        self.rib = rib

    def compare(self, other) -> None:  # pragma: no cover
        pass

    def dump(self, ts: datetime) -> Graph:
        if ts == self._last_dump_ts and self._last_graph is not None:
            return self._last_graph
        if self.rib is None:
            raise RuntimeError(
                "RibHegemonyObserver has no RIB reference (pass rib= or call set_rib)"
            )

        self.last_edge_hegemony = edge_hegemony(self.rib.data, self.alpha)
        graph = _hegemony_graph(self.last_edge_hegemony, self.min_hegemony)

        self._last_dump_ts, self._last_graph = ts, graph
        return graph

quickrib.observers.hegemony.node_hegemony(rib, alpha=0.2)

Per-ASN hegemony over a reconstructed RIB.

Each ASN is credited once per path (set(as_path)), where :func:edge_hegemony credits each consecutive link and so counts a link twice if a path traverses it twice. Paths that revisit an ASN are the only case where the two differ, and they are rare enough in real data that the asymmetry has never been worth the cost of de-duplicating edges per path.

Source code in src/quickrib/observers/hegemony.py

def node_hegemony(rib: RIBData, alpha=0.2) -> dict[str, float]:
    """Per-ASN hegemony over a reconstructed RIB.

    Each ASN is credited once per path (``set(as_path)``), where
    :func:`edge_hegemony` credits each *consecutive link* and so counts a link
    twice if a path traverses it twice. Paths that revisit an ASN are the only
    case where the two differ, and they are rare enough in real data that the
    asymmetry has never been worth the cost of de-duplicating edges per path.
    """

    node_bcs = defaultdict(list) # mapping ASN to list of their betweenness centrality (for each VP [vp1, vp2, ...])
    n_vps = sum(len(vps) for vps in rib.values())
    for vps in rib.values():
        for pfxs in vps.values():
            covered = radix_size(pfxs)
            if not covered:
                # A VP holding nothing routes nothing: it has no betweenness to
                # contribute, and normalising by its coverage would divide by zero.
                continue
            vp_bcs = defaultdict(float) # mapping ASN to betweenness centrality for a given VP
            score_increment = 1.0 / covered
            for rnode in pfxs.nodes():
                as_path = rnode.data["as-path"]
                weight = radix_prefix_size(pfxs, rnode)
                for asn in set(as_path):
                    vp_bcs[asn] += weight * score_increment
            for node, bc in vp_bcs.items():
                node_bcs[node].append(bc)

    return _trim_across_vps(node_bcs, n_vps, alpha)

quickrib.observers.hegemony.edge_hegemony(rib, alpha=0.2)

Per-link hegemony over a reconstructed RIB. See :func:node_hegemony for how the two treat a path that revisits an ASN.

Source code in src/quickrib/observers/hegemony.py

def edge_hegemony(rib: RIBData, alpha=0.2) -> dict[tuple[str, str], float]:
    """Per-link hegemony over a reconstructed RIB. See :func:`node_hegemony` for
    how the two treat a path that revisits an ASN."""

    edge_bcs = defaultdict(list) # mapping edge to list of its betweenness centrality (for each VP)
    n_vps = sum(len(vps) for vps in rib.values())
    for vps in rib.values():
        for pfxs in vps.values():
            covered = radix_size(pfxs)
            if not covered:
                # See node_hegemony: an empty VP normalises by zero.
                continue
            vp_bcs = defaultdict(float) # mapping edge to betweenness centrality for a given VP
            score_increment = 1.0 / covered
            for rnode in pfxs.nodes():
                weight = radix_prefix_size(pfxs, rnode)
                for edge in _canon_edges(rnode.data["as-path"]):
                    vp_bcs[edge] += weight * score_increment
            for edge, bc in vp_bcs.items():
                edge_bcs[edge].append(bc)

    return _trim_across_vps(edge_bcs, n_vps, alpha)

quickrib.observers.hegemony.trim_mean(data, alpha, force_trim=True)

Mean of data with its most extreme values cut from both ends.

Kept as the single-sample entry point (and the reference the vectorised :func:_trim_across_vps is tested against); it no longer goes through scipy.stats.trim_mean, whose per-call overhead dominated when hegemony called it once per edge.

Source code in src/quickrib/observers/hegemony.py

def trim_mean(data, alpha: float, force_trim: bool = True) -> float:
    """Mean of ``data`` with its most extreme values cut from both ends.

    Kept as the single-sample entry point (and the reference the vectorised
    :func:`_trim_across_vps` is tested against); it no longer goes through
    ``scipy.stats.trim_mean``, whose per-call overhead dominated when hegemony
    called it once per edge.
    """
    values = sorted(data)
    cut = _trim_count(len(values), alpha, force_trim)
    kept = values[cut : len(values) - cut]
    return float(np.mean(kept))

IODA Signal

quickrib.observers.ioda.IODAObserver

Bases: Observer

Incremental IODA signal.

The observer is told which peers are full-feed before the RIB build. QuickRIB.initialize_processing has already made that selection and hands it over through :meth:set_full_feed_peers, so the observer can count how many of them see each prefix directly in update_rib and never needs to read the RIB back. The first dump turns those counts into the visible set and the per-AS trees; from then on the update_* hot path maintains them in O(1) per element and a dump only recomputes the block count of the ASes whose visible set actually changed.

The state is a chain, each level derived from the one above it::

_pfx_peers    prefix -> how many full-feed peers currently see it
_pfx_origins  prefix -> origin ASN -> how many of them report that origin
_visible      the prefixes whose peer count is above the quorum
_trees        origin ASN -> radix holding *its* visible prefixes
_totals       origin ASN -> block count of that radix, i.e. the signal

An update touches the top two directly. Only a change that crosses the quorum, or that adds or drops an origin for an already-visible prefix, reaches _trees, and even then it only marks the AS dirty rather than recounting it. The block count is paid once per dump, not once per message.

Because the criterion is QuickRIB's own, running the pipeline with fullfeed_only=True is not just allowed but cheaper: the peers the observer would ignore never enter the RIB in the first place.

The full-feed population is fixed at that hand-over rather than recomputed every bin as IODA does. Peer table sizes move by a fraction of a percent over a few hours, so the membership does not drift; :meth:full_feed_drift reports it against a RIB if it ever does.

Source code in src/quickrib/observers/ioda.py

class IODAObserver(Observer):
    """Incremental IODA signal.

    The observer is told which peers are full-feed before the RIB build.
    ``QuickRIB.initialize_processing`` has already made that selection and hands
    it over through :meth:`set_full_feed_peers`, so the observer can count how
    many of them see each prefix directly in ``update_rib`` and never needs to
    read the RIB back. The first ``dump`` turns those counts into the visible set and the
    per-AS trees; from then on the ``update_*`` hot path maintains them in O(1)
    per element and a dump only recomputes the block count of the ASes whose
    visible set actually changed.

    The state is a chain, each level derived from the one above it::

        _pfx_peers    prefix -> how many full-feed peers currently see it
        _pfx_origins  prefix -> origin ASN -> how many of them report that origin
        _visible      the prefixes whose peer count is above the quorum
        _trees        origin ASN -> radix holding *its* visible prefixes
        _totals       origin ASN -> block count of that radix, i.e. the signal

    An update touches the top two directly. Only a change that crosses the
    quorum, or that adds or drops an origin for an already-visible prefix,
    reaches ``_trees``, and even then it only marks the AS dirty rather than
    recounting it. The block count is paid once per dump, not once per message.

    Because the criterion is QuickRIB's own, running the pipeline with
    ``fullfeed_only=True`` is not just allowed but cheaper: the peers the
    observer would ignore never enter the RIB in the first place.

    The full-feed population is fixed at that hand-over rather than recomputed
    every bin as IODA does. Peer table sizes move by a fraction of a percent over
    a few hours, so the membership does not drift; :meth:`full_feed_drift`
    reports it against a RIB if it ever does.
    """

    def __init__(
        self,
        ff_peers: Optional[Iterable[VP]] = None,
        *,
        ip_version: Literal[4, 6] = 4,
        visible_ratio: float = VISIBLE_RATIO,
        full_feed_ratio: float = FULL_FEED_RATIO,
        entities: Optional[Iterable[str]] = None,
        step: int = IODA_STEP,
        name: str = "ioda",
    ) -> None:
        self.ip_version: Literal[4, 6] = ip_version
        self.visible_ratio = visible_ratio
        self.full_feed_ratio = full_feed_ratio
        self.entities = set(entities) if entities is not None else None
        self.step = step
        self.name = name
        self.series: dict[int, dict[str, int]] = {}

        self._want_v6 = ip_version == 6
        self.ff_peers: set[VP] = set()
        self._quorum = 0.0
        self._have_peers = False
        self._built = False

        self._pfx_peers: dict[str, int] = {}
        "prefix -> number of full-feed peers seeing it"
        self._pfx_origins: dict[str, dict[str, int]] = {}
        "prefix -> origin ASN -> number of full-feed peers reporting that origin"
        self._visible: set[str] = set()
        "prefixes currently above the quorum: the only ones present in the trees"
        self._trees: dict[str, radix.Radix] = {}
        "origin ASN -> radix of its visible prefixes, and the block count of it"
        self._totals: dict[str, int] = {}
        "origin ASN -> its block count as of the last dump; the emitted signal"
        self._dirty: set[str] = set()
        "ASes whose tree changed since the last dump; only these are recomputed"

        if ff_peers is not None:
            self.set_full_feed_peers(ff_peers)

    # ------------------------------------------------------------------- setup

    def set_full_feed_peers(self, ff_peers: Iterable[VP]) -> None:
        """Adopt the full-feed selection, as ``(collector, peer_asn, peer_ip)``.

        Called by :class:`~quickrib.rib_table.RIBTable` once ``QuickRIB`` has
        made its selection, which is before any element is replayed.
        """
        self.ff_peers = set(ff_peers)
        self._quorum = self.visible_ratio * len(self.ff_peers)
        self._have_peers = True
        logger.info(
            "IODA: %d full-feed peers, an IPv%d prefix is visible above %.1f of them",
            len(self.ff_peers),
            self.ip_version,
            self._quorum,
        )

    def set_rib(self, rib: RIBTable) -> None:  # pragma: no cover - nothing needed
        pass

    def full_feed_drift(self, rib: RIBTable) -> set[VP]:
        """Vantage points whose full-feed status in ``rib`` differs from the pinned set.

        Empty in a well-behaved run. A non-empty result means the window is long
        enough that a per-bin re-evaluation would have picked a different peer
        population than the one pinned before the build.
        """
        current = full_feed_peers(rib.data, ratio=self.full_feed_ratio)
        return current ^ self.ff_peers

    # ---------------------------------------------------------------- hot path

    def _attach(self, origin: str, prefix: str) -> None:
        """Put a now-visible ``prefix`` into ``origin``'s tree and defer the recount."""
        tree = self._trees.get(origin)
        if tree is None:
            tree = self._trees[origin] = radix.Radix()
        tree.add(prefix)
        self._dirty.add(origin)

    def _detach(self, origin: str, prefix: str) -> None:
        """The inverse of :meth:`_attach`; a prefix absent from the tree is not an error."""
        tree = self._trees.get(origin)
        if tree is None:
            return
        try:
            tree.delete(prefix)
        except KeyError:
            return
        self._dirty.add(origin)

    def _set_visible(self, prefix: str, visible: bool) -> None:
        """Move ``prefix`` in or out of every origin AS's tree when it crosses the quorum.

        Most updates leave visibility where it was, since one peer gaining or
        losing a route rarely moves a prefix across half the fleet. The common
        case is therefore the early return, with no tree touched at all.
        """
        if visible == (prefix in self._visible):
            return
        if visible:
            self._visible.add(prefix)
            for origin in self._pfx_origins.get(prefix, ()):
                self._attach(origin, prefix)
        else:
            self._visible.discard(prefix)
            for origin in self._pfx_origins.get(prefix, ()):
                self._detach(origin, prefix)

    def _add_origin(self, prefix: str, origin: str) -> None:
        """Reference-count ``origin`` for ``prefix`` across the full-feed peers.

        The tree only changes on the 0 -> 1 transition, and only while the prefix
        is visible: an origin that some other peer still reports is already filed,
        and an invisible prefix is in no tree to begin with.
        """
        origins = self._pfx_origins.get(prefix)
        if origins is None:
            self._pfx_origins[prefix] = {origin: 1}
            if prefix in self._visible:
                self._attach(origin, prefix)
            return
        count = origins.get(origin, 0)
        origins[origin] = count + 1
        if count == 0 and prefix in self._visible:
            self._attach(origin, prefix)

    def _remove_origin(self, prefix: str, origin: str) -> None:
        """Drop one peer's vote for ``origin``; the last one out clears the tree."""
        origins = self._pfx_origins.get(prefix)
        if origins is None:
            return
        count = origins.get(origin, 0) - 1
        if count > 0:
            origins[origin] = count
            return
        origins.pop(origin, None)
        if not origins:
            self._pfx_origins.pop(prefix, None)
        if prefix in self._visible:
            self._detach(origin, prefix)


    def _tracked(self, bgpelem: AnyElement, prefix: str) -> bool:
        """Whether this element is an IPv4/IPv6 route from a full-feed peer."""
        if (":" in prefix) != self._want_v6:
            return False
        return (
            bgpelem.collector,
            bgpelem.peer_asn,
            bgpelem.peer_address,
        ) in self.ff_peers

    def update_rib(self, bgpelem: ParsedElement) -> None:
        # Counting only: visibility and the trees are derived in one go by
        # _build_trees once the whole dump has been read (see `dump`).
        fields = bgpelem.fields
        prefix = fields["prefix"]
        if not self._tracked(bgpelem, prefix):
            return
        self._pfx_peers[prefix] = self._pfx_peers.get(prefix, 0) + 1
        self._add_origin(prefix, fields["as-path"][-1])

    def update_announcement(
        self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData] = None
    ) -> None:
        fields = bgpelem.fields
        prefix = fields["prefix"]
        if not self._tracked(bgpelem, prefix):
            return

        origin = fields["as-path"][-1]
        if old_data is None:
            # The peer did not have this prefix, so it is one more peer seeing
            # it. This is the only case where the visibility count can rise.
            seen_by = self._pfx_peers.get(prefix, 0) + 1
            self._pfx_peers[prefix] = seen_by
            self._add_origin(prefix, origin)
            if self._built:
                self._set_visible(prefix, seen_by > self._quorum)
        else:
            # A re-announcement: the peer already counted, so only the origin can
            # have moved. A path or community change that keeps the origin is a
            # no-op here, and that is the overwhelming majority of updates.
            old_origin = old_data["as-path"][-1]
            if old_origin != origin:
                self._add_origin(prefix, origin)
                self._remove_origin(prefix, old_origin)

    def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData] = None) -> None:
        # `data` is None when the peer did not have the prefix: nothing counted,
        # so nothing to take back. Otherwise it carries the path being withdrawn,
        # which is the only place the origin to decrement can come from.
        if data is None:
            return
        prefix = bgpelem.fields["prefix"]
        if not self._tracked(bgpelem, prefix):
            return

        seen_by = self._pfx_peers.get(prefix, 0) - 1
        if seen_by > 0:
            self._pfx_peers[prefix] = seen_by
        else:
            # Nobody sees it any more: drop it rather than keep a zero (or a
            # negative, if a withdrawal ever outruns its announcement) for the
            # rest of the run.
            self._pfx_peers.pop(prefix, None)
            seen_by = 0
        if self._built:
            self._set_visible(prefix, seen_by > self._quorum)
        self._remove_origin(prefix, data["as-path"][-1])

    # --------------------------------------------------------------- reporting

    def _build_trees(self) -> None:
        """Turn the counts gathered during the build into the visible set and trees.

        Deferred to the first dump because that is the first moment the counts
        are complete: until the RIB dump has been read in full, a prefix's peer
        count is still climbing and testing it against the quorum would be
        meaningless.
        """
        if not self._have_peers:
            raise RuntimeError(
                "IODAObserver does not know the full-feed peers (pass ff_peers=, "
                "call set_full_feed_peers, or run it through the QuickRIB pipeline)"
            )
        for prefix, seen_by in self._pfx_peers.items():
            if seen_by > self._quorum:
                self._visible.add(prefix)
                for origin in self._pfx_origins[prefix]:
                    self._attach(origin, prefix)
        self._built = True

    def dump(self, ts: datetime.datetime) -> dict[str, int]:
        if not self._built:
            self._build_trees()

        # Only the ASes whose tree moved since the last dump are recounted; the
        # rest keep the total they already had. An AS also keeps its slot once
        # seen, so a total falling to zero is reported as a zero rather than
        # vanishing. A zero is what a complete blackout looks like in the series.
        for origin in self._dirty:
            self._totals[origin] = radix_block_count(self._trees[origin])
        self._dirty.clear()

        if self.entities is None:
            totals = dict(self._totals)
        else:
            totals = {asn: self._totals.get(asn, 0) for asn in self.entities}
        self.series[_bin_of(ts, self.step)] = totals
        return totals

    def compare(self, other) -> None:  # pragma: no cover
        pass

set_full_feed_peers(ff_peers)

Adopt the full-feed selection, as (collector, peer_asn, peer_ip).

Called by :class:~quickrib.rib_table.RIBTable once QuickRIB has made its selection, which is before any element is replayed.

Source code in src/quickrib/observers/ioda.py

def set_full_feed_peers(self, ff_peers: Iterable[VP]) -> None:
    """Adopt the full-feed selection, as ``(collector, peer_asn, peer_ip)``.

    Called by :class:`~quickrib.rib_table.RIBTable` once ``QuickRIB`` has
    made its selection, which is before any element is replayed.
    """
    self.ff_peers = set(ff_peers)
    self._quorum = self.visible_ratio * len(self.ff_peers)
    self._have_peers = True
    logger.info(
        "IODA: %d full-feed peers, an IPv%d prefix is visible above %.1f of them",
        len(self.ff_peers),
        self.ip_version,
        self._quorum,
    )

full_feed_drift(rib)

Vantage points whose full-feed status in rib differs from the pinned set.

Empty in a well-behaved run. A non-empty result means the window is long enough that a per-bin re-evaluation would have picked a different peer population than the one pinned before the build.

Source code in src/quickrib/observers/ioda.py

def full_feed_drift(self, rib: RIBTable) -> set[VP]:
    """Vantage points whose full-feed status in ``rib`` differs from the pinned set.

    Empty in a well-behaved run. A non-empty result means the window is long
    enough that a per-bin re-evaluation would have picked a different peer
    population than the one pinned before the build.
    """
    current = full_feed_peers(rib.data, ratio=self.full_feed_ratio)
    return current ^ self.ff_peers

quickrib.observers.ioda.RibIODAObserver

Bases: Observer

Post-hoc IODA signal: recompute :func:visible_slash24s at every dump.

The reference observer. It rescans the whole RIB each time, which is O(peers x prefixes) per dump: correct, but far too slow to run at IODA's 5-minute cadence over a long window. Use :class:IODAObserver for that.

Source code in src/quickrib/observers/ioda.py

class RibIODAObserver(Observer):
    """Post-hoc IODA signal: recompute :func:`visible_slash24s` at every dump.

    The reference observer. It rescans the whole RIB each time, which is
    O(peers x prefixes) per dump: correct, but far too slow to run at IODA's
    5-minute cadence over a long window. Use :class:`IODAObserver` for that.
    """

    def __init__(
        self,
        rib: Optional[RIBTable] = None,
        *,
        ip_version: Literal[4, 6] = 4,
        ff_peers: Optional[set[VP]] = None,
        visible_ratio: float = VISIBLE_RATIO,
        full_feed_ratio: float = FULL_FEED_RATIO,
        entities: Optional[Iterable[str]] = None,
        step: int = IODA_STEP,
        name: str = "ioda_rib",
    ) -> None:
        self.rib = rib
        self.ip_version: Literal[4, 6] = ip_version
        self.ff_peers = ff_peers
        self.visible_ratio = visible_ratio
        self.full_feed_ratio = full_feed_ratio
        self.entities = set(entities) if entities is not None else None
        self.step = step
        self.name = name
        self.series: dict[int, dict[str, int]] = {}


    def update_rib(self, bgpelem: ParsedElement) -> None:
        pass

    def update_announcement(
        self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData] = None
    ) -> None:
        pass

    def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData] = None) -> None:
        pass

    def set_rib(self, rib: RIBTable) -> None:
        self.rib = rib

    def set_full_feed_peers(self, ff_peers: set[VP]) -> None:
        """Receive the pipeline's full-feed selection (see :class:`IODAObserver`)."""
        self.ff_peers = set(ff_peers)

    def dump(self, ts: datetime.datetime) -> dict[str, int]:
        if self.rib is None:
            raise RuntimeError(
                "RibIODAObserver has no RIB reference (pass rib= or call set_rib)"
            )
        totals = visible_slash24s(
            self.rib.data,
            ip_version=self.ip_version,
            ff_peers=self.ff_peers,
            visible_ratio=self.visible_ratio,
            full_feed_ratio=self.full_feed_ratio,
        )
        if self.entities is not None:
            totals = {asn: totals.get(asn, 0) for asn in self.entities}
        self.series[_bin_of(ts, self.step)] = totals
        return totals

    def compare(self, other) -> None:  # pragma: no cover
        pass

set_full_feed_peers(ff_peers)

Receive the pipeline's full-feed selection (see :class:IODAObserver).

Source code in src/quickrib/observers/ioda.py

def set_full_feed_peers(self, ff_peers: set[VP]) -> None:
    """Receive the pipeline's full-feed selection (see :class:`IODAObserver`)."""
    self.ff_peers = set(ff_peers)

quickrib.observers.ioda.visible_slash24s(rib_data, *, ip_version=4, ff_peers=None, visible_ratio=VISIBLE_RATIO, full_feed_ratio=FULL_FEED_RATIO)

The IODA BGP signal: origin ASN -> number of visible /24s (or /48s).

ff_peers pins the full-feed population instead of deriving it from rib_data; :class:IODAObserver uses that to keep the two implementations comparable at a shared instant.

Source code in src/quickrib/observers/ioda.py

def visible_slash24s(
    rib_data: RIBData,
    *,
    ip_version: Literal[4, 6] = 4,
    ff_peers: Optional[set[VP]] = None,
    visible_ratio: float = VISIBLE_RATIO,
    full_feed_ratio: float = FULL_FEED_RATIO,
) -> dict[str, int]:
    """The IODA BGP signal: origin ASN -> number of visible /24s (or /48s).

    ``ff_peers`` pins the full-feed population instead of deriving it from
    ``rib_data``; :class:`IODAObserver` uses that to keep the two
    implementations comparable at a shared instant.
    """
    if ff_peers is None:
        ff_peers = full_feed_peers(rib_data, ratio=full_feed_ratio)
    quorum = visible_ratio * len(ff_peers)
    want_v6 = ip_version == 6

    # Pass 1: how many full-feed peers see each prefix, and who originates it.
    pfx_peers: dict[str, int] = defaultdict(int)
    pfx_origins: dict[str, set[str]] = defaultdict(set)
    for collector, peers in rib_data.items():
        for (peer_asn, peer_ip), peer_table in peers.items():
            if (collector, peer_asn, peer_ip) not in ff_peers:
                continue
            for rnode in peer_table.nodes():
                prefix = rnode.prefix
                if (":" in prefix) != want_v6:
                    continue
                pfx_peers[prefix] += 1
                pfx_origins[prefix].add(rnode.data["as-path"][-1])

    # Pass 2: keep the prefixes above the quorum, filed under every origin seen
    # for them. Under MOAS each origin is credited the whole prefix.
    trees: dict[str, radix.Radix] = defaultdict(radix.Radix)
    for prefix, seen_by in pfx_peers.items():
        if seen_by > quorum:
            for origin in pfx_origins[prefix]:
                trees[origin].add(prefix)

    # Pass 3: each AS's signal is the union of its visible prefixes, in blocks.
    return {origin: radix_block_count(tree) for origin, tree in trees.items()}

quickrib.observers.ioda.full_feed_peers(rib_data, *, ratio=FULL_FEED_RATIO)

The full-feed vantage points of rib_data, by QuickRIB's rule.

A peer is full-feed if it carries more than ratio times the number of unique prefixes its collector saw, across both address families. This is the same computation QuickRIB.initialize_processing performs on the RIB-dump stream. Deriving it here is the post-hoc path; a pipeline run hands the already-computed selection to the observer instead.

Source code in src/quickrib/observers/ioda.py

def full_feed_peers(
    rib_data: RIBData, *, ratio: float = FULL_FEED_RATIO
) -> set[VP]:
    """The full-feed vantage points of ``rib_data``, by QuickRIB's rule.

    A peer is full-feed if it carries more than ``ratio`` times the number of
    unique prefixes its collector saw, across both address families. This is the
    same computation ``QuickRIB.initialize_processing`` performs on the RIB-dump
    stream. Deriving it here is the post-hoc path; a pipeline run hands the
    already-computed selection to the observer instead.
    """
    full_feed: set[VP] = set()
    for collector, peers in rib_data.items():
        unique: set[str] = set()
        sizes: dict[tuple[int, str], int] = {}
        for peer, peer_table in peers.items():
            prefixes = peer_table.prefixes()
            sizes[peer] = len(prefixes)
            unique.update(prefixes)
        threshold = ratio * len(unique)
        for (peer_asn, peer_ip), size in sizes.items():
            if size > threshold:
                full_feed.add((collector, peer_asn, peer_ip))
    return full_feed

Outage detection

quickrib.observers.ioda.evaluate_series(series, *, step=IODA_STEP, history=ALERT_HISTORY, threshold=ALERT_THRESHOLD)

Score one entity's time series bin by bin against its own recent past.

Returns one record per bin that has a full history behind it, with the value, the historyValue (the median baseline) and the resulting level. Bins without enough history are skipped, exactly as IODA cannot alert before it has a baseline.

A bin that is itself alerting does not feed the baseline: otherwise a long outage erodes its own median and silently declares itself normal while still down. Matching the API's historyValue on real series is what settles this; tests/test_ioda.py pins the agreement.

Source code in src/quickrib/observers/ioda.py

def evaluate_series(
    series: Mapping[int, int],
    *,
    step: int = IODA_STEP,
    history: int = ALERT_HISTORY,
    threshold: float = ALERT_THRESHOLD,
) -> list[AlertRecord]:
    """Score one entity's time series bin by bin against its own recent past.

    Returns one record per bin that has a **full** ``history`` behind it, with
    the ``value``, the ``historyValue`` (the median baseline) and the resulting
    ``level``. Bins without enough history are skipped, exactly as IODA cannot
    alert before it has a baseline.

    A bin that is itself alerting does **not** feed the baseline: otherwise a
    long outage erodes its own median and silently declares itself normal while
    still down. Matching the API's ``historyValue`` on real series is what
    settles this; ``tests/test_ioda.py`` pins the agreement.
    """
    window = history // step
    if window <= 0:
        raise ValueError(
            f"history ({history}s) must be at least one step ({step}s) long; "
            "a zero-length baseline would score every bin against the whole series"
        )
    baseline_values: list[int] = []
    records: list[AlertRecord] = []
    # Carried across bins: while the level is critical the value is withheld from
    # the baseline, so an outage cannot lower the bar it is being judged against.
    level = "normal"
    for ts in sorted(series):
        value = series[ts]
        if len(baseline_values) >= window:
            baseline = median(baseline_values[-window:])
            alerting = baseline > 0 and value < threshold * baseline
            level = "critical" if alerting else "normal"
            records.append(
                {
                    "time": ts,
                    "value": value,
                    "historyValue": baseline,
                    "level": level,
                    "condition": f"< {threshold:g}" if alerting else "normal",
                    "method": "median",
                }
            )
        if level == "normal":
            baseline_values.append(value)
            # Trim in batches rather than per bin: only the last `window` values
            # are ever read, and popping from the front of a list is O(n).
            if len(baseline_values) > 2 * window:
                del baseline_values[:-window]
    return records

quickrib.observers.ioda.detect_alerts(series, **kwargs)

IODA-shaped alerts for one entity: the bins where its level changes.

Mirrors /v2/outages/alerts, which reports transitions rather than every bin. The first evaluated bin is reported only if it is already alerting.

Source code in src/quickrib/observers/ioda.py

def detect_alerts(series: Mapping[int, int], **kwargs) -> list[AlertRecord]:
    """IODA-shaped alerts for one entity: the bins where its level *changes*.

    Mirrors ``/v2/outages/alerts``, which reports transitions rather than every
    bin. The first evaluated bin is reported only if it is already alerting.
    """
    alerts: list[AlertRecord] = []
    level = "normal"
    for record in evaluate_series(series, **kwargs):
        if record["level"] != level:
            level = record["level"]
            alerts.append(record)
    return alerts

quickrib.observers.ioda.detect_events(series, *, location='', location_name='', datasource='bgp', step=IODA_STEP, **kwargs)

IODA-shaped outage events for one entity, as /v2/outages/events returns them.

An event spans from a bin turning critical to the bin that turns normal again; one still alerting at the end of the series is closed on its last bin. score follows IODA's formula, 500 * sum(1 - value / historyValue) over the event's bins.

Source code in src/quickrib/observers/ioda.py

def detect_events(
    series: Mapping[int, int],
    *,
    location: str = "",
    location_name: str = "",
    datasource: str = "bgp",
    step: int = IODA_STEP,
    **kwargs,
) -> list[OutageEvent]:
    """IODA-shaped outage events for one entity, as ``/v2/outages/events`` returns them.

    An event spans from a bin turning ``critical`` to the bin that turns
    ``normal`` again; one still alerting at the end of the series is closed on
    its last bin. ``score`` follows IODA's formula, ``500 * sum(1 - value /
    historyValue)`` over the event's bins.
    """
    records = evaluate_series(series, step=step, **kwargs)
    events: list[OutageEvent] = []
    start: Optional[int] = None  # None while no event is open
    drop = 0.0  # running sum of (1 - value / historyValue) over the open event

    def close(start: int, drop: float, end: int) -> None:
        events.append(
            {
                "location": location,
                "start": start,
                "duration": end - start,
                "uncertainty": None,
                "method": "median",
                "datasource": datasource,
                "status": 0,
                "fraction": None,
                "score": SCORE_SCALE * drop,
                "location_name": location_name,
                "overlaps_window": False,
            }
        )

    for record in records:
        if record["level"] == "critical":
            if start is None:
                start, drop = record["time"], 0.0
            drop += 1 - record["value"] / record["historyValue"]
        elif start is not None:
            close(start, drop, record["time"])
            start = None
    # An event still alerting when the series runs out is closed on the bin after
    # the last one, so its duration covers every bin it actually spans.
    if start is not None and records:
        close(start, drop, records[-1]["time"] + step)
    return events

BGPlay

quickrib.observers.bgplay.BGPlayObserver

Bases: Observer

Collect the BGPlay view of one resource while the pipeline replays BGP.

Parameters:

Name	Type	Description	Default
resource	str | Iterable[str]	Prefix, IP address, ASxxx / bare AS number, or a comma-separated list of those, following the API's resource grammar. A prefix matches exactly (querying 8.0.0.0/9 does not pull in its more specifics, matching the API); an AS matches every path whose origin it is, so a path towards one of its prefixes that originates elsewhere is excluded. Query that prefix to see the hijack.	required
start_time	datetime	The query_starttime. It is also the boundary between the two halves of the output: an A/W at or before it folds into initial_state, one after it becomes an event. RIB-dump entries always fold into initial_state, whatever their timestamp, since they describe the table rather than a change to it.	required
end_time	Optional[datetime]	The query_endtime, reported as-is. Updates past it are still recorded; bound the replay with QuickRIB's own end_time.	None
resolve_owners	bool	Fill nodes[].owner from the RIPEstat as-names data call when serialising. Off by default, since it is a network round-trip and the rest of the payload is computed entirely offline.	False

Notes

The hot path is one set lookup on the prefix plus, in AS mode, one on the path origin, and nothing is allocated before an element matches. All the formatting (ISO timestamps, integer paths, community filtering, sorting) happens once in :meth:to_dict.

A withdrawal carries no AS path, so it can only be matched by the prefix filter. In pure AS mode the payload therefore has no W events at all, which is exactly what the API returns.

Source code in src/quickrib/observers/bgplay.py

class BGPlayObserver(Observer):
    """Collect the BGPlay view of one resource while the pipeline replays BGP.

    Parameters
    ----------
    resource
        Prefix, IP address, ``ASxxx`` / bare AS number, or a comma-separated
        list of those, following the API's ``resource`` grammar.
        A prefix matches *exactly* (querying ``8.0.0.0/9`` does not pull in
        its more specifics, matching the API); an AS matches every path whose
        **origin** it is, so a path towards one of its prefixes that originates
        elsewhere is excluded. Query that prefix to see the hijack.
    start_time
        The ``query_starttime``. It is also the boundary between the two halves
        of the output: an ``A``/``W`` at or before it folds into
        ``initial_state``, one after it becomes an ``event``. RIB-dump entries
        always fold into ``initial_state``, whatever their timestamp, since they
        describe the table rather than a change to it.
    end_time
        The ``query_endtime``, reported as-is. Updates past it are still
        recorded; bound the replay with ``QuickRIB``'s own ``end_time``.
    resolve_owners
        Fill ``nodes[].owner`` from the RIPEstat ``as-names`` data call when
        serialising. Off by default, since it is a network round-trip and the
        rest of the payload is computed entirely offline.

    Notes
    -----
    The hot path is one set lookup on the prefix plus, in AS mode, one on the
    path origin, and nothing is allocated before an element matches. All the
    formatting (ISO timestamps, integer paths, community filtering, sorting)
    happens once in :meth:`to_dict`.

    A withdrawal carries no AS path, so it can only be matched by the prefix
    filter. In pure AS mode the payload therefore has no ``W`` events at all,
    which is exactly what the API returns.
    """

    def __init__(
        self,
        resource: str | Iterable[str],
        start_time: datetime.datetime,
        end_time: Optional[datetime.datetime] = None,
        name: str = "bgplay",
        resolve_owners: bool = False,
    ) -> None:
        self.name = name
        self.resolve_owners = resolve_owners

        self.resource, self._prefix_filter, self._origin_filter = _parse_resource(
            resource
        )
        self.start_time = start_time
        self.end_time = end_time
        self._start_ts = start_time.timestamp()

        # Prefixes actually observed for the resource: the API's `targets`.
        self._targets: set[str] = set()
        # (source_id, prefix) -> (path, communities); the API's `initial_state`.
        self._initial: dict[tuple[str, str], tuple[list[str], list[str]]] = {}
        # (time, type, source_id, prefix, path, communities); the API's `events`.
        self._events: list[tuple[float, str, str, str, list[str], list[str]]] = []
        # source_id -> (peer_asn, peer_ip), for the API's `sources`.
        self._sources: dict[str, tuple[int, str]] = {}

    # ------------------------------------------------------------------ hot path

    def _record(self, bgpelem: AnyElement, pfx: str) -> None:
        """Route a matched element into ``initial_state`` or ``events``."""
        sid = source_id(bgpelem.collector, bgpelem.peer_address)
        self._sources[sid] = (bgpelem.peer_asn, bgpelem.peer_address)
        etype = bgpelem.type

        # Discriminate on `type`, never on the Python class: `ParsedElement` and
        # `WithdrawalElement` are static narrowings of the *same*
        # `pybgpflux.BGPElement` the pipeline yields, so an isinstance check
        # against them is false for every real element. A withdrawal's `fields`
        # then carries only the prefix, since no parser puts an `as-path` on a
        # withdrawal, and reading the path off it raises.
        if bgpelem.type == "W":
            path: list[str] = []
            communities: list[str] = []
        else:
            fields = bgpelem.fields
            path = fields["as-path"]
            communities = fields["communities"]

        # A RIB-dump entry is the table, not a change to it, so it always lands
        # in initial_state even when its dump timestamp is past start_time.
        if etype == "R" or bgpelem.time <= self._start_ts:
            if etype == "W":
                self._initial.pop((sid, pfx), None)
            else:
                self._initial[(sid, pfx)] = (path, communities)
            return

        self._events.append((bgpelem.time, etype, sid, pfx, path, communities))

    def _match(self, fields: ParsedFields) -> Optional[str]:
        """Return the matched prefix, or ``None`` if the element is unrelated.

        A prefix resource matches on the prefix alone; an AS resource matches on
        the origin of this very path, so the same prefix can match one
        announcement and not the next.
        """
        pfx = fields["prefix"]
        if pfx in self._prefix_filter:
            return pfx
        if self._origin_filter:
            path = fields["as-path"]
            if path and path[-1] in self._origin_filter:
                return pfx
        return None

    def update_rib(self, bgpelem: ParsedElement) -> None:
        pfx = self._match(bgpelem.fields)
        if pfx is not None:
            self._targets.add(pfx)
            self._record(bgpelem, pfx)

    def update_announcement(
        self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData] = None
    ) -> None:
        pfx = self._match(bgpelem.fields)
        if pfx is not None:
            self._targets.add(pfx)
            self._record(bgpelem, pfx)

    def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData] = None) -> None:
        # A withdrawal carries no path, so only the prefix filter can match it.
        pfx = bgpelem.fields["prefix"]
        if pfx in self._prefix_filter:
            self._targets.add(pfx)
            self._record(bgpelem, pfx)


    def set_rib(self, rib: RIBTable) -> None:  # pragma: no cover
        pass

    # ---------------------------------------------------------------- rendering

    def _sorted_events(self) -> list[EventRecord]:
        """Events in timestamp order.

        Nothing is de-duplicated. A repeated announcement is real and the API
        reports it twice, so the observer must too. This used to collapse exact
        repeats inside the window where QuickRIB's build and update streams
        overlapped. That overlap is gone: `update_rib` starts strictly after
        each collector's dump instant, so an element reaches an observer once.
        """
        return sorted(self._events, key=lambda event: event[0])

    def as_numbers(self, events: Optional[list[EventRecord]] = None) -> set[int | str]:
        """Every ASN on an ``initial_state`` or announcement path.

        ``events`` lets :meth:`to_dict` pass the list it has already sorted
        rather than have it sorted again.
        """
        if events is None:
            events = self._sorted_events()
        asns: set[int | str] = set()
        for path, _ in self._initial.values():
            asns.update(_int_path(path))
        for _, etype, _, _, path, _ in events:
            if etype != "W":
                asns.update(_int_path(path))
        return asns

    def to_dict(self, resolve_owners: Optional[bool] = None) -> Payload:
        """Render the collected data as the API's ``data`` object."""
        if resolve_owners is None:
            resolve_owners = self.resolve_owners

        # One key cache for both `initial_state` and `targets`: parsing a prefix
        # is far more expensive than the sort itself.
        sort_keys = {pfx: prefix_sort_key(pfx) for pfx in self._targets}
        initial_state = [
            {
                "target_prefix": pfx,
                "source_id": sid,
                "path": _int_path(path),
                "community": _standard_communities(communities),
            }
            for (sid, pfx), (path, communities) in sorted(
                self._initial.items(), key=lambda item: (item[0][0], sort_keys[item[0][1]])
            )
        ]

        events = []
        # Stable sort on time keeps same-second updates in stream order, the way
        # the API keeps them in arrival order.
        ordered = self._sorted_events()
        for seq, (ts, etype, sid, pfx, path, communities) in enumerate(ordered):
            attrs: Payload = {"source_id": sid, "target_prefix": pfx}
            if etype != "W":
                attrs["path"] = _int_path(path)
                attrs["community"] = _standard_communities(communities)
            events.append(
                {
                    "seq": seq,
                    "timestamp": _iso(ts),
                    "type": etype,
                    "attrs": attrs,
                }
            )

        asns = sorted(self.as_numbers(ordered))
        owners = fetch_as_owners(asns) if (resolve_owners and asns) else {}

        return {
            "resource": self.resource,
            "query_starttime": _iso(self.start_time),
            "query_endtime": _iso(self.end_time) if self.end_time else None,
            "initial_state": initial_state,
            "events": events,
            "nodes": [
                {"as_number": asn, "owner": owners.get(asn, "") if isinstance(asn, int) else ""}
                for asn in asns
            ],
            "sources": [
                {
                    "id": sid,
                    "as_number": self._sources[sid][0],
                    "ip": self._sources[sid][1],
                    "rrc": _rrc_of(sid),
                }
                for sid in sorted(self._sources)
            ],
            "targets": [
                {"prefix": pfx} for pfx in sorted(self._targets, key=lambda p: sort_keys[p])
            ],
        }

    def dump(self, ts: datetime.datetime) -> Payload:
        """Return the BGPlay payload accumulated so far.

        Unlike a windowed observer this is cumulative: ``initial_state`` is the
        table at ``start_time`` and ``events`` grows for the whole run, which is
        exactly what one API response holds. Use :meth:`write_json` to persist
        it.
        """
        return self.to_dict()

    def compare(self, other: Observer | Payload) -> None:
        """Log how this payload differs from ``other`` (another observer or a payload)."""
        if isinstance(other, BGPlayObserver):
            theirs: Payload = other.to_dict()
        else:
            assert isinstance(other, dict), "compare() takes a BGPlayObserver or a payload"
            theirs = other
        report = diff_bgplay(self.to_dict(), theirs)
        for line in format_diff(report):
            logger.info(line)

as_numbers(events=None)

Every ASN on an initial_state or announcement path.

events lets :meth:to_dict pass the list it has already sorted rather than have it sorted again.

Source code in src/quickrib/observers/bgplay.py

def as_numbers(self, events: Optional[list[EventRecord]] = None) -> set[int | str]:
    """Every ASN on an ``initial_state`` or announcement path.

    ``events`` lets :meth:`to_dict` pass the list it has already sorted
    rather than have it sorted again.
    """
    if events is None:
        events = self._sorted_events()
    asns: set[int | str] = set()
    for path, _ in self._initial.values():
        asns.update(_int_path(path))
    for _, etype, _, _, path, _ in events:
        if etype != "W":
            asns.update(_int_path(path))
    return asns

to_dict(resolve_owners=None)

Render the collected data as the API's data object.

Source code in src/quickrib/observers/bgplay.py

def to_dict(self, resolve_owners: Optional[bool] = None) -> Payload:
    """Render the collected data as the API's ``data`` object."""
    if resolve_owners is None:
        resolve_owners = self.resolve_owners

    # One key cache for both `initial_state` and `targets`: parsing a prefix
    # is far more expensive than the sort itself.
    sort_keys = {pfx: prefix_sort_key(pfx) for pfx in self._targets}
    initial_state = [
        {
            "target_prefix": pfx,
            "source_id": sid,
            "path": _int_path(path),
            "community": _standard_communities(communities),
        }
        for (sid, pfx), (path, communities) in sorted(
            self._initial.items(), key=lambda item: (item[0][0], sort_keys[item[0][1]])
        )
    ]

    events = []
    # Stable sort on time keeps same-second updates in stream order, the way
    # the API keeps them in arrival order.
    ordered = self._sorted_events()
    for seq, (ts, etype, sid, pfx, path, communities) in enumerate(ordered):
        attrs: Payload = {"source_id": sid, "target_prefix": pfx}
        if etype != "W":
            attrs["path"] = _int_path(path)
            attrs["community"] = _standard_communities(communities)
        events.append(
            {
                "seq": seq,
                "timestamp": _iso(ts),
                "type": etype,
                "attrs": attrs,
            }
        )

    asns = sorted(self.as_numbers(ordered))
    owners = fetch_as_owners(asns) if (resolve_owners and asns) else {}

    return {
        "resource": self.resource,
        "query_starttime": _iso(self.start_time),
        "query_endtime": _iso(self.end_time) if self.end_time else None,
        "initial_state": initial_state,
        "events": events,
        "nodes": [
            {"as_number": asn, "owner": owners.get(asn, "") if isinstance(asn, int) else ""}
            for asn in asns
        ],
        "sources": [
            {
                "id": sid,
                "as_number": self._sources[sid][0],
                "ip": self._sources[sid][1],
                "rrc": _rrc_of(sid),
            }
            for sid in sorted(self._sources)
        ],
        "targets": [
            {"prefix": pfx} for pfx in sorted(self._targets, key=lambda p: sort_keys[p])
        ],
    }

dump(ts)

Return the BGPlay payload accumulated so far.

Unlike a windowed observer this is cumulative: initial_state is the table at start_time and events grows for the whole run, which is exactly what one API response holds. Use :meth:write_json to persist it.

Source code in src/quickrib/observers/bgplay.py

def dump(self, ts: datetime.datetime) -> Payload:
    """Return the BGPlay payload accumulated so far.

    Unlike a windowed observer this is cumulative: ``initial_state`` is the
    table at ``start_time`` and ``events`` grows for the whole run, which is
    exactly what one API response holds. Use :meth:`write_json` to persist
    it.
    """
    return self.to_dict()

compare(other)

Log how this payload differs from other (another observer or a payload).

Source code in src/quickrib/observers/bgplay.py

def compare(self, other: Observer | Payload) -> None:
    """Log how this payload differs from ``other`` (another observer or a payload)."""
    if isinstance(other, BGPlayObserver):
        theirs: Payload = other.to_dict()
    else:
        assert isinstance(other, dict), "compare() takes a BGPlayObserver or a payload"
        theirs = other
    report = diff_bgplay(self.to_dict(), theirs)
    for line in format_diff(report):
        logger.info(line)

quickrib.observers.bgplay.diff_bgplay(ours, theirs)

Compare two BGPlay data objects, ours against a reference.

seq and nodes[].owner are ignored (see the module docstring), and each section is compared as a multiset, so a lost or invented duplicate event shows up instead of collapsing. The same peer re-announcing the same path in the same second is real, and the API does report it twice. Communities are compared as sets, since the two sides may order them differently.

Returns, per section, only_ours / only_theirs (Counter of keys), mismatched (a set of keys present on both sides with differing content) and a jaccard similarity. Event ordering within one second is not compared; see the module docstring.

Source code in src/quickrib/observers/bgplay.py

def diff_bgplay(ours: Payload, theirs: Payload) -> Payload:
    """Compare two BGPlay ``data`` objects, ours against a reference.

    ``seq`` and ``nodes[].owner`` are ignored (see the module docstring), and
    each section is compared as a **multiset**, so a lost or invented duplicate
    event shows up instead of collapsing. The same peer re-announcing the same
    path in the same second is real, and the API does report it twice.
    Communities are compared as sets, since the two sides may order them
    differently.

    Returns, per section, ``only_ours`` / ``only_theirs`` (``Counter`` of keys),
    ``mismatched`` (a set of keys present on both sides with differing content)
    and a ``jaccard`` similarity. Event ordering *within* one second is not
    compared; see the module docstring.
    """

    def _compare(ours_keys: Counter[Any], theirs_keys: Counter[Any]) -> Payload:
        union = (ours_keys | theirs_keys).total()
        return {
            "only_ours": ours_keys - theirs_keys,
            "only_theirs": theirs_keys - ours_keys,
            "jaccard": (ours_keys & theirs_keys).total() / union if union else 1.0,
        }

    our_state = {_state_key(entry): entry for entry in ours["initial_state"]}
    their_state = {_state_key(entry): entry for entry in theirs["initial_state"]}
    state = _compare(Counter(our_state.keys()), Counter(their_state.keys()))
    state["mismatched"] = {
        key
        for key in set(our_state) & set(their_state)
        if our_state[key]["path"] != their_state[key]["path"]
        or set(our_state[key]["community"]) != set(their_state[key]["community"])
    }

    events = _compare(
        Counter(_event_key(event) for event in ours["events"]),
        Counter(_event_key(event) for event in theirs["events"]),
    )
    events["mismatched"] = set()

    report = {"initial_state": state, "events": events}
    for section, key in (
        ("nodes", "as_number"),
        ("sources", "id"),
        ("targets", "prefix"),
    ):
        entry = _compare(
            Counter(item[key] for item in ours[section]),
            Counter(item[key] for item in theirs[section]),
        )
        entry["mismatched"] = set()
        report[section] = entry
    return report

quickrib.observers.bgplay.is_equivalent(report)

True when :func:diff_bgplay found no difference at all.

Source code in src/quickrib/observers/bgplay.py

def is_equivalent(report: Payload) -> bool:
    """True when :func:`diff_bgplay` found no difference at all."""
    return all(
        not entry["only_ours"] and not entry["only_theirs"] and not entry["mismatched"]
        for entry in report.values()
    )

quickrib.observers.bgplay.format_diff(report)

Render :func:diff_bgplay output as human-readable lines.

Source code in src/quickrib/observers/bgplay.py

def format_diff(report: Payload) -> list[str]:
    """Render :func:`diff_bgplay` output as human-readable lines."""
    lines = []
    for section, entry in report.items():
        lines.append(
            f"{section}: jaccard={entry['jaccard']:.4f} "
            f"only_ours={entry['only_ours'].total()} "
            f"only_theirs={entry['only_theirs'].total()} "
            f"mismatched={len(entry['mismatched'])}"
        )
    return lines

Update Classification

quickrib.observers.update_tagger.UpdateTagFlags

Bases: IntFlag

Non-exclusive BGP update tags. One update can have multiple flags simultaneously.

Source code in src/quickrib/observers/update_tagger.py

class UpdateTagFlags(IntFlag):
    """
    Non-exclusive BGP update tags. 
    One update can have multiple flags simultaneously.
    """
    NONE = 0
    UPDATE_MESSAGE = auto()

    # Change size
    REMOVE_PREFIX = auto() #
    NEW_PREFIX = auto() #
    # Update entry -> AS Path
    ORIGIN_CHANGE = auto() #
    TRANSIT_CHANGE = auto()
    PATH_SWITCHING = auto() #
    PREPENDING_ADD = auto() # 
    PREPENDING_CHANGE = auto() #
    PREPENDING_REMOVE = auto() #
    # Update entry -> Other attributes
    COMMUNITY_CHANGE = auto() #
    OTHER_ATTRIBUTE_CHANGE = auto()
    # No change
    DUPLICATE_WITHDRAWAL = auto()
    DUPLICATE_ANNOUNCE = auto()

quickrib.observers.update_tagger.UpdateTagger

Bases: Observer

Tags every update with the BLT classification flags it satisfies.

The tags are returned, not printed: a subclass such as :class:UpdateTagsCounter accumulates them, and the calling program decides what to do with the result. Set log_tags to trace individual messages through logging.DEBUG. That is one record per BGP message, so it is off by default and belongs on a narrow filter, not a full replay.

Source code in src/quickrib/observers/update_tagger.py

class UpdateTagger(Observer):
    """Tags every update with the BLT classification flags it satisfies.

    The tags are *returned*, not printed: a subclass such as
    :class:`UpdateTagsCounter` accumulates them, and the calling program decides
    what to do with the result. Set ``log_tags`` to trace individual messages
    through ``logging.DEBUG``. That is one record per BGP message, so it is off
    by default and belongs on a narrow filter, not a full replay.
    """

    def __init__(self, name: str = "update_tagger", log_tags: bool = False):
        self.name = name
        self.log_tags = log_tags

    def _log(self, bgpelem, flags: "UpdateTagFlags") -> None:
        if self.log_tags:
            logger.debug("%s %s", bgpelem, self._get_tag_names(flags))


    @staticmethod
    def _get_tag_names(flags: UpdateTagFlags) -> str:
        active_tags = [
            tag.name
            for tag in UpdateTagFlags
            if tag & flags and tag.name is not None and tag.name != "NONE"
        ]
        return " | ".join(active_tags) if active_tags else "NONE"

    def update_rib(self, bgpelem: ParsedElement):
        pass

    def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData]) -> UpdateTagFlags:
        if data:
            flags = UpdateTagFlags.REMOVE_PREFIX | UpdateTagFlags.UPDATE_MESSAGE
        else:
            flags = UpdateTagFlags.DUPLICATE_WITHDRAWAL | UpdateTagFlags.UPDATE_MESSAGE
        self._log(bgpelem, flags)
        return flags


    def update_announcement(self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData]) -> UpdateTagFlags:
        """
        Calculates non-exclusive tags for an announcement.
        """
        # Handle New Prefix (Exclusive to other logic for efficiency)
        if not old_data:
            flags = UpdateTagFlags.NEW_PREFIX | UpdateTagFlags.UPDATE_MESSAGE
            self._log(bgpelem, flags)
            return flags

        # Accumulate non-exclusive tags
        flags = UpdateTagFlags.NONE
        flags |= UpdateTagFlags.UPDATE_MESSAGE
        new_path = data["as-path"]
        old_path = old_data["as-path"]

        # Path switching logic (Using the injected rnode context)
        if history := data.get("as-path-history"):
            try:
                # If current path matches the path from two updates ago
                if history[-2] == new_path and old_path != new_path:
                    flags |= UpdateTagFlags.PATH_SWITCHING
            except IndexError:
                pass

        # Origin or Transit change logic
        if old_path != new_path:
            if old_path[-1] != new_path[-1]:
                flags |= UpdateTagFlags.ORIGIN_CHANGE
            else:
                flags |= UpdateTagFlags.TRANSIT_CHANGE

        # Community Change logic
        if old_data["communities"] != data["communities"]:
            flags |= UpdateTagFlags.COMMUNITY_CHANGE

        # Prepending logic
        new_path_diff = len(new_path) - len(set(new_path))
        old_path_diff = len(old_path) - len(set(old_path))
        if new_path_diff or old_path_diff:
            if new_path_diff > old_path_diff:
                flags |= UpdateTagFlags.PREPENDING_ADD
                flags |= UpdateTagFlags.PREPENDING_CHANGE
            elif new_path_diff < old_path_diff:
                flags |= UpdateTagFlags.PREPENDING_REMOVE
                flags |= UpdateTagFlags.PREPENDING_CHANGE
            elif new_path_diff == old_path_diff and new_path != old_path:
                flags |= UpdateTagFlags.PREPENDING_CHANGE

        # If after all checks flags is still NONE, it was likely a duplicate/identical announce
        if flags == UpdateTagFlags.UPDATE_MESSAGE and old_path == new_path:
            flags = UpdateTagFlags.UPDATE_MESSAGE | UpdateTagFlags.DUPLICATE_ANNOUNCE
        self._log(bgpelem, flags)
        return flags

    def dump(self, ts: datetime) -> Optional[dict[str, int]]:
        return None

update_announcement(bgpelem, data, old_data)

Calculates non-exclusive tags for an announcement.

Source code in src/quickrib/observers/update_tagger.py

def update_announcement(self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData]) -> UpdateTagFlags:
    """
    Calculates non-exclusive tags for an announcement.
    """
    # Handle New Prefix (Exclusive to other logic for efficiency)
    if not old_data:
        flags = UpdateTagFlags.NEW_PREFIX | UpdateTagFlags.UPDATE_MESSAGE
        self._log(bgpelem, flags)
        return flags

    # Accumulate non-exclusive tags
    flags = UpdateTagFlags.NONE
    flags |= UpdateTagFlags.UPDATE_MESSAGE
    new_path = data["as-path"]
    old_path = old_data["as-path"]

    # Path switching logic (Using the injected rnode context)
    if history := data.get("as-path-history"):
        try:
            # If current path matches the path from two updates ago
            if history[-2] == new_path and old_path != new_path:
                flags |= UpdateTagFlags.PATH_SWITCHING
        except IndexError:
            pass

    # Origin or Transit change logic
    if old_path != new_path:
        if old_path[-1] != new_path[-1]:
            flags |= UpdateTagFlags.ORIGIN_CHANGE
        else:
            flags |= UpdateTagFlags.TRANSIT_CHANGE

    # Community Change logic
    if old_data["communities"] != data["communities"]:
        flags |= UpdateTagFlags.COMMUNITY_CHANGE

    # Prepending logic
    new_path_diff = len(new_path) - len(set(new_path))
    old_path_diff = len(old_path) - len(set(old_path))
    if new_path_diff or old_path_diff:
        if new_path_diff > old_path_diff:
            flags |= UpdateTagFlags.PREPENDING_ADD
            flags |= UpdateTagFlags.PREPENDING_CHANGE
        elif new_path_diff < old_path_diff:
            flags |= UpdateTagFlags.PREPENDING_REMOVE
            flags |= UpdateTagFlags.PREPENDING_CHANGE
        elif new_path_diff == old_path_diff and new_path != old_path:
            flags |= UpdateTagFlags.PREPENDING_CHANGE

    # If after all checks flags is still NONE, it was likely a duplicate/identical announce
    if flags == UpdateTagFlags.UPDATE_MESSAGE and old_path == new_path:
        flags = UpdateTagFlags.UPDATE_MESSAGE | UpdateTagFlags.DUPLICATE_ANNOUNCE
    self._log(bgpelem, flags)
    return flags

quickrib.observers.update_tagger.UpdateTagsCounter

Bases: UpdateTagger

Counts how often each tag fires, and reports the totals once per window.

Source code in src/quickrib/observers/update_tagger.py

class UpdateTagsCounter(UpdateTagger):
    """Counts how often each tag fires, and reports the totals once per window."""

    def __init__(self, name: str = "update_tags_counter", log_tags: bool = False):
        super().__init__(name, log_tags=log_tags)
        self.tag_counts: Counter[str] = Counter()
        # Pre-cache flag members to avoid .__dict__ or .members calls in hot loop.
        # `Flag.name` is Optional because a *composite* flag has none; every
        # member here is a single named one, hence the `or ""` formality.
        self._tag_list: list[tuple[UpdateTagFlags, str]] = [
            (f, f.name or "") for f in UpdateTagFlags if f is not UpdateTagFlags.NONE
        ]

    def _apply_counts(self, flags: UpdateTagFlags) -> None:
        # Iterating the value yields only the members that are set: one step per
        # tag that fired, rather than one test per tag that exists.
        for flag in flags:
            name = flag.name
            if name is not None:
                self.tag_counts[name] += 1

    def update_rib(self, bgpelem: ParsedElement):
        pass

    def update_withdrawal(self, bgpelem: WithdrawalElement, data: Optional[RIBNodeData]) -> UpdateTagFlags:
        tags = super().update_withdrawal(bgpelem, data)
        self._apply_counts(tags)
        return tags

    def update_announcement(self, bgpelem: ParsedElement, data: RIBNodeData, old_data: Optional[RIBNodeData]) -> UpdateTagFlags:
        tags = super().update_announcement(bgpelem, data, old_data)
        self._apply_counts(tags)
        return tags

    def dump(self, ts: datetime) -> dict[str, int]:
        """Return this window's tag totals, and start the next window empty.

        Every tag is present, including the ones that did not fire: a caller
        reading a series wants a zero, not a missing key.
        """
        counts = {name: self.tag_counts.get(name, 0) for _flag, name in self._tag_list}
        logger.info(
            "Update tags for the window ending %s: %s",
            ts,
            {name: count for name, count in counts.items() if count} or "none",
        )
        self.tag_counts.clear()
        return counts

dump(ts)

Return this window's tag totals, and start the next window empty.

Every tag is present, including the ones that did not fire: a caller reading a series wants a zero, not a missing key.

Source code in src/quickrib/observers/update_tagger.py

def dump(self, ts: datetime) -> dict[str, int]:
    """Return this window's tag totals, and start the next window empty.

    Every tag is present, including the ones that did not fire: a caller
    reading a series wants a zero, not a missing key.
    """
    counts = {name: self.tag_counts.get(name, 0) for _flag, name in self._tag_list}
    logger.info(
        "Update tags for the window ending %s: %s",
        ts,
        {name: count for name, count in counts.items() if count} or "none",
    )
    self.tag_counts.clear()
    return counts

Anomaly Detection

quickrib.observers.mad.MADConfig pydantic-model

Bases: BaseModel

Hyper-parameters for a MAD observer.

Show JSON schema:

{
  "description": "Hyper-parameters for a MAD observer.",
  "properties": {
    "window_size": {
      "description": "Number of historical snapshots in the window",
      "title": "Window Size",
      "type": "integer"
    },
    "H_stats_dampening": {
      "default": 0.01,
      "description": "Denominator floor for the scoring statistic",
      "title": "H Stats Dampening",
      "type": "number"
    },
    "mode": {
      "default": "mZ-score",
      "description": "Statistic used to compare a snapshot's coefficients to its history",
      "enum": [
        "Z-score",
        "mZ-score"
      ],
      "title": "Mode",
      "type": "string"
    }
  },
  "required": [
    "window_size"
  ],
  "title": "MADConfig",
  "type": "object"
}

Fields:

• window_size (int)
• H_stats_dampening (float)
• mode (Literal['Z-score', 'mZ-score'])

Source code in src/quickrib/observers/mad.py

class MADConfig(BaseModel):
    """Hyper-parameters for a MAD observer."""

    window_size: int = Field(description="Number of historical snapshots in the window")
    H_stats_dampening: float = Field(
        default=1e-2, description="Denominator floor for the scoring statistic"
    )
    mode: Literal["Z-score", "mZ-score"] = Field(
        default="mZ-score",
        description="Statistic used to compare a snapshot's coefficients to its history",
    )

window_size pydantic-field

Number of historical snapshots in the window

H_stats_dampening = 0.01 pydantic-field

Denominator floor for the scoring statistic

mode = 'mZ-score' pydantic-field

Statistic used to compare a snapshot's coefficients to its history

quickrib.observers.mad.MADObserver

Bases: Observer

Whole-graph MAD (MADglob).

Source code in src/quickrib/observers/mad.py

class MADObserver(Observer):
    """Whole-graph MAD (``MADglob``)."""

    def __init__(
        self,
        hegemony_observer: HegemonyObserver,
        window_size: int,
        H_stats_dampening: float = 1e-2,
        mode: Literal["Z-score", "mZ-score"] = "mZ-score",
        name: str = "mad",
        *,
        drives_hegemony: bool = True,
    ) -> None:
        self.hegemony_observer = hegemony_observer
        self.window_size = window_size
        self.H_stats_dampening = H_stats_dampening
        self.mode = mode
        self.stats_func = compute_Z_score if mode == "Z-score" else compute_mZ_score
        self.drives_hegemony = drives_hegemony
        self.name = name

        self.graphs: deque[Graph] = deque(maxlen=window_size + 1)
        self.scores: dict[datetime, float] = {}

    @classmethod
    def from_config(
        cls, hegemony_observer: HegemonyObserver, config: MADConfig, **kwargs
    ) -> "MADObserver":
        return cls(
            hegemony_observer,
            config.window_size,
            config.H_stats_dampening,
            config.mode,
            **kwargs,
        )

    # ---- hot path: forwarded to the owned hegemony observer ------------------


    def update_rib(self, bgpelem) -> None:
        if self.drives_hegemony:
            self.hegemony_observer.update_rib(bgpelem)

    def update_withdrawal(self, bgpelem, data) -> None:
        if self.drives_hegemony:
            self.hegemony_observer.update_withdrawal(bgpelem, data)

    def update_announcement(self, bgpelem, data, old_data) -> None:
        if self.drives_hegemony:
            self.hegemony_observer.update_announcement(bgpelem, data, old_data)

    def set_rib(self, rib) -> None:  # pragma: no cover - nothing to do
        pass

    def compare(self, other) -> None:  # pragma: no cover
        pass

    # ---- scoring -------------------------------------------------------------

    def _score_window(self, graphs: list[Graph]) -> float:
        scores = anomaly_scoring_subgraph(
            graphs[-1], graphs[:-1], self.stats_func, H_stats_dampening=self.H_stats_dampening
        )
        return float(np.sum(np.abs(scores)))

    def dump(self, ts: datetime) -> Any:
        """The whole-graph anomaly score for this window, or ``None`` until the
        history window is full.

        Typed as ``Any`` because the family disagrees on the shape of a result:
        :class:`MADStarObserver` returns one score *per AS*. Each subclass says
        what it returns.
        """
        graph = self.hegemony_observer.dump(ts)
        self.graphs.append(graph)
        if len(self.graphs) < self.window_size + 1:
            return None  # the window is not full yet, so there is nothing to score
        score = self._score_window(list(self.graphs))
        logger.info("MAD anomaly score at %s: %s", ts.isoformat(), score)
        self.scores[ts] = score
        return score

dump(ts)

The whole-graph anomaly score for this window, or None until the history window is full.

Typed as Any because the family disagrees on the shape of a result: :class:MADStarObserver returns one score per AS. Each subclass says what it returns.

Source code in src/quickrib/observers/mad.py

def dump(self, ts: datetime) -> Any:
    """The whole-graph anomaly score for this window, or ``None`` until the
    history window is full.

    Typed as ``Any`` because the family disagrees on the shape of a result:
    :class:`MADStarObserver` returns one score *per AS*. Each subclass says
    what it returns.
    """
    graph = self.hegemony_observer.dump(ts)
    self.graphs.append(graph)
    if len(self.graphs) < self.window_size + 1:
        return None  # the window is not full yet, so there is nothing to score
    score = self._score_window(list(self.graphs))
    logger.info("MAD anomaly score at %s: %s", ts.isoformat(), score)
    self.scores[ts] = score
    return score

quickrib.observers.mad.MADStaticObserver

Bases: MADObserver

Fast approximation (MADstatic): the edge dictionary is frozen at the first window, so later snapshots are only projected onto that fixed basis (edges that appear later are ignored).

Source code in src/quickrib/observers/mad.py

class MADStaticObserver(MADObserver):
    """Fast approximation (``MADstatic``): the edge dictionary is frozen at the
    first window, so later snapshots are only projected onto that fixed basis
    (edges that appear later are ignored)."""

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self._first_dump = True
        self.links: Optional[set[tuple[str, str]]] = None
        self.dictionary: Optional[SubgraphDictionary] = None
        self.rolling_coeffs: deque[np.ndarray] = deque(maxlen=self.window_size + 1)

    def dump(self, ts: datetime) -> Any:
        """The anomaly score for this window, or ``None`` while the window is not
        full (or before a non-empty first window has fixed the basis)."""
        graph = self.hegemony_observer.dump(ts)

        if self._first_dump:
            if not graph.to_edgelist():
                # The basis is whatever the first window holds, so an empty first
                # window would freeze an empty basis and score everything after it
                # as zero. Wait for a window with edges instead.
                logger.info("MADstatic: first window at %s is empty, deferring the "
                            "dictionary to the next one", ts.isoformat())
                return None
            self.links = graph.to_edgelist()
            self.dictionary = SubgraphDictionary(self.links, graph.to_weightfn())
            self._first_dump = False
        # Both are set on the first dump and never cleared afterwards.
        assert self.links is not None and self.dictionary is not None

        coeffs = np.hstack(
            decomposition_coefficients(graph.get_subgraph(self.links), self.dictionary)
        )
        self.rolling_coeffs.append(coeffs)
        if len(self.rolling_coeffs) < self.window_size + 1:
            return None
        rolling = list(self.rolling_coeffs)
        scores = self.stats_func(
            rolling[-1], np.vstack(rolling[:-1]), self.H_stats_dampening
        )
        score = float(np.sum(np.abs(scores)))
        logger.info("MADstatic anomaly score at %s: %s", ts.isoformat(), score)
        self.scores[ts] = score
        return score

dump(ts)

The anomaly score for this window, or None while the window is not full (or before a non-empty first window has fixed the basis).

Source code in src/quickrib/observers/mad.py

def dump(self, ts: datetime) -> Any:
    """The anomaly score for this window, or ``None`` while the window is not
    full (or before a non-empty first window has fixed the basis)."""
    graph = self.hegemony_observer.dump(ts)

    if self._first_dump:
        if not graph.to_edgelist():
            # The basis is whatever the first window holds, so an empty first
            # window would freeze an empty basis and score everything after it
            # as zero. Wait for a window with edges instead.
            logger.info("MADstatic: first window at %s is empty, deferring the "
                        "dictionary to the next one", ts.isoformat())
            return None
        self.links = graph.to_edgelist()
        self.dictionary = SubgraphDictionary(self.links, graph.to_weightfn())
        self._first_dump = False
    # Both are set on the first dump and never cleared afterwards.
    assert self.links is not None and self.dictionary is not None

    coeffs = np.hstack(
        decomposition_coefficients(graph.get_subgraph(self.links), self.dictionary)
    )
    self.rolling_coeffs.append(coeffs)
    if len(self.rolling_coeffs) < self.window_size + 1:
        return None
    rolling = list(self.rolling_coeffs)
    scores = self.stats_func(
        rolling[-1], np.vstack(rolling[:-1]), self.H_stats_dampening
    )
    score = float(np.sum(np.abs(scores)))
    logger.info("MADstatic anomaly score at %s: %s", ts.isoformat(), score)
    self.scores[ts] = score
    return score

quickrib.observers.mad.MADStarObserver

Bases: MADObserver

Per-AS star sub-graphs (MADstar): an anomaly score per AS per window.

Source code in src/quickrib/observers/mad.py

class MADStarObserver(MADObserver):
    """Per-AS star sub-graphs (``MADstar``): an anomaly score per AS per window."""

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.node_to_graphs: defaultdict[str, deque[Graph]] = defaultdict(
            lambda: deque(maxlen=self.window_size + 1)
        )
        self.node_to_scores: defaultdict[datetime, dict[str, float]] = defaultdict(dict)

    def dump(self, ts: datetime) -> Any:
        """This window's anomaly score per AS, empty until the window is full."""
        full_graph = self.hegemony_observer.dump(ts)
        stars = graph2graphstar(full_graph)
        logger.debug("MADstar: scoring %d star sub-graphs at %s", len(stars), ts.isoformat())

        # A node absent from this window gets an empty star, not a skipped slot.
        # Appending only for the nodes present would splice non-consecutive
        # windows into one history. An AS *vanishing* from the graph is itself
        # an anomaly, so it is the one case that must not go unscored.
        for node in self.node_to_graphs.keys() - stars.keys():
            stars[node] = Graph()

        for node, star in stars.items():
            history = self.node_to_graphs[node]
            history.append(star)
            if len(history) == self.window_size + 1:
                self.node_to_scores[ts][node] = self._score_window(list(history))
        # `.get`, not `[ts]`: reading a defaultdict creates the key, which would
        # record a window that scored nothing as an empty one.
        return self.node_to_scores.get(ts, {})

dump(ts)

This window's anomaly score per AS, empty until the window is full.

Source code in src/quickrib/observers/mad.py

def dump(self, ts: datetime) -> Any:
    """This window's anomaly score per AS, empty until the window is full."""
    full_graph = self.hegemony_observer.dump(ts)
    stars = graph2graphstar(full_graph)
    logger.debug("MADstar: scoring %d star sub-graphs at %s", len(stars), ts.isoformat())

    # A node absent from this window gets an empty star, not a skipped slot.
    # Appending only for the nodes present would splice non-consecutive
    # windows into one history. An AS *vanishing* from the graph is itself
    # an anomaly, so it is the one case that must not go unscored.
    for node in self.node_to_graphs.keys() - stars.keys():
        stars[node] = Graph()

    for node, star in stars.items():
        history = self.node_to_graphs[node]
        history.append(star)
        if len(history) == self.window_size + 1:
            self.node_to_scores[ts][node] = self._score_window(list(history))
    # `.get`, not `[ts]`: reading a defaultdict creates the key, which would
    # record a window that scored nothing as an empty one.
    return self.node_to_scores.get(ts, {})

Testing

quickrib.observers.tester.TesterObserver

Bases: Observer

Source code in src/quickrib/observers/tester.py

class TesterObserver(Observer):
    __test__ = False  # its name starts with "Test"; keep pytest from collecting it

    def __init__(self, config: "QuickRIBConfig", name: str = "tester"):
        super().__init__(name)
        self.config = config

        self.n_rib_entries = defaultdict(int)
        self.n_updates = defaultdict(int)
        self.n_withdrawals = defaultdict(int)
        self.n_announcements = defaultdict(int)

        self.n_rib_entries_per_peer = defaultdict(lambda: defaultdict(int))
        self.n_withdrawals_per_peer = defaultdict(lambda: defaultdict(int))
        self.n_announcements_per_peer = defaultdict(lambda: defaultdict(int))

        # New: Track [first_timestamp, last_timestamp] per peer
        self.peer_rib_times = defaultdict(lambda: defaultdict(lambda: [float('inf'), float('-inf')]))
        self.peer_update_times = defaultdict(lambda: defaultdict(lambda: [float('inf'), float('-inf')]))

    def update_rib(self, bgpelem: ParsedElement):
        is_valid_bgpelement(bgpelem, self.config)
        assert bgpelem.type == "R"

        peer_key = (bgpelem.peer_asn, bgpelem.peer_address)
        self.n_rib_entries[bgpelem.collector] += 1
        self.n_rib_entries_per_peer[bgpelem.collector][peer_key] += 1

        # Track RIB times
        ts = bgpelem.time
        times = self.peer_rib_times[bgpelem.collector][peer_key]
        times[0] = min(times[0], ts)
        times[1] = max(times[1], ts)

    def update_withdrawal(self, bgpelem: WithdrawalElement, data):
        is_valid_bgpelement(bgpelem, self.config)
        assert bgpelem.type == "W"

        peer_key = (bgpelem.peer_asn, bgpelem.peer_address)
        self.n_updates[bgpelem.collector] += 1
        self.n_withdrawals[bgpelem.collector] += 1
        self.n_withdrawals_per_peer[bgpelem.collector][peer_key] += 1

        # Track Update times
        ts = bgpelem.time
        times = self.peer_update_times[bgpelem.collector][peer_key]
        times[0] = min(times[0], ts)
        times[1] = max(times[1], ts)

    def update_announcement(self, bgpelem: ParsedElement, data, old_data):
        is_valid_bgpelement(bgpelem, self.config)
        assert bgpelem.type == "A"

        peer_key = (bgpelem.peer_asn, bgpelem.peer_address)
        self.n_updates[bgpelem.collector] += 1
        self.n_announcements[bgpelem.collector] += 1
        self.n_announcements_per_peer[bgpelem.collector][peer_key] += 1

        # Track Update times
        ts = bgpelem.time
        times = self.peer_update_times[bgpelem.collector][peer_key]
        times[0] = min(times[0], ts)
        times[1] = max(times[1], ts)

    def dump(self, ts: datetime.datetime):
        pass

    def compare(self, other):
        # not concerned with checks
        pass

    def __str__(self) -> str:
        collectors = set()
        for d in (self.n_rib_entries, self.n_updates, self.n_withdrawals, self.n_announcements,
                  self.n_rib_entries_per_peer, self.n_withdrawals_per_peer, self.n_announcements_per_peer,
                  self.peer_rib_times, self.peer_update_times):
            collectors.update(d.keys())

        if not collectors:
            return "<TesterObserver State: Empty (No data collected)>"

        # Helper to format timestamps gracefully
        def fmt_ts(ts):
            if ts in (float('inf'), float('-inf')):
                return "N/A"
            return datetime.datetime.fromtimestamp(ts, tz=datetime.UTC).strftime('%Y-%m-%d %H:%M:%S')

        lines = ["<TesterObserver Internal State>"]

        for coll in sorted(collectors):
            lines.append(f"\n[Collector: {coll}]")
            lines.append("  Totals:")
            lines.append(f"    RIB Entries:   {self.n_rib_entries.get(coll, 0)}")
            lines.append(f"    Updates:       {self.n_updates.get(coll, 0)}")
            lines.append(f"    Withdrawals:   {self.n_withdrawals.get(coll, 0)}")
            lines.append(f"    Announcements: {self.n_announcements.get(coll, 0)}")

            peers = set()
            for peer_dict in (self.n_rib_entries_per_peer.get(coll, {}),
                              self.n_withdrawals_per_peer.get(coll, {}),
                              self.n_announcements_per_peer.get(coll, {}),
                              self.peer_rib_times.get(coll, {}),
                              self.peer_update_times.get(coll, {})):
                peers.update(peer_dict.keys())

            if peers:
                lines.append("  Per-Peer Breakdown:")
                for peer in sorted(peers):
                    asn, addr = peer
                    rib = self.n_rib_entries_per_peer.get(coll, {}).get(peer, 0)
                    withd = self.n_withdrawals_per_peer.get(coll, {}).get(peer, 0)
                    ann = self.n_announcements_per_peer.get(coll, {}).get(peer, 0)

                    rib_times = self.peer_rib_times.get(coll, {}).get(peer, [float('inf'), float('-inf')])
                    upd_times = self.peer_update_times.get(coll, {}).get(peer, [float('inf'), float('-inf')])

                    lines.append(f"    - Peer ASN: {asn:<6} IP: {addr}")
                    lines.append(f"        Counts -> RIB: {rib:<5} | Announce: {ann:<5} | Withdraw: {withd}")
                    lines.append(f"        RIB    -> First: {fmt_ts(rib_times[0]):<19} | Last: {fmt_ts(rib_times[1])}")
                    lines.append(f"        Update -> First: {fmt_ts(upd_times[0]):<19} | Last: {fmt_ts(upd_times[1])}")

        lines.append("\n</TesterObserver>")
        return "\n".join(lines)