tcjlj/OpenFlare
1
0
Fork 0
mirror of https://github.com/Rain-kl/OpenFlare.git synced 2026-10-07 16:16:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: openresty状态上报

Browse Source
This commit is contained in:
ryan
2026-03-12 15:09:34 +08:00
parent ecc71428e4
commit 42ca18681c
26 changed files with 1232 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
atsf_agent/internal/agent/runner.go
+100 -15
View File
@@ -27,6 +27,11 @@ type Updater interface {
CheckAndUpdate(ctx context.Context, repo string, options UpdateOptions) error
}
type RuntimeManager interface {
CheckHealth(ctx context.Context) error
Restart(ctx context.Context) error
}
type UpdateOptions struct {
Channel string
TagName string
@@ -39,12 +44,14 @@ type Runner struct {
HeartbeatService HeartbeatService
SyncService SyncService
Updater Updater
RuntimeManager RuntimeManager
autoUpdate bool
updateNow bool
updateRepo string
updateChan string
updateTag string
autoUpdate bool
updateNow bool
updateRepo string
updateChan string
updateTag string
restartOpenrestyNow bool
}
func (r *Runner) Run(ctx context.Context) error {
@@ -60,12 +67,14 @@ func (r *Runner) Run(ctx context.Context) error {
} else {
log.Printf("agent startup sync completed")
}
r.refreshOpenrestyHealth(ctx)
settings, hbErr := r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID))
if hbErr != nil {
log.Printf("agent startup heartbeat failed: %v", hbErr)
} else {
log.Printf("agent startup heartbeat succeeded: node_id=%s", nodeID)
r.applySettings(settings)
r.tryRestartOpenresty(ctx)
}
} else if err = r.tryRegister(ctx, &nodeID); err != nil {
log.Printf("agent initial discovery register failed: %v", err)
@@ -88,6 +97,7 @@ func (r *Runner) Run(ctx context.Context) error {
}
continue
}
r.refreshOpenrestyHealth(ctx)
settings, hbErr := r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID))
if hbErr != nil {
log.Printf("agent heartbeat failed: %v", hbErr)
@@ -96,6 +106,7 @@ func (r *Runner) Run(ctx context.Context) error {
heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration())
syncTicker.Reset(r.Config.SyncInterval.Duration())
}
r.tryRestartOpenresty(ctx)
r.tryAutoUpdate(ctx)
}
case <-syncTicker.C:
@@ -143,9 +154,28 @@ func (r *Runner) applySettings(settings *protocol.AgentSettings) bool {
r.updateRepo = strings.TrimSpace(settings.UpdateRepo)
r.updateChan = strings.TrimSpace(settings.UpdateChannel)
r.updateTag = strings.TrimSpace(settings.UpdateTag)
r.restartOpenrestyNow = settings.RestartOpenrestyNow
return changed
}
func (r *Runner) tryRestartOpenresty(ctx context.Context) {
if !r.restartOpenrestyNow {
return
}
r.restartOpenrestyNow = false
if r.RuntimeManager == nil {
return
}
log.Printf("agent openresty restart requested by server")
if err := r.RuntimeManager.Restart(ctx); err != nil {
log.Printf("agent openresty restart failed: %v", err)
r.recordOpenrestyUnhealthy(err, false)
return
}
log.Printf("agent openresty restart succeeded")
r.recordOpenrestyHealthy()
}
func (r *Runner) tryAutoUpdate(ctx context.Context) {
force := r.updateNow
shouldCheck := r.autoUpdate || force
@@ -224,15 +254,70 @@ func (r *Runner) recordSyncError(err error) {
}
}
func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
snapshot, _ := r.StateStore.Load()
return protocol.NodePayload{
NodeID: nodeID,
Name: r.Config.NodeName,
IP: r.Config.NodeIP,
AgentVersion: r.Config.AgentVersion,
NginxVersion: r.Config.NginxVersion,
CurrentVersion: snapshot.CurrentVersion,
LastError: snapshot.LastError,
func (r *Runner) refreshOpenrestyHealth(ctx context.Context) {
if r.RuntimeManager == nil || r.StateStore == nil {
return
}
if err := r.RuntimeManager.CheckHealth(ctx); err != nil {
r.recordOpenrestyUnhealthy(err, true)
return
}
r.recordOpenrestyHealthy()
}
func (r *Runner) recordOpenrestyHealthy() {
if r.StateStore == nil {
return
}
snapshot, err := r.StateStore.Load()
if err != nil {
log.Printf("load state before recording openresty health failed: %v", err)
return
}
if snapshot.OpenrestyStatus == protocol.OpenrestyStatusHealthy && strings.TrimSpace(snapshot.OpenrestyMessage) == "" {
return
}
snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
snapshot.OpenrestyMessage = ""
if err = r.StateStore.Save(snapshot); err != nil {
log.Printf("save state after recording openresty health failed: %v", err)
}
}
func (r *Runner) recordOpenrestyUnhealthy(err error, fallbackOnly bool) {
if err == nil || r.StateStore == nil {
return
}
snapshot, loadErr := r.StateStore.Load()
if loadErr != nil {
log.Printf("load state before recording openresty error failed: %v", loadErr)
return
}
message := strings.TrimSpace(err.Error())
if !fallbackOnly || strings.TrimSpace(snapshot.OpenrestyMessage) == "" {
snapshot.OpenrestyMessage = message
}
snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
if saveErr := r.StateStore.Save(snapshot); saveErr != nil {
log.Printf("save state after recording openresty error failed: %v", saveErr)
}
}
func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
snapshot, _ := r.StateStore.Load()
openrestyStatus := strings.TrimSpace(snapshot.OpenrestyStatus)
if openrestyStatus == "" {
openrestyStatus = protocol.OpenrestyStatusUnknown
}
return protocol.NodePayload{
NodeID: nodeID,
Name: r.Config.NodeName,
IP: r.Config.NodeIP,
AgentVersion: r.Config.AgentVersion,
NginxVersion: r.Config.NginxVersion,
CurrentVersion: snapshot.CurrentVersion,
LastError: snapshot.LastError,
OpenrestyStatus: openrestyStatus,
OpenrestyMessage: snapshot.OpenrestyMessage,
}
}
atsf_agent/internal/agent/runner_test.go
+105 -9
View File
@@ -15,14 +15,16 @@ import (
)
type fakeHeartbeatService struct {
mu sync.Mutex
registerCalls int
heartbeatCalls int
registerErr error
registerResp *protocol.RegisterNodeResponse
heartbeatErrs []error
onHeartbeat func(int)
lastToken string
mu sync.Mutex
registerCalls int
heartbeatCalls int
registerErr error
registerResp *protocol.RegisterNodeResponse
heartbeatErrs []error
heartbeatSettings []*protocol.AgentSettings
heartbeatPayloads []protocol.NodePayload
onHeartbeat func(int)
lastToken string
}
func (f *fakeHeartbeatService) Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) {
@@ -36,16 +38,21 @@ func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.N
f.mu.Lock()
f.heartbeatCalls++
callIndex := f.heartbeatCalls
f.heartbeatPayloads = append(f.heartbeatPayloads, payload)
var err error
if len(f.heartbeatErrs) >= callIndex {
err = f.heartbeatErrs[callIndex-1]
}
var settings *protocol.AgentSettings
if len(f.heartbeatSettings) >= callIndex {
settings = f.heartbeatSettings[callIndex-1]
}
onHeartbeat := f.onHeartbeat
f.mu.Unlock()
if onHeartbeat != nil {
onHeartbeat(callIndex)
}
return nil, err
return settings, err
}
func (f *fakeHeartbeatService) SetToken(token string) {
@@ -63,6 +70,30 @@ type fakeSyncService struct {
onSyncOnceCall func(int)
}
type fakeRuntimeManager struct {
mu sync.Mutex
healthErr error
restartErr error
restartCalls int
clearHealthOnRestart bool
}
func (f *fakeRuntimeManager) CheckHealth(ctx context.Context) error {
f.mu.Lock()
defer f.mu.Unlock()
return f.healthErr
}
func (f *fakeRuntimeManager) Restart(ctx context.Context) error {
f.mu.Lock()
defer f.mu.Unlock()
f.restartCalls++
if f.clearHealthOnRestart && f.restartErr == nil {
f.healthErr = nil
}
return f.restartErr
}
func (f *fakeSyncService) SyncOnStartup(ctx context.Context) error {
f.mu.Lock()
defer f.mu.Unlock()
@@ -182,6 +213,71 @@ func TestRunnerDoesNotExitOnHeartbeatOrSyncError(t *testing.T) {
}
}
func TestRunnerReportsOpenrestyHealthAndExecutesRestart(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
if err := stateStore.Save(&state.Snapshot{
OpenrestyStatus: protocol.OpenrestyStatusUnhealthy,
OpenrestyMessage: "docker run openresty failed: bind 80 already allocated",
}); err != nil {
t.Fatalf("failed to seed state: %v", err)
}
heartbeatService := &fakeHeartbeatService{
heartbeatSettings: []*protocol.AgentSettings{{RestartOpenrestyNow: true}},
onHeartbeat: func(callCount int) {
if callCount >= 1 {
cancel()
}
},
}
runtimeManager := &fakeRuntimeManager{
healthErr: errors.New("docker openresty container is not running"),
clearHealthOnRestart: true,
}
runner := &Runner{
Config: &config.Config{
AgentToken: "agent-token",
NodeName: "edge-01",
NodeIP: "10.0.0.8",
AgentVersion: config.AgentVersion,
NginxVersion: "1.27.1.2",
HeartbeatInterval: config.MillisecondDuration(10 * time.Millisecond),
SyncInterval: config.MillisecondDuration(100 * time.Millisecond),
},
StateStore: stateStore,
HeartbeatService: heartbeatService,
SyncService: &fakeSyncService{},
RuntimeManager: runtimeManager,
}
err := runner.Run(ctx)
if !errors.Is(err, context.Canceled) {
t.Fatalf("expected context cancellation, got %v", err)
}
if len(heartbeatService.heartbeatPayloads) == 0 {
t.Fatal("expected at least one heartbeat payload")
}
payload := heartbeatService.heartbeatPayloads[0]
if payload.OpenrestyStatus != protocol.OpenrestyStatusUnhealthy {
t.Fatalf("expected unhealthy openresty status in heartbeat payload, got %q", payload.OpenrestyStatus)
}
if payload.OpenrestyMessage != "docker run openresty failed: bind 80 already allocated" {
t.Fatalf("unexpected openresty message: %q", payload.OpenrestyMessage)
}
if runtimeManager.restartCalls != 1 {
t.Fatalf("expected one openresty restart attempt, got %d", runtimeManager.restartCalls)
}
snapshot, loadErr := stateStore.Load()
if loadErr != nil {
t.Fatalf("failed to load state: %v", loadErr)
}
if snapshot.OpenrestyStatus != protocol.OpenrestyStatusHealthy || snapshot.OpenrestyMessage != "" {
t.Fatal("expected restart success to mark openresty healthy")
}
}
func TestRunnerDiscoveryRegisterUpdatesTokenAndNodeID(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()