fix: 修复6个边缘侧问题 + 单元测试

1. 删除 ConfigAgent (与 MQTT 配置更新重叠且未启用)
2. OTA 完整安装流程: 解压tar.gz -> 校验ELF -> 替换二进制 -> systemctl重启
3. StreamManager stopAll 添加 Wait() 防止僵尸进程
4. InferClient socket 读写添加 timeout 防止永久阻塞
5. Heartbeat 集成 NPU 监控 (npu-smi + fallback 脚本)
6. 配置更新时动态重启 ffmpeg 以应用新 fps

internal/control/config_agent.go

@@ -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
-	}
-}

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
+}

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()
+}

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}

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")
+	}
+}