From c94fe0df5a8be41dcf9f6cd0b95b678f9a373a0f Mon Sep 17 00:00:00 2001 From: Ali Mahdavi Date: Thu, 13 Aug 2026 23:07:36 +0200 Subject: [PATCH] fix fresh cluster bootstrap retries --- cmd/flux-agent/bootstrap_identity.go | 23 +++++++- cmd/flux-agent/bootstrap_identity_test.go | 68 +++++++++++++++++++++++ cmd/flux-agent/etcd_start.go | 24 +++++--- cmd/flux-agent/proxy.go | 15 ++++- cmd/flux-agent/recovery_test.go | 40 +++++++++++++ 5 files changed, 161 insertions(+), 9 deletions(-) diff --git a/cmd/flux-agent/bootstrap_identity.go b/cmd/flux-agent/bootstrap_identity.go index a1fbd98..b86d2c3 100644 --- a/cmd/flux-agent/bootstrap_identity.go +++ b/cmd/flux-agent/bootstrap_identity.go @@ -54,7 +54,7 @@ func localNodeIdentity(cfg *config.Config) nodeIdentity { PatroniScope: cfg.PatroniScope, NodeName: cfg.MyName, PGDataEmpty: directoryIsEmpty("/var/lib/postgresql/data"), - EtcdDataEmpty: directoryIsEmpty("/var/lib/etcd"), + EtcdDataEmpty: etcdDataDirectoryIsFresh("/var/lib/etcd"), MembershipView: membershipNames(cfg.EtcdInitialCluster), } if !identity.PGDataEmpty { @@ -88,6 +88,27 @@ func directoryIsEmpty(path string) bool { return err == nil && len(entries) == 0 } +// etcdDataDirectoryIsFresh applies the same strict empty-directory policy as +// directoryIsEmpty, except for the agent-owned .etcd3_api breadcrumb. The +// marker is written before execing etcd and is not etcd state; treating it as +// preserved state prevents an otherwise fresh node from ever retrying a safe +// bootstrap after an unsuccessful etcd launch. +func etcdDataDirectoryIsFresh(path string) bool { + entries, err := os.ReadDir(path) + if os.IsNotExist(err) { + return true + } + if err != nil { + return false + } + for _, entry := range entries { + if entry.Name() != ".etcd3_api" || !entry.Type().IsRegular() { + return false + } + } + return true +} + // identityProbeToken authenticates the read-only peer probe without exposing // the replication password itself. The endpoint still returns no secrets, but // requiring a shared HMAC prevents arbitrary Internet clients from inventorying diff --git a/cmd/flux-agent/bootstrap_identity_test.go b/cmd/flux-agent/bootstrap_identity_test.go index b8f5d7b..a74a606 100644 --- a/cmd/flux-agent/bootstrap_identity_test.go +++ b/cmd/flux-agent/bootstrap_identity_test.go @@ -153,6 +153,34 @@ func TestDirectoryIsEmptyIsStrict(t *testing.T) { } } +func TestEtcdDataDirectoryIsFreshAllowsOnlyAgentMarker(t *testing.T) { + dir := t.TempDir() + marker := filepath.Join(dir, ".etcd3_api") + if err := os.WriteFile(marker, []byte("etcd3\n"), 0o600); err != nil { + t.Fatal(err) + } + if !etcdDataDirectoryIsFresh(dir) { + t.Fatal("the agent-owned marker must not count as persisted etcd state") + } + + if err := os.Mkdir(filepath.Join(dir, "member"), 0o700); err != nil { + t.Fatal(err) + } + if etcdDataDirectoryIsFresh(dir) { + t.Fatal("any etcd state alongside the marker must block fresh bootstrap") + } +} + +func TestEtcdDataDirectoryIsFreshRejectsMarkerDirectory(t *testing.T) { + dir := t.TempDir() + if err := os.Mkdir(filepath.Join(dir, ".etcd3_api"), 0o700); err != nil { + t.Fatal(err) + } + if etcdDataDirectoryIsFresh(dir) { + t.Fatal("only the regular marker file may be ignored") + } +} + func TestIdentityEndpointRequiresSharedProbeToken(t *testing.T) { cfg := bootstrapTestConfig() @@ -178,3 +206,43 @@ func TestIdentityEndpointRequiresSharedProbeToken(t *testing.T) { t.Fatalf("identity = %+v, want app=%q node=%q", identity, cfg.AppName, cfg.MyName) } } + +func TestIdentityEndpointReloadsCurrentCredentialsAndMembership(t *testing.T) { + startup := bootstrapTestConfig() + startup.PostgresReplicationPassword = "stale-secret" + startup.EtcdInitialCluster = "node-10-0-0-1=https://10.0.0.1:12380" + + current := *startup + current.PostgresReplicationPassword = "current-secret" + current.EtcdInitialCluster = "node-10-0-0-1=https://10.0.0.1:12380," + + "node-10-0-0-2=https://10.0.0.2:12380," + + "node-10-0-0-3=https://10.0.0.3:12380" + + recorder := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, clusterIdentityPath, nil) + req.Header.Set("X-Flux-Cluster-Probe", identityProbeToken(¤t)) + loads := 0 + writeCurrentIdentityResponse(recorder, startup, req, func(cfg *config.Config) error { + loads++ + *cfg = current + return nil + }) + + if loads != 1 { + t.Fatalf("cluster_env reloads = %d, want 1", loads) + } + if recorder.Code != http.StatusOK { + t.Fatalf("identity status = %d, want %d", recorder.Code, http.StatusOK) + } + var identity nodeIdentity + if err := json.Unmarshal(recorder.Body.Bytes(), &identity); err != nil { + t.Fatal(err) + } + wantView := membershipNames(current.EtcdInitialCluster) + if !reflect.DeepEqual(identity.MembershipView, wantView) { + t.Fatalf("membership view = %v, want %v", identity.MembershipView, wantView) + } + if startup.PostgresReplicationPassword != "stale-secret" { + t.Fatal("identity reload mutated the proxy's shared startup config") + } +} diff --git a/cmd/flux-agent/etcd_start.go b/cmd/flux-agent/etcd_start.go index 464b9b8..5309868 100644 --- a/cmd/flux-agent/etcd_start.go +++ b/cmd/flux-agent/etcd_start.go @@ -334,19 +334,29 @@ func verifyLocalDataWithTempEtcd(cfg *config.Config, dataDir, token string, sslO // ETCD_INITIAL_CLUSTER string derived from actual registered members (needed // to satisfy etcd v3.3's "member count is unequal" check). func tryJoinExisting(cfg *config.Config, sslOpts, otherIPs []string, forceRejoin bool, maxRetries int) (string, string, error) { - delay := time.Duration(cfg.EtcdJoinRetryDelaySeconds) * time.Second + return tryJoinExistingWithLoader(cfg, sslOpts, otherIPs, forceRejoin, maxRetries, config.LoadClusterEnv) +} + +func tryJoinExistingWithLoader(cfg *config.Config, sslOpts, otherIPs []string, forceRejoin bool, maxRetries int, + loadClusterEnv func(*config.Config) error) (string, string, error) { if maxRetries < 1 { maxRetries = 1 } anyReachable := false - peerURL := fmt.Sprintf("%s://%s:%d", cfg.EtcdProtocol(), cfg.MyIP, cfg.HostEtcdPeerPort) - - if len(otherIPs) == 0 { - return "", "", myerrors.NewNoPeersReachable() - } for attempt := 1; attempt <= maxRetries; attempt++ { + // The reconciliation daemon may replace an incomplete startup topology + // while this process is waiting for Flux placement to settle. Refresh the + // inputs, not only the network operation, on every attempt. Keep the last + // usable snapshot if a concurrent read temporarily fails. + if err := loadClusterEnv(cfg); err != nil { + pkglog.Warnf("cannot refresh %s for peer discovery: %v", config.ClusterEnvFile, err) + } else { + otherIPs = otherIPsFromInitialCluster(cfg.EtcdInitialCluster, cfg.MyName) + } + delay := time.Duration(cfg.EtcdJoinRetryDelaySeconds) * time.Second + peerURL := fmt.Sprintf("%s://%s:%d", cfg.EtcdProtocol(), cfg.MyIP, cfg.HostEtcdPeerPort) pkglog.Infof("peer discovery attempt %d/%d", attempt, maxRetries) for _, ip := range otherIPs { endpoint := fmt.Sprintf("%s://%s:%d", cfg.EtcdProtocol(), ip, cfg.HostEtcdClientPort) @@ -451,7 +461,7 @@ func decideBootstrap(cfg *config.Config, dataDir string) string { expectedCount := len(strings.Split(cfg.EtcdInitialCluster, ",")) etcdDataEmpty := !fileExists(dataDir + "/member/snap/db") pgDataEmpty := !dirExists("/var/lib/postgresql/data/global") - strictlyEmpty := directoryIsEmpty(dataDir) && directoryIsEmpty("/var/lib/postgresql/data") + strictlyEmpty := etcdDataDirectoryIsFresh(dataDir) && directoryIsEmpty("/var/lib/postgresql/data") candidate := bootstrapCandidate(cfg.EtcdInitialCluster) diff --git a/cmd/flux-agent/proxy.go b/cmd/flux-agent/proxy.go index 31fe822..19c0299 100644 --- a/cmd/flux-agent/proxy.go +++ b/cmd/flux-agent/proxy.go @@ -257,7 +257,7 @@ func handleProxyConn(client net.Conn, primaryIP string, cfg *config.Config, host return } defer req.Body.Close() - writeIdentityResponse(&connResponseWriter{Conn: client}, cfg, req) + writeCurrentIdentityResponse(&connResponseWriter{Conn: client}, cfg, req, config.LoadClusterEnv) return } if primaryIP == "" { @@ -284,6 +284,19 @@ func handleProxyConn(client net.Conn, primaryIP string, cfg *config.Config, host <-done } +// writeCurrentIdentityResponse reloads the configuration snapshot used for +// authentication and membership reporting. The daemon can update cluster_env +// without restarting the proxy, so serving the startup snapshot here can make +// otherwise matching peers reject each other's probe tokens or topology. +func writeCurrentIdentityResponse(w http.ResponseWriter, cfg *config.Config, r *http.Request, + loadClusterEnv func(*config.Config) error) { + current := *cfg + if err := loadClusterEnv(¤t); err != nil { + pkglog.Warnf("proxy: could not refresh %s for identity probe: %v", config.ClusterEnvFile, err) + } + writeIdentityResponse(w, ¤t, r) +} + // connResponseWriter is the minimal http.ResponseWriter needed to serve the // identity document on the existing TCP proxy listener. type connResponseWriter struct { diff --git a/cmd/flux-agent/recovery_test.go b/cmd/flux-agent/recovery_test.go index 84d0d98..6cb5624 100644 --- a/cmd/flux-agent/recovery_test.go +++ b/cmd/flux-agent/recovery_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "errors" "strings" "testing" "time" @@ -9,6 +10,45 @@ import ( "github.com/RunOnFlux/flux-pg-cluster/internal/config" ) +func TestJoinDiscoveryReloadsTopologyOnEveryAttempt(t *testing.T) { + cfg := recoveryTestConfig() + cfg.EtcdInitialCluster = "node-82-67-53-6=https://82.67.53.6:12380" + cfg.EtcdJoinRetryDelaySeconds = 0 + + loads := 0 + _, _, err := tryJoinExistingWithLoader(cfg, nil, nil, false, 3, func(cfg *config.Config) error { + loads++ + // Keep this as a self-only view so the test performs no network I/O. The + // important regression is that an initially empty peer list no longer + // returns before loading, and that all retry attempts refresh the input. + cfg.EtcdInitialCluster = "node-82-67-53-6=https://82.67.53.6:12380" + return nil + }) + if err == nil { + t.Fatal("empty peer topology should still report no reachable peers") + } + if loads != 3 { + t.Fatalf("cluster_env reloads = %d, want one for each of 3 attempts", loads) + } +} + +func TestJoinDiscoveryRetriesWhenReloadFailsWithNoTopology(t *testing.T) { + cfg := recoveryTestConfig() + cfg.EtcdJoinRetryDelaySeconds = 0 + + loads := 0 + _, _, err := tryJoinExistingWithLoader(cfg, nil, nil, false, 2, func(*config.Config) error { + loads++ + return errors.New("temporary read failure") + }) + if err == nil { + t.Fatal("unreachable fallback peers should report no reachable peers") + } + if loads != 2 { + t.Fatalf("cluster_env reloads = %d, want 2", loads) + } +} + func TestBootstrapRetryDelayOutlivesSupervisorStartupWindow(t *testing.T) { if got := bootstrapRetryDelay(0); got != 2*time.Second { t.Fatalf("minimum retry delay = %s, want 2s", got)