fix: validate atomic configuration updates and expire stale monitoring snapshots

This commit is contained in:
way
2026-10-18 18:15:22 +08:00
parent c82d9ab3c2
commit 9b83937d60
10 changed files with 244 additions and 85 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
# PIGWay Raspberry Pi Agent v3.11.2 # PIGWay Raspberry Pi Agent v3.11.3
[简体中文](README.zh-CN.md) | English [简体中文](README.zh-CN.md) | English
+1 -1
View File
@@ -1,4 +1,4 @@
# PIGWay Raspberry Pi Agent v3.11.2 # PIGWay Raspberry Pi Agent v3.11.3
简体中文 | [English](README.md) 简体中文 | [English](README.md)
+7 -5
View File
@@ -1,7 +1,9 @@
# Raspberry Pi Agent v3.11.2 # Raspberry Pi Agent v3.11.3
新增磁盘只由用户手动扫描登记,已知磁盘容量及根分区兼容指标每周更新;服务重启复用运行缓存。保留现有CPU、内存、网络、硬件插件及监控协议。 - 监控快照超过三秒停止续期硬件插件租约,状态 API 对超过十秒的状态返回 503,避免采集停滞仍显示为新状态。
- 配置 API 校验有限数值、范围、阈值关系和服务目标;保留名称及进程表达式中的字面百分号。
- 配置写入使用进程内事务锁、唯一备份及原子替换,避免并发覆盖。
- 告警日志按严重程度记录,严重程度变化单独记事件;主要告警优先按严重程度排序,磁盘提示使用实际阈值。
- 根分区缓存恢复期限与七天采集周期一致。
磁盘清单持久化,支持名称/显示设置;整机重启后容量缓存首次初始化不自动发现新盘。没有调整风扇、RGB或OLED控制协议。 本地 17 项回归测试通过,覆盖并发配置、旧状态拒绝与恢复、快照停滞、告警和缓存。本记录不作为当前生产状态的证明。
验证:10项后端测试通过;pigway-gateway生产升级后原两个服务正常,容量周期为604800秒,配置和令牌校验和保留。未执行整机重启或七天自然周期测试。
+18 -14
View File
@@ -1,20 +1,24 @@
ecf1884bb89b86ed4573d4d922a971abde43772bf7b2ffa6a999a12fbfb72caf .gitignore 516ed5bf52a34f3a842dcc050e015106f33cd856987cff1e8005ddeee73452ea .gitignore
a16429e43b2876ac4587479c3c3b5c7e3b5c666d7e889c2178a34a4ae9de8deb README.md 9c90aa203d27dea1ff6493ff401ce252daa50b340672aa02b84e6729387ffe78 README.md
5a81a035871415f88fee6f12e20a42c8ff606f90e13fd383b8dd9407fe8aa909 README.zh-CN.md cf797200392a367fbe2f4fc79deabcd430249299947bd2d1d8d8cace49bfb97b README.zh-CN.md
dd98abfbdfc25c58783084b286d8324a49e5fb0410935f25be04ae9f75f7a2fc RELEASE_NOTES.md
c5853f283c0951b43f2bacbc49e66c26d3a605a184d427828e042eb9236d2740 app/network_rates.py c5853f283c0951b43f2bacbc49e66c26d3a605a184d427828e042eb9236d2740 app/network_rates.py
69877b4ad2b8307db69495d629988ff4bca838dea9e36d839165029fdb6ed4a8 app/pigway_pi_control.py 5f553c03839a2acb86265b24db656b358b4cde6b3c617eef97c9316868578e1b app/pigway_pi_control.py
757eb6a83ca07740a2389af9b51c6fd5d940b4eb86abbc603f231409416d45db app/pigway_pi_control_api.py c3278a0c09f7529b50460755eed5470893d6598002ac0e0d47bf6247b100aa77 app/pigway_pi_control_api.py
3a8a368bbf299eb2d6cccc968a4aacf070058bddebdc5176083a97ac579d720d app/plugin_api.py 9e011761ddb9583c26505fd85d6b484c440b649cb6169922512020f95a5a7d42 app/plugin_api.py
81ca44b8bdc384c7f27a39da8e785af2450241c65200edee61db95f48d67f058 app/storage_cache.py 6429ba4e65baae379e892b8c2f15a5a158924b72060a692f0a3b447f8bebfc80 app/storage_cache.py
d5a783b64a10fa43191a813ef73ef8f762e359b0bd42ec155724fa46216298b0 config/pigway-pi-control.conf 14e0e3337dde66309418658d14d1c61f1742c0f327b76a57126c72f006448165 config/pigway-pi-control.conf
e3bf3e7070ea62c0526791bed922688c3fb49fee36ca9c375d080119968feb72 docs/HARDWARE_PLUGIN_API.md e3bf3e7070ea62c0526791bed922688c3fb49fee36ca9c375d080119968feb72 docs/HARDWARE_PLUGIN_API.md
b54ec1f3e9fb15473910686c54142a85d387e44ea5df427a4cf1b06052e789b3 docs/RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.md b54ec1f3e9fb15473910686c54142a85d387e44ea5df427a4cf1b06052e789b3 docs/RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.md
9222a2b6a38f6b6d7d88b30c05413c68c05f69ee499c8479eca05ee7c62e87c3 docs/RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.zh-CN.md 9222a2b6a38f6b6d7d88b30c05413c68c05f69ee499c8479eca05ee7c62e87c3 docs/RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.zh-CN.md
1b0aeaf4997bdb02b967d538cf276ad7462a3e22ec35fc0d046d779ef56cd1f8 install.sh
8a593c290da5bd0d037aee86c35264512afd574a2a0f60f2c29e9db63e0109bd systemd/pigway-pi-control-api.service
1cb55de6b104a233066495e9786319305fa4f1b6f25483c108ab4f5ac9aa5700 systemd/pigway-pi-control.service
8b97229abfb31c68464eb5754bc3a8fec7a6a78e216d74a86b7cc89486700740 tests/test_network_rates.py
f78e31e0eeee4e02532885ace5c8f6f7011d084e4c7440b704408bc88495e40e tests/test_storage_cache.py
9f59e7313ee58e687ae6dd29d175c67082674254ac5c5f54168b2735b64c20a4 uninstall.sh
67f56bdb128d64f434ca52e45245fd2f3548da3852b3e06cc8def8c804502bf0 docs/README.md 67f56bdb128d64f434ca52e45245fd2f3548da3852b3e06cc8def8c804502bf0 docs/README.md
6552741bb5639520c331a8c38d5d99c90f0ab3d9b1f11e571cb789c386bd0577 docs/STORAGE.md 6552741bb5639520c331a8c38d5d99c90f0ab3d9b1f11e571cb789c386bd0577 docs/STORAGE.md
0eddd7adb18bfdf801e93fde28aa06ef28db3011b4e5f60015f0c641d62c6de1 install.sh
8a593c290da5bd0d037aee86c35264512afd574a2a0f60f2c29e9db63e0109bd systemd/pigway-pi-control-api.service
1cb55de6b104a233066495e9786319305fa4f1b6f25483c108ab4f5ac9aa5700 systemd/pigway-pi-control.service
a58ff2e2ae49f481618cbb8d83b7cfbaa08ed4361eecbcab54755c886a6163eb tests/test_audit_regressions.py
8b97229abfb31c68464eb5754bc3a8fec7a6a78e216d74a86b7cc89486700740 tests/test_network_rates.py
56a7b171f7b6d93721b4888908929b36095a4cb73700a041e37f9bc796309c5e tests/test_storage_cache.py
ab555730f3f3955a1338d5d4282addfb2ea1703f6a50541146dc49adbca5694b tests/test_storage_config.py
7ef40a31988bc258eb490d39732a1790e6e533b045c3f3ff277d2e0fd274dfa1 tests/test_storage_inventory.py
9f59e7313ee58e687ae6dd29d175c67082674254ac5c5f54168b2735b64c20a4 uninstall.sh
+7 -6
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import time,socket,subprocess,configparser,signal,re,argparse,json,os,platform import time,socket,subprocess,configparser,signal,re,argparse,json,os,platform
from pathlib import Path from pathlib import Path
APP_VERSION="3.11.2" APP_VERSION="3.11.3"
import sys import sys
sys.path.append("/usr/local/share/pigway-pi-control") sys.path.append("/usr/local/share/pigway-pi-control")
from plugin_api import Publisher,registered,discover from plugin_api import Publisher,registered,discover
@@ -185,7 +185,7 @@ def internet():
except: return False except: return False
def load_cfg(): def load_cfg():
c=configparser.ConfigParser(); c.read(CFG) c=configparser.ConfigParser(interpolation=None); c.read(CFG)
watches=[] watches=[]
if c.has_section("services"): if c.has_section("services"):
for label,target in c["services"].items(): for label,target in c["services"].items():
@@ -331,10 +331,10 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
else: alert_manager.clear("MEMORY_HIGH") else: alert_manager.clear("MEMORY_HIGH")
if disk is not None: if disk is not None:
if disk>=DISK_CRIT: if disk>=DISK_CRIT:
alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%","FREE < 5%","CHECK STORAGE") alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%",f"USED >= {DISK_CRIT:g}%","CHECK STORAGE")
alert_manager.clear("DISK_HIGH") alert_manager.clear("DISK_HIGH")
elif disk>=DISK_HIGH: elif disk>=DISK_HIGH:
alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%","FREE < 10%","CHECK STORAGE") alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%",f"USED >= {DISK_HIGH:g}%","CHECK STORAGE")
alert_manager.clear("DISK_CRITICAL") alert_manager.clear("DISK_CRITICAL")
else: else:
alert_manager.clear("DISK_HIGH"); alert_manager.clear("DISK_CRITICAL") alert_manager.clear("DISK_HIGH"); alert_manager.clear("DISK_CRITICAL")
@@ -408,13 +408,14 @@ class AlertManager:
"first_seen":old["first_seen"] if old else now,"metadata":metadata or {}} "first_seen":old["first_seen"] if old else now,"metadata":metadata or {}}
value["last_changed"]=old["last_changed"] if old and all(old[k]==v for k,v in value.items()) else now value["last_changed"]=old["last_changed"] if old and all(old[k]==v for k,v in value.items()) else now
self.requests[request_id]=value self.requests[request_id]=value
if old is None:log("WARN","ALERT_ACTIVE",alert=request_id,severity=severity,detail=l2) if old is None or old["severity"]!=severity:
log({1:"WARN",2:"CRITICAL",3:"EMERGENCY"}[severity],"ALERT_ACTIVE" if old is None else "ALERT_CHANGED",alert=request_id,severity=severity,detail=l2)
def clear(self,request_id): def clear(self,request_id):
if self.requests.pop(request_id,None) is not None:log("INFO","ALERT_RECOVERED",alert=request_id) if self.requests.pop(request_id,None) is not None:log("INFO","ALERT_RECOVERED",alert=request_id)
def status_snapshot(self): def status_snapshot(self):
values=sorted(self.requests.values(),key=lambda a:(-a["priority"],a["first_seen"],a["id"])) values=sorted(self.requests.values(),key=lambda a:(-a["severity"],-a["priority"],a["first_seen"],a["id"]))
return {"owner":values[0]["id"] if values else "NORMAL_HOME","alerts":values} return {"owner":values[0]["id"] if values else "NORMAL_HOME","alerts":values}
+67 -52
View File
@@ -2,26 +2,30 @@
import sys import sys
sys.path.append("/usr/local/share/pigway-pi-control") sys.path.append("/usr/local/share/pigway-pi-control")
from plugin_api import discover,registered,request as plugin_request,socket_path from plugin_api import discover,registered,request as plugin_request,socket_path
import math,functools,tempfile
import configparser,ipaddress,json,re,secrets,shutil,subprocess,threading,time,socket import configparser,ipaddress,json,re,secrets,shutil,subprocess,threading,time,socket
from concurrent.futures import ThreadPoolExecutor,as_completed
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from pathlib import Path from pathlib import Path
STORAGE_SCAN_LOCK=threading.Lock() from urllib.parse import parse_qs,urlparse
from urllib.parse import parse_qs,urlencode,urlparse
from urllib.request import Request,urlopen
APP_VERSION="3.11.2" STORAGE_SCAN_LOCK=threading.Lock()
CONFIG_LOCK=threading.RLock()
def config_transaction(function):
@functools.wraps(function)
def locked(*args,**kwargs):
with CONFIG_LOCK:
return function(*args,**kwargs)
return locked
APP_VERSION="3.11.3"
CFG=Path("/etc/pigway-pi-control.conf") CFG=Path("/etc/pigway-pi-control.conf")
STATUS=Path("/run/pigway-pi-control/status.json") STATUS=Path("/run/pigway-pi-control/status.json")
TOKEN=Path("/etc/pigway-pi-control-api.token") TOKEN=Path("/etc/pigway-pi-control-api.token")
ALLOWED_SECTIONS={"device","services","processes","alerts","timing","dark_mode","api","storage_names","storage_visible"} ALLOWED_SECTIONS={"device","services","processes","alerts","timing","dark_mode","api","storage_names","storage_visible"}
NAME_RE=re.compile(r"^[A-Za-z0-9_.-]{1,64}$") NAME_RE=re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
UNIT_RE=re.compile(r"^[A-Za-z0-9_.@:-]{1,128}\.service$") UNIT_RE=re.compile(r"^[A-Za-z0-9_.@:-]{1,128}\.service$")
RGB_KEY_RE=re.compile(r"^(cpu|power|memory|storage|network|service|normal)_[rgb]$")
FLASH_KEY_RE=re.compile(r"^(warning|critical|emergency|normal)_flash_(on|off)_ms$")
NORMAL_MODES={"off","solid","flash","flow","breathe","marquee","rainbow","colorful"}
EFFECT_COLORS={"red","green","blue","yellow","purple","cyan","white"}
PROTECTED_UNITS={"pigway-pi-control-api.service"} PROTECTED_UNITS={"pigway-pi-control-api.service"}
def plugin_inventory(): def plugin_inventory():
@@ -36,6 +40,7 @@ def find_plugin(plugin_id):
raise ValueError("unknown plugin") raise ValueError("unknown plugin")
@config_transaction
def plugin_operation(body): def plugin_operation(body):
plugin_id=str(body.get("id","")) plugin_id=str(body.get("id",""))
if not NAME_RE.fullmatch(plugin_id):raise ValueError("invalid plugin id") if not NAME_RE.fullmatch(plugin_id):raise ValueError("invalid plugin id")
@@ -66,7 +71,7 @@ def machine_identity():
return {"id":identifier,"name":name,"hostname":socket.gethostname(),"local":True,"url":""} return {"id":identifier,"name":name,"hostname":socket.gethostname(),"local":True,"url":""}
def load_cfg(): def load_cfg():
cfg=configparser.ConfigParser(); cfg.read(CFG) cfg=configparser.ConfigParser(interpolation=None); cfg.read(CFG)
return cfg return cfg
def api_setting(key,default,cast=str): def api_setting(key,default,cast=str):
@@ -91,36 +96,39 @@ def validate_value(section,key,value,existing):
if section=="timing" and key in {"oled_refresh_interval","page_interval"}:raise ValueError("display timing is managed by the hardware plugin") if section=="timing" and key in {"oled_refresh_interval","page_interval"}:raise ValueError("display timing is managed by the hardware plugin")
if section not in {"services","processes"} and key not in existing.get(section,set()): if section not in {"services","processes"} and key not in existing.get(section,set()):
raise ValueError(f"unsupported option: {section}.{key}") raise ValueError(f"unsupported option: {section}.{key}")
if section=="hardware" and key=="cooling_hat_enabled": if section in {"alerts","timing"}:
if value.lower() not in ("true","false"): raise ValueError("cooling_hat_enabled must be true or false") number=float(value)
elif section=="led" and key=="alert_mode": if not math.isfinite(number):raise ValueError(f"{section}.{key} must be finite")
if value.lower() not in ("breathe","flash"): raise ValueError("alert_mode must be breathe or flash") low,high=(0,86400)
elif section=="led" and key=="custom_min_hold_ms": if section=="timing":
if not 1500<=int(value)<=60000: raise ValueError("custom_min_hold_ms must be 1500..60000") low=0 if key in {"startup_grace_seconds","network_failure_seconds"} else 0.2
elif RGB_KEY_RE.fullmatch(key): high=3600
number=int(value) elif key in {"cpu_high","cpu_recover","memory_high","memory_recover","disk_high","disk_critical"}:low,high=0,100
if not 0<=number<=255: raise ValueError(f"{key} must be 0..255") elif key in {"temp_high","temp_critical"}:low,high=-50,150
elif FLASH_KEY_RE.fullmatch(key): elif key in {"wifi_weak","wifi_recover"}:low,high=-120,0
number=int(value) elif key=="ethernet_min_speed_mbps":low,high=0,1000000
if not 50<=number<=60000: raise ValueError(f"{key} must be 50..60000") if not low<=number<=high:raise ValueError(f"{section}.{key} must be {low}..{high}")
elif section=="led" and key=="normal_mode": if key in {"cpu_high_seconds","memory_high_seconds","ethernet_min_speed_mbps"} and not value.isdecimal():
if value.lower() not in NORMAL_MODES: raise ValueError("normal_mode must be off, solid, flash, flow, breathe, marquee, rainbow or colorful") raise ValueError(f"{key} must be a nonnegative integer")
elif section=="led" and key=="normal_effect_speed": elif section=="dark_mode" and value.lower() not in {"on","off","true","false","0","1","yes","no"}:
if int(value) not in (1,2,3): raise ValueError("normal_effect_speed must be 1, 2 or 3") raise ValueError("LED setting must be on or off")
elif section=="led" and key=="normal_effect_color": elif section=="services" and not UNIT_RE.fullmatch(value):raise ValueError("invalid systemd service name")
if value.lower() not in EFFECT_COLORS: raise ValueError("normal_effect_color must be red, green, blue, yellow, purple, cyan or white") elif section=="api" and key in {"port","log_limit"}:
elif section=="api" and key=="port": limit=65535 if key=="port" else 1000
number=int(value) if not 1<=int(value)<=limit:raise ValueError(f"api.{key} must be 1..{limit}")
if not 1<=number<=65535: raise ValueError("api.port must be 1..65535") elif section=="api" and key=="bind":ipaddress.ip_address(value)
elif section=="api" and key=="log_limit":
number=int(value)
if not 1<=number<=1000: raise ValueError("api.log_limit must be 1..1000")
elif section=="api" and key=="bind":
ipaddress.ip_address(value)
elif section=="device" and key=="identifier" and value.lower()!="auto" and not NAME_RE.fullmatch(value): elif section=="device" and key=="identifier" and value.lower()!="auto" and not NAME_RE.fullmatch(value):
raise ValueError("device.identifier must be auto or 1..64 letters, numbers, dot, underscore or dash") raise ValueError("device.identifier must be auto or 1..64 letters, numbers, dot, underscore or dash")
def validate_candidate(parsed):
for lower,upper in (("cpu_recover","cpu_high"),("memory_recover","memory_high"),
("temp_high","temp_critical"),("disk_high","disk_critical"),("wifi_weak","wifi_recover")):
if parsed.has_option("alerts",lower) and parsed.has_option("alerts",upper):
if not parsed.getfloat("alerts",lower)<parsed.getfloat("alerts",upper):
raise ValueError(f"alerts.{lower} must be less than alerts.{upper}")
def replace_ini_value(section,key,value): def replace_ini_value(section,key,value):
return replace_ini_text(CFG.read_text(),section,key,value) return replace_ini_text(CFG.read_text(),section,key,value)
@@ -139,13 +147,23 @@ def replace_ini_text(text,section,key,value):
if value is not None:return text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n") if value is not None:return text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
return text return text
@config_transaction
def save_ini_text(text,prefix="api"): def save_ini_text(text,prefix="api"):
parsed=configparser.ConfigParser(); parsed.read_string(text) parsed=configparser.ConfigParser(interpolation=None); parsed.read_string(text)
backup=CFG.with_name(f"{CFG.name}.bak.{prefix}-{time.strftime('%Y%m%d-%H%M%S')}") validate_candidate(parsed)
backup=CFG.with_name(f"{CFG.name}.bak.{prefix}-{time.time_ns()}")
shutil.copy2(CFG,backup) shutil.copy2(CFG,backup)
temporary=CFG.with_suffix(".tmp"); temporary.write_text(text); temporary.chmod(0o644); temporary.replace(CFG) temporary=None
try:
with tempfile.NamedTemporaryFile(mode="w",dir=CFG.parent,prefix=CFG.name+".",delete=False) as file:
temporary=Path(file.name);file.write(text)
temporary.chmod(0o644);temporary.replace(CFG)
finally:
if temporary is not None:temporary.unlink(missing_ok=True)
return str(backup) return str(backup)
@config_transaction
def update_ini(updates): def update_ini(updates):
if not isinstance(updates,dict): raise ValueError("updates must be an object") if not isinstance(updates,dict): raise ValueError("updates must be an object")
text=CFG.read_text() text=CFG.read_text()
@@ -175,12 +193,7 @@ def update_ini(updates):
text=text[:start]+replacement+text[stop:] text=text[:start]+replacement+text[stop:]
else: else:
text=text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n") text=text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
parsed=configparser.ConfigParser(); parsed.read_string(text) return save_ini_text(text,"api")
backup=CFG.with_name(f"{CFG.name}.bak.api-{time.strftime('%Y%m%d-%H%M%S')}")
shutil.copy2(CFG,backup)
temporary=CFG.with_suffix(".tmp")
temporary.write_text(text); temporary.chmod(0o644); temporary.replace(CFG)
return str(backup)
def journal_query(since=None,until=None,severity="",event="",search="",sort="timestamp",order="desc",limit=100,offset=0): def journal_query(since=None,until=None,severity="",event="",search="",sort="timestamp",order="desc",limit=100,offset=0):
command=["journalctl","-u","pigway-pi-control.service","--no-pager","-o","json"] command=["journalctl","-u","pigway-pi-control.service","--no-pager","-o","json"]
@@ -244,6 +257,7 @@ def service_inventory():
item["protected"]=item["unit"] in PROTECTED_UNITS item["protected"]=item["unit"] in PROTECTED_UNITS
return sorted(units.values(),key=lambda x:(not x["monitored"],x["unit"])) return sorted(units.values(),key=lambda x:(not x["monitored"],x["unit"]))
@config_transaction
def edit_service_monitor(operation,name,target,note="",previous_name=""): def edit_service_monitor(operation,name,target,note="",previous_name=""):
if operation not in {"save","delete","pause","resume"}: raise ValueError("invalid monitor operation") if operation not in {"save","delete","pause","resume"}: raise ValueError("invalid monitor operation")
if not NAME_RE.fullmatch(name): raise ValueError("invalid monitor name") if not NAME_RE.fullmatch(name): raise ValueError("invalid monitor name")
@@ -281,10 +295,7 @@ def edit_service_monitor(operation,name,target,note="",previous_name=""):
change("service_monitor_disabled",previous_name,None) change("service_monitor_disabled",previous_name,None)
change("services",name,target); change("service_notes",name,note or None) change("services",name,target); change("service_notes",name,note or None)
if was_paused: change("service_monitor_disabled",name,"true") if was_paused: change("service_monitor_disabled",name,"true")
parsed=configparser.ConfigParser(); parsed.read_string(text) backup=save_ini_text(text,"api")
backup=CFG.with_name(f"{CFG.name}.bak.api-{time.strftime('%Y%m%d-%H%M%S')}")
shutil.copy2(CFG,backup)
temporary=CFG.with_suffix(".tmp"); temporary.write_text(text); temporary.chmod(0o644); temporary.replace(CFG)
subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10) subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10)
return str(backup) return str(backup)
@@ -298,6 +309,7 @@ def process_monitors():
result.append({"name":name.upper(),"target":target,"up":up}) result.append({"name":name.upper(),"target":target,"up":up})
return result return result
@config_transaction
def edit_process_monitor(operation,name,target="",previous_name=""): def edit_process_monitor(operation,name,target="",previous_name=""):
if operation not in {"save","delete"}: raise ValueError("invalid process monitor operation") if operation not in {"save","delete"}: raise ValueError("invalid process monitor operation")
if not NAME_RE.fullmatch(name): raise ValueError("invalid process monitor name") if not NAME_RE.fullmatch(name): raise ValueError("invalid process monitor name")
@@ -311,7 +323,7 @@ def edit_process_monitor(operation,name,target="",previous_name=""):
else: else:
if previous_name and previous_name.lower()!=name.lower(): text=replace_ini_text(text,"processes",previous_name,None) if previous_name and previous_name.lower()!=name.lower(): text=replace_ini_text(text,"processes",previous_name,None)
text=replace_ini_text(text,"processes",name,target) text=replace_ini_text(text,"processes",name,target)
parsed=configparser.ConfigParser(); parsed.read_string(text) parsed=configparser.ConfigParser(interpolation=None); parsed.read_string(text)
backup=save_ini_text(text,"api") backup=save_ini_text(text,"api")
subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10) subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10)
return backup return backup
@@ -359,7 +371,10 @@ class Handler(BaseHTTPRequestHandler):
self.send_json(200,{"ok":True,"version":APP_VERSION,"status_available":STATUS.exists()});return self.send_json(200,{"ok":True,"version":APP_VERSION,"status_available":STATUS.exists()});return
if parsed.path=="/api/v1/status": if parsed.path=="/api/v1/status":
if not STATUS.exists(): self.send_json(503,{"error":"agent status unavailable"}); return if not STATUS.exists(): self.send_json(503,{"error":"agent status unavailable"}); return
self.send_json(200,json.loads(STATUS.read_text())); return state=json.loads(STATUS.read_text())
if not 0<=time.time()-state.get("timestamp",0)<=10:
self.send_json(503,{"error":"agent status stale"});return
self.send_json(200,state); return
if parsed.path=="/api/v1/logs": if parsed.path=="/api/v1/logs":
query=parse_qs(parsed.query) query=parse_qs(parsed.query)
value=lambda key,default="":query.get(key,[default])[0] value=lambda key,default="":query.get(key,[default])[0]
+9 -4
View File
@@ -4,6 +4,7 @@ from contextlib import closing
import http.client import http.client
import json import json
import socket import socket
import time
import threading import threading
import uuid import uuid
from pathlib import Path from pathlib import Path
@@ -65,14 +66,15 @@ class Publisher:
"""One replaceable snapshot, not an unbounded queue of animation commands.""" """One replaceable snapshot, not an unbounded queue of animation commands."""
def __init__(self,plugins,source,log): def __init__(self,plugins,source,log):
self.plugins=plugins;self.source=source;self.log=log self.plugins=plugins;self.source=source;self.log=log
self.session=str(uuid.uuid4());self.revision=0;self.snapshot=None self.session=str(uuid.uuid4());self.revision=0;self.snapshot=None;self.published_at=None
self.lock=threading.Lock();self.stop_event=threading.Event();self.status={} self.lock=threading.Lock();self.stop_event=threading.Event();self.status={}
self.threads=[threading.Thread(target=self.run,args=(key,path),name='plugin-'+key,daemon=True) self.threads=[threading.Thread(target=self.run,args=(key,path),name='plugin-'+key,daemon=True)
for key,path in plugins.items()] for key,path in plugins.items()]
for thread in self.threads:thread.start() for thread in self.threads:thread.start()
def publish(self,state): def publish(self,state):
with self.lock:self.snapshot=copy.deepcopy(state) with self.lock:
self.snapshot=copy.deepcopy(state);self.published_at=time.monotonic()
def current_status(self): def current_status(self):
with self.lock:return copy.deepcopy(self.status) with self.lock:return copy.deepcopy(self.status)
@@ -80,8 +82,11 @@ class Publisher:
def run(self,key,path): def run(self,key,path):
previous_error=None;revision=0;descriptor=None previous_error=None;revision=0;descriptor=None
while not self.stop_event.is_set(): while not self.stop_event.is_set():
with self.lock:state=self.snapshot with self.lock:
if state is not None: state=self.snapshot
fresh=self.published_at is not None and time.monotonic()-self.published_at<3
# Never renew a hardware lease from a stalled monitoring loop.
if state is not None and fresh:
revision+=1 revision+=1
payload={'api_version':1,'source':self.source,'session':self.session,'revision':revision, payload={'api_version':1,'source':self.source,'session':self.session,'revision':revision,
'ttl_seconds':10,'system':state['system'],'network':state['network'],'alerts':state['display']['alerts']} 'ttl_seconds':10,'system':state['system'],'network':state['network'],'alerts':state['display']['alerts']}
+1 -1
View File
@@ -174,7 +174,7 @@ class StorageCache:
values = data['values'] values = data['values']
if all(isinstance(values[k], int) and not isinstance(values[k], bool) and values[k] >= 0 for k in ('total', 'used', 'free')) and values['total'] > 0 and values['used'] <= values['total']: if all(isinstance(values[k], int) and not isinstance(values[k], bool) and values[k] >= 0 for k in ('total', 'used', 'free')) and values['total'] > 0 and values['used'] <= values['total']:
at = data['observed_at'] at = data['observed_at']
if isinstance(at, (int, float)) and 0 <= self.clock() - at < 90000: if isinstance(at, (int, float)) and 0 <= self.clock() - at < 604800:
self.values, self.observed_at = values, at self.values, self.observed_at = values, at
if not self.error:self.next_update_at = at + 604800 if not self.error:self.next_update_at = at + 604800
except (OSError, ValueError, KeyError, TypeError): except (OSError, ValueError, KeyError, TypeError):
+1 -1
View File
@@ -35,7 +35,7 @@ fi
python3 - <<'PY' python3 - <<'PY'
from pathlib import Path from pathlib import Path
import configparser,re import configparser,re
p=Path('/etc/pigway-pi-control.conf');text=p.read_text();c=configparser.ConfigParser();c.read_string(text) p=Path('/etc/pigway-pi-control.conf');text=p.read_text();c=configparser.ConfigParser(interpolation=None);c.read_string(text)
for section,values in {'device':{'identifier':'auto','name':'auto'},'plugins':{}, for section,values in {'device':{'identifier':'auto','name':'auto'},'plugins':{},
'alerts':{'ethernet_min_speed_mbps':'1000','ethernet_speed_seconds':'15'}, 'alerts':{'ethernet_min_speed_mbps':'1000','ethernet_speed_seconds':'15'},
'api':{'bind':'0.0.0.0','port':'6001','log_limit':'200'}}.items(): 'api':{'bind':'0.0.0.0','port':'6001','log_limit':'200'}}.items():
+132
View File
@@ -0,0 +1,132 @@
"""Regression checks for configuration, alert truth and stale snapshot handling."""
import ast
import concurrent.futures
import json
import sys
import tempfile
import threading
import time
import unittest
from http.server import ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from urllib.error import HTTPError
from urllib.request import Request, urlopen
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'app'))
import pigway_pi_control_api as api
import plugin_api
from storage_cache import StorageCache
class ConfigTests(unittest.TestCase):
def setUp(self):
self.directory=tempfile.TemporaryDirectory();self.addCleanup(self.directory.cleanup)
self.config=Path(self.directory.name)/'agent.conf'
self.config.write_text((ROOT/'config/pigway-pi-control.conf').read_text())
self.override=patch.object(api,'CFG',self.config);self.override.start();self.addCleanup(self.override.stop)
def test_invalid_values_and_relationships_leave_configuration_unchanged(self):
original=self.config.read_text()
for updates in [
{'timing':{'sensor_interval':'-1'}},{'timing':{'sensor_interval':'0'}},
{'alerts':{'cpu_high':'nan'}},{'alerts':{'memory_high':'inf'}},
{'alerts':{'cpu_high':'101'}},{'alerts':{'cpu_high':'50'}},
{'alerts':{'temp_high':'80'}},{'alerts':{'disk_high':'99'}},
{'alerts':{'wifi_weak':'-60'}},{'alerts':{'cpu_high_seconds':'1.5'}},
{'services':{'BAD':'not-a-unit'}},
]:
with self.subTest(updates=updates):
with self.assertRaises(ValueError):api.update_ini(updates)
self.assertEqual(self.config.read_text(),original)
api.update_ini({'alerts':{'cpu_high':'60','cpu_recover':'50'}})
self.assertEqual(api.load_cfg().getfloat('alerts','cpu_high'),60)
def test_literal_percent_in_identity_and_process_matches(self):
api.update_ini({'device':{'name':'Office 100%'},'processes':{'APP':'worker --ratio=100%'}})
self.assertEqual(api.machine_identity()['name'],'Office 100%')
self.assertEqual(api.load_cfg().get('processes','app'),'worker --ratio=100%')
def test_parallel_writes_preserve_every_edit_and_backup(self):
def save(index):return api.update_ini({'processes':{f'APP{index}':f'worker-{index}'}})
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
backups=list(pool.map(save,range(20)))
self.assertEqual(len(set(backups)),20)
cfg=api.load_cfg()
for index in range(20):self.assertEqual(cfg.get('processes',f'app{index}'),f'worker-{index}')
self.assertFalse(list(self.config.parent.glob('agent.conf.*tmp*')))
class StateTests(unittest.TestCase):
def test_weekly_root_cache_survives_restart_after_first_day(self):
with tempfile.TemporaryDirectory() as directory:
reads=[]
def reader(_):reads.append(1);return SimpleNamespace(total=100,used=20,free=80)
path=Path(directory)/'cache'
StorageCache(path,reader,lambda:1000).sample()
restored=StorageCache(path,reader,lambda:101000)
restored.sample();self.assertEqual(len(reads),1)
self.assertEqual(restored.next_update_at,605800)
StorageCache(path,reader,lambda:605800).sample();self.assertEqual(len(reads),2)
def test_publisher_stops_renewing_stale_data_and_recovers(self):
now=[0];calls=[]
publisher=plugin_api.Publisher({},'test',lambda *a,**k:None)
class Event:
def is_set(self):return now[0]>=8
def wait(self,_):
now[0]+=1
if now[0]==6:publisher.publish(state)
publisher.stop_event=Event()
state={'system':{},'network':{},'display':{'alerts':[]}}
def request(path,method,endpoint,payload=None):
if method=='PUT':calls.append((now[0],payload['revision']))
return {'api_version':1}
with patch.object(plugin_api.time,'monotonic',lambda:now[0]),patch.object(plugin_api,'request',request):
publisher.publish(state);publisher.run('test','unused')
self.assertEqual(calls,[(0,1),(1,2),(2,3),(6,4),(7,5)])
def test_status_api_rejects_stale_agent_then_accepts_fresh_data(self):
with tempfile.TemporaryDirectory() as directory:
path=Path(directory)/'status.json';path.write_text(json.dumps({'timestamp':time.time()-60}))
token=Path(directory)/'token';token.write_text('test-token')
with patch.object(api,'STATUS',path),patch.object(api,'TOKEN',token):
server=ThreadingHTTPServer(('127.0.0.1',0),api.Handler)
worker=threading.Thread(target=server.serve_forever);worker.start()
try:
req=Request(f'http://127.0.0.1:{server.server_port}/api/v1/status',headers={'Authorization':'Bearer test-token'})
with self.assertRaises(HTTPError) as caught:urlopen(req)
self.assertEqual(caught.exception.code,503)
caught.exception.close()
path.write_text(json.dumps({'timestamp':time.time()}))
with urlopen(req) as response:self.assertEqual(response.status,200)
finally:server.shutdown();worker.join();server.server_close()
def test_alert_severity_owner_changes_and_configured_disk_text(self):
tree=ast.parse((ROOT/'app/pigway_pi_control.py').read_text())
names={'AlertManager','alert_priority','update_active_alerts'}
nodes=[n for n in tree.body if isinstance(n,(ast.ClassDef,ast.FunctionDef)) and n.name in names]
logs=[]
env={'time':time,'log':lambda *args,**kwargs:logs.append((args,kwargs)),
'ALERT_PRIORITIES':{'CPU_HIGH':45,'DISK_CRITICAL':55,'DISK_HIGH':40,'TEMP_HIGH':50,'TEMP_CRITICAL':90},
'ALERT_CATEGORIES':{'CPU_HIGH':'CPU','DISK_CRITICAL':'STORAGE','DISK_HIGH':'STORAGE','TEMP_HIGH':'CPU','TEMP_CRITICAL':'CPU'}}
exec(compile(ast.Module(body=nodes,type_ignores=[]),'monitor-functions','exec'),env)
manager=env['AlertManager']()
manager.request('CPU_HIGH',1,'CPU');manager.request('DISK_CRITICAL',2,'DISK')
self.assertEqual(logs[-1][0][0],'CRITICAL')
manager.request('TEMP_HIGH',3,'TEMP')
self.assertEqual(manager.status_snapshot()['owner'],'TEMP_HIGH')
self.assertEqual(logs[-1][0][0],'EMERGENCY')
before=len(logs);manager.request('TEMP_HIGH',3,'TEMP');self.assertEqual(len(logs),before)
manager.request('TEMP_HIGH',1,'TEMP');self.assertEqual(logs[-1][0][1],'ALERT_CHANGED')
env.update(alert_manager=manager,cpu_hi=0,mem_hi=0,CPU_HIGH=90,CPU_RECOVER=75,CPU_HIGH_SEC=30,
MEM_HIGH=90,MEM_RECOVER=85,MEM_HIGH_SEC=30,TEMP_HIGH=68,TEMP_CRIT=75,
DISK_HIGH=70,DISK_CRIT=80,link_ok=True,net_ok=True,watches=[],watch_state=[],
net_kind='NET',health={},fs_readonly=False,ethernet_slow_active=False,
transition=lambda *a,**k:None,net_iface='',net_speed_mbps=None,ETHERNET_MIN_SPEED=1000)
env['update_active_alerts'](10,40,20,85)
self.assertEqual(manager.requests['DISK_CRITICAL']['l3'],'USED >= 80%')
env['update_active_alerts'](10,40,20,75)
self.assertEqual(manager.requests['DISK_HIGH']['l3'],'USED >= 70%')