Network Interface Monitor and Alerter

Problem Statement

In production cloud environments, bare-metal servers, and edge nodes, unexpected network traffic spikes can quickly lead to degraded application performance, noisy-neighbor bottlenecks, or exorbitant data egress cloud bills. Standard system metrics often aggregate traffic over wide intervals (e.g., 5-minute cloud watch averages), delaying visibility into transient saturations or denial-of-service attempts.

This script provides real-time, lightweight network throughput monitoring directly on the server host. By polling interface traffic standard counters, calculating instantaneous megabits-per-second (Mbps) rates, and evaluating them against configurable thresholds, sysadmins and DevOps engineers can instantly detect anomalous bandwidth consumption and receive immediate alerts via HTTP webhooks (such as Slack, Teams, or custom alerting backends).

How It Works

The tool relies on psutil to inspect system network interface counters (psutil.net_io_counters) and requests to dispatch alerting payloads over HTTP.

At runtime, the monitor reads baseline byte counts (bytes_sent and bytes_recv) for a specified target network interface (e.g., eth0, ens33, wlan0). After waiting for a user-configured interval (defaulting to 2 seconds), it fetches updated counters. The difference in total byte counters over the elapsed duration is calculated and converted into Megabits per second (Mbps). If the calculated throughput exceeds the threshold, the script triggers an HTTP POST request carrying metric metadata to a configured webhook endpoint. To prevent alert fatigue, a configurable cooldown timer enforces a grace period between consecutive alert dispatches.

Full Script

#!/usrbin/env python3
"""
Network Interface Monitor and Alerter

Monitors network throughput on a specified interface using psutil.
Calculates instantaneous traffic in Mbps and dispatches HTTP webhook alerts
when throughput breaches a specified threshold.
"""

import argparse
import logging
import os
import sys
import time
from typing import Optional

import psutil
import requests

# Configure logging format and level
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("net_monitor")

# Configuration fallback defaults via Environment Variables
DEFAULT_INTERFACE = os.environ.get("MONITOR_INTERFACE", "eth0")
DEFAULT_THRESHOLD_MBPS = float(os.environ.get("THRESHOLD_MBPS", "100.0"))
DEFAULT_INTERVAL = float(os.environ.get("MONITOR_INTERVAL", "2.0"))
DEFAULT_COOLDOWN = float(os.environ.get("ALERT_COOLDOWN", "60.0"))
DEFAULT_WEBHOOK_URL = os.environ.get("WEBHOOK_URL", "")


def list_available_interfaces() -> list[str]:
    """Retrieve a list of valid active network interface names."""
    return list(psutil.net_io_counters(pernic=True).keys())


def get_interface_counters(interface: str) -> tuple[int, int]:
    """
    Fetch cumulative sent and received byte counts for a specific interface.
    
    Returns:
        tuple[int, int]: (bytes_sent, bytes_recv)
    """
    io_counters = psutil.net_io_counters(pernic=True)
    if interface not in io_counters:
        available = ", ".join(io_counters.keys())
        raise ValueError(
            f"Interface '{interface}' not found. Available interfaces: {available}"
        )
    
    stats = io_counters[interface]
    return stats.bytes_sent, stats.bytes_recv


def bytes_to_mbps(delta_bytes: int, elapsed_seconds: float) -> float:
    """
    Convert raw byte delta over time into Megabits per second (Mbps).
    
    1 Byte = 8 bits
    1 Megabit = 1,000,000 bits (SI standard for networking speed)
    """
    if elapsed_seconds <= 0:
        return 0.0
    total_bits = delta_bytes * 8
    bits_per_second = total_bits / elapsed_seconds
    return round(bits_per_second / 1_000_000.0, 2)


def send_webhook_alert(
    webhook_url: str,
    interface: str,
    current_mbps: float,
    threshold_mbps: float,
    sent_mbps: float,
    recv_mbps: float,
) -> bool:
    """
    Dispatch an HTTP POST payload alerting on a bandwidth threshold breach.
    """
    payload = {
        "text": f"🚨 *Bandwidth Threshold Exceeded* 🚨",
        "interface": interface,
        "current_mbps": current_mbps,
        "threshold_mbps": threshold_mbps,
        "tx_mbps": sent_mbps,
        "rx_mbps": recv_mbps,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime()),
    }

    try:
        response = requests.post(webhook_url, json=payload, timeout=5.0)
        response.raise_for_status()
        logger.info("Successfully dispatched alert payload to webhook.")
        return True
    except requests.exceptions.RequestException as exc:
        logger.error(f"Failed to send webhook alert: {exc}")
        return False


