Interface Utilization Report

#!/usr/bin/env python3
"""
Interface Utilization Report

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/interface-utilization

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "netmiko>=4.3" "PyYAML>=6.0"
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

try:
    import yaml
except ImportError:
    yaml = None

from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
import re
import time
from dataclasses import dataclass

# ---------------------------------------------------------------------------
# Inventory loading and connection
# ---------------------------------------------------------------------------

NETMIKO_DRIVER = {
    "ios": "cisco_ios",
    "iosxe": "cisco_ios",
    "iosxr": "cisco_xr",
    "nxos": "cisco_nxos",
    "junos": "juniper_junos",
    "eos": "arista_eos",
    "panos": "paloalto_panos",
    "fortios": "fortinet",
}


def load_inventory(path: str) -> list[dict]:
    """
    Read devices.yaml (or .json) and return a list of device dicts.

    Applies the `defaults:` block to every device, validates the `os` field,
    and fails with a readable message rather than a stack trace.
    """
    p = Path(path)
    if not p.exists():
        raise SystemExit(
            f"Inventory file not found: {path}\n"
            f"Create it with:\n\n"
            f"  devices:\n"
            f"    - host: 192.168.1.1\n"
            f"      os: ios\n"
        )

    raw = p.read_text(encoding="utf-8")
    try:
        if p.suffix.lower() in (".yaml", ".yml"):
            if yaml is None:
                raise SystemExit("PyYAML is not installed. Run:  pip install PyYAML")
            data = yaml.safe_load(raw)
        else:
            data = json.loads(raw)
    except SystemExit:
        raise
    except Exception as exc:
        raise SystemExit(f"Could not parse {path}: {exc}\n"
                         f"Check for tabs (use spaces) and missing spaces after colons.")

    if isinstance(data, dict):
        defaults = data.get("defaults", {}) or {}
        devices = data.get("devices", []) or []
    else:
        defaults, devices = {}, data

    if not devices:
        raise SystemExit(f"No devices found in {path}. The file needs a `devices:` list.")

    out = []
    for i, d in enumerate(devices, 1):
        if not isinstance(d, dict):
            raise SystemExit(f"Device #{i} in {path} is not a mapping. "
                             f"Each entry needs at least `host:` and `os:`.")
        merged = {**defaults, **d}
        if "host" not in merged:
            raise SystemExit(f"Device #{i} in {path} has no `host:` field.")
        os_key = merged.get("os")
        if not os_key:
            raise SystemExit(f"Device '{merged['host']}' has no `os:` field. "
                             f"Valid values: {', '.join(sorted(NETMIKO_DRIVER))}")
        if os_key not in NETMIKO_DRIVER:
            raise SystemExit(
                f"Device '{merged['host']}' has os '{os_key}', which is not supported.\n"
                f"Valid values: {', '.join(sorted(NETMIKO_DRIVER))}\n"
                f"Note: 'cisco' is a vendor, not an OS - use ios, iosxe, iosxr, or nxos."
            )
        merged.setdefault("vrfs", [])
        out.append(merged)
    return out


def credentials() -> tuple[str, str, str]:
    """Read credentials from the environment and fail clearly if missing."""
    user = os.environ.get("NET_USER", "")
    pwd = os.environ.get("NET_PASS", "")
    enable = os.environ.get("NET_ENABLE", "")
    if not user or not pwd:
        raise SystemExit(
            "Set NET_USER and NET_PASS in your environment before running.\n\n"
            "  Linux/macOS:   export NET_USER='admin'\n"
            "                 export NET_PASS='your-password'\n\n"
            "  PowerShell:    $env:NET_USER = 'admin'\n"
            "                 $env:NET_PASS = 'your-password'\n"
        )
    return user, pwd, enable


def connect(dev: dict):
    """Open an SSH session to one device using netmiko."""
    user, pwd, enable = credentials()
    params = {
        "device_type": NETMIKO_DRIVER[dev["os"]],
        "host": dev["host"],
        "username": dev.get("username") or user,
        "password": dev.get("password") or pwd,
        "port": dev.get("port", 22),
        "conn_timeout": 20,
        "fast_cli": False,
    }
    secret = dev.get("secret") or enable
    if secret:
        params["secret"] = secret

    conn = ConnectHandler(**params)
    if secret and dev["os"] in ("ios", "iosxe", "nxos"):
        conn.enable()
    return conn


# ---------------------------------------------------------------------------
# Commands and parsing
# ---------------------------------------------------------------------------

COMMAND_MAP = {
    "ios": "show interfaces",
    "iosxe": "show interfaces",
    "iosxr": "show interfaces",
    "nxos": "show interface",
    "junos": "show interfaces detail",
    "eos": "show interfaces",
    "panos": "show interface all",
    "fortios": "get system interface physical",
}

# Interface speed, so a rate can be expressed as a percentage of capacity.
SPEED_CMD = {
    "ios": "show interfaces status",
    "iosxe": "show interfaces status",
    "nxos": "show interface status",
    "eos": "show interfaces status",
}

# 32-bit counters wrap at 2^32. 64-bit ones effectively never do. A sample
# that goes backwards is either a wrap or a counter reset; both make the
# delta meaningless, so it is reported rather than silently dropped.
WRAP_32 = 2 ** 32


@dataclass
class Sample:
    in_bytes: int
    out_bytes: int
    in_errors: int = 0
    out_errors: int = 0
    discards: int = 0


def parse_counters(output: str, os_key: str) -> dict[str, Sample]:
    """Return {interface: Sample} from one poll of the interface counters."""
    counters: dict[str, Sample] = {}
    cur: str | None = None

    if os_key in ("ios", "iosxe", "iosxr", "nxos", "eos"):
        acc: dict[str, dict] = {}
        for line in output.splitlines():
            m = re.match(r"^(\S+)\s+is\s+\S+", line)
            if m and not line.startswith((" ", "\t")):
                cur = m.group(1)
                acc.setdefault(cur, {})
            if not cur:
                continue
            for key, pat in (
                ("in_bytes", r"(\d+) bytes input"),
                ("out_bytes", r"(\d+) bytes output"),
                ("in_errors", r"(\d+) input errors"),
                ("out_errors", r"(\d+) output errors"),
            ):
                g = re.search(pat, line)
                if g:
                    acc[cur][key] = int(g.group(1))
            g = re.search(r"Total output drops:\s*(\d+)", line)
            if g:
                acc[cur]["discards"] = int(g.group(1))
        for name, v in acc.items():
            if "in_bytes" in v and "out_bytes" in v:
                counters[name] = Sample(
                    v["in_bytes"], v["out_bytes"],
                    v.get("in_errors", 0), v.get("out_errors", 0), v.get("discards", 0))

    elif os_key == "junos":
        acc: dict[str, dict] = {}
        for line in output.splitlines():
            m = re.match(r"\s*Physical interface:\s+(\S+?),", line)
            if m:
                cur = m.group(1)
                acc.setdefault(cur, {})
            if not cur:
                continue
            for key, pat in (
                ("in_bytes", r"Input bytes\s*:\s*(\d+)"),
                ("out_bytes", r"Output bytes\s*:\s*(\d+)"),
                ("in_errors", r"Input errors:\s*(\d+)"),
                ("out_errors", r"Output errors:\s*(\d+)"),
            ):
                g = re.search(pat, line)
                if g:
                    acc[cur][key] = int(g.group(1))
        for name, v in acc.items():
            if "in_bytes" in v and "out_bytes" in v:
                counters[name] = Sample(v["in_bytes"], v["out_bytes"],
                                        v.get("in_errors", 0), v.get("out_errors", 0))

    elif os_key == "panos":
        for line in output.splitlines():
            m = re.match(r"(\S+)\s+\S+\s+\S+.*?rx:(\d+)\s+tx:(\d+)", line)
            if m:
                counters[m.group(1)] = Sample(int(m.group(2)), int(m.group(3)))

    elif os_key == "fortios":
        acc: dict[str, dict] = {}
        for line in output.splitlines():
            m = re.match(r"==\s*\[(\S+)\]", line)
            if m:
                cur = m.group(1)
                acc.setdefault(cur, {})
            if not cur:
                continue
            for key, pat in (("in_bytes", r"rx_bytes=(\d+)"),
                             ("out_bytes", r"tx_bytes=(\d+)"),
                             ("in_errors", r"rx_errors=(\d+)"),
                             ("out_errors", r"tx_errors=(\d+)")):
                g = re.search(pat, line)
                if g:
                    acc[cur][key] = int(g.group(1))
        for name, v in acc.items():
            if "in_bytes" in v and "out_bytes" in v:
                counters[name] = Sample(v["in_bytes"], v["out_bytes"],
                                        v.get("in_errors", 0), v.get("out_errors", 0))

    return counters


def parse_speeds(output: str) -> dict[str, int]:
    """
    Read negotiated speed per interface so utilisation is a true percentage.

    `show interfaces status` prints e.g. a-1000, 10G, 1000. A hardcoded
    assumption of 1 Gbps reports a saturated 100 Mbps port as 10% busy.
    """
    speeds: dict[str, int] = {}
    for line in output.splitlines():
        parts = line.split()
        if len(parts) < 2:
            continue
        name = parts[0]
        for tok in parts[1:]:
            m = re.fullmatch(r"a?-?(\d+)([GMgm])?", tok)
            if not m:
                continue
            val = int(m.group(1))
            unit = (m.group(2) or "M").upper()
            mbps = val * 1000 if unit == "G" else val
            if mbps in (10, 100, 1000, 2500, 5000, 10000, 25000, 40000, 100000):
                speeds[name] = mbps
                break
    return speeds


def delta(prev: int, now: int) -> int | None:
    """Counter delta, or None when the counter wrapped or was reset."""
    if now >= prev:
        return now - prev
    if prev < WRAP_32 and now < WRAP_32:
        return (WRAP_32 - prev) + now        # plausible 32-bit wrap
    return None                               # reset or 64-bit wrap: unusable


# ---------------------------------------------------------------------------
# Per-device sampling
# ---------------------------------------------------------------------------

def sample_device(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    command = COMMAND_MAP[os_key]
    default_speed = int(dev.get("speed_mbps", args.default_speed))
    conn = None

    try:
        conn = connect(dev)

        speeds: dict[str, int] = {}
        if SPEED_CMD.get(os_key) and not args.assume_speed:
            try:
                speeds = parse_speeds(conn.send_command(SPEED_CMD[os_key], read_timeout=60))
            except Exception:  # noqa: BLE001 - speed is an optimisation, not required
                speeds = {}

        first = parse_counters(conn.send_command(command, read_timeout=args.read_timeout), os_key)
        time.sleep(args.interval)
        second = parse_counters(conn.send_command(command, read_timeout=args.read_timeout), os_key)

        rows = []
        for iface, a in first.items():
            b = second.get(iface)
            if not b:
                continue

            d_in = delta(a.in_bytes, b.in_bytes)
            d_out = delta(a.out_bytes, b.out_bytes)
            if d_in is None or d_out is None:
                rows.append({"iface": iface, "wrapped": True})
                continue

            in_mbps = d_in * 8 / args.interval / 1_000_000
            out_mbps = d_out * 8 / args.interval / 1_000_000
            speed = speeds.get(iface, default_speed)
            pct = max(in_mbps, out_mbps) / speed * 100 if speed else 0.0

            err_in = delta(a.in_errors, b.in_errors) or 0
            err_out = delta(a.out_errors, b.out_errors) or 0
            drops = delta(a.discards, b.discards) or 0

            if args.only_active and in_mbps < 0.01 and out_mbps < 0.01:
                continue

            rows.append({
                "iface": iface, "wrapped": False,
                "in_mbps": round(in_mbps, 2), "out_mbps": round(out_mbps, 2),
                "speed_mbps": speed, "pct": round(pct, 1),
                "in_errors": err_in, "out_errors": err_out, "discards": drops,
                "speed_known": iface in speeds,
            })

        return {"host": host, "os": os_key, "status": "ok",
                "rows": sorted(rows, key=lambda r: r.get("pct", 0), reverse=True)}

    except NetmikoAuthenticationException as exc:
        return {"host": host, "os": os_key, "status": "error", "rows": [],
                "error": f"authentication failed - check NET_USER/NET_PASS ({exc})"}
    except NetmikoTimeoutException as exc:
        return {"host": host, "os": os_key, "status": "error", "rows": [],
                "error": f"unreachable or SSH timed out ({exc})"}
    except Exception as exc:  # noqa: BLE001 - one device must not abort the run
        return {"host": host, "os": os_key, "status": "error", "rows": [],
                "error": f"{type(exc).__name__}: {exc}"}
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Measure real interface throughput by sampling counters twice.",
        epilog="Credentials come from NET_USER and NET_PASS. See the docstring at "
               "the top of this file for the full inventory format.",
    )
    ap.add_argument("--inventory", required=True,
                    help="Path to devices.yaml listing which devices to sample")
    ap.add_argument("--interval", type=int, default=30,
                    help="Seconds between the two counter samples. Longer smooths out "
                         "bursts; shorter reacts faster. Default: 30")
    ap.add_argument("--threshold", type=float, default=80.0,
                    help="Flag interfaces at or above this percent of capacity. Default: 80")
    ap.add_argument("--default-speed", type=int, default=1000,
                    help="Interface speed in Mbps when it cannot be read from the "
                         "device. Default: 1000")
    ap.add_argument("--assume-speed", action="store_true",
                    help="Skip reading per-interface speed and use --default-speed "
                         "everywhere. Faster, less accurate")
    ap.add_argument("--only-active", action="store_true",
                    help="Hide interfaces carrying essentially no traffic")
    ap.add_argument("--top", type=int, default=0,
                    help="Show only the N busiest interfaces per device. 0 = all")
    ap.add_argument("--limit", help="Only sample hosts whose name contains this text")
    ap.add_argument("--read-timeout", type=int, default=120,
                    help="Seconds to wait per command. Default: 120")
    ap.add_argument("--workers", type=int, default=10,
                    help="How many devices to sample at once. Default: 10")
    ap.add_argument("--json-out", help="Write machine-readable results here")
    args = ap.parse_args()

    devices = load_inventory(args.inventory)
    if args.limit:
        devices = [d for d in devices if args.limit in d["host"]]
        if not devices:
            raise SystemExit(f"No device name contains '{args.limit}'.")

    credentials()  # fail fast before opening any connection

    # All devices are sampled concurrently, so the whole run takes roughly
    # --interval seconds regardless of how many devices there are.
    print(f"Sampling {len(devices)} device(s): two polls {args.interval}s apart, "
          f"all in parallel...")
    print("=" * 86)

    results = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(sample_device, d, args): d["host"] for d in devices}
        for fut in as_completed(futures):
            host = futures[fut]
            try:
                results.append(fut.result())
            except Exception as exc:  # noqa: BLE001
                results.append({"host": host, "os": "?", "status": "error", "rows": [],
                                "error": f"worker crashed: {exc}"})

    high = 0
    erroring = 0
    for r in sorted(results, key=lambda x: x["host"]):
        if r["status"] == "error":
            print(f"[ERROR] {r['host']} ({r['os']})  {r['error']}")
            continue

        rows = r["rows"][: args.top] if args.top else r["rows"]
        print(f"\n{r['host']} ({r['os']})  {len(r['rows'])} interface(s)")
        if not rows:
            print("   no interface carried measurable traffic")
            continue

        for row in rows:
            if row["wrapped"]:
                print(f"   {row['iface']:<22} counter wrapped or was reset - "
                      f"rate not measurable this cycle")
                continue
            flag = "  [HIGH]" if row["pct"] >= args.threshold else ""
            if row["pct"] >= args.threshold:
                high += 1
            speed_mark = "" if row["speed_known"] else "*"
            errs = ""
            if row["in_errors"] or row["out_errors"] or row["discards"]:
                errs = (f"  err in/out {row['in_errors']}/{row['out_errors']}"
                        f"  drops {row['discards']}")
                erroring += 1
            print(f"   {row['iface']:<22} in {row['in_mbps']:>8.2f}  "
                  f"out {row['out_mbps']:>8.2f} Mbps  "
                  f"{row['pct']:>5.1f}% of {row['speed_mbps']}M{speed_mark}{flag}{errs}")

    print("\n" + "=" * 86)
    print(f"{high} interface(s) at or above {args.threshold}% of capacity. "
          f"{erroring} with errors or drops in this window.")
    if any(not row.get("speed_known", True)
           for r in results for row in r["rows"] if not row.get("wrapped")):
        print(f"* speed could not be read from the device; "
              f"assumed {args.default_speed} Mbps")

    if args.json_out:
        Path(args.json_out).write_text(json.dumps(results, indent=2), encoding="utf-8")

    return 1 if high else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)