feat: enhance logging for agent registration, heartbeat, and configuration sync processes

This commit is contained in:
ryan
2026-03-10 14:55:35 +08:00
parent f396c8c74e
commit e7dc18e6ca
7 changed files with 215 additions and 80 deletions
+3
View File
@@ -24,6 +24,7 @@ func main() {
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
log.Printf("agent config loaded: server=%s node=%s ip=%s heartbeat_interval=%s sync_interval=%s route_config=%s cert_dir=%s", cfg.ServerURL, cfg.NodeName, cfg.NodeIP, cfg.HeartbeatInterval, cfg.SyncInterval, cfg.RouteConfigPath, cfg.CertDir)
client := httpclient.New(cfg.ServerURL, cfg.AgentToken, cfg.RequestTimeout) client := httpclient.New(cfg.ServerURL, cfg.AgentToken, cfg.RequestTimeout)
stateStore := state.NewStore(cfg.StatePath) stateStore := state.NewStore(cfg.StatePath)
@@ -49,8 +50,10 @@ func main() {
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop() defer stop()
log.Printf("agent process started")
if err = runner.Run(ctx); err != nil && err != context.Canceled { if err = runner.Run(ctx); err != nil && err != context.Canceled {
log.Fatal(err) log.Fatal(err)
} }
log.Printf("agent process stopped")
} }
+12
View File
@@ -32,15 +32,22 @@ func (r *Runner) Run(ctx context.Context) error {
if err != nil { if err != nil {
return err 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 { if err = r.HeartbeatService.Register(ctx, r.nodePayload(nodeID)); err != nil {
log.Printf("agent register failed: %v", err) 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 { if err = r.SyncService.SyncOnStartup(ctx); err != nil {
r.recordSyncError(err) r.recordSyncError(err)
log.Printf("agent startup sync failed: %v", 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 { if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil {
log.Printf("agent startup heartbeat failed: %v", err) log.Printf("agent startup heartbeat failed: %v", err)
} else {
log.Printf("agent startup heartbeat succeeded: node_id=%s", nodeID)
} }
heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval) heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval)
@@ -51,15 +58,19 @@ func (r *Runner) Run(ctx context.Context) error {
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
log.Printf("agent runner shutting down: %v", ctx.Err())
return ctx.Err() return ctx.Err()
case <-heartbeatTicker.C: case <-heartbeatTicker.C:
if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil { if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil {
log.Printf("agent heartbeat failed: %v", err) log.Printf("agent heartbeat failed: %v", err)
} }
case <-syncTicker.C: case <-syncTicker.C:
log.Printf("agent sync tick: node_id=%s", nodeID)
if err = r.SyncService.SyncOnce(ctx); err != nil { if err = r.SyncService.SyncOnce(ctx); err != nil {
r.recordSyncError(err) r.recordSyncError(err)
log.Printf("agent sync failed: %v", err) log.Printf("agent sync failed: %v", err)
} else {
log.Printf("agent sync completed")
} }
} }
} }
@@ -75,6 +86,7 @@ func (r *Runner) recordSyncError(err error) {
return return
} }
snapshot.LastError = err.Error() snapshot.LastError = err.Error()
log.Printf("recording sync error into state: %s", snapshot.LastError)
if saveErr := r.StateStore.Save(snapshot); saveErr != nil { if saveErr := r.StateStore.Save(snapshot); saveErr != nil {
log.Printf("save state after sync error failed: %v", saveErr) log.Printf("save state after sync error failed: %v", saveErr)
} }
+30 -1
View File
@@ -5,6 +5,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"log"
"net/http" "net/http"
"strings" "strings"
"time" "time"
@@ -29,6 +30,7 @@ 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) 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) return c.postJSON(ctx, "/api/agent/nodes/register", payload, nil)
} }
@@ -37,6 +39,7 @@ func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) er
} }
func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) { func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) {
log.Printf("http get active config request")
resp := protocol.APIResponse[protocol.ActiveConfigResponse]{} resp := protocol.APIResponse[protocol.ActiveConfigResponse]{}
if err := c.getJSON(ctx, "/api/agent/config-versions/active", &resp); err != nil { if err := c.getJSON(ctx, "/api/agent/config-versions/active", &resp); err != nil {
return nil, err return nil, err
@@ -44,10 +47,12 @@ func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigRes
if !resp.Success { if !resp.Success {
return nil, errors.New(resp.Message) return nil, errors.New(resp.Message)
} }
log.Printf("http get active config response: version=%s checksum=%s support_files=%d", resp.Data.Version, resp.Data.Checksum, len(resp.Data.SupportFiles))
return &resp.Data, nil return &resp.Data, nil
} }
func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error { func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error {
log.Printf("http report apply log request: node_id=%s version=%s result=%s", payload.NodeID, payload.Version, payload.Result)
return c.postJSON(ctx, "/api/agent/apply-logs", payload, nil) return c.postJSON(ctx, "/api/agent/apply-logs", payload, nil)
} }
@@ -75,23 +80,47 @@ func (c *Client) postJSON(ctx context.Context, path string, body any, target any
} }
func (c *Client) do(req *http.Request, target any) error { func (c *Client) do(req *http.Request, target any) error {
if !isHeartbeatRequest(req) {
log.Printf("http request start: method=%s path=%s", req.Method, req.URL.Path)
}
res, err := c.httpClient.Do(req) res, err := c.httpClient.Do(req)
if err != nil { if err != nil {
log.Printf("http request failed: method=%s path=%s error=%v", req.Method, req.URL.Path, err)
return err return err
} }
defer res.Body.Close() defer res.Body.Close()
if res.StatusCode != http.StatusOK { if res.StatusCode != http.StatusOK {
log.Printf("http request returned non-200: method=%s path=%s status=%s", req.Method, req.URL.Path, res.Status)
return errors.New(res.Status) return errors.New(res.Status)
} }
if target == nil { if target == nil {
var wrapper protocol.APIResponse[json.RawMessage] var wrapper protocol.APIResponse[json.RawMessage]
if err = json.NewDecoder(res.Body).Decode(&wrapper); err != nil { if err = json.NewDecoder(res.Body).Decode(&wrapper); err != nil {
log.Printf("http response decode failed: method=%s path=%s error=%v", req.Method, req.URL.Path, err)
return err return err
} }
if !wrapper.Success { if !wrapper.Success {
log.Printf("http api response failed: method=%s path=%s message=%s", req.Method, req.URL.Path, wrapper.Message)
return errors.New(wrapper.Message) return errors.New(wrapper.Message)
} }
if !isHeartbeatRequest(req) {
log.Printf("http request succeeded: method=%s path=%s", req.Method, req.URL.Path)
}
return nil return nil
} }
return json.NewDecoder(res.Body).Decode(target) if err = json.NewDecoder(res.Body).Decode(target); err != nil {
log.Printf("http response decode failed: method=%s path=%s error=%v", req.Method, req.URL.Path, err)
return err
}
if !isHeartbeatRequest(req) {
log.Printf("http request succeeded: method=%s path=%s", req.Method, req.URL.Path)
}
return nil
}
func isHeartbeatRequest(req *http.Request) bool {
if req == nil || req.URL == nil {
return false
}
return req.Method == http.MethodPost && req.URL.Path == "/api/agent/nodes/heartbeat"
} }
+28 -1
View File
@@ -6,6 +6,7 @@ import (
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt" "fmt"
"log"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@@ -41,18 +42,22 @@ type PathExecutor struct {
} }
func (e *PathExecutor) Test(ctx context.Context) error { func (e *PathExecutor) Test(ctx context.Context) error {
log.Printf("running nginx test with binary: %s", e.Path)
output, err := e.Runner.Run(ctx, e.Path, "-t") output, err := e.Runner.Run(ctx, e.Path, "-t")
if err != nil { if err != nil {
return fmt.Errorf("nginx -t failed: %w: %s", err, string(output)) return fmt.Errorf("nginx -t failed: %w: %s", err, string(output))
} }
log.Printf("nginx test succeeded with binary: %s", e.Path)
return nil return nil
} }
func (e *PathExecutor) Reload(ctx context.Context) error { func (e *PathExecutor) Reload(ctx context.Context) error {
log.Printf("running nginx reload with binary: %s", e.Path)
output, err := e.Runner.Run(ctx, e.Path, "-s", "reload") output, err := e.Runner.Run(ctx, e.Path, "-s", "reload")
if err != nil { if err != nil {
return fmt.Errorf("nginx reload failed: %w: %s", err, string(output)) return fmt.Errorf("nginx reload failed: %w: %s", err, string(output))
} }
log.Printf("nginx reload succeeded with binary: %s", e.Path)
return nil return nil
} }
@@ -71,6 +76,7 @@ type DockerExecutor struct {
} }
func (e *DockerExecutor) Test(ctx context.Context) error { func (e *DockerExecutor) Test(ctx context.Context) error {
log.Printf("running docker nginx test: container=%s image=%s", e.ContainerName, e.Image)
output, err := e.Runner.Run( output, err := e.Runner.Run(
ctx, ctx,
e.DockerBinary, e.DockerBinary,
@@ -87,6 +93,7 @@ func (e *DockerExecutor) Test(ctx context.Context) error {
if err != nil { if err != nil {
return fmt.Errorf("docker nginx -t failed: %w: %s", err, string(output)) return fmt.Errorf("docker nginx -t failed: %w: %s", err, string(output))
} }
log.Printf("docker nginx test succeeded: container=%s", e.ContainerName)
return nil return nil
} }
@@ -95,6 +102,7 @@ func (e *DockerExecutor) Reload(ctx context.Context) error {
} }
func (e *DockerExecutor) EnsureRuntime(ctx context.Context, recreate bool) error { func (e *DockerExecutor) EnsureRuntime(ctx context.Context, recreate bool) error {
log.Printf("ensuring docker nginx runtime: container=%s recreate=%t", e.ContainerName, recreate)
output, err := e.Runner.Run(ctx, e.DockerBinary, "inspect", "-f", "{{.State.Running}}", e.ContainerName) output, err := e.Runner.Run(ctx, e.DockerBinary, "inspect", "-f", "{{.State.Running}}", e.ContainerName)
if err == nil { if err == nil {
if recreate { if recreate {
@@ -104,6 +112,7 @@ func (e *DockerExecutor) EnsureRuntime(ctx context.Context, recreate bool) error
return e.runContainer(ctx) return e.runContainer(ctx)
} }
if strings.TrimSpace(string(output)) == "true" { if strings.TrimSpace(string(output)) == "true" {
log.Printf("docker nginx runtime already healthy: container=%s", e.ContainerName)
return nil return nil
} }
if err := e.removeContainer(ctx); err != nil { if err := e.removeContainer(ctx); err != nil {
@@ -115,6 +124,7 @@ func (e *DockerExecutor) EnsureRuntime(ctx context.Context, recreate bool) error
} }
func (e *DockerExecutor) removeContainer(ctx context.Context) error { func (e *DockerExecutor) removeContainer(ctx context.Context) error {
log.Printf("removing docker nginx container: container=%s", e.ContainerName)
output, err := e.Runner.Run(ctx, e.DockerBinary, "rm", "-f", e.ContainerName) output, err := e.Runner.Run(ctx, e.DockerBinary, "rm", "-f", e.ContainerName)
if err != nil { if err != nil {
text := string(output) text := string(output)
@@ -123,10 +133,12 @@ func (e *DockerExecutor) removeContainer(ctx context.Context) error {
} }
return fmt.Errorf("docker rm nginx failed: %w: %s", err, text) return fmt.Errorf("docker rm nginx failed: %w: %s", err, text)
} }
log.Printf("docker nginx container removed: container=%s", e.ContainerName)
return nil return nil
} }
func (e *DockerExecutor) runContainer(ctx context.Context) error { func (e *DockerExecutor) runContainer(ctx context.Context) error {
log.Printf("starting docker nginx container: container=%s image=%s", e.ContainerName, e.Image)
runArgs := []string{ runArgs := []string{
"run", "-d", "run", "-d",
"--name", e.ContainerName, "--name", e.ContainerName,
@@ -140,6 +152,7 @@ func (e *DockerExecutor) runContainer(ctx context.Context) error {
if runErr != nil { if runErr != nil {
return fmt.Errorf("docker run nginx failed: %w: %s", runErr, string(runOutput)) return fmt.Errorf("docker run nginx failed: %w: %s", runErr, string(runOutput))
} }
log.Printf("docker nginx container started: container=%s", e.ContainerName)
return nil return nil
} }
@@ -151,27 +164,33 @@ type Manager struct {
} }
func (m *Manager) Apply(ctx context.Context, content string, supportFiles []protocol.SupportFile) error { func (m *Manager) Apply(ctx context.Context, content string, supportFiles []protocol.SupportFile) error {
log.Printf("nginx apply started: route_config=%s support_files=%d", m.RouteConfigPath, len(supportFiles))
backup, err := m.backup() backup, err := m.backup()
if err != nil { if err != nil {
return err return err
} }
if err = m.writeSupportFiles(supportFiles); err != nil { if err = m.writeSupportFiles(supportFiles); err != nil {
log.Printf("writing support files failed, restoring backup: error=%v", err)
_ = m.restore(backup) _ = m.restore(backup)
return err return err
} }
renderedContent := m.renderConfig(content) renderedContent := m.renderConfig(content)
if err = os.WriteFile(m.RouteConfigPath, []byte(renderedContent), 0o644); err != nil { if err = os.WriteFile(m.RouteConfigPath, []byte(renderedContent), 0o644); err != nil {
log.Printf("writing nginx route config failed, restoring backup: error=%v", err)
_ = m.restore(backup) _ = m.restore(backup)
return err return err
} }
if err = m.Executor.Test(ctx); err != nil { if err = m.Executor.Test(ctx); err != nil {
log.Printf("nginx test failed after config write, restoring backup: error=%v", err)
_ = m.restore(backup) _ = m.restore(backup)
return err return err
} }
if err = m.Executor.Reload(ctx); err != nil { if err = m.Executor.Reload(ctx); err != nil {
log.Printf("nginx reload failed after config write, restoring backup: error=%v", err)
_ = m.restore(backup) _ = m.restore(backup)
return err return err
} }
log.Printf("nginx apply completed successfully: route_config=%s", m.RouteConfigPath)
return nil return nil
} }
@@ -179,6 +198,7 @@ func (m *Manager) EnsureRuntime(ctx context.Context, recreate bool) error {
if m.Executor == nil { if m.Executor == nil {
return errors.New("executor 未配置") return errors.New("executor 未配置")
} }
log.Printf("nginx ensure runtime requested: recreate=%t", recreate)
return m.Executor.EnsureRuntime(ctx, recreate) return m.Executor.EnsureRuntime(ctx, recreate)
} }
@@ -201,7 +221,9 @@ func (m *Manager) CurrentChecksum() (string, error) {
if err != nil { if err != nil {
return "", err return "", err
} }
return bundleChecksum(normalized, files), nil result := bundleChecksum(normalized, files)
log.Printf("nginx current checksum calculated: route_config=%s checksum=%s support_files=%d", m.RouteConfigPath, result, len(files))
return result, nil
} }
type ExecutorOptions struct { type ExecutorOptions struct {
@@ -272,6 +294,7 @@ func (m *Manager) backup() (*backupState, error) {
return nil, err return nil, err
} }
state.Files = files state.Files = files
log.Printf("nginx backup captured: route_exists=%t support_files=%d", state.RouteExisted, len(state.Files))
return state, nil return state, nil
} }
@@ -279,6 +302,7 @@ func (m *Manager) restore(state *backupState) error {
if state == nil { if state == nil {
return nil return nil
} }
log.Printf("restoring nginx backup: route_existed=%t support_files=%d", state.RouteExisted, len(state.Files))
if state.RouteExisted { if state.RouteExisted {
if err := os.WriteFile(m.RouteConfigPath, state.RouteData, 0o644); err != nil { if err := os.WriteFile(m.RouteConfigPath, state.RouteData, 0o644); err != nil {
return err return err
@@ -304,6 +328,7 @@ func (m *Manager) restore(state *backupState) error {
return err return err
} }
} }
log.Printf("nginx backup restored")
return nil return nil
} }
@@ -311,6 +336,7 @@ func (m *Manager) writeSupportFiles(supportFiles []protocol.SupportFile) error {
if m.CertDir == "" { if m.CertDir == "" {
return nil return nil
} }
log.Printf("writing nginx support files: cert_dir=%s count=%d", m.CertDir, len(supportFiles))
if err := os.RemoveAll(m.CertDir); err != nil && !os.IsNotExist(err) { if err := os.RemoveAll(m.CertDir); err != nil && !os.IsNotExist(err) {
return err return err
} }
@@ -326,6 +352,7 @@ func (m *Manager) writeSupportFiles(supportFiles []protocol.SupportFile) error {
return err return err
} }
} }
log.Printf("nginx support files written: cert_dir=%s count=%d", m.CertDir, len(supportFiles))
return nil return nil
} }
+26 -2
View File
@@ -2,6 +2,7 @@ package sync
import ( import (
"context" "context"
"log"
"atsflare-agent/internal/protocol" "atsflare-agent/internal/protocol"
"atsflare-agent/internal/state" "atsflare-agent/internal/state"
@@ -46,33 +47,48 @@ func (s *Service) SyncOnStartup(ctx context.Context) error {
} }
func (s *Service) sync(ctx context.Context, startup bool) error { func (s *Service) sync(ctx context.Context, startup bool) error {
mode := "periodic"
if startup {
mode = "startup"
}
log.Printf("sync started: mode=%s", mode)
snapshot, err := s.stateStore.Load() snapshot, err := s.stateStore.Load()
if err != nil { if err != nil {
return err return err
} }
config, err := s.client.GetActiveConfig(ctx) config, err := s.client.GetActiveConfig(ctx)
if err != nil { if err != nil {
log.Printf("fetch active config failed: mode=%s error=%v", mode, err)
return err return err
} }
log.Printf("active config fetched: mode=%s version=%s checksum=%s support_files=%d", mode, config.Version, config.Checksum, len(config.SupportFiles))
currentChecksum, err := s.nginxManager.CurrentChecksum() currentChecksum, err := s.nginxManager.CurrentChecksum()
if err != nil { if err != nil {
return err return err
} }
log.Printf("current local checksum loaded: mode=%s checksum=%s", mode, currentChecksum)
if currentChecksum == config.Checksum { if currentChecksum == config.Checksum {
log.Printf("local nginx config already up to date: mode=%s version=%s", mode, config.Version)
if startup { if startup {
log.Printf("ensuring nginx runtime on startup: version=%s", config.Version)
if err = s.nginxManager.EnsureRuntime(ctx, true); err != nil { if err = s.nginxManager.EnsureRuntime(ctx, true); err != nil {
return err return err
} }
log.Printf("nginx runtime ensured on startup: version=%s", config.Version)
} }
snapshot.CurrentVersion = config.Version snapshot.CurrentVersion = config.Version
snapshot.CurrentChecksum = config.Checksum snapshot.CurrentChecksum = config.Checksum
snapshot.LastError = "" snapshot.LastError = ""
log.Printf("sync finished without changes: mode=%s version=%s", mode, config.Version)
return s.stateStore.Save(snapshot) return s.stateStore.Save(snapshot)
} }
if snapshot.CurrentVersion == config.Version && snapshot.CurrentChecksum == config.Checksum && !startup { if snapshot.CurrentVersion == config.Version && snapshot.CurrentChecksum == config.Checksum && !startup {
log.Printf("skipping apply because state already records target version/checksum: version=%s checksum=%s", config.Version, config.Checksum)
return nil return nil
} }
log.Printf("applying new nginx config: mode=%s from_version=%s to_version=%s old_checksum=%s new_checksum=%s", mode, snapshot.CurrentVersion, config.Version, currentChecksum, config.Checksum)
if err = s.nginxManager.Apply(ctx, config.RenderedConfig, config.SupportFiles); err != nil { if err = s.nginxManager.Apply(ctx, config.RenderedConfig, config.SupportFiles); err != nil {
log.Printf("apply nginx config failed: mode=%s version=%s error=%v", mode, config.Version, err)
snapshot.LastError = err.Error() snapshot.LastError = err.Error()
_ = s.stateStore.Save(snapshot) _ = s.stateStore.Save(snapshot)
reportErr := s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{ reportErr := s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{
@@ -82,20 +98,28 @@ func (s *Service) sync(ctx context.Context, startup bool) error {
Message: err.Error(), Message: err.Error(),
}) })
if reportErr != nil { if reportErr != nil {
log.Printf("report failed apply log failed: version=%s error=%v", config.Version, reportErr)
return reportErr return reportErr
} }
log.Printf("failed apply log reported: version=%s", config.Version)
return err return err
} }
log.Printf("nginx config applied successfully: mode=%s version=%s", mode, config.Version)
snapshot.CurrentVersion = config.Version snapshot.CurrentVersion = config.Version
snapshot.CurrentChecksum = config.Checksum snapshot.CurrentChecksum = config.Checksum
snapshot.LastError = "" snapshot.LastError = ""
if err = s.stateStore.Save(snapshot); err != nil { if err = s.stateStore.Save(snapshot); err != nil {
return err return err
} }
return s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{ if err = s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{
NodeID: snapshot.NodeID, NodeID: snapshot.NodeID,
Version: config.Version, Version: config.Version,
Result: ApplyResultSuccess, Result: ApplyResultSuccess,
Message: "apply success", Message: "apply success",
}) }); err != nil {
log.Printf("report successful apply log failed: version=%s error=%v", config.Version, err)
return err
}
log.Printf("successful apply log reported: version=%s", config.Version)
return nil
} }
+85 -75
View File
@@ -1,75 +1,85 @@
package main package main
import ( import (
"embed" "embed"
"gin-template/common" "fmt"
"gin-template/middleware" "gin-template/common"
"gin-template/model" "gin-template/middleware"
"gin-template/router" "gin-template/model"
"github.com/gin-contrib/sessions" "gin-template/router"
"github.com/gin-contrib/sessions/cookie" "github.com/gin-contrib/sessions"
"github.com/gin-contrib/sessions/redis" "github.com/gin-contrib/sessions/cookie"
"github.com/gin-gonic/gin" "github.com/gin-contrib/sessions/redis"
"log" "github.com/gin-gonic/gin"
"os" "log"
"strconv" "os"
) "strconv"
)
//go:embed web/build
var buildFS embed.FS //go:embed web/build
var buildFS embed.FS
//go:embed web/build/index.html
var indexPage []byte //go:embed web/build/index.html
var indexPage []byte
func main() {
common.SetupGinLog() func main() {
common.SysLog("Gin Template " + common.Version + " started") common.SetupGinLog()
if os.Getenv("GIN_MODE") != "debug" { common.SysLog("Gin Template " + common.Version + " started")
gin.SetMode(gin.ReleaseMode) if os.Getenv("GIN_MODE") != "debug" {
} gin.SetMode(gin.ReleaseMode)
// Initialize SQL Database }
err := model.InitDB() // Initialize SQL Database
if err != nil { err := model.InitDB()
common.FatalLog(err) if err != nil {
} common.FatalLog(err)
defer func() { }
err := model.CloseDB() defer func() {
if err != nil { err := model.CloseDB()
common.FatalLog(err) if err != nil {
} common.FatalLog(err)
}() }
}()
// Initialize Redis
err = common.InitRedisClient() // Initialize Redis
if err != nil { err = common.InitRedisClient()
common.FatalLog(err) if err != nil {
} common.FatalLog(err)
}
// Initialize options
model.InitOptionMap() // Initialize options
model.InitOptionMap()
// Initialize HTTP server
server := gin.Default() // Initialize HTTP server
//server.Use(gzip.Gzip(gzip.DefaultCompression)) server := gin.Default()
server.Use(middleware.CORS()) //server.Use(gzip.Gzip(gzip.DefaultCompression))
server.Use(middleware.CORS())
// Initialize session store
if common.RedisEnabled { // Initialize session store
opt := common.ParseRedisOption() if common.RedisEnabled {
store, _ := redis.NewStore(opt.MinIdleConns, opt.Network, opt.Addr, opt.Password, []byte(common.SessionSecret)) opt := common.ParseRedisOption()
server.Use(sessions.Sessions("session", store)) store, _ := redis.NewStore(opt.MinIdleConns, opt.Network, opt.Addr, opt.Password, []byte(common.SessionSecret))
} else { server.Use(sessions.Sessions("session", store))
store := cookie.NewStore([]byte(common.SessionSecret)) } else {
server.Use(sessions.Sessions("session", store)) store := cookie.NewStore([]byte(common.SessionSecret))
} server.Use(sessions.Sessions("session", store))
}
router.SetRouter(server, buildFS, indexPage)
var port = os.Getenv("PORT") router.SetRouter(server, buildFS, indexPage)
if port == "" { var port = os.Getenv("PORT")
port = strconv.Itoa(*common.Port) if port == "" {
} port = strconv.Itoa(*common.Port)
err = server.Run(":" + port) }
if err != nil { common.SysLog(fmt.Sprintf("server config: port=%s gin_mode=%s sqlite_path=%s redis_enabled=%t upload_path=%s log_dir=%s agent_token_configured=%t node_offline_threshold=%s", port, gin.Mode(), common.SQLitePath, common.RedisEnabled, common.UploadPath, valueOrDefault(*common.LogDir, "stdout"), common.AgentToken != "", common.NodeOfflineThreshold))
log.Println(err) common.SysLog(fmt.Sprintf("server listening on :%s", port))
} err = server.Run(":" + port)
} if err != nil {
log.Println(err)
}
}
func valueOrDefault(value string, fallback string) string {
if value == "" {
return fallback
}
return value
}
+31 -1
View File
@@ -62,16 +62,19 @@ type NodeView struct {
} }
func RegisterNode(payload AgentNodePayload) (*model.Node, error) { 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) return upsertNode(payload)
} }
func HeartbeatNode(payload AgentNodePayload) (*model.Node, error) { 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) return upsertNode(payload)
} }
func GetActiveConfigForAgent() (*AgentConfigResponse, error) { func GetActiveConfigForAgent() (*AgentConfigResponse, error) {
version, err := model.GetActiveConfigVersion() version, err := model.GetActiveConfigVersion()
if err != nil { if err != nil {
common.SysError("agent requested active config but no active version is available")
return nil, err return nil, err
} }
var supportFiles []SupportFile var supportFiles []SupportFile
@@ -80,6 +83,7 @@ func GetActiveConfigForAgent() (*AgentConfigResponse, error) {
return nil, err return nil, err
} }
} }
common.SysLog("agent fetched active config: version=" + version.Version + " checksum=" + version.Checksum)
return &AgentConfigResponse{ return &AgentConfigResponse{
Version: version.Version, Version: version.Version,
Checksum: version.Checksum, Checksum: version.Checksum,
@@ -104,6 +108,7 @@ func ReportApplyLog(payload ApplyLogPayload) (*model.ApplyLog, error) {
if payload.Result != ApplyResultOK && payload.Result != ApplyResultFailed { if payload.Result != ApplyResultOK && payload.Result != ApplyResultFailed {
return nil, errors.New("result 仅支持 success 或 failed") return nil, errors.New("result 仅支持 success 或 failed")
} }
common.SysLog("agent apply log received: node_id=" + payload.NodeID + " version=" + payload.Version + " result=" + payload.Result)
log := &model.ApplyLog{ log := &model.ApplyLog{
NodeID: payload.NodeID, NodeID: payload.NodeID,
@@ -133,6 +138,11 @@ func ReportApplyLog(payload ApplyLogPayload) (*model.ApplyLog, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
if payload.Result == ApplyResultOK {
common.SysLog("agent apply reported success: node_id=" + payload.NodeID + " version=" + payload.Version)
} else {
common.SysError("agent apply reported failure: node_id=" + payload.NodeID + " version=" + payload.Version + " message=" + payload.Message)
}
return log, nil return log, nil
} }
@@ -143,6 +153,16 @@ func ListNodeViews() ([]*NodeView, error) {
} }
views := make([]*NodeView, 0, len(nodes)) views := make([]*NodeView, 0, len(nodes))
for _, node := range nodes { for _, node := range nodes {
computedStatus := computeNodeStatus(node.LastSeenAt)
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 {
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{ view := &NodeView{
ID: node.ID, ID: node.ID,
NodeID: node.NodeID, NodeID: node.NodeID,
@@ -150,7 +170,7 @@ func ListNodeViews() ([]*NodeView, error) {
IP: node.IP, IP: node.IP,
AgentVersion: node.AgentVersion, AgentVersion: node.AgentVersion,
NginxVersion: node.NginxVersion, NginxVersion: node.NginxVersion,
Status: computeNodeStatus(node.LastSeenAt), Status: computedStatus,
CurrentVersion: node.CurrentVersion, CurrentVersion: node.CurrentVersion,
LastSeenAt: node.LastSeenAt, LastSeenAt: node.LastSeenAt,
LastError: node.LastError, LastError: node.LastError,
@@ -203,6 +223,10 @@ func upsertNode(payload AgentNodePayload) (*model.Node, error) {
NodeID: payload.NodeID, NodeID: payload.NodeID,
} }
} }
previousStatus := node.Status
previousIP := node.IP
previousVersion := node.CurrentVersion
previousAgentVersion := node.AgentVersion
node.Name = payload.Name node.Name = payload.Name
node.IP = payload.IP node.IP = payload.IP
node.AgentVersion = payload.AgentVersion node.AgentVersion = payload.AgentVersion
@@ -215,11 +239,17 @@ func upsertNode(payload AgentNodePayload) (*model.Node, error) {
if err = model.DB.Create(node).Error; err != nil { if err = model.DB.Create(node).Error; err != nil {
return nil, err return nil, err
} }
common.SysLog("node online: node_id=" + node.NodeID + " name=" + node.Name + " ip=" + node.IP + " agent_version=" + node.AgentVersion)
return node, nil 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 { 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 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 node, nil
} }