def parse_args() -> argparse.Namespace:
    """Parse command line arguments with environment variable fallbacks."""
    parser = argparse.ArgumentParser(
        description="Monitor network interface throughput and alert on high bandwidth usage."
    )
    parser.add_argument(
        "-i",
        "--interface",
        type=str,
        default=DEFAULT_INTERFACE,
        help=f"Network interface name (default: {DEFAULT_INTERFACE})",
    )
    parser.add_argument(
        "-t",
        "--threshold-mbps",
        type=float,
        default=DEFAULT_THRESHOLD_MBPS,
        help=f"Alerting threshold in Mbps (default: {DEFAULT_THRESHOLD_MBPS})",
    )
    parser.add_argument(
        "-s",
        "--interval",
        type=float,
        default=DEFAULT_INTERVAL,
        help=f"Polling interval in seconds (default: {DEFAULT_INTERVAL})",
    )
    parser.add_argument(
        "-c",
        "--cooldown",
        type=float,
        default=DEFAULT_COOLDOWN,
        help=f"Minimum seconds between consecutive alerts (default: {DEFAULT_COOLDOWN})",
    )
    parser.add_argument(
        "-w",
        "--webhook-url",
        type=str,
        default=DEFAULT_WEBHOOK_URL,
        help="HTTP Webhook URL for alerts (env: WEBHOOK_URL)",
    )
    parser.add_argument(
        "-n",
        "--count",
        type=int,
        default=0,
        help="Number of samples to record before exiting (0 = run indefinitely)",
    )
    return parser.parse_args()


def monitor_loop(
    interface: str,
    threshold_mbps: float,
    interval: float,
    cooldown: float,
    webhook_url: str,
    max_count: int,
) -> None:
    """Main sampling loop monitoring traffic and managing alert triggers."""
    logger.info(f"Starting monitoring on interface: '{interface}'")
    logger.info(f"Threshold: {threshold_mbps} Mbps | Interval: {interval}s | Cooldown: {cooldown}s")

    # Initial baseline measurements
    last_sent, last_recv = get_interface_counters(interface)
    last_time = time.time()
    last_alert_time = 0.0
    iteration = 0

    while True:
        time.sleep(interval)
        current_time = time.time()
        
        try:
            current_sent, current_recv = get_interface_counters(interface)
        except ValueError as err:
            logger.error(err)
            break

        elapsed = current_time - last_time
        
        # Calculate deltas
        bytes_sent_delta = max(0, current_sent - last_sent)
        bytes_recv_delta = max(0, current_recv - last_recv)
        total_bytes_delta = bytes_sent_delta + bytes_recv_delta

        # Calculate rate in Mbps
        tx_mbps = bytes_to_mbps(bytes_sent_delta, elapsed)
        rx_mbps = bytes_to_mbps(bytes_recv_delta, elapsed)
        total_mbps = bytes_to_mbps(total_bytes_delta, elapsed)

        logger.info(
            f"[{interface}] Total: {total_mbps:.2f} Mbps (TX: {tx_mbps:.2f} Mbps, RX: {rx_mbps:.2f} Mbps)"
        )

        # Check threshold condition
        if total_mbps >= threshold_mbps:
            logger.warning(
                f"HIGH TRAFFIC DETECTED: {total_mbps:.2f} Mbps exceeds limit ({threshold_mbps:.2f} Mbps)"
            )
            
            time_since_last_alert = current_time - last_alert_time
            if time_since_last_alert >= cooldown:
                if webhook_url:
                    if send_webhook_alert(
                        webhook_url, interface, total_mbps, threshold_mbps, tx_mbps, rx_mbps
                    ):
                        last_alert_time = current_time
                else:
                    logger.warning("Threshold breached, but no --webhook-url was provided.")
            else:
                remaining_cooldown = cooldown - time_since_last_alert
                logger.info(f"Alert suppressed. Cooldown active for {remaining_cooldown:.1f}s.")

        # Update state for next cycle
        last_sent, last_recv = current_sent, current_recv
        last_time = current_time
        iteration += 1

        if max_count > 0 and iteration >= max_count:
            logger.info("Reached maximum sample count. Exiting.")
            break


def main() -> None:
    args = parse_args()

    # Verify selected interface exists before looping
    available_ifaces = list_available_interfaces()
    if args.interface not in available_ifaces:
        logger.error(f"Invalid interface '{args.interface}'.")
        logger.error(f"Available interfaces on this host: {', '.join(available_ifaces)}")
        sys.exit(1)

    try:
        monitor_loop(
            interface=args.interface,
            threshold_mbps=args.threshold_mbps,
            interval=args.interval,
            cooldown=args.cooldown,
            webhook_url=args.webhook_url,
            max_count=args.count,
        )
    except KeyboardInterrupt:
        logger.info("Termination signal received. Exiting gracefully.")
        sys.exit(0)


if __name__ == "__main__":
    main()

Configuration

The script supports configuration through both command-line arguments and standard environment variables. CLI parameters take precedence over environment variable settings.

Usage Examples

1. Basic High-Bandwidth Monitoring (CLI Arguments)

Monitor a target interface (eth0), setting the high-traffic trigger to 900 Mbps:

python3 net_monitor.py --interface eth0 --threshold-mbps 900

Expected Output:

2023-10-24 10:00:00,102 [INFO] Starting monitoring on interface: 'eth0'
2023-10-24 10:00:00,102 [INFO] Threshold: 900.0 Mbps | Interval: 2.0s | Cooldown: 60.0s
2023-10-24 10:00:02,106 [INFO] [eth0] Total: 12.40 Mbps (TX: 1.10 Mbps, RX: 11.30 Mbps)
2023-10-24 10:00:04,110 [INFO] [eth0] Total: 942.15 Mbps (TX: 10.00 Mbps, RX: 932.15 Mbps)
2023-10-24 10:00:04,111 [WARNING] HIGH TRAFFIC DETECTED: 942.15 Mbps exceeds limit (900.00 Mbps)
2023-10-24 10:00:04,111 [WARNING] Threshold breached, but no --webhook-url was provided.

2. Environment Variables with Webhook Alerts

Integrate with continuous deployment or systemd units using environment variables:

export MONITOR_INTERFACE="ens33"
export THRESHOLD_MBPS="50.0"
export WEBHOOK_URL="https://hooks.slack.com/services/T00/B00/X00"

python3 net_monitor.py --cooldown 120

Expected Output:

2023-10-24 10:05:00,011 [INFO] Starting monitoring on interface: 'ens33'
2023-10-24 10:05:00,011 [INFO] Threshold: 50.0 Mbps | Interval: 2.0s | Cooldown: 120.0s
2023-10-24 10:05:02,014 [INFO] [ens33] Total: 85.20 Mbps (TX: 40.10 Mbps, RX: 45.10 Mbps)
2023-10-24 10:05:02,014 [WARNING] HIGH TRAFFIC DETECTED: 85.20 Mbps exceeds limit (50.00 Mbps)
2023-10-24 10:05:02,340 [INFO] Successfully dispatched alert payload to webhook.

3. Limited Run Diagnostics Mode

Sample the loop 3 times on interface wlan0 with a lower polling interval to gather baseline statistics:

python3 net_monitor.py -i wlan0 -s 1.0 -n 3

Error Handling

The script handles failure modes gracefully without crashing unexpected service workloads:

• Invalid or Missing Interface: At startup, list_available_interfaces() queries active host adapters. If the supplied interface name doesn't exist, the script logs an explicit error presenting all available interfaces and exits with code 1.
• Network Webhook Outages: Network POST requests to alerting endpoints use strict execution timeouts (timeout=5.0). HTTP failures, connection errors, or non-2xx status responses are caught via requests.exceptions.RequestException, logged cleanly as errors, and allow the monitoring loop to continue running uninterrupted.
• Counter Resets / Interface Flapping: Interface disconnects or counter overflows (e.g., interface hardware reset) are safely handled by enforcing max(0, current - last) on byte deltas, preventing negative rate anomalies.
• Interruption Handling: KeyboardInterrupt (SIGINT) is intercepted to provide clean output and zero exit codes when terminated manually.

Extending Ideas

1. Direction-Specific Alerting: Split threshold evaluations into separate ingress (RX) and egress (TX) triggers. This helps isolate download spikes (potential DDoS / large pull) from unexpected outbound data leaks (data exfiltration).
2. Prometheus Metrics Endpoint: Replace stdout logs with an embedded prometheus_client server exposing metrics like network_throughput_mbps{interface="eth0", direction="tx"}.
3. Adaptive Dynamic Thresholds: Calculate a moving average over a sliding window (e.g., 5-minute exponentially weighted moving average) and alert when current rates exceed N standard deviations above the baseline.
4. Automated Remediations: Execute system commands upon threshold violation (e.g., invoking iptables or nftables rate-limiting scripts when traffic exceeds emergency levels).