Compare commits

..
7 changed files with 465 additions and 78 deletions
+41 -4
View File
@@ -1,8 +1,8 @@
# PIGWay Pi Control v3.6.0
# PIGWay Pi Control v3.7.0
Raspberry Pi 4B + 配套 128×32 OLED / 风扇 / RGB 散热扩展板的本机硬件监控 Agent。
v3.6.0 在本地硬件监控、显示仲裁和可配置 RGB 告警基础上,增加轻量状态/日志/配置 API 与同源 Web 管理界面。
v3.7.0 在现有本机硬件监控基础上增加多机状态、日志和配置管理;未登记其他节点时继续作为完整的单机版本运行。
## 已有核心功能
@@ -84,7 +84,7 @@ CHECK STORAGE
只在 **状态发生变化** 时记录关键事件,避免每秒刷屏。例如:
```text
level=INFO event=START version=3.6.0
level=INFO event=START version=3.7.0
level=INFO event=NET_SWITCH old_if=eth0 new_if=wlan0 ...
level=WARN event=NETWORK_DOWN ...
level=INFO event=NETWORK_RECOVERED ...
@@ -138,6 +138,36 @@ RGB 保留卖家 MCU 0x0D 寄存器协议。实机校准确认静态自定义颜
Pi 红色 PWR 灯关机后恢复硬件默认亮起的行为不强行修改,可作为“系统已关机但 UPS 仍供电”的直观提示。
## v3.7.0 多机监控
每台机器在 `[device]` 中设置稳定标识和显示名称;保留 `auto` 时读取本机 hostname。作为 Web 聚合入口的机器可在 `[hosts]` 中登记其他已安装 PIGWay Pi Control 的节点:
```ini
[device]
identifier = living-room
name = 客厅树莓派
[hosts]
study = http://192.168.1.20:6001
nas-room = http://192.168.1.21:6001
```
未设置 `[hosts]` 时,API 和 Web 保持单机使用方式。聚合入口通过各节点现有 HTTP API 读取状态、journal 和配置;节点不可达时会在机器卡片上显示离线,不影响其他节点。
新增接口:
```text
GET /api/v1/hosts
PUT /api/v1/hosts
GET /api/v1/fleet/status
GET /api/v1/fleet/logs
PUT /api/v1/fleet/config
```
运行状态按机器显示独立卡片,包括机器标识、hostname、型号、序列号、板卡版本、操作系统、内核、架构、CPU 核数和 Agent 版本。内存和根分区同时显示百分比、已用量、总量及可用/剩余量;Wi-Fi 以 0–100% 为主值,并保留原始 dBm。OLED 的 Wi-Fi 主值同样使用百分比。
事件日志增加机器列、机器筛选和机器排序。系统配置可选择目标机器;批量同步只发送当前已修改的配置项,并逐台返回成功或失败结果。远端 Bearer Token 只保存在当前浏览器的 localStorage 中,不写入聚合机器配置文件。
## v3.6.0 API 与 Web
安装后访问:
@@ -165,6 +195,10 @@ GET /api/v1/config
GET /api/v1/services
```
日志接口支持 `since`、`until` Unix 时间戳,`severity`、`event`、`search` 筛选,`timestamp` / `severity` / `event` / `message` 排序,以及 `limit` / `offset` 分页。Web 提供今天、昨天、本周、本月和全部快捷范围。日志直接读取 systemd journal 的现有保留数据,不建立重复的 SQLite 日志副本。
Web 界面支持中文和 English 切换,并将结构化 journal 事件类型显示为当前界面的语言;journal 内部稳定的事件标识保持不变,便于脚本筛选。外观可选择自动、深色或浅色,自动模式跟随浏览器/操作系统配色。语言和外观偏好只保存在当前浏览器中。
配置写入接口:
```text
@@ -175,7 +209,9 @@ Content-Type: application/json
{"updates":{"led":{"service_b":"96"},"api":{"port":"6001"}}}
```
服务监控页面会列出本机 systemd 服务,并标识“已监控/未监控”。添加、编辑或删除监控使用 `PUT /api/v1/services`;启动或停止服务使用 `PUT /api/v1/services/control`,两者均要求 Bearer Token。SSH、向日葵、Tailscale 和 PIGWay 自身服务不能从 Web 停止,以免中断远程管理或控制进程。监控备注保存在配置文件的 `[service_notes]`,不会参与硬件 Agent 的监控判断。
服务监控页面会列出本机 systemd 服务,并标识“已监控/未监控”。添加、编辑或删除监控使用 `PUT /api/v1/services`;启动或停止服务使用 `PUT /api/v1/services/control`,两者均要求 Bearer Token。API 服务不能通过自己的请求停止自身,其他服务由用户自行管理。监控备注保存在配置文件的 `[service_notes]`,不会参与硬件 Agent 的监控判断。
“暂停监控/恢复监控”只改变 PIGWay 是否检查该服务,不会启动或停止 systemd unit;暂停状态保存在 `[service_monitor_disabled]`。只有明确标注为“启动服务/停止服务”的按钮才会调用 systemctl。
读取 token:
@@ -236,3 +272,4 @@ sudo /usr/local/sbin/pigway-pi-control --diagnose
- v3.2.0:主网络接口 / Wi-Fi RSSI / ETH速率 / 双网切换
- v3.5.0:System Health、显示仲裁与可配置 RGB 告警
- **v3.6.0:轻量状态/日志/配置 API 与同源 Web 管理界面**
- **v3.7.0:多机状态卡片、聚合日志、机器配置选择与批量同步**
+6 -6
View File
@@ -1,10 +1,10 @@
5442d660875ec28909b1183e4e2aee604e9f2e82d229c3806498392e7a4546e3 README.md
96dfd20c3ad16ca62a46d66eb5ffb82b1f7f987e66c4dae5310609b1989b7dae README.md
72c46162c33f9587c6ccb01ad53fbaaf4ecec8cd21014b909e87e8707e253e36 app/oled_font_5x7.bin
d71d949643d31461f2e62ded3818e1a0f1711d1420756ad0cc32eb04cde0fad3 app/pigway_pi_control.py
bf52a12d3208b5330f3e2980187e70330b930e950cfd5165b4e6605febfef65d app/pigway_pi_control_api.py
75fa662ed21138a8e394de83e46281dab039433cc0d7a8f787e1a049f4b47ec8 config/pigway-pi-control.conf
f34eac1df180a3da3a6998d650c06435fd89728c10b090f6162df82e53986dd0 install.sh
d0fa68951097a90eb6f51b36e56dc935d8fd7632c3de0ad6bff0b448960a7c03 app/pigway_pi_control.py
034931590d83969f9f24222cc57800ab193588beea3d3dcedb98f1690000370d app/pigway_pi_control_api.py
ee0f682e8016f75f906d974f00c3e3340edddf30447d441ddfd61a7a34c9e83c config/pigway-pi-control.conf
f9fbb4defa45b43abe0ce66973b1f4d553369b3a43c01d04f206c573f7dc96ca install.sh
4f6e8db3698a62f57823d3298a0e6814593de6fa3d31e4751b7ea4a76e553e35 systemd/pigway-pi-control.service
aa21cef9a06593371ccff005335b6c0c826fbf4b5c48e4255b5f4b75ce6e9217 systemd/pigway-pi-control-api.service
9f59e7313ee58e687ae6dd29d175c67082674254ac5c5f54168b2735b64c20a4 uninstall.sh
2a3e37414845e023850af248a8ae4f5aacc06814cee88950e6e2564a34bac2c3 web/index.html
3a40109f11bea2d44e9d61bed4a90499a5047c4fc534f97aaa90becbf9e57017 web/index.html
+70 -11
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
import time,socket,subprocess,configparser,signal,re,errno,threading,argparse,json
import time,socket,subprocess,configparser,signal,re,errno,threading,argparse,json,os,platform
from pathlib import Path
APP_VERSION="3.6.0"
APP_VERSION="3.7.0"
from smbus2 import SMBus
from PIL import Image
from shutil import disk_usage
@@ -29,6 +29,41 @@ def load_average():
x=Path("/proc/loadavg").read_text().split();return (float(x[0]),float(x[1]),float(x[2]))
except Exception:return (0.0,0.0,0.0)
def memory_usage():
try:
values={}
for line in Path("/proc/meminfo").read_text().splitlines():
key,value=line.split(":",1); values[key]=int(value.split()[0])*1024
total=values["MemTotal"]; available=values.get("MemAvailable",values.get("MemFree",0))
used=max(0,total-available)
return total,used,available,(used*100.0/total if total else 0.0)
except Exception:return 0,0,0,0.0
def wifi_percent(metric):
try:return max(0,min(100,round((float(metric)+100.0)*2.0)))
except Exception:return None
def device_details():
info={}
try:
for line in Path("/proc/cpuinfo").read_text().splitlines():
if ":" in line:
key,value=(x.strip() for x in line.split(":",1))
if key in ("Serial","Revision","Hardware","Model"): info[key.lower()]=value
except Exception: pass
try: info["model"]=Path("/proc/device-tree/model").read_bytes().rstrip(b"\0").decode()
except Exception: pass
try:
os_release={}
for line in Path("/etc/os-release").read_text().splitlines():
if "=" in line:
key,value=line.split("=",1); os_release[key]=value.strip().strip('"')
info["os"]=os_release.get("PRETTY_NAME",platform.platform())
except Exception: info["os"]=platform.platform()
info.update({"hostname":socket.gethostname(),"kernel":platform.release(),
"architecture":platform.machine(),"cpu_count":os.cpu_count() or 0})
return info
def cpu_freq_mhz():
for q in ("/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq","/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_cur_freq"):
try:return int(Path(q).read_text().strip())/1000.0
@@ -210,7 +245,9 @@ def load_cfg():
if c.has_section("services"):
for label,target in c["services"].items():
target=target.strip()
if target and not target.startswith("#"): watches.append(("service",label.upper(),target))
paused=c.getboolean("service_monitor_disabled",label,fallback=False)
if target and not target.startswith("#") and not paused:
watches.append(("service",label.upper(),target))
if c.has_section("processes"):
for label,target in c["processes"].items():
target=target.strip()
@@ -222,6 +259,12 @@ def cf(section,key,default,cast=float):
try: return cast(cfg.get(section,key))
except: return default
DEVICE_ID=cf("device","identifier","auto",str).strip()
if not DEVICE_ID or DEVICE_ID.lower()=="auto": DEVICE_ID=socket.gethostname()
DEVICE_NAME=cf("device","name",DEVICE_ID,str).strip() or DEVICE_ID
if DEVICE_NAME.lower()=="auto": DEVICE_NAME=DEVICE_ID
DEVICE_INFO=device_details()
SENSOR_INTERVAL=cf("timing","sensor_interval",1.0)
OLED_REFRESH=cf("timing","oled_refresh_interval",1.0)
PAGE_INTERVAL=cf("timing","page_interval",10.0)
@@ -304,6 +347,7 @@ def check_watch(w):
def diagnose():
print("Agent:")
print(f"version={APP_VERSION} uptime_seconds={uptime_seconds()} config={CFG}")
print(f"machine_id={DEVICE_ID} machine_name={DEVICE_NAME} hostname={socket.gethostname()}")
print("loop_errors=UNAVAILABLE i2c_errors=UNAVAILABLE scope=separate_diagnostic_process")
print("System:")
print(f"load_average={load_average()} cpu_freq_mhz={cpu_freq_mhz():.0f} root_readonly={root_readonly()}")
@@ -316,7 +360,7 @@ def diagnose():
if "error" in power: print("error="+power["error"])
iface,kind,metric,label,address=primary_network()
print("Network:")
print(f"primary_interface={iface or 'NONE'} kind={kind} metric={metric} ip_label={label} IP={address} internet={internet()}")
print(f"primary_interface={iface or 'NONE'} kind={kind} metric={metric} signal_percent={wifi_percent(metric) if kind=='WIF' else 'N/A'} ip_label={label} IP={address} internet={internet()}")
print("I2C:")
# SMBus receive-byte performs a read only; no register selection or display commands.
try:
@@ -489,10 +533,11 @@ def render_home(cpu,temp,mem,disk):
f"DSK {disk:.1f}%",
ip_label+" "+display_ip,
]
signal=wifi_percent(net_metric) if net_kind=="WIF" else None
right=[
f"MEM {mem:.1f}%",
f"FAN {FAN_NAME[fan_level]}",
f"{net_kind} {net_metric}",
f"WIF {signal}%" if signal is not None else f"{net_kind} {net_metric}",
"",
]
for row in range(4):
@@ -666,7 +711,8 @@ class DisplayController:
}
def _rgb_worker(self):
previous_error=None; applied_mode="OFF"; applied_flash=False; animation_started=time.monotonic()
previous_error=None; applied_mode="OFF"; applied_flash=False
flash_on=None; animation_started=time.monotonic()
while not self.stop_event.is_set():
try:
with self.lock: wanted_mode=self.current_rgb_mode
@@ -683,6 +729,7 @@ class DisplayController:
else: raise ValueError(f"unsupported RGB mode: {wanted_mode}")
log("INFO","RGB_MODE",old=applied_mode,new=wanted_mode)
applied_mode=wanted_mode; animation_started=time.monotonic()
flash_on=True if wanted_mode.endswith("_FLASH") else None
applied_flash=applied_mode.endswith("_FLASH")
if applied_flash:
elapsed_ms=(time.monotonic()-animation_started)*1000.0
@@ -691,10 +738,12 @@ class DisplayController:
elif "_CRITICAL_" in applied_mode: on_ms,off_ms=CRITICAL_FLASH
else: on_ms,off_ms=WARNING_FLASH
on=elapsed_ms%(on_ms+off_ms)<on_ms
with bus_lock:
if applied_mode=="NORMAL_FLASH":
rgb_static(NORMAL_COLOR,"normal_flash_on") if on else rgb_off()
else: rgb_flash(applied_mode,on)
if on!=flash_on:
with bus_lock:
if applied_mode=="NORMAL_FLASH":
rgb_static(NORMAL_COLOR,"normal_flash_on") if on else rgb_off()
else: rgb_flash(applied_mode,on)
flash_on=on
if previous_error is not None: log("INFO","RGB_RECOVERED")
previous_error=None
except Exception as e:
@@ -723,14 +772,24 @@ def cleanup():
def write_status(cpu,temp,mem,disk):
display_state=display.status_snapshot()
memory_total,memory_used,memory_available,memory_calculated=memory_usage()
disk_total,disk_used,disk_free=disk_usage("/")
signal=wifi_percent(net_metric) if net_kind=="WIF" else None
state={
"version":APP_VERSION,"timestamp":int(time.time()),"uptime_seconds":uptime_seconds(),
"machine":{"id":DEVICE_ID,"name":DEVICE_NAME,"hostname":socket.gethostname()},
"device":DEVICE_INFO,
"system":{"load_average":load_average(),"cpu_percent":round(cpu,1),
"cpu_freq_mhz":round(cpu_freq_mhz()),"temperature_c":round(temp,1),
"memory_percent":round(mem,1),"disk_percent":round(disk,1),
"memory_percent":round(memory_calculated if memory_total else mem,1),
"memory_total_bytes":memory_total,"memory_used_bytes":memory_used,
"memory_available_bytes":memory_available,
"disk_percent":round(disk,1),"disk_total_bytes":disk_total,
"disk_used_bytes":disk_used,"disk_free_bytes":disk_free,
"root_readonly":bool(fs_readonly),"fan_level":fan_level,"fan_name":FAN_NAME[fan_level]},
"power":health,
"network":{"interface":net_iface,"kind":net_kind,"metric":net_metric,
"signal_percent":signal,
"ip_label":ip_label,"ip":ip,"link":bool(link_ok),"internet":bool(net_ok)},
"watches":[{"type":w[0],"name":w[1],"target":w[2],"up":bool(ok)}
for w,ok in zip(watches,watch_state)],
+246 -24
View File
@@ -1,24 +1,86 @@
#!/usr/bin/env python3
import configparser,ipaddress,json,re,secrets,shutil,subprocess,threading,time
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 pathlib import Path
from urllib.parse import parse_qs,urlparse
from urllib.parse import parse_qs,urlencode,urlparse
from urllib.request import Request,urlopen
APP_VERSION="3.6.0"
APP_VERSION="3.7.0"
CFG=Path("/etc/pigway-pi-control.conf")
STATUS=Path("/run/pigway-pi-control/status.json")
TOKEN=Path("/etc/pigway-pi-control-api.token")
WEB=Path("/usr/local/share/pigway-pi-control/web/index.html")
ALLOWED_SECTIONS={"services","processes","alerts","fan","led","timing","dark_mode","display","api"}
ALLOWED_SECTIONS={"device","hosts","services","processes","alerts","fan","led","timing","dark_mode","display","api"}
NAME_RE=re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
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={"ssh.service","sshd.service","runawesun.service","tailscaled.service",
"pigway-pi-control.service","pigway-pi-control-api.service"}
PROTECTED_UNITS={"pigway-pi-control-api.service"}
def machine_identity():
cfg=load_cfg(); identifier=cfg.get("device","identifier",fallback="auto").strip()
if not identifier or identifier.lower()=="auto": identifier=socket.gethostname()
name=cfg.get("device","name",fallback=identifier).strip() or identifier
if name.lower()=="auto": name=identifier
return {"id":identifier,"name":name,"hostname":socket.gethostname(),"local":True,"url":""}
def normalize_url(value):
parsed=urlparse(str(value).strip())
if parsed.scheme not in {"http","https"} or not parsed.hostname or parsed.username or parsed.password:
raise ValueError("host URL must be an http(s) URL without credentials")
return f"{parsed.scheme}://{parsed.netloc}".rstrip("/")
def host_registry():
local=machine_identity(); result=[local]
cfg=load_cfg()
if cfg.has_section("hosts"):
for host_id,url in cfg.items("hosts"):
try: base=normalize_url(url)
except ValueError: continue
if host_id==local["id"]: continue
result.append({"id":host_id,"name":host_id,"hostname":"","local":False,"url":base})
return result
def find_host(host_id):
for host in host_registry():
if host["id"]==host_id:return host
raise ValueError("unknown host")
def remote_json(base,path,method="GET",payload=None,token=""):
data=None if payload is None else json.dumps(payload,separators=(",",":")).encode()
headers={"Accept":"application/json"}
if data is not None: headers["Content-Type"]="application/json"
if token: headers["Authorization"]=f"Bearer {token}"
request=Request(base+path,data=data,headers=headers,method=method)
with urlopen(request,timeout=5) as response:
return json.loads(response.read())
def fleet_status():
hosts=host_registry(); results=[]
try:
status=json.loads(STATUS.read_text())
status.setdefault("machine",machine_identity())
results.append({"host":hosts[0],"online":True,"status":status})
except Exception as exc:
results.append({"host":hosts[0],"online":False,"error":str(exc)})
def fetch(host):
try:
status=remote_json(host["url"],"/api/v1/status")
machine=status.get("machine",{})
merged={**host,"name":machine.get("name") or host["name"],"hostname":machine.get("hostname","")}
return {"host":merged,"online":True,"status":status}
except Exception as exc:return {"host":host,"online":False,"error":str(exc)}
peers=hosts[1:]
with ThreadPoolExecutor(max_workers=min(8,max(1,len(peers)))) as pool:
futures=[pool.submit(fetch,h) for h in peers]
for future in as_completed(futures): results.append(future.result())
order={h["id"]:i for i,h in enumerate(hosts)}
results.sort(key=lambda x:order.get(x["host"]["id"],999))
return {"hosts":results,"timestamp":int(time.time())}
def load_cfg():
cfg=configparser.ConfigParser(); cfg.read(CFG)
@@ -33,7 +95,7 @@ def config_json():
return {section:dict(cfg.items(section)) for section in cfg.sections() if section in ALLOWED_SECTIONS}
def validate_value(section,key,value,existing):
if section not in {"services","processes"} and key not in existing.get(section,set()):
if section not in {"services","processes","hosts"} and key not in existing.get(section,set()):
raise ValueError(f"unsupported option: {section}.{key}")
if RGB_KEY_RE.fullmatch(key):
number=int(value)
@@ -55,6 +117,42 @@ def validate_value(section,key,value,existing):
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):
raise ValueError("device.identifier must be auto or 1..64 letters, numbers, dot, underscore or dash")
elif section=="hosts":
normalize_url(value)
def replace_ini_value(section,key,value):
text=CFG.read_text()
section_match=re.search(rf"(?mi)^\[{re.escape(section)}\]\s*$",text)
if not section_match:
if value is not None: text=text.rstrip()+f"\n\n[{section}]\n{key} = {value}\n"
return text
next_section=re.search(r"(?m)^\[.+\]\s*$",text[section_match.end():])
end=section_match.end()+(next_section.start() if next_section else len(text)-section_match.end())
chunk=text[section_match.end():end]
key_match=re.search(rf"(?mi)^\s*{re.escape(key)}\s*=.*(?:\n|$)",chunk)
if key_match:
start=section_match.end()+key_match.start(); stop=section_match.end()+key_match.end()
return text[:start]+((f"{key} = {value}\n") if value is not None else "")+text[stop:]
if value is not None:return text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
return text
def save_ini_text(text,prefix="api"):
parsed=configparser.ConfigParser(); parsed.read_string(text)
backup=CFG.with_name(f"{CFG.name}.bak.{prefix}-{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 edit_host(operation,host_id,url=""):
if operation not in {"save","delete"}: raise ValueError("invalid host operation")
if not NAME_RE.fullmatch(host_id): raise ValueError("invalid host identifier")
local=machine_identity()["id"]
if host_id==local: raise ValueError("local host cannot be changed in the remote host list")
if operation=="save": url=normalize_url(url)
text=replace_ini_value("hosts",host_id,url if operation=="save" else None)
return save_ini_text(text,"hosts")
def update_ini(updates):
if not isinstance(updates,dict): raise ValueError("updates must be an object")
@@ -92,18 +190,83 @@ def update_ini(updates):
temporary.write_text(text); temporary.chmod(0o644); temporary.replace(CFG)
return str(backup)
def journal(limit):
result=subprocess.run(["journalctl","-u","pigway-pi-control.service","-n",str(limit),
"--no-pager","-o","json"],capture_output=True,text=True,timeout=5)
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"]
if since is not None: command.extend(["--since",f"@{since}"])
if until is not None: command.extend(["--until",f"@{until}"])
result=subprocess.run(command,capture_output=True,text=True,timeout=15)
if result.returncode: raise RuntimeError(result.stderr.strip() or "journalctl failed")
rows=[]
for line in result.stdout.splitlines():
try:
item=json.loads(line)
rows.append({"timestamp":item.get("__REALTIME_TIMESTAMP"),"priority":item.get("PRIORITY"),
"message":item.get("MESSAGE","")})
message=item.get("MESSAGE","")
level_match=re.search(r"(?:^|\s)level=([A-Z]+)(?:\s|$)",message)
event_match=re.search(r"(?:^|\s)event=([^ ]+)",message)
priority=str(item.get("PRIORITY","6"))
level=(level_match.group(1) if level_match else
{"0":"EMERGENCY","1":"ALERT","2":"CRITICAL","3":"ERROR","4":"WARN","5":"NOTICE","6":"INFO","7":"DEBUG"}.get(priority,"INFO"))
rows.append({"timestamp":int(item.get("__REALTIME_TIMESTAMP",0)),"priority":priority,
"severity":level,"event":event_match.group(1) if event_match else "SYSTEM",
"message":message})
except json.JSONDecodeError: pass
return rows
severities={x.strip().upper() for x in severity.split(",") if x.strip()}
if severities: rows=[x for x in rows if x["severity"] in severities]
event_types=sorted({x["event"] for x in rows})
if event: rows=[x for x in rows if x["event"]==event]
if search:
needle=search.casefold(); rows=[x for x in rows if needle in x["message"].casefold()]
severity_rank={"EMERGENCY":0,"ALERT":1,"CRITICAL":2,"ERROR":3,"WARN":4,"NOTICE":5,"INFO":6,"DEBUG":7}
keys={"timestamp":lambda x:x["timestamp"],"severity":lambda x:severity_rank.get(x["severity"],99),
"event":lambda x:x["event"],"message":lambda x:x["message"]}
rows.sort(key=keys[sort],reverse=order=="desc")
total=len(rows)
return {"logs":rows[offset:offset+limit],"total":total,"offset":offset,"limit":limit,
"event_types":event_types,"retention":"systemd-journal"}
def fleet_journal_query(since=None,until=None,severity="",event="",search="",machine="",
sort="timestamp",order="desc",limit=100,offset=0):
hosts=host_registry(); rows=[]; event_types=set(); machines=[]
def decorate(payload,host):
machine_info=payload.get("machine") or {"id":host["id"],"name":host["name"]}
host_id=machine_info.get("id") or host["id"]
host_name=machine_info.get("name") or host["name"]
decorated=[]
for row in payload.get("logs",[]):
decorated.append({**row,"machine_id":host_id,"machine_name":host_name})
return decorated,set(payload.get("event_types",[])),{"id":host_id,"name":host_name}
local_payload=journal_query(since,until,severity,event,search,"timestamp","desc",50000,0)
local_payload["machine"]=machine_identity()
local_rows,local_events,local_machine=decorate(local_payload,hosts[0])
rows.extend(local_rows); event_types.update(local_events); machines.append(local_machine)
query={"limit":50000,"offset":0,"sort":"timestamp","order":"desc"}
for key,value in (("since",since),("until",until),("severity",severity),("event",event),("search",search)):
if value not in (None,""): query[key]=value
def fetch(host):
payload=remote_json(host["url"],"/api/v1/logs?"+urlencode(query))
try:
status=remote_json(host["url"],"/api/v1/status")
payload["machine"]=status.get("machine",{})
except Exception: pass
return decorate(payload,host)
peers=hosts[1:]
with ThreadPoolExecutor(max_workers=min(8,max(1,len(peers)))) as pool:
futures=[pool.submit(fetch,h) for h in peers]
for future in as_completed(futures):
try:
remote_rows,remote_events,remote_machine=future.result()
rows.extend(remote_rows); event_types.update(remote_events); machines.append(remote_machine)
except Exception: pass
if machine: rows=[row for row in rows if row["machine_id"]==machine]
severity_rank={"EMERGENCY":0,"ALERT":1,"CRITICAL":2,"ERROR":3,"WARN":4,"NOTICE":5,"INFO":6,"DEBUG":7}
keys={"timestamp":lambda x:x["timestamp"],"severity":lambda x:severity_rank.get(x["severity"],99),
"event":lambda x:x["event"],"message":lambda x:x["message"],
"machine":lambda x:(x["machine_name"],x["machine_id"])}
rows.sort(key=keys[sort],reverse=order=="desc")
total=len(rows)
return {"logs":rows[offset:offset+limit],"total":total,"offset":offset,"limit":limit,
"event_types":sorted(event_types),"machines":sorted(machines,key=lambda x:(x["name"],x["id"])),
"retention":"systemd-journal"}
def service_inventory():
cfg=load_cfg()
@@ -125,19 +288,23 @@ def service_inventory():
item.update({"active":parts[2],"description":parts[4] if len(parts)>4 else ""})
for name,target in monitored.values():
item=units.setdefault(target,{"unit":target,"enabled":"not-found","active":"inactive","description":""})
item.update({"monitored":True,"monitor_name":name.upper(),"note":notes.get(name,"")})
paused=cfg.getboolean("service_monitor_disabled",name,fallback=False)
item.update({"monitored":True,"paused":paused,"monitor_name":name.upper(),"note":notes.get(name,"")})
for item in units.values():
item.setdefault("monitored",False); item.setdefault("monitor_name",""); item.setdefault("note","")
item.setdefault("monitored",False); item.setdefault("paused",False)
item.setdefault("monitor_name",""); item.setdefault("note","")
item["protected"]=item["unit"] in PROTECTED_UNITS
return sorted(units.values(),key=lambda x:(not x["monitored"],x["unit"]))
def edit_service_monitor(operation,name,target,note="",previous_name=""):
if operation not in {"save","delete"}: raise ValueError("operation must be save or delete")
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 UNIT_RE.fullmatch(target): raise ValueError("invalid systemd service name")
if len(note)>120 or any(c in note for c in "\r\n\x00"): raise ValueError("invalid note")
if previous_name and not NAME_RE.fullmatch(previous_name): raise ValueError("invalid previous monitor name")
cfg=load_cfg(); old_target=cfg.get("services",name,fallback=None)
previous=previous_name or name
was_paused=cfg.getboolean("service_monitor_disabled",previous,fallback=False)
text=CFG.read_text()
def change(section,key,value):
nonlocal text
@@ -156,11 +323,16 @@ def edit_service_monitor(operation,name,target,note="",previous_name=""):
text=text[:end].rstrip()+f"\n{key} = {value}\n\n"+text[end:].lstrip("\n")
if operation=="delete":
if old_target is None: raise ValueError("monitor not found")
change("services",name,None); change("service_notes",name,None)
change("services",name,None); change("service_notes",name,None); change("service_monitor_disabled",name,None)
elif operation in {"pause","resume"}:
if old_target is None: raise ValueError("monitor not found")
change("service_monitor_disabled",name,"true" if operation=="pause" else None)
else:
if previous_name and previous_name.lower()!=name.lower():
change("services",previous_name,None); change("service_notes",previous_name,None)
change("service_monitor_disabled",previous_name,None)
change("services",name,target); change("service_notes",name,note or None)
if was_paused: change("service_monitor_disabled",name,"true")
parsed=configparser.ConfigParser(); parsed.read_string(text)
backup=CFG.with_name(f"{CFG.name}.bak.api-{time.strftime('%Y%m%d-%H%M%S')}")
shutil.copy2(CFG,backup)
@@ -176,7 +348,7 @@ def control_service(unit,action):
if result.returncode: raise RuntimeError(result.stderr.strip() or f"systemctl {action} failed")
class Handler(BaseHTTPRequestHandler):
server_version="PIGWayAPI/3.6"
server_version="PIGWayAPI/3.7"
def log_message(self,fmt,*args):
return
@@ -211,25 +383,75 @@ class Handler(BaseHTTPRequestHandler):
self.send_header("Content-Length",str(len(data))); self.security_headers(); self.end_headers(); self.wfile.write(data); return
if parsed.path=="/api/v1/health":
self.send_json(200,{"ok":True,"version":APP_VERSION,"status_available":STATUS.exists()}); return
if parsed.path=="/api/v1/hosts":
self.send_json(200,{"hosts":host_registry()}); return
if parsed.path=="/api/v1/fleet/status":
self.send_json(200,fleet_status()); return
if parsed.path=="/api/v1/status":
if not STATUS.exists(): self.send_json(503,{"error":"agent status unavailable"}); return
self.send_json(200,json.loads(STATUS.read_text())); return
if parsed.path=="/api/v1/logs":
limit=min(max(int(parse_qs(parsed.query).get("limit",[api_setting("log_limit",200,int)])[0]),1),1000)
self.send_json(200,{"logs":journal(limit)}); return
if parsed.path in {"/api/v1/logs","/api/v1/fleet/logs"}:
query=parse_qs(parsed.query)
value=lambda key,default="":query.get(key,[default])[0]
limit=min(max(int(value("limit",api_setting("log_limit",200,int))),1),50000)
offset=max(int(value("offset",0)),0)
since=int(value("since")) if value("since") else None
until=int(value("until")) if value("until") else None
if since is not None and until is not None and since>until: raise ValueError("since must not be after until")
sort=value("sort","timestamp"); order=value("order","desc")
allowed_sort={"timestamp","severity","event","message"}
if parsed.path=="/api/v1/fleet/logs": allowed_sort.add("machine")
if sort not in allowed_sort: raise ValueError("invalid log sort")
if order not in {"asc","desc"}: raise ValueError("invalid log order")
severity=value("severity").upper(); event=value("event"); search=value("search")
if len(event)>80 or len(search)>120: raise ValueError("log filter is too long")
if parsed.path=="/api/v1/fleet/logs":
self.send_json(200,fleet_journal_query(since,until,severity,event,search,value("machine"),sort,order,limit,offset)); return
self.send_json(200,journal_query(since,until,severity,event,search,sort,order,limit,offset)); return
if parsed.path=="/api/v1/config":
self.send_json(200,{"config":config_json(),"write_requires_token":True}); return
query=parse_qs(parsed.query); host_id=query.get("host",[""])[0]
if host_id and host_id!=machine_identity()["id"]:
host=find_host(host_id); self.send_json(200,remote_json(host["url"],"/api/v1/config")); return
self.send_json(200,{"config":config_json(),"write_requires_token":True,"machine":machine_identity()}); return
if parsed.path=="/api/v1/services":
self.send_json(200,{"services":service_inventory()}); return
query=parse_qs(parsed.query); host_id=query.get("host",[""])[0]
if host_id and host_id!=machine_identity()["id"]:
host=find_host(host_id); self.send_json(200,remote_json(host["url"],"/api/v1/services")); return
self.send_json(200,{"services":service_inventory(),"machine":machine_identity()}); return
self.send_json(404,{"error":"not found"})
except ValueError as e: self.send_json(400,{"error":str(e)})
except Exception as e: self.send_json(500,{"error":str(e)})
def do_PUT(self):
path=urlparse(self.path).path
if path not in {"/api/v1/config","/api/v1/services","/api/v1/services/control"}: self.send_json(404,{"error":"not found"}); return
if path not in {"/api/v1/config","/api/v1/services","/api/v1/services/control","/api/v1/hosts","/api/v1/fleet/config"}: self.send_json(404,{"error":"not found"}); return
if not self.authorized(): self.send_json(401,{"error":"bearer token required"}); return
try:
body=self.read_json()
if path=="/api/v1/hosts":
backup=edit_host(str(body.get("operation","")),str(body.get("id","")),str(body.get("url","")))
self.send_json(200,{"ok":True,"backup":backup,"hosts":host_registry()}); return
if path=="/api/v1/fleet/config":
updates=body.get("updates"); targets=body.get("targets",[]); tokens=body.get("tokens",{})
if not isinstance(targets,list) or not targets or len(targets)>32: raise ValueError("targets must contain 1..32 host identifiers")
if not isinstance(tokens,dict): raise ValueError("tokens must be an object")
results=[]; local_id=machine_identity()["id"]; restart_local_api=False
for host_id in dict.fromkeys(str(x) for x in targets):
try:
if host_id==local_id:
backup=update_ini(updates)
subprocess.run(["systemctl","restart","pigway-pi-control.service"],check=True,timeout=10)
restart_local_api="api" in updates
results.append({"id":host_id,"ok":True,"backup":backup})
else:
host=find_host(host_id); token=str(tokens.get(host_id,""))
if not token: raise ValueError("remote bearer token required")
response=remote_json(host["url"],"/api/v1/config","PUT",{"updates":updates},token)
results.append({"id":host_id,"ok":True,"backup":response.get("backup","")})
except Exception as exc: results.append({"id":host_id,"ok":False,"error":str(exc)})
self.send_json(200,{"ok":all(x["ok"] for x in results),"results":results})
if restart_local_api: threading.Timer(0.2,self.server.shutdown).start()
return
if path=="/api/v1/services/control":
control_service(str(body.get("unit","")),str(body.get("action","")))
print(f"SERVICE_CONTROL action={body.get('action')} unit={body.get('unit')}",flush=True)
+23 -1
View File
@@ -1,5 +1,5 @@
# =============================================================================
# PIGWay Pi Control v3.6.0 - 用户配置文件
# PIGWay Pi Control v3.7.0 - 用户配置文件
# =============================================================================
#
# 正式安装后的配置路径:
@@ -25,6 +25,22 @@
# =============================================================================
[device]
# 多机监控中使用的稳定机器标识。auto 表示读取本机 hostname。
# 多台机器的 identifier 必须唯一;修改后重启硬件 Agent 和 API 服务。
identifier = auto
# Web 页面显示名称。auto 表示使用 identifier。
name = auto
[hosts]
# 在作为聚合入口的机器上登记其他节点,每行格式:
# 唯一标识 = http://节点IP:6001
# 示例:
# living-room = http://192.168.1.20:6001
[api]
# API 与 Web 使用同一监听地址和端口。0.0.0.0 允许局域网访问;
# 如只允许本机访问可改为 127.0.0.1。修改后重启 API 服务生效。
@@ -271,6 +287,12 @@ AWESUN = runawesun.service
# NGINX = nginx.service
[service_monitor_disabled]
# Web 中“暂停监控”的状态保存在这里。
# 名称对应 [services] 左侧名称;true 表示保留配置和备注,但 Agent 暂不检查。
# 通常由 Web 自动维护,无需手工编辑。
[processes]
# =============================================================================
# 普通进程监控
+4 -2
View File
@@ -102,6 +102,8 @@ from pathlib import Path
path=Path(sys.argv[1])
text=path.read_text()
sections={
"device":(("identifier","auto"),("name","auto")),
"hosts":(),
"led":(
("normal_mode","off"),
("normal_r",0),("normal_g",0),("normal_b",0),
@@ -163,7 +165,7 @@ echo "[7/8] 安装后自检..."
FAIL=0
check() { local label="$1"; shift; if "$@"; then printf " %-30s PASS\n" "$label"; else printf " %-30s FAIL\n" "$label"; FAIL=1; fi; }
APP=/usr/local/sbin/pigway-pi-control
check "Version 3.6.0" grep -Fq 'APP_VERSION="3.6.0"' "$APP"
check "Version 3.7.0" grep -Fq 'APP_VERSION="3.7.0"' "$APP"
check "OLED physical 128x32" grep -Fq 'W,H=128,32' "$APP"
check "OLED MUX 0x1F" grep -Fq '0xA8,0x1F' "$APP"
check "OLED pages 0..3" grep -Fq 'cmd(0x22);cmd(0);cmd(3)' "$APP"
@@ -178,7 +180,7 @@ else
fi
check "Service running" systemctl is-active --quiet pigway-pi-control.service
check "API/Web service running" systemctl is-active --quiet pigway-pi-control-api.service
check "API/Web version 3.6.0" grep -Fq 'APP_VERSION="3.6.0"' /usr/local/sbin/pigway-pi-control-api
check "API/Web version 3.7.0" grep -Fq 'APP_VERSION="3.7.0"' /usr/local/sbin/pigway-pi-control-api
check "Web UI installed" test -s /usr/local/share/pigway-pi-control/web/index.html
check "P0 health monitor" grep -Fq 'throttled_state' "$APP"
check "Diagnostic command" grep -Fq -- '--diagnose' "$APP"
+75 -30
View File
File diff suppressed because one or more lines are too long