[优化] 优化 WAF IP 组同步功能及相关文档更新

This commit is contained in:
ryan
2026-06-01 13:55:04 +08:00
parent a092935623
commit a8e8a940a0
27 changed files with 735 additions and 43 deletions
+32 -1
View File
@@ -24,6 +24,8 @@ type SyncService interface {
SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error
SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
WAFIPGroupChecksums() (map[string]string, error)
ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error
}
type Updater interface {
@@ -161,6 +163,7 @@ func (r *Runner) performHeartbeatCycle(ctx context.Context, nodeID string, start
}
slog.Debug("agent heartbeat succeeded", "mode", mode, "node_id", nodeID)
changed := r.applySettings(heartbeatResult.AgentSettings)
r.applyWAFIPGroups(ctx, heartbeatResult.WAFIPGroups)
if startup {
if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
r.recordSyncError(err)
@@ -307,6 +310,14 @@ func (r *Runner) handleWebSocketMessage(ctx context.Context, message protocol.WS
slog.Error("agent ws triggered force sync failed", "version", target.Version, "error", err)
}
return false, nil
case protocol.WSMessageTypeWAFIPGroups:
var groups []protocol.WAFIPGroup
if err := json.Unmarshal(message.Payload, &groups); err != nil {
slog.Debug("agent ws waf ip groups decode failed", "error", err)
return false, nil
}
r.applyWAFIPGroups(ctx, groups)
return false, nil
case protocol.WSMessageTypePing:
slog.Debug("agent ws ping received")
return false, conn.SendPong()
@@ -470,6 +481,7 @@ func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error {
heartbeatResult = &protocol.HeartbeatResult{}
}
r.applySettings(heartbeatResult.AgentSettings)
r.applyWAFIPGroups(ctx, heartbeatResult.WAFIPGroups)
if err = r.SyncService.SyncOnStartup(ctx, heartbeatResult.ActiveConfig); err != nil {
r.recordSyncError(err)
slog.Error("agent post-register startup sync failed", "error", err)
@@ -563,7 +575,7 @@ func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
}
metricSnapshot := observability.BuildSnapshot(r.Config, r.StateStore, managedOpenRestyMetrics)
healthEvents := observability.BuildHealthEvents(snapshot)
return protocol.NodePayload{
payload := protocol.NodePayload{
NodeID: nodeID,
Name: r.Config.NodeName,
IP: r.Config.NodeIP,
@@ -579,6 +591,25 @@ func (r *Runner) nodePayload(nodeID string) protocol.NodePayload {
AccessLogs: accessLogs,
HealthEvents: healthEvents,
}
if r.SyncService != nil {
checksums, err := r.SyncService.WAFIPGroupChecksums()
if err != nil {
slog.Debug("load local waf ip group checksums failed", "error", err)
} else if len(checksums) > 0 {
payload.WAFIPGroupChecksums = checksums
}
}
return payload
}
func (r *Runner) applyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) {
if len(groups) == 0 || r.SyncService == nil {
return
}
if err := r.SyncService.ApplyWAFIPGroups(ctx, groups); err != nil {
r.recordSyncError(err)
slog.Error("agent apply waf ip groups failed", "error", err)
}
}
func (r *Runner) prepareHeartbeatPayload(nodeID string) (protocol.NodePayload, []int64) {
@@ -70,6 +70,8 @@ type fakeSyncService struct {
syncOnceCalls int
lastTarget *protocol.ActiveConfigMeta
onSyncOnceCall func(int)
wafChecksums map[string]string
wafGroups []protocol.WAFIPGroup
}
type fakeRuntimeManager struct {
@@ -135,6 +137,20 @@ func (f *fakeSyncService) ForceSyncOnce(ctx context.Context, target *protocol.Ac
return f.syncOnceErr
}
func (f *fakeSyncService) WAFIPGroupChecksums() (map[string]string, error) {
if f.wafChecksums == nil {
return map[string]string{}, nil
}
return f.wafChecksums, nil
}
func (f *fakeSyncService) ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error {
f.mu.Lock()
defer f.mu.Unlock()
f.wafGroups = append(f.wafGroups, groups...)
return nil
}
type fakeWebSocketConnection struct {
pongCalls int
}