Approach: sample /proc/<pid> every 30s, keep a sliding window of the last 10 minutes (20 samples). For robustness against PID reuse, record the process starttime (from /proc/[pid]/stat field 22) and only compare samples with the same starttime. If RSS rises >20% from the window's oldest sample to current, print an alert including PID, starttime, RSS values, command line and owner.python
#!/usr/bin/env python3
import time, os, pwd
from collections import deque
SAMPLE_INTERVAL = 30 # seconds
WINDOW_SECONDS = 10 * 60 # 10 minutes
MAX_SAMPLES = WINDOW_SECONDS // SAMPLE_INTERVAL
THRESHOLD = 0.20 # 20%
def read_proc_stat(pid):
p = f"/proc/{pid}"
try:
# get RSS (kB) from /proc/<pid>/status VmRSS
with open(os.path.join(p, "status")) as f:
lines = f.readlines()
rss = None
for ln in lines:
if ln.startswith("VmRSS:"):
rss = int(ln.split()[1]) # in kB
break
# get starttime (field 22) and cmdline
with open(os.path.join(p, "stat")) as f:
stat = f.read().split()
starttime = stat[21]
with open(os.path.join(p, "cmdline")) as f:
cmd = f.read().replace("\0", " ").strip()
owner = pwd.getpwuid(os.stat(p).st_uid).pw_name
return int(rss) if rss is not None else 0, starttime, cmd, owner
except Exception:
return None # process gone or permission error
def monitor(pid):
window = deque(maxlen=MAX_SAMPLES) # store tuples (timestamp, rss_kb, starttime)
while True:
info = read_proc_stat(pid)
ts = time.time()
if info is None:
print(f"[{time.ctime()}] PID {pid} not present or inaccessible")
window.clear()
else:
rss, starttime, cmd, owner = info
window.append((ts, rss, starttime))
# find oldest sample with same starttime
same_proc_samples = [s for s in window if s[2] == starttime]
if len(same_proc_samples) >= 2:
old_ts, old_rss, _ = same_proc_samples[0]
new_ts, new_rss, _ = same_proc_samples[-1]
# compare growth
if old_rss > 0 and (new_rss - old_rss) / old_rss > THRESHOLD:
print(f"[ALERT {time.ctime()}] PID {pid} (start {starttime}) RSS grew >{THRESHOLD*100}%: "
f"{old_rss}kB -> {new_rss}kB. cmd='{cmd}' owner={owner}")
# optional: take action, e.g., notify, capture heap, restart
time.sleep(SAMPLE_INTERVAL)
if __name__ == "__main__":
import sys
if len(sys.argv) != 2:
print("usage: monitor_rss.py <pid>")
raise SystemExit(1)
monitor(int(sys.argv[1]))
Key points:- Uses VmRSS (resident set) from /proc/<pid>/status in kB.- Guard against PID reuse with starttime from /proc/<pid>/stat.- Sliding window length = 10 minutes (20 samples at 30s).Limitations:- Requires permission to read /proc/<pid> (root or same user).- /proc VmRSS is Linux-specific and reports current RSS only.- Short sampling interval may miss very short-lived spikes between samples.- Starttime resolution ties to boot-time jiffies; in rare cases of collisions it may not fully prevent reuse confusion.Running at scale:- Don’t run one script per PID on many hosts. Instead: - Integrate into existing monitoring (Prometheus exporter or node_exporter textfile) to expose rss metrics per PID or per service. - Push metrics to a centralized TSDB and run alerts (Prometheus alerting rule comparing t-10m to now). - Use service labels (systemd/container IDs) rather than PIDs where possible. - Run with limited privileges (capabilities) and aggregate efficiently (batch reads of /proc) to reduce overhead.- Add rate-limiting, backoff, logging, and automated remediation hooks (notify PagerDuty, capture profiles, restart).