diff --git a/README.md b/README.md index 6f6e50ed..005c8b27 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ This branch contains the compatibility runtime used by PastureStack Server to op ## Runtime image -PastureStack Server creates this image automatically when a user adds a load-balancer service: +The currently published runtime image used when PastureStack Server creates a load-balancer service is: ```text ghcr.io/pasturestack/load-balancer-service:v0.9.27 @@ -14,6 +14,11 @@ ghcr.io/pasturestack/load-balancer-service:v0.9.27 Release deployments pin the image by digest. The image is not intended to be started by itself because the control plane supplies metadata, API credentials, service links, port rules, and health-check configuration. +This branch prepares the `v0.9.28` source candidate for drain weight recovery. +The candidate has passed the focused local race tests and static checks described +below. Protected pull-request CI and native PastureStack load-balancer QA remain +release qualification gates; `v0.9.28` has not been tagged or published. + ## Compatibility The controller accepts role-specific credentials when the orchestration engine supplies them: @@ -27,6 +32,36 @@ It also accepts the generic `CATTLE_ACCESS_KEY` and `CATTLE_SECRET_KEY` pair use The HAProxy logging helper keeps its configuration under the standard `/etc/rsyslog.d` path and does not require a daemon PID file. Logging initialization is validated independently before HAProxy configuration is applied. +### Drain weight recovery (v0.9.28 source candidate) + +The controller records each backend's runtime weight before setting it to zero +for `target.drain`, including drains that already have zero active sessions. +After drain completion or timeout, a healthy endpoint still present in desired +metadata can recover its original weight even when its ID, IP, and generated +configuration have not changed. No reload is forced solely for this recovery. + +Only weights changed by this controller are eligible. Existing zero weights, +manual MAINT/DRAIN, configured `disabled` or weight zero, and an operator's later +nonzero weight are preserved. Active drains follow a HAProxy reload; a completed +drain does not adopt an unexplained zero in a different HAProxy process. Recovery +ownership is process-local and is not reconstructed by guessing after a controller +restart. + +Pending recovery retries every five seconds and after successful configuration +application. Runtime socket operations have a two-second deadline. A failed or +unverified weight change retains recovery ownership and produces an ERROR with +the endpoint, backend, and intended weight. Successful restoration is logged only +after runtime readback. A metadata target in `stopping` keeps desired weight zero +even when port rules are merged. This lifecycle zero retains recovery ownership +until the same endpoint returns to an eligible running state; an explicit user +weight zero or `disabled` remains protected. Across a HAProxy PID change, an +active drain is reapplied only to an unchanged configured initial weight, never +to a runtime custom weight observed in the replacement process. +If the desired configured weight changes while draining, recovery honors that +new configured weight instead of undoing the user's edit with the old snapshot. + +The backend state fields follow the [HAProxy Runtime API management contract](https://docs.haproxy.org/3.4/management.html#show%20servers%20state). + Protocol-critical `CATTLE_*` variables, `io.rancher.*` labels, metadata paths, API types, and provider identifiers remain compatibility interfaces. They do not represent PastureStack branding or affiliation. ## Build and test diff --git a/config/config.go b/config/config.go index d2e747d0..4837a588 100644 --- a/config/config.go +++ b/config/config.go @@ -74,6 +74,7 @@ type Endpoint struct { IsCname bool DrainTimeout string Weight int + Stopping bool } type FrontendService struct { diff --git a/controller/rancher/drain_weight_test.go b/controller/rancher/drain_weight_test.go new file mode 100644 index 00000000..0087e62b --- /dev/null +++ b/controller/rancher/drain_weight_test.go @@ -0,0 +1,52 @@ +package rancher + +import ( + "testing" + + "github.com/PastureStack/load-balancer-controller/internal/metadata" +) + +type drainWeightMetadata struct { + tMetaFetcher + state string +} + +func (mf drainWeightMetadata) GetService(link string) (*metadata.Service, error) { + ip := "10.42.0.1" + if link == "default/two" { + ip = "10.42.0.2" + } + return &metadata.Service{Kind: "service", State: "active", Containers: []metadata.Container{{Name: link, UUID: link, PrimaryIp: ip, State: mf.state}}}, nil +} + +func TestStoppingAndRestartedTargetsPreserveDesiredWeightsAcrossRuleMerges(t *testing.T) { + var oldNames []string + for _, state := range []string{"running", "stopping", "running", "starting"} { + c := &LoadBalancerController{MetaFetcher: drainWeightMetadata{state: state}, CertFetcher: tCertFetcher{}, LBProvider: &tProvider{}} + meta := &LBMetadata{PortRules: []metadata.PortRule{ + {Protocol: "http", Service: "default/one", SourcePort: 80, TargetPort: 8080, Weight: 17}, + {Protocol: "http", Service: "default/two", SourcePort: 80, TargetPort: 8080, Weight: 17}, + }} + configs, err := c.BuildConfigFromMetadata("lb", "env", "", "any", meta) + if err != nil { + t.Fatal(err) + } + endpoints := configs[0].FrontendServices[0].BackendServices[0].Endpoints + if len(endpoints) != 2 { + t.Fatalf("merged endpoint count=%d", len(endpoints)) + } + for i, ep := range endpoints { + want := 17 + if state == "stopping" { + want = 0 + } + if ep.Weight != want { + t.Fatalf("state=%s endpoint=%s weight=%d want=%d", state, ep.Name, ep.Weight, want) + } + if oldNames != nil && ep.Name != oldNames[i] { + t.Fatalf("state transition changed endpoint identity: %s", state) + } + } + oldNames = []string{endpoints[0].Name, endpoints[1].Name} + } +} diff --git a/controller/rancher/rancher.go b/controller/rancher/rancher.go index e1d748ff..65684b17 100644 --- a/controller/rancher/rancher.go +++ b/controller/rancher/rancher.go @@ -382,7 +382,9 @@ func (lbc *LoadBalancerController) BuildConfigFromMetadata(lbName, envUUID, self for _, ep := range eps { if _, ok := epMap[ep.IP]; !ok { epMap[ep.IP] = ep.IP - ep.Weight = rule.Weight + if !ep.Stopping { + ep.Weight = rule.Weight + } backend.Endpoints = append(backend.Endpoints, ep) } } @@ -408,7 +410,9 @@ func (lbc *LoadBalancerController) BuildConfigFromMetadata(lbName, envUUID, self epMap := make(map[string]string) for _, ep := range eps { epMap[ep.IP] = ep.IP - ep.Weight = rule.Weight + if !ep.Stopping { + ep.Weight = rule.Weight + } } allEps[pathUUID] = epMap } @@ -765,6 +769,7 @@ func getContainerEndpoint(c *metadata.Container, targetPort int, selfHostUUID st } if strings.EqualFold(c.State, "stopping") { ep.Weight = 0 + ep.Stopping = true } if localServicePreference != "any" && !strings.EqualFold(c.HostUUID, selfHostUUID) { diff --git a/provider/haproxy/drain_recovery.go b/provider/haproxy/drain_recovery.go new file mode 100644 index 00000000..722583c6 --- /dev/null +++ b/provider/haproxy/drain_recovery.go @@ -0,0 +1,298 @@ +package haproxy + +import ( + "errors" + "fmt" + "strconv" + "strings" + + "github.com/PastureStack/load-balancer-controller/config" + utils "github.com/PastureStack/load-balancer-controller/utils" + log "github.com/sirupsen/logrus" +) + +// Only a weight that this controller changed can be restored. In particular, +// an unowned zero weight, MAINT/DRAIN admin state or operator's nonzero weight +// is not evidence of a stale controller drain. +type drainWeight struct { + weight int + initialWeight int + pid string + retry bool +} + +type desiredEndpoint struct { + weight int + disabled bool + temporaryZero bool +} + +type serverState struct { + backend string + name string + weight int + initialWeight int + admin int + operational int +} + +// HAProxy's forced/inherited/configuration MAINT and forced/inherited DRAIN, +// plus an FQDN changed through the runtime API, belong to the operator/config. +// RMAINT (0x20) is a transient DNS failure: wait without discarding ownership. +const operatorAdminState = 0x1f | 0x40 + +func (dm *drainMgr) setDesired(lbConfig *config.LoadBalancerConfig) { + desired := make(map[string]desiredEndpoint) + for _, fe := range lbConfig.FrontendServices { + for _, be := range fe.BackendServices { + for _, ep := range be.Endpoints { + state := desiredEndpoint{weight: 1, temporaryZero: ep.Stopping} + fields := strings.Fields(ep.Config) + for i, field := range fields { + if field == "disabled" { + state.disabled = true + } + if field == "weight" && i+1 < len(fields) { + if weight, err := strconv.Atoi(fields[i+1]); err == nil { + state.weight = weight + } + } + } + desired[be.UUID+"/"+ep.Name] = state + } + } + } + dm.mu.Lock() + dm.desired = desired + dm.mu.Unlock() +} + +func (dm *drainMgr) readServerStates() (map[string]serverState, error) { + lines, err := dm.runCommand("show servers state") + if err != nil { + return nil, err + } + states := make(map[string]serverState) + var header []string + for _, line := range lines { + if strings.HasPrefix(line, "# ") { + header = strings.Fields(strings.TrimPrefix(line, "# ")) + continue + } + if header == nil || strings.TrimSpace(line) == "" { + continue + } + values := strings.Fields(line) + if len(values) != len(header) { + return nil, errors.New("invalid HAProxy server state column count") + } + fields := make(map[string]string, len(header)) + for i, key := range header { + fields[key] = values[i] + } + weight, weightErr := strconv.Atoi(fields["srv_uweight"]) + initialWeight, initialErr := strconv.Atoi(fields["srv_iweight"]) + admin, adminErr := strconv.Atoi(fields["srv_admin_state"]) + operational, opErr := strconv.Atoi(fields["srv_op_state"]) + if fields["be_name"] == "" || fields["srv_name"] == "" || weightErr != nil || initialErr != nil || adminErr != nil || opErr != nil || weight < 0 || weight > 256 || initialWeight < 0 || initialWeight > 256 || admin < 0 || operational < 0 || operational > 3 { + return nil, errors.New("invalid HAProxy server state fields") + } + key := fields["be_name"] + "/" + fields["srv_name"] + states[key] = serverState{backend: key, name: fields["srv_name"], weight: weight, initialWeight: initialWeight, admin: admin, operational: operational} + } + if header == nil { + return nil, errors.New("HAProxy server state header is missing") + } + for _, required := range []string{"be_name", "srv_name", "srv_uweight", "srv_iweight", "srv_admin_state", "srv_op_state"} { + found := false + for _, field := range header { + found = found || field == required + } + if !found { + return nil, fmt.Errorf("HAProxy server state header is missing %s", required) + } + } + return states, nil +} + +func (dm *drainMgr) setWeight(backend string, weight int) error { + lines, err := dm.runCommand(fmt.Sprintf("set server %s weight %d", backend, weight)) + if err != nil { + return err + } + for _, line := range lines { + if strings.TrimSpace(line) != "" { + return errors.New("HAProxy rejected weight change: " + utils.SanitizeLogText(line)) + } + } + return nil +} + +func (dm *drainMgr) setWeightVerified(backend string, weight int) error { + if err := dm.setWeight(backend, weight); err != nil { + return err + } + states, err := dm.readServerStates() + if err != nil { + return err + } + state, exists := states[backend] + if !exists || state.weight != weight { + return fmt.Errorf("weight verification failed: backend=%s wantWeight=%d", backend, weight) + } + return nil +} + +// Caller holds mu. Save provenance before issuing a command: a socket timeout +// can occur after HAProxy applied the zero, and must not lose its recovery key. +func (dm *drainMgr) drainEndpoint(name, pid string, states map[string]serverState) error { + if dm.weights == nil { + dm.weights = make(map[string]map[string]drainWeight) + } + if dm.weights[name] == nil { + dm.weights[name] = make(map[string]drainWeight) + } + var failures []error + for key, state := range states { + if state.name != name || state.admin != 0 || state.weight == 0 { + continue + } + if _, exists := dm.weights[name][key]; !exists { + dm.weights[name][key] = drainWeight{weight: state.weight, initialWeight: state.initialWeight, pid: pid} + } else { + owned := dm.weights[name][key] + owned.pid = pid + dm.weights[name][key] = owned + } + if err := dm.setWeightVerified(key, 0); err != nil { + failures = append(failures, fmt.Errorf("backend=%s: %w", key, err)) + } + } + return errors.Join(failures...) +} + +func (lbp *Provider) reconcileDrains() error { + if lbp.recoveryBlocked { + return nil + } + dm := lbp.drainMgr + dm.mu.Lock() + defer dm.mu.Unlock() + if len(dm.weights) == 0 || dm.desired == nil { + return nil + } + pid, err := lbp.cfg.readPid() + if err != nil { + return fmt.Errorf("read HAProxy pid for drain recovery: %w", err) + } + states, err := dm.readServerStates() + if err != nil { + return fmt.Errorf("read HAProxy state for drain recovery: %w", err) + } + var failures []error + for name, weights := range dm.weights { + _, active := dm.drainList[name] + if active { + // A reload replaces the process and resets runtime weights. Reapply + // active drains without losing the weight captured before draining. + for key, original := range weights { + state, exists := states[key] + desired, present := dm.desired[key] + if !exists || !present || desired.disabled || desired.weight == 0 && !desired.temporaryZero || state.admin&operatorAdminState != 0 { + delete(weights, key) + continue + } + if state.admin != 0 { + continue + } + if original.pid == pid && !original.retry { + if state.weight != 0 { + delete(weights, key) + } // Observed operator override after our verified zero. + continue + } + if original.pid != pid { + if desired.temporaryZero && desired.weight == 0 && state.initialWeight == 0 && state.weight == 0 { + // This replacement process was initialized from our stopping + // configuration, not an unexplained runtime zero. + original.pid = pid + original.retry = false + weights[key] = original + continue + } + if state.weight == 0 { + continue + } // An unowned zero in the replacement process. + if state.weight != state.initialWeight { + log.Infof("Preserved runtime weight override after reload: endpoint=%s backend=%s weight=%d initialWeight=%d", name, key, state.weight, state.initialWeight) + delete(weights, key) + continue + } + if state.initialWeight != desired.weight { + continue + } // Cannot prove this is the applied desired config. + } + // Only the original owned backends follow a reload; never adopt + // another backend's intentional zero/disabled runtime state. + original.pid = pid + original.retry = true + weights[key] = original + if state.weight != 0 { + if err := dm.setWeightVerified(key, 0); err != nil { + failures = append(failures, fmt.Errorf("reapply active drain backend=%s: %w", key, err)) + continue + } + } + original.retry = false + weights[key] = original + } + continue + } + for key, original := range weights { + desired, present := dm.desired[key] + state, exists := states[key] + if !present || !exists || desired.disabled || desired.weight == 0 && !desired.temporaryZero || state.admin&operatorAdminState != 0 || state.weight != 0 { + delete(weights, key) + continue + } + if desired.temporaryZero && desired.weight == 0 { + // DrainComplete and stopping metadata can arrive in either order. + // Keep the provenance while this lifecycle zero is desired. If a + // reload initialized that zero, bind only that proven generation. + if original.pid != pid && state.initialWeight == 0 { + original.pid = pid + weights[key] = original + } + continue + } + if original.pid != pid { + // No proof this controller zeroed this replacement process. A + // zero there can be an operator action; do not infer ownership. + log.Warnf("Drain recovery ownership expired: endpoint=%s backend=%s oldPID=%s currentPID=%s", name, key, original.pid, pid) + delete(weights, key) + continue + } + if state.admin != 0 || state.operational != 2 { + continue + } // Wait for actual health recovery. + restoreWeight := original.weight + if desired.weight != original.initialWeight { + // A later desired-config edit supersedes the old runtime weight. + // Unchanged config still recovers an operator's original custom + // runtime weight, rather than resetting it to the config default. + restoreWeight = desired.weight + } + if err := dm.setWeightVerified(key, restoreWeight); err != nil { + failures = append(failures, fmt.Errorf("restore endpoint=%s backend=%s weight=%d: %w", name, key, restoreWeight, err)) + continue + } + // Delete the only proof of drain ownership after verified readback. + delete(weights, key) + log.Infof("Recovered controller drain weight: endpoint=%s backend=%s weight=%d", name, key, restoreWeight) + } + if len(weights) == 0 { + delete(dm.weights, name) + } + } + return errors.Join(failures...) +} diff --git a/provider/haproxy/drain_recovery_test.go b/provider/haproxy/drain_recovery_test.go new file mode 100644 index 00000000..c13d8b33 --- /dev/null +++ b/provider/haproxy/drain_recovery_test.go @@ -0,0 +1,581 @@ +package haproxy + +import ( + "errors" + "fmt" + "net" + "os" + "path/filepath" + "reflect" + "runtime" + "sort" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/PastureStack/load-balancer-controller/config" +) + +type drainRuntime struct { + states map[string]serverState + commands []string + fail map[string]int + reject map[string]string + ignoreWrites bool + sessions int + initialWeights map[string]int +} + +func (r *drainRuntime) command(cmd string) ([]string, error) { + r.commands = append(r.commands, cmd) + if r.fail[cmd] > 0 { + r.fail[cmd]-- + return nil, errors.New("socket timed out") + } + if response, ok := r.reject[cmd]; ok { + return []string{response}, nil + } + switch cmd { + case "show stat": + rows := []string{"# pxname,svname,scur"} + for _, state := range r.states { + rows = append(rows, fmt.Sprintf("%s,%s,%d", strings.Split(state.backend, "/")[0], state.name, r.sessions)) + } + return rows, nil + case "show servers state": + rows := []string{"1", "# be_name srv_name srv_uweight srv_iweight srv_admin_state srv_op_state"} + var keys []string + for key := range r.states { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + state := r.states[key] + initial, ok := r.initialWeights[key] + if !ok { + initial = 1 + } + rows = append(rows, fmt.Sprintf("%s %s %d %d %d %d", strings.Split(key, "/")[0], state.name, state.weight, initial, state.admin, state.operational)) + } + return rows, nil + } + fields := strings.Fields(cmd) + if len(fields) == 5 && fields[0] == "set" && fields[1] == "server" && fields[3] == "weight" { + state, ok := r.states[fields[2]] + if !ok { + return []string{"No such server"}, nil + } + weight, err := strconv.Atoi(fields[4]) + if err != nil { + return nil, err + } + if !r.ignoreWrites { + state.weight = weight + r.states[fields[2]] = state + } + return nil, nil + } + return nil, fmt.Errorf("unexpected command %q", cmd) +} + +func recoveryFixture(t *testing.T) (*Provider, *drainRuntime, *config.Endpoint, *config.LoadBalancerConfig) { + t.Helper() + r := &drainRuntime{states: map[string]serverState{ + "be-one/target": {backend: "be-one/target", name: "target", weight: 7, operational: 2}, + "be-two/target": {backend: "be-two/target", name: "target", weight: 23, operational: 2}, + }, fail: make(map[string]int), reject: make(map[string]string), initialWeights: make(map[string]int)} + dir := t.TempDir() + pid := filepath.Join(dir, "haproxy.pid") + if err := os.WriteFile(pid, []byte("123\n"), 0600); err != nil { + t.Fatal(err) + } + p := &Provider{cfg: &haproxyConfig{PidFile: pid, CertDir: filepath.Join(dir, "certs"), Config: filepath.Join(dir, "haproxy.cfg"), Template: "test_data/haproxy_template.cfg", reloadCommand: func() error { return nil }}, drainMgr: &drainMgr{drainList: make(map[string]string), mu: &sync.RWMutex{}, command: r.command, olderProcesses: func() (bool, error) { return false, nil }}} + ep := &config.Endpoint{Name: "target", IP: "10.42.0.2", Port: 80, Weight: 1, Config: "check"} + cfg := &config.LoadBalancerConfig{FrontendServices: config.FrontendServices{&config.FrontendService{Protocol: config.HTTPProto, BackendServices: config.BackendServices{&config.BackendService{UUID: "be-one", Endpoints: config.Endpoints{ep}}, &config.BackendService{UUID: "be-two", Endpoints: config.Endpoints{ep}}}}}} + p.drainMgr.setDesired(cfg) + return p, r, ep, cfg +} + +func runtimeWeights(r *drainRuntime) map[string]int { + weights := make(map[string]int) + for key, state := range r.states { + weights[key] = state.weight + } + return weights +} + +func TestDrainWithoutSessionsStillRecoversBothBackendWeights(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + if p.DrainEndpoint(ep) { + t.Fatal("zero sessions must not delay the event reply") + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 0, "be-two/target": 0}) { + t.Fatal(runtimeWeights(r)) + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } + if len(p.drainMgr.weights) != 0 { + t.Fatal("verified recovery must clear ownership") + } +} + +func TestRepeatedDrainAndTimeoutPreserveOriginalWeights(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + r.sessions = 2 + if !p.DrainEndpoint(ep) || !p.DrainEndpoint(ep) { + t.Fatal("active duplicate drain not retained") + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states["be-one/target"].weight != 0 { + t.Fatal("active drain recovered early") + } + p.RemoveEndpointFromDrain(ep) + p.RemoveEndpointFromDrain(ep) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestDrainCompletionWhileStoppingRetainsOwnershipUntilSameTargetRuns(t *testing.T) { + p, r, ep, cfg := recoveryFixture(t) + r.sessions = 1 + if !p.DrainEndpoint(ep) { + t.Fatal("expected active drain") + } + // Metadata observes stopping before the drain-complete reply. These are + // separate ordered events, not a synthetic final-state fixture. + ep.Stopping = true + ep.Config = "check weight 0" + p.drainMgr.setDesired(cfg) + p.RemoveEndpointFromDrain(ep) + for i := 0; i < 2; i++ { + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if len(p.drainMgr.weights[ep.Name]) != 2 { + t.Fatal("stopping sync discarded original weights") + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 0, "be-two/target": 0}) { + t.Fatal("stopping endpoint recovered early") + } + } + // The same target comes back without a process/config reload to rescue it. + ep.Stopping = false + ep.Config = "check" + p.drainMgr.setDesired(cfg) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestActiveDrainReloadDoesNotOverwriteNewProcessManualWeight(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + r.sessions = 1 + if !p.DrainEndpoint(ep) { + t.Fatal("expected active drain") + } + if err := os.WriteFile(p.cfg.PidFile, []byte("456\n"), 0600); err != nil { + t.Fatal(err) + } + // Reload initializes both at configured weight 1. An operator then changes + // only one backend before the controller observes this new process. + for key, state := range r.states { + state.weight = 1 + r.states[key] = state + } + state := r.states["be-one/target"] + state.weight = 31 + r.states[state.backend] = state + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states["be-one/target"].weight != 31 { + t.Fatal("reload reconciliation overwrote manual weight") + } + if r.states["be-two/target"].weight != 0 { + t.Fatal("reload default did not resume active drain") + } + p.RemoveEndpointFromDrain(ep) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 31, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestCompletedDrainStoppingReloadRetainsOriginalWeightsUntilRunning(t *testing.T) { + p, r, ep, cfg := recoveryFixture(t) + r.sessions = 1 + p.DrainEndpoint(ep) + p.RemoveEndpointFromDrain(ep) // Completion can precede stopping metadata. + ep.Stopping = true + ep.Config = "check weight 0" + p.drainMgr.setDesired(cfg) + if err := os.WriteFile(p.cfg.PidFile, []byte("456\n"), 0600); err != nil { + t.Fatal(err) + } + for key := range r.states { + r.initialWeights[key] = 0 + } // Reload of the controller's stopping config. + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + for _, original := range p.drainMgr.weights[ep.Name] { + if original.pid != "456" { + t.Fatal("proven stopping config generation was not retained") + } + } + ep.Stopping = false + ep.Config = "check" + p.drainMgr.setDesired(cfg) + // No subsequent reload: the runtime is still zero in the same PID. + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestActiveDrainReloadDoesNotAdoptNewProcessManualZero(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + r.sessions = 1 + p.DrainEndpoint(ep) + if err := os.WriteFile(p.cfg.PidFile, []byte("456\n"), 0600); err != nil { + t.Fatal(err) + } + // The replacement has configured initial weight 1, but its current zero is + // not from any command issued by this controller in the new generation. + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + p.RemoveEndpointFromDrain(ep) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 0, "be-two/target": 0}) { + t.Fatal("adopted unowned replacement-process zeros") + } +} + +func TestActiveDrainReloadHonorsChangedDesiredCustomWeight(t *testing.T) { + p, r, ep, cfg := recoveryFixture(t) + r.sessions = 1 + p.DrainEndpoint(ep) + // A desired-config edit is a different source of authority from the + // runtime weight captured before draining. Do not undo the user's edit. + ep.Config = "check weight 37" + p.drainMgr.setDesired(cfg) + if err := os.WriteFile(p.cfg.PidFile, []byte("456\n"), 0600); err != nil { + t.Fatal(err) + } + for key, state := range r.states { + state.weight = 37 + r.states[key] = state + r.initialWeights[key] = 37 + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + p.RemoveEndpointFromDrain(ep) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 37, "be-two/target": 37}) { + t.Fatal("recovery undid the desired custom weight:", runtimeWeights(r)) + } +} + +func TestSameEndpointAndUnchangedConfigRecoverWithoutReload(t *testing.T) { + p, r, ep, cfg := recoveryFixture(t) + var previous string + reloads := 0 + p.cfg.reloadCommand = func() error { + body, err := os.ReadFile(p.cfg.Config) + if err != nil { + return err + } + if string(body) != previous { + reloads++ + previous = string(body) + } + return nil + } + if err := p.applyHaproxyConfig(cfg); err != nil { + t.Fatal(err) + } + p.DrainEndpoint(ep) + if err := p.applyHaproxyConfig(cfg); err != nil { + t.Fatal(err) + } + if reloads != 1 { + t.Fatalf("unchanged config caused %d reloads", reloads) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestDrainRecoveryWaitsForHealthAndRetriesSocketFailure(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + p.DrainEndpoint(ep) + state := r.states["be-one/target"] + state.operational = 0 + r.states[state.backend] = state + r.fail["set server be-two/target weight 23"] = 1 + if err := p.reconcileDrains(); err == nil { + t.Fatal("missing restore failure") + } + if len(p.drainMgr.weights[ep.Name]) != 2 { + t.Fatal("failure lost recovery provenance") + } + state.operational = 2 + r.states[state.backend] = state + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestDrainCommandFailureAndRejectedReplyKeepRecoveryOwnership(t *testing.T) { + for _, failure := range []string{"timeout", "rejected"} { + t.Run(failure, func(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + if failure == "timeout" { + r.fail["set server be-one/target weight 0"] = 1 + } else { + r.reject["set server be-one/target weight 0"] = "Permission denied" + } + if p.DrainEndpoint(ep) { + t.Fatal("partial drain reported success") + } + if len(p.drainMgr.weights[ep.Name]) != 2 { + t.Fatal("partial failure lost ownership") + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } + }) + } +} + +func TestDrainRecoveryRequiresVerifiedWeightBeforeForgetting(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + p.DrainEndpoint(ep) + r.ignoreWrites = true + if err := p.reconcileDrains(); err == nil { + t.Fatal("unapplied restore acknowledged as successful") + } + if len(p.drainMgr.weights[ep.Name]) != 2 { + t.Fatal("unverified restore deleted ownership") + } + r.ignoreWrites = false + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } +} + +func TestDrainRecoveryPreservesManualStateAndDesiredZero(t *testing.T) { + for _, mode := range []string{"preexisting-zero", "manual-MAINT", "manual-DRAIN", "manual-custom-weight", "desired-zero", "desired-disabled", "removed"} { + t.Run(mode, func(t *testing.T) { + p, r, ep, cfg := recoveryFixture(t) + if mode == "preexisting-zero" { + state := r.states["be-one/target"] + state.weight = 0 + r.states[state.backend] = state + } + p.DrainEndpoint(ep) + state := r.states["be-one/target"] + switch mode { + case "manual-MAINT": + state.admin = 1 + case "manual-DRAIN": + state.admin = 8 + case "manual-custom-weight": + state.weight = 31 + case "desired-zero": + ep.Config = "check weight 0" + p.drainMgr.setDesired(cfg) + case "desired-disabled": + ep.Config = "check disabled" + p.drainMgr.setDesired(cfg) + case "removed": + p.drainMgr.setDesired(&config.LoadBalancerConfig{}) + } + r.states[state.backend] = state + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if got := r.states[state.backend]; got.weight != state.weight || got.admin != state.admin { + t.Fatalf("operator state changed: before=%+v after=%+v", state, got) + } + }) + } +} + +func TestActiveDrainSurvivesReloadAndRecoversOriginalCustomWeights(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + r.sessions = 1 + p.DrainEndpoint(ep) + if err := os.WriteFile(p.cfg.PidFile, []byte("456\n"), 0600); err != nil { + t.Fatal(err) + } + for key, state := range r.states { + state.weight = 1 + r.states[key] = state + } + r.fail["set server be-two/target weight 0"] = 1 + if err := p.reconcileDrains(); err == nil { + t.Fatal("missing reapply failure") + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 0, "be-two/target": 0}) { + t.Fatal(runtimeWeights(r)) + } + p.RemoveEndpointFromDrain(ep) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runtimeWeights(r), map[string]int{"be-one/target": 7, "be-two/target": 23}) { + t.Fatal(runtimeWeights(r)) + } +} + +func TestActiveDrainDoesNotOverwriteManualRuntimeWeight(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + r.sessions = 1 + p.DrainEndpoint(ep) + state := r.states["be-one/target"] + state.weight = 31 + r.states[state.backend] = state + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states[state.backend].weight != 31 { + t.Fatal("active reconciliation overwrote operator weight") + } + p.RemoveEndpointFromDrain(ep) + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states[state.backend].weight != 31 { + t.Fatal("recovery overwrote operator weight") + } +} + +func TestDrainRecoveryRetainsOwnershipDuringTransientResolutionFailure(t *testing.T) { + p, r, ep, _ := recoveryFixture(t) + p.DrainEndpoint(ep) + state := r.states["be-one/target"] + state.admin = 0x20 + state.operational = 0 + r.states[state.backend] = state + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if _, owned := p.drainMgr.weights[ep.Name][state.backend]; !owned { + t.Fatal("transient DNS maintenance lost recovery ownership") + } + state.admin = 0 + state.operational = 2 + r.states[state.backend] = state + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states[state.backend].weight != 7 { + t.Fatal("DNS recovery retained stale zero") + } +} + +func TestRecoveryDoesNotAdoptReplacementProcessOrFailedReload(t *testing.T) { + p, r, ep, cfg := recoveryFixture(t) + p.DrainEndpoint(ep) + p.cfg.reloadCommand = func() error { return errors.New("reload failed") } + if err := p.applyHaproxyConfig(cfg); err == nil { + t.Fatal("missing reload error") + } + if r.states["be-one/target"].weight != 0 || len(p.drainMgr.weights[ep.Name]) != 2 { + t.Fatal("failed reload restored or forgot state") + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states["be-one/target"].weight != 0 || len(p.drainMgr.weights[ep.Name]) != 2 { + t.Fatal("timer used stale desired state after failed reload") + } + p.recoveryBlocked = false // Separately verify ownership after a process replacement. + if err := os.WriteFile(p.cfg.PidFile, []byte("456\n"), 0600); err != nil { + t.Fatal(err) + } + if err := p.reconcileDrains(); err != nil { + t.Fatal(err) + } + if r.states["be-one/target"].weight != 0 { + t.Fatal("adopted replacement process's unowned zero") + } +} + +func TestRuntimeCommandHasReadDeadline(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("Unix socket fixture runs in Linux validation") + } + socket := filepath.Join(t.TempDir(), "ha.sock") + listener, err := net.Listen("unix", socket) + if err != nil { + t.Fatal(err) + } + defer listener.Close() + done := make(chan struct{}) + go func() { + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + <-done + }() + defer close(done) + dm := &drainMgr{haproxySock: socket} + start := time.Now() + if _, err := dm.runCommand("show servers state"); err == nil { + t.Fatal("hung socket did not time out") + } + if elapsed := time.Since(start); elapsed < time.Second || elapsed > 4*time.Second { + t.Fatalf("unexpected deadline %s", elapsed) + } +} + +func TestMalformedRuntimeResponsesFailClosed(t *testing.T) { + for _, response := range [][]string{{"error"}, {"1", "# unknown"}, {"1", "# be_name srv_name srv_uweight srv_admin_state srv_op_state", "be target bad 0 2"}, {"1", "# be_name srv_name srv_uweight srv_admin_state srv_op_state", "be target 1 0 2 extra"}} { + dm := &drainMgr{command: func(string) ([]string, error) { return response, nil }} + if _, err := dm.readServerStates(); err == nil { + t.Fatalf("accepted malformed state %q", response) + } + } + dm := &drainMgr{command: func(string) ([]string, error) { return []string{"# pxname,svname,scur", "be,target,0,unexpected"}, nil }} + if _, err := dm.ReadStats(); err == nil { + t.Fatal("accepted malformed stats") + } +} diff --git a/provider/haproxy/haproxy.go b/provider/haproxy/haproxy.go index b87711b4..63615826 100644 --- a/provider/haproxy/haproxy.go +++ b/provider/haproxy/haproxy.go @@ -47,20 +47,23 @@ func init() { type Stat map[string]string type Provider struct { - cfg *haproxyConfig - stopCh chan struct{} - init bool - drainMgr *drainMgr + cfg *haproxyConfig + stopCh chan struct{} + init bool + runtimeMu sync.Mutex + recoveryBlocked bool + drainMgr *drainMgr } type haproxyConfig struct { - Name string - ReloadCmd string - StartCmd string - Config string - Template string - CertDir string - PidFile string + Name string + ReloadCmd string + StartCmd string + Config string + Template string + CertDir string + PidFile string + reloadCommand func() error } type drainManager interface { @@ -69,9 +72,13 @@ type drainManager interface { } type drainMgr struct { - drainList map[string]string - mu *sync.RWMutex - haproxySock string + drainList map[string]string + mu *sync.RWMutex + haproxySock string + weights map[string]map[string]drainWeight + desired map[string]desiredEndpoint + command func(string) ([]string, error) + olderProcesses func() (bool, error) } func (cfg *haproxyConfig) write(lbConfig *config.LoadBalancerConfig) (err error) { @@ -138,6 +145,11 @@ func (cfg *haproxyConfig) readPid() (string, error) { } func (lbp *Provider) applyHaproxyConfig(lbConfig *config.LoadBalancerConfig) error { + lbp.runtimeMu.Lock() + defer lbp.runtimeMu.Unlock() + // A failed desired-config update must not let the retry timer restore + // weights using the previous (possibly now disabled) desired target set. + lbp.recoveryBlocked = true // copy certificates if _, err := os.Stat(lbp.cfg.CertDir); os.IsNotExist(err) { if err = os.Mkdir(lbp.cfg.CertDir, 0755); err != nil { @@ -179,7 +191,12 @@ func (lbp *Provider) applyHaproxyConfig(lbConfig *config.LoadBalancerConfig) err return err } - return lbp.cfg.reload() + if err := lbp.cfg.reload(); err != nil { + return err + } + lbp.drainMgr.setDesired(lbConfig) + lbp.recoveryBlocked = false + return lbp.reconcileDrains() } func (lbp *Provider) ApplyConfig(lbConfig *config.LoadBalancerConfig) error { @@ -208,6 +225,8 @@ func (lbp *Provider) IsEndpointUpForDrain(ep *config.Endpoint) bool { } func (lbp *Provider) DrainEndpoint(ep *config.Endpoint) bool { + lbp.runtimeMu.Lock() + defer lbp.runtimeMu.Unlock() pid, err := lbp.cfg.readPid() if err != nil { @@ -218,6 +237,8 @@ func (lbp *Provider) DrainEndpoint(ep *config.Endpoint) bool { } func (lbp *Provider) RemoveEndpointFromDrain(ep *config.Endpoint) { + lbp.runtimeMu.Lock() + defer lbp.runtimeMu.Unlock() lbp.drainMgr.RemoveEndpointFromDrain(ep) } @@ -314,8 +335,7 @@ func (dm *drainMgr) readCurrentStats(ep *config.Endpoint) (string, []string, err if escur != "" { escurInt, err := strconv.Atoi(escur) if err != nil { - log.Errorf("Error %v while converting scur %v, defaulting to 0", err, escur) - escurInt = 0 + return "", backendNames, fmt.Errorf("invalid HAProxy current session count for endpoint %s", ep.Name) } totalScur = totalScur + escurInt } @@ -331,6 +351,9 @@ func (dm *drainMgr) readCurrentStats(ep *config.Endpoint) (string, []string, err } func (dm *drainMgr) olderPidExists() (bool, error) { + if dm.olderProcesses != nil { + return dm.olderProcesses() + } output, err := exec.Command("pidof", "haproxy").Output() if err != nil { return true, fmt.Errorf("error listing old process %v", err) @@ -348,24 +371,23 @@ func (dm *drainMgr) olderPidExists() (bool, error) { } func (dm *drainMgr) AddEndpointForDrain(ep *config.Endpoint, pid string) bool { - //check if drain is needed by looking at Stats if ep != nil && ep.Name != "" { - scur, pxnames, err := dm.readCurrentStats(ep) + dm.mu.Lock() + defer dm.mu.Unlock() + if _, exists := dm.drainList[ep.Name]; exists { + return true + } + scur, _, err := dm.readCurrentStats(ep) if err != nil { log.Errorf("AddEndpointForDrain: Error %v", err) return false } - //set weight to zero on the socket anyway to stop accepting new connections - setWeightFailed := false - for _, pxname := range pxnames { - err = dm.SetWeightForDrain(pxname) - if err != nil { - log.Errorf("AddEndpointForDrain: Error %v setting weight via socket for Endpoint backend %v", err, pxname) - setWeightFailed = true - } + states, err := dm.readServerStates() + if err == nil { + err = dm.drainEndpoint(ep.Name, pid, states) } - - if setWeightFailed { + if err != nil { + log.Errorf("AddEndpointForDrain: endpoint=%s error=%s; retaining recovery ownership", ep.Name, utils.SanitizeLogText(err.Error())) return false } @@ -385,9 +407,7 @@ func (dm *drainMgr) AddEndpointForDrain(ep *config.Endpoint, pid string) bool { pid = "" } // add to drainList - dm.mu.Lock() dm.drainList[ep.Name] = pid - dm.mu.Unlock() return true } return false @@ -434,7 +454,7 @@ func (dm *drainMgr) ReadStats() (map[string][]Stat, error) { values := strings.Split(line, ",") if len(keys) != len(values) { - log.Errorf("Invalid stat line: %s", line) + return nil, fmt.Errorf("invalid HAProxy stat column count: got %d, want %d", len(values), len(keys)) } stat := Stat{} @@ -450,19 +470,21 @@ func (dm *drainMgr) ReadStats() (map[string][]Stat, error) { } func (dm *drainMgr) SetWeightForDrain(backend string) error { - _, err := dm.runCommand("set server " + backend + " weight 0") - if err != nil { - return err - } - return nil + return dm.setWeight(backend, 0) } func (dm *drainMgr) runCommand(cmd string) ([]string, error) { - conn, err := net.Dial("unix", dm.haproxySock) + if dm.command != nil { + return dm.command(cmd) + } + conn, err := net.DialTimeout("unix", dm.haproxySock, 2*time.Second) if err != nil { return nil, err } defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + return nil, err + } _, err = conn.Write([]byte(cmd + "\n")) if err != nil { @@ -491,6 +513,9 @@ func (cfg *haproxyConfig) start() error { } func (cfg *haproxyConfig) reload() error { + if cfg.reloadCommand != nil { + return cfg.reloadCommand() + } output, err := exec.Command("sh", "-c", cfg.ReloadCmd).CombinedOutput() msg := fmt.Sprintf("%v -- %v", cfg.Name, string(output)) if string(output) != "" { @@ -513,7 +538,21 @@ func (lbp *Provider) IsHealthy() bool { func (lbp *Provider) Run(syncEndpointsQueue *utils.TaskQueue) { lbp.StartHaproxy() lbp.init = false - <-lbp.stopCh + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + select { + case <-lbp.stopCh: + return + case <-ticker.C: + lbp.runtimeMu.Lock() + err := lbp.reconcileDrains() + lbp.runtimeMu.Unlock() + if err != nil { + log.Errorf("Drain runtime reconciliation failed: %s; retrying", utils.SanitizeLogText(err.Error())) + } + } + } } func (lbp *Provider) Stop() error { diff --git a/provider/haproxy/haproxy_config_util.go b/provider/haproxy/haproxy_config_util.go index 437b2232..450eb6aa 100644 --- a/provider/haproxy/haproxy_config_util.go +++ b/provider/haproxy/haproxy_config_util.go @@ -230,9 +230,10 @@ func BuildCustomConfig(lbConfig *config.LoadBalancerConfig, customConfig string, ep.Config = fmt.Sprintf("%s cookie %s", ep.Config, ep.Name) } - //check if the endpoint is in drain list. If yes then add weight 0 to ep.Config - inDrainList, _ := drainMgr.isEndpointUpForDrain(ep) - if inDrainList || (ep.Weight != 1 && ep.Weight >= 0 && ep.Weight < 257) { + // Draining is runtime state, not desired configuration. Reconcile + // owned weights separately even when the reload script skips an + // unchanged configuration; never reload merely to undo a drain. + if ep.Weight != 1 && ep.Weight >= 0 && ep.Weight < 257 { ep.Config = fmt.Sprintf("%s weight %d", ep.Config, ep.Weight) } }