From 05e75549d20e8133089c7e24eda34e08a4bd2023 Mon Sep 17 00:00:00 2001 From: ryan Date: Tue, 10 Mar 2026 15:33:06 +0800 Subject: [PATCH] feat: Implement node management features including creation, updating, and deletion - Added CreateNode, UpdateNode, and DeleteNode functions in the controller for managing nodes. - Introduced NodeInput struct for input validation during node creation and updates. - Enhanced the Node model to include DiscoveryToken and Pending status. - Updated the AgentRegister and AgentHeartbeat functions to utilize the new node management logic. - Refactored the API router to include new routes for node management. - Improved the frontend Node component to support node creation, editing, and deletion with appropriate UI feedback. - Added tests to ensure the new functionality works as expected. --- atsf_agent/internal/agent/runner.go | 84 +++++-- atsf_agent/internal/agent/runner_test.go | 92 +++++++- atsf_agent/internal/config/config.go | 63 +++++- atsf_agent/internal/heartbeat/service.go | 9 +- atsf_agent/internal/httpclient/client.go | 17 +- atsf_agent/internal/protocol/agent_api.go | 6 + atsf_server/controller/agent.go | 27 ++- atsf_server/controller/node.go | 87 +++++++ atsf_server/middleware/agent-auth.go | 31 ++- atsf_server/model/node.go | 122 +++++----- atsf_server/router/api-router.go | 20 +- atsf_server/router/api_phase1_test.go | 15 +- atsf_server/router/api_phase2_test.go | 74 ++++-- atsf_server/service/agent.go | 154 ++++++------- atsf_server/service/node.go | 148 ++++++++++++ atsf_server/web/src/pages/Node/index.js | 263 +++++++++++++++------- 16 files changed, 916 insertions(+), 296 deletions(-) create mode 100644 atsf_server/controller/node.go create mode 100644 atsf_server/service/node.go diff --git a/atsf_agent/internal/agent/runner.go b/atsf_agent/internal/agent/runner.go index ddbdddf6..37d01fdd 100644 --- a/atsf_agent/internal/agent/runner.go +++ b/atsf_agent/internal/agent/runner.go @@ -2,7 +2,9 @@ package agent import ( "context" + "errors" "log" + "strings" "time" "atsflare-agent/internal/config" @@ -11,8 +13,9 @@ import ( ) type HeartbeatService interface { - Register(ctx context.Context, payload protocol.NodePayload) error + Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) Heartbeat(ctx context.Context, payload protocol.NodePayload) error + SetToken(token string) } type SyncService interface { @@ -33,21 +36,20 @@ func (r *Runner) Run(ctx context.Context) error { return err } log.Printf("agent runner started: node_id=%s node=%s ip=%s", nodeID, r.Config.NodeName, r.Config.NodeIP) - if err = r.HeartbeatService.Register(ctx, r.nodePayload(nodeID)); err != nil { - log.Printf("agent register failed: %v", err) - } else { - log.Printf("agent register succeeded: node_id=%s", nodeID) - } - if err = r.SyncService.SyncOnStartup(ctx); err != nil { - r.recordSyncError(err) - log.Printf("agent startup sync failed: %v", err) - } else { - log.Printf("agent startup sync completed") - } - if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil { - log.Printf("agent startup heartbeat failed: %v", err) - } else { - log.Printf("agent startup heartbeat succeeded: node_id=%s", nodeID) + if r.hasAgentToken() { + if err = r.SyncService.SyncOnStartup(ctx); err != nil { + r.recordSyncError(err) + log.Printf("agent startup sync failed: %v", err) + } else { + log.Printf("agent startup sync completed") + } + if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil { + log.Printf("agent startup heartbeat failed: %v", err) + } else { + log.Printf("agent startup heartbeat succeeded: node_id=%s", nodeID) + } + } else if err = r.tryRegister(ctx, &nodeID); err != nil { + log.Printf("agent initial discovery register failed: %v", err) } heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval) @@ -61,10 +63,19 @@ func (r *Runner) Run(ctx context.Context) error { log.Printf("agent runner shutting down: %v", ctx.Err()) return ctx.Err() case <-heartbeatTicker.C: + if !r.hasAgentToken() { + if err = r.tryRegister(ctx, &nodeID); err != nil { + log.Printf("agent discovery register failed: %v", err) + } + continue + } if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil { log.Printf("agent heartbeat failed: %v", err) } case <-syncTicker.C: + if !r.hasAgentToken() { + continue + } log.Printf("agent sync tick: node_id=%s", nodeID) if err = r.SyncService.SyncOnce(ctx); err != nil { r.recordSyncError(err) @@ -76,6 +87,47 @@ func (r *Runner) Run(ctx context.Context) error { } } +func (r *Runner) hasAgentToken() bool { + return strings.TrimSpace(r.Config.AgentToken) != "" +} + +func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error { + if strings.TrimSpace(r.Config.DiscoveryToken) == "" { + return errors.New("agent_token 为空且未配置 discovery_token") + } + log.Printf("agent discovery registration started") + response, err := r.HeartbeatService.Register(ctx, r.nodePayload(*nodeID)) + if err != nil { + return err + } + if response == nil || strings.TrimSpace(response.AgentToken) == "" || strings.TrimSpace(response.NodeID) == "" { + return errors.New("discovery register response 缺少 node_id 或 agent_token") + } + snapshot, err := r.StateStore.Load() + if err != nil { + return err + } + snapshot.NodeID = response.NodeID + if err = r.StateStore.Save(snapshot); err != nil { + return err + } + r.Config.AgentToken = response.AgentToken + r.Config.DiscoveryToken = "" + if err = r.Config.Save(); err != nil { + return err + } + r.HeartbeatService.SetToken(response.AgentToken) + *nodeID = response.NodeID + log.Printf("agent discovery registration succeeded: node_id=%s", response.NodeID) + if err = r.SyncService.SyncOnStartup(ctx); err != nil { + r.recordSyncError(err) + log.Printf("agent post-register startup sync failed: %v", err) + } else { + log.Printf("agent post-register startup sync completed") + } + return nil +} + func (r *Runner) recordSyncError(err error) { if err == nil || r.StateStore == nil { return diff --git a/atsf_agent/internal/agent/runner_test.go b/atsf_agent/internal/agent/runner_test.go index 0dc3a293..6e66e775 100644 --- a/atsf_agent/internal/agent/runner_test.go +++ b/atsf_agent/internal/agent/runner_test.go @@ -3,6 +3,7 @@ package agent import ( "context" "errors" + "os" "path/filepath" "sync" "testing" @@ -18,15 +19,17 @@ type fakeHeartbeatService struct { registerCalls int heartbeatCalls int registerErr error + registerResp *protocol.RegisterNodeResponse heartbeatErrs []error onHeartbeat func(int) + lastToken string } -func (f *fakeHeartbeatService) Register(ctx context.Context, payload protocol.NodePayload) error { +func (f *fakeHeartbeatService) Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) { f.mu.Lock() defer f.mu.Unlock() f.registerCalls++ - return f.registerErr + return f.registerResp, f.registerErr } func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.NodePayload) error { @@ -45,6 +48,12 @@ func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.N return err } +func (f *fakeHeartbeatService) SetToken(token string) { + f.mu.Lock() + defer f.mu.Unlock() + f.lastToken = token +} + type fakeSyncService struct { mu sync.Mutex startupErr error @@ -90,6 +99,7 @@ func TestRunnerKeepsHeartbeatWhenStartupSyncFails(t *testing.T) { } runner := &Runner{ Config: &config.Config{ + AgentToken: "agent-token", NodeName: "edge-01", NodeIP: "10.0.0.8", AgentVersion: "0.1.0", @@ -106,8 +116,8 @@ func TestRunnerKeepsHeartbeatWhenStartupSyncFails(t *testing.T) { if !errors.Is(err, context.Canceled) { t.Fatalf("expected context cancellation, got %v", err) } - if heartbeatService.registerCalls != 1 { - t.Fatalf("expected 1 register call, got %d", heartbeatService.registerCalls) + if heartbeatService.registerCalls != 0 { + t.Fatalf("expected no discovery register call, got %d", heartbeatService.registerCalls) } if heartbeatService.heartbeatCalls < 2 { t.Fatalf("expected heartbeat loop to continue, got %d heartbeat calls", heartbeatService.heartbeatCalls) @@ -140,6 +150,7 @@ func TestRunnerDoesNotExitOnHeartbeatOrSyncError(t *testing.T) { } runner := &Runner{ Config: &config.Config{ + AgentToken: "agent-token", NodeName: "edge-01", NodeIP: "10.0.0.8", AgentVersion: "0.1.0", @@ -156,8 +167,8 @@ func TestRunnerDoesNotExitOnHeartbeatOrSyncError(t *testing.T) { if !errors.Is(err, context.Canceled) { t.Fatalf("expected context cancellation, got %v", err) } - if heartbeatService.registerCalls != 1 { - t.Fatalf("expected register attempt, got %d", heartbeatService.registerCalls) + if heartbeatService.registerCalls != 0 { + t.Fatalf("expected no register attempt, got %d", heartbeatService.registerCalls) } if syncService.syncOnceCalls == 0 { t.Fatal("expected sync loop to continue after heartbeat/register errors") @@ -170,3 +181,72 @@ func TestRunnerDoesNotExitOnHeartbeatOrSyncError(t *testing.T) { t.Fatalf("expected sync error to be recorded, got %q", snapshot.LastError) } } + +func TestRunnerDiscoveryRegisterUpdatesTokenAndNodeID(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json")) + heartbeatService := &fakeHeartbeatService{ + registerResp: &protocol.RegisterNodeResponse{ + NodeID: "node-server-assigned", + AgentToken: "agent-token-issued", + Name: "edge-01", + }, + onHeartbeat: func(callCount int) { + if callCount >= 1 { + cancel() + } + }, + } + syncService := &fakeSyncService{} + configPath := filepath.Join(t.TempDir(), "agent.json") + if err := os.WriteFile(configPath, []byte(`{"server_url":"http://127.0.0.1:3000","discovery_token":"discovery-token","node_name":"edge-01","node_ip":"10.0.0.8"}`), 0o644); err != nil { + t.Fatalf("failed to seed config file: %v", err) + } + cfg, err := config.Load(configPath) + if err != nil { + t.Fatalf("failed to load config: %v", err) + } + runner := &Runner{ + Config: &config.Config{ + ServerURL: cfg.ServerURL, + DiscoveryToken: cfg.DiscoveryToken, + NodeName: cfg.NodeName, + NodeIP: cfg.NodeIP, + AgentVersion: "0.1.0", + NginxVersion: "1.25.5", + HeartbeatInterval: 10 * time.Millisecond, + SyncInterval: 20 * time.Millisecond, + }, + StateStore: stateStore, + HeartbeatService: heartbeatService, + SyncService: syncService, + } + runner.Config = cfg + runner.Config.AgentVersion = "0.1.0" + runner.Config.NginxVersion = "1.25.5" + runner.Config.HeartbeatInterval = 10 * time.Millisecond + runner.Config.SyncInterval = 20 * time.Millisecond + + err = runner.Run(ctx) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context cancellation, got %v", err) + } + if heartbeatService.registerCalls == 0 { + t.Fatal("expected discovery register to be attempted") + } + if heartbeatService.lastToken != "agent-token-issued" { + t.Fatalf("expected client token to be updated, got %q", heartbeatService.lastToken) + } + snapshot, loadErr := stateStore.Load() + if loadErr != nil { + t.Fatalf("failed to load state: %v", loadErr) + } + if snapshot.NodeID != "node-server-assigned" { + t.Fatalf("expected node id to be replaced, got %q", snapshot.NodeID) + } + if runner.Config.AgentToken != "agent-token-issued" || runner.Config.DiscoveryToken != "" { + t.Fatal("expected config token rotation to complete") + } +} diff --git a/atsf_agent/internal/config/config.go b/atsf_agent/internal/config/config.go index e192e6b1..e7505658 100644 --- a/atsf_agent/internal/config/config.go +++ b/atsf_agent/internal/config/config.go @@ -3,6 +3,7 @@ package config import ( "encoding/json" "errors" + "net" "os" pathpkg "path" "path/filepath" @@ -20,6 +21,7 @@ const ( type Config struct { ServerURL string `json:"server_url"` AgentToken string `json:"agent_token"` + DiscoveryToken string `json:"discovery_token"` NodeName string `json:"node_name"` NodeIP string `json:"node_ip"` AgentVersion string `json:"agent_version"` @@ -36,6 +38,7 @@ type Config struct { HeartbeatInterval time.Duration `json:"heartbeat_interval"` SyncInterval time.Duration `json:"sync_interval"` RequestTimeout time.Duration `json:"request_timeout"` + configPath string } func Load(path string) (*Config, error) { @@ -47,6 +50,7 @@ func Load(path string) (*Config, error) { if err = json.Unmarshal(data, cfg); err != nil { return nil, err } + cfg.configPath = path applyDefaults(cfg, filepath.Dir(path)) if err = validate(cfg); err != nil { return nil, err @@ -71,6 +75,12 @@ func applyDefaults(cfg *Config, baseDir string) { if cfg.DataDir == "" { cfg.DataDir = filepath.Join(baseDir, "data") } + if cfg.NodeName == "" { + cfg.NodeName = detectHostname() + } + if cfg.NodeIP == "" { + cfg.NodeIP = detectNodeIP() + } if cfg.NginxPath == "" { cfg.RouteConfigPath = joinManagedPath(cfg.DataDir, defaultDockerRouteConfigRelativePath) cfg.StatePath = joinManagedPath(cfg.DataDir, defaultDockerStateRelativePath) @@ -137,8 +147,8 @@ func validate(cfg *Config) error { if cfg.ServerURL == "" { return errors.New("server_url 不能为空") } - if cfg.AgentToken == "" { - return errors.New("agent_token 不能为空") + if strings.TrimSpace(cfg.AgentToken) == "" && strings.TrimSpace(cfg.DiscoveryToken) == "" { + return errors.New("agent_token 和 discovery_token 不能同时为空") } if cfg.NodeName == "" { return errors.New("node_name 不能为空") @@ -148,3 +158,52 @@ func validate(cfg *Config) error { } return nil } + +func (cfg *Config) Save() error { + if cfg == nil { + return errors.New("config 不能为空") + } + if cfg.configPath == "" { + return errors.New("config path 未初始化") + } + data, err := json.MarshalIndent(cfg, "", " ") + if err != nil { + return err + } + return os.WriteFile(cfg.configPath, data, 0o644) +} + +func detectHostname() string { + host, err := os.Hostname() + if err != nil { + return "" + } + return strings.TrimSpace(host) +} + +func detectNodeIP() string { + interfaces, err := net.Interfaces() + if err != nil { + return "" + } + for _, iface := range interfaces { + if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagLoopback != 0 { + continue + } + addrs, err := iface.Addrs() + if err != nil { + continue + } + for _, addr := range addrs { + ipNet, ok := addr.(*net.IPNet) + if !ok || ipNet.IP == nil || ipNet.IP.IsLoopback() { + continue + } + ipv4 := ipNet.IP.To4() + if ipv4 != nil { + return ipv4.String() + } + } + } + return "" +} diff --git a/atsf_agent/internal/heartbeat/service.go b/atsf_agent/internal/heartbeat/service.go index 18c87602..faf91828 100644 --- a/atsf_agent/internal/heartbeat/service.go +++ b/atsf_agent/internal/heartbeat/service.go @@ -7,8 +7,9 @@ import ( ) type Client interface { - RegisterNode(ctx context.Context, payload protocol.NodePayload) error + RegisterNode(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) Heartbeat(ctx context.Context, payload protocol.NodePayload) error + SetToken(token string) } type Service struct { @@ -19,10 +20,14 @@ func New(client Client) *Service { return &Service{client: client} } -func (s *Service) Register(ctx context.Context, payload protocol.NodePayload) error { +func (s *Service) Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) { return s.client.RegisterNode(ctx, payload) } func (s *Service) Heartbeat(ctx context.Context, payload protocol.NodePayload) error { return s.client.Heartbeat(ctx, payload) } + +func (s *Service) SetToken(token string) { + s.client.SetToken(token) +} diff --git a/atsf_agent/internal/httpclient/client.go b/atsf_agent/internal/httpclient/client.go index 99a183dd..2f842fa4 100644 --- a/atsf_agent/internal/httpclient/client.go +++ b/atsf_agent/internal/httpclient/client.go @@ -29,9 +29,17 @@ func New(baseURL string, token string, timeout time.Duration) *Client { } } -func (c *Client) RegisterNode(ctx context.Context, payload protocol.NodePayload) error { +func (c *Client) RegisterNode(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) { log.Printf("http register node request: node_id=%s current_version=%s", payload.NodeID, payload.CurrentVersion) - return c.postJSON(ctx, "/api/agent/nodes/register", payload, nil) + resp := protocol.APIResponse[protocol.RegisterNodeResponse]{} + if err := c.postJSON(ctx, "/api/agent/nodes/register", payload, &resp); err != nil { + return nil, err + } + if !resp.Success { + return nil, errors.New(resp.Message) + } + log.Printf("http register node response: node_id=%s", resp.Data.NodeID) + return &resp.Data, nil } func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) error { @@ -56,6 +64,11 @@ func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPa return c.postJSON(ctx, "/api/agent/apply-logs", payload, nil) } +func (c *Client) SetToken(token string) { + c.token = strings.TrimSpace(token) + log.Printf("http client token updated") +} + func (c *Client) getJSON(ctx context.Context, path string, target any) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil) if err != nil { diff --git a/atsf_agent/internal/protocol/agent_api.go b/atsf_agent/internal/protocol/agent_api.go index e40390d2..47db8804 100644 --- a/atsf_agent/internal/protocol/agent_api.go +++ b/atsf_agent/internal/protocol/agent_api.go @@ -16,6 +16,12 @@ type NodePayload struct { LastError string `json:"last_error"` } +type RegisterNodeResponse struct { + NodeID string `json:"node_id"` + AgentToken string `json:"agent_token"` + Name string `json:"name"` +} + type ApplyLogPayload struct { NodeID string `json:"node_id"` Version string `json:"version"` diff --git a/atsf_server/controller/agent.go b/atsf_server/controller/agent.go index 4863711f..d1d685cf 100644 --- a/atsf_server/controller/agent.go +++ b/atsf_server/controller/agent.go @@ -2,6 +2,7 @@ package controller import ( "encoding/json" + "gin-template/model" "gin-template/service" "github.com/gin-gonic/gin" "net/http" @@ -16,7 +17,15 @@ func AgentRegister(c *gin.Context) { }) return } - node, err := service.RegisterNode(payload) + discoveryNode, ok := c.Get("discovery_node") + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "无权进行此操作，Discovery Token 无效", + }) + return + } + result, err := service.RegisterNode(discoveryNode.(*model.Node), payload) if err != nil { c.JSON(http.StatusOK, gin.H{ "success": false, @@ -27,7 +36,7 @@ func AgentRegister(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ "success": true, "message": "", - "data": node, + "data": result, }) } @@ -40,7 +49,15 @@ func AgentHeartbeat(c *gin.Context) { }) return } - node, err := service.HeartbeatNode(payload) + authNode, ok := c.Get("agent_node") + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "无权进行此操作，Agent Token 无效", + }) + return + } + node, err := service.HeartbeatNode(authNode.(*model.Node), payload) if err != nil { c.JSON(http.StatusOK, gin.H{ "success": false, @@ -80,6 +97,10 @@ func AgentReportApplyLog(c *gin.Context) { }) return } + authNode, ok := c.Get("agent_node") + if ok { + payload.NodeID = authNode.(*model.Node).NodeID + } log, err := service.ReportApplyLog(payload) if err != nil { c.JSON(http.StatusOK, gin.H{ diff --git a/atsf_server/controller/node.go b/atsf_server/controller/node.go new file mode 100644 index 00000000..a8dbcc5f --- /dev/null +++ b/atsf_server/controller/node.go @@ -0,0 +1,87 @@ +package controller + +import ( + "encoding/json" + "gin-template/service" + "github.com/gin-gonic/gin" + "net/http" + "strconv" +) + +func CreateNode(c *gin.Context) { + var input service.NodeInput + if err := json.NewDecoder(c.Request.Body).Decode(&input); err != nil { + c.JSON(http.StatusBadRequest, gin.H{ + "success": false, + "message": "无效的参数", + }) + return + } + node, err := service.CreateNode(input) + if err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": err.Error(), + }) + return + } + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": node, + }) +} + +func UpdateNode(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil || id == 0 { + c.JSON(http.StatusBadRequest, gin.H{ + "success": false, + "message": "无效的参数", + }) + return + } + var input service.NodeInput + if err = json.NewDecoder(c.Request.Body).Decode(&input); err != nil { + c.JSON(http.StatusBadRequest, gin.H{ + "success": false, + "message": "无效的参数", + }) + return + } + node, err := service.UpdateNode(uint(id), input) + if err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": err.Error(), + }) + return + } + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + "data": node, + }) +} + +func DeleteNode(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil || id == 0 { + c.JSON(http.StatusBadRequest, gin.H{ + "success": false, + "message": "无效的参数", + }) + return + } + if err = service.DeleteNode(uint(id)); err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": err.Error(), + }) + return + } + c.JSON(http.StatusOK, gin.H{ + "success": true, + "message": "", + }) +} diff --git a/atsf_server/middleware/agent-auth.go b/atsf_server/middleware/agent-auth.go index 4cbfd15c..416f853a 100644 --- a/atsf_server/middleware/agent-auth.go +++ b/atsf_server/middleware/agent-auth.go @@ -1,7 +1,7 @@ package middleware import ( - "gin-template/common" + "gin-template/service" "github.com/gin-gonic/gin" "net/http" ) @@ -9,15 +9,8 @@ import ( func AgentAuth() func(c *gin.Context) { return func(c *gin.Context) { token := c.GetHeader("X-Agent-Token") - if common.AgentToken == "" { - c.JSON(http.StatusUnauthorized, gin.H{ - "success": false, - "message": "Agent Token 未配置", - }) - c.Abort() - return - } - if token == "" || token != common.AgentToken { + node, err := service.AuthenticateAgentToken(token) + if err != nil { c.JSON(http.StatusUnauthorized, gin.H{ "success": false, "message": "无权进行此操作，Agent Token 无效", @@ -25,6 +18,24 @@ func AgentAuth() func(c *gin.Context) { c.Abort() return } + c.Set("agent_node", node) + c.Next() + } +} + +func AgentDiscoveryAuth() func(c *gin.Context) { + return func(c *gin.Context) { + token := c.GetHeader("X-Agent-Token") + node, err := service.AuthenticateDiscoveryToken(token) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{ + "success": false, + "message": "无权进行此操作，Discovery Token 无效", + }) + c.Abort() + return + } + c.Set("discovery_node", node) c.Next() } } diff --git a/atsf_server/model/node.go b/atsf_server/model/node.go index 207f8b72..43f1d22c 100644 --- a/atsf_server/model/node.go +++ b/atsf_server/model/node.go @@ -1,61 +1,61 @@ -package model - -import "time" - -type Node struct { - ID uint `json:"id" gorm:"primaryKey"` - NodeID string `json:"node_id" gorm:"uniqueIndex;size:64;not null"` - Name string `json:"name" gorm:"size:128;not null"` - IP string `json:"ip" gorm:"size:64;not null"` - AgentToken string `json:"-" gorm:"size:128;uniqueIndex"` - DiscoveryToken string `json:"-" gorm:"size:128;uniqueIndex"` - AgentVersion string `json:"agent_version" gorm:"size:64;not null"` - NginxVersion string `json:"nginx_version" gorm:"size:64"` - Status string `json:"status" gorm:"size:16;not null;default:'offline'"` - CurrentVersion string `json:"current_version" gorm:"size:32"` - LastSeenAt time.Time `json:"last_seen_at"` - LastError string `json:"last_error" gorm:"size:1024"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` -} - -func ListNodes() (nodes []*Node, err error) { - err = DB.Order("id desc").Find(&nodes).Error - return nodes, err -} - -func GetNodeByNodeID(nodeID string) (*Node, error) { - node := &Node{} - err := DB.Where("node_id = ?", nodeID).First(node).Error - return node, err -} - -func GetNodeByID(id uint) (*Node, error) { - node := &Node{} - err := DB.First(node, id).Error - return node, err -} - -func GetNodeByAgentToken(token string) (*Node, error) { - node := &Node{} - err := DB.Where("agent_token = ?", token).First(node).Error - return node, err -} - -func GetNodeByDiscoveryToken(token string) (*Node, error) { - node := &Node{} - err := DB.Where("discovery_token = ?", token).First(node).Error - return node, err -} - -func (node *Node) Insert() error { - return DB.Create(node).Error -} - -func (node *Node) Update() error { - return DB.Save(node).Error -} - -func (node *Node) Delete() error { - return DB.Delete(node).Error -} +package model + +import "time" + +type Node struct { + ID uint `json:"id" gorm:"primaryKey"` + NodeID string `json:"node_id" gorm:"uniqueIndex;size:64;not null"` + Name string `json:"name" gorm:"size:128;not null"` + IP string `json:"ip" gorm:"size:64;not null"` + AgentToken string `json:"-" gorm:"size:128;index"` + DiscoveryToken string `json:"-" gorm:"size:128;index"` + AgentVersion string `json:"agent_version" gorm:"size:64;not null"` + NginxVersion string `json:"nginx_version" gorm:"size:64"` + Status string `json:"status" gorm:"size:16;not null;default:'offline'"` + CurrentVersion string `json:"current_version" gorm:"size:32"` + LastSeenAt time.Time `json:"last_seen_at"` + LastError string `json:"last_error" gorm:"size:1024"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +func ListNodes() (nodes []*Node, err error) { + err = DB.Order("id desc").Find(&nodes).Error + return nodes, err +} + +func GetNodeByNodeID(nodeID string) (*Node, error) { + node := &Node{} + err := DB.Where("node_id = ?", nodeID).First(node).Error + return node, err +} + +func GetNodeByID(id uint) (*Node, error) { + node := &Node{} + err := DB.First(node, id).Error + return node, err +} + +func GetNodeByAgentToken(token string) (*Node, error) { + node := &Node{} + err := DB.Where("agent_token = ?", token).First(node).Error + return node, err +} + +func GetNodeByDiscoveryToken(token string) (*Node, error) { + node := &Node{} + err := DB.Where("discovery_token = ?", token).First(node).Error + return node, err +} + +func (node *Node) Insert() error { + return DB.Create(node).Error +} + +func (node *Node) Update() error { + return DB.Save(node).Error +} + +func (node *Node) Delete() error { + return DB.Delete(node).Error +} diff --git a/atsf_server/router/api-router.go b/atsf_server/router/api-router.go index 1e65b615..8c2c4263 100644 --- a/atsf_server/router/api-router.go +++ b/atsf_server/router/api-router.go @@ -99,6 +99,9 @@ func SetApiRouter(router *gin.Engine) { nodeRoute.Use(middleware.AdminAuth()) { nodeRoute.GET("/", controller.GetNodes) + nodeRoute.POST("/", controller.CreateNode) + nodeRoute.PUT("/:id", controller.UpdateNode) + nodeRoute.DELETE("/:id", controller.DeleteNode) } applyLogRoute := apiRouter.Group("/apply-logs") applyLogRoute.Use(middleware.AdminAuth()) @@ -106,12 +109,19 @@ func SetApiRouter(router *gin.Engine) { applyLogRoute.GET("/", controller.GetApplyLogs) } agentRoute := apiRouter.Group("/agent") - agentRoute.Use(middleware.AgentAuth()) { - agentRoute.POST("/nodes/register", controller.AgentRegister) - agentRoute.POST("/nodes/heartbeat", controller.AgentHeartbeat) - agentRoute.GET("/config-versions/active", controller.AgentGetActiveConfig) - agentRoute.POST("/apply-logs", controller.AgentReportApplyLog) + discoveryRoute := agentRoute.Group("/") + discoveryRoute.Use(middleware.AgentDiscoveryAuth()) + { + discoveryRoute.POST("/nodes/register", controller.AgentRegister) + } + authorizedRoute := agentRoute.Group("/") + authorizedRoute.Use(middleware.AgentAuth()) + { + authorizedRoute.POST("/nodes/heartbeat", controller.AgentHeartbeat) + authorizedRoute.GET("/config-versions/active", controller.AgentGetActiveConfig) + authorizedRoute.POST("/apply-logs", controller.AgentReportApplyLog) + } } } } diff --git a/atsf_server/router/api_phase1_test.go b/atsf_server/router/api_phase1_test.go index 90e9cec2..d0f9d6ba 100644 --- a/atsf_server/router/api_phase1_test.go +++ b/atsf_server/router/api_phase1_test.go @@ -11,6 +11,7 @@ import ( "gin-template/common" "gin-template/model" "gin-template/router" + "gin-template/service" "github.com/gin-contrib/sessions" "github.com/gin-contrib/sessions/cookie" "github.com/gin-gonic/gin" @@ -199,8 +200,20 @@ func TestPhase1HTTPSAndCertificateImportLifecycle(t *testing.T) { if !strings.Contains(version.SupportFilesJSON, ".crt") || !strings.Contains(version.SupportFilesJSON, ".key") { t.Fatal("expected support files json to contain certificate artifacts") } + if err := (&model.Node{ + NodeID: "phase1-node", + Name: "phase1-node", + IP: "10.0.0.8", + AgentToken: common.AgentToken, + AgentVersion: "0.1.0", + NginxVersion: "1.25.5", + Status: service.NodeStatusOnline, + LastSeenAt: time.Now(), + }).Insert(); err != nil { + t.Fatalf("failed to seed phase1 node: %v", err) + } - agentResp := performAgentJSONRequest(t, engine, http.MethodGet, "/api/agent/config-versions/active", nil) + agentResp := performAgentJSONRequestWithToken(t, engine, common.AgentToken, http.MethodGet, "/api/agent/config-versions/active", nil) var activeConfig map[string]any decodeResponseData(t, agentResp, &activeConfig) supportFiles, ok := activeConfig["support_files"].([]any) diff --git a/atsf_server/router/api_phase2_test.go b/atsf_server/router/api_phase2_test.go index 9d17f95c..f2657b62 100644 --- a/atsf_server/router/api_phase2_test.go +++ b/atsf_server/router/api_phase2_test.go @@ -19,7 +19,6 @@ import ( func TestPhase2AgentLifecycle(t *testing.T) { gin.SetMode(gin.TestMode) common.RedisEnabled = false - common.AgentToken = "phase2-agent-token" setupTestDB(t) engine := gin.New() @@ -34,11 +33,20 @@ func TestPhase2AgentLifecycle(t *testing.T) { unauthorizedRecorder := httptest.NewRecorder() engine.ServeHTTP(unauthorizedRecorder, unauthorizedRequest) if unauthorizedRecorder.Code != http.StatusUnauthorized { - t.Fatalf("expected unauthorized status for missing agent token, got %d", unauthorizedRecorder.Code) + t.Fatalf("expected unauthorized status for missing discovery token, got %d", unauthorizedRecorder.Code) + } + + createdNodeResp := performJSONRequest(t, engine, adminToken, http.MethodPost, "/api/nodes/", map[string]any{ + "name": "shanghai-edge-1", + }) + var createdNode service.NodeView + decodeResponseData(t, createdNodeResp, &createdNode) + if createdNode.DiscoveryToken == "" || !createdNode.Pending { + t.Fatal("expected created node to expose discovery token while pending") } nodePayload := map[string]any{ - "node_id": "node-001", + "node_id": "local-node-id", "name": "shanghai-edge-1", "ip": "10.0.0.8", "agent_version": "0.1.0", @@ -46,15 +54,15 @@ func TestPhase2AgentLifecycle(t *testing.T) { "current_version": "", "last_error": "", } - resp := performAgentJSONRequest(t, engine, http.MethodPost, "/api/agent/nodes/register", nodePayload) - var registeredNode model.Node - decodeResponseData(t, resp, ®isteredNode) - if registeredNode.NodeID != "node-001" || registeredNode.Status != service.NodeStatusOnline { - t.Fatal("expected node registration to persist online node state") + resp := performAgentJSONRequestWithToken(t, engine, createdNode.DiscoveryToken, http.MethodPost, "/api/agent/nodes/register", nodePayload) + var registration service.AgentRegistrationResponse + decodeResponseData(t, resp, ®istration) + if registration.NodeID != createdNode.NodeID || registration.AgentToken == "" { + t.Fatal("expected discovery registration to return assigned node_id and agent token") } heartbeatPayload := map[string]any{ - "node_id": "node-001", + "node_id": "spoofed-node-id", "name": "shanghai-edge-1", "ip": "10.0.0.9", "agent_version": "0.1.1", @@ -62,21 +70,22 @@ func TestPhase2AgentLifecycle(t *testing.T) { "current_version": "", "last_error": "", } - resp = performAgentJSONRequest(t, engine, http.MethodPost, "/api/agent/nodes/heartbeat", heartbeatPayload) + resp = performAgentJSONRequestWithToken(t, engine, registration.AgentToken, http.MethodPost, "/api/agent/nodes/heartbeat", heartbeatPayload) + var registeredNode model.Node decodeResponseData(t, resp, ®isteredNode) - if registeredNode.IP != "10.0.0.9" || registeredNode.AgentVersion != "0.1.1" { + if registeredNode.IP != "10.0.0.9" || registeredNode.AgentVersion != "0.1.1" || registeredNode.NodeID != createdNode.NodeID { t.Fatal("expected heartbeat to update node metadata") } - activeConfigResp := performAgentJSONRequest(t, engine, http.MethodGet, "/api/agent/config-versions/active", nil) + activeConfigResp := performAgentJSONRequestWithToken(t, engine, registration.AgentToken, http.MethodGet, "/api/agent/config-versions/active", nil) var activeConfig service.AgentConfigResponse decodeResponseData(t, activeConfigResp, &activeConfig) if activeConfig.Version == "" || activeConfig.RenderedConfig == "" || activeConfig.Checksum == "" { t.Fatal("expected active config response to contain version payload") } - successApplyResp := performAgentJSONRequest(t, engine, http.MethodPost, "/api/agent/apply-logs", map[string]any{ - "node_id": "node-001", + successApplyResp := performAgentJSONRequestWithToken(t, engine, registration.AgentToken, http.MethodPost, "/api/agent/apply-logs", map[string]any{ + "node_id": "spoofed-node-id", "version": activeConfig.Version, "result": service.ApplyResultOK, "message": "apply ok", @@ -87,8 +96,8 @@ func TestPhase2AgentLifecycle(t *testing.T) { t.Fatal("expected apply log success to be recorded") } - failedApplyResp := performAgentJSONRequest(t, engine, http.MethodPost, "/api/agent/apply-logs", map[string]any{ - "node_id": "node-001", + failedApplyResp := performAgentJSONRequestWithToken(t, engine, registration.AgentToken, http.MethodPost, "/api/agent/apply-logs", map[string]any{ + "node_id": "spoofed-node-id", "version": activeConfig.Version, "result": service.ApplyResultFailed, "message": "nginx reload failed", @@ -105,6 +114,9 @@ func TestPhase2AgentLifecycle(t *testing.T) { if len(nodes) != 1 { t.Fatalf("expected 1 node, got %d", len(nodes)) } + if nodes[0].Pending || nodes[0].DiscoveryToken != "" { + t.Fatal("expected registered node to clear pending discovery state") + } if nodes[0].LatestApplyResult != service.ApplyResultFailed || nodes[0].LatestApplyMessage != "nginx reload failed" { t.Fatal("expected node list to expose latest apply status") } @@ -115,15 +127,23 @@ func TestPhase2AgentLifecycle(t *testing.T) { t.Fatal("expected node last_error to reflect failed apply") } - logsResp := performJSONRequest(t, engine, adminToken, http.MethodGet, "/api/apply-logs/?node_id=node-001", nil) + logsResp := performJSONRequest(t, engine, adminToken, http.MethodGet, "/api/apply-logs/?node_id="+createdNode.NodeID, nil) var logs []model.ApplyLog decodeResponseData(t, logsResp, &logs) if len(logs) != 2 { t.Fatalf("expected 2 apply logs, got %d", len(logs)) } + updatedNodeResp := performJSONRequest(t, engine, adminToken, http.MethodPut, "/api/nodes/"+toString(createdNode.ID), map[string]any{ + "name": "shanghai-edge-1-renamed", + }) + decodeResponseData(t, updatedNodeResp, &createdNode) + if createdNode.Name != "shanghai-edge-1-renamed" { + t.Fatal("expected node name to be editable") + } + oldTime := time.Now().Add(-common.NodeOfflineThreshold - time.Minute) - if err := model.DB.Model(&model.Node{}).Where("node_id = ?", "node-001").Update("last_seen_at", oldTime).Error; err != nil { + if err := model.DB.Model(&model.Node{}).Where("node_id = ?", createdNode.NodeID).Update("last_seen_at", oldTime).Error; err != nil { t.Fatalf("failed to update node last_seen_at: %v", err) } nodesResp = performJSONRequest(t, engine, adminToken, http.MethodGet, "/api/nodes/", nil) @@ -131,9 +151,23 @@ func TestPhase2AgentLifecycle(t *testing.T) { if nodes[0].Status != service.NodeStatusOffline { t.Fatal("expected node to be shown as offline after timeout") } + + deleteResp := performJSONRequest(t, engine, adminToken, http.MethodDelete, "/api/nodes/"+toString(createdNode.ID), nil) + if !deleteResp.Success { + t.Fatalf("expected delete node success, got %s", deleteResp.Message) + } + + deniedReq := httptest.NewRequest(http.MethodPost, "/api/agent/nodes/heartbeat", bytes.NewReader([]byte(`{"ip":"10.0.0.9","agent_version":"0.1.1"}`))) + deniedReq.Header.Set("Content-Type", "application/json") + deniedReq.Header.Set("X-Agent-Token", registration.AgentToken) + deniedRecorder := httptest.NewRecorder() + engine.ServeHTTP(deniedRecorder, deniedReq) + if deniedRecorder.Code != http.StatusUnauthorized { + t.Fatalf("expected deleted node token to be rejected, got %d", deniedRecorder.Code) + } } -func performAgentJSONRequest(t *testing.T, engine http.Handler, method string, path string, body any) apiResponse { +func performAgentJSONRequestWithToken(t *testing.T, engine http.Handler, token string, method string, path string, body any) apiResponse { t.Helper() var payload []byte var err error @@ -147,7 +181,7 @@ func performAgentJSONRequest(t *testing.T, engine http.Handler, method string, p if body != nil { req.Header.Set("Content-Type", "application/json") } - req.Header.Set("X-Agent-Token", common.AgentToken) + req.Header.Set("X-Agent-Token", token) recorder := httptest.NewRecorder() engine.ServeHTTP(recorder, req) if recorder.Code != http.StatusOK { diff --git a/atsf_server/service/agent.go b/atsf_server/service/agent.go index 9aaf76e5..ea4dc3af 100644 --- a/atsf_server/service/agent.go +++ b/atsf_server/service/agent.go @@ -14,6 +14,7 @@ import ( const ( NodeStatusOnline = "online" NodeStatusOffline = "offline" + NodeStatusPending = "pending" ApplyResultOK = "success" ApplyResultFailed = "failed" ) @@ -48,6 +49,8 @@ type NodeView struct { NodeID string `json:"node_id"` Name string `json:"name"` IP string `json:"ip"` + DiscoveryToken string `json:"discovery_token,omitempty"` + Pending bool `json:"pending"` AgentVersion string `json:"agent_version"` NginxVersion string `json:"nginx_version"` Status string `json:"status"` @@ -61,14 +64,61 @@ type NodeView struct { UpdatedAt time.Time `json:"updated_at"` } -func RegisterNode(payload AgentNodePayload) (*model.Node, error) { - common.SysLog("agent register request received: node_id=" + strings.TrimSpace(payload.NodeID) + " name=" + strings.TrimSpace(payload.Name) + " ip=" + strings.TrimSpace(payload.IP)) - return upsertNode(payload) +func RegisterNode(node *model.Node, payload AgentNodePayload) (*AgentRegistrationResponse, error) { + common.SysLog("agent discovery register request received: name=" + strings.TrimSpace(payload.Name) + " ip=" + strings.TrimSpace(payload.IP)) + payload.Name = strings.TrimSpace(payload.Name) + payload.IP = strings.TrimSpace(payload.IP) + payload.AgentVersion = strings.TrimSpace(payload.AgentVersion) + payload.NginxVersion = strings.TrimSpace(payload.NginxVersion) + payload.CurrentVersion = strings.TrimSpace(payload.CurrentVersion) + payload.LastError = strings.TrimSpace(payload.LastError) + if node == nil { + return nil, errors.New("节点不存在") + } + if payload.IP == "" { + return nil, errors.New("ip 不能为空") + } + if payload.AgentVersion == "" { + return nil, errors.New("agent_version 不能为空") + } + agentToken, err := newRandomToken() + if err != nil { + return nil, err + } + applyNodeRuntime(node, payload, true) + node.AgentToken = agentToken + node.DiscoveryToken = "" + if err = node.Update(); err != nil { + return nil, err + } + common.SysLog("agent discovery register succeeded: node_id=" + node.NodeID + " name=" + node.Name) + return &AgentRegistrationResponse{ + NodeID: node.NodeID, + AgentToken: node.AgentToken, + Name: node.Name, + }, nil } -func HeartbeatNode(payload AgentNodePayload) (*model.Node, error) { - common.SysLog("agent heartbeat received: node_id=" + strings.TrimSpace(payload.NodeID) + " current_version=" + strings.TrimSpace(payload.CurrentVersion)) - return upsertNode(payload) +func HeartbeatNode(node *model.Node, payload AgentNodePayload) (*model.Node, error) { + common.SysLog("agent heartbeat received: node_id=" + node.NodeID + " current_version=" + strings.TrimSpace(payload.CurrentVersion)) + payload.NodeID = node.NodeID + payload.Name = strings.TrimSpace(payload.Name) + payload.IP = strings.TrimSpace(payload.IP) + payload.AgentVersion = strings.TrimSpace(payload.AgentVersion) + payload.NginxVersion = strings.TrimSpace(payload.NginxVersion) + payload.CurrentVersion = strings.TrimSpace(payload.CurrentVersion) + payload.LastError = strings.TrimSpace(payload.LastError) + if payload.IP == "" { + return nil, errors.New("ip 不能为空") + } + if payload.AgentVersion == "" { + return nil, errors.New("agent_version 不能为空") + } + applyNodeRuntime(node, payload, true) + if err := model.DB.Model(node).Select("ip", "agent_version", "nginx_version", "status", "current_version", "last_seen_at", "last_error").Updates(node).Error; err != nil { + return nil, err + } + return node, nil } func GetActiveConfigForAgent() (*AgentConfigResponse, error) { @@ -153,30 +203,18 @@ func ListNodeViews() ([]*NodeView, error) { } views := make([]*NodeView, 0, len(nodes)) for _, node := range nodes { - computedStatus := computeNodeStatus(node.LastSeenAt) + computedStatus := computeNodeStatus(node) if node.Status != computedStatus { if computedStatus == NodeStatusOffline { common.SysError("node offline: node_id=" + node.NodeID + " name=" + node.Name + " ip=" + node.IP + " last_seen_at=" + node.LastSeenAt.Format(time.RFC3339)) - } else { + } else if computedStatus == NodeStatusOnline { common.SysLog("node online: node_id=" + node.NodeID + " name=" + node.Name + " ip=" + node.IP) } _ = model.DB.Model(node).Update("status", computedStatus).Error node.Status = computedStatus } - view := &NodeView{ - ID: node.ID, - NodeID: node.NodeID, - Name: node.Name, - IP: node.IP, - AgentVersion: node.AgentVersion, - NginxVersion: node.NginxVersion, - Status: computedStatus, - CurrentVersion: node.CurrentVersion, - LastSeenAt: node.LastSeenAt, - LastError: node.LastError, - CreatedAt: node.CreatedAt, - UpdatedAt: node.UpdatedAt, - } + view := buildNodeView(node) + view.Status = computedStatus if log, err := model.GetLatestApplyLog(node.NodeID); err == nil { view.LatestApplyResult = log.Result view.LatestApplyMessage = log.Message @@ -192,72 +230,20 @@ func ListApplyLogs(nodeID string) ([]*model.ApplyLog, error) { } func upsertNode(payload AgentNodePayload) (*model.Node, error) { - now := time.Now() - payload.NodeID = strings.TrimSpace(payload.NodeID) - payload.Name = strings.TrimSpace(payload.Name) - payload.IP = strings.TrimSpace(payload.IP) - payload.AgentVersion = strings.TrimSpace(payload.AgentVersion) - payload.NginxVersion = strings.TrimSpace(payload.NginxVersion) - payload.CurrentVersion = strings.TrimSpace(payload.CurrentVersion) - payload.LastError = strings.TrimSpace(payload.LastError) - if payload.NodeID == "" { - return nil, errors.New("node_id 不能为空") - } - if payload.Name == "" { - return nil, errors.New("name 不能为空") - } - if payload.IP == "" { - return nil, errors.New("ip 不能为空") - } - if payload.AgentVersion == "" { - return nil, errors.New("agent_version 不能为空") - } - - node := &model.Node{} - err := model.DB.Where("node_id = ?", payload.NodeID).First(node).Error - if err != nil { - if !errors.Is(err, gorm.ErrRecordNotFound) { - return nil, err - } - node = &model.Node{ - NodeID: payload.NodeID, - } - } - previousStatus := node.Status - previousIP := node.IP - previousVersion := node.CurrentVersion - previousAgentVersion := node.AgentVersion - node.Name = payload.Name - node.IP = payload.IP - node.AgentVersion = payload.AgentVersion - node.NginxVersion = payload.NginxVersion - node.Status = NodeStatusOnline - node.CurrentVersion = payload.CurrentVersion - node.LastSeenAt = now - node.LastError = payload.LastError - if node.ID == 0 { - if err = model.DB.Create(node).Error; err != nil { - return nil, err - } - common.SysLog("node online: node_id=" + node.NodeID + " name=" + node.Name + " ip=" + node.IP + " agent_version=" + node.AgentVersion) - return node, nil - } - if err = model.DB.Model(node).Select("name", "ip", "agent_version", "nginx_version", "status", "current_version", "last_seen_at", "last_error").Updates(node).Error; err != nil { - return nil, err - } - if previousStatus != NodeStatusOnline { - common.SysLog("node online: node_id=" + node.NodeID + " name=" + node.Name + " ip=" + node.IP + " agent_version=" + node.AgentVersion) - } else if previousIP != node.IP || previousVersion != node.CurrentVersion || previousAgentVersion != node.AgentVersion { - common.SysLog("node metadata updated: node_id=" + node.NodeID + " ip=" + previousIP + "->" + node.IP + " agent_version=" + previousAgentVersion + "->" + node.AgentVersion + " current_version=" + previousVersion + "->" + node.CurrentVersion) - } - return node, nil + return nil, errors.New("不再支持匿名自动注册") } -func computeNodeStatus(lastSeenAt time.Time) string { - if lastSeenAt.IsZero() { +func computeNodeStatus(node *model.Node) string { + if node == nil { return NodeStatusOffline } - if time.Since(lastSeenAt) > common.NodeOfflineThreshold { + if strings.TrimSpace(node.AgentToken) == "" && strings.TrimSpace(node.DiscoveryToken) != "" { + return NodeStatusPending + } + if node.LastSeenAt.IsZero() { + return NodeStatusOffline + } + if time.Since(node.LastSeenAt) > common.NodeOfflineThreshold { return NodeStatusOffline } return NodeStatusOnline diff --git a/atsf_server/service/node.go b/atsf_server/service/node.go new file mode 100644 index 00000000..3433e522 --- /dev/null +++ b/atsf_server/service/node.go @@ -0,0 +1,148 @@ +package service + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "gin-template/common" + "gin-template/model" + "strings" + "time" +) + +type NodeInput struct { + Name string `json:"name"` +} + +type AgentRegistrationResponse struct { + NodeID string `json:"node_id"` + AgentToken string `json:"agent_token"` + Name string `json:"name"` +} + +func CreateNode(input NodeInput) (*NodeView, error) { + name := strings.TrimSpace(input.Name) + if name == "" { + return nil, errors.New("节点名不能为空") + } + node := &model.Node{ + Name: name, + IP: "", + AgentVersion: "", + NginxVersion: "", + Status: NodeStatusPending, + } + var err error + node.NodeID, err = newServerNodeID() + if err != nil { + return nil, err + } + node.DiscoveryToken, err = newRandomToken() + if err != nil { + return nil, err + } + if err := node.Insert(); err != nil { + if isUniqueConstraintError(err) { + return nil, errors.New("节点标识生成冲突，请重试") + } + return nil, err + } + common.SysLog("node created: name=" + node.Name + " node_id=" + node.NodeID) + return buildNodeView(node), nil +} + +func UpdateNode(id uint, input NodeInput) (*NodeView, error) { + name := strings.TrimSpace(input.Name) + if name == "" { + return nil, errors.New("节点名不能为空") + } + node, err := model.GetNodeByID(id) + if err != nil { + return nil, err + } + node.Name = name + if err = node.Update(); err != nil { + return nil, err + } + common.SysLog("node updated: name=" + node.Name + " node_id=" + node.NodeID) + return buildNodeView(node), nil +} + +func DeleteNode(id uint) error { + node, err := model.GetNodeByID(id) + if err != nil { + return err + } + common.SysLog("node deleted: name=" + node.Name + " node_id=" + node.NodeID) + return node.Delete() +} + +func AuthenticateAgentToken(token string) (*model.Node, error) { + token = strings.TrimSpace(token) + if token == "" { + return nil, errors.New("缺少 Agent Token") + } + return model.GetNodeByAgentToken(token) +} + +func AuthenticateDiscoveryToken(token string) (*model.Node, error) { + token = strings.TrimSpace(token) + if token == "" { + return nil, errors.New("缺少 Discovery Token") + } + return model.GetNodeByDiscoveryToken(token) +} + +func buildNodeView(node *model.Node) *NodeView { + status := computeNodeStatus(node) + view := &NodeView{ + ID: node.ID, + NodeID: node.NodeID, + Name: node.Name, + IP: node.IP, + AgentVersion: node.AgentVersion, + NginxVersion: node.NginxVersion, + Status: status, + CurrentVersion: node.CurrentVersion, + LastSeenAt: node.LastSeenAt, + LastError: node.LastError, + CreatedAt: node.CreatedAt, + UpdatedAt: node.UpdatedAt, + Pending: status == NodeStatusPending, + } + if status == NodeStatusPending { + view.DiscoveryToken = node.DiscoveryToken + } + return view +} + +func applyNodeRuntime(node *model.Node, payload AgentNodePayload, preserveName bool) { + if !preserveName || strings.TrimSpace(node.Name) == "" { + if strings.TrimSpace(payload.Name) != "" { + node.Name = strings.TrimSpace(payload.Name) + } + } + node.IP = strings.TrimSpace(payload.IP) + node.AgentVersion = strings.TrimSpace(payload.AgentVersion) + node.NginxVersion = strings.TrimSpace(payload.NginxVersion) + node.Status = NodeStatusOnline + node.CurrentVersion = strings.TrimSpace(payload.CurrentVersion) + node.LastSeenAt = time.Now() + node.LastError = strings.TrimSpace(payload.LastError) +} + +func newRandomToken() (string, error) { + buf := make([]byte, 16) + if _, err := rand.Read(buf); err != nil { + return "", err + } + return hex.EncodeToString(buf), nil +} + +func newServerNodeID() (string, error) { + token, err := newRandomToken() + if err != nil { + return "", err + } + return "node-" + token, nil +} diff --git a/atsf_server/web/src/pages/Node/index.js b/atsf_server/web/src/pages/Node/index.js index 60f0ef46..bdd82584 100644 --- a/atsf_server/web/src/pages/Node/index.js +++ b/atsf_server/web/src/pages/Node/index.js @@ -1,84 +1,179 @@ -import React, { useEffect, useState } from 'react'; -import { Header, Label, Segment, Table } from 'semantic-ui-react'; -import { API, formatDateTime, showError } from '../../helpers'; - -const renderStatus = (status) => { - if (status === 'online') { - return ; - } - return ; -}; - -const renderApply = (result) => { - if (result === 'success') { - return ; - } - if (result === 'failed') { - return ; - } - return ; -}; - -const Node = () => { - const [nodes, setNodes] = useState([]); - const [loading, setLoading] = useState(false); - - const loadNodes = async () => { - setLoading(true); - const res = await API.get('/api/nodes/'); - const { success, message, data } = res.data; - if (success) { - setNodes(data || []); - } else { - showError(message); - } - setLoading(false); - }; - - useEffect(() => { - loadNodes().then(); - }, []); - - return ( - -
节点状态
-

查看节点在线状态、当前版本和最近一次应用结果。

- - - - - 节点名 - Node ID - IP - 状态 - Agent / Nginx - 当前版本 - 最近应用 - 最近心跳 - 错误 - - - - {nodes.map((node) => ( - - {node.name} - {node.node_id} - {node.ip} - {renderStatus(node.status)} - {node.agent_version} / {node.nginx_version || 'unknown'} - {node.current_version || '未应用'} - - {renderApply(node.latest_apply_result)} -
{node.latest_apply_message || '暂无记录'}
- - {formatDateTime(node.last_seen_at)} - {node.last_error || '无'} - - ))} - -
- - ); -}; - -export default Node; +import React, { useEffect, useState } from 'react'; +import { Button, Form, Header, Label, Segment, Table } from 'semantic-ui-react'; +import { API, formatDateTime, showError, showSuccess } from '../../helpers'; + +const initialForm = { + name: '', +}; + +const renderStatus = (status) => { + if (status === 'online') { + return ; + } + if (status === 'pending') { + return ; + } + return ; +}; + +const renderApply = (result) => { + if (result === 'success') { + return ; + } + if (result === 'failed') { + return ; + } + return ; +}; + +const Node = () => { + const [nodes, setNodes] = useState([]); + const [loading, setLoading] = useState(false); + const [submitting, setSubmitting] = useState(false); + const [form, setForm] = useState(initialForm); + const [editingId, setEditingId] = useState(null); + + const loadNodes = async () => { + setLoading(true); + const res = await API.get('/api/nodes/'); + const { success, message, data } = res.data; + if (success) { + setNodes(data || []); + } else { + showError(message); + } + setLoading(false); + }; + + useEffect(() => { + loadNodes().then(); + }, []); + + const resetForm = () => { + setForm(initialForm); + setEditingId(null); + }; + + const submitNode = async () => { + setSubmitting(true); + const payload = { + name: form.name.trim(), + }; + const res = editingId + ? await API.put(`/api/nodes/${editingId}`, payload) + : await API.post('/api/nodes/', payload); + const { success, message } = res.data; + if (success) { + showSuccess(editingId ? '节点已更新' : '节点已创建'); + resetForm(); + await loadNodes(); + } else { + showError(message); + } + setSubmitting(false); + }; + + const beginEdit = (node) => { + setEditingId(node.id); + setForm({ + name: node.name || '', + }); + }; + + const deleteNode = async (node) => { + if (!window.confirm(`确认删除节点“${node.name}”吗？删除后该节点需要重新创建并重新接入。`)) { + return; + } + const res = await API.delete(`/api/nodes/${node.id}`); + const { success, message } = res.data; + if (success) { + showSuccess('节点已删除'); + if (editingId === node.id) { + resetForm(); + } + await loadNodes(); + } else { + showError(message); + } + }; + + return ( + +
节点管理
+

创建节点、查看接入令牌，并跟踪节点在线状态、版本与最近一次应用结果。

+ +
+ + setForm({ ...form, name: value })} + /> + + + {editingId ? ( + + ) : null} +
+ + + + + 节点名 + Node ID + Discovery Token + IP + 状态 + Agent / Nginx + 当前版本 + 最近应用 + 最近心跳 + 错误 + 操作 + + + + {nodes.map((node) => ( + + {node.name} + {node.node_id} + + {node.pending && node.discovery_token ? ( + <> + +
{node.discovery_token}
+ + ) : '已激活'} + + {node.ip} + {renderStatus(node.status)} + {node.agent_version} / {node.nginx_version || 'unknown'} + {node.current_version || '未应用'} + + {renderApply(node.latest_apply_result)} +
{node.latest_apply_message || '暂无记录'}
+ + {node.last_seen_at ? formatDateTime(node.last_seen_at) : '暂无'} + {node.last_error || '无'} + + + + + + ))} + +
+ + ); +}; + +export default Node;