301 lines
10 KiB
Python
301 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
watch_remote_top.py
|
|
|
|
Prompts for a remote host/IP, SSH username, and password (interactively,
|
|
password not echoed), then every 10 minutes:
|
|
- SSHes in and runs `top -bn1 -o %MEM` (one-shot, non-interactive, sorted by RAM)
|
|
- Saves the full raw output to a local log file
|
|
- Parses out CPU usage, RAM usage, and the top RAM-consuming process into a
|
|
local CSV summary
|
|
|
|
Everything is saved on THIS machine, not the remote host.
|
|
|
|
Requires: paramiko
|
|
pip install paramiko
|
|
|
|
Usage:
|
|
python3 watch_remote_top.py
|
|
python3 watch_remote_top.py --once # single run, no loop
|
|
python3 watch_remote_top.py --interval 5 # poll every 5 minutes instead of 10
|
|
python3 watch_remote_top.py --log-dir ./logs # where to save logs (default: ./TopLogs)
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import getpass
|
|
import re
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import paramiko
|
|
except ImportError:
|
|
sys.exit(
|
|
"The 'paramiko' package is required but not installed.\n"
|
|
"Install it with: pip install paramiko"
|
|
)
|
|
|
|
# Plain `-bn1` (no -o): BusyBox's top doesn't support -o for sorting, so we
|
|
# capture unsorted output and sort by memory ourselves in parse_top_output().
|
|
REMOTE_COMMAND = "top -bn1"
|
|
|
|
|
|
def get_connection_details():
|
|
host = input("Remote host/IP: ").strip()
|
|
username = input("SSH username: ").strip()
|
|
password = getpass.getpass("SSH password: ")
|
|
return host, username, password
|
|
|
|
|
|
def run_remote_top(host, username, password, timeout=10):
|
|
"""SSH in, run the top command once, return the raw stdout as a string."""
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
try:
|
|
client.connect(
|
|
hostname=host,
|
|
username=username,
|
|
password=password,
|
|
timeout=timeout,
|
|
banner_timeout=timeout,
|
|
auth_timeout=timeout,
|
|
)
|
|
stdin, stdout, stderr = client.exec_command(REMOTE_COMMAND, timeout=timeout)
|
|
output = stdout.read().decode(errors="replace")
|
|
err = stderr.read().decode(errors="replace")
|
|
if not output and err:
|
|
raise RuntimeError(f"Remote command produced no output. stderr: {err.strip()}")
|
|
return output
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def _to_mib(value, unit):
|
|
"""Convert a K/M/G-suffixed value (as used by BusyBox's `Mem:` line) to MiB."""
|
|
value = float(value)
|
|
unit = (unit or "").upper()
|
|
if unit.startswith("K"):
|
|
return value / 1024
|
|
if unit.startswith("G"):
|
|
return value * 1024
|
|
return value # already M / MiB, or no unit
|
|
|
|
|
|
def parse_top_output(raw_output):
|
|
"""
|
|
Pull CPU used %, mem used/total/free (MiB), mem used %, and the top
|
|
RAM-consuming process out of raw `top -bn1` output. Handles both the
|
|
standard Linux (procps) format and BusyBox's leaner top format, since
|
|
the remote command has no reliable way to sort by memory itself
|
|
(BusyBox top doesn't support -o), so the top process is found by
|
|
scanning and sorting the process rows here instead.
|
|
"""
|
|
lines = raw_output.splitlines()
|
|
|
|
cpu_used = cpu_idle = None
|
|
mem_total = mem_used = mem_free = mem_used_pct = None
|
|
top_process = None
|
|
|
|
is_procps = any("%Cpu(s)" in line for line in lines)
|
|
is_busybox = any(re.match(r"^CPU:", line.strip()) for line in lines)
|
|
|
|
if is_procps:
|
|
for line in lines:
|
|
if "%Cpu(s)" in line:
|
|
m = re.search(r"([\d.]+)\s*id", line)
|
|
if m:
|
|
cpu_idle = float(m.group(1))
|
|
cpu_used = round(100 - cpu_idle, 1)
|
|
|
|
if re.search(r"(MiB|KiB|GiB)\s+Mem", line):
|
|
m_total = re.search(r"([\d.]+)\s+total", line)
|
|
m_free = re.search(r"([\d.]+)\s+free", line)
|
|
m_used = re.search(r"([\d.]+)\s+used", line)
|
|
if m_total:
|
|
mem_total = float(m_total.group(1))
|
|
if m_free:
|
|
mem_free = float(m_free.group(1))
|
|
if m_used:
|
|
mem_used = float(m_used.group(1))
|
|
if mem_total and mem_used:
|
|
mem_used_pct = round((mem_used / mem_total) * 100, 1)
|
|
|
|
# PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
|
|
header_idx, mem_col_idx = None, None
|
|
for i, line in enumerate(lines):
|
|
if re.match(r"^\s*PID\s+USER", line):
|
|
header_cols = line.split()
|
|
if "%MEM" in header_cols:
|
|
header_idx = i
|
|
mem_col_idx = header_cols.index("%MEM")
|
|
break
|
|
|
|
if header_idx is not None and mem_col_idx is not None:
|
|
best_pct = -1.0
|
|
for line in lines[header_idx + 1:]:
|
|
cols = line.split()
|
|
if len(cols) <= mem_col_idx:
|
|
continue
|
|
try:
|
|
pct = float(cols[mem_col_idx])
|
|
except ValueError:
|
|
continue
|
|
if pct > best_pct:
|
|
best_pct = pct
|
|
top_process = f"{cols[-1]} ({pct}% MEM)"
|
|
|
|
elif is_busybox:
|
|
# e.g. "Mem: 51380K used, 178308K free, 0K shrd, 3844K buff, 21520K cached"
|
|
for line in lines:
|
|
m = re.search(
|
|
r"Mem:\s*([\d.]+)\s*([KMG])?\s*used,\s*([\d.]+)\s*([KMG])?\s*free",
|
|
line,
|
|
re.IGNORECASE,
|
|
)
|
|
if m:
|
|
mem_used = round(_to_mib(m.group(1), m.group(2)), 1)
|
|
mem_free = round(_to_mib(m.group(3), m.group(4)), 1)
|
|
mem_total = round(mem_used + mem_free, 1)
|
|
if mem_total:
|
|
mem_used_pct = round((mem_used / mem_total) * 100, 1)
|
|
|
|
# e.g. "CPU: 2.3% usr 1.1% sys 0.0% nic 96.4% idle ..."
|
|
m_idle = re.search(r"([\d.]+)%?\s*idle", line, re.IGNORECASE)
|
|
if m_idle:
|
|
cpu_idle = float(m_idle.group(1))
|
|
cpu_used = round(100 - cpu_idle, 1)
|
|
|
|
# e.g. " PID PPID USER STAT VSZ %VSZ %CPU COMMAND"
|
|
header_idx, mem_col_idx = None, None
|
|
for i, line in enumerate(lines):
|
|
cols = line.split()
|
|
if cols[:1] == ["PID"] and "USER" in cols:
|
|
header_idx = i
|
|
if "%VSZ" in cols:
|
|
mem_col_idx = cols.index("%VSZ")
|
|
elif "%MEM" in cols:
|
|
mem_col_idx = cols.index("%MEM")
|
|
break
|
|
|
|
if header_idx is not None and mem_col_idx is not None:
|
|
best_pct = -1.0
|
|
for line in lines[header_idx + 1:]:
|
|
cols = line.split()
|
|
if len(cols) <= mem_col_idx:
|
|
continue
|
|
try:
|
|
pct = float(cols[mem_col_idx].rstrip("%"))
|
|
except ValueError:
|
|
continue
|
|
if pct > best_pct:
|
|
best_pct = pct
|
|
top_process = f"{cols[-1]} ({pct}% VSZ)"
|
|
|
|
return {
|
|
"cpu_used": cpu_used,
|
|
"cpu_idle": cpu_idle,
|
|
"mem_total": mem_total,
|
|
"mem_used": mem_used,
|
|
"mem_free": mem_free,
|
|
"mem_used_pct": mem_used_pct,
|
|
"top_process": top_process,
|
|
}
|
|
|
|
|
|
def ensure_csv_header(csv_path):
|
|
if not csv_path.exists():
|
|
with open(csv_path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow(
|
|
[
|
|
"Timestamp",
|
|
"CPU_Used_Pct",
|
|
"CPU_Idle_Pct",
|
|
"Mem_Total_MiB",
|
|
"Mem_Used_MiB",
|
|
"Mem_Free_MiB",
|
|
"Mem_Used_Pct",
|
|
"Top_RAM_Process",
|
|
]
|
|
)
|
|
|
|
|
|
def collect_snapshot(host, username, password, full_log_path, summary_csv_path):
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
print(f"[{timestamp}] Connecting to {host}...")
|
|
|
|
try:
|
|
raw_output = run_remote_top(host, username, password)
|
|
except Exception as exc:
|
|
print(f"[{timestamp}] SSH/command failed: {exc}", file=sys.stderr)
|
|
with open(summary_csv_path, "a", newline="", encoding="utf-8") as f:
|
|
csv.writer(f).writerow([timestamp, "ERROR", "", "", "", "", "", f"SSH failed: {exc}"])
|
|
return
|
|
|
|
# Save full raw snapshot
|
|
with open(full_log_path, "a", encoding="utf-8") as f:
|
|
f.write(f"===== {timestamp} =====\n")
|
|
f.write(raw_output)
|
|
f.write("\n\n")
|
|
|
|
parsed = parse_top_output(raw_output)
|
|
|
|
with open(summary_csv_path, "a", newline="", encoding="utf-8") as f:
|
|
csv.writer(f).writerow(
|
|
[
|
|
timestamp,
|
|
parsed["cpu_used"],
|
|
parsed["cpu_idle"],
|
|
parsed["mem_total"],
|
|
parsed["mem_used"],
|
|
parsed["mem_free"],
|
|
parsed["mem_used_pct"],
|
|
parsed["top_process"],
|
|
]
|
|
)
|
|
|
|
print(
|
|
f"[{timestamp}] CPU used: {parsed['cpu_used']}% | "
|
|
f"Mem used: {parsed['mem_used_pct']}% "
|
|
f"({parsed['mem_used']}/{parsed['mem_total']} MiB) | "
|
|
f"Top RAM proc: {parsed['top_process']}"
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Poll a remote Linux host's `top` output over SSH and log it locally.")
|
|
parser.add_argument("--once", action="store_true", help="Run a single collection and exit instead of looping.")
|
|
parser.add_argument("--interval", type=int, default=10, help="Polling interval in minutes (default: 10).")
|
|
parser.add_argument("--log-dir", type=str, default="TopLogs", help="Local folder to save logs in (default: ./TopLogs).")
|
|
args = parser.parse_args()
|
|
|
|
host, username, password = get_connection_details()
|
|
|
|
log_dir = Path(args.log_dir)
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
full_log_path = log_dir / "top_full_snapshots.log"
|
|
summary_csv_path = log_dir / "cpu_ram_summary.csv"
|
|
ensure_csv_header(summary_csv_path)
|
|
|
|
print(f"\nFull snapshots -> {full_log_path}")
|
|
print(f"Summary CSV -> {summary_csv_path}\n")
|
|
|
|
if args.once:
|
|
collect_snapshot(host, username, password, full_log_path, summary_csv_path)
|
|
return
|
|
|
|
print(f"Polling {host} every {args.interval} minute(s). Press Ctrl+C to stop.\n")
|
|
try:
|
|
while True:
|
|
collect_snapshot(host, username, password, full_log_path, summary_csv_path)
|
|
time.sleep(args.interval * 60)
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|