package control import ( "bytes" "context" "encoding/json" "fmt" "log" "net/http" "os/exec" "strconv" "strings" "time" "tianyan-edge/internal/config" ) type Heartbeat struct { cfg *config.Config } func NewHeartbeat(cfg *config.Config) *Heartbeat { return &Heartbeat{cfg: cfg} } 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(): return case <-ticker.C: h.send(client) } } } func (h *Heartbeat) send(client *http.Client) { payload := map[string]any{ "edge_id": h.cfg.EdgeID, "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)) req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+h.cfg.EdgeToken) resp, err := client.Do(req) if err != nil { log.Printf("heartbeat: %v", err) return } 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 }