diff --git a/DEPLOYMENT.md b/DEPLOYMENT.md index 647b163..9ad7cd5 100644 --- a/DEPLOYMENT.md +++ b/DEPLOYMENT.md @@ -1,197 +1,286 @@ # AI-tianyan 边缘部署操作指南 -## 前置条件 +## 系统环境 | 项目 | 要求 | |---|---| -| 硬件 | Atlas 200I DK2 (Ascend 310B4) 或 Ascend 310B4 设备 | +| 硬件 | Atlas 200I A2 (Ascend 310B4) 或同系列设备 | | OS | Ubuntu 22.04 aarch64 | -| Go | 1.18+ | -| Python | 3.9+ | -| CANN | 6.2.RC2 (Ascend Toolkit) | +| Python | 3.9.x (Miniconda 或系统自带) | +| CANN | 6.2.RC2 (Ascend Toolkit V100R003C11) | +| Go | 1.18+ (仅编译时需要) | | 云端 | 101.36.73.102 (API/MQTT/InfluxDB/OTA/ZLMediaKit) | +> 如果目标设备环境完全一致（同镜像版本），可直接复用已编译的 Go 二进制和 .om 模型，无需重新编译。 + --- -## 步骤 1: 拉取代码 +## 部署方式 + +### 方式一：离线包部署（推荐，最快） + +适用于环境一致的设备，所有构建产物已包含在内。 + +#### 1. 在源设备打包 + +```bash +cd /root +tar czf ai-tianyan.tar.gz --exclude='.git' --exclude='__pycache__' --exclude='*.pyc' AI-tianyan/ +``` + +#### 2. 传输到目标设备 + +```bash +# 网络传输 +scp ai-tianyan.tar.gz root@目标IP:/root/ + +# 或 U盘拷贝 +``` + +#### 3. 目标设备上安装 + +```bash +# 确保 pip 存在 +python3 -m pip --version || apt update && apt install -y python3-pip + +# 解压 +cd /root +tar xzf ai-tianyan.tar.gz +cd AI-tianyan + +# 一键安装（部署文件 + Python依赖 + systemd服务 + 自动启动） +bash scripts/install.sh +``` + +#### 4. 修改配置 + +```bash +# 生成新的设备 UUID +NEW_UUID=$(cat /proc/sys/kernel/random/uuid | tr -d '-') + +# 更新配置 +sudo sed -i "s/device_uuid:.*/device_uuid: $NEW_UUID/" /opt/tianyan-edge/config/edge.yaml +sudo sed -i "s/edge_id:.*/edge_id: edge-demo-002/" /opt/tianyan-edge/config/edge.yaml + +# 如有不同的摄像头流地址 +sudo vi /opt/tianyan-edge/config/edge.yaml +``` + +#### 5. 验证 + +```bash +sudo systemctl status edge-agent edge-infer +journalctl -u edge-agent -n 50 --no-pager +``` + +--- + +### 方式二：Git 源码部署 + +适用于需要自定义代码或环境有差异的设备。 + +#### 1. 拉取代码 ```bash git clone http://101.36.73.102:3112/fyah/AI-tianyan.git cd AI-tianyan ``` ---- - -## 步骤 2: 转换 YOLO 模型 (.onnx → .om) +#### 2. 编译 Go 边缘代理 + +```bash +export GOPROXY=https://goproxy.cn,direct +bash scripts/build.sh +# 输出: build/edge-agent +``` + +#### 3. 转换模型（仅 ATC 版本不同时需要） ```bash -# 在 CANN 环境设备上执行 source /usr/local/Ascend/ascend-toolkit/set_env.sh - -# 执行 ATC 转换 bash scripts/atc_convert.sh - -# 或手动转换 +# 或手动: atc --model=yolov8n.onnx \ --framework=5 \ --output=model/model.om \ --soc_version=Ascend310B4 \ --input_format=NCHW \ - --input_shape="images:1,3,640,640" - -# 复制模型到指定位置 -cp yolov8n.om model/model.om + --input_shape="images:1,3,640,640" \ + --output_type=FP16 ``` ---- - -## 步骤 3: 编译 Go 边缘代理 +#### 4. 安装部署 ```bash -# 设置 Go 代理（国内环境） -export GOPROXY=https://goproxy.cn,direct +# 确保 pip 存在 +python3 -m pip --version || apt update && apt install -y python3-pip -# 构建 -bash scripts/build.sh -# 或使用 make build -# 输出: build/edge-agent -``` +# 安装 Python 依赖 +python3 -m pip install opencv-python-headless numpy ---- - -## 步骤 4: 安装部署 - -```bash # 一键安装 bash scripts/install.sh -# 或 make install - -# 安装脚本会: -# 1. 创建 /opt/tianyan-edge/{bin,python,config,systemd,staging,model} -# 2. 复制 edge-agent 到 /opt/tianyan-edge/bin/ -# 3. 复制 infer_server.py 到 /opt/tianyan-edge/python/ -# 4. 安装 Python 依赖 (opencv-python-headless, numpy) -# 5. 安装 systemd 服务文件 -# 6. 启用并启动服务 ``` --- -## 步骤 5: 配置 edge.yaml +## 安装脚本说明 -```bash -sudo vi /opt/tianyan-edge/config/edge.yaml +`scripts/install.sh` 会自动执行以下操作： + +``` +1. 创建目录: /opt/tianyan-edge/{bin,python,config,systemd,staging,model} +2. 复制 edge-agent -> /opt/tianyan-edge/bin/edge-agent +3. 复制 infer_server.py -> /opt/tianyan-edge/python/ +4. 复制 edge.yaml.template -> /opt/tianyan-edge/config/edge.yaml (仅在不存在时) +5. 安装 Python 依赖 (opencv-python-headless, numpy) +6. 安装 systemd 服务文件 +7. daemon-reload + enable + restart 服务 ``` -关键配置项: +--- + +## 配置文件说明 + +安装后配置位于 `/opt/tianyan-edge/config/edge.yaml`： ```yaml -device_uuid: 8541db9f77826e39605ef2c032f8fb93 # 设备唯一标识 -edge_id: edge-demo-001 -cloud_url: http://101.36.73.102:8004 # 云端 API -mqtt_broker: tcp://101.36.73.102:1883 # MQTT 消息 +device_uuid: 8541db9f77826e39605ef2c032f8fb93 # 设备唯一标识，每台设备必须不同 +edge_id: edge-demo-001 # 设备名称，便于识别 +cloud_url: http://101.36.73.102:8004 # 云端 API 地址 +mqtt_broker: tcp://101.36.73.102:1883 # MQTT Broker +mqtt_user: "" # MQTT 认证用户名 +mqtt_pass: "" # MQTT 认证密码 +edge_token: "" # 设备认证 Token rtsp_urls: - - http://101.36.73.102:8080/rtp/34020000002000000003_34020000001310000001.live.flv -infer_socket: /tmp/edge-infer.sock -infer_fps: 2 # 推理帧率 -conf_threshold: 0.2 # 检测阈值 -ota_url: http://101.36.73.102:8087 # OTA 更新地址 -version: 1.0.0 + - http://101.36.73.102:8080/rtp/xxx.live.flv # 视频流地址 (支持多路) +infer_socket: /tmp/edge-infer.sock # Go与Python通信的Unix Socket +infer_fps: 2 # 推理帧率 (每秒抽样数) +infer_workers: 3 # 并发推理 worker 数 +conf_threshold: 0.2 # 检测置信度阈值 +dedup_window_sec: 30 # 事件去重窗口 (秒) +ota_url: http://101.36.73.102:8087 # OTA 自动更新地址 +version: 1.0.0 # 固件版本号 ``` +> 新设备部署时 **必须修改**: `device_uuid`、`edge_id`、`rtsp_urls` + --- -## 步骤 6: 启动服务 +## 服务管理 -### 方式 A: systemd (推荐) +### 启动/停止/重启 ```bash -# 启动推理服务 -sudo systemctl start edge-infer +sudo systemctl start edge-infer # 启动 NPU 推理服务 +sudo systemctl start edge-agent # 启动边缘代理 +sudo systemctl stop edge-infer edge-agent +sudo systemctl restart edge-infer edge-agent +``` -# 启动边缘代理 -sudo systemctl start edge-agent +### 开机自启 -# 设置开机自启 +```bash sudo systemctl enable edge-infer edge-agent +``` -# 查看状态 +### 查看状态 + +```bash sudo systemctl status edge-agent edge-infer -sudo journalctl -u edge-agent -f +sudo journalctl -u edge-agent -f # 实时查看边缘代理日志 +sudo journalctl -u edge-infer -f # 实时查看推理服务日志 ``` -### 方式 B: 手动启动 +### 手动调试模式 ```bash -# 启动推理服务 +# 停止 systemd 服务 +sudo systemctl stop edge-agent edge-infer + +# 手动启动推理服务 bash -lc 'source /usr/local/Ascend/ascend-toolkit/set_env.sh && \ - NAMES_FILE=/root/AI-tianyan/model/names.txt \ - CONF_THRESHOLD=0.15 \ - python3 python/infer_server.py' & + NAMES_FILE=/opt/tianyan-edge/model/names.txt \ + OM_MODEL=/opt/tianyan-edge/model/model.om \ + CONF_THRESHOLD=0.2 \ + OUTPUT_FORMAT=raw \ + python3 /opt/tianyan-edge/python/infer_server.py' -# 启动边缘代理 -./build/edge-agent -config config/edge.yaml -``` - -### 方式 C: 一键脚本 - -```bash -bash start.sh +# 另一个终端启动边缘代理 +/opt/tianyan-edge/bin/edge-agent -config /opt/tianyan-edge/config/edge.yaml ``` --- -## 步骤 7: 验证运行 +## 运行验证 ```bash # 检查进程 -ps aux | grep -E 'edge-agent|infer_server|ffmpeg' - -# 检查网络连接 -ss -tunp | grep 101.36.73.102 +ps aux | grep -E 'edge-agent|infer_server' # 检查 NPU 状态 npu-smi info -# 检查日志 -sudo journalctl -u edge-agent -n 50 -sudo journalctl -u edge-infer -n 50 -tail -f logs/agent.log logs/infer.log +# 检查网络连接 (应看到连接到 101.36.73.102) +ss -tunp | grep 101.36.73.102 + +# 检查 Unix Socket +ls -la /tmp/edge-infer.sock + +# 检查模型文件 +ls -lh /opt/tianyan-edge/model/model.om /opt/tianyan-edge/model/names.txt + +# 查看系统日志 +tail -f /opt/tianyan-edge/logs/agent.log 2>/dev/null +tail -f /opt/tianyan-edge/logs/infer.log 2>/dev/null ``` --- -## 步骤 8: 配置 Telegraf 指标上报 (可选) +## Docker 部署（可选） ```bash -# Telegraf 已安装 (v1.21.4) -# 配置文件: /etc/telegraf/telegraf.conf -# 已配置上报到 InfluxDB (101.36.73.102:18086) - -# 重启 Telegraf 使配置生效 -sudo systemctl restart telegraf -sudo systemctl status telegraf +cd AI-tianyan +docker compose up -d +docker compose logs -f tianyan-edge ``` +要求：Docker 已安装，CANN 驱动已就绪，NPU 设备节点 `/dev/davinci0` 等存在。 + --- -## 常见问题排查 +## 常见问题 | 问题 | 解决方法 | |---|---| -| `acl.init failed` | 确认 `set_env.sh` 已执行，检查 `LD_LIBRARY_PATH` | -| `model.om not found` | 重新执行 ATC 转换，确认路径正确 | -| 无法连接 MQTT | 检查 `mqtt_broker` 地址和 `edge_token` | -| 拉流失败 | 确认 ZLMediaKit 端口 8080 可达 | -| 编译失败 | `GOPROXY=https://goproxy.cn,direct go mod download` | +| `acl.init failed` | 确认 `source set_env.sh` 已执行，检查 `LD_LIBRARY_PATH` 包含 CANN 路径 | +| `model.om not found` | 确认 `/opt/tianyan-edge/model/model.om` 存在，重新执行 ATC 转换 | +| `python3 -m pip: command not found` | `apt install -y python3-pip` | +| `opencv-python-headless 安装失败` | `apt install -y python3-dev gcc g++ libgl1-mesa-glx` 后再 pip install | +| 无法连接 MQTT | 检查 `mqtt_broker` 地址是否正确，确认网络可达 | +| 拉流失败 | 确认 ZLMediaKit 端口 8080 可达，流地址正确 | +| Go 编译失败 | `export GOPROXY=https://goproxy.cn,direct && go mod download` | +| systemd 启动后立即退出 | `journalctl -u edge-agent -n 100` 查看详细错误 | +| 多设备 UUID 冲突 | 每台设备必须有不同的 `device_uuid`，用 `cat /proc/sys/kernel/random/uuid` 生成 | --- -## 更新部署 (OTA) +## 卸载 ```bash -# 云端推送新版本后，edge-agent 自动检测并下载 -# 也可手动更新: +bash scripts/uninstall.sh +# 或 make uninstall +``` + +--- + +## OTA 远程更新 + +云端推送新版本后，edge-agent 会自动检测并下载更新到 staging 目录。 +也可手动更新： + +```bash +cd /root/AI-tianyan git pull make build make install diff --git a/internal/control/config_agent.go b/internal/control/config_agent.go deleted file mode 100644 index 88ee804..0000000 --- a/internal/control/config_agent.go +++ /dev/null @@ -1,59 +0,0 @@ -package control - -import ( - "context" - "encoding/json" - "fmt" - "log" - "net/http" - "time" - - "tianyan-edge/internal/config" -) - -type ConfigAgent struct { - cfg *config.Config -} - -func NewConfigAgent(cfg *config.Config) *ConfigAgent { - return &ConfigAgent{cfg: cfg} -} - -func (a *ConfigAgent) Run(ctx context.Context) { - client := &http.Client{Timeout: 5 * time.Second} - ticker := time.NewTicker(60 * time.Second) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - a.pull(client) - } - } -} - -func (a *ConfigAgent) pull(client *http.Client) { - url := fmt.Sprintf("%s/api/v1/edge/config?edge_id=%s", a.cfg.CloudURL, a.cfg.EdgeID) - req, _ := http.NewRequest(http.MethodGet, url, nil) - req.Header.Set("Authorization", "Bearer "+a.cfg.EdgeToken) - resp, err := client.Do(req) - if err != nil { - log.Printf("config pull: %v", err) - return - } - defer resp.Body.Close() - if resp.StatusCode != 200 { - return - } - var data map[string]any - if err := json.NewDecoder(resp.Body).Decode(&data); err != nil { - return - } - if v, ok := data["infer_fps"].(float64); ok && int(v) > 0 { - a.cfg.InferFPS = int(v) - } - if v, ok := data["conf_threshold"].(float64); ok { - a.cfg.ConfThreshold = v - } -} diff --git a/internal/control/heartbeat.go b/internal/control/heartbeat.go index 42d3aef..924adf4 100644 --- a/internal/control/heartbeat.go +++ b/internal/control/heartbeat.go @@ -7,6 +7,9 @@ import ( "fmt" "log" "net/http" + "os/exec" + "strconv" + "strings" "time" "tianyan-edge/internal/config" @@ -24,6 +27,10 @@ func (h *Heartbeat) Run(ctx context.Context) { client := &http.Client{Timeout: 5 * time.Second} ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() + + // Send first heartbeat immediately + h.send(client) + for { select { case <-ctx.Done(): @@ -40,6 +47,12 @@ func (h *Heartbeat) send(client *http.Client) { "version": h.cfg.Version, "ts": float64(time.Now().UnixMilli()) / 1000.0, } + + // Collect NPU stats if available + if npu := collectNPUStats(); npu != nil { + payload["npu"] = npu + } + body, _ := json.Marshal(payload) url := fmt.Sprintf("%s/api/v1/edge/heartbeat", h.cfg.CloudURL) req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) @@ -52,3 +65,113 @@ func (h *Heartbeat) send(client *http.Client) { } resp.Body.Close() } + +// NPUStats holds parsed NPU telemetry +type NPUStats struct { + TempC int `json:"temp_c"` + UtilPct int `json:"util_pct"` + MemUsedMB int `json:"mem_used_mb"` + MemTotalMB int `json:"mem_total_mb"` + MemPct int `json:"mem_pct"` +} + +// collectNPUStats runs npu-smi and parses the output +func collectNPUStats() *NPUStats { + // Try npu-smi first (Ascend devices), fall back to the helper script + var commonOut, memOut []byte + var err error + + commonOut, err = exec.Command("/usr/local/sbin/npu-smi", "info", "-t", "common", "-i", "0").CombinedOutput() + if err != nil { + // Fallback to shell script + commonOut, err = exec.Command("bash", "/opt/tianyan-edge/scripts/npu_info.sh").CombinedOutput() + if err != nil { + log.Printf("npu: monitoring unavailable: %v", err) + return nil + } + // If script succeeded, it outputs Influx line protocol — parse that + line := strings.TrimSpace(string(commonOut)) + return parseInfluxLine(line) + } + + memOut, _ = exec.Command("/usr/local/sbin/npu-smi", "info", "-t", "memory", "-i", "0").CombinedOutput() + + return parseNpuSmiOutput(string(commonOut), string(memOut)) +} + +func parseNpuSmiOutput(common, memory string) *NPUStats { + stats := &NPUStats{} + + for _, line := range strings.Split(common, "\n") { + line = strings.TrimSpace(line) + if strings.Contains(line, "Temperature") { + if v := extractLastNumber(line); v != 0 { + stats.TempC = v + } + } + if strings.Contains(line, "Aicore Usage Rate") { + if v := extractLastNumber(line); v != 0 { + stats.UtilPct = v + } + } + if strings.Contains(line, "Memory Usage Rate") { + if v := extractLastNumber(line); v != 0 { + stats.MemPct = v + } + } + } + + for _, line := range strings.Split(memory, "\n") { + line = strings.TrimSpace(line) + if strings.Contains(line, "Capacity") { + if v := extractLastNumber(line); v != 0 { + stats.MemTotalMB = v + } + } + } + + if stats.MemTotalMB > 0 && stats.MemPct > 0 { + stats.MemUsedMB = stats.MemTotalMB * stats.MemPct / 100 + } + + return stats +} + +func parseInfluxLine(line string) *NPUStats { + // Format: npu_status,device=0 temp=X,utilization=Y,memory_used=Z,memory_total=W + stats := &NPUStats{} + parts := strings.Split(line, " ") + if len(parts) < 2 { + return nil + } + for _, field := range strings.Split(parts[1], ",") { + kv := strings.SplitN(field, "=", 2) + if len(kv) != 2 { + continue + } + v, _ := strconv.Atoi(kv[1]) + switch kv[0] { + case "temp": + stats.TempC = v + case "utilization": + stats.UtilPct = v + case "memory_used": + stats.MemUsedMB = v + case "memory_total": + stats.MemTotalMB = v + } + } + return stats +} + +func extractLastNumber(line string) int { + fields := strings.Fields(line) + if len(fields) == 0 { + return 0 + } + // Last field might have units like "%" or "C" + raw := fields[len(fields)-1] + raw = strings.TrimRight(raw, "%C ") + v, _ := strconv.Atoi(raw) + return v +} diff --git a/internal/control/heartbeat_test.go b/internal/control/heartbeat_test.go new file mode 100644 index 0000000..f97d6db --- /dev/null +++ b/internal/control/heartbeat_test.go @@ -0,0 +1,96 @@ +package control + +import ( + "testing" +) + +func TestParseNpuSmiOutput(t *testing.T) { + common := `Temperature(C) : 46 +Aicore Usage Rate(%) : 25 +Memory Usage Rate(%) : 86` + memory := `Capacity(MB) : 3513` + + stats := parseNpuSmiOutput(common, memory) + if stats == nil { + t.Fatal("expected non-nil stats") + } + if stats.TempC != 46 { + t.Errorf("temp: got %d want 46", stats.TempC) + } + if stats.UtilPct != 25 { + t.Errorf("util: got %d want 25", stats.UtilPct) + } + if stats.MemTotalMB != 3513 { + t.Errorf("mem_total: got %d want 3513", stats.MemTotalMB) + } + if stats.MemPct != 86 { + t.Errorf("mem_pct: got %d want 86", stats.MemPct) + } + // 3513 * 86 / 100 = 3021 + if stats.MemUsedMB != 3021 { + t.Errorf("mem_used: got %d want 3021", stats.MemUsedMB) + } +} + +func TestParseInfluxLine(t *testing.T) { + line := "npu_status,device=0 temp=46,utilization=0,memory_used=3021,memory_total=3513" + stats := parseInfluxLine(line) + if stats == nil { + t.Fatal("expected non-nil stats") + } + if stats.TempC != 46 { + t.Errorf("temp: got %d want 46", stats.TempC) + } + if stats.UtilPct != 0 { + t.Errorf("util: got %d want 0", stats.UtilPct) + } + if stats.MemUsedMB != 3021 { + t.Errorf("mem_used: got %d want 3021", stats.MemUsedMB) + } + if stats.MemTotalMB != 3513 { + t.Errorf("mem_total: got %d want 3513", stats.MemTotalMB) + } +} + +func TestParseInfluxLineInvalid(t *testing.T) { + // Input with no space-separated fields returns nil + stats := parseInfluxLine("garbage") + if stats != nil { + t.Error("expected nil stats for garbage input without fields") + } + + // Input with fields but no valid data returns zero-valued stats + stats2 := parseInfluxLine("npu_status,device=0 ") + if stats2 == nil { + t.Fatal("expected non-nil stats for input with fields") + } + if stats2.TempC != 0 || stats2.UtilPct != 0 { + t.Errorf("expected zero values, got temp=%d util=%d", stats2.TempC, stats2.UtilPct) + } +} + +func TestExtractLastNumber(t *testing.T) { + tests := []struct { + input string + want int + }{ + {"Temperature(C) : 46", 46}, + {"Aicore Usage Rate(%) : 25%", 25}, + {" 86%", 86}, + {"no numbers here", 0}, + {"", 0}, + {"value: 123C", 123}, + } + for _, tt := range tests { + got := extractLastNumber(tt.input) + if got != tt.want { + t.Errorf("extractLastNumber(%q) = %d, want %d", tt.input, got, tt.want) + } + } +} + +func TestInMaintenanceWindow(t *testing.T) { + // Can't test specific hours without mocking time.Time, + // but verify it doesn't panic + _ = inMaintenanceWindow() +} diff --git a/internal/control/ota_agent.go b/internal/control/ota_agent.go index d9194f1..99539ec 100644 --- a/internal/control/ota_agent.go +++ b/internal/control/ota_agent.go @@ -1,7 +1,9 @@ package control import ( + "archive/tar" "bytes" + "compress/gzip" "context" "crypto/sha256" "encoding/hex" @@ -11,6 +13,7 @@ import ( "log" "net/http" "os" + "os/exec" "path/filepath" "time" @@ -29,6 +32,12 @@ func (o *OTAAgent) Run(ctx context.Context) { client := &http.Client{Timeout: 30 * time.Second} ticker := time.NewTicker(10 * time.Minute) defer ticker.Stop() + + // Check once on startup if in maintenance window + if inMaintenanceWindow() { + o.check(client) + } + for { select { case <-ctx.Done(): @@ -69,10 +78,16 @@ func (o *OTAAgent) check(client *http.Client) { return } if err := o.downloadAndVerify(client, mf); err != nil { + log.Printf("ota: download/verify failed: %v", err) o.report(client, target, "failed") return } - o.report(client, target, "verified") + if err := o.install(target); err != nil { + log.Printf("ota: install failed: %v", err) + o.report(client, target, "install_failed") + return + } + o.report(client, target, "installed") } func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]string) error { @@ -93,13 +108,150 @@ func (o *OTAAgent) downloadAndVerify(client *http.Client, mf map[string]string) sum := sha256.Sum256(data) actual := hex.EncodeToString(sum[:]) if actual != expected { - return fmt.Errorf("checksum mismatch") + return fmt.Errorf("checksum mismatch: expected %s got %s", expected, actual) } _ = os.MkdirAll("/opt/tianyan-edge/staging", 0o755) path := filepath.Join("/opt/tianyan-edge/staging", target+".tar.gz") return os.WriteFile(path, data, 0o644) } +func (o *OTAAgent) install(target string) error { + stagingPath := filepath.Join("/opt/tianyan-edge/staging", target+".tar.gz") + data, err := os.ReadFile(stagingPath) + if err != nil { + return fmt.Errorf("read staging file: %w", err) + } + + // Extract tar.gz into a temp directory + tmpDir, err := os.MkdirTemp("/opt/tianyan-edge/staging", "ota-"+target) + if err != nil { + return fmt.Errorf("create temp dir: %w", err) + } + defer os.RemoveAll(tmpDir) + + if err := extractTarGz(bytes.NewReader(data), tmpDir); err != nil { + return fmt.Errorf("extract: %w", err) + } + + // Look for edge-agent binary in extracted content + newBinary := filepath.Join(tmpDir, "edge-agent") + if _, err := os.Stat(newBinary); os.IsNotExist(err) { + // Try finding it recursively + found := "" + var stopWalk = fmt.Errorf("found") + filepath.Walk(tmpDir, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + if info.Name() == "edge-agent" && !info.IsDir() { + found = path + return stopWalk + } + return nil + }) + if found == "" { + return fmt.Errorf("edge-agent binary not found in package") + } + newBinary = found + } + + // Verify the new binary is actually an ELF executable + header, err := os.Open(newBinary) + if err != nil { + return fmt.Errorf("open new binary: %w", err) + } + magic := make([]byte, 4) + header.Read(magic) + header.Close() + if !bytes.Equal(magic, []byte{0x7f, 0x45, 0x4c, 0x46}) { // ELF magic + return fmt.Errorf("new binary is not a valid ELF executable") + } + + // Backup current binary + backupPath := "/opt/tianyan-edge/edge-agent.bak" + if err := copyFile("/opt/tianyan-edge/edge-agent", backupPath); err != nil { + log.Printf("ota: backup failed (non-fatal): %v", err) + } + + // Replace binary + if err := copyFile(newBinary, "/opt/tianyan-edge/edge-agent"); err != nil { + return fmt.Errorf("replace binary: %w", err) + } + if err := os.Chmod("/opt/tianyan-edge/edge-agent", 0755); err != nil { + return fmt.Errorf("chmod: %w", err) + } + + // Update version in config + o.cfg.Version = target + _ = o.cfg.Save() + + // Clean up staging + os.Remove(stagingPath) + + // Restart service via systemd + log.Println("ota: binary replaced, restarting service...") + cmd := exec.Command("systemctl", "restart", "edge-agent") + if err := cmd.Run(); err != nil { + log.Printf("ota: systemctl restart failed: %v (binary replaced, manual restart needed)", err) + // Binary is already replaced, service will use new version on next boot + } + + return nil +} + +func extractTarGz(r io.Reader, dest string) error { + gzr, err := gzip.NewReader(r) + if err != nil { + return err + } + defer gzr.Close() + + tr := tar.NewReader(gzr) + for { + hdr, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return err + } + target := filepath.Join(dest, hdr.Name) + switch hdr.Typeflag { + case tar.TypeDir: + if err := os.MkdirAll(target, os.FileMode(hdr.Mode)); err != nil { + return err + } + case tar.TypeReg: + os.MkdirAll(filepath.Dir(target), 0755) + f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY, os.FileMode(hdr.Mode)) + if err != nil { + return err + } + if _, err := io.Copy(f, tr); err != nil { + f.Close() + return err + } + f.Close() + } + } + return nil +} + +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + _, err = io.Copy(out, in) + return err +} + func (o *OTAAgent) report(client *http.Client, version, status string) { payload := map[string]any{"edge_id": o.cfg.EdgeID, "version": version, "status": status, "ts": float64(time.Now().UnixMilli()) / 1000.0} diff --git a/internal/control/ota_agent_test.go b/internal/control/ota_agent_test.go new file mode 100644 index 0000000..31c51b5 --- /dev/null +++ b/internal/control/ota_agent_test.go @@ -0,0 +1,106 @@ +package control + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "os" + "path/filepath" + "testing" +) + +func TestExtractTarGz(t *testing.T) { + // Create a test tar.gz with a text file and a binary-like file + var buf bytes.Buffer + gw := gzip.NewWriter(&buf) + tw := tar.NewWriter(gw) + + // Add a directory + tw.WriteHeader(&tar.Header{ + Name: "testdir/", + Typeflag: tar.TypeDir, + Mode: 0755, + }) + + // Add a text file + content := []byte("hello world") + tw.WriteHeader(&tar.Header{ + Name: "testdir/hello.txt", + Size: int64(len(content)), + Mode: 0644, + }) + tw.Write(content) + + // Add a binary file (fake ELF header) + binContent := []byte{0x7f, 0x45, 0x4c, 0x46, 0x01, 0x02} + tw.WriteHeader(&tar.Header{ + Name: "testdir/edge-agent", + Size: int64(len(binContent)), + Mode: 0755, + }) + tw.Write(binContent) + + tw.Close() + gw.Close() + + // Extract to temp dir + dest, _ := os.MkdirTemp("", "ota-test-*") + defer os.RemoveAll(dest) + + if err := extractTarGz(bytes.NewReader(buf.Bytes()), dest); err != nil { + t.Fatalf("extractTarGz failed: %v", err) + } + + // Verify extracted files + helloPath := filepath.Join(dest, "testdir", "hello.txt") + data, err := os.ReadFile(helloPath) + if err != nil { + t.Fatalf("hello.txt not found: %v", err) + } + if string(data) != "hello world" { + t.Errorf("hello.txt content: got %q want %q", string(data), "hello world") + } + + binPath := filepath.Join(dest, "testdir", "edge-agent") + data, err = os.ReadFile(binPath) + if err != nil { + t.Fatalf("edge-agent not found: %v", err) + } + if !bytes.Equal(data, binContent) { + t.Errorf("edge-agent content mismatch") + } +} + +func TestExtractTarGzInvalid(t *testing.T) { + dest, _ := os.MkdirTemp("", "ota-test-*") + defer os.RemoveAll(dest) + + if err := extractTarGz(bytes.NewReader([]byte("not a gzip")), dest); err == nil { + t.Error("expected error for invalid gzip") + } +} + +func TestCopyFile(t *testing.T) { + src, _ := os.CreateTemp("", "src-*") + src.Write([]byte("test data")) + src.Close() + defer os.Remove(src.Name()) + + dst := src.Name() + ".copy" + defer os.Remove(dst) + + if err := copyFile(src.Name(), dst); err != nil { + t.Fatalf("copyFile failed: %v", err) + } + + data, _ := os.ReadFile(dst) + if string(data) != "test data" { + t.Errorf("copy content: got %q want %q", string(data), "test data") + } +} + +func TestCopyFileNotFound(t *testing.T) { + if err := copyFile("/nonexistent/file", "/tmp/dst"); err == nil { + t.Error("expected error for nonexistent source") + } +} diff --git a/internal/event/uploader_test.go b/internal/event/uploader_test.go new file mode 100644 index 0000000..658776e --- /dev/null +++ b/internal/event/uploader_test.go @@ -0,0 +1,25 @@ +package event + +import ( + "testing" +) + +func TestBufferCap(t *testing.T) { + u := &Uploader{buf: []SuspectedEvent{}} + for i := 0; i < 600; i++ { + u.buffer(SuspectedEvent{EventID: "test"}) + } + if len(u.buf) != maxBuffer { + t.Errorf("buffer len: got %d want %d", len(u.buf), maxBuffer) + } +} + +func TestUploadError(t *testing.T) { + err := &UploadError{Fatal: true, Msg: "test error"} + if err.Error() != "test error" { + t.Errorf("Error(): got %q want %q", err.Error(), "test error") + } + if !err.Fatal { + t.Error("Fatal flag should be true") + } +} diff --git a/internal/infer/client.go b/internal/infer/client.go index c1c2a9b..14cceec 100644 --- a/internal/infer/client.go +++ b/internal/infer/client.go @@ -208,10 +208,17 @@ func (c *Client) send(conn net.Conn, f stream.Frame) error { if err != nil { return err } + conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + defer conn.SetWriteDeadline(time.Time{}) return writeMsg(conn, data) } func (c *Client) recv(conn net.Conn) (*resultMsg, error) { + // Set read deadline to prevent blocking forever if inference server hangs + // 30s timeout covers even slow NPU inference on large images + conn.SetReadDeadline(time.Now().Add(30 * time.Second)) + defer conn.SetReadDeadline(time.Time{}) // reset after read + data, err := readMsg(conn) if err != nil { return nil, err diff --git a/internal/infer/client_test.go b/internal/infer/client_test.go new file mode 100644 index 0000000..f9297dd --- /dev/null +++ b/internal/infer/client_test.go @@ -0,0 +1,80 @@ +package infer + +import ( + "testing" + + "tianyan-edge/internal/config" +) + +func TestIsDuplicate(t *testing.T) { + cfg := &config.Config{DedupWindowSec: 30} + c := &Client{ + cfg: cfg, + dedup: &dedupState{seen: make(map[string]float64)}, + } + + // First detection should NOT be duplicate (uses default config dedup window of 30s) + // ts=100.0, key="cam-001|person" — first time, not duplicate + if c.isDuplicate("cam-001", "person", 100.0) { + t.Error("first detection should not be duplicate") + } + + // Same device+class within window should BE duplicate + // ts=105.0, 105-100=5 < 30, duplicate + if !c.isDuplicate("cam-001", "person", 105.0) { + t.Error("same event within window should be duplicate") + } + + // Same device+class outside window should NOT be duplicate + // ts=200.0, 200-105=95 > 30, not duplicate (updates the timestamp) + if c.isDuplicate("cam-001", "person", 200.0) { + t.Error("same event outside window should not be duplicate") + } + + // Different class should NOT be duplicate + if c.isDuplicate("cam-001", "car", 105.0) { + t.Error("different class should not be duplicate") + } + + // Different device should NOT be duplicate + if c.isDuplicate("cam-002", "person", 105.0) { + t.Error("different device should not be duplicate") + } +} + +func TestDedupMapCleanup(t *testing.T) { + cfg := &config.Config{DedupWindowSec: 30} + c := &Client{ + cfg: cfg, + dedup: &dedupState{seen: make(map[string]float64)}, + } + + // Fill up the map with old entries + for i := 0; i < 5001; i++ { + c.dedup.seen["cam-001|person"] = float64(i * 10) + } + // Trigger cleanup by adding a new entry + c.isDuplicate("cam-001", "person", 99999.0) + + if len(c.dedup.seen) > 5000 { + t.Errorf("dedup map should be cleaned up, got %d entries", len(c.dedup.seen)) + } +} + +func TestMustUUID(t *testing.T) { + u1 := mustUUID() + u2 := mustUUID() + if u1 == u2 { + t.Error("UUIDs should be unique") + } + // Check format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx (32 hex chars with dashes) + parts := 0 + for _, c := range u1 { + if c == '-' { + parts++ + } + } + if parts != 4 { + t.Errorf("UUID format: expected 4 dashes, got %d", parts) + } +} diff --git a/internal/stream/ingestor.go b/internal/stream/ingestor.go index 5e55ebd..a2957d7 100644 --- a/internal/stream/ingestor.go +++ b/internal/stream/ingestor.go @@ -22,10 +22,15 @@ type Frame struct { TS float64 } +type streamProcess struct { + cmd *exec.Cmd + fps int +} + type StreamManager struct { cfg *config.Config frames chan<- Frame - processes map[string]*exec.Cmd + processes map[string]*streamProcess // url -> {cmd, fps} mu sync.Mutex updates <-chan *config.Config } @@ -34,14 +39,14 @@ func NewStreamManager(cfg *config.Config, frames chan<- Frame, updates <-chan *c return &StreamManager{ cfg: cfg, frames: frames, - processes: make(map[string]*exec.Cmd), + processes: make(map[string]*streamProcess), updates: updates, } } func (sm *StreamManager) Run(ctx context.Context) { // Initial start - sm.applyStreams(ctx, sm.cfg.RTSPURLs) + sm.applyStreams(ctx, sm.cfg.RTSPURLs, sm.cfg.InferFPS) for { select { @@ -49,19 +54,19 @@ func (sm *StreamManager) Run(ctx context.Context) { sm.stopAll() return case newCfg := <-sm.updates: - log.Printf("stream: applying new config, urls=%d", len(newCfg.RTSPURLs)) - sm.applyStreams(ctx, newCfg.RTSPURLs) + log.Printf("stream: applying new config, urls=%d fps=%d", len(newCfg.RTSPURLs), newCfg.InferFPS) + sm.applyStreams(ctx, newCfg.RTSPURLs, newCfg.InferFPS) } } } -func (sm *StreamManager) applyStreams(ctx context.Context, urls []string) { +func (sm *StreamManager) applyStreams(ctx context.Context, urls []string, fps int) { sm.mu.Lock() defer sm.mu.Unlock() // Identify URLs to remove (present in sm.processes but not in new urls) - toRemove := map[string]*exec.Cmd{} - for url, cmd := range sm.processes { + toRemove := map[string]*streamProcess{} + for url, sp := range sm.processes { found := false for _, u := range urls { if u == url { @@ -70,27 +75,35 @@ func (sm *StreamManager) applyStreams(ctx context.Context, urls []string) { } } if !found { - toRemove[url] = cmd + toRemove[url] = sp } } // Stop removed streams - for url, cmd := range toRemove { + for url, sp := range toRemove { log.Printf("stream: stopping %s", url) - cmd.Process.Kill() - cmd.Wait() + sm.stopProcess(sp) delete(sm.processes, url) } - // Start new streams + // Restart streams with changed fps + for url, sp := range sm.processes { + if sp.fps != fps { + log.Printf("stream: restarting %s (fps %d -> %d)", url, sp.fps, fps) + sm.stopProcess(sp) + delete(sm.processes, url) + } + } + + // Start new / restarted streams for i, url := range urls { if _, exists := sm.processes[url]; !exists { - go sm.streamLoop(ctx, i, url) + go sm.streamLoop(ctx, i, url, fps) } } } -func (sm *StreamManager) streamLoop(ctx context.Context, idx int, url string) { +func (sm *StreamManager) streamLoop(ctx context.Context, idx int, url string, fps int) { deviceID := fmt.Sprintf("cam-%03d", idx) for { select { @@ -99,9 +112,17 @@ func (sm *StreamManager) streamLoop(ctx context.Context, idx int, url string) { default: } + // Use current config fps (may have been updated) + currentFps := fps + sm.mu.Lock() + if sm.cfg != nil && sm.cfg.InferFPS > 0 { + currentFps = sm.cfg.InferFPS + } + sm.mu.Unlock() + cmd := exec.CommandContext(ctx, "ffmpeg", "-i", url, - "-vf", fmt.Sprintf("fps=%d", sm.cfg.InferFPS), + "-vf", fmt.Sprintf("fps=%d", currentFps), "-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "5", "-", ) stdout, err := cmd.StdoutPipe() @@ -111,20 +132,29 @@ func (sm *StreamManager) streamLoop(ctx context.Context, idx int, url string) { continue } + sp := &streamProcess{cmd: cmd, fps: currentFps} sm.mu.Lock() - sm.processes[url] = cmd + sm.processes[url] = sp sm.mu.Unlock() if err := cmd.Start(); err != nil { log.Printf("stream[%d] start failed: %v", idx, err) + sm.mu.Lock() + delete(sm.processes, url) + sm.mu.Unlock() time.Sleep(5 * time.Second) continue } - log.Printf("stream[%d] connected url=%s", idx, url) + log.Printf("stream[%d] connected fps=%d url=%s", idx, currentFps, url) sm.readFrames(ctx, stdout, idx, deviceID, url) cmd.Wait() + // Clean up from map after exit + sm.mu.Lock() + delete(sm.processes, url) + sm.mu.Unlock() + select { case <-ctx.Done(): return @@ -175,12 +205,20 @@ func (sm *StreamManager) readFrames(ctx context.Context, r io.Reader, idx int, d } } +func (sm *StreamManager) stopProcess(sp *streamProcess) { + if sp == nil || sp.cmd == nil || sp.cmd.Process == nil { + return + } + sp.cmd.Process.Kill() + sp.cmd.Wait() // Prevent zombie processes +} + func (sm *StreamManager) stopAll() { sm.mu.Lock() defer sm.mu.Unlock() - for url, cmd := range sm.processes { + for url, sp := range sm.processes { log.Printf("stream: stopping all %s", url) - cmd.Process.Kill() + sm.stopProcess(sp) delete(sm.processes, url) } } diff --git a/systemd/edge-agent.service b/systemd/edge-agent.service index 7f7c661..2bb76e5 100644 --- a/systemd/edge-agent.service +++ b/systemd/edge-agent.service @@ -8,7 +8,7 @@ Type=simple WorkingDirectory=/opt/tianyan-edge Environment="PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" Environment="LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/arm64-linux/lib64:/usr/local/Ascend/driver/lib64" -ExecStart=/opt/tianyan-edge/edge-agent -config /opt/tianyan-edge/config/edge.yaml +ExecStart=/opt/tianyan-edge/bin/edge-agent -config /opt/tianyan-edge/config/edge.yaml Restart=on-failure RestartSec=5 StartLimitBurst=5