修复 Agent 部署 Pages 问题

This commit is contained in:
ryan
2026-06-21 11:53:54 +08:00
parent 36f11c6ecb
commit 13a375e042
19 changed files with 755 additions and 59 deletions
+123 -19
View File
@@ -18,6 +18,7 @@ import (
"strings"
"github.com/Rain-kl/Wavelet/internal/apps/agent/protocol"
"github.com/Rain-kl/Wavelet/internal/apps/agent/state"
)
const (
@@ -45,10 +46,85 @@ type pagesDeploymentMarker struct {
Checksum string `json:"checksum"`
}
func (s *Service) syncPagesDeployments(ctx context.Context, config *protocol.ActiveConfigResponse) error {
deployments, err := referencedPagesDeployments(config)
if err != nil {
return err
func pagesDeploymentStateHash(item state.PagesDeployment) string {
if hash := strings.TrimSpace(item.Hash); hash != "" {
return hash
}
return strings.TrimSpace(item.Checksum)
}
func snapshotPagesDeployments(snapshot *state.Snapshot) []pagesDeploymentSource {
if snapshot == nil || snapshot.PagesDeployments == nil {
return nil
}
result := make([]pagesDeploymentSource, 0, len(snapshot.PagesDeployments))
for _, item := range snapshot.PagesDeployments {
result = append(result, pagesDeploymentSource{
DeploymentID: item.DeploymentID,
Checksum: pagesDeploymentStateHash(item),
})
}
return result
}
func setSnapshotPagesDeployments(snapshot *state.Snapshot, deployments []pagesDeploymentSource) {
if snapshot == nil {
return
}
if len(deployments) == 0 {
snapshot.PagesDeployments = []state.PagesDeployment{}
return
}
snapshot.PagesDeployments = make([]state.PagesDeployment, len(deployments))
for i, deployment := range deployments {
snapshot.PagesDeployments[i] = state.PagesDeployment{
DeploymentID: deployment.DeploymentID,
Hash: strings.TrimSpace(deployment.Checksum),
}
}
}
func updateSnapshotPagesDeploymentHash(snapshot *state.Snapshot, deployment pagesDeploymentSource) {
if snapshot == nil || snapshot.PagesDeployments == nil {
return
}
hash := strings.TrimSpace(deployment.Checksum)
for i := range snapshot.PagesDeployments {
if snapshot.PagesDeployments[i].DeploymentID != deployment.DeploymentID {
continue
}
snapshot.PagesDeployments[i].Hash = hash
snapshot.PagesDeployments[i].Checksum = ""
return
}
}
func pagesDiscoveryNeeded(snapshot *state.Snapshot) bool {
return snapshot == nil || snapshot.PagesDeployments == nil
}
func pagesSyncNeeded(snapshot *state.Snapshot) bool {
return snapshot != nil && snapshot.PagesDeployments != nil && len(snapshot.PagesDeployments) > 0
}
func pagesReconcileNeeded(snapshot *state.Snapshot) bool {
if pagesDiscoveryNeeded(snapshot) {
return true
}
return pagesSyncNeeded(snapshot)
}
func (s *Service) syncPagesDeployments(ctx context.Context, snapshot *state.Snapshot, config *protocol.ActiveConfigResponse) error {
var deployments []pagesDeploymentSource
var err error
if config != nil {
deployments, err = referencedPagesDeployments(config)
if err != nil {
return err
}
setSnapshotPagesDeployments(snapshot, deployments)
} else {
deployments = snapshotPagesDeployments(snapshot)
}
if len(deployments) == 0 {
return nil
@@ -57,32 +133,60 @@ func (s *Service) syncPagesDeployments(ctx context.Context, config *protocol.Act
return errors.New("pages_dir is required when active config references Pages deployments")
}
for _, deployment := range deployments {
if err := s.ensurePagesDeployment(ctx, deployment); err != nil {
if err := s.ensurePagesDeployment(ctx, snapshot, deployment); err != nil {
return err
}
}
return nil
}
func (s *Service) ensurePagesDeployment(ctx context.Context, deployment pagesDeploymentSource) error {
currentDir := pagesCurrentDir(s.pagesDir, deployment.DeploymentID)
if markerMatches(currentDir, deployment) {
return nil
}
packageBytes, err := s.client.DownloadPagesDeploymentPackage(ctx, deployment.DeploymentID)
func (s *Service) ensurePagesDeployment(ctx context.Context, snapshot *state.Snapshot, deployment pagesDeploymentSource) error {
serverHash, err := s.client.GetPagesDeploymentHash(ctx, deployment.DeploymentID)
if err != nil {
return fmt.Errorf("download Pages deployment %d: %w", deployment.DeploymentID, err)
return fmt.Errorf("fetch Pages deployment %d hash: %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)
serverHash = strings.TrimSpace(serverHash)
if serverHash == "" {
return fmt.Errorf("pages deployment %d hash is empty", deployment.DeploymentID)
}
releaseDir := pagesReleaseDir(s.pagesDir, deployment.DeploymentID, deployment.Checksum)
if !markerMatches(releaseDir, deployment) {
if err := extractPagesPackage(packageBytes, releaseDir, deployment); err != nil {
return err
effective := pagesDeploymentSource{
DeploymentID: deployment.DeploymentID,
Checksum: serverHash,
}
updateSnapshotPagesDeploymentHash(snapshot, effective)
releaseDir := pagesReleaseDir(s.pagesDir, effective.DeploymentID, effective.Checksum)
if pagesReleaseReady(releaseDir, effective) {
return switchPagesCurrentDir(s.pagesDir, effective.DeploymentID, releaseDir)
}
packageBytes, err := s.client.DownloadPagesDeploymentPackage(ctx, effective.DeploymentID)
if err != nil {
return fmt.Errorf("download Pages deployment %d: %w", effective.DeploymentID, err)
}
if got := checksumBytes(packageBytes); got != effective.Checksum {
return fmt.Errorf("pages deployment %d checksum mismatch: expected %s, got %s", effective.DeploymentID, effective.Checksum, got)
}
if err := extractPagesPackage(packageBytes, releaseDir, effective); err != nil {
return err
}
return switchPagesCurrentDir(s.pagesDir, effective.DeploymentID, releaseDir)
}
func pagesReleaseReady(dir string, deployment pagesDeploymentSource) bool {
if !markerMatches(dir, deployment) {
return false
}
entries, err := os.ReadDir(dir) //nolint:gosec // dir is managed PagesDir
if err != nil {
return false
}
for _, entry := range entries {
if entry.Name() == ".openflare-pages.json" {
continue
}
return true
}
return switchPagesCurrentDir(s.pagesDir, deployment.DeploymentID, releaseDir)
return false
}
func referencedPagesDeployments(config *protocol.ActiveConfigResponse) ([]pagesDeploymentSource, error) {
+5 -1
View File
@@ -29,6 +29,7 @@ const (
// ConfigClient is the interface for communicating with the server control plane.
type ConfigClient interface {
GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error)
GetPagesDeploymentHash(ctx context.Context, deploymentID uint) (string, error)
DownloadPagesDeploymentPackage(ctx context.Context, deploymentID uint) ([]byte, error)
ReportApplyLog(ctx context.Context, payload protocol.ApplyLogPayload) error
SyncWAFIPGroups(ctx context.Context, payload protocol.WAFIPGroupSyncRequest) (*protocol.WAFIPGroupSyncResponse, error)
@@ -152,6 +153,9 @@ func (s *Service) applyIfNeeded(ctx context.Context, mode string, startup bool,
clearBlockedTarget(snapshot)
}
if snapshot.CurrentVersion == config.Version && snapshot.CurrentChecksum == config.Checksum && !startup {
if err := s.syncPagesDeployments(ctx, snapshot, config); err != nil {
return err
}
slog.Debug("skipping apply because state already records target version/checksum", "version", config.Version, "checksum", config.Checksum)
return s.stateStore.Save(snapshot)
}
@@ -163,7 +167,7 @@ func (s *Service) applyRenderedConfig(ctx context.Context, mode string, snapshot
if err != nil {
return err
}
if err := s.syncPagesDeployments(ctx, config); err != nil {
if err := s.syncPagesDeployments(ctx, snapshot, config); err != nil {
return err
}
mainConfigChecksum := checksumString(rendered.mainConfig)
+227 -9
View File
@@ -31,7 +31,9 @@ type fakeClient struct {
config protocol.ActiveConfigResponse
reports []protocol.ApplyLogPayload
pagesPackages map[uint][]byte
pagesHashes map[uint]string
fetchCalls int
hashCalls int
}
type fakeManager struct {
@@ -76,6 +78,21 @@ func (f *fakeClient) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfi
return &f.config, nil
}
func (f *fakeClient) GetPagesDeploymentHash(ctx context.Context, deploymentID uint) (string, error) {
f.hashCalls++
if f.pagesHashes != nil {
if hash, ok := f.pagesHashes[deploymentID]; ok {
return hash, nil
}
}
if f.pagesPackages != nil {
if packageBytes, ok := f.pagesPackages[deploymentID]; ok {
return testBytesChecksum(packageBytes), nil
}
}
return "", fmt.Errorf("missing Pages hash %d", deploymentID)
}
func (f *fakeClient) DownloadPagesDeploymentPackage(ctx context.Context, deploymentID uint) ([]byte, error) {
if f.pagesPackages == nil {
return nil, fmt.Errorf("missing Pages package %d", deploymentID)
@@ -501,8 +518,8 @@ func TestSyncOnceReportsNoopWhenVersionChangesButChecksumMatches(t *testing.T) {
}); err != nil {
t.Fatalf("SyncOnce failed: %v", err)
}
if client.fetchCalls != 0 {
t.Fatalf("expected checksum match to skip config fetch, got %d", client.fetchCalls)
if client.fetchCalls != 1 {
t.Fatalf("expected checksum match to fetch active config once for Pages reconciliation, got %d", client.fetchCalls)
}
if len(manager.applyMainContents) != 0 {
t.Fatal("expected checksum match to skip apply")
@@ -568,9 +585,10 @@ func TestSyncOnceDoesNotRepeatNoopReportWhenStateAlreadyMatches(t *testing.T) {
t.Fatalf("EnsureNodeID failed: %v", err)
}
if err = stateStore.Save(&state.Snapshot{
NodeID: nodeID,
CurrentVersion: "20260309-003",
CurrentChecksum: "checksum-3",
NodeID: nodeID,
CurrentVersion: "20260309-003",
CurrentChecksum: "checksum-3",
PagesDeployments: []state.PagesDeployment{},
}); err != nil {
t.Fatalf("failed to seed state: %v", err)
}
@@ -583,6 +601,9 @@ func TestSyncOnceDoesNotRepeatNoopReportWhenStateAlreadyMatches(t *testing.T) {
}); err != nil {
t.Fatalf("SyncOnce failed: %v", err)
}
if client.fetchCalls != 0 {
t.Fatalf("expected no active config fetch when Pages releases are already reconciled, got %d", client.fetchCalls)
}
if len(client.reports) != 0 {
t.Fatalf("expected matching state to skip duplicate noop report, got %+v", client.reports)
}
@@ -907,9 +928,10 @@ func TestSyncOnceSkipsFetchWhenHeartbeatChecksumMatches(t *testing.T) {
t.Fatalf("EnsureNodeID failed: %v", err)
}
if err = stateStore.Save(&state.Snapshot{
NodeID: nodeID,
CurrentVersion: client.config.Version,
CurrentChecksum: client.config.Checksum,
NodeID: nodeID,
CurrentVersion: client.config.Version,
CurrentChecksum: client.config.Checksum,
PagesDeployments: []state.PagesDeployment{},
}); err != nil {
t.Fatalf("failed to seed state: %v", err)
}
@@ -923,13 +945,209 @@ func TestSyncOnceSkipsFetchWhenHeartbeatChecksumMatches(t *testing.T) {
t.Fatalf("SyncOnce failed: %v", err)
}
if client.fetchCalls != 0 {
t.Fatalf("expected no active config fetch when heartbeat checksum matches, got %d", client.fetchCalls)
t.Fatalf("expected no active config fetch when Pages releases are already reconciled, got %d", client.fetchCalls)
}
if len(manager.applyMainContents) != 0 {
t.Fatal("expected checksum match to skip apply")
}
if len(client.reports) != 0 {
t.Fatal("expected no apply log when no config change is needed")
}
}
func TestSyncOnceDownloadsPagesDeploymentWhenChecksumMatches(t *testing.T) {
packageBytes := testPagesPackage(t, map[string]string{"index.html": "hello"})
checksum := testBytesChecksum(packageBytes)
client := &fakeClient{
config: protocol.ActiveConfigResponse{
Version: "20260309-101",
Checksum: "pages-config-checksum",
SourceConfigJSON: testPagesSourceConfigJSON(7, checksum),
CreatedAt: time.Now().Format(time.RFC3339),
},
pagesPackages: map[uint][]byte{7: packageBytes},
}
stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
nodeID, err := stateStore.EnsureNodeID()
if err != nil {
t.Fatalf("EnsureNodeID failed: %v", err)
}
if err = stateStore.Save(&state.Snapshot{
NodeID: nodeID,
CurrentVersion: client.config.Version,
CurrentChecksum: client.config.Checksum,
}); err != nil {
t.Fatalf("save state failed: %v", err)
}
manager := &fakeManager{currentChecksum: client.config.Checksum}
service := New(client, manager, stateStore)
pagesDir := t.TempDir()
service.SetPagesDir(pagesDir)
if err = service.SyncOnce(context.Background(), &protocol.ActiveConfigMeta{
Version: client.config.Version,
Checksum: client.config.Checksum,
}); err != nil {
t.Fatalf("SyncOnce failed: %v", err)
}
data, err := os.ReadFile(filepath.Join(pagesDir, "deployments", "7", "current", "index.html"))
if err != nil {
t.Fatalf("expected Pages file to be extracted when checksum already matches: %v", err)
}
if string(data) != "hello" {
t.Fatalf("unexpected Pages file content: %s", string(data))
}
if len(manager.applyMainContents) != 0 {
t.Fatal("expected checksum match to skip OpenResty apply")
}
if client.fetchCalls != 1 {
t.Fatalf("expected one active config fetch on first reconcile, got %d", client.fetchCalls)
}
client.fetchCalls = 0
if err = service.SyncOnce(context.Background(), &protocol.ActiveConfigMeta{
Version: client.config.Version,
Checksum: client.config.Checksum,
}); err != nil {
t.Fatalf("second SyncOnce failed: %v", err)
}
if client.fetchCalls != 0 {
t.Fatalf("expected no active config fetch after Pages release is ready, got %d", client.fetchCalls)
}
snapshot, err := stateStore.Load()
if err != nil {
t.Fatalf("failed to load state: %v", err)
}
if len(snapshot.PagesDeployments) != 1 || snapshot.PagesDeployments[0].DeploymentID != 7 || snapshot.PagesDeployments[0].Hash != checksum {
t.Fatalf("expected Pages deployment refs to be cached in state, got %+v", snapshot.PagesDeployments)
}
if client.hashCalls == 0 {
t.Fatal("expected Pages hash check during reconcile")
}
}
func TestSyncOnceRedownloadsPagesDeploymentWhenServerHashChanges(t *testing.T) {
initialPackage := testPagesPackage(t, map[string]string{"index.html": "v1"})
updatedPackage := testPagesPackage(t, map[string]string{"index.html": "v2"})
initialHash := testBytesChecksum(initialPackage)
updatedHash := testBytesChecksum(updatedPackage)
deploymentID := uint(12)
client := &fakeClient{
config: protocol.ActiveConfigResponse{
Version: "20260309-107",
Checksum: "pages-config-checksum",
SourceConfigJSON: testPagesSourceConfigJSON(deploymentID, initialHash),
CreatedAt: time.Now().Format(time.RFC3339),
},
pagesPackages: map[uint][]byte{deploymentID: initialPackage},
pagesHashes: map[uint]string{deploymentID: initialHash},
}
stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
nodeID, err := stateStore.EnsureNodeID()
if err != nil {
t.Fatalf("EnsureNodeID failed: %v", err)
}
if err = stateStore.Save(&state.Snapshot{
NodeID: nodeID,
CurrentVersion: client.config.Version,
CurrentChecksum: client.config.Checksum,
PagesDeployments: []state.PagesDeployment{{
DeploymentID: deploymentID,
Hash: initialHash,
}},
}); err != nil {
t.Fatalf("save state failed: %v", err)
}
pagesDir := t.TempDir()
releaseDir := pagesReleaseDir(pagesDir, deploymentID, initialHash)
if err = extractPagesPackage(initialPackage, releaseDir, pagesDeploymentSource{
DeploymentID: deploymentID,
Checksum: initialHash,
}); err != nil {
t.Fatalf("seed release failed: %v", err)
}
if err = switchPagesCurrentDir(pagesDir, deploymentID, releaseDir); err != nil {
t.Fatalf("seed current dir failed: %v", err)
}
client.pagesPackages[deploymentID] = updatedPackage
client.pagesHashes[deploymentID] = updatedHash
manager := &fakeManager{currentChecksum: client.config.Checksum}
service := New(client, manager, stateStore)
service.SetPagesDir(pagesDir)
if err = service.SyncOnce(context.Background(), &protocol.ActiveConfigMeta{
Version: client.config.Version,
Checksum: client.config.Checksum,
}); err != nil {
t.Fatalf("SyncOnce failed: %v", err)
}
data, err := os.ReadFile(filepath.Join(pagesDir, "deployments", "12", "current", "index.html"))
if err != nil {
t.Fatalf("expected updated Pages file: %v", err)
}
if string(data) != "v2" {
t.Fatalf("unexpected Pages file content after hash change: %s", string(data))
}
if client.fetchCalls != 0 {
t.Fatalf("expected hash-only reconcile without active config fetch, got %d", client.fetchCalls)
}
}
func TestSyncOnceRedownloadsPagesDeploymentWhenReleaseDirOnlyHasMarker(t *testing.T) {
packageBytes := testPagesPackage(t, map[string]string{"index.html": "hello"})
checksum := testBytesChecksum(packageBytes)
deploymentID := uint(11)
client := &fakeClient{
config: protocol.ActiveConfigResponse{
Version: "20260309-106",
Checksum: "pages-config-checksum",
SourceConfigJSON: testPagesSourceConfigJSON(deploymentID, checksum),
CreatedAt: time.Now().Format(time.RFC3339),
},
pagesPackages: map[uint][]byte{deploymentID: packageBytes},
}
stateStore := state.NewStore(filepath.Join(t.TempDir(), "state.json"))
nodeID, err := stateStore.EnsureNodeID()
if err != nil {
t.Fatalf("EnsureNodeID failed: %v", err)
}
if err = stateStore.Save(&state.Snapshot{
NodeID: nodeID,
CurrentVersion: client.config.Version,
CurrentChecksum: client.config.Checksum,
}); err != nil {
t.Fatalf("save state failed: %v", err)
}
pagesDir := t.TempDir()
releaseDir := pagesReleaseDir(pagesDir, deploymentID, checksum)
if err = os.MkdirAll(releaseDir, pagesDirPerm); err != nil {
t.Fatalf("mkdir release dir failed: %v", err)
}
if err = writePagesMarker(releaseDir, pagesDeploymentSource{DeploymentID: deploymentID, Checksum: checksum}); err != nil {
t.Fatalf("write marker failed: %v", err)
}
if pagesReleaseReady(releaseDir, pagesDeploymentSource{DeploymentID: deploymentID, Checksum: checksum}) {
t.Fatal("expected marker-only release dir to be treated as not ready")
}
manager := &fakeManager{currentChecksum: client.config.Checksum}
service := New(client, manager, stateStore)
service.SetPagesDir(pagesDir)
if err = service.SyncOnce(context.Background(), &protocol.ActiveConfigMeta{
Version: client.config.Version,
Checksum: client.config.Checksum,
}); err != nil {
t.Fatalf("SyncOnce failed: %v", err)
}
data, err := os.ReadFile(filepath.Join(pagesDir, "deployments", "11", "current", "index.html"))
if err != nil {
t.Fatalf("expected Pages file to be extracted after marker-only release dir: %v", err)
}
if string(data) != "hello" {
t.Fatalf("unexpected Pages file content: %s", string(data))
}
}
func testPagesPackage(t *testing.T, files map[string]string) []byte {
t.Helper()
var buffer bytes.Buffer
+32
View File
@@ -68,6 +68,20 @@ func (s *Service) syncMatchingChecksum(ctx context.Context, mode string, startup
}
func (s *Service) finishUpToDateSync(ctx context.Context, mode string, snapshot *state.Snapshot, target *protocol.ActiveConfigMeta) error {
if pagesReconcileNeeded(snapshot) {
if pagesDiscoveryNeeded(snapshot) {
config, err := s.client.GetActiveConfig(ctx)
if err != nil {
slog.Error("fetch active config failed", "mode", mode, "error", err)
return err
}
if err := s.syncPagesDeployments(ctx, snapshot, config); err != nil {
return err
}
} else if err := s.syncPagesDeployments(ctx, snapshot, nil); err != nil {
return err
}
}
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 {
@@ -97,6 +111,21 @@ func (s *Service) syncMismatchedChecksum(ctx context.Context, mode string, start
clearBlockedTarget(snapshot)
}
if snapshot.CurrentVersion == target.Version && snapshot.CurrentChecksum == target.Checksum && !startup {
if pagesReconcileNeeded(snapshot) {
if pagesDiscoveryNeeded(snapshot) {
config, fetchErr := s.client.GetActiveConfig(ctx)
if fetchErr != nil {
slog.Error("fetch active config failed", "mode", mode, "error", fetchErr)
return fetchErr
}
if err := s.syncPagesDeployments(ctx, snapshot, config); err != nil {
return err
}
} else if err := s.syncPagesDeployments(ctx, snapshot, nil); err != nil {
return err
}
return s.stateStore.Save(snapshot)
}
slog.Debug("skipping config fetch because state already records target version/checksum", "version", target.Version, "checksum", target.Checksum)
return s.stateStore.Save(snapshot)
}
@@ -110,6 +139,9 @@ func (s *Service) syncMismatchedChecksum(ctx context.Context, mode string, start
}
func (s *Service) handleUpToDateConfig(ctx context.Context, mode string, snapshot *state.Snapshot, config *protocol.ActiveConfigResponse) error {
if err := s.syncPagesDeployments(ctx, snapshot, config); err != nil {
return err
}
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)