Files
BitMonitor/cmd/server/main.go
T

803 lines
24 KiB
Go

package main
import (
_ "embed"
"encoding/json"
"flag"
"fmt"
"log"
"net/http"
"os"
"path/filepath"
"runtime"
"strconv"
"sync"
"time"
"github.com/gorilla/websocket"
"gopkg.in/yaml.v3"
)
// ═══════════════════════════════════════════════
// 日志
// ═══════════════════════════════════════════════
func init() {
log.SetFlags(0)
log.SetOutput(&logWriter{})
}
type logWriter struct{}
func (w *logWriter) Write(p []byte) (int, error) {
ts := time.Now().Format("15:04:05")
return fmt.Fprintf(os.Stderr, "%s %s", ts, p)
}
func logInfo(tag, format string, args ...interface{}) {
msg := fmt.Sprintf(format, args...)
log.Printf("[%-8s] %s", tag, msg)
}
// ═══════════════════════════════════════════════
// 版本
// ═══════════════════════════════════════════════
var (
Version = "dev"
BuildTime = "unknown"
)
// ═══════════════════════════════════════════════
// 路径工具
// ═══════════════════════════════════════════════
func exeDir() string {
p, err := os.Executable()
if err != nil {
return "."
}
return filepath.Dir(p)
}
func resolvePath(baseDir, p string) string {
if filepath.IsAbs(p) {
return p
}
return filepath.Join(baseDir, p)
}
// ═══════════════════════════════════════════════
// 配置结构
// ═══════════════════════════════════════════════
type AuthConfig struct {
MinerID string `yaml:"miner_id"`
Password string `yaml:"password"`
}
type ServerConfig struct {
Listen string `yaml:"listen"`
DevicesFile string `yaml:"devices_file"`
Auth AuthConfig `yaml:"auth"`
MinerUID string `yaml:"miner_uid"`
MinerAPIBase string `yaml:"miner_api_base"`
FetchInterval int `yaml:"fetch_interval"`
HistoryMax int `yaml:"history_max"`
}
func defaultConfig() ServerConfig {
return ServerConfig{
Listen: ":8080",
DevicesFile: "data/devices.yaml",
Auth: AuthConfig{MinerID: "admin", Password: "123456"},
MinerAPIBase: "https://pool.kryptex.com/prl/api/v3/miner/workers",
FetchInterval: 30,
HistoryMax: 200,
}
}
// ═══════════════════════════════════════════════
// 配置加载(不存在则自动生成)
// ═══════════════════════════════════════════════
// ═══════════════════════════════════════════════
// 路径工具
// ═══════════════════════════════════════════════
// baseDir: 优先工作目录,其次 exe 目录
func baseDir() string {
// 当前工作目录
if wd, err := os.Getwd(); err == nil {
return wd
}
// exe 目录(打包后使用)
if p, err := os.Executable(); err == nil {
return filepath.Dir(p)
}
return "."
}
// ═══════════════════════════════════════════════
// 配置加载(不存在则自动生成)
// ═══════════════════════════════════════════════
// 搜索顺序: explicit > 工作目录 > exe 目录
func findConfigPaths(explicit string) []string {
var paths []string
if explicit != "" {
paths = append(paths, explicit)
}
// 工作目录(go run 时 = 项目目录)
if wd, err := os.Getwd(); err == nil {
paths = append(paths, filepath.Join(wd, "server.yaml"))
}
// exe 目录(打包分发时 = exe 所在目录)
if p, err := os.Executable(); err == nil {
d := filepath.Dir(p)
if wd, _ := os.Getwd(); d != wd {
paths = append(paths, filepath.Join(d, "server.yaml"))
}
}
return paths
}
func loadOrCreateConfig(explicitPath string) (ServerConfig, string) {
cfg := defaultConfig()
paths := findConfigPaths(explicitPath)
// 1. 尝试找到已有配置文件
for _, p := range paths {
abs, _ := filepath.Abs(p)
raw, err := os.ReadFile(abs)
if err != nil {
continue
}
if err := yaml.Unmarshal(raw, &cfg); err != nil {
logInfo("config", "解析 %s 失败: %v", abs, err)
continue
}
fillDefaults(&cfg)
logInfo("config", "已加载: %s", abs)
logInfo("config", "listen=%s devices=%s", cfg.Listen, cfg.DevicesFile)
return cfg, abs
}
// 2. 全部不存在 → 在第一个搜索路径生成默认配置
genPath, _ := filepath.Abs(paths[0])
fillDefaults(&cfg)
out, _ := yaml.Marshal(&cfg)
hdr := "# Hardware Monitor Server 配置\n#\n"
hdr += "# listen : 监听地址\n"
hdr += "# devices_file : 设备注册表路径 (相对于配置文件所在目录)\n"
hdr += "#\n"
hdr += "# auth:\n"
hdr += "# miner_id : 登录设备ID\n"
hdr += "# password : 登录密码 (明文)\n"
hdr += "#\n"
hdr += "# miner_uid : Kryptex 钱包地址\n"
hdr += "# miner_api_base: 矿池 API 地址\n"
hdr += "# fetch_interval: 拉取间隔 (秒)\n"
hdr += "# history_max : 算力历史最大条数\n"
hdr += "#\n"
hdr += "# 删除此文件可恢复默认配置\n\n"
dir := filepath.Dir(genPath)
os.MkdirAll(dir, 0755)
os.WriteFile(genPath, append([]byte(hdr), out...), 0644)
logInfo("config", "已生成默认配置: %s", genPath)
return cfg, genPath
}
func fillDefaults(cfg *ServerConfig) {
if cfg.Listen == "" {
cfg.Listen = ":8080"
}
if cfg.DevicesFile == "" {
cfg.DevicesFile = "data/devices.yaml"
}
if cfg.Auth.MinerID == "" {
cfg.Auth.MinerID = "admin"
}
if cfg.Auth.Password == "" {
cfg.Auth.Password = "123456"
}
if cfg.MinerAPIBase == "" {
cfg.MinerAPIBase = "https://pool.kryptex.com/prl/api/v3/miner/workers"
}
if cfg.FetchInterval <= 0 {
cfg.FetchInterval = 30
}
if cfg.HistoryMax <= 0 {
cfg.HistoryMax = 200
}
}
// ═══════════════════════════════════════════════
// 设备注册表
// ═══════════════════════════════════════════════
type DeviceFile struct {
Devices map[string]*DeviceRecord `yaml:"devices"`
}
type DeviceRecord struct {
Name string `yaml:"machine_name" json:"machine_name"`
FirstSeen time.Time `yaml:"first_seen" json:"first_seen"`
LastSeen time.Time `yaml:"last_seen" json:"last_seen"`
Board BoardInfo `yaml:"motherboard" json:"motherboard"`
CPUModel string `yaml:"cpu_model" json:"cpu_model"`
CPUCoresP int `yaml:"cpu_cores_physical" json:"cpu_cores_physical"`
CPUCoresL int `yaml:"cpu_cores_logical" json:"cpu_cores_logical"`
GPUs []GPURec `yaml:"gpus" json:"gpus"`
}
type BoardInfo struct {
Manufacturer string `yaml:"manufacturer,omitempty" json:"manufacturer,omitempty"`
Product string `yaml:"product,omitempty" json:"product,omitempty"`
}
type GPURec struct {
Name string `yaml:"name" json:"name"`
VRAM float64 `yaml:"vram_total_mb" json:"vram_total_mb"`
}
// ═══════════════════════════════════════════════
// 矿池数据
// ═══════════════════════════════════════════════
type MiningData struct {
Status string `json:"status"`
Coin string `json:"coin"`
Valid int `json:"valid"`
Stale int `json:"stale"`
Invalid int `json:"invalid"`
Hashrate30m float64 `json:"hashrate_30m"`
Hashrate3h float64 `json:"hashrate_3h"`
Hashrate24h float64 `json:"hashrate_24h"`
Agent string `json:"agent"`
Country string `json:"country"`
LastActive int64 `json:"last_active"`
OpenedAt int64 `json:"opened_at"`
FetchedAt time.Time `json:"fetched_at"`
}
type HashratePoint struct {
T string `json:"t"`
V float64 `json:"v"`
}
// ═══════════════════════════════════════════════
// 消息结构
// ═══════════════════════════════════════════════
type AgentPayload struct {
MachineName string `json:"machine_name"`
Motherboard struct {
Manufacturer string `json:"manufacturer"`
Product string `json:"product"`
} `json:"motherboard"`
CPU struct {
Model string `json:"model"`
CoresPhysical int `json:"cores_physical"`
CoresLogical int `json:"cores_logical"`
} `json:"cpu"`
GPUs []struct {
Name string `json:"name"`
VRAM float64 `json:"vram_total_mb"`
} `json:"gpus"`
}
type BrowserDevice struct {
Info *DeviceRecord `json:"info"`
Live json.RawMessage `json:"live,omitempty"`
Online bool `json:"online"`
Mining *MiningData `json:"mining,omitempty"`
History []HashratePoint `json:"hash_history,omitempty"`
}
type BrowserMsg struct {
Type string `json:"type"`
Msg string `json:"msg,omitempty"`
Machine string `json:"machine,omitempty"`
Device *BrowserDevice `json:"device,omitempty"`
Mining *MiningData `json:"mining,omitempty"`
History []HashratePoint `json:"hash_history,omitempty"`
Devices map[string]*BrowserDevice `json:"devices,omitempty"`
}
type BrowserCmd struct {
Type string `json:"type"`
Action string `json:"action"`
Machine string `json:"machine"`
MinerID string `json:"miner_id"`
Password string `json:"password"`
}
// ═══════════════════════════════════════════════
// Hub
// ═══════════════════════════════════════════════
type Hub struct {
mu sync.RWMutex
fileMu sync.Mutex
records map[string]*DeviceRecord
live map[string]json.RawMessage
monitors map[*websocket.Conn]struct{}
miningData map[string]*MiningData
hashHist map[string][]HashratePoint
devicesAbs string
config ServerConfig
}
func NewHub(cfg ServerConfig, devicesAbs string) *Hub {
h := &Hub{
records: make(map[string]*DeviceRecord),
live: make(map[string]json.RawMessage),
monitors: make(map[*websocket.Conn]struct{}),
miningData: make(map[string]*MiningData),
hashHist: make(map[string][]HashratePoint),
devicesAbs: devicesAbs,
config: cfg,
}
h.loadYAML()
return h
}
func (h *Hub) loadYAML() {
data, err := os.ReadFile(h.devicesAbs)
if err != nil {
return
}
var f DeviceFile
if yaml.Unmarshal(data, &f) != nil || f.Devices == nil {
return
}
h.records = f.Devices
logInfo("yaml", "加载 %d 台设备", len(h.records))
for name, r := range h.records {
log.Printf(" · %-20s CPU: %-36s %s", name, r.CPUModel, r.LastSeen.Format("01-02 15:04"))
}
}
func (h *Hub) saveYAML() {
h.mu.RLock()
snap := make(map[string]*DeviceRecord, len(h.records))
for k, v := range h.records {
c := *v
snap[k] = &c
}
h.mu.RUnlock()
out, err := yaml.Marshal(&DeviceFile{Devices: snap})
if err != nil {
return
}
hdr := fmt.Sprintf("# Hardware Monitor — 设备注册表\n# 更新: %s\n\n", time.Now().Format("2006-01-02 15:04:05"))
dir := filepath.Dir(h.devicesAbs)
os.MkdirAll(dir, 0755)
h.fileMu.Lock()
defer h.fileMu.Unlock()
os.WriteFile(h.devicesAbs, append([]byte(hdr), out...), 0644)
}
func (h *Hub) UpdateDevice(raw json.RawMessage) {
var p AgentPayload
if json.Unmarshal(raw, &p) != nil || p.MachineName == "" {
return
}
name := p.MachineName
now := time.Now()
h.mu.Lock()
rec, existed := h.records[name]
if !existed {
rec = &DeviceRecord{Name: name, FirstSeen: now}
h.records[name] = rec
}
rec.LastSeen = now
rec.Board = BoardInfo{Manufacturer: p.Motherboard.Manufacturer, Product: p.Motherboard.Product}
rec.CPUModel = p.CPU.Model
rec.CPUCoresP = p.CPU.CoresPhysical
rec.CPUCoresL = p.CPU.CoresLogical
rec.GPUs = make([]GPURec, 0, len(p.GPUs))
for _, g := range p.GPUs {
rec.GPUs = append(rec.GPUs, GPURec{Name: g.Name, VRAM: g.VRAM})
}
cp := *rec
h.live[name] = raw
var md *MiningData
var hist []HashratePoint
if m, ok := h.miningData[name]; ok {
md = m
}
if hs, ok := h.hashHist[name]; ok {
hist = make([]HashratePoint, len(hs))
copy(hist, hs)
}
h.mu.Unlock()
if !existed {
logInfo("new", "%s CPU=%s", name, cp.CPUModel)
}
h.saveYAML()
h.broadcast(BrowserMsg{
Type: "device", Machine: name,
Device: &BrowserDevice{Info: &cp, Live: raw, Online: true, Mining: md, History: hist},
})
}
func (h *Hub) MarkOffline(name string) {
h.mu.Lock()
_, had := h.live[name]
delete(h.live, name)
rec, ok := h.records[name]
var c *DeviceRecord
if ok {
cp := *rec
c = &cp
}
var md *MiningData
var hist []HashratePoint
if m, ok := h.miningData[name]; ok {
md = m
}
if hs, ok := h.hashHist[name]; ok {
hist = make([]HashratePoint, len(hs))
copy(hist, hs)
}
h.mu.Unlock()
if had && c != nil {
h.broadcast(BrowserMsg{
Type: "device", Machine: name,
Device: &BrowserDevice{Info: c, Online: false, Mining: md, History: hist},
})
}
}
func (h *Hub) DeleteDevice(name string) {
h.mu.Lock()
_, ok := h.records[name]
delete(h.records, name)
delete(h.live, name)
h.mu.Unlock()
if ok {
h.saveYAML()
h.broadcast(BrowserMsg{Type: "remove", Machine: name})
logInfo("delete", "%s", name)
}
}
func (h *Hub) AddMonitor(c *websocket.Conn) {
h.mu.Lock()
h.monitors[c] = struct{}{}
h.mu.Unlock()
c.WriteJSON(BrowserMsg{Type: "snapshot", Devices: h.snapshot()})
logInfo("monitor+", "浏览器连接 (%d 在线)", len(h.monitors))
}
func (h *Hub) RemoveMonitor(c *websocket.Conn) {
h.mu.Lock()
delete(h.monitors, c)
h.mu.Unlock()
c.Close()
logInfo("monitor-", "浏览器断开 (%d 在线)", len(h.monitors))
}
func (h *Hub) snapshot() map[string]*BrowserDevice {
h.mu.RLock()
defer h.mu.RUnlock()
out := make(map[string]*BrowserDevice, len(h.records))
for name, rec := range h.records {
bd := &BrowserDevice{Info: rec, Online: false}
if live, ok := h.live[name]; ok {
bd.Live = live
bd.Online = true
}
if md, ok := h.miningData[name]; ok {
bd.Mining = md
if hs, ok := h.hashHist[name]; ok {
bd.History = make([]HashratePoint, len(hs))
copy(bd.History, hs)
}
}
out[name] = bd
}
return out
}
func (h *Hub) broadcast(msg BrowserMsg) {
data, _ := json.Marshal(msg)
h.mu.RLock()
defer h.mu.RUnlock()
for c := range h.monitors {
if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
go h.RemoveMonitor(c)
}
}
}
func (h *Hub) CleanStale() {
tick := time.NewTicker(10 * time.Second)
defer tick.Stop()
for range tick.C {
h.mu.RLock()
var stale []string
for name, rec := range h.records {
if _, online := h.live[name]; online && time.Since(rec.LastSeen) > 20*time.Second {
stale = append(stale, name)
}
}
h.mu.RUnlock()
for _, name := range stale {
h.MarkOffline(name)
logInfo("offline", "%s", name)
}
}
}
// ═══════════════════════════════════════════════
// 矿池拉取
// ═══════════════════════════════════════════════
func (h *Hub) StartMinerFetcher() {
if h.config.MinerUID == "" {
logInfo("miner", "未配置 miner_uid,跳过")
return
}
url := h.config.MinerAPIBase + "/" + h.config.MinerUID
interval := time.Duration(h.config.FetchInterval) * time.Second
logInfo("miner", "启动 (每 %v)", interval)
h.fetchMinerData(url)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
h.mu.RLock()
n := len(h.monitors)
h.mu.RUnlock()
if n > 0 {
h.fetchMinerData(url)
}
}
}
func (h *Hub) fetchMinerData(url string) {
client := &http.Client{Timeout: 15 * time.Second}
resp, err := client.Get(url)
if err != nil {
logInfo("miner", "请求失败: %v", err)
return
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
logInfo("miner", "HTTP %d", resp.StatusCode)
return
}
var apiResp struct {
Results []struct {
Worker string `json:"worker"`
Status string `json:"status"`
Coin string `json:"coin"`
Valid int `json:"valid"`
Stale int `json:"stale"`
Invalid int `json:"invalid"`
AvgHashrate24h string `json:"avg_hashrate_24h"`
AvgHashrate3h string `json:"avg_hashrate_3h"`
AvgHashrate30m string `json:"avg_hashrate_30m"`
Agent string `json:"agent"`
Country string `json:"country"`
LastActive int64 `json:"last_active"`
OpenedAt int64 `json:"opened_at"` // ★ 新增
} `json:"results"`
}
if json.NewDecoder(resp.Body).Decode(&apiResp) != nil {
return
}
now := time.Now()
h.mu.Lock()
for _, w := range apiResp.Results {
hr30m, _ := strconv.ParseFloat(w.AvgHashrate30m, 64)
hr3h, _ := strconv.ParseFloat(w.AvgHashrate3h, 64)
hr24h, _ := strconv.ParseFloat(w.AvgHashrate24h, 64)
h.miningData[w.Worker] = &MiningData{
Status: w.Status, Coin: w.Coin, Valid: w.Valid, Stale: w.Stale, Invalid: w.Invalid,
Hashrate30m: hr30m, Hashrate3h: hr3h, Hashrate24h: hr24h,
Agent: w.Agent, Country: w.Country,
LastActive: w.LastActive, OpenedAt: w.OpenedAt, // ★ 加 OpenedAt
FetchedAt: now,
}
h.hashHist[w.Worker] = append(h.hashHist[w.Worker], HashratePoint{T: now.Format(time.RFC3339), V: hr30m})
if len(h.hashHist[w.Worker]) > h.config.HistoryMax {
h.hashHist[w.Worker] = h.hashHist[w.Worker][len(h.hashHist[w.Worker])-h.config.HistoryMax:]
}
}
h.mu.Unlock()
for _, w := range apiResp.Results {
h.mu.RLock()
md := h.miningData[w.Worker]
hist := make([]HashratePoint, len(h.hashHist[w.Worker]))
copy(hist, h.hashHist[w.Worker])
h.mu.RUnlock()
h.broadcast(BrowserMsg{Type: "mining", Machine: w.Worker, Mining: md, History: hist})
}
logInfo("miner", "更新 %d 个 worker", len(apiResp.Results))
}
// ═══════════════════════════════════════════════
// WebSocket
// ═══════════════════════════════════════════════
var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
ReadBufferSize: 65536,
WriteBufferSize: 262144,
}
func handleAgent(conn *websocket.Conn, hub *Hub) {
defer conn.Close()
addr := conn.RemoteAddr().String()
logInfo("agent+", "%s", addr)
defer logInfo("agent-", "%s", addr)
for {
_, raw, err := conn.ReadMessage()
if err != nil {
break
}
hub.UpdateDevice(raw)
}
}
func handleMonitor(conn *websocket.Conn, hub *Hub) {
addr := conn.RemoteAddr().String()
defer conn.Close()
// ① 要求认证
conn.WriteJSON(BrowserMsg{Type: "auth_required"})
// ② 等待认证 (15 秒超时)
conn.SetReadDeadline(time.Now().Add(15 * time.Second))
var authOK bool
for {
_, raw, err := conn.ReadMessage()
if err != nil {
logInfo("monitor", "认证超时: %s", addr)
return
}
var cmd BrowserCmd
if json.Unmarshal(raw, &cmd) != nil {
continue
}
if cmd.Type != "auth" {
continue
}
// 校验
if cmd.MinerID == hub.config.Auth.MinerID && cmd.Password == hub.config.Auth.Password {
authOK = true
break
}
logInfo("monitor", "认证失败: %s", addr)
conn.WriteJSON(BrowserMsg{Type: "auth_fail", Msg: "设备ID或密码错误"})
}
if !authOK {
return
}
// ③ 认证通过
conn.SetReadDeadline(time.Time{})
conn.WriteJSON(BrowserMsg{Type: "auth_ok"})
logInfo("monitor+", "认证通过: %s", addr)
// ④ 进入正常数据流
hub.AddMonitor(conn)
defer hub.RemoveMonitor(conn)
for {
_, raw, err := conn.ReadMessage()
if err != nil {
break
}
var cmd BrowserCmd
if json.Unmarshal(raw, &cmd) == nil && cmd.Action == "delete" && cmd.Machine != "" {
hub.DeleteDevice(cmd.Machine)
}
}
}
// ═══════════════════════════════════════════════
// HTML
// ═══════════════════════════════════════════════
//go:embed templates/index.html
var indexHTML string
// ═══════════════════════════════════════════════
// main
// ═══════════════════════════════════════════════
func main() {
showVersion := flag.Bool("version", false, "显示版本信息")
configFlag := flag.String("config", "", "配置文件路径 (默认: 当前目录 server.yaml)")
flag.Parse()
if *showVersion {
fmt.Printf("Hardware Monitor Server %s\n", Version)
fmt.Printf("Build: %s / %s\n", BuildTime, runtime.GOOS+"/"+runtime.GOARCH)
return
}
base := baseDir() // ★ 改这里
cfgPath := *configFlag
if cfgPath == "" {
cfgPath = filepath.Join(base, "server.yaml")
}
cfg, cfgAbs := loadOrCreateConfig(cfgPath)
devicesAbs := resolvePath(filepath.Dir(cfgAbs), cfg.DevicesFile)
os.MkdirAll(filepath.Dir(devicesAbs), 0755)
hub := NewHub(cfg, devicesAbs)
go hub.CleanStale()
go hub.StartMinerFetcher()
mux := http.NewServeMux()
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/" {
http.NotFound(w, r)
return
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
fmt.Fprint(w, indexHTML)
})
mux.HandleFunc("/ws/agent", func(w http.ResponseWriter, r *http.Request) {
if c, err := upgrader.Upgrade(w, r, nil); err == nil {
go handleAgent(c, hub)
}
})
mux.HandleFunc("/ws/monitor", func(w http.ResponseWriter, r *http.Request) {
if c, err := upgrader.Upgrade(w, r, nil); err == nil {
go handleMonitor(c, hub)
}
})
fmt.Println()
fmt.Println(" ╔═══════════════════════════════════════════╗")
fmt.Println(" ║ Hardware Monitor Server ║")
fmt.Printf(" ║ Version %-28s ║\n", Version)
fmt.Println(" ╠═══════════════════════════════════════════╣")
fmt.Printf(" ║ Dashboard : http://localhost%-13s ║\n", cfg.Listen)
fmt.Printf(" ║ Agent WS : ws://localhost%-12s/ws ║\n", cfg.Listen)
if cfgAbs != "" {
fmt.Printf(" ║ Config : %-29s║\n", trunc(cfgAbs, 29))
}
fmt.Printf(" ║ Devices : %-29s║\n", trunc(devicesAbs, 29))
if cfg.MinerUID != "" {
uid := cfg.MinerUID
if len(uid) > 22 {
uid = uid[:22] + "..."
}
fmt.Printf(" ║ Miner : %-29s║\n", uid)
} else {
fmt.Println(" ║ Miner : 未配置 ║")
}
fmt.Println(" ╚═══════════════════════════════════════════╝")
fmt.Println()
log.Fatal(http.ListenAndServe(cfg.Listen, mux))
}
func trunc(s string, n int) string {
if len(s) <= n {
return s
}
return "..." + s[len(s)-n+3:]
}