Compare commits

...
7 Commits
23 changed files with 798 additions and 100 deletions
+22
View File
@@ -0,0 +1,22 @@
name: Verify
on:
push:
branches: [main]
pull_request:
workflow_dispatch:
permissions:
contents: read
jobs:
test:
# Fork code may run on GitHub-hosted runners, never the trusted Gitea runner.
if: github.server_url == 'https://github.com' || github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
with:
persist-credentials: false
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
with:
python-version: '3.13'
- run: python -m unittest discover -s tests -v
- run: git diff --check
+7
View File
@@ -3,3 +3,10 @@ __pycache__/
*.pyc *.pyc
*.pyo *.pyo
*.log *.log
.data/
dist/
/storage-release.json
/ops/deploy-storage.py
/ops/release-storage.py
.pytest_cache/
.ruff_cache/
+9
View File
@@ -0,0 +1,9 @@
# 贡献指南 / Contributing
安装和运行步骤见项目首页;使用 Python 3.13 或 3.14,修改保持小而集中。运行 `python3 -m unittest discover -s tests -v`,Shell 脚本使用 `bash -n`,提交前运行 `git diff --check`。
测试只能使用临时目录、模拟服务和设备桩;不得访问真实设备、执行生产写入或提交密钥、令牌、数据库、照片及部署回执。真实设备验证由维护者另行执行。
请通过仓库 Pull Request 提交,说明问题、行为变化和验证结果。行为修复应覆盖对应回归风险。新增代码按本项目 MIT 许可证提供,保留第三方许可及来源。
See README.md for installation. Keep changes focused, test with mocks, and submit a pull request describing the behavior and checks. Never include credentials or production data. Contributions use the MIT license; preserve third-party notices.
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 PIGWay
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+4 -8
View File
@@ -1,7 +1,9 @@
# PIGWay Raspberry Pi Agent v3.7.1 # PIGWay Raspberry Pi Agent v3.11.3
[简体中文](README.zh-CN.md) | English [简体中文](README.zh-CN.md) | English
Disk discovery is manual: Device Console → System configuration → select the device → Storage settings → Scan disks. The inventory persists across restarts; known filesystem capacity updates weekly. Names and visibility belong to the Agent. See [deployment and validation](docs/STORAGE.md).
PIGWay Raspberry Pi Agent is a lightweight local monitoring service for Raspberry Pi. It monitors CPU, memory, temperature, disk, networking, services, processes, system health alerts, and journal events. Ethernet monitoring reads the negotiated link speed and can alert when it remains below the configured minimum. It has no embedded Web UI and does not require a hardware plugin or Device Console. PIGWay Raspberry Pi Agent is a lightweight local monitoring service for Raspberry Pi. It monitors CPU, memory, temperature, disk, networking, services, processes, system health alerts, and journal events. Ethernet monitoring reads the negotiated link speed and can alert when it remains below the configured minimum. It has no embedded Web UI and does not require a hardware plugin or Device Console.
@@ -56,10 +58,4 @@ Logs follow the system journal retention policy; no additional database is creat
This project targets Raspberry Pi. Other Linux platforms are not covered by its support claim and may have separately maintained adapters. Repository naming does not change the existing `pigway-agent` adapter ID or `pigway-pi-control` service, configuration and API identifiers. This project targets Raspberry Pi. Other Linux platforms are not covered by its support claim and may have separately maintained adapters. Repository naming does not change the existing `pigway-agent` adapter ID or `pigway-pi-control` service, configuration and API identifiers.
## v3.8.0 网络流量 [Documentation](docs/README.md)
主接口每个网络采集周期读取 sysfs 的 rx_bytes/tx_bytes,以单调时间差计算字节/秒,API 新增 rx_bytes_per_second 和 tx_bytes_per_second。首次采样、接口切换、计数器重置或读取失败返回空值;不修改监控告警、硬件插件控制或 API 可选安装行为。2026-10-14 已在 Gateway 部署,实机上下行速率与 Console 标准历史指标验证通过。
## v3.9.0 采样时间
API 状态新增 system.observed_at、network.observed_at 和 collection.interval_seconds。Console 可每秒读取缓存,网络等低频指标保留实际采样时间,供历史去重。数据来源与原网络采样周期保持。
+5 -1
View File
@@ -1,7 +1,9 @@
# PIGWay Raspberry Pi Agent v3.7.1 # PIGWay Raspberry Pi Agent v3.11.3
简体中文 | [English](README.md) 简体中文 | [English](README.md)
新增磁盘由 Console → 系统配置 → 选择设备 → 磁盘设置 → 扫描磁盘手动发现;已保存清单跨重启保留,已知磁盘容量每周刷新。支持在 Agent 保存名称和隐藏设置。见 [部署与验证](docs/STORAGE.md)。
极轻的树莓派本地监控服务:CPU、内存、温度、磁盘、网络、服务/进程、系统健康告警与 journal 日志。有线网络会读取真实协商速率,并可在持续低于配置阈值时告警。无 Web 界面;无需硬件插件,也无需管理中心。 极轻的树莓派本地监控服务:CPU、内存、温度、磁盘、网络、服务/进程、系统健康告警与 journal 日志。有线网络会读取真实协商速率,并可在持续低于配置阈值时告警。无 Web 界面;无需硬件插件,也无需管理中心。
@@ -57,3 +59,5 @@ sudo journalctl -u pigway-pi-control.service -f
## 名称与兼容性 ## 名称与兼容性
本项目支持范围明确为树莓派;其他 Linux 平台以后独立维护项目。仓库改名不改变现有 `pigway-pi-control` 服务名、配置路径和 API。远程 API 仍可选择启用,默认关闭。 本项目支持范围明确为树莓派;其他 Linux 平台以后独立维护项目。仓库改名不改变现有 `pigway-pi-control` 服务名、配置路径和 API。远程 API 仍可选择启用,默认关闭。
[文档目录](docs/README.md)
+13
View File
@@ -0,0 +1,13 @@
# pigway-raspberry-pi-agent v3.11.3
- 监控快照超过三秒停止续期硬件插件租约,状态 API 对超过十秒的状态返回 503,避免采集停滞仍显示为新状态。
- 配置 API 校验有限数值、范围、阈值关系和服务目标;保留名称及进程表达式中的字面百分号。
- 配置写入使用进程内事务锁、唯一备份及原子替换,避免并发覆盖。
- 告警日志按严重程度记录,严重程度变化单独记事件;主要告警优先按严重程度排序,磁盘提示使用实际阈值。
- 根分区缓存恢复期限与七天采集周期一致。
本地 17 项回归测试通过,覆盖并发配置、旧状态拒绝与恢复、快照停滞、告警和缓存。本记录不作为当前生产状态的证明。
统一开源许可证、贡献和安全反馈入口;完善当前版本的文档导航、源码分发和校验和。插件 ZIP 携带许可证。运行功能边界保持项目首页所述。
本轮 17 项 Python 测试通过。源码语法、文档入口、打包和校验和检查通过;生产运行状态以实际部署验收为准。
+7
View File
@@ -0,0 +1,7 @@
# 安全反馈 / Security
仅支持当前发布版本。请勿在公开 Issue 中发布密钥、令牌、个人数据或可直接利用的完整漏洞细节。
请通过仓库维护者 [way](https://tea.pigway.com/way) 的私有联系渠道报告,包含受影响版本、复现条件、影响和最小化示例。若尚未建立私有渠道,先在 Issue 请求私下联系,不附敏感细节。
Only the current release is supported. Contact the maintainer privately before disclosing exploit details; include the version, impact and a minimal reproduction. Do not attach credentials or private data to public issues.
+21 -10
View File
@@ -1,17 +1,28 @@
ecf1884bb89b86ed4573d4d922a971abde43772bf7b2ffa6a999a12fbfb72caf .gitignore 1db9d786fdc6c2ae84b7d317e981b23e303a5bc0b72704d43558b29373d97ade .github/workflows/verify.yml
ba2329dc02e9e09f5989c0dc6f174e0ce36b126b94249092a7ec97eb0e2a1d9d README.md 4035aba62136040755ffdf2faf1556a49215e829c16ee44263e0ab7530fbf521 .gitignore
64a9708701f8c8d872a587e8a2845afff463797b06a2791319daadd705933d31 README.zh-CN.md 79f34290cfa1a6f0928162909848e6597d46183d0de6b213bf52b12546f5fc33 CONTRIBUTING.md
6eb7728e6e73f9568ca6fd5ad6383d13ffc638f2870fefa2d86f72f39685c0fc LICENSE
9c90aa203d27dea1ff6493ff401ce252daa50b340672aa02b84e6729387ffe78 README.md
cf797200392a367fbe2f4fc79deabcd430249299947bd2d1d8d8cace49bfb97b README.zh-CN.md
54fa1c8f12d277d3309d02885d83bb317cdb0eda0ead1f9e16ccf2960c01b552 RELEASE_NOTES.md
6cead6c1378cd840f655adc3d467e2cd6fd3ab368d7200a3afbfe85d9ed159be SECURITY.md
c5853f283c0951b43f2bacbc49e66c26d3a605a184d427828e042eb9236d2740 app/network_rates.py c5853f283c0951b43f2bacbc49e66c26d3a605a184d427828e042eb9236d2740 app/network_rates.py
439602f2f6188558f1f16b6630a65438806d98dabd4c6f7fa5270e9f3845bf54 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
d5a783b64a10fa43191a813ef73ef8f762e359b0bd42ec155724fa46216298b0 config/pigway-pi-control.conf 6429ba4e65baae379e892b8c2f15a5a158924b72060a692f0a3b447f8bebfc80 app/storage_cache.py
9d45f22221012a7fdb3406ce74a6113f618e9930fe460efb3869efd099e3b1d8 docs/DEPLOYMENT-v3.8.0-2026-10-14.md 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
c104443a4c951ea7916f74473510362f684df7d916bf2a436b779513ba3fe396 install.sh 3041278ff521c303d95711d38f31f43dfd38c8ed94999a7a86514b1fa3e09f60 docs/README.md
6552741bb5639520c331a8c38d5d99c90f0ab3d9b1f11e571cb789c386bd0577 docs/STORAGE.md
a38fae07a9151923706ff708598bf24e89ef4ed7fb9449c3f7b4d013d6e1be93 install.sh
8a593c290da5bd0d037aee86c35264512afd574a2a0f60f2c29e9db63e0109bd systemd/pigway-pi-control-api.service 8a593c290da5bd0d037aee86c35264512afd574a2a0f60f2c29e9db63e0109bd systemd/pigway-pi-control-api.service
1cb55de6b104a233066495e9786319305fa4f1b6f25483c108ab4f5ac9aa5700 systemd/pigway-pi-control.service 1cb55de6b104a233066495e9786319305fa4f1b6f25483c108ab4f5ac9aa5700 systemd/pigway-pi-control.service
a58ff2e2ae49f481618cbb8d83b7cfbaa08ed4361eecbcab54755c886a6163eb tests/test_audit_regressions.py
8b97229abfb31c68464eb5754bc3a8fec7a6a78e216d74a86b7cc89486700740 tests/test_network_rates.py 8b97229abfb31c68464eb5754bc3a8fec7a6a78e216d74a86b7cc89486700740 tests/test_network_rates.py
9f59e7313ee58e687ae6dd29d175c67082674254ac5c5f54168b2735b64c20a4 uninstall.sh 56a7b171f7b6d93721b4888908929b36095a4cb73700a041e37f9bc796309c5e tests/test_storage_cache.py
ab555730f3f3955a1338d5d4282addfb2ea1703f6a50541146dc49adbca5694b tests/test_storage_config.py
7ef40a31988bc258eb490d39732a1790e6e533b045c3f3ff277d2e0fd274dfa1 tests/test_storage_inventory.py
273fa86cdb77a0610ef64a3ad1788fecbe465bfbdf9568fa8928304b6be2cfaa uninstall.sh
+35 -17
View File
@@ -1,12 +1,13 @@
#!/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.9.0" 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 shutil import disk_usage
from plugin_api import Publisher,registered,discover from plugin_api import Publisher,registered,discover
from network_rates import NetworkRates from network_rates import NetworkRates
from storage_cache import StorageCache,StorageInventory
storage_cache=StorageCache("/run/pigway-pi-control/storage-cache.json")
CFG="/etc/pigway-pi-control.conf" CFG="/etc/pigway-pi-control.conf"
STATUS_PATH="/run/pigway-pi-control/status.json" STATUS_PATH="/run/pigway-pi-control/status.json"
@@ -144,7 +145,8 @@ def mem_pct():
k,v=l.split(":",1); vals[k]=int(v.split()[0]) k,v=l.split(":",1); vals[k]=int(v.split()[0])
return 100*(vals["MemTotal"]-vals["MemAvailable"])/vals["MemTotal"] return 100*(vals["MemTotal"]-vals["MemAvailable"])/vals["MemTotal"]
def disk_pct(): def disk_pct():
d=disk_usage("/"); return 100*d.used/d.total d=storage_cache.sample()
return 100*d["used"]/d["total"] if d else None
def _run(args,timeout=2): def _run(args,timeout=2):
try: return subprocess.check_output(args,text=True,stderr=subprocess.DEVNULL,timeout=timeout).strip() try: return subprocess.check_output(args,text=True,stderr=subprocess.DEVNULL,timeout=timeout).strip()
except Exception: return "" except Exception: return ""
@@ -183,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():
@@ -197,6 +199,7 @@ def load_cfg():
if target and not target.startswith("#"): watches.append(("process",label.upper(),target)) if target and not target.startswith("#"): watches.append(("process",label.upper(),target))
return c,watches return c,watches
cfg,watches=load_cfg() cfg,watches=load_cfg()
storage_inventory=StorageInventory(cache_path=Path(CFG).with_suffix(".storage.json"),capacity_path="/run/pigway-pi-control/storage-inventory-usage.json")
def cf(section,key,default,cast=float): def cf(section,key,default,cast=float):
try: return cast(cfg.get(section,key)) try: return cast(cfg.get(section,key))
@@ -326,14 +329,15 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
alert_manager.clear("TEMP_HIGH"); alert_manager.clear("TEMP_CRITICAL") alert_manager.clear("TEMP_HIGH"); alert_manager.clear("TEMP_CRITICAL")
if mem_hi>=MEM_HIGH_SEC: alert_manager.request("MEMORY_HIGH",1,"MEMORY HIGH",f"USED {mem:.1f}%","CHECK PROCESS","") if mem_hi>=MEM_HIGH_SEC: alert_manager.request("MEMORY_HIGH",1,"MEMORY HIGH",f"USED {mem:.1f}%","CHECK PROCESS","")
else: alert_manager.clear("MEMORY_HIGH") else: alert_manager.clear("MEMORY_HIGH")
if disk>=DISK_CRIT: if disk is not None:
alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%","FREE < 5%","CHECK STORAGE") if disk>=DISK_CRIT:
alert_manager.clear("DISK_HIGH") alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%",f"USED >= {DISK_CRIT:g}%","CHECK STORAGE")
elif disk>=DISK_HIGH: alert_manager.clear("DISK_HIGH")
alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%","FREE < 10%","CHECK STORAGE") elif disk>=DISK_HIGH:
alert_manager.clear("DISK_CRITICAL") alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%",f"USED >= {DISK_HIGH:g}%","CHECK STORAGE")
else: alert_manager.clear("DISK_CRITICAL")
alert_manager.clear("DISK_HIGH"); alert_manager.clear("DISK_CRITICAL") else:
alert_manager.clear("DISK_HIGH"); alert_manager.clear("DISK_CRITICAL")
if not link_ok: if not link_ok:
alert_manager.request("NETWORK_DOWN",2,"NETWORK DOWN","NO LAN / WIFI","NO IP ADDRESS","") alert_manager.request("NETWORK_DOWN",2,"NETWORK DOWN","NO LAN / WIFI","NO IP ADDRESS","")
alert_manager.clear("INTERNET_DOWN") alert_manager.clear("INTERNET_DOWN")
@@ -404,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}
@@ -423,11 +428,20 @@ def cleanup():
def write_status(cpu,temp,mem,disk): def write_status(cpu,temp,mem,disk):
display_state=alert_manager.status_snapshot() display_state=alert_manager.status_snapshot()
memory_total,memory_used,memory_available,memory_calculated=memory_usage() memory_total,memory_used,memory_available,memory_calculated=memory_usage()
disk_total,disk_used,disk_free=disk_usage("/") storage=storage_cache.sample()
disks=storage_inventory.sample(cfg)
disk_total,disk_used,disk_free=(storage[k] for k in ("total","used","free")) if storage else (None,None,None)
signal=wifi_percent(net_metric) if net_kind=="WIF" else None signal=wifi_percent(net_metric) if net_kind=="WIF" else None
state={ state={
"version":APP_VERSION,"timestamp":int(time.time()), "version":APP_VERSION,"timestamp":int(time.time()),
"collection":{"interval_seconds":min(1.0,SENSOR_INTERVAL,NETWORK_INTERVAL,SERVICE_INTERVAL,HEALTH_INTERVAL)},"uptime_seconds":uptime_seconds(), "collection":{"interval_seconds":min(1.0,SENSOR_INTERVAL,NETWORK_INTERVAL,SERVICE_INTERVAL,HEALTH_INTERVAL),
"refresh":{
"storage_inventory":{"interval_seconds":604800,"observed_at":storage_inventory.observed_at,"next_update_at":storage_inventory.next_update_at,"error":storage_inventory.error},
"system":{"interval_seconds":SENSOR_INTERVAL,"observed_at":sensor_observed_at,"next_update_at":sensor_observed_at+SENSOR_INTERVAL if sensor_observed_at else None},
"network":{"interval_seconds":NETWORK_INTERVAL,"observed_at":network_observed_at,"next_update_at":network_observed_at+NETWORK_INTERVAL if network_observed_at else None},
"storage":{"interval_seconds":604800,"observed_at":storage_cache.observed_at,"next_update_at":storage_cache.next_update_at,"error":storage_cache.error},
"display":{"interval_seconds":1,"next_update_at":time.time()+1},
"alerts":{"interval_seconds":1,"next_update_at":time.time()+1}}},"uptime_seconds":uptime_seconds(),
"machine":{"id":DEVICE_ID,"name":DEVICE_NAME,"hostname":socket.gethostname()}, "machine":{"id":DEVICE_ID,"name":DEVICE_NAME,"hostname":socket.gethostname()},
"device":DEVICE_INFO, "device":DEVICE_INFO,
"system":{"observed_at":sensor_observed_at,"load_average":load_average(),"cpu_percent":round(cpu,1), "system":{"observed_at":sensor_observed_at,"load_average":load_average(),"cpu_percent":round(cpu,1),
@@ -435,7 +449,11 @@ def write_status(cpu,temp,mem,disk):
"memory_percent":round(memory_calculated if memory_total else mem,1), "memory_percent":round(memory_calculated if memory_total else mem,1),
"memory_total_bytes":memory_total,"memory_used_bytes":memory_used, "memory_total_bytes":memory_total,"memory_used_bytes":memory_used,
"memory_available_bytes":memory_available, "memory_available_bytes":memory_available,
"disk_percent":round(disk,1),"disk_total_bytes":disk_total, "storage_observed_at":storage_cache.observed_at,
"storage_inventory":disks,"storage_inventory_observed_at":storage_inventory.observed_at,
"storage_inventory_error":storage_inventory.error,
"storage_scanned_at":storage_inventory.scanned_at,"storage_discovery_mode":"manual",
"disk_percent":round(disk,1) if disk is not None else None,"disk_total_bytes":disk_total,
"disk_used_bytes":disk_used,"disk_free_bytes":disk_free, "disk_used_bytes":disk_used,"disk_free_bytes":disk_free,
"root_readonly":bool(fs_readonly)}, "root_readonly":bool(fs_readonly)},
"power":health, "power":health,
+86 -53
View File
@@ -2,25 +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
from urllib.parse import parse_qs,urlencode,urlparse from urllib.parse import parse_qs,urlparse
from urllib.request import Request,urlopen
APP_VERSION="3.7.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"} 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():
@@ -35,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")
@@ -65,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):
@@ -74,45 +80,55 @@ def api_setting(key,default,cast=str):
def config_json(): def config_json():
cfg=load_cfg() cfg=load_cfg()
return {section:{k:v for k,v in cfg.items(section) if not (section=="timing" and k in {"oled_refresh_interval","page_interval"})} for section in cfg.sections() if section in ALLOWED_SECTIONS} return {section:{k:v for k,v in cfg.items(section,raw=True) if not (section=="timing" and k in {"oled_refresh_interval","page_interval"})} for section in cfg.sections() if section in ALLOWED_SECTIONS}
def validate_value(section,key,value,existing): def validate_value(section,key,value,existing):
if section in {"storage_names","storage_visible"}:
if not re.fullmatch(r"[a-f0-9]{32}",key):raise ValueError("invalid storage identifier")
if section=="storage_visible":
if value.lower() not in {"true","false"}:raise ValueError("visibility must be true or false")
elif len(value)>64 or any(ord(c)<32 for c in value):
raise ValueError("storage name must be at most 64 characters without control characters")
return
if section=="api" and key=="enabled": if section=="api" and key=="enabled":
if value.lower() not in {"true","false"}:raise ValueError("enabled must be true or false") if value.lower() not in {"true","false"}:raise ValueError("enabled must be true or false")
return return
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)
@@ -131,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()
@@ -167,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"]
@@ -236,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")
@@ -273,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)
@@ -290,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")
@@ -303,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
@@ -351,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]
@@ -388,6 +411,16 @@ class Handler(BaseHTTPRequestHandler):
def do_PUT(self): def do_PUT(self):
path=urlparse(self.path).path path=urlparse(self.path).path
if path=="/api/v1/storage/scan":
if not self.authorized():self.send_json(401,{"error":"bearer token required"});return
try:
from storage_cache import StorageInventory
with STORAGE_SCAN_LOCK:
result=StorageInventory(cache_path=CFG.with_suffix(".storage.json")).scan()
self.send_json(200,result)
except Exception:
self.send_json(503,{"error":"storage scan failed; previous inventory retained"})
return
if path not in {"/api/v1/plugins","/api/v1/plugin/config","/api/v1/config","/api/v1/services","/api/v1/services/control","/api/v1/processes"}: self.send_json(404,{"error":"not found"}); return if path not in {"/api/v1/plugins","/api/v1/plugin/config","/api/v1/config","/api/v1/services","/api/v1/services/control","/api/v1/processes"}: self.send_json(404,{"error":"not found"}); return
if not self.authorized(): self.send_json(401,{"error":"bearer token required"}); return if not self.authorized(): self.send_json(401,{"error":"bearer token required"}); return
try: try:
+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']}
+210
View File
@@ -0,0 +1,210 @@
"""Daily filesystem capacity cache, persisted in RAM across monitor restarts."""
import json
import time
from pathlib import Path
from shutil import disk_usage
class StorageInventory:
"""Read host block devices without mounting, probing SMART, or writing disks."""
def __init__(self, clock=time.time, runner=None, mounts_path="/proc/1/mountinfo", usage=None, cache_path=None, capacity_path=None):
import subprocess
self.clock = clock
self.runner = runner or (lambda: subprocess.check_output(
["lsblk", "--json", "--bytes", "--paths", "--output",
"NAME,KNAME,TYPE,SIZE,FSTYPE,UUID,LABEL,MODEL,SERIAL,MAJ:MIN"],
text=True, timeout=5, stderr=subprocess.DEVNULL))
self.mounts_path = Path(mounts_path)
self.usage = usage or (lambda mount: disk_usage("/proc/1/root" + mount))
self.next_update_at = 0
self.observed_at = None
self.error = False
self.devices = []
self.cache_path = Path(cache_path) if cache_path else None
self.scanned_at = None
self.capacity_path = Path(capacity_path) if capacity_path else None
if self.capacity_path:
try:
saved=json.loads(self.capacity_path.read_text())
at=saved['observed_at']
if isinstance(at,(int,float)) and 0 <= self.clock()-at < 604800 and isinstance(saved['devices'],list):
self.devices=saved['devices'];self.observed_at=at
self.scanned_at=saved['scanned_at'];self.next_update_at=at+604800
except (OSError,ValueError,KeyError,TypeError):pass
def scan(self):
if self.cache_path is None:
raise ValueError("storage inventory cache path required")
try:
previous = json.loads(self.cache_path.read_text())
if 0 <= self.clock() - previous["scanned_at"] < 60:
return {"accepted": True, "scanned_at": previous["scanned_at"], "cooldown": True}
except (OSError, ValueError, KeyError, TypeError):
pass
data = json.loads(self.runner())
self.discover(data)
document = {"scanned_at": self.clock(), "inventory": data}
temporary = self.cache_path.with_suffix(".tmp")
with open(temporary,"w",opener=lambda path, flags: __import__("os").open(path, flags, 0o600)) as file:
json.dump(document, file)
temporary.replace(self.cache_path)
return {"accepted": True, "scanned_at": document["scanned_at"], "cooldown": False}
def sample(self, config=None):
data = None
changed = False
if self.cache_path is not None:
try:
document = json.loads(self.cache_path.read_text())
changed = document["scanned_at"] != self.scanned_at
data = document["inventory"]
self.scanned_at = document["scanned_at"]
except FileNotFoundError:
pass
except (OSError, ValueError, KeyError, TypeError):
self.error = True
if data is not None and (changed or self.clock() >= self.next_update_at):
try:
self.devices = self.discover(data)
self.observed_at = self.clock()
self.error = False
except (OSError, ValueError, TypeError, KeyError, __import__("subprocess").SubprocessError):
self.error = True
self.next_update_at = self.clock() + (3600 if self.error else 604800)
if self.capacity_path and not self.error:
try:
self.capacity_path.write_text(json.dumps(dict(devices=self.devices,observed_at=self.observed_at,scanned_at=self.scanned_at)))
except OSError:pass
names = dict(config.items("storage_names", raw=True)) if config is not None and config.has_section("storage_names") else {}
visible = dict(config.items("storage_visible", raw=True)) if config is not None and config.has_section("storage_visible") else {}
result = []
for device in self.devices:
item = dict(device)
item["name"] = names.get(item["id"], "auto")
if item["name"].lower() == "auto":
item["name"] = item["label"] or item["mountpoint"] or item["device"]
item["visible"] = visible.get(item["id"], "true").lower() != "false"
if self.error:
item["state"] = "unknown"
result.append(item)
known = {d["id"] for d in result}
for identifier in sorted((names.keys() | visible.keys()) - known):
result.append(dict(id=identifier, name=names.get(identifier, identifier),
visible=visible.get(identifier, "true").lower() != "false",
state="unknown" if self.error else "missing", device="", mountpoint="",
label="", capacity_bytes=None, used_bytes=None, free_bytes=None,
used_percent=None))
return result
def discover(self, data):
import hashlib
import re
mounts, sources = {}, {}
for line in self.mounts_path.read_text().splitlines():
left, right = line.split(" - ", 1)
fields = left.split()
# Bind mounts and Btrfs subvolumes must not duplicate filesystem capacity.
mount = re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), fields[4])
row = (fields[3] != "/", len(mount), mount)
mounts.setdefault(fields[2], []).append(row)
source = right.split()[1]
if source.startswith("/dev/"):
sources.setdefault(str(Path(source).resolve()), []).append(row)
devices = data["blockdevices"]
if not isinstance(devices, list):
raise ValueError("invalid block device inventory")
result, seen = [], set()
def walk(device, serial="", parent=""):
kind = device.get("type")
if kind in {"loop", "rom", "zram"}:
return
serial = str(device.get("serial") or serial).strip()
children = device.get("children") or []
uuid = str(device.get("uuid") or "").strip()
fs = str(device.get("fstype") or "")
# A parent disk with partitions/RAID/LVM is represented by its usable leaves.
if not children and kind not in {"loop", "rom"} and fs not in {"swap", "linux_raid_member", "LVM2_member", "crypto_LUKS"}:
partition = re.search(r"(\d+)$", str(device.get("kname") or device.get("name") or ""))
suffix = ":" + partition[1] if kind == "part" and partition else ""
source = "uuid:" + uuid if uuid else "serial:" + serial + suffix if serial else "device:" + str(device.get("name"))
identifier = hashlib.sha256(source.encode()).hexdigest()[:32]
if identifier not in seen:
seen.add(identifier)
candidates = mounts.get(device.get("maj:min"), []) or sources.get(str(Path(device["name"]).resolve()), [])
mount = sorted(candidates)[0][2] if candidates else ""
item = dict(id=identifier, device=str(device.get("name") or ""),
label=str(device.get("label") or ""), uuid=uuid,
stable_identity=bool(uuid or serial), model=str(device.get("model") or "").strip(),
mountpoint=mount, filesystem=fs, state="mounted" if mount else "unmounted",
capacity_bytes=int(device.get("size") or 0), used_bytes=None,
free_bytes=None, used_percent=None)
if mount:
try:
usage = self.usage(mount)
if usage.total <= 0:
raise ValueError("invalid filesystem capacity")
item.update(capacity_bytes=usage.total, used_bytes=usage.used,
free_bytes=usage.free, used_percent=round(100 * usage.used / usage.total, 1))
except (OSError, ValueError):
item["state"] = "unknown"
result.append(item)
for child in children:
walk(child, serial, str(device.get("name") or parent))
for device in devices:
walk(device)
return sorted(result, key=lambda d: (d["mountpoint"] != "/", d["mountpoint"], d["device"]))
class StorageCache:
def __init__(self, path, reader=disk_usage, clock=time.time):
self.path, self.reader, self.clock = Path(path), reader, clock
self.values = None
self.observed_at = None
self.next_update_at = 0
self.error = False
try:
data = json.loads(self.path.read_text())
if not isinstance(data,dict):raise ValueError('invalid cache')
retry = data.get('next_update_at')
if data.get('error') is True and isinstance(retry,(int,float)) and self.clock() < retry <= self.clock()+3600:
self.error, self.next_update_at = True, retry
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']:
at = data['observed_at']
if isinstance(at, (int, float)) and 0 <= self.clock() - at < 604800:
self.values, self.observed_at = values, at
if not self.error:self.next_update_at = at + 604800
except (OSError, ValueError, KeyError, TypeError):
pass
def sample(self):
now = self.clock()
if now < self.next_update_at:
return self.values
try:
usage = self.reader('/')
if usage.total <= 0 or not 0 <= usage.used <= usage.total or usage.free < 0:
raise ValueError('invalid capacity')
except (OSError, ValueError):
self.error = True
self.next_update_at = now + 3600
self.persist()
return self.values
self.values = dict(total=usage.total, used=usage.used, free=usage.free)
self.observed_at = now
self.next_update_at = now + 604800
self.error = False
self.persist()
return self.values
def persist(self):
try:
self.path.parent.mkdir(parents=True, exist_ok=True)
self.path.write_text(json.dumps(dict(values=self.values, observed_at=self.observed_at,
next_update_at=self.next_update_at, error=self.error)))
except OSError:
# The in-memory sample remains valid even if the RAM cache cannot be persisted.
pass
+6
View File
@@ -183,3 +183,9 @@ SSH = ssh.service
# #
# 示例默认全部注释,不会监控任何普通进程。 # 示例默认全部注释,不会监控任何普通进程。
# MYAPP = /opt/myapp/server # MYAPP = /opt/myapp/server
# Use status.system.storage_inventory IDs, or edit through Console.
# auto uses the filesystem label, mountpoint, or device path.
[storage_names]
[storage_visible]
# <32-character ID> = false hides a disk while retaining collection.
-5
View File
@@ -1,5 +0,0 @@
# Agent v3.8.0 网络流量
2026-10-14 Gateway 已部署。单独的 NetworkRates 模块读取主接口 sysfs 计数器;网络更新周期计算速率,写入已有状态/API。首次采样、接口切换、复位、读取失败不伪造零值。idle 返回真实 0。安装器加入模块依赖,保持 API 可选安装。采样测试覆盖初始、零流量、传输、复位、切换及失效;Python 编译、安装器语法检查通过。实机服务 active,状态版本 3.8.0、eth0 上传/下载数值有效,未发现本轮循环错误;Console 两设备在线,接入插件生成对应历史指标。
旧程序备份 `/home/way/pigway-agent-backup-v3.8.0-20261014/pigway-pi-control`,目录 0700。仅更新监控程序和新增模块并重启监控服务,没有变更硬件插件或 API 配置。
+10
View File
@@ -0,0 +1,10 @@
# 文档目录
安装、连接及日常使用从[中文首页](../README.zh-CN.md)或[英文首页](../README.md)开始。本目录收录仍适用的功能、配置与开发说明;版本变化见[版本记录](../RELEASE_NOTES.md)。
- [本地硬件插件协议](HARDWARE_PLUGIN_API.md)
- [Raspberry Pi OS Lite (Debian 13 Trixie) fresh installation](RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.md)
- [Raspberry Pi OS Lite(Debian 13 Trixie)全新安装](RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.zh-CN.md)
- [磁盘发现、刷新与显示](STORAGE.md)
[贡献指南](../CONTRIBUTING.md) · [安全反馈](../SECURITY.md) · [许可证](../LICENSE)
+21
View File
@@ -0,0 +1,21 @@
# 磁盘发现与显示
## 使用
在 Console → 系统配置 → 选择设备 → 磁盘设置中点击“扫描磁盘”,显式发现新增磁盘。打开页面、普通刷新和重启不执行新增磁盘发现。扫描失败保留已有清单;不执行挂载、格式化或分区操作。
已知磁盘容量每周刷新,CPU、内存与网络按自身周期采集。摘要显示实际采集时间;缺失指标不补零,采集失败不据此确认告警恢复。
显示名称、同组顺序与隐藏偏好统一保存到 Console,首页与配置页共用;Agent 已有名称和隐藏定义作为默认值。隐藏只影响显示,不停止采集或删除历史。
## Agent 行为
认证接口 `PUT /api/v1/storage/scan` 读取块设备和挂载点,持久化清单到对应配置旁的 `.storage.json`。监控服务读取清单;新增磁盘须再次手动扫描。根分区指标保留兼容,服务重启可复用容量缓存,整机重启后的容量初始化不自动发现新磁盘。
UUID 优先作为标识,其次为设备序列号和分区编号,再退回路径并标记 `stable_identity=false`。重复挂载及 loop/swap 等伪设备不展示;未挂载设备不伪造使用率。配置过的磁盘在扫描未发现时保留为断开状态。
磁盘扫描不执行 SMART 或自检;SMART 能力以本项目的独立说明为准。
## 相关说明
[中文首页](../README.zh-CN.md) · [文档目录](README.md)
+6 -2
View File
@@ -12,13 +12,15 @@ for arg in "$@"; do
*) echo "ERROR: unknown option: $arg (hardware is installed separately)" >&2; exit 2 ;; *) echo "ERROR: unknown option: $arg (hardware is installed separately)" >&2; exit 2 ;;
esac esac
done done
for required in app/pigway_pi_control.py app/plugin_api.py app/network_rates.py app/pigway_pi_control_api.py config/pigway-pi-control.conf systemd/pigway-pi-control.service systemd/pigway-pi-control-api.service; do for required in app/pigway_pi_control.py app/plugin_api.py app/network_rates.py app/storage_cache.py app/pigway_pi_control_api.py config/pigway-pi-control.conf systemd/pigway-pi-control.service systemd/pigway-pi-control-api.service; do
[ -f "$required" ] || { echo "ERROR: required file missing: $required" >&2; exit 1; } [ -f "$required" ] || { echo "ERROR: required file missing: $required" >&2; exit 1; }
done done
[ "$(id -u)" -eq 0 ] || { echo 'ERROR: run installer as root' >&2; exit 1; } [ "$(id -u)" -eq 0 ] || { echo 'ERROR: run installer as root' >&2; exit 1; }
# The monitor has no I2C, GPIO, Pillow, smbus2 or cooling-board requirement. # The monitor has no I2C, GPIO, Pillow, smbus2 or cooling-board requirement.
command -v python3 >/dev/null || { apt-get update; apt-get install -y python3; } command -v python3 >/dev/null || { apt-get update; apt-get install -y python3; }
command -v lsblk >/dev/null || { apt-get update; apt-get install -y util-linux; }
mkdir -p /usr/local/share/pigway-pi-control mkdir -p /usr/local/share/pigway-pi-control
install -m 0644 app/storage_cache.py /usr/local/share/pigway-pi-control/storage_cache.py
install -m 0644 app/network_rates.py /usr/local/share/pigway-pi-control/network_rates.py install -m 0644 app/network_rates.py /usr/local/share/pigway-pi-control/network_rates.py
install -m 0644 app/plugin_api.py /usr/local/share/pigway-pi-control/plugin_api.py install -m 0644 app/plugin_api.py /usr/local/share/pigway-pi-control/plugin_api.py
install -m 0755 app/pigway_pi_control.py /usr/local/sbin/pigway-pi-control install -m 0755 app/pigway_pi_control.py /usr/local/sbin/pigway-pi-control
@@ -33,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():
@@ -67,6 +69,8 @@ sys.path.insert(0,'app')
import pigway_pi_control_api as api import pigway_pi_control_api as api
api.save_ini_text(api.replace_ini_value('api','enabled',os.environ['API_ENABLED']),'install') api.save_ini_text(api.replace_ini_value('api','enabled',os.environ['API_ENABLED']),'install')
PYAPI PYAPI
install -d -m 0755 /usr/local/share/doc/pigway-raspberry-pi-agent
install -m 0644 LICENSE /usr/local/share/doc/pigway-raspberry-pi-agent/
systemctl daemon-reload systemctl daemon-reload
if [ "$NO_START" -eq 0 ]; then if [ "$NO_START" -eq 0 ]; then
systemctl enable pigway-pi-control.service >/dev/null systemctl enable pigway-pi-control.service >/dev/null
+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%')
+41
View File
@@ -0,0 +1,41 @@
import sys,tempfile,unittest
from pathlib import Path
from types import SimpleNamespace
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'app'))
from storage_cache import StorageCache
class StorageTests(unittest.TestCase):
def test_weekly_reads_and_restart_cache(self):
with tempfile.TemporaryDirectory() as directory:
now=[1000];reads=[]
def reader(path):
reads.append(path);return SimpleNamespace(total=1000,used=200,free=800)
p=Path(directory)/'cache.json'
cache=StorageCache(p,reader,lambda:now[0])
self.assertEqual(cache.sample()['used'],200)
for i in range(20):now[0]+=1;cache.sample()
self.assertEqual(len(reads),1)
restored=StorageCache(p,reader,lambda:now[0]);restored.sample()
self.assertEqual(len(reads),1)
now[0]=605800;restored.sample()
self.assertEqual(len(reads),2)
def test_failure_does_not_fabricate_zero_or_retry_each_second(self):
with tempfile.TemporaryDirectory() as directory:
now=[1000];reads=[]
def fail(path):reads.append(path);raise OSError('unavailable')
cache=StorageCache(Path(directory)/'cache.json',fail,lambda:now[0])
self.assertIsNone(cache.sample());self.assertTrue(cache.error)
now[0]+=5;self.assertIsNone(cache.sample())
self.assertEqual(len(reads),1)
def test_failed_read_cooldown_survives_restart(self):
with tempfile.TemporaryDirectory() as directory:
reads=[]
def fail(path):reads.append(path);raise OSError('unavailable')
p=Path(directory)/'cache.json'
StorageCache(p,fail,lambda:1000).sample()
restored=StorageCache(p,fail,lambda:1005)
self.assertTrue(restored.error);self.assertIsNone(restored.sample())
self.assertEqual(len(reads),1)
+35
View File
@@ -0,0 +1,35 @@
import importlib.util
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "app"))
stem = "pigway_ubuntu_agent_api" if (ROOT / "app/pigway_ubuntu_agent_api.py").exists() else "pigway_pi_control_api"
spec = importlib.util.spec_from_file_location("storage_api_under_test", ROOT / "app" / (stem + ".py"))
api = importlib.util.module_from_spec(spec)
spec.loader.exec_module(api)
class StorageConfigTests(unittest.TestCase):
def test_config_round_trip_and_invalid_edits_are_atomic(self):
with tempfile.TemporaryDirectory() as directory:
config = Path(directory) / "agent.conf"
config.write_text("[device]\nname = auto\n")
identifier = "a" * 32
with patch.object(api, "CFG", config):
api.update_ini({"storage_names": {identifier: "媒体盘 100%"}, "storage_visible": {identifier: "false"}})
self.assertEqual(api.config_json()["storage_names"][identifier], "媒体盘 100%")
self.assertEqual(api.config_json()["storage_visible"][identifier], "false")
original = config.read_text()
for updates in [
{"storage_visible": {identifier: "yes"}},
{"storage_names": {"invalid-id": "Bad"}},
{"storage_names": {identifier: "x" * 65}},
{"storage_names": {identifier: "bad\tname"}},
]:
with self.assertRaises(ValueError):
api.update_ini(updates)
self.assertEqual(config.read_text(), original)
+97
View File
@@ -0,0 +1,97 @@
import configparser
import json
import sys
import tempfile
import unittest
from collections import namedtuple
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "app"))
from storage_cache import StorageInventory
Usage = namedtuple("Usage", "total used free")
class StorageInventoryTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.mounts = Path(self.tmp.name) / "mountinfo"
self.mounts.write_text("1 0 8:1 / / rw - ext4 /dev/sda1 rw\n2 0 8:2 / /media/My\\040Disk rw - ext4 /dev/sda2 rw\n3 0 8:2 /folder /bind rw - ext4 /dev/sda2 rw\n")
self.data = {"blockdevices": [
{"name": "/dev/sda", "type": "disk", "serial": "serial", "children": [
{"name": "/dev/sda1", "type": "part", "uuid": "root", "maj:min": "8:1", "fstype": "ext4", "size": 1000},
{"name": "/dev/sda2", "type": "part", "uuid": "data", "maj:min": "8:2", "fstype": "ext4", "size": 2000},
]},
{"name": "/dev/sdb", "type": "disk", "size": 3000},
{"name": "/dev/loop0", "type": "loop", "size": 4000},
{"name": "/dev/zram0", "type": "disk", "fstype": "swap"},
]}
self.now = 100
self.reader = StorageInventory(clock=lambda: self.now, runner=lambda: json.dumps(self.data),
mounts_path=self.mounts, usage=lambda p: Usage(1000, 0, 1000),
cache_path=Path(self.tmp.name)/"inventory.json")
self.reader.scan()
def test_weekly_usage_cache_survives_service_restart(self):
capacity=Path(self.tmp.name)/'usage.json'
self.reader.capacity_path=capacity
self.reader.sample()
def forbidden(*args):raise AssertionError('Not due; should not read capacity or discover disks')
restarted=StorageInventory(clock=lambda:self.now+100,runner=forbidden,mounts_path=self.mounts,
usage=forbidden,cache_path=self.reader.cache_path,capacity_path=capacity)
self.assertEqual(len(restarted.sample()),3)
self.assertEqual(restarted.observed_at,self.now)
self.assertEqual(restarted.next_update_at,self.now+604800)
def test_mounts_raw_disk_and_pseudo_filters(self):
disks = self.reader.sample()
self.assertEqual(len(disks), 3)
self.assertEqual(disks[0]["mountpoint"], "/")
self.assertEqual(next(d for d in disks if d.get("uuid") == "data")["mountpoint"], "/media/My Disk")
self.assertEqual(disks[0]["used_percent"], 0)
raw = next(d for d in disks if d["device"] == "/dev/sdb")
self.assertEqual(raw["state"], "unmounted")
self.assertIsNone(raw["used_bytes"])
def test_uuid_survives_renumber_and_configuration(self):
disk = self.reader.sample()[0]
self.data["blockdevices"][0]["children"][0]["name"] = "/dev/sdz1"
self.now += 61
self.reader.scan()
config = configparser.ConfigParser()
config.read_string(f'[storage_names]\n{disk["id"]} = 数据盘 100%\n[storage_visible]\n{disk["id"]} = false\n')
result = next(d for d in self.reader.sample(config) if d["id"] == disk["id"])
self.assertEqual(result["name"], "数据盘 100%")
self.assertFalse(result["visible"])
def test_removal_and_discovery_error_are_distinct(self):
disk = self.reader.sample()[0]
config = configparser.ConfigParser()
config.read_dict({"storage_names": {disk["id"]: "System"}})
self.reader.runner = lambda: "invalid json"
self.now += 61
with self.assertRaises(ValueError):self.reader.scan()
self.assertEqual(self.reader.sample(config)[0]["state"], "mounted")
self.reader.runner = lambda: '{"blockdevices":[]}'
self.now += 61
self.reader.scan()
self.assertEqual(self.reader.sample(config)[0]["state"], "missing")
self.assertFalse(self.reader.error)
def test_hotplug_requires_manual_scan_and_inventory_survives_restart(self):
self.assertEqual(len(self.reader.sample()), 3)
self.data["blockdevices"].append({"name": "/dev/sdc", "type": "disk", "uuid": "new", "size": 9000})
self.now += 29
self.assertEqual(len(self.reader.sample()), 3)
self.now += 100000
self.assertEqual(len(self.reader.sample()), 3)
self.reader.scan()
self.assertEqual(len(self.reader.sample()), 4)
def forbidden():raise AssertionError("automatic enumeration is forbidden")
restored=StorageInventory(clock=lambda:self.now,runner=forbidden,mounts_path=self.mounts,
usage=lambda p:Usage(1000,0,1000),cache_path=self.reader.cache_path)
self.assertEqual(len(restored.sample()),4)
if __name__ == "__main__":
unittest.main()
+1
View File
@@ -7,6 +7,7 @@ rm -f /usr/local/sbin/pigway-pi-control
rm -f /usr/local/sbin/pigway-pi-control-api rm -f /usr/local/sbin/pigway-pi-control-api
rm -f /etc/pigway-pi-control-api.token rm -f /etc/pigway-pi-control-api.token
rm -rf /usr/local/share/pigway-pi-control rm -rf /usr/local/share/pigway-pi-control
rm -rf /usr/local/share/doc/pigway-raspberry-pi-agent
systemctl daemon-reload systemctl daemon-reload
echo "程序已卸载。为防误删自定义内容,配置 /etc/pigway-pi-control.conf 已保留。" echo "程序已卸载。为防误删自定义内容,配置 /etc/pigway-pi-control.conf 已保留。"
echo "如确定不要配置,可手工:sudo rm /etc/pigway-pi-control.conf" echo "如确定不要配置,可手工:sudo rm /etc/pigway-pi-control.conf"