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

fix lint

Browse Source
This commit is contained in:
ryan
2026-06-19 15:13:24 +08:00
parent 0b34792709
commit 32861c5db9
376 changed files with 3648 additions and 19957 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/apps/agent/agent/runner.go
+64 -37
View File
@@ -1,3 +1,4 @@
// Package agent implements the local OpenFlare agent runtime loop.
package agent
import (
@@ -16,12 +17,14 @@ import (
edgeheartbeat "github.com/Rain-kl/Wavelet/internal/apps/edge/heartbeat"
)
// HeartbeatService handles node registration and periodic heartbeat reporting.
type HeartbeatService interface {
Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error)
Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error)
SetToken(token string)
}
// SyncService handles configuration synchronisation between the agent and the server.
type SyncService interface {
SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error
SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
@@ -30,30 +33,36 @@ type SyncService interface {
ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error
}
// RuntimeManager manages the lifecycle and health checks of the OpenResty runtime.
type RuntimeManager interface {
CheckHealth(ctx context.Context) error
Restart(ctx context.Context) error
}
// WebSocketService manages the persistent WebSocket connection to the server.
type WebSocketService interface {
Connect(ctx context.Context) (protocol.WebSocketConnection, error)
SetToken(token string)
URL() string
}
const websocketBackoffDefaultDelay = 30 * time.Second
// Runner coordinates the agent's heartbeat, configuration sync, and WebSocket upgrade lifecycle.
type Runner struct {
Config *config.Config
StateStore *state.Store
HeartbeatCycle *agentheartbeat.Cycle
HeartbeatService HeartbeatService
SyncService SyncService
RuntimeManager RuntimeManager
WebSocketService WebSocketService
Config *config.Config
StateStore *state.Store
HeartbeatCycle *agentheartbeat.Cycle
HeartbeatService HeartbeatService
SyncService SyncService
RuntimeManager RuntimeManager
WebSocketService WebSocketService
restartOpenrestyNow bool
websocketUpgradeEnabled bool
}
// Run starts the agent's main loop, performing heartbeats and upgrading to WebSocket when available.
func (r *Runner) Run(ctx context.Context) error {
if r.HeartbeatCycle != nil {
r.HeartbeatCycle.RecordSyncError = r.recordSyncError
@@ -63,13 +72,7 @@ func (r *Runner) Run(ctx context.Context) error {
return err
}
slog.Info("agent runner started", "node_id", nodeID, "node", r.Config.NodeName, "ip", r.Config.NodeIP)
if r.hasAccessToken() {
if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, true); hbErr != nil {
slog.Error("agent startup heartbeat failed", "error", hbErr)
}
} else if err = r.tryRegister(ctx, &nodeID); err != nil {
slog.Error("agent initial discovery register failed", "error", err)
}
r.runStartupAuth(ctx, &nodeID)
heartbeatTicker := time.NewTicker(r.Config.HeartbeatInterval.Duration())
defer heartbeatTicker.Stop()
@@ -108,42 +111,66 @@ func (r *Runner) Run(ctx context.Context) error {
delay := wsBackoff.Next()
nextWSAttempt = time.Now().Add(delay)
slog.Debug("agent ws disconnected; resuming http heartbeat", "retry_after", delay, "error", wsErr)
if r.hasAccessToken() {
if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil {
slog.Error("agent heartbeat after ws disconnect failed", "error", hbErr)
}
}
r.handleWSDisconnect(ctx, nodeID)
case <-heartbeatTicker.C:
if wsDone != nil {
continue
}
if !r.hasAccessToken() {
if err = r.tryRegister(ctx, &nodeID); err != nil {
slog.Error("agent discovery register failed", "error", err)
}
continue
}
if changed, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil {
slog.Error("agent heartbeat failed", "error", hbErr)
} else {
if changed {
heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration())
}
tryStartWebSocket()
}
r.handleHeartbeatTick(ctx, &nodeID, heartbeatTicker, tryStartWebSocket)
}
}
}
func (r *Runner) runStartupAuth(ctx context.Context, nodeID *string) {
if r.hasAccessToken() {
if _, hbErr := r.performHeartbeatCycle(ctx, *nodeID, true); hbErr != nil {
slog.Error("agent startup heartbeat failed", "error", hbErr)
}
return
}
if err := r.tryRegister(ctx, nodeID); err != nil {
slog.Error("agent initial discovery register failed", "error", err)
}
}
func (r *Runner) handleWSDisconnect(ctx context.Context, nodeID string) {
if !r.hasAccessToken() {
return
}
if _, hbErr := r.performHeartbeatCycle(ctx, nodeID, false); hbErr != nil {
slog.Error("agent heartbeat after ws disconnect failed", "error", hbErr)
}
}
func (r *Runner) handleHeartbeatTick(ctx context.Context, nodeID *string, heartbeatTicker *time.Ticker, tryStartWebSocket func()) {
if !r.hasAccessToken() {
if err := r.tryRegister(ctx, nodeID); err != nil {
slog.Error("agent discovery register failed", "error", err)
}
return
}
changed, hbErr := r.performHeartbeatCycle(ctx, *nodeID, false)
if hbErr != nil {
slog.Error("agent heartbeat failed", "error", hbErr)
return
}
if changed {
heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration())
}
tryStartWebSocket()
}
func (r *Runner) performHeartbeatCycle(ctx context.Context, nodeID string, startup bool) (bool, error) {
r.refreshOpenrestyHealth(ctx)
return r.HeartbeatCycle.Perform(ctx, nodeID, startup, r)
}
// Apply applies the provided agent settings and reports whether the heartbeat interval changed.
func (r *Runner) Apply(settings *protocol.AgentSettings) bool {
return r.applySettings(settings)
}
// RestartOpenrestyIfNeeded restarts OpenResty when a server-requested restart is pending.
func (r *Runner) RestartOpenrestyIfNeeded(ctx context.Context) {
r.tryRestartOpenresty(ctx)
}
@@ -251,7 +278,7 @@ func (r *Runner) runWebSocket(ctx context.Context, nodeID string, conn protocol.
func (r *Runner) sendWebSocketStatus(ctx context.Context, nodeID string, conn protocol.WebSocketConnection) error {
r.refreshOpenrestyHealth(ctx)
payload, ackWindows := r.HeartbeatCycle.PrepareHeartbeatPayload(nodeID)
payload, ackWindows := r.HeartbeatCycle.PrepareHeartbeatPayload(ctx, nodeID)
if err := conn.SendStatus(payload); err != nil {
return err
}
@@ -338,7 +365,7 @@ func newWebSocketBackoff() *webSocketBackoff {
func (backoff *webSocketBackoff) Next() time.Duration {
if backoff == nil || len(backoff.delays) == 0 {
return 30 * time.Second
return websocketBackoffDefaultDelay
}
if backoff.index >= len(backoff.delays) {
return backoff.delays[len(backoff.delays)-1]
@@ -402,7 +429,7 @@ func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error {
return errors.New("agent_token 为空且未配置 discovery_token")
}
slog.Info("agent discovery registration started")
response, err := r.HeartbeatService.Register(ctx, r.HeartbeatCycle.NodePayload(*nodeID))
response, err := r.HeartbeatService.Register(ctx, r.HeartbeatCycle.NodePayload(ctx, *nodeID))
if err != nil {
return err
}
@@ -501,4 +528,4 @@ func (r *Runner) recordOpenrestyUnhealthy(err error, fallbackOnly bool) {
if saveErr := r.StateStore.Save(snapshot); saveErr != nil {
slog.Error("save state after recording openresty error failed", "error", saveErr)
}
}
}
internal/apps/agent/agent/runner_test.go
+2 -2
View File
@@ -395,7 +395,7 @@ func TestRunnerHeartbeatPayloadIncludesObservabilityExtensions(t *testing.T) {
t.Fatalf("failed to prepare access log: %v", err)
}
firstPayload := runner.HeartbeatCycle.NodePayload("node-observe")
firstPayload := runner.HeartbeatCycle.NodePayload(context.Background(), "node-observe")
if firstPayload.Profile == nil {
t.Fatal("expected first heartbeat payload to include system profile")
}
@@ -412,7 +412,7 @@ func TestRunnerHeartbeatPayloadIncludesObservabilityExtensions(t *testing.T) {
t.Fatalf("expected health events for openresty and sync error, got %+v", firstPayload.HealthEvents)
}
secondPayload := runner.HeartbeatCycle.NodePayload("node-observe")
secondPayload := runner.HeartbeatCycle.NodePayload(context.Background(), "node-observe")
if secondPayload.Profile != nil {
t.Fatal("expected unchanged profile to be omitted on subsequent heartbeat")
}
internal/apps/agent/config/config.go
+68 -65
View File
@@ -1,3 +1,4 @@
// Package config loads and persists agent daemon configuration.
package config
import (
@@ -30,8 +31,12 @@ const (
defaultObservabilityReplayMinutes = 15
defaultMMDBUpdateInterval = 24 * time.Hour
defaultMMDBDownloadURL = "https://raw.githubusercontent.com/Loyalsoldier/geoip/release/GeoLite2-Country.mmdb"
defaultHeartbeatInterval = 10 * time.Second
defaultRequestTimeout = 10 * time.Second
configFilePerm = 0o600
)
// Config holds the full runtime configuration for the OpenFlare agent.
type Config struct {
ServerURL string `json:"server_url"`
AccessToken string `json:"agent_token"`
@@ -61,7 +66,7 @@ type Config struct {
StatePath string `json:"state_path"`
HeartbeatInterval MillisecondDuration `json:"heartbeat_interval"`
RequestTimeout MillisecondDuration `json:"request_timeout"`
configPath string
configPath string `json:"-"`
}
type configFile struct {
@@ -93,8 +98,17 @@ type configFile struct {
RequestTimeout MillisecondDuration `json:"request_timeout"`
}
func transferPersistedConfig(dst, src any) error {
data, err := json.Marshal(src)
if err != nil {
return err
}
return json.Unmarshal(data, dst)
}
// Load reads and parses the agent configuration file at the given path.
func Load(path string) (*Config, error) {
data, err := os.ReadFile(path)
data, err := os.ReadFile(path) //nolint:gosec // path is the configured agent config location
if err != nil && !os.IsNotExist(err) {
return nil, err
}
@@ -107,33 +121,11 @@ func Load(path string) (*Config, error) {
if err != nil && !hasEnvConfig() {
return nil, err
}
cfg := &Config{
ServerURL: file.ServerURL,
AccessToken: file.AccessToken,
DiscoveryToken: file.DiscoveryToken,
NodeName: file.NodeName,
NodeIP: file.NodeIP,
OpenrestyPath: file.OpenrestyPath,
OpenrestyResolvers: append([]string{}, file.OpenrestyResolvers...),
DataDir: file.DataDir,
MainConfigPath: file.MainConfigPath,
RouteConfigPath: file.RouteConfigPath,
AccessLogPath: file.AccessLogPath,
CertDir: file.CertDir,
OpenrestyCertDir: file.OpenrestyCertDir,
LuaDir: file.LuaDir,
OpenrestyLuaDir: file.OpenrestyLuaDir,
RuntimeConfigDir: file.RuntimeConfigDir,
PagesDir: file.PagesDir,
MMDBPath: file.MMDBPath,
MMDBUpdateInterval: file.MMDBUpdateInterval,
MMDBDownloadURL: file.MMDBDownloadURL,
OpenrestyObservabilityPort: file.OpenrestyObservabilityPort,
ObservabilityBufferPath: file.ObservabilityBufferPath,
ObservabilityReplayMinutes: file.ObservabilityReplayMinutes,
StatePath: file.StatePath,
HeartbeatInterval: file.HeartbeatInterval,
RequestTimeout: file.RequestTimeout,
cfg := &Config{}
if err == nil {
if err = transferPersistedConfig(cfg, file); err != nil {
return nil, err
}
}
cfg.configPath = path
applyEnvOverrides(cfg)
@@ -148,51 +140,58 @@ func applyDefaults(cfg *Config, baseDir string) {
baseDir = filepath.Clean(baseDir)
cfg.Version = Version
cfg.OpenrestyResolvers = utils.UniqueAndCleanStringSlice(cfg.OpenrestyResolvers)
applyAgentIdentityDefaults(cfg)
applyAgentPathDefaults(cfg, baseDir)
applyAgentTimingDefaults(cfg)
normalizeManagedPaths(cfg)
}
func applyAgentIdentityDefaults(cfg *Config) {
if cfg.OpenrestyPath == "" {
cfg.OpenrestyPath = "openresty"
}
if cfg.DataDir == "" {
cfg.DataDir = filepath.Join(baseDir, "data")
}
if cfg.NodeName == "" {
cfg.NodeName = detectHostname()
}
if cfg.NodeIP == "" {
cfg.NodeIP = nodeip.Detect()
}
if cfg.MainConfigPath == "" {
cfg.MainConfigPath = joinManagedPath(cfg.DataDir, defaultMainConfigRelativePath)
}
func applyAgentPathDefaults(cfg *Config, baseDir string) {
if cfg.DataDir == "" {
cfg.DataDir = filepath.Join(baseDir, "data")
}
if cfg.RouteConfigPath == "" {
cfg.RouteConfigPath = joinManagedPath(cfg.DataDir, defaultRouteConfigRelativePath)
type managedPathDefault struct {
target *string
relative string
}
if cfg.AccessLogPath == "" {
cfg.AccessLogPath = joinManagedPath(cfg.DataDir, defaultAccessLogRelativePath)
pathDefaults := []managedPathDefault{
{&cfg.MainConfigPath, defaultMainConfigRelativePath},
{&cfg.RouteConfigPath, defaultRouteConfigRelativePath},
{&cfg.AccessLogPath, defaultAccessLogRelativePath},
{&cfg.StatePath, defaultStateRelativePath},
{&cfg.CertDir, defaultCertDirRelativePath},
{&cfg.LuaDir, defaultLuaDirRelativePath},
{&cfg.RuntimeConfigDir, defaultRuntimeConfigDirRelativePath},
{&cfg.PagesDir, defaultPagesDirRelativePath},
{&cfg.MMDBPath, defaultMMDBRelativePath},
{&cfg.ObservabilityBufferPath, defaultObservabilityBufferRelativePath},
}
if cfg.StatePath == "" {
cfg.StatePath = joinManagedPath(cfg.DataDir, defaultStateRelativePath)
}
if cfg.CertDir == "" {
cfg.CertDir = joinManagedPath(cfg.DataDir, defaultCertDirRelativePath)
for _, item := range pathDefaults {
if strings.TrimSpace(*item.target) == "" {
*item.target = joinManagedPath(cfg.DataDir, item.relative)
}
}
if cfg.OpenrestyCertDir == "" {
cfg.OpenrestyCertDir = cfg.CertDir
}
if cfg.LuaDir == "" {
cfg.LuaDir = joinManagedPath(cfg.DataDir, defaultLuaDirRelativePath)
}
if cfg.OpenrestyLuaDir == "" {
cfg.OpenrestyLuaDir = cfg.LuaDir
}
if cfg.RuntimeConfigDir == "" {
cfg.RuntimeConfigDir = joinManagedPath(cfg.DataDir, defaultRuntimeConfigDirRelativePath)
}
if cfg.PagesDir == "" {
cfg.PagesDir = joinManagedPath(cfg.DataDir, defaultPagesDirRelativePath)
}
if cfg.MMDBPath == "" {
cfg.MMDBPath = joinManagedPath(cfg.DataDir, defaultMMDBRelativePath)
}
}
func applyAgentTimingDefaults(cfg *Config) {
if cfg.MMDBUpdateInterval <= 0 {
cfg.MMDBUpdateInterval = MillisecondDuration(defaultMMDBUpdateInterval)
}
@@ -202,19 +201,15 @@ func applyDefaults(cfg *Config, baseDir string) {
if cfg.OpenrestyObservabilityPort <= 0 {
cfg.OpenrestyObservabilityPort = defaultOpenRestyObservabilityPort
}
if cfg.ObservabilityBufferPath == "" {
cfg.ObservabilityBufferPath = joinManagedPath(cfg.DataDir, defaultObservabilityBufferRelativePath)
}
if cfg.ObservabilityReplayMinutes <= 0 {
cfg.ObservabilityReplayMinutes = defaultObservabilityReplayMinutes
}
if cfg.HeartbeatInterval <= 0 {
cfg.HeartbeatInterval = MillisecondDuration(10 * time.Second)
cfg.HeartbeatInterval = MillisecondDuration(defaultHeartbeatInterval)
}
if cfg.RequestTimeout <= 0 {
cfg.RequestTimeout = MillisecondDuration(10 * time.Second)
cfg.RequestTimeout = MillisecondDuration(defaultRequestTimeout)
}
normalizeManagedPaths(cfg)
}
func normalizeManagedPaths(cfg *Config) {
@@ -360,6 +355,7 @@ func validate(cfg *Config) error {
return nil
}
// InitialAuthToken returns the agent access token, falling back to the discovery token if absent.
func (cfg *Config) InitialAuthToken() string {
if cfg == nil {
return ""
@@ -370,6 +366,15 @@ func (cfg *Config) InitialAuthToken() string {
return strings.TrimSpace(cfg.DiscoveryToken)
}
func (cfg *Config) toConfigFile() configFile {
var file configFile
if err := transferPersistedConfig(&file, cfg); err != nil {
return configFile{}
}
return file
}
// Save persists the current configuration back to its original file path.
func (cfg *Config) Save() error {
if cfg == nil {
return errors.New("config 不能为空")
@@ -377,11 +382,11 @@ func (cfg *Config) Save() error {
if cfg.configPath == "" {
return errors.New("config path 未初始化")
}
data, err := json.MarshalIndent(cfg, "", " ")
data, err := json.MarshalIndent(cfg.toConfigFile(), "", " ") //nolint:gosec // agent token must be persisted in local config file
if err != nil {
return err
}
return os.WriteFile(cfg.configPath, data, 0o644)
return os.WriteFile(cfg.configPath, data, configFilePerm)
}
func detectHostname() string {
@@ -391,5 +396,3 @@ func detectHostname() string {
}
return strings.TrimSpace(host)
}
internal/apps/agent/config/duration.go
+2 -1
View File
@@ -2,4 +2,5 @@ package config
import edgeconfig "github.com/Rain-kl/Wavelet/internal/apps/edge/config"
type MillisecondDuration = edgeconfig.MillisecondDuration
// MillisecondDuration is an alias for the edge config millisecond-precision duration type.
type MillisecondDuration = edgeconfig.MillisecondDuration
internal/apps/agent/config/version.go
+1
View File
@@ -1,3 +1,4 @@
package config
// Version is the current agent version string, overridden at build time.
var Version = "dev"
internal/apps/agent/geoipdata/data.go
+4
View File
@@ -1,8 +1,12 @@
// Package geoipdata embeds the default MaxMind GeoLite2 country database.
package geoipdata
import "embed"
// FS holds the embedded GeoLite2-Country.mmdb database.
//
//go:embed GeoLite2-Country.mmdb
var FS embed.FS
// DefaultMMDBName is the filename of the embedded MaxMind country database.
const DefaultMMDBName = "GeoLite2-Country.mmdb"
internal/apps/agent/geoipupdate/updater.go
+13 -3
View File
@@ -1,3 +1,4 @@
// Package geoipupdate schedules local MaxMind GeoIP database updates for the agent.
package geoipupdate
import (
@@ -13,12 +14,20 @@ import (
"github.com/Rain-kl/Wavelet/pkg/geoip"
)
const (
mmdbDirPerm = 0o750
mmdbFilePerm = 0o600
)
// Updater periodically downloads a fresh GeoIP MMDB file and seeds the
// initial embedded database when none is present on disk.
type Updater struct {
MMDBPath string
DownloadURL string
UpdateInterval time.Duration
}
// EnsureInitialDatabase seeds the MMDB file from the embedded database if it does not exist on disk.
func (u *Updater) EnsureInitialDatabase() error {
path := filepath.Clean(u.MMDBPath)
if path == "" || path == "." {
@@ -33,16 +42,17 @@ func (u *Updater) EnsureInitialDatabase() error {
if err != nil {
return fmt.Errorf("read embedded mmdb failed: %w", err)
}
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(path), mmdbDirPerm); err != nil {
return fmt.Errorf("create mmdb directory failed: %w", err)
}
if err := os.WriteFile(path, data, 0o644); err != nil {
if err := os.WriteFile(path, data, mmdbFilePerm); err != nil {
return fmt.Errorf("write initial mmdb failed: %w", err)
}
slog.Info("initialized GeoIP mmdb from embedded database", "path", path, "size", len(data))
return nil
}
// Run starts the periodic GeoIP update loop and blocks until ctx is cancelled.
func (u *Updater) Run(ctx context.Context) {
if u == nil || u.MMDBPath == "" || u.UpdateInterval <= 0 {
return
@@ -57,7 +67,7 @@ func (u *Updater) Run(ctx context.Context) {
case <-ctx.Done():
return
case <-ticker.C:
if err := geoip.DownloadMaxMindDatabase(u.MMDBPath, u.DownloadURL); err != nil {
if err := geoip.DownloadMaxMindDatabase(ctx, u.MMDBPath, u.DownloadURL); err != nil {
slog.Warn("update GeoIP mmdb failed", "path", u.MMDBPath, "error", err)
continue
}
internal/apps/agent/heartbeat/cycle.go
+18 -11
View File
@@ -1,3 +1,5 @@
// Package heartbeat implements the periodic heartbeat cycle executed by the agent,
// including payload preparation, config sync, WAF IP group application, and observability buffering.
package heartbeat
import (
@@ -14,10 +16,7 @@ import (
edgeheartbeat "github.com/Rain-kl/Wavelet/internal/apps/edge/heartbeat"
)
type HeartbeatClient interface {
Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error)
}
// SyncService is the interface used by Cycle to sync active configuration and WAF IP groups.
type SyncService interface {
SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error
SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error
@@ -25,23 +24,26 @@ type SyncService interface {
ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error
}
// SettingsApplier is the interface used by Cycle to apply agent settings received from the server.
type SettingsApplier interface {
Apply(settings *protocol.AgentSettings) (intervalChanged bool)
RestartOpenrestyIfNeeded(ctx context.Context)
}
// Cycle holds the dependencies required to execute a single agent heartbeat cycle.
type Cycle struct {
Config *config.Config
StateStore *state.Store
ObservabilityBuffer *state.ObservabilityBufferStore
Heartbeat HeartbeatClient
Heartbeat API
Sync SyncService
Updater *updater.Service
RecordSyncError func(err error)
}
// Perform executes one complete heartbeat cycle: sends the heartbeat, syncs config, and applies settings.
func (c *Cycle) Perform(ctx context.Context, nodeID string, startup bool, settings SettingsApplier) (bool, error) {
payload, ackWindows := c.PrepareHeartbeatPayload(nodeID)
payload, ackWindows := c.PrepareHeartbeatPayload(ctx, nodeID)
heartbeatResult, err := c.Heartbeat.Heartbeat(ctx, payload)
if err != nil {
return false, err
@@ -79,14 +81,15 @@ func (c *Cycle) Perform(ctx context.Context, nodeID string, startup bool, settin
return changed, nil
}
func (c *Cycle) NodePayload(nodeID string) protocol.NodePayload {
// NodePayload builds and returns the full NodePayload to be sent in a heartbeat request.
func (c *Cycle) NodePayload(ctx context.Context, nodeID string) protocol.NodePayload {
snapshot, _ := c.StateStore.Load()
openrestyStatus := strings.TrimSpace(snapshot.OpenrestyStatus)
if openrestyStatus == "" {
openrestyStatus = protocol.OpenrestyStatusUnknown
}
profile := observability.BuildProfile(c.Config, c.StateStore)
managedOpenRestyMetrics := observability.CollectManagedOpenRestyMetrics(c.Config)
managedOpenRestyMetrics := observability.CollectManagedOpenRestyMetrics(ctx, c.Config)
trafficReport, accessLogs, fallbackMetrics := observability.BuildTrafficObservability(c.Config, c.StateStore, managedOpenRestyMetrics)
if managedOpenRestyMetrics == nil {
managedOpenRestyMetrics = fallbackMetrics
@@ -122,8 +125,9 @@ func (c *Cycle) NodePayload(nodeID string) protocol.NodePayload {
return payload
}
func (c *Cycle) PrepareHeartbeatPayload(nodeID string) (protocol.NodePayload, []int64) {
payload := c.NodePayload(nodeID)
// PrepareHeartbeatPayload constructs the heartbeat payload with buffered observability records and returns the window timestamps to acknowledge.
func (c *Cycle) PrepareHeartbeatPayload(ctx context.Context, nodeID string) (protocol.NodePayload, []int64) {
payload := c.NodePayload(ctx, nodeID)
if c.ObservabilityBuffer == nil || (payload.Snapshot == nil && payload.TrafficReport == nil && len(payload.AccessLogs) == 0) {
return payload, nil
}
@@ -173,6 +177,7 @@ func (c *Cycle) PrepareHeartbeatPayload(nodeID string) (protocol.NodePayload, []
return payload, ackWindows
}
// AckObservabilityWindows acknowledges the given observability window timestamps in the buffer store.
func (c *Cycle) AckObservabilityWindows(windowStartedAtUnix []int64) {
if c.ObservabilityBuffer == nil || len(windowStartedAtUnix) == 0 {
return
@@ -183,6 +188,7 @@ func (c *Cycle) AckObservabilityWindows(windowStartedAtUnix []int64) {
}
}
// ApplyWAFIPGroups applies the WAF IP groups received from the server via the SyncService.
func (c *Cycle) ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) {
if len(groups) == 0 || c.Sync == nil {
return
@@ -199,6 +205,7 @@ func (c *Cycle) recordSyncError(err error) {
}
}
// AgentSettingsToAutoUpdate converts AgentSettings to an AutoUpdateSettings value used by the edge heartbeat updater.
func AgentSettingsToAutoUpdate(settings *protocol.AgentSettings) *edgeheartbeat.AutoUpdateSettings {
if settings == nil {
return nil
@@ -214,4 +221,4 @@ func AgentSettingsToAutoUpdate(settings *protocol.AgentSettings) *edgeheartbeat.
func agentSettingsToAutoUpdate(settings *protocol.AgentSettings) *edgeheartbeat.AutoUpdateSettings {
return AgentSettingsToAutoUpdate(settings)
}
}
internal/apps/agent/heartbeat/service.go
+17 -4
View File
@@ -6,28 +6,41 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
)
type Client interface {
// RemoteClient is the interface that abstracts the remote API calls performed by Service.
type RemoteClient interface {
RegisterNode(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error)
Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error)
SetToken(token string)
}
type Service struct {
client Client
// API abstracts registration and heartbeat operations used by Cycle.
type API interface {
Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error)
Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error)
SetToken(token string)
}
func New(client Client) *Service {
// Service wraps a RemoteClient to expose agent registration and heartbeat operations.
type Service struct {
client RemoteClient
}
// New creates a new Service backed by the given RemoteClient.
func New(client RemoteClient) *Service {
return &Service{client: client}
}
// Register sends a node registration request to the server.
func (s *Service) Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) {
return s.client.RegisterNode(ctx, payload)
}
// Heartbeat sends a heartbeat to the server using the service client.
func (s *Service) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error) {
return s.client.Heartbeat(ctx, payload)
}
// SetToken sets the authentication token for the service client.
func (s *Service) SetToken(token string) {
s.client.SetToken(token)
}
internal/apps/agent/httpclient/client.go
+13 -3
View File
@@ -1,3 +1,4 @@
// Package httpclient provides an authenticated HTTP client for the agent.
package httpclient
import (
@@ -8,20 +9,23 @@ import (
"net/http"
"time"
edgehttp "github.com/Rain-kl/Wavelet/internal/apps/edge/httpclient"
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
edgehttp "github.com/Rain-kl/Wavelet/internal/apps/edge/httpclient"
)
// Client is a HTTP client used by the agent to communicate with the control plane server.
type Client struct {
base *edgehttp.Client
}
// New creates a new Client instance with the specified base URL, token, and timeout.
func New(baseURL string, token string, timeout time.Duration) *Client {
return &Client{
base: edgehttp.New(baseURL, token, timeout, "X-Agent-Token"),
}
}
// RegisterNode registers the agent node with the control plane server.
func (c *Client) RegisterNode(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error) {
resp := protocol.APIResponse[protocol.RegisterNodeResponse]{}
if err := c.base.PostJSON(ctx, "/api/v1/agent/nodes/register", payload, &resp); err != nil {
@@ -33,6 +37,7 @@ func (c *Client) RegisterNode(ctx context.Context, payload protocol.NodePayload)
return &resp.Data, nil
}
// Heartbeat sends a heartbeat payload to the control plane and returns the response result.
func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.HeartbeatResult, error) {
resp := protocol.APIResponse[protocol.HeartbeatData]{}
if err := c.base.PostJSON(ctx, "/api/v1/agent/nodes/heartbeat", payload, &resp); err != nil {
@@ -48,6 +53,7 @@ func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*
}, nil
}
// GetActiveConfig retrieves the current active configuration from the control plane server.
func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) {
resp := protocol.APIResponse[protocol.ActiveConfigResponse]{}
if err := c.base.GetJSON(ctx, "/api/v1/agent/config-versions/active", &resp); err != nil {
@@ -59,6 +65,7 @@ func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigRes
return &resp.Data, nil
}
// ReportApplyLog reports the configuration application logs back to the control plane.
func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error {
resp := protocol.APIResponse[json.RawMessage]{}
if err := c.base.PostJSON(ctx, "/api/v1/agent/apply-logs", payload, &resp); err != nil {
@@ -67,6 +74,7 @@ func (c *Client) ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPa
return edgehttp.APIError(resp.ErrorMsg)
}
// SyncWAFIPGroups synchronizes WAF IP groups with the control plane server.
func (c *Client) SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGroupSyncRequest) (*protocol.WAFIPGroupSyncResponse, error) {
resp := protocol.APIResponse[protocol.WAFIPGroupSyncResponse]{}
if err := c.base.PostJSON(ctx, "/api/v1/agent/waf/ip-groups/sync", payload, &resp); err != nil {
@@ -78,18 +86,20 @@ func (c *Client) SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGrou
return &resp.Data, nil
}
// DownloadPagesDeploymentPackage downloads the deployment package for the given Pages deployment ID.
func (c *Client) DownloadPagesDeploymentPackage(ctx context.Context, deploymentID uint) ([]byte, error) {
res, err := c.base.DoRaw(ctx, http.MethodGet, fmt.Sprintf("/api/v1/agent/pages/deployments/%d/package", deploymentID), nil)
if err != nil {
return nil, err
}
defer res.Body.Close()
defer func() { _ = res.Body.Close() }()
if res.StatusCode != http.StatusOK {
return nil, edgehttp.ReadHTTPError(res)
}
return io.ReadAll(res.Body)
}
// SetToken updates the authentication token used for API requests.
func (c *Client) SetToken(token string) {
c.base.SetToken(token)
}
}
internal/apps/agent/logging/logger.go
+3 -1
View File
@@ -1,7 +1,9 @@
// Package logging configures structured logging for the agent process.
package logging
import edgelogging "github.com/Rain-kl/Wavelet/internal/apps/edge/logging"
// Setup initialises structured logging for the agent process.
func Setup() {
edgelogging.Setup(edgelogging.Options{AddSource: true})
}
}
internal/apps/agent/nginx/manager.go
+98 -52
View File
@@ -1,3 +1,4 @@
// Package nginx manages OpenResty configuration, runtime, and supporting assets.
package nginx
import (
@@ -26,10 +27,26 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
)
// RuntimeConfigDirPlaceholder is substituted into generated configs at apply time.
const RuntimeConfigDirPlaceholder = "__OPENFLARE_RUNTIME_CONFIG_DIR__"
const ResolverDirectivePlaceholder = "__OPENFLARE_RESOLVER_DIRECTIVE__"
const WAFIPGroupsConfigFileName = "waf_ip_groups.json"
// ResolverDirectivePlaceholder is substituted into generated configs at apply time.
const ResolverDirectivePlaceholder = "__OPENFLARE_RESOLVER_DIRECTIVE__"
// WAFIPGroupsConfigFileName is the runtime filename for synced WAF IP group data.
const WAFIPGroupsConfigFileName = "waf_ip_groups.json"
const powConfigFileName = "pow_config.json"
const (
nginxConfigFilePerm = 0o644
nginxPrivateKeyFilePerm = 0o600
nginxDirPerm = 0o755
stubStatusCheckTimeout = 1500 * time.Millisecond
nginxVersionSubmatchCount = 2
resolverAddressCapacity = 2
)
// Executor controls OpenResty validation, reload, health, and lifecycle operations.
type Executor interface {
Test(ctx context.Context) error
Reload(ctx context.Context) error
@@ -38,44 +55,49 @@ type Executor interface {
Restart(ctx context.Context) error
}
// CommandRunner executes external commands on behalf of an Executor.
type CommandRunner interface {
Run(ctx context.Context, name string, args ...string) ([]byte, error)
}
// OSCommandRunner runs commands using the host operating system.
type OSCommandRunner struct{}
// Run executes the named command and returns its combined output.
func (r *OSCommandRunner) Run(ctx context.Context, name string, args ...string) ([]byte, error) {
slog.Debug("OSCommandRunner starting command", "name", name, "args", args)
tmpFile, err := os.CreateTemp("", "openflare-cmd-*")
if err != nil {
slog.Error("OSCommandRunner failed to create temp file, falling back to CombinedOutput", "error", err)
cmd := exec.CommandContext(ctx, name, args...)
cmd := exec.CommandContext(ctx, name, args...) //nolint:gosec // command name and args come from trusted OpenResty management paths
output, outErr := cmd.CombinedOutput()
slog.Debug("OSCommandRunner finished CombinedOutput", "name", name, "error", outErr)
return output, outErr
}
defer os.Remove(tmpFile.Name())
defer func() { _ = os.Remove(tmpFile.Name()) }()
cmd := exec.CommandContext(ctx, name, args...)
cmd := exec.CommandContext(ctx, name, args...) //nolint:gosec // command name and args come from trusted OpenResty management paths
cmd.Stdout = tmpFile
cmd.Stderr = tmpFile
slog.Debug("OSCommandRunner executing cmd.Run()", "name", name)
runErr := cmd.Run()
slog.Debug("OSCommandRunner cmd.Run() returned", "name", name, "error", runErr)
tmpFile.Close()
_ = tmpFile.Close()
output, _ := os.ReadFile(tmpFile.Name())
slog.Debug("OSCommandRunner command complete", "name", name, "output_len", len(output))
return output, runErr
}
// PathExecutor runs OpenResty using a configured binary and config path.
type PathExecutor struct {
Path string
ConfigPath string
Runner CommandRunner
}
// Test validates the current OpenResty configuration.
func (e *PathExecutor) Test(ctx context.Context) error {
slog.Debug("running openresty test with binary", "path", e.Path, "config", e.ConfigPath)
output, err := e.Runner.Run(ctx, e.Path, "-t", "-c", e.ConfigPath)
@@ -86,6 +108,7 @@ func (e *PathExecutor) Test(ctx context.Context) error {
return nil
}
// Reload reloads OpenResty or starts it when no runtime process is running.
func (e *PathExecutor) Reload(ctx context.Context) error {
slog.Debug("running openresty reload with binary", "path", e.Path, "config", e.ConfigPath)
output, err := e.Runner.Run(ctx, e.Path, "-s", "reload", "-c", e.ConfigPath)
@@ -104,6 +127,7 @@ func (e *PathExecutor) Reload(ctx context.Context) error {
return nil
}
// EnsureRuntime validates configuration and reloads the OpenResty runtime.
func (e *PathExecutor) EnsureRuntime(ctx context.Context, _ bool) error {
if err := e.Test(ctx); err != nil {
return err
@@ -111,10 +135,12 @@ func (e *PathExecutor) EnsureRuntime(ctx context.Context, _ bool) error {
return e.Reload(ctx)
}
// CheckHealth reports whether the OpenResty configuration is valid.
func (e *PathExecutor) CheckHealth(ctx context.Context) error {
return e.Test(ctx)
}
// Restart stops and starts the OpenResty runtime process.
func (e *PathExecutor) Restart(ctx context.Context) error {
slog.Info("restarting openresty with binary", "path", e.Path, "config", e.ConfigPath)
output, err := e.Runner.Run(ctx, e.Path, "-s", "quit", "-c", e.ConfigPath)
@@ -132,6 +158,7 @@ func (e *PathExecutor) Restart(ctx context.Context) error {
return nil
}
// Manager applies OpenResty configuration and manages runtime assets.
type Manager struct {
MainConfigPath string
RouteConfigPath string
@@ -148,9 +175,12 @@ type Manager struct {
Executor Executor
}
// ApplyStatus reports the outcome of an OpenResty configuration apply.
type ApplyStatus string
// Apply outcome status values.
const (
// ApplyStatusSuccess indicates the configuration was applied successfully.
ApplyStatusSuccess ApplyStatus = "success"
ApplyStatusWarning ApplyStatus = "warning"
ApplyStatusFatal ApplyStatus = "fatal"
@@ -188,6 +218,7 @@ const safeDefaultFallbackObservabilityServerBlock = `
}
`
// ApplyOutcome contains the status and message from a configuration apply.
type ApplyOutcome struct {
Status ApplyStatus
Message string
@@ -197,6 +228,7 @@ type wafIPGroupsRuntimeConfig struct {
Groups map[string]protocol.WAFIPGroup `json:"groups"`
}
// Apply writes, validates, and activates new OpenResty configuration files.
func (m *Manager) Apply(ctx context.Context, mainConfig string, routeConfig string, supportFiles []protocol.SupportFile) ApplyOutcome {
slog.Info("openresty apply started", "main_config", m.MainConfigPath, "route_config", m.RouteConfigPath, "cert_files", len(supportFiles))
backup, err := m.backup()
@@ -236,11 +268,11 @@ func (m *Manager) writeTargetFiles(mainConfig string, routeConfig string, suppor
slog.Warn("runtime-resolved hostname upstreams detected without available resolvers; hostname origin requests may fail until resolvers are configured")
}
renderedMainConfig := m.renderMainConfig(mainConfig)
if err := os.WriteFile(m.MainConfigPath, []byte(renderedMainConfig), 0o644); err != nil {
if err := os.WriteFile(m.MainConfigPath, []byte(renderedMainConfig), nginxConfigFilePerm); err != nil {
return err
}
renderedRouteConfig := m.renderRouteConfig(routeConfig)
if err := os.WriteFile(m.RouteConfigPath, []byte(renderedRouteConfig), 0o644); err != nil {
if err := os.WriteFile(m.RouteConfigPath, []byte(renderedRouteConfig), nginxConfigFilePerm); err != nil {
return err
}
return nil
@@ -293,6 +325,7 @@ func fatalApplyOutcome(err error) ApplyOutcome {
}
}
// EnsureLuaAssets synchronizes managed Lua and static assets to the runtime directory.
func (m *Manager) EnsureLuaAssets() error {
if strings.TrimSpace(m.LuaDir) == "" {
return nil
@@ -317,12 +350,13 @@ func (m *Manager) EnsureLuaAssets() error {
files = append(files, managedFile{
Path: filepath.ToSlash(relativePath),
Content: []byte(file.Content),
Mode: 0o644,
Mode: nginxConfigFilePerm,
})
}
return syncManagedFiles(m.LuaDir, files)
}
// EnsureRuntime validates and reloads the current OpenResty runtime configuration.
func (m *Manager) EnsureRuntime(ctx context.Context, recreate bool) error {
if m.Executor == nil {
return errors.New("executor 未配置")
@@ -331,6 +365,7 @@ func (m *Manager) EnsureRuntime(ctx context.Context, recreate bool) error {
return m.Executor.EnsureRuntime(ctx, recreate)
}
// EnsureSafeFallbackRuntime starts a minimal safe default OpenResty runtime.
func (m *Manager) EnsureSafeFallbackRuntime(ctx context.Context, reason string) error {
if m.Executor == nil {
return errors.New("executor 未配置")
@@ -350,6 +385,7 @@ func (m *Manager) EnsureSafeFallbackRuntime(ctx context.Context, reason string)
return nil
}
// CheckHealth verifies that OpenResty configuration and health endpoints are available.
func (m *Manager) CheckHealth(ctx context.Context) error {
if m.Executor == nil {
return errors.New("executor 未配置")
@@ -365,6 +401,7 @@ func (m *Manager) CheckHealth(ctx context.Context) error {
return m.checkStubStatus(ctx)
}
// Restart restarts the OpenResty runtime process.
func (m *Manager) Restart(ctx context.Context) error {
if m.Executor == nil {
return errors.New("executor 未配置")
@@ -373,6 +410,7 @@ func (m *Manager) Restart(ctx context.Context) error {
return m.Executor.Restart(ctx)
}
// CurrentChecksum returns a stable checksum for the active OpenResty configuration bundle.
func (m *Manager) CurrentChecksum() (string, error) {
if m.RouteConfigPath == "" {
return "", errors.New("route config path 不能为空")
@@ -435,6 +473,7 @@ func (m *Manager) CurrentChecksum() (string, error) {
return result, nil
}
// WAFIPGroupChecksums returns checksums for locally synced WAF IP groups.
func (m *Manager) WAFIPGroupChecksums() (map[string]string, error) {
config, err := m.readWAFIPGroupsRuntimeConfig()
if err != nil {
@@ -449,6 +488,7 @@ func (m *Manager) WAFIPGroupChecksums() (map[string]string, error) {
return result, nil
}
// SyncWAFIPGroups writes WAF IP group definitions to the runtime config directory.
func (m *Manager) SyncWAFIPGroups(groups []protocol.WAFIPGroup) error {
if m.RuntimeConfigDir == "" || len(groups) == 0 {
return nil
@@ -470,11 +510,11 @@ func (m *Manager) SyncWAFIPGroups(groups []protocol.WAFIPGroup) error {
if err != nil {
return err
}
if err := os.MkdirAll(m.RuntimeConfigDir, 0o755); err != nil {
if err := os.MkdirAll(m.RuntimeConfigDir, nginxDirPerm); err != nil {
return err
}
path := filepath.Join(m.RuntimeConfigDir, WAFIPGroupsConfigFileName)
if err := os.WriteFile(path, data, 0o644); err != nil {
if err := os.WriteFile(path, data, nginxConfigFilePerm); err != nil {
return fmt.Errorf("write %s: %w", WAFIPGroupsConfigFileName, err)
}
slog.Info("synced waf ip groups", "path", path, "group_count", len(groups))
@@ -487,7 +527,7 @@ func (m *Manager) readWAFIPGroupsRuntimeConfig() (*wafIPGroupsRuntimeConfig, err
return config, nil
}
path := filepath.Join(m.RuntimeConfigDir, WAFIPGroupsConfigFileName)
data, err := os.ReadFile(path)
data, err := os.ReadFile(path) //nolint:gosec // path is under managed RuntimeConfigDir
if err != nil {
if os.IsNotExist(err) {
return config, nil
@@ -506,6 +546,7 @@ func (m *Manager) readWAFIPGroupsRuntimeConfig() (*wafIPGroupsRuntimeConfig, err
return config, nil
}
// ExecutorOptions configures construction of an OpenResty Executor.
type ExecutorOptions struct {
NginxPath string
MainConfigPath string
@@ -517,6 +558,7 @@ type ExecutorOptions struct {
OpenrestyObservabilityPort int
}
// NewExecutor creates an Executor backed by a configured OpenResty binary.
func NewExecutor(options ExecutorOptions) Executor {
runner := &OSCommandRunner{}
return &PathExecutor{
@@ -526,6 +568,7 @@ func NewExecutor(options ExecutorOptions) Executor {
}
}
// DetectVersion returns the OpenResty version reported by the configured binary.
func DetectVersion(ctx context.Context, options ExecutorOptions) string {
version, err := detectVersion(ctx, options, &OSCommandRunner{})
if err != nil {
@@ -556,7 +599,7 @@ func detectVersion(ctx context.Context, options ExecutorOptions, runner CommandR
func parseExtVersion(output string) string {
matches := nginxVersionPattern.FindStringSubmatch(output)
if len(matches) != 2 {
if len(matches) != nginxVersionSubmatchCount {
return ""
}
return matches[1]
@@ -606,24 +649,24 @@ func (m *Manager) backup() (*backupState, error) {
if m.RouteConfigPath == "" {
return nil, errors.New("route config path 不能为空")
}
if err := os.MkdirAll(filepath.Dir(m.MainConfigPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(m.MainConfigPath), nginxDirPerm); err != nil {
return nil, err
}
if err := os.MkdirAll(filepath.Dir(m.RouteConfigPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(m.RouteConfigPath), nginxDirPerm); err != nil {
return nil, err
}
if m.AccessLogPath != "" {
if err := os.MkdirAll(filepath.Dir(m.AccessLogPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(m.AccessLogPath), nginxDirPerm); err != nil {
return nil, err
}
}
if m.CertDir != "" {
if err := os.MkdirAll(m.CertDir, 0o755); err != nil {
if err := os.MkdirAll(m.CertDir, nginxDirPerm); err != nil {
return nil, err
}
}
if m.RuntimeConfigDir != "" {
if err := os.MkdirAll(m.RuntimeConfigDir, 0o755); err != nil {
if err := os.MkdirAll(m.RuntimeConfigDir, nginxDirPerm); err != nil {
return nil, err
}
}
@@ -672,14 +715,14 @@ func (m *Manager) restore(state *backupState) error {
}
slog.Warn("restoring nginx backup", "main_existed", state.MainExisted, "route_existed", state.RouteExisted, "cert_files", len(state.Files))
if state.MainExisted {
if err := os.WriteFile(m.MainConfigPath, state.MainData, 0o644); err != nil {
if err := os.WriteFile(m.MainConfigPath, state.MainData, nginxConfigFilePerm); err != nil {
return err
}
} else if err := os.Remove(m.MainConfigPath); err != nil && !os.IsNotExist(err) {
return err
}
if state.RouteExisted {
if err := os.WriteFile(m.RouteConfigPath, state.RouteData, 0o644); err != nil {
if err := os.WriteFile(m.RouteConfigPath, state.RouteData, nginxConfigFilePerm); err != nil {
return err
}
} else if err := os.Remove(m.RouteConfigPath); err != nil && !os.IsNotExist(err) {
@@ -690,7 +733,7 @@ func (m *Manager) restore(state *backupState) error {
return err
}
}
if err := m.restoreRuntimeConfig(state.PowConfig, "pow_config.json"); err != nil {
if err := m.restoreRuntimeConfig(state.PowConfig, powConfigFileName); err != nil {
return err
}
if err := m.restoreRuntimeConfig(state.WAFConfig, "waf_config.json"); err != nil {
@@ -710,10 +753,10 @@ func (m *Manager) writePowConfig(supportFiles []protocol.SupportFile) error {
if m.RuntimeConfigDir == "" {
return nil
}
configPath := filepath.Join(m.RuntimeConfigDir, "pow_config.json")
configPath := filepath.Join(m.RuntimeConfigDir, powConfigFileName)
for _, file := range supportFiles {
if file.Path == "pow_config.json" {
if err := os.WriteFile(configPath, []byte(file.Content), 0o644); err != nil {
if file.Path == powConfigFileName {
if err := os.WriteFile(configPath, []byte(file.Content), nginxConfigFilePerm); err != nil {
return fmt.Errorf("write pow_config.json: %w", err)
}
slog.Info("wrote pow config", "path", configPath, "size", len(file.Content))
@@ -723,10 +766,10 @@ func (m *Manager) writePowConfig(supportFiles []protocol.SupportFile) error {
if err := os.Remove(configPath); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("remove pow_config.json: %w", err)
}
if err := removeLegacyPowConfig(filepath.Join(m.LuaDir, "pow_config.json")); err != nil {
if err := removeLegacyPowConfig(filepath.Join(m.LuaDir, powConfigFileName)); err != nil {
return err
}
if err := removeLegacyPowConfig(filepath.Join(m.CertDir, "pow_config.json")); err != nil {
if err := removeLegacyPowConfig(filepath.Join(m.CertDir, powConfigFileName)); err != nil {
return err
}
return nil
@@ -739,7 +782,7 @@ func (m *Manager) writeWAFConfig(supportFiles []protocol.SupportFile) error {
configPath := filepath.Join(m.RuntimeConfigDir, "waf_config.json")
for _, file := range supportFiles {
if file.Path == "waf_config.json" {
if err := os.WriteFile(configPath, []byte(file.Content), 0o644); err != nil {
if err := os.WriteFile(configPath, []byte(file.Content), nginxConfigFilePerm); err != nil {
return fmt.Errorf("write waf_config.json: %w", err)
}
slog.Info("wrote waf config", "path", configPath, "size", len(file.Content))
@@ -759,7 +802,7 @@ func (m *Manager) writeSourceConfig(supportFiles []protocol.SupportFile) error {
configPath := filepath.Join(m.RuntimeConfigDir, openrestyrender.SourceConfigFileName)
for _, file := range supportFiles {
if file.Path == openrestyrender.SourceConfigFileName {
if err := os.WriteFile(configPath, []byte(file.Content), 0o644); err != nil {
if err := os.WriteFile(configPath, []byte(file.Content), nginxConfigFilePerm); err != nil {
return fmt.Errorf("write %s: %w", openrestyrender.SourceConfigFileName, err)
}
slog.Info("wrote openresty source config", "path", configPath, "size", len(file.Content))
@@ -775,7 +818,7 @@ func (m *Manager) writeSourceConfig(supportFiles []protocol.SupportFile) error {
func (m *Manager) writeManagedCertFiles(certFiles []protocol.SupportFile) error {
files := make([]managedFile, 0, len(certFiles))
for _, file := range certFiles {
if file.Path == "pow_config.json" || file.Path == "waf_config.json" || file.Path == openrestyrender.SourceConfigFileName {
if file.Path == powConfigFileName || file.Path == "waf_config.json" || file.Path == openrestyrender.SourceConfigFileName {
continue
}
targetPath, err := m.certFileTargetPath(file.Path)
@@ -817,10 +860,10 @@ func (m *Manager) readCertFiles() ([]protocol.SupportFile, error) {
if err != nil {
return err
}
if filepath.ToSlash(relativePath) == "pow_config.json" {
if filepath.ToSlash(relativePath) == powConfigFileName {
return nil
}
data, err := os.ReadFile(path)
data, err := os.ReadFile(path) //nolint:gosec // path is under managed baseDir walk root
if err != nil {
return err
}
@@ -840,7 +883,7 @@ func (m *Manager) readCertFiles() ([]protocol.SupportFile, error) {
}
func (m *Manager) readPowConfigFile() (*protocol.SupportFile, error) {
return m.readRuntimeConfigFile("pow_config.json")
return m.readRuntimeConfigFile(powConfigFileName)
}
func (m *Manager) readRuntimeConfigFile(name string) (*protocol.SupportFile, error) {
@@ -848,7 +891,7 @@ func (m *Manager) readRuntimeConfigFile(name string) (*protocol.SupportFile, err
return nil, nil
}
configPath := filepath.Join(m.RuntimeConfigDir, name)
data, err := os.ReadFile(configPath)
data, err := os.ReadFile(configPath) //nolint:gosec // configPath is under managed RuntimeConfigDir
if err != nil {
if os.IsNotExist(err) {
return nil, nil
@@ -894,7 +937,7 @@ func (m *Manager) restoreRuntimeConfig(file *protocol.SupportFile, name string)
}
return nil
}
return os.WriteFile(configPath, []byte(file.Content), 0o644)
return os.WriteFile(configPath, []byte(file.Content), nginxConfigFilePerm)
}
func (m *Manager) writeSafeDefaultFallbackFiles() error {
@@ -904,16 +947,16 @@ func (m *Manager) writeSafeDefaultFallbackFiles() error {
if strings.TrimSpace(m.RouteConfigPath) == "" {
return errors.New("route config path 不能为空")
}
if err := os.MkdirAll(filepath.Dir(m.MainConfigPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(m.MainConfigPath), nginxDirPerm); err != nil {
return err
}
if err := os.MkdirAll(filepath.Dir(m.RouteConfigPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(m.RouteConfigPath), nginxDirPerm); err != nil {
return err
}
if err := os.WriteFile(m.RouteConfigPath, nil, 0o644); err != nil {
if err := os.WriteFile(m.RouteConfigPath, nil, nginxConfigFilePerm); err != nil {
return err
}
if err := os.WriteFile(m.MainConfigPath, []byte(m.safeDefaultFallbackMainConfig()), 0o644); err != nil {
if err := os.WriteFile(m.MainConfigPath, []byte(m.safeDefaultFallbackMainConfig()), nginxConfigFilePerm); err != nil {
return err
}
return nil
@@ -928,10 +971,10 @@ func (m *Manager) safeDefaultFallbackMainConfig() string {
}
func (m *Manager) checkStubStatus(ctx context.Context) error {
ctx, cancel := context.WithTimeout(ctx, 1500*time.Millisecond)
ctx, cancel := context.WithTimeout(ctx, stubStatusCheckTimeout)
defer cancel()
openrestyStubUrl := fmt.Sprintf("http://127.0.0.1:%d/openflare/stub_status", m.OpenrestyObservabilityPort)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, openrestyStubUrl, nil)
openrestyStubURL := fmt.Sprintf("http://127.0.0.1:%d/openflare/stub_status", m.OpenrestyObservabilityPort)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, openrestyStubURL, nil)
if err != nil {
return err
}
@@ -939,11 +982,11 @@ func (m *Manager) checkStubStatus(ctx context.Context) error {
if err != nil {
return fmt.Errorf("openresty health endpoint unreachable: %w", err)
}
defer resp.Body.Close()
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("openresty health endpoint returned %s", resp.Status)
}
slog.Debug("openresty health endpoint is healthy", "url", openrestyStubUrl)
slog.Debug("openresty health endpoint is healthy", "url", openrestyStubURL)
return nil
}
@@ -986,11 +1029,11 @@ func (m *Manager) certFileTargetPath(relativePath string) (string, error) {
func certFileMode(relativePath string) fs.FileMode {
switch strings.ToLower(filepath.Ext(strings.TrimSpace(relativePath))) {
case ".crt", ".pem":
return 0o644
return nginxConfigFilePerm
case ".key":
return 0o600
return nginxPrivateKeyFilePerm
default:
return 0o644
return nginxConfigFilePerm
}
}
@@ -1029,7 +1072,7 @@ func syncManagedFiles(baseDir string, files []managedFile) error {
} else if err != nil && !os.IsNotExist(err) {
return err
}
if err := os.MkdirAll(baseDir, 0o755); err != nil {
if err := os.MkdirAll(baseDir, nginxDirPerm); err != nil {
return err
}
@@ -1060,14 +1103,14 @@ func syncManagedFiles(baseDir string, files []managedFile) error {
if _, ok := desired[filepath.Clean(relativePath)]; ok {
return nil
}
return os.Remove(path)
return os.Remove(path) //nolint:gosec // path is resolved under the managed baseDir walk root
}); err != nil {
return err
}
for _, file := range desired {
targetPath := filepath.Join(baseDir, file.Path)
if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(targetPath), nginxDirPerm); err != nil {
return err
}
if err := os.WriteFile(targetPath, file.Content, file.Mode); err != nil {
@@ -1119,10 +1162,10 @@ func (m *Manager) ensureMimeTypes() error {
} else if !os.IsNotExist(err) {
return err
}
if err := os.MkdirAll(configDir, 0o755); err != nil {
if err := os.MkdirAll(configDir, nginxDirPerm); err != nil {
return err
}
return os.WriteFile(mimeTypesPath, []byte(DefaultMimeTypes), 0o644)
return os.WriteFile(mimeTypesPath, []byte(DefaultMimeTypes), nginxConfigFilePerm)
}
func (m *Manager) renderRouteConfig(content string) string {
@@ -1183,6 +1226,7 @@ func (m *Manager) managedWAFLuaFiles() []protocol.SupportFile {
return files
}
// ObservabilityListenAddress returns the localhost listen address for stub_status.
func ObservabilityListenAddress(port int) string {
if port <= 0 {
return ""
@@ -1190,6 +1234,7 @@ func ObservabilityListenAddress(port int) string {
return fmt.Sprintf("127.0.0.1:%d", port)
}
// ResolverDirective renders the nginx resolver block for runtime upstream lookups.
func ResolverDirective(explicitResolvers []string) string {
resolvers := resolverAddresses(explicitResolvers)
if len(resolvers) == 0 {
@@ -1211,7 +1256,7 @@ func resolverAddresses(explicitResolvers []string) []string {
func parseResolverAddresses(content string, dockerMode bool) []string {
lines := strings.Split(content, "\n")
resolvers := make([]string, 0, 2)
resolvers := make([]string, 0, resolverAddressCapacity)
seen := make(map[string]struct{})
for _, line := range lines {
fields := strings.Fields(strings.TrimSpace(line))
@@ -1242,6 +1287,7 @@ func isUsableDockerResolver(addr string) bool {
return !ip.IsLoopback() && !ip.IsUnspecified()
}
// RequiresRuntimeResolver reports whether originURL needs a runtime DNS resolver.
func RequiresRuntimeResolver(originURL string) bool {
parsed, err := url.Parse(strings.TrimSpace(originURL))
if err != nil || parsed.Hostname() == "" {
internal/apps/agent/nginx/mimetypes.go
+1
View File
@@ -1,5 +1,6 @@
package nginx
// DefaultMimeTypes is the embedded nginx mime.types map used by generated configs.
const DefaultMimeTypes = `
types {
text/html html htm shtml;
internal/apps/agent/nginx/observability_assets.go
+1
View File
@@ -146,6 +146,7 @@ ngx.header.content_type = "application/json"
ngx.say(cjson.encode(payload))
`
// ManagedObservabilityLuaFiles returns embedded Lua assets for OpenResty observability.
func ManagedObservabilityLuaFiles() []protocol.SupportFile {
return []protocol.SupportFile{
{Path: "init.lua", Content: openRestyObservabilityInitLua},
internal/apps/agent/nginx/pow_assets.go
+2
View File
@@ -542,6 +542,7 @@ end
return M
`
// ManagedPowLuaFiles returns embedded Lua assets for proof-of-work challenges.
func ManagedPowLuaFiles() []protocol.SupportFile {
return []protocol.SupportFile{
{Path: "pow/runtime.lua", Content: openRestyPowRuntimeLua},
@@ -552,6 +553,7 @@ func ManagedPowLuaFiles() []protocol.SupportFile {
}
}
// ManagedPowStaticFiles returns embedded static assets served by the PoW module.
func ManagedPowStaticFiles() ([]protocol.SupportFile, error) {
var files []protocol.SupportFile
entries, err := powStaticFS.ReadDir("pow_static")
internal/apps/agent/nginx/waf_assets.go
+1
View File
@@ -302,6 +302,7 @@ end
return require("waf.runtime").check()
`
// ManagedWAFLuaFiles returns the embedded Lua source files that must be deployed to the WAF runtime directory.
func ManagedWAFLuaFiles() []protocol.SupportFile {
return []protocol.SupportFile{
{Path: "waf/runtime.lua", Content: openRestyWAFRuntimeLua},
internal/apps/agent/observability/collector.go
+9 -2
View File
@@ -1,3 +1,4 @@
// Package observability provides system and service level observability data collection for the agent.
package observability
import (
@@ -15,6 +16,9 @@ import (
edgeobs "github.com/Rain-kl/Wavelet/internal/apps/edge/observability"
)
const nodeHealthEventInitialCapacity = 2
// BuildProfile collects the system profile and returns it only if the fingerprint has changed.
func BuildProfile(cfg *config.Config, stateStore *state.Store) *protocol.NodeSystemProfile {
profile := collectProfile(cfg)
if profile == nil {
@@ -38,6 +42,7 @@ func BuildProfile(cfg *config.Config, stateStore *state.Store) *protocol.NodeSys
return profile
}
// BuildSnapshot captures current system metrics and returns a metric snapshot.
func BuildSnapshot(cfg *config.Config, stateStore *state.Store) *protocol.NodeMetricSnapshot {
now := time.Now().UTC()
metric := &protocol.NodeMetricSnapshot{
@@ -79,6 +84,7 @@ func BuildSnapshot(cfg *config.Config, stateStore *state.Store) *protocol.NodeMe
return metric
}
// BuildOpenrestyObservation builds the OpenResty observation protocol model from the managed metrics.
func BuildOpenrestyObservation(managed *ManagedOpenRestyMetrics) *protocol.NodeOpenrestyObservation {
if managed == nil {
return nil
@@ -91,11 +97,12 @@ func BuildOpenrestyObservation(managed *ManagedOpenRestyMetrics) *protocol.NodeO
}
}
// BuildHealthEvents converts system snapshot health state into a list of health events.
func BuildHealthEvents(snapshot *state.Snapshot) []protocol.NodeHealthEvent {
if snapshot == nil {
return []protocol.NodeHealthEvent{}
}
events := make([]protocol.NodeHealthEvent, 0, 2)
events := make([]protocol.NodeHealthEvent, 0, nodeHealthEventInitialCapacity)
nowUnix := time.Now().UTC().Unix()
if strings.TrimSpace(snapshot.OpenrestyStatus) == protocol.OpenrestyStatusUnhealthy {
events = append(events, protocol.NodeHealthEvent{
@@ -147,4 +154,4 @@ func fingerprintProfile(profile *protocol.NodeSystemProfile) string {
}
sum := sha256.Sum256(raw)
return hex.EncodeToString(sum[:])
}
}
internal/apps/agent/observability/openresty_local.go
+26 -12
View File
@@ -1,6 +1,7 @@
package observability
import (
"context"
"encoding/json"
"fmt"
"io"
@@ -14,11 +15,15 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
)
const openRestyObservabilityPath = "/openflare/observability"
const openRestyStubStatusPath = "/openflare/stub_status"
const (
openRestyObservabilityPath = "/openflare/observability"
openRestyStubStatusPath = "/openflare/stub_status"
stubStatusActiveMatchGroupCount = 2
)
var stubStatusActivePattern = regexp.MustCompile(`Active connections:\s+(\d+)`)
// ManagedOpenRestyMetrics holds metrics collected from the local OpenResty instance.
type ManagedOpenRestyMetrics struct {
TrafficReport *protocol.NodeTrafficReport
OpenrestyRxBytes int64
@@ -39,7 +44,8 @@ type openRestyObservabilityResponse struct {
OpenrestyTxBytes int64 `json:"openresty_tx_bytes"`
}
func CollectManagedOpenRestyMetrics(cfg *config.Config) *ManagedOpenRestyMetrics {
// CollectManagedOpenRestyMetrics collects metrics from the local OpenResty observability endpoints.
func CollectManagedOpenRestyMetrics(ctx context.Context, cfg *config.Config) *ManagedOpenRestyMetrics {
if cfg == nil || cfg.OpenrestyObservabilityPort <= 0 {
return nil
}
@@ -48,7 +54,7 @@ func CollectManagedOpenRestyMetrics(cfg *config.Config) *ManagedOpenRestyMetrics
client := &http.Client{Timeout: 1500 * time.Millisecond}
observabilityResp := openRestyObservabilityResponse{}
if err := fetchLocalJSON(client, baseURL+openRestyObservabilityPath, &observabilityResp); err != nil {
if err := fetchLocalJSON(ctx, client, baseURL+openRestyObservabilityPath, &observabilityResp); err != nil {
return nil
}
@@ -67,31 +73,39 @@ func CollectManagedOpenRestyMetrics(cfg *config.Config) *ManagedOpenRestyMetrics
OpenrestyTxBytes: observabilityResp.OpenrestyTxBytes,
}
if text, err := fetchLocalText(client, baseURL+openRestyStubStatusPath); err == nil {
if text, err := fetchLocalText(ctx, client, baseURL+openRestyStubStatusPath); err == nil {
result.OpenrestyConnections = parseStubStatusActiveConnections(text)
}
return result
}
func fetchLocalJSON(client *http.Client, url string, target any) error {
resp, err := client.Get(url)
func fetchLocalJSON(ctx context.Context, client *http.Client, url string, target any) error {
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return err
}
defer resp.Body.Close()
resp, err := client.Do(req)
if err != nil {
return err
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("unexpected local observability status: %s", resp.Status)
}
return json.NewDecoder(resp.Body).Decode(target)
}
func fetchLocalText(client *http.Client, url string) (string, error) {
resp, err := client.Get(url)
func fetchLocalText(ctx context.Context, client *http.Client, url string) (string, error) {
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return "", err
}
defer resp.Body.Close()
resp, err := client.Do(req)
if err != nil {
return "", err
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("unexpected local stub status: %s", resp.Status)
}
@@ -104,7 +118,7 @@ func fetchLocalText(client *http.Client, url string) (string, error) {
func parseStubStatusActiveConnections(raw string) int64 {
matches := stubStatusActivePattern.FindStringSubmatch(raw)
if len(matches) != 2 {
if len(matches) != stubStatusActiveMatchGroupCount {
return 0
}
value, err := strconv.ParseInt(matches[1], 10, 64)
internal/apps/agent/observability/openresty_local_test.go
+3 -2
View File
@@ -1,6 +1,7 @@
package observability
import (
"context"
"net"
"net/http"
"net/http/httptest"
@@ -31,7 +32,7 @@ func TestCollectManagedOpenRestyMetrics(t *testing.T) {
server.Start()
defer server.Close()
metrics := CollectManagedOpenRestyMetrics(&config.Config{
metrics := CollectManagedOpenRestyMetrics(context.Background(), &config.Config{
OpenrestyObservabilityPort: port,
})
if metrics == nil || metrics.TrafficReport == nil {
@@ -67,7 +68,7 @@ func TestNormalizeCountMapDropsEmptyKeys(t *testing.T) {
func TestCollectManagedOpenRestyMetricsHandlesUnavailableEndpoint(t *testing.T) {
cfg := &config.Config{OpenrestyObservabilityPort: 1}
if metrics := CollectManagedOpenRestyMetrics(cfg); metrics != nil {
if metrics := CollectManagedOpenRestyMetrics(context.Background(), cfg); metrics != nil {
t.Fatalf("expected nil metrics for unavailable endpoint, got %+v", metrics)
}
}
internal/apps/agent/observability/traffic.go
+12 -4
View File
@@ -6,6 +6,7 @@ import (
"errors"
"io"
"log/slog"
"net/http"
"os"
"regexp"
"sort"
@@ -28,6 +29,11 @@ type accessLogRecord struct {
RequestLength int64 `json:"request_length"`
}
const (
combinedAccessLogMatchGroupCount = 5
trafficTopDomainsLimit = 8
)
var combinedAccessLogPattern = regexp.MustCompile(`^(\S+)\s+\S+\s+\S+\s+\[([^]]+)]\s+"\S+\s+(\S+)(?:\s+[^"]*)?"\s+(\d{3})\s+\S+`)
type trafficAggregate struct {
@@ -43,11 +49,13 @@ type trafficAggregate struct {
logs []protocol.NodeAccessLog
}
// BuildTrafficReport generates a traffic report using access logs or falling back to managed metrics.
func BuildTrafficReport(cfg *config.Config, stateStore *state.Store, managed *ManagedOpenRestyMetrics) *protocol.NodeTrafficReport {
report, _, _ := BuildTrafficObservability(cfg, stateStore, managed)
return report
}
// BuildTrafficObservability returns the traffic report, parsed access logs, and managed metrics.
func BuildTrafficObservability(cfg *config.Config, stateStore *state.Store, managed *ManagedOpenRestyMetrics) (*protocol.NodeTrafficReport, []protocol.NodeAccessLog, *ManagedOpenRestyMetrics) {
if cfg == nil || stateStore == nil {
if managed != nil && managed.TrafficReport != nil {
@@ -78,7 +86,7 @@ func readAccessLogDelta(cfg *config.Config, stateStore *state.Store) *trafficAgg
}
logPath := managedAccessLogPath(cfg)
file, err := os.Open(logPath)
file, err := os.Open(logPath) //nolint:gosec // path is the configured managed access log location
if err != nil {
if os.IsNotExist(err) {
if snapshot.AccessLogOffset != 0 {
@@ -167,7 +175,7 @@ func (aggregate *trafficAggregate) consume(line []byte) {
}
aggregate.requestCount++
if record.Status >= 500 {
if record.Status >= http.StatusInternalServerError {
aggregate.errorCount++
}
if record.Status > 0 {
@@ -234,7 +242,7 @@ func parseJSONAccessLogRecord(raw string) (parsedAccessLogRecord, bool) {
func parseCombinedAccessLogRecord(raw string) (parsedAccessLogRecord, bool) {
matches := combinedAccessLogPattern.FindStringSubmatch(raw)
if len(matches) != 5 {
if len(matches) != combinedAccessLogMatchGroupCount {
return parsedAccessLogRecord{}, false
}
timestamp, err := parseAccessLogTime(matches[2])
@@ -265,7 +273,7 @@ func (aggregate *trafficAggregate) report() *protocol.NodeTrafficReport {
ErrorCount: aggregate.errorCount,
UniqueVisitorCount: int64(len(aggregate.visitors)),
StatusCodes: cloneTrafficCounts(aggregate.statusCodes, 0),
TopDomains: topCounts(aggregate.topDomains, 8),
TopDomains: topCounts(aggregate.topDomains, trafficTopDomainsLimit),
SourceCountries: map[string]int64{},
}
}
internal/apps/agent/protocol/alias.go
+65 -9
View File
@@ -1,43 +1,99 @@
// Package protocol defines type aliases and constants for the agent protocol.
package protocol
import pkgprotocol "github.com/Rain-kl/Wavelet/pkg/protocol"
// APIResponse is an alias for pkgprotocol.APIResponse.
type APIResponse[T any] = pkgprotocol.APIResponse[T]
// HeartbeatData is an alias for pkgprotocol.HeartbeatData.
type HeartbeatData = pkgprotocol.HeartbeatData
// HeartbeatResult is an alias for pkgprotocol.HeartbeatResult.
type HeartbeatResult = pkgprotocol.HeartbeatResult
// AgentSettings is an alias for pkgprotocol.AgentSettings.
type AgentSettings = pkgprotocol.AgentSettings
// WSMessage is an alias for pkgprotocol.WSMessage.
type WSMessage = pkgprotocol.WSMessage
// WSOutboundMessage is an alias for pkgprotocol.WSOutboundMessage.
type WSOutboundMessage = pkgprotocol.WSOutboundMessage
// WebSocketConnection is an alias for pkgprotocol.WebSocketConnection.
type WebSocketConnection = pkgprotocol.WebSocketConnection
// NodePayload is an alias for pkgprotocol.NodePayload.
type NodePayload = pkgprotocol.NodePayload
// NodeSystemProfile is an alias for pkgprotocol.NodeSystemProfile.
type NodeSystemProfile = pkgprotocol.NodeSystemProfile
// NodeMetricSnapshot is an alias for pkgprotocol.NodeMetricSnapshot.
type NodeMetricSnapshot = pkgprotocol.NodeMetricSnapshot
// NodeOpenrestyObservation is an alias for pkgprotocol.NodeOpenrestyObservation.
type NodeOpenrestyObservation = pkgprotocol.NodeOpenrestyObservation
// NodeTrafficReport is an alias for pkgprotocol.NodeTrafficReport.
type NodeTrafficReport = pkgprotocol.NodeTrafficReport
// NodeAccessLog is an alias for pkgprotocol.NodeAccessLog.
type NodeAccessLog = pkgprotocol.NodeAccessLog
// BufferedObservabilityRecord is an alias for pkgprotocol.BufferedObservabilityRecord.
type BufferedObservabilityRecord = pkgprotocol.BufferedObservabilityRecord
// NodeHealthEvent is an alias for pkgprotocol.NodeHealthEvent.
type NodeHealthEvent = pkgprotocol.NodeHealthEvent
// RegisterNodeResponse is an alias for pkgprotocol.RegisterNodeResponse.
type RegisterNodeResponse = pkgprotocol.RegisterNodeResponse
// ApplyLogPayload is an alias for pkgprotocol.ApplyLogPayload.
type ApplyLogPayload = pkgprotocol.ApplyLogPayload
// ActiveConfigResponse is an alias for pkgprotocol.ActiveConfigResponse.
type ActiveConfigResponse = pkgprotocol.ActiveConfigResponse
// ActiveConfigMeta is an alias for pkgprotocol.ActiveConfigMeta.
type ActiveConfigMeta = pkgprotocol.ActiveConfigMeta
// WAFIPGroup is an alias for pkgprotocol.WAFIPGroup.
type WAFIPGroup = pkgprotocol.WAFIPGroup
// WAFIPGroupSyncRequest is an alias for pkgprotocol.WAFIPGroupSyncRequest.
type WAFIPGroupSyncRequest = pkgprotocol.WAFIPGroupSyncRequest
// WAFIPGroupSyncResponse is an alias for pkgprotocol.WAFIPGroupSyncResponse.
type WAFIPGroupSyncResponse = pkgprotocol.WAFIPGroupSyncResponse
// SupportFile is an alias for pkgprotocol.SupportFile.
type SupportFile = pkgprotocol.SupportFile
const (
WSMessageTypeStatus = pkgprotocol.WSMessageTypeStatus
WSMessageTypeSettings = pkgprotocol.WSMessageTypeSettings
WSMessageTypeActiveConfig = pkgprotocol.WSMessageTypeActiveConfig
// WSMessageTypeStatus is an alias for pkgprotocol.WSMessageTypeStatus.
WSMessageTypeStatus = pkgprotocol.WSMessageTypeStatus
// WSMessageTypeSettings is an alias for pkgprotocol.WSMessageTypeSettings.
WSMessageTypeSettings = pkgprotocol.WSMessageTypeSettings
// WSMessageTypeActiveConfig is an alias for pkgprotocol.WSMessageTypeActiveConfig.
WSMessageTypeActiveConfig = pkgprotocol.WSMessageTypeActiveConfig
// WSMessageTypeForceSyncConfig is an alias for pkgprotocol.WSMessageTypeForceSyncConfig.
WSMessageTypeForceSyncConfig = pkgprotocol.WSMessageTypeForceSyncConfig
WSMessageTypeWAFIPGroups = pkgprotocol.WSMessageTypeWAFIPGroups
WSMessageTypePing = pkgprotocol.WSMessageTypePing
WSMessageTypePong = pkgprotocol.WSMessageTypePong
// WSMessageTypeWAFIPGroups is an alias for pkgprotocol.WSMessageTypeWAFIPGroups.
WSMessageTypeWAFIPGroups = pkgprotocol.WSMessageTypeWAFIPGroups
// WSMessageTypePing is an alias for pkgprotocol.WSMessageTypePing.
WSMessageTypePing = pkgprotocol.WSMessageTypePing
// WSMessageTypePong is an alias for pkgprotocol.WSMessageTypePong.
WSMessageTypePong = pkgprotocol.WSMessageTypePong
)
const (
OpenrestyStatusHealthy = pkgprotocol.OpenrestyStatusHealthy
// OpenrestyStatusHealthy is an alias for pkgprotocol.OpenrestyStatusHealthy.
OpenrestyStatusHealthy = pkgprotocol.OpenrestyStatusHealthy
// OpenrestyStatusUnhealthy is an alias for pkgprotocol.OpenrestyStatusUnhealthy.
OpenrestyStatusUnhealthy = pkgprotocol.OpenrestyStatusUnhealthy
OpenrestyStatusUnknown = pkgprotocol.OpenrestyStatusUnknown
)
// OpenrestyStatusUnknown is an alias for pkgprotocol.OpenrestyStatusUnknown.
OpenrestyStatusUnknown = pkgprotocol.OpenrestyStatusUnknown
)
internal/apps/agent/state/observability_buffer.go
+10 -2
View File
@@ -1,3 +1,4 @@
// Package state persists agent runtime state and observability snapshots.
package state
import (
@@ -13,6 +14,7 @@ import (
const observabilityBufferWindowSeconds = 60
// ObservabilityBufferRecord stores observability data for a single time window.
type ObservabilityBufferRecord struct {
WindowStartedAtUnix int64 `json:"window_started_at_unix"`
Snapshot *protocol.NodeMetricSnapshot `json:"snapshot,omitempty"`
@@ -22,15 +24,18 @@ type ObservabilityBufferRecord struct {
QueuedAtUnix int64 `json:"queued_at_unix"`
}
// ObservabilityBufferStore persists observability records to disk for replay on heartbeat.
type ObservabilityBufferStore struct {
path string
mu sync.Mutex
}
// NewObservabilityBufferStore creates a store backed by the file at path.
func NewObservabilityBufferStore(path string) *ObservabilityBufferStore {
return &ObservabilityBufferStore{path: filepath.Clean(path)}
}
// Upsert inserts or merges an observability record and prunes entries older than retainAfterUnix.
func (s *ObservabilityBufferStore) Upsert(record ObservabilityBufferRecord, retainAfterUnix int64) error {
if s == nil || record.WindowStartedAtUnix <= 0 || (record.Snapshot == nil && record.OpenrestyObservation == nil && record.TrafficReport == nil && len(record.AccessLogs) == 0) {
return nil
@@ -113,6 +118,7 @@ func accessLogKey(item protocol.NodeAccessLog) string {
return strconv.FormatInt(item.LoggedAtUnix, 10) + "|" + item.RemoteAddr + "|" + item.Host + "|" + item.Path + "|" + strconv.Itoa(item.StatusCode)
}
// Replayable returns buffered records from windows before currentWindowStartedAtUnix.
func (s *ObservabilityBufferStore) Replayable(currentWindowStartedAtUnix int64, retainAfterUnix int64) ([]ObservabilityBufferRecord, error) {
if s == nil {
return nil, nil
@@ -138,6 +144,7 @@ func (s *ObservabilityBufferStore) Replayable(currentWindowStartedAtUnix int64,
return result, nil
}
// Ack removes acknowledged observability windows and prunes entries older than retainAfterUnix.
func (s *ObservabilityBufferStore) Ack(windowStartedAtUnix []int64, retainAfterUnix int64) error {
if s == nil || len(windowStartedAtUnix) == 0 {
return nil
@@ -185,16 +192,17 @@ func (s *ObservabilityBufferStore) loadUnlocked() ([]ObservabilityBufferRecord,
}
func (s *ObservabilityBufferStore) saveUnlocked(records []ObservabilityBufferRecord) error {
if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(s.path), stateDirPerm); err != nil {
return err
}
data, err := json.MarshalIndent(records, "", " ")
if err != nil {
return err
}
return os.WriteFile(s.path, data, 0o644)
return os.WriteFile(s.path, data, stateFilePerm)
}
// ObservabilityWindowStartedAt calculates the start of the 60-second window for the given metrics, openresty observation, or traffic report.
func ObservabilityWindowStartedAt(snapshot *protocol.NodeMetricSnapshot, openresty *protocol.NodeOpenrestyObservation, traffic *protocol.NodeTrafficReport) int64 {
if traffic != nil && traffic.WindowStartedAtUnix > 0 {
return traffic.WindowStartedAtUnix - (traffic.WindowStartedAtUnix % observabilityBufferWindowSeconds)
internal/apps/agent/state/state.go
+15 -3
View File
@@ -9,6 +9,13 @@ import (
"sync"
)
const (
stateDirPerm = 0o750
stateFilePerm = 0o600
nodeIDRandomBytes = 8
)
// Snapshot represents the state of the agent at a given point in time.
type Snapshot struct {
NodeID string `json:"node_id"`
CurrentVersion string `json:"current_version"`
@@ -26,21 +33,25 @@ type Snapshot struct {
AccessLogOffset int64 `json:"access_log_offset"`
}
// Store manages the storage and retrieval of the agent state snapshot.
type Store struct {
path string
mu sync.Mutex
}
// NewStore creates a new Store instance at the given path.
func NewStore(path string) *Store {
return &Store{path: filepath.Clean(path)}
}
// Load loads the snapshot from the store.
func (s *Store) Load() (*Snapshot, error) {
s.mu.Lock()
defer s.mu.Unlock()
return s.loadUnlocked()
}
// EnsureNodeID returns the existing node ID, or generates and saves a new one if it does not exist.
func (s *Store) EnsureNodeID() (string, error) {
s.mu.Lock()
defer s.mu.Unlock()
@@ -62,6 +73,7 @@ func (s *Store) EnsureNodeID() (string, error) {
return snapshot.NodeID, nil
}
// Save saves the given snapshot to the store.
func (s *Store) Save(snapshot *Snapshot) error {
s.mu.Lock()
defer s.mu.Unlock()
@@ -87,18 +99,18 @@ func (s *Store) loadUnlocked() (*Snapshot, error) {
}
func (s *Store) saveUnlocked(snapshot *Snapshot) error {
if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(s.path), stateDirPerm); err != nil {
return err
}
data, err := json.MarshalIndent(snapshot, "", " ")
if err != nil {
return err
}
return os.WriteFile(s.path, data, 0o644)
return os.WriteFile(s.path, data, stateFilePerm)
}
func newNodeID() (string, error) {
buf := make([]byte, 8)
buf := make([]byte, nodeIDRandomBytes)
if _, err := rand.Read(buf); err != nil {
return "", err
}
internal/apps/agent/sync/pages.go
+41 -25
View File
@@ -1,3 +1,4 @@
// Package sync applies control-plane configuration to the local agent runtime.
package sync
import (
@@ -10,6 +11,7 @@ import (
"errors"
"fmt"
"io"
"math"
"os"
"path"
"path/filepath"
@@ -18,6 +20,12 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
)
const (
pagesMaxExtractedFileBytes = 100 * 1024 * 1024
pagesDirPerm = 0o755
pagesManifestFilePerm = 0o644
)
type pagesSourceDocument struct {
Routes []pagesSourceRoute `json:"routes"`
}
@@ -66,7 +74,7 @@ func (s *Service) ensurePagesDeployment(ctx context.Context, deployment pagesDep
return fmt.Errorf("download Pages deployment %d: %w", deployment.DeploymentID, err)
}
if got := checksumBytes(packageBytes); got != deployment.Checksum {
return fmt.Errorf("Pages deployment %d checksum mismatch: expected %s, got %s", deployment.DeploymentID, deployment.Checksum, got)
return fmt.Errorf("pages deployment %d checksum mismatch: expected %s, got %s", deployment.DeploymentID, deployment.Checksum, got)
}
releaseDir := pagesReleaseDir(s.pagesDir, deployment.DeploymentID, deployment.Checksum)
if !markerMatches(releaseDir, deployment) {
@@ -83,7 +91,7 @@ func referencedPagesDeployments(config *protocol.ActiveConfigResponse) ([]pagesD
}
var doc pagesSourceDocument
if err := json.Unmarshal([]byte(config.SourceConfigJSON), &doc); err != nil {
return nil, fmt.Errorf("decode Pages references: %w", err)
return nil, fmt.Errorf("decode pages references: %w", err)
}
seen := make(map[uint]struct{})
result := make([]pagesDeploymentSource, 0)
@@ -94,7 +102,7 @@ func referencedPagesDeployments(config *protocol.ActiveConfigResponse) ([]pagesD
deploymentID := route.PagesDeployment.DeploymentID
checksum := strings.TrimSpace(route.PagesDeployment.Checksum)
if deploymentID == 0 || checksum == "" {
return nil, errors.New("Pages deployment snapshot is incomplete")
return nil, errors.New("pages deployment snapshot is incomplete")
}
if _, ok := seen[deploymentID]; ok {
continue
@@ -152,7 +160,7 @@ func findCommonRootPrefix(files []*zip.File) (string, error) {
func extractPagesPackage(packageBytes []byte, releaseDir string, deployment pagesDeploymentSource) error {
tmpDir := releaseDir + ".tmp"
_ = os.RemoveAll(tmpDir)
if err := os.MkdirAll(tmpDir, 0o755); err != nil {
if err := os.MkdirAll(tmpDir, pagesDirPerm); err != nil {
return err
}
reader, err := zip.NewReader(bytes.NewReader(packageBytes), int64(len(packageBytes)))
@@ -182,7 +190,7 @@ func extractPagesPackage(packageBytes []byte, releaseDir string, deployment page
}
if item.FileInfo().Mode()&os.ModeSymlink != 0 {
_ = os.RemoveAll(tmpDir)
return fmt.Errorf("Pages package contains unsupported symlink: %s", relativePath)
return fmt.Errorf("pages package contains unsupported symlink: %s", relativePath)
}
if err := extractPagesFile(item, filepath.Join(tmpDir, relativePath)); err != nil {
_ = os.RemoveAll(tmpDir)
@@ -197,21 +205,32 @@ func extractPagesPackage(packageBytes []byte, releaseDir string, deployment page
return os.Rename(tmpDir, releaseDir)
}
func pagesZipEntryCopyLimit(size uint64) (int64, error) {
if size == 0 || size > pagesMaxExtractedFileBytes || size > uint64(math.MaxInt64) {
return 0, errors.New("pages file size out of bounds")
}
return int64(size), nil //nolint:gosec // size is bounded to math.MaxInt64 above
}
func extractPagesFile(item *zip.File, targetPath string) error {
if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(targetPath), pagesDirPerm); err != nil {
return err
}
source, err := item.Open()
if err != nil {
return err
}
defer source.Close()
target, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, item.FileInfo().Mode().Perm())
defer func() { _ = source.Close() }()
target, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, item.FileInfo().Mode().Perm()) //nolint:gosec // targetPath is under managed PagesDir from validated zip entry
if err != nil {
return err
}
defer target.Close()
_, err = io.Copy(target, source)
defer func() { _ = target.Close() }()
limit, err := pagesZipEntryCopyLimit(item.UncompressedSize64)
if err != nil {
return fmt.Errorf("%s: %w", item.Name, err)
}
_, err = io.CopyN(target, source, limit)
return err
}
@@ -219,7 +238,7 @@ func switchPagesCurrentDir(baseDir string, deploymentID uint, releaseDir string)
currentDir := pagesCurrentDir(baseDir, deploymentID)
previousDir := currentDir + ".previous"
_ = os.RemoveAll(previousDir)
if err := os.MkdirAll(filepath.Dir(currentDir), 0o755); err != nil {
if err := os.MkdirAll(filepath.Dir(currentDir), pagesDirPerm); err != nil {
return err
}
if _, err := os.Stat(currentDir); err == nil {
@@ -249,25 +268,25 @@ func copyPagesDir(sourceDir string, targetDir string) error {
}
targetPath := filepath.Join(targetDir, relativePath)
if entry.IsDir() {
return os.MkdirAll(targetPath, 0o755)
return os.MkdirAll(targetPath, pagesDirPerm)
}
info, err := entry.Info()
if err != nil {
return err
}
input, err := os.Open(sourcePath)
input, err := os.Open(sourcePath) //nolint:gosec // sourcePath is under managed PagesDir walk root
if err != nil {
return err
}
defer input.Close()
if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil {
defer func() { _ = input.Close() }()
if err := os.MkdirAll(filepath.Dir(targetPath), pagesDirPerm); err != nil {
return err
}
output, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, info.Mode().Perm())
output, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, info.Mode().Perm()) //nolint:gosec // targetPath is under managed PagesDir walk root
if err != nil {
return err
}
defer output.Close()
defer func() { _ = output.Close() }()
_, err = io.Copy(output, input)
return err
})
@@ -279,20 +298,20 @@ func normalizePagesArchivePath(raw string) (string, bool, error) {
return "", true, nil
}
if strings.HasPrefix(name, "/") {
return "", false, fmt.Errorf("Pages package contains absolute path: %s", raw)
return "", false, fmt.Errorf("pages package contains absolute path: %s", raw)
}
cleaned := path.Clean(name)
if cleaned == "." {
return "", true, nil
}
if cleaned == ".." || strings.HasPrefix(cleaned, "../") || strings.Contains(cleaned, "/../") {
return "", false, fmt.Errorf("Pages package path escapes deployment root: %s", raw)
return "", false, fmt.Errorf("pages package path escapes deployment root: %s", raw)
}
return filepath.FromSlash(cleaned), false, nil
}
func markerMatches(dir string, deployment pagesDeploymentSource) bool {
data, err := os.ReadFile(filepath.Join(dir, ".openflare-pages.json"))
data, err := os.ReadFile(filepath.Join(dir, ".openflare-pages.json")) //nolint:gosec // dir is managed PagesDir
if err != nil {
return false
}
@@ -304,14 +323,11 @@ func markerMatches(dir string, deployment pagesDeploymentSource) bool {
}
func writePagesMarker(dir string, deployment pagesDeploymentSource) error {
data, err := json.Marshal(pagesDeploymentMarker{
DeploymentID: deployment.DeploymentID,
Checksum: deployment.Checksum,
})
data, err := json.Marshal(pagesDeploymentMarker(deployment))
if err != nil {
return err
}
return os.WriteFile(filepath.Join(dir, ".openflare-pages.json"), data, 0o644)
return os.WriteFile(filepath.Join(dir, ".openflare-pages.json"), data, pagesManifestFilePerm)
}
func pagesCurrentDir(baseDir string, deploymentID uint) string {
internal/apps/agent/sync/service.go
+36 -154
View File
@@ -18,12 +18,14 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/agent/state"
)
// Apply result constants indicate the outcome reported back to the server.
const (
ApplyResultSuccess = "success"
ApplyResultWarning = "warning"
ApplyResultFailed = "failed"
)
// ConfigClient is the interface for communicating with the server control plane.
type ConfigClient interface {
GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error)
DownloadPagesDeploymentPackage(ctx context.Context, deploymentID uint) ([]byte, error)
@@ -31,6 +33,7 @@ type ConfigClient interface {
SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGroupSyncRequest) (*protocol.WAFIPGroupSyncResponse, error)
}
// NginxManager is the interface for managing the local OpenResty instance.
type NginxManager interface {
Apply(ctx context.Context, mainConfig string, routeConfig string, supportFiles []protocol.SupportFile) nginx.ApplyOutcome
EnsureRuntime(ctx context.Context, recreate bool) error
@@ -40,6 +43,7 @@ type NginxManager interface {
SyncWAFIPGroups(groups []protocol.WAFIPGroup) error
}
// Service orchestrates configuration synchronisation between the server and the local OpenResty instance.
type Service struct {
client ConfigClient
nginxManager NginxManager
@@ -47,10 +51,12 @@ type Service struct {
pagesDir string
}
// SetPagesDir sets the local directory used for pages deployment packages.
func (s *Service) SetPagesDir(path string) {
s.pagesDir = strings.TrimSpace(path)
}
// New creates a new Service with the given client, nginx manager, and state store.
func New(client ConfigClient, nginxManager NginxManager, stateStore *state.Store) *Service {
return &Service{
client: client,
@@ -59,100 +65,34 @@ func New(client ConfigClient, nginxManager NginxManager, stateStore *state.Store
}
}
// SyncOnce performs a single periodic sync against the given active config summary.
func (s *Service) SyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
return s.sync(ctx, false, target)
}
// SyncOnStartup performs an initial sync at agent startup, applying config even when checksums already match.
func (s *Service) SyncOnStartup(ctx context.Context, target *protocol.ActiveConfigMeta) error {
return s.sync(ctx, true, target)
}
func (s *Service) sync(ctx context.Context, startup bool, target *protocol.ActiveConfigMeta) error {
mode := "periodic"
if startup {
mode = "startup"
}
snapshot, err := s.stateStore.Load()
mode := syncMode(startup)
snapshot, currentChecksum, err := s.loadSyncState()
if err != nil {
return err
}
currentChecksum, err := s.nginxManager.CurrentChecksum()
if err != nil {
return err
}
if target != nil {
target.Version = strings.TrimSpace(target.Version)
target.Checksum = strings.TrimSpace(target.Checksum)
}
normalizeSyncTarget(target)
if target == nil || target.Version == "" || target.Checksum == "" {
if !startup {
slog.Debug("skipping sync because heartbeat returned no active config summary", "mode", mode)
return nil
}
slog.Debug("sync startup fallback: active config summary unavailable, fetching active config directly")
config, fetchErr := s.client.GetActiveConfig(ctx)
if fetchErr != nil {
slog.Error("fetch active config failed", "mode", mode, "error", fetchErr)
return fetchErr
}
target = &protocol.ActiveConfigMeta{
Version: config.Version,
Checksum: config.Checksum,
}
return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
return s.syncWithoutTarget(ctx, mode, startup, snapshot, currentChecksum)
}
if currentChecksum == target.Checksum {
if startup {
config, fetchErr := s.client.GetActiveConfig(ctx)
if fetchErr != nil {
slog.Error("fetch active config failed", "mode", mode, "error", fetchErr)
return fetchErr
}
return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
}
slog.Debug("local openresty config already up to date", "mode", mode, "version", target.Version)
shouldReport := shouldReportNoopApply(snapshot, target.Version, target.Checksum)
if shouldReport {
if err = s.reportNoopApply(ctx, snapshot.NodeID, target.Version, target.Checksum, "", "", 0); err != nil {
return err
}
}
snapshot.CurrentVersion = target.Version
snapshot.CurrentChecksum = target.Checksum
clearBlockedTarget(snapshot)
snapshot.LastError = ""
slog.Debug("sync finished without changes", "mode", mode, "version", target.Version)
return s.stateStore.Save(snapshot)
return s.syncMatchingChecksum(ctx, mode, startup, snapshot, currentChecksum, target)
}
if isBlockedTarget(snapshot, target.Version, target.Checksum) {
slog.Warn("skipping blocked config version after previous failed apply", "mode", mode, "version", target.Version, "checksum", target.Checksum)
if startup {
if err = s.ensureRuntimeForCurrentConfig(ctx, mode, snapshot, currentChecksum); err != nil {
return err
}
return s.stateStore.Save(snapshot)
}
return nil
}
if hasBlockedTarget(snapshot) {
clearBlockedTarget(snapshot)
}
if snapshot.CurrentVersion == target.Version && snapshot.CurrentChecksum == target.Checksum && !startup {
slog.Debug("skipping config fetch because state already records target version/checksum", "version", target.Version, "checksum", target.Checksum)
return s.stateStore.Save(snapshot)
}
config, err := s.client.GetActiveConfig(ctx)
if err != nil {
slog.Error("fetch active config failed", "mode", mode, "error", err)
return err
}
return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
return s.syncMismatchedChecksum(ctx, mode, startup, snapshot, currentChecksum, target)
}
// ForceSyncOnce clears any blocked target state then unconditionally fetches and applies the active config.
func (s *Service) ForceSyncOnce(ctx context.Context, target *protocol.ActiveConfigMeta) error {
snapshot, err := s.stateStore.Load()
if err != nil {
@@ -174,6 +114,7 @@ func (s *Service) ForceSyncOnce(ctx context.Context, target *protocol.ActiveConf
return s.applyIfNeeded(ctx, "force", true, snapshot, currentChecksum, target, config)
}
// WAFIPGroupChecksums returns the current per-group checksums held by the nginx manager.
func (s *Service) WAFIPGroupChecksums() (map[string]string, error) {
if s.nginxManager == nil {
return map[string]string{}, nil
@@ -181,7 +122,8 @@ func (s *Service) WAFIPGroupChecksums() (map[string]string, error) {
return s.nginxManager.WAFIPGroupChecksums()
}
func (s *Service) ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPGroup) error {
// ApplyWAFIPGroups writes the given WAF IP groups to the nginx manager.
func (s *Service) ApplyWAFIPGroups(_ context.Context, groups []protocol.WAFIPGroup) error {
if len(groups) == 0 || s.nginxManager == nil {
return nil
}
@@ -190,36 +132,13 @@ func (s *Service) ApplyWAFIPGroups(ctx context.Context, groups []protocol.WAFIPG
func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string, target *protocol.ActiveConfigMeta, config *protocol.ActiveConfigResponse) error {
if currentChecksum == config.Checksum && !startup {
slog.Debug("local openresty config already up to date", "mode", mode, "version", config.Version)
shouldReport := shouldReportNoopApply(snapshot, config.Version, config.Checksum)
if shouldReport {
rendered, renderErr := renderActiveConfig(config)
if renderErr != nil {
return renderErr
}
if err := s.reportNoopApply(ctx, snapshot.NodeID, config.Version, config.Checksum, checksumString(rendered.mainConfig), checksumString(rendered.routeConfig), len(rendered.supportFiles)); err != nil {
return err
}
}
snapshot.CurrentVersion = config.Version
snapshot.CurrentChecksum = config.Checksum
clearBlockedTarget(snapshot)
snapshot.LastError = ""
slog.Debug("sync finished without changes", "mode", mode, "version", config.Version)
return s.stateStore.Save(snapshot)
return s.handleUpToDateConfig(ctx, mode, snapshot, config)
}
if target != nil && (target.Version != config.Version || target.Checksum != config.Checksum) {
slog.Warn("active config changed between heartbeat and fetch", "heartbeat_version", target.Version, "heartbeat_checksum", target.Checksum, "fetched_version", config.Version, "fetched_checksum", config.Checksum)
}
if isBlockedTarget(snapshot, config.Version, config.Checksum) {
slog.Warn("skipping blocked config after fetch because the same version previously failed", "mode", mode, "version", config.Version, "checksum", config.Checksum)
if startup {
if err := s.ensureRuntimeForCurrentConfig(ctx, mode, snapshot, currentChecksum); err != nil {
return err
}
return s.stateStore.Save(snapshot)
}
return nil
if handled, err := s.handleBlockedConfigAfterFetch(ctx, mode, startup, snapshot, currentChecksum, config); handled {
return err
}
if hasBlockedTarget(snapshot) {
clearBlockedTarget(snapshot)
@@ -228,6 +147,10 @@ func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool,
slog.Debug("skipping apply because state already records target version/checksum", "version", config.Version, "checksum", config.Checksum)
return s.stateStore.Save(snapshot)
}
return s.applyRenderedConfig(ctx, mode, snapshot, currentChecksum, config)
}
func (s *Service) applyRenderedConfig(ctx context.Context, mode string, snapshot *state.Snapshot, currentChecksum string, config *protocol.ActiveConfigResponse) error {
rendered, err := renderActiveConfig(config)
if err != nil {
return err
@@ -238,49 +161,8 @@ func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool,
mainConfigChecksum := checksumString(rendered.mainConfig)
routeConfigChecksum := checksumString(rendered.routeConfig)
slog.Info("applying new openresty config", "mode", mode, "from_version", snapshot.CurrentVersion, "to_version", config.Version, "old_checksum", currentChecksum, "new_checksum", config.Checksum)
outcome := s.nginxManager.Apply(ctx, rendered.mainConfig, rendered.routeConfig, rendered.supportFiles)
message := strings.TrimSpace(outcome.Message)
if outcome.Status == "" {
outcome.Status = nginx.ApplyStatusFatal
if message == "" {
message = "openresty apply returned empty outcome"
}
}
reportResult := ApplyResultFailed
switch outcome.Status {
case nginx.ApplyStatusSuccess:
slog.Info("openresty config applied successfully", "mode", mode, "version", config.Version)
snapshot.CurrentVersion = config.Version
snapshot.CurrentChecksum = config.Checksum
clearBlockedTarget(snapshot)
snapshot.LastError = ""
snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
snapshot.OpenrestyMessage = ""
reportResult = ApplyResultSuccess
if message == "" {
message = "apply success"
}
case nginx.ApplyStatusWarning:
if message == "" {
message = "apply rolled back to previous config"
}
slog.Warn("openresty config apply rolled back", "mode", mode, "version", config.Version, "message", message)
markBlockedTarget(snapshot, config.Version, config.Checksum, message)
snapshot.LastError = message
snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
snapshot.OpenrestyMessage = message
reportResult = ApplyResultWarning
default:
if message == "" {
message = "openresty apply failed"
}
slog.Error("apply openresty config failed", "mode", mode, "version", config.Version, "message", message)
markBlockedTarget(snapshot, config.Version, config.Checksum, message)
snapshot.LastError = message
snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
snapshot.OpenrestyMessage = message
}
outcome, message := normalizeApplyOutcome(s.nginxManager.Apply(ctx, rendered.mainConfig, rendered.routeConfig, rendered.supportFiles))
applyResult := updateSnapshotFromApplyOutcome(mode, snapshot, config, outcome, message)
if err := s.stateStore.Save(snapshot); err != nil {
return err
@@ -288,25 +170,25 @@ func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool,
if err := s.client.ReportApplyLog(ctx, protocol.ApplyLogPayload{
NodeID: snapshot.NodeID,
Version: config.Version,
Result: reportResult,
Message: message,
Result: applyResult.reportResult,
Message: applyResult.message,
Checksum: config.Checksum,
MainConfigChecksum: mainConfigChecksum,
RouteConfigChecksum: routeConfigChecksum,
SupportFileCount: len(rendered.supportFiles),
}); err != nil {
slog.Error("report apply log failed", "version", config.Version, "result", reportResult, "error", err)
slog.Error("report apply log failed", "version", config.Version, "result", applyResult.reportResult, "error", err)
return err
}
if reportResult == ApplyResultFailed {
if applyResult.reportResult == ApplyResultFailed {
slog.Warn("failed apply log reported", "version", config.Version)
return outcomeError(config.Version, message)
return outcomeError(config.Version, applyResult.message)
}
if err := s.syncReferencedWAFIPGroups(ctx, rendered.supportFiles); err != nil {
slog.Error("sync referenced waf ip groups failed", "version", config.Version, "error", err)
return err
}
slog.Debug("apply log reported", "version", config.Version, "result", reportResult)
slog.Debug("apply log reported", "version", config.Version, "result", applyResult.reportResult)
return nil
}
@@ -476,13 +358,13 @@ func (s *Service) ensureRuntimeForCurrentConfig(ctx context.Context, mode string
if err := s.nginxManager.EnsureRuntime(ctx, true); err != nil {
if strings.TrimSpace(snapshot.CurrentChecksum) == "" {
reason := fmt.Sprintf("blocked config %s has no historical config and current local config cannot start: %v", strings.TrimSpace(snapshot.BlockedVersion), err)
if fallbackErr := s.nginxManager.EnsureSafeFallbackRuntime(ctx, reason); fallbackErr == nil {
fallbackErr := s.nginxManager.EnsureSafeFallbackRuntime(ctx, reason)
if fallbackErr == nil {
snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
snapshot.OpenrestyMessage = "safe default fallback runtime started"
return nil
} else {
err = fmt.Errorf("%v; fallback recovery failed: %w", err, fallbackErr)
}
err = fmt.Errorf("%v; fallback recovery failed: %w", err, fallbackErr)
}
snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
snapshot.OpenrestyMessage = err.Error()
internal/apps/agent/sync/sync_helpers.go
+205
View File
@@ -0,0 +1,205 @@
package sync
import (
"context"
"log/slog"
"strings"
"github.com/Rain-kl/Wavelet/internal/apps/agent/nginx"
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
"github.com/Rain-kl/Wavelet/internal/apps/agent/state"
)
func syncMode(startup bool) string {
if startup {
return "startup"
}
return "periodic"
}
func normalizeSyncTarget(target *protocol.ActiveConfigMeta) {
if target == nil {
return
}
target.Version = strings.TrimSpace(target.Version)
target.Checksum = strings.TrimSpace(target.Checksum)
}
func (s *Service) loadSyncState() (*state.Snapshot, string, error) {
snapshot, err := s.stateStore.Load()
if err != nil {
return nil, "", err
}
currentChecksum, err := s.nginxManager.CurrentChecksum()
if err != nil {
return nil, "", err
}
return snapshot, currentChecksum, nil
}
func (s *Service) syncWithoutTarget(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string) error {
if !startup {
slog.Debug("skipping sync because heartbeat returned no active config summary", "mode", mode)
return nil
}
slog.Debug("sync startup fallback: active config summary unavailable, fetching active config directly")
config, fetchErr := s.client.GetActiveConfig(ctx)
if fetchErr != nil {
slog.Error("fetch active config failed", "mode", mode, "error", fetchErr)
return fetchErr
}
target := &protocol.ActiveConfigMeta{
Version: config.Version,
Checksum: config.Checksum,
}
return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
}
func (s *Service) syncMatchingChecksum(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string, target *protocol.ActiveConfigMeta) error {
if startup {
config, fetchErr := s.client.GetActiveConfig(ctx)
if fetchErr != nil {
slog.Error("fetch active config failed", "mode", mode, "error", fetchErr)
return fetchErr
}
return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
}
return s.finishUpToDateSync(ctx, mode, snapshot, target)
}
func (s *Service) finishUpToDateSync(ctx context.Context, mode string, snapshot *state.Snapshot, target *protocol.ActiveConfigMeta) error {
slog.Debug("local openresty config already up to date", "mode", mode, "version", target.Version)
if shouldReportNoopApply(snapshot, target.Version, target.Checksum) {
if err := s.reportNoopApply(ctx, snapshot.NodeID, target.Version, target.Checksum, "", "", 0); err != nil {
return err
}
}
snapshot.CurrentVersion = target.Version
snapshot.CurrentChecksum = target.Checksum
clearBlockedTarget(snapshot)
snapshot.LastError = ""
slog.Debug("sync finished without changes", "mode", mode, "version", target.Version)
return s.stateStore.Save(snapshot)
}
func (s *Service) syncMismatchedChecksum(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string, target *protocol.ActiveConfigMeta) error {
if isBlockedTarget(snapshot, target.Version, target.Checksum) {
slog.Warn("skipping blocked config version after previous failed apply", "mode", mode, "version", target.Version, "checksum", target.Checksum)
if startup {
if err := s.ensureRuntimeForCurrentConfig(ctx, mode, snapshot, currentChecksum); err != nil {
return err
}
return s.stateStore.Save(snapshot)
}
return nil
}
if hasBlockedTarget(snapshot) {
clearBlockedTarget(snapshot)
}
if snapshot.CurrentVersion == target.Version && snapshot.CurrentChecksum == target.Checksum && !startup {
slog.Debug("skipping config fetch because state already records target version/checksum", "version", target.Version, "checksum", target.Checksum)
return s.stateStore.Save(snapshot)
}
config, err := s.client.GetActiveConfig(ctx)
if err != nil {
slog.Error("fetch active config failed", "mode", mode, "error", err)
return err
}
return s.applyIfNeeded(ctx, mode, startup, snapshot, currentChecksum, target, config)
}
func (s *Service) handleUpToDateConfig(ctx context.Context, mode string, snapshot *state.Snapshot, config *protocol.ActiveConfigResponse) error {
slog.Debug("local openresty config already up to date", "mode", mode, "version", config.Version)
if shouldReportNoopApply(snapshot, config.Version, config.Checksum) {
rendered, renderErr := renderActiveConfig(config)
if renderErr != nil {
return renderErr
}
if err := s.reportNoopApply(
ctx,
snapshot.NodeID,
config.Version,
config.Checksum,
checksumString(rendered.mainConfig),
checksumString(rendered.routeConfig),
len(rendered.supportFiles),
); err != nil {
return err
}
}
snapshot.CurrentVersion = config.Version
snapshot.CurrentChecksum = config.Checksum
clearBlockedTarget(snapshot)
snapshot.LastError = ""
slog.Debug("sync finished without changes", "mode", mode, "version", config.Version)
return s.stateStore.Save(snapshot)
}
func (s *Service) handleBlockedConfigAfterFetch(ctx context.Context, mode string, startup bool, snapshot *state.Snapshot, currentChecksum string, config *protocol.ActiveConfigResponse) (bool, error) {
if !isBlockedTarget(snapshot, config.Version, config.Checksum) {
return false, nil
}
slog.Warn("skipping blocked config after fetch because the same version previously failed", "mode", mode, "version", config.Version, "checksum", config.Checksum)
if startup {
if err := s.ensureRuntimeForCurrentConfig(ctx, mode, snapshot, currentChecksum); err != nil {
return true, err
}
return true, s.stateStore.Save(snapshot)
}
return true, nil
}
type applyOutcomeResult struct {
reportResult string
message string
}
func normalizeApplyOutcome(outcome nginx.ApplyOutcome) (nginx.ApplyOutcome, string) {
message := strings.TrimSpace(outcome.Message)
if outcome.Status == "" {
outcome.Status = nginx.ApplyStatusFatal
if message == "" {
message = "openresty apply returned empty outcome"
}
}
return outcome, message
}
func updateSnapshotFromApplyOutcome(mode string, snapshot *state.Snapshot, config *protocol.ActiveConfigResponse, outcome nginx.ApplyOutcome, message string) applyOutcomeResult {
result := applyOutcomeResult{reportResult: ApplyResultFailed, message: message}
switch outcome.Status {
case nginx.ApplyStatusSuccess:
slog.Info("openresty config applied successfully", "mode", mode, "version", config.Version)
snapshot.CurrentVersion = config.Version
snapshot.CurrentChecksum = config.Checksum
clearBlockedTarget(snapshot)
snapshot.LastError = ""
snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
snapshot.OpenrestyMessage = ""
result.reportResult = ApplyResultSuccess
if result.message == "" {
result.message = "apply success"
}
case nginx.ApplyStatusWarning:
if result.message == "" {
result.message = "apply rolled back to previous config"
}
slog.Warn("openresty config apply rolled back", "mode", mode, "version", config.Version, "message", result.message)
markBlockedTarget(snapshot, config.Version, config.Checksum, result.message)
snapshot.LastError = result.message
snapshot.OpenrestyStatus = protocol.OpenrestyStatusHealthy
snapshot.OpenrestyMessage = result.message
result.reportResult = ApplyResultWarning
default:
if result.message == "" {
result.message = "openresty apply failed"
}
slog.Error("apply openresty config failed", "mode", mode, "version", config.Version, "message", result.message)
markBlockedTarget(snapshot, config.Version, config.Checksum, result.message)
snapshot.LastError = result.message
snapshot.OpenrestyStatus = protocol.OpenrestyStatusUnhealthy
snapshot.OpenrestyMessage = result.message
}
return result
}
internal/apps/agent/updater/updater.go
+7 -2
View File
@@ -1,17 +1,22 @@
// Package updater provides agent self-update integration with the edge updater.
package updater
import (
edgeupdater "github.com/Rain-kl/Wavelet/internal/apps/edge/updater"
"github.com/Rain-kl/Wavelet/internal/apps/agent/config"
edgeupdater "github.com/Rain-kl/Wavelet/internal/apps/edge/updater"
)
// Service is an alias for the edge updater service type used by the agent.
type Service = edgeupdater.Service
// UpdateOptions is an alias for the edge updater options type.
type UpdateOptions = edgeupdater.UpdateOptions
// New creates and returns a new agent updater Service with the agent-specific configuration.
func New() *Service {
return edgeupdater.New(edgeupdater.Config{
LocalVersion: config.Version,
AssetPrefix: "openflare-agent",
LogLabel: "agent",
})
}
}
internal/apps/agent/wsclient/client.go
+12 -1
View File
@@ -1,3 +1,4 @@
// Package wsclient provides the agent-side WebSocket client for connecting to the OpenFlare server.
package wsclient
import (
@@ -8,28 +9,38 @@ import (
edgews "github.com/Rain-kl/Wavelet/internal/apps/edge/wsclient"
)
// WSMessage is an alias for the WebSocket message type.
type WSMessage = edgews.WSMessage
// MessageHandler is an alias for the WebSocket message handler function type.
type MessageHandler = edgews.MessageHandler
// Connection is an alias for the AgentConnection interface.
type Connection = edgews.AgentConnection
// Client wraps the connection client for agent WebSockets.
type Client struct {
inner *edgews.Client
}
// New creates a new Client instance.
func New(baseURL, token string, timeout time.Duration) *Client {
return &Client{
inner: edgews.New(edgews.PresetAgent, baseURL, token, timeout),
}
}
// SetToken updates the client's token.
func (c *Client) SetToken(token string) {
c.inner.SetToken(token)
}
// URL returns the WebSocket client's target URL.
func (c *Client) URL() string {
return c.inner.URL()
}
// Connect establishes a WebSocket connection to the server and returns the connection handle.
func (c *Client) Connect(ctx context.Context) (protocol.WebSocketConnection, error) {
return c.inner.ConnectAgent(ctx)
}
}