diff --git a/pkg/router/template/configmanager/haproxy/backend.go b/pkg/router/template/configmanager/haproxy/backend.go index d8b8293ee..86ee72016 100644 --- a/pkg/router/template/configmanager/haproxy/backend.go +++ b/pkg/router/template/configmanager/haproxy/backend.go @@ -303,12 +303,24 @@ func (b *Backend) FindServer(id string) (*backendServer, error) { return nil, fmt.Errorf("no server found for id: %s", id) } +// IsHealthCheckNotConfiguredError returns true if the provided error is due to an attempt to +// enable health check on a backend server whose health check was not configured. +func (b *Backend) IsHealthCheckNotConfiguredError(err error) bool { + return err != nil && strings.Contains(err.Error(), "Health check was not configured on this server") +} + +// IsServerAlreadyExistsError returns true if the provided error is due to an attempt to add a +// new backend server, but the server name was already used. +func (b *Backend) IsServerAlreadyExistsError(err error) bool { + return err != nil && strings.Contains(err.Error(), "Already exists a server ") +} + // AddServer dynamically adds a new backend server. It detects if the server already exists, and if so tries to remove it. // It returns a failure in case HAProxy refuses to dynamically add the server for any reason, or if the existing server // cannot be removed, e.g., it still have active or steady and established connection(s) to its backend server endpoint. -func (b *Backend) AddServer(cfg *templaterouter.ServiceAliasConfig, svc *templaterouter.ServiceUnit, ep templaterouter.Endpoint, weight int32, workingDir, defaultDestinationCA string) error { - if err := b.innerAddServer(cfg, svc, ep, weight, workingDir, defaultDestinationCA); err != nil { - if !strings.Contains(err.Error(), "Already exists a server ") { +func (b *Backend) AddServer(cfg *templaterouter.ServiceAliasConfig, svc *templaterouter.ServiceUnit, ep *templaterouter.Endpoint, workingDir, defaultDestinationCA string) error { + if err := b.innerAddServer(cfg, svc, ep, workingDir, defaultDestinationCA); err != nil { + if !b.IsServerAlreadyExistsError(err) { return err } // Failed due to already existing server left behind, in maintenance mode, due to in-flight connections. @@ -317,11 +329,11 @@ func (b *Backend) AddServer(cfg *templaterouter.ServiceAliasConfig, svc *templat // No way, need to fail which will ask for a fork-and-reload. This will leave the existing connections in the old process. return err } - if err := b.innerAddServer(cfg, svc, ep, weight, workingDir, defaultDestinationCA); err != nil { + if err := b.innerAddServer(cfg, svc, ep, workingDir, defaultDestinationCA); err != nil { return err } } - if err := b.innerSetServerState(ep, true, weight); err != nil { + if err := b.innerSetServerState(ep, true); err != nil { return err } @@ -331,33 +343,83 @@ func (b *Backend) AddServer(cfg *templaterouter.ServiceAliasConfig, svc *templat } // UpdateServer dynamically updates the backend server with new address and weight. -func (b *Backend) UpdateServer(ep templaterouter.Endpoint, weight int32, isPassthrough bool) error { - // missing to properly populate the current servers when created, should be done in the next phase. - // After that we can update only changed attributes. - // https://redhat.atlassian.net/browse/NE-2646 - if err := b.innerUpdateServerAddr(ep); err != nil { +func (b *Backend) UpdateServer(cfg *templaterouter.ServiceAliasConfig, svc *templaterouter.ServiceUnit, oldEP, newEP *templaterouter.Endpoint, isPassthrough bool, workingDir, defaultDestinationCA string) (added bool, err error) { + oldIsH2 := strings.TrimPrefix(oldEP.AppProtocol, "kubernetes.io/") == "h2c" + newIsH2 := strings.TrimPrefix(newEP.AppProtocol, "kubernetes.io/") == "h2c" + if oldIsH2 != newIsH2 || oldEP.VerifyHostname != newEP.VerifyHostname { + // changes require to remove+add endpoints, an error is returned in case this cannot be done, e.g., existing connections + return true, b.ReplaceServer(cfg, svc, oldEP, newEP, workingDir, defaultDestinationCA) + } + + // changes that can be applied in the running server + if oldEP.IP != newEP.IP || oldEP.Port != newEP.Port { + if err := b.innerUpdateServerAddrPort(newEP); err != nil { + return false, err + } + } + + if oldEP.Weight != newEP.Weight { + return false, b.UpdateServerWeight(oldEP, newEP, isPassthrough) + } + + return false, nil +} + +// UpdateServerWeight updates the weight of the backend server represented by the new endpoint. +// It also updates the state of the enpoint to `drain` or `ready` if the weight going to, or coming from `0`. +func (b *Backend) UpdateServerWeight(oldEP, newEP *templaterouter.Endpoint, isPassthrough bool) error { + if err := b.innerUpdateServerWeight(newEP, isPassthrough); err != nil { + return err + } + if (oldEP.Weight <= 0) != (newEP.Weight <= 0) { + if err := b.innerSetServerState(newEP, true); err != nil { + return err + } + } + return nil +} + +// ReplaceServer dynamically replaces the backend server by removing it and adding again with new configuration. +// Note that a failure adding the backend server should result in the server being missed in the configuration, +// because of that it is important to not replace servers in case of single replica, this would cause an outage +// until HAProxy is reloaded. +func (b *Backend) ReplaceServer(cfg *templaterouter.ServiceAliasConfig, svc *templaterouter.ServiceUnit, oldEP, newEP *templaterouter.Endpoint, workingDir, defaultDestinationCA string) error { + if err := b.innerSetServerState(oldEP, false); err != nil { + return err + } + if err := b.innerDeleteServer(oldEP); err != nil { + if rollbackErr := b.innerSetServerState(oldEP, true); rollbackErr != nil { + return fmt.Errorf("deleting old server: %v; restoring old server state: %v", err, rollbackErr) + } return err } - return b.innerUpdateServerWeight(ep, weight, isPassthrough) + if err := b.innerAddServer(cfg, svc, newEP, workingDir, defaultDestinationCA); err != nil { + return err + } + return b.innerSetServerState(newEP, true) } // EnableHealthCheck dynamically enables health check on a backend server that already declares the health check interval. -func (b *Backend) EnableHealthCheck(ep templaterouter.Endpoint) error { +func (b *Backend) EnableHealthCheck(ep *templaterouter.Endpoint) error { return b.innerSetHealthCheck(ep, true) } // DisableHealthCheck dynamically disables health check on a backend server. -func (b *Backend) DisableHealthCheck(ep templaterouter.Endpoint) error { - return b.innerSetHealthCheck(ep, false) +func (b *Backend) DisableHealthCheck(ep *templaterouter.Endpoint) error { + if err := b.innerSetHealthCheck(ep, false); err != nil { + return err + } + // manually set the new health state after disabling the automatic check + return b.innerSetServerHealth(ep, true) } // DeleteServer dynamically removes the backend server from the load balance. The backend server is put in maintenance mode // and returns `removed` as false in case it has active or steady and established connections, so these connections continue // to be handled and new ones are directed to other servers. An error only happens if the server cannot be put in maintenance // mode, any failure trying to remove the server is logged and just return removed as false. -func (b *Backend) DeleteServer(ep templaterouter.Endpoint) (removed bool, err error) { +func (b *Backend) DeleteServer(ep *templaterouter.Endpoint) (removed bool, err error) { // put in maintenance mode first, this is a pre-requisite to remove a backend server. - if err := b.innerSetServerState(ep, false, 0); err != nil { + if err := b.innerSetServerState(ep, false); err != nil { return false, err } if err := b.innerDeleteServer(ep); err != nil { @@ -367,7 +429,7 @@ func (b *Backend) DeleteServer(ep templaterouter.Endpoint) (removed bool, err er return true, nil } -func (b *Backend) innerAddServer(cfg *templaterouter.ServiceAliasConfig, svc *templaterouter.ServiceUnit, ep templaterouter.Endpoint, weight int32, workingDir, defaultDestinationCA string) error { +func (b *Backend) innerAddServer(cfg *templaterouter.ServiceAliasConfig, svc *templaterouter.ServiceUnit, ep *templaterouter.Endpoint, workingDir, defaultDestinationCA string) error { // This should always follow the template, changes here should be reflected there, both regular and passthrough backends // // TODO: either read this configuration from the template, or instead make the template read from here. @@ -376,7 +438,7 @@ func (b *Backend) innerAddServer(cfg *templaterouter.ServiceAliasConfig, svc *te // // https://redhat.atlassian.net/browse/NE-2646 - cmd := fmt.Sprintf("add server %s/%s %s:%s weight %d", b.name, ep.ID, ep.IP, ep.Port, weight) + cmd := fmt.Sprintf("add server %s/%s %s:%s weight %d", b.name, ep.ID, ep.IP, ep.Port, ep.Weight) switch cfg.TLSTermination { case v1.TLSTerminationReencrypt: @@ -384,7 +446,7 @@ func (b *Backend) innerAddServer(cfg *templaterouter.ServiceAliasConfig, svc *te if disableHTTP2, _ := strconv.ParseBool(os.Getenv("ROUTER_DISABLE_HTTP2")); !disableHTTP2 { cmd += " alpn h2,http/1.1" } - if cfg.VerifyServiceHostname { + if ep.VerifyHostname { cmd += " verifyhost " + svc.Hostname } if cert := cfg.Certificates[cfg.Host+"_pod"]; len(cert.Contents) > 0 { @@ -396,7 +458,7 @@ func (b *Backend) innerAddServer(cfg *templaterouter.ServiceAliasConfig, svc *te } cmd += " check-ssl" case "", v1.TLSTerminationEdge: - if ep.AppProtocol == "h2c" || ep.AppProtocol == "kubernetes.io/h2c" { + if strings.TrimPrefix(ep.AppProtocol, "kubernetes.io/") == "h2c" { cmd += " proto h2" } case v1.TLSTerminationPassthrough: @@ -419,23 +481,22 @@ func (b *Backend) innerAddServer(cfg *templaterouter.ServiceAliasConfig, svc *te return execCommand(b.client, apiAddServer, cmd) } -func (b *Backend) innerUpdateServerAddr(ep templaterouter.Endpoint) error { +func (b *Backend) innerUpdateServerAddrPort(ep *templaterouter.Endpoint) error { cmd := fmt.Sprintf("set server %s/%s addr %s port %s", b.name, ep.ID, ep.IP, ep.Port) return execCommand(b.client, apiSetServerAddr, cmd) } -func (b *Backend) innerUpdateServerWeight(ep templaterouter.Endpoint, weight int32, isPassthrough bool) error { - cmd := fmt.Sprintf("set server %s/%s", b.name, ep.ID) - if isPassthrough { - // https://github.com/openshift/router/blob/896390778ebe15f57f87e6ca78f11c96e64c2652/pkg/router/template/configmanager/haproxy/manager.go#L446-L454 - cmd += " weight 100%" - } else { - cmd = fmt.Sprintf("%s weight %d", cmd, weight) +func (b *Backend) innerUpdateServerWeight(ep *templaterouter.Endpoint, isPassthrough bool) error { + // https://github.com/openshift/router/blob/896390778ebe15f57f87e6ca78f11c96e64c2652/pkg/router/template/configmanager/haproxy/manager.go#L446-L454 + weight := "100%" // hardcoded for passthrough + if !isPassthrough { + weight = strconv.Itoa(int(ep.Weight)) } + cmd := fmt.Sprintf("set server %s/%s weight %s", b.name, ep.ID, weight) return execCommand(b.client, apiSetServerWeight, cmd) } -func (b *Backend) innerSetHealthCheck(ep templaterouter.Endpoint, enable bool) error { +func (b *Backend) innerSetHealthCheck(ep *templaterouter.Endpoint, enable bool) error { enableStr := "enable" if !enable { enableStr = "disable" @@ -444,18 +505,27 @@ func (b *Backend) innerSetHealthCheck(ep templaterouter.Endpoint, enable bool) e return execCommand(b.client, apiSetHealth, cmd) } -func (b *Backend) innerSetServerState(ep templaterouter.Endpoint, ready bool, weight int32) error { +func (b *Backend) innerSetServerHealth(ep *templaterouter.Endpoint, up bool) error { + upStr := "up" + if !up { + upStr = "down" + } + cmd := fmt.Sprintf("set server %s/%s health %s", b.name, ep.ID, upStr) + return execCommand(b.client, apiSetServerHealth, cmd) +} + +func (b *Backend) innerSetServerState(ep *templaterouter.Endpoint, ready bool) error { state := "ready" if !ready { state = "maint" - } else if weight <= 0 { + } else if ep.Weight <= 0 { state = "drain" } cmd := fmt.Sprintf("set server %s/%s state %s", b.name, ep.ID, state) return execCommand(b.client, apiSetServerState, cmd) } -func (b *Backend) innerDeleteServer(ep templaterouter.Endpoint) error { +func (b *Backend) innerDeleteServer(ep *templaterouter.Endpoint) error { cmd := fmt.Sprintf("del server %s/%s", b.name, ep.ID) return execCommand(b.client, apiDelServer, cmd) } @@ -560,6 +630,7 @@ const ( apiDelServer apiSetHealth apiSetServerAddr + apiSetServerHealth apiSetServerWeight apiSetServerState ) @@ -582,7 +653,7 @@ func execCommand(client HAProxyClient, api apiType, cmd string) error { valid = response == "Server deleted." case apiSetServerAddr: valid = response == "nothing changed" || strings.HasPrefix(response, "IP changed from ") || strings.HasPrefix(response, "port changed from ") || strings.HasPrefix(response, "no need to change ") - case apiSetHealth, apiSetServerWeight, apiSetServerState: + case apiSetHealth, apiSetServerHealth, apiSetServerWeight, apiSetServerState: valid = false // any response from these api calls mean there is a failure default: // fail fast in case of a dev error diff --git a/pkg/router/template/configmanager/haproxy/backend_test.go b/pkg/router/template/configmanager/haproxy/backend_test.go index 03a48a80c..cd3374ce4 100644 --- a/pkg/router/template/configmanager/haproxy/backend_test.go +++ b/pkg/router/template/configmanager/haproxy/backend_test.go @@ -16,6 +16,7 @@ func TestBackendDynamicUpdate(t *testing.T) { cmdAdd cmd = "add" cmdDel cmd = "del" cmdUpdate cmd = "update" + cmdReplace cmd = "replace" cmdEnableHealth cmd = "enable-health" cmdDisableHealth cmd = "disable-health" ) @@ -24,20 +25,26 @@ func TestBackendDynamicUpdate(t *testing.T) { cmd cmd backendName *templaterouter.ServiceAliasConfigKey // default: "route1" endpointID *string // default: "server1" + oldIP *string // default: "10.0.1.11" ip *string // default: "10.0.1.11" + oldPort *string // default: "9000" port *string // default: "9000" + oldWeight *int32 // default: 1 weight *int32 // default: 1 - workingDir *string // default: "tmp" + workingDir *string // default: "/tmp" publicHostname string serviceHostname string tlsTermination routev1.TLSTerminationType - verifyHostname bool - appProtocol string + oldVerifyHost bool + verifyHost bool + oldAppProto string + appProto string certificates map[string]templaterouter.Certificate annotations map[string]string envvars []string defaultCA string cmdCustomResp []string // 1:1 to `cmdExpected`, trailing empty items can be omited. + addedExpected bool errExpected string removedExpected bool cmdExpected []string @@ -105,10 +112,19 @@ func TestBackendDynamicUpdate(t *testing.T) { "set server route1/server1 state ready", }, }, - "should add edge termination h2 server": { + "should add edge termination h2 server with h2c proto": { cmd: cmdAdd, tlsTermination: routev1.TLSTerminationEdge, - appProtocol: "h2c", + appProto: "h2c", + cmdExpected: []string{ + "add server route1/server1 10.0.1.11:9000 weight 1 proto h2 check inter 5000ms", + "set server route1/server1 state ready", + }, + }, + "should add edge termination h2 server with kubernetes.io/h2c proto": { + cmd: cmdAdd, + tlsTermination: routev1.TLSTerminationEdge, + appProto: "kubernetes.io/h2c", cmdExpected: []string{ "add server route1/server1 10.0.1.11:9000 weight 1 proto h2 check inter 5000ms", "set server route1/server1 state ready", @@ -125,7 +141,7 @@ func TestBackendDynamicUpdate(t *testing.T) { "should add reencrypt termination server with verify host": { cmd: cmdAdd, tlsTermination: routev1.TLSTerminationReencrypt, - verifyHostname: true, + verifyHost: true, serviceHostname: "route1.default.svc", cmdExpected: []string{ "add server route1/server1 10.0.1.11:9000 weight 1 ssl alpn h2,http/1.1 verifyhost route1.default.svc verify none check-ssl check inter 5000ms", @@ -182,7 +198,6 @@ func TestBackendDynamicUpdate(t *testing.T) { cmd: cmdUpdate, weight: ptr.To[int32](10), cmdExpected: []string{ - "set server route1/server1 addr 10.0.1.11 port 9000", "set server route1/server1 weight 10", }, }, @@ -191,16 +206,126 @@ func TestBackendDynamicUpdate(t *testing.T) { tlsTermination: routev1.TLSTerminationPassthrough, weight: ptr.To[int32](10), cmdExpected: []string{ - "set server route1/server1 addr 10.0.1.11 port 9000", "set server route1/server1 weight 100%", }, }, + "should update server changing port": { + cmd: cmdUpdate, + oldPort: ptr.To("8000"), + port: ptr.To("9000"), + cmdExpected: []string{ + "set server route1/server1 addr 10.0.1.11 port 9000", + }, + }, + "should update server changing weight": { + cmd: cmdUpdate, + oldWeight: ptr.To[int32](10), + weight: ptr.To[int32](20), + cmdExpected: []string{ + "set server route1/server1 weight 20", + }, + }, + "should update server changing backend proto": { + cmd: cmdUpdate, + oldAppProto: "", + appProto: "h2c", + addedExpected: true, + cmdExpected: []string{ + "set server route1/server1 state maint", + "del server route1/server1", + "add server route1/server1 10.0.1.11:9000 weight 1 proto h2 check inter 5000ms", + "set server route1/server1 state ready", + }, + }, "should fail if failing to update server": { cmd: cmdUpdate, + oldWeight: ptr.To[int32](2), + weight: ptr.To[int32](1), cmdCustomResp: []string{"Some unknown updating error."}, errExpected: "unexpected response from haproxy: Some unknown updating error.", cmdExpected: []string{ - "set server route1/server1 addr 10.0.1.11 port 9000", + "set server route1/server1 weight 1", + }, + }, + "should fail if failing to update server changing proto": { + cmd: cmdUpdate, + oldAppProto: "", + appProto: "h2c", + addedExpected: true, + cmdCustomResp: []string{ + "", // first cmd + "Some unknown deleting error.", // second cmd + }, + errExpected: "unexpected response from haproxy: Some unknown deleting error.", + cmdExpected: []string{ + "set server route1/server1 state maint", + "del server route1/server1", + "set server route1/server1 state ready", + }, + }, + + // + // replacing + "should replace an existing server": { + cmd: cmdReplace, + cmdExpected: []string{ + "set server route1/server1 state maint", + "del server route1/server1", + "add server route1/server1 10.0.1.11:9000 weight 1 check inter 5000ms", + "set server route1/server1 state ready", + }, + }, + "should fail if failing to disable on replace server": { + cmd: cmdReplace, + cmdCustomResp: []string{ + "Some unknown set server error.", + }, + errExpected: "unexpected response from haproxy: Some unknown set server error.", + cmdExpected: []string{ + "set server route1/server1 state maint", + }, + }, + "should fail if failing to delete on replace server": { + cmd: cmdReplace, + cmdCustomResp: []string{ + "", // first cmd + "Some unknown deleting error.", // second cmd + }, + errExpected: "unexpected response from haproxy: Some unknown deleting error.", + cmdExpected: []string{ + "set server route1/server1 state maint", + "del server route1/server1", + "set server route1/server1 state ready", + }, + }, + "should fail if failing to add on replace server": { + cmd: cmdReplace, + cmdCustomResp: []string{ + "", // first cmd + "", // second cmd + "Some unknown add server error.", // third cmd + }, + errExpected: "unexpected response from haproxy: Some unknown add server error.", + cmdExpected: []string{ + "set server route1/server1 state maint", + "del server route1/server1", + "add server route1/server1 10.0.1.11:9000 weight 1 check inter 5000ms", + }, + }, + "should fail if failing to set server state on replace server": { + cmd: cmdReplace, + cmdCustomResp: []string{ + "", // first cmd + "", // second cmd + "", // third cmd + "Some unknown set server state error.", // forth cmd + }, + errExpected: "unexpected response from haproxy: Some unknown set server state error.", + cmdExpected: []string{ + "set server route1/server1 state maint", + "del server route1/server1", + "add server route1/server1 10.0.1.11:9000 weight 1 check inter 5000ms", + "set server route1/server1 state ready", }, }, @@ -224,6 +349,7 @@ func TestBackendDynamicUpdate(t *testing.T) { cmd: cmdDisableHealth, cmdExpected: []string{ "disable health route1/server1", + "set server route1/server1 health up", }, }, "should fail if failing to disable health check": { @@ -272,14 +398,17 @@ func TestBackendDynamicUpdate(t *testing.T) { t.Run(name, func(t *testing.T) { backendName := ptr.Deref(test.backendName, "route1") endpointID := ptr.Deref(test.endpointID, "server1") + oldIP := ptr.Deref(test.oldIP, "10.0.1.11") ip := ptr.Deref(test.ip, "10.0.1.11") + oldPort := ptr.Deref(test.oldPort, "9000") port := ptr.Deref(test.port, "9000") - weight := ptr.Deref(test.weight, 1) + oldWeight := ptr.Deref(test.oldWeight, 1) + newWeight := ptr.Deref(test.weight, 1) workingDir := ptr.Deref(test.workingDir, "/tmp") cfg := &templaterouter.ServiceAliasConfig{ TLSTermination: test.tlsTermination, - VerifyServiceHostname: test.verifyHostname, + VerifyServiceHostname: test.verifyHost, Host: test.publicHostname, Certificates: test.certificates, Annotations: test.annotations, @@ -287,11 +416,21 @@ func TestBackendDynamicUpdate(t *testing.T) { svc := &templaterouter.ServiceUnit{ Hostname: test.serviceHostname, } - ep := templaterouter.Endpoint{ - ID: endpointID, - IP: ip, - Port: port, - AppProtocol: test.appProtocol, + oldEP := &templaterouter.Endpoint{ + ID: endpointID, + IP: oldIP, + Port: oldPort, + AppProtocol: test.oldAppProto, + Weight: oldWeight, + VerifyHostname: test.oldVerifyHost, + } + newEP := &templaterouter.Endpoint{ + ID: endpointID, + IP: ip, + Port: port, + AppProtocol: test.appProto, + Weight: newWeight, + VerifyHostname: test.verifyHost, } isPassthrough := test.tlsTermination == routev1.TLSTerminationPassthrough client := &fakeClient{cmdCustomResp: test.cmdCustomResp} @@ -299,18 +438,21 @@ func TestBackendDynamicUpdate(t *testing.T) { b := newBackend(backendName, client) var removed bool + var addedFromUpdate bool var err error switch test.cmd { case cmdAdd: - err = b.AddServer(cfg, svc, ep, weight, workingDir, test.defaultCA) + err = b.AddServer(cfg, svc, newEP, workingDir, test.defaultCA) case cmdDel: - removed, err = b.DeleteServer(ep) + removed, err = b.DeleteServer(newEP) case cmdUpdate: - err = b.UpdateServer(ep, weight, isPassthrough) + addedFromUpdate, err = b.UpdateServer(cfg, svc, oldEP, newEP, isPassthrough, workingDir, test.defaultCA) + case cmdReplace: + err = b.ReplaceServer(cfg, svc, oldEP, newEP, workingDir, test.defaultCA) case cmdEnableHealth: - err = b.EnableHealthCheck(ep) + err = b.EnableHealthCheck(newEP) case cmdDisableHealth: - err = b.DisableHealthCheck(ep) + err = b.DisableHealthCheck(newEP) default: t.Errorf("invalid cmd: %s", test.cmd) } @@ -320,11 +462,11 @@ func TestBackendDynamicUpdate(t *testing.T) { } else { require.NoError(t, err) } + assert.Equal(t, test.addedExpected, addedFromUpdate) assert.Equal(t, test.removedExpected, removed) assert.Equal(t, test.cmdExpected, client.executedCmds) }) } - } type fakeClient struct { diff --git a/pkg/router/template/configmanager/haproxy/blueprint_plugin_test.go b/pkg/router/template/configmanager/haproxy/blueprint_plugin_test.go index 537819012..5cca18270 100644 --- a/pkg/router/template/configmanager/haproxy/blueprint_plugin_test.go +++ b/pkg/router/template/configmanager/haproxy/blueprint_plugin_test.go @@ -52,7 +52,7 @@ func (cm *fakeConfigManager) RemoveRoute(id templaterouter.ServiceAliasConfigKey return nil } -func (cm *fakeConfigManager) ReplaceRouteEndpoints(id templaterouter.ServiceAliasConfigKey, svc *templaterouter.ServiceUnit, oldEndpoints, newEndpoints []templaterouter.Endpoint, weight int32) error { +func (cm *fakeConfigManager) ReplaceRouteEndpoints(id templaterouter.ServiceAliasConfigKey, svc *templaterouter.ServiceUnit, oldEndpoints, newEndpoints []templaterouter.Endpoint, activeEndpoints int) error { return nil } diff --git a/pkg/router/template/configmanager/haproxy/manager.go b/pkg/router/template/configmanager/haproxy/manager.go index d812c5bba..1151c7095 100644 --- a/pkg/router/template/configmanager/haproxy/manager.go +++ b/pkg/router/template/configmanager/haproxy/manager.go @@ -408,7 +408,7 @@ func (cm *haproxyConfigManager) RemoveRoute(id templaterouter.ServiceAliasConfig return err } for _, server := range servers { - if _, err := backend.DeleteServer(templaterouter.Endpoint{ID: server.Name}); err != nil { + if _, err := backend.DeleteServer(&templaterouter.Endpoint{ID: server.Name}); err != nil { return err } } @@ -419,8 +419,8 @@ func (cm *haproxyConfigManager) RemoveRoute(id templaterouter.ServiceAliasConfig // ReplaceRouteEndpoints dynamically replaces a subset of the endpoints for // a route - modifies a subset of the servers on an haproxy backend. -func (cm *haproxyConfigManager) ReplaceRouteEndpoints(id templaterouter.ServiceAliasConfigKey, svc *templaterouter.ServiceUnit, oldEndpoints, newEndpoints []templaterouter.Endpoint, weight int32) error { - log.V(4).Info("replacing route endpoints", "id", id, "weight", weight) +func (cm *haproxyConfigManager) ReplaceRouteEndpoints(id templaterouter.ServiceAliasConfigKey, svc *templaterouter.ServiceUnit, oldEndpoints, newEndpoints []templaterouter.Endpoint, activeEndpoints int) error { + log.V(4).Info("replacing route endpoints", "id", id) if cm.isReloading() { return fmt.Errorf("Router reload in progress, cannot dynamically add endpoints for %s", id) } @@ -447,75 +447,108 @@ func (cm *haproxyConfigManager) ReplaceRouteEndpoints(id templaterouter.ServiceA return err } - addedEndpoints := make(map[string]templaterouter.Endpoint) - modifiedEndpoints := make(map[string]templaterouter.Endpoint) - for _, ep := range newEndpoints { - existing := slices.ContainsFunc(oldEndpoints, func(v2ep templaterouter.Endpoint) bool { - return v2ep.ID == ep.ID + type epPair struct{ oldEP, newEP *templaterouter.Endpoint } + addedEndpoints := make(map[string]*templaterouter.Endpoint) + modifiedEndpoints := make(map[string]epPair) + for i := range newEndpoints { + newEP := newEndpoints[i] + j := slices.IndexFunc(oldEndpoints, func(oldEP templaterouter.Endpoint) bool { + return oldEP.ID == newEP.ID }) - if existing { - modifiedEndpoints[ep.ID] = ep + if j >= 0 { + oldEP := oldEndpoints[j] + if !reflect.DeepEqual(oldEP, newEP) { + if oldEP.NoHealthCheck != newEP.NoHealthCheck { + // This is not a frequent update and it is currently challenging to implement. + // Taking the simple route for now, stopping here before any dynamic update and ask for a reload. + return fmt.Errorf("detected change in idled configuration in service %q, need to reload", svc.Name) + } + modifiedEndpoints[newEP.ID] = epPair{oldEP: &oldEP, newEP: &newEP} + } } else { - addedEndpoints[ep.ID] = ep + addedEndpoints[newEP.ID] = &newEP } } - deletedEndpoints := make(map[string]templaterouter.Endpoint) - for _, ep := range oldEndpoints { - if v2ep, ok := modifiedEndpoints[ep.ID]; ok { - if reflect.DeepEqual(ep, v2ep) { - // endpoint was unchanged. - delete(modifiedEndpoints, v2ep.ID) - continue - } - epUsesH2C := ep.AppProtocol == "h2c" || ep.AppProtocol == "kubernetes.io/h2c" - v2epUsesH2c := v2ep.AppProtocol == "h2c" || v2ep.AppProtocol == "kubernetes.io/h2c" - if (epUsesH2C || v2epUsesH2c) && epUsesH2C != v2epUsesH2c { - return fmt.Errorf("endpoint %s changed appProtocol from %q to %q, dynamically updating proto is unsupported - route will be updated on next reload", ep.ID, ep.AppProtocol, v2ep.AppProtocol) - } - } else { - deletedEndpoints[ep.ID] = ep + deletedEndpoints := make(map[string]*templaterouter.Endpoint) + for i := range oldEndpoints { + oldEP := oldEndpoints[i] + found := slices.ContainsFunc(newEndpoints, func(newEP templaterouter.Endpoint) bool { + return oldEP.ID == newEP.ID + }) + if !found { + deletedEndpoints[oldEP.ID] = &oldEP } } // there is a configuration change if any of the tracking maps have endpoint(s) - configChanged = len(deletedEndpoints)+len(modifiedEndpoints)+len(addedEndpoints) > 0 + configChanged = len(deletedEndpoints) > 0 || len(modifiedEndpoints) > 0 || len(addedEndpoints) > 0 log.V(4).Info("processing endpoint changes", "added", addedEndpoints, "deleted", deletedEndpoints, "modified", modifiedEndpoints) // Aggregating errors instead of failing fast in the first API error. This ensures that the old // process has a more accurate configuration in case it lives longer due to persistent connections. var errs []error - - for name, ep := range deletedEndpoints { - if _, err := backend.DeleteServer(ep); err != nil { - errs = append(errs, fmt.Errorf("error deleting backend server %s: %w", name, err)) + for name, ep := range addedEndpoints { + if err := backend.AddServer(entry.backend, svc, ep, cm.workingDir, cm.defaultDestinationCA); err != nil { + errs = append(errs, fmt.Errorf("error adding backend server %s: %w", name, err)) } } - for name, ep := range modifiedEndpoints { - if err := backend.UpdateServer(ep, weight, entry.termination == routev1.TLSTerminationPassthrough); err != nil { + var addedFromUpdate []*templaterouter.Endpoint + for name, epPair := range modifiedEndpoints { + oldEP := epPair.oldEP + newEP := epPair.newEP + if added, err := backend.UpdateServer(entry.backend, svc, oldEP, newEP, entry.termination == routev1.TLSTerminationPassthrough, cm.workingDir, cm.defaultDestinationCA); err != nil { errs = append(errs, fmt.Errorf("error updating backend server %s: %w", name, err)) + } else if added { + addedFromUpdate = append(addedFromUpdate, newEP) } } - for name, ep := range addedEndpoints { - if err := backend.AddServer(entry.backend, svc, ep, weight, cm.workingDir, cm.defaultDestinationCA); err != nil { - errs = append(errs, fmt.Errorf("error adding backend server %s: %w", name, err)) + for name, ep := range deletedEndpoints { + if _, err := backend.DeleteServer(ep); err != nil { + errs = append(errs, fmt.Errorf("error deleting backend server %s: %w", name, err)) } } + if len(errs) > 0 { + return errors.Join(errs...) + } // Checking health check. We need to: // * enable new endpoints if `cfg.ActiveEndpoints > 1` - // * enable also the only former endpoint if scaling from 1 to 2 or more - // * disable the only current endpoint if scaling to 1 - if len(newEndpoints) > 1 { - var newEPs []templaterouter.Endpoint + // * enable also the only former endpoint if scaling out from 1 to 2 or more + // * disable the only current endpoint if scaling in to 1 + if activeEndpoints > 1 { + var newEPs []*templaterouter.Endpoint for _, ep := range addedEndpoints { // enabling for all the new added endpoints newEPs = append(newEPs, ep) } + for _, ep := range addedFromUpdate { + newEPs = append(newEPs, ep) + } if len(oldEndpoints) == 1 { - // enabling also for the former single endpoint as well - newEPs = append(newEPs, oldEndpoints[0]) + ep := &oldEndpoints[0] + _, deleted := deletedEndpoints[ep.ID] + if !ep.NoHealthCheck && !deleted { + // The backend was previously in the single server scenario, so health check should be enabled. + // Dynamically enabling health check only works if health check is configured, and we only + // configure health check upfront in dynamically added servers. + // So, we are trying to enable health check first, and if HAProxy responds that it is not + // configured, we'll need to remove and add it again. + err := backend.EnableHealthCheck(ep) + if backend.IsHealthCheckNotConfiguredError(err) { + // Health check not configured on this server. Replace it dynamically to reconfigure with health check. + err = backend.ReplaceServer(entry.backend, svc, ep, ep, cm.workingDir, cm.defaultDestinationCA) + if err == nil { + // Server replaced successfully, mark to enable health check later. + newEPs = append(newEPs, ep) + } + } + if err != nil { + // Failed either enabling health check or replacing backend server. + errs = append(errs, err) + } + } } for _, ep := range newEPs { if !ep.NoHealthCheck { @@ -524,12 +557,10 @@ func (cm *haproxyConfigManager) ReplaceRouteEndpoints(id templaterouter.ServiceA } } } - } else if len(newEndpoints) == 1 && len(oldEndpoints) != 1 { + } else if len(newEndpoints) == 1 { // the single backend server scenario, health check should be disabled - if ep := newEndpoints[0]; !ep.NoHealthCheck { - if err := backend.DisableHealthCheck(ep); err != nil { - errs = append(errs, err) - } + if err := backend.DisableHealthCheck(&newEndpoints[0]); err != nil { + errs = append(errs, err) } } @@ -565,7 +596,7 @@ func (cm *haproxyConfigManager) RemoveRouteEndpoints(id templaterouter.ServiceAl var errs []error for _, ep := range endpoints { log.V(4).Info("deleting server for endpoint", "endpoint", ep.ID) - if _, err := backend.DeleteServer(ep); err != nil { + if _, err := backend.DeleteServer(&ep); err != nil { errs = append(errs, fmt.Errorf("error deleting server %s: %w", ep.ID, err)) } } @@ -737,6 +768,12 @@ func (cm *haproxyConfigManager) reset() { // findMatchingBlueprint finds a matching blueprint route that can be used // as a "surrogate" for the route. func (cm *haproxyConfigManager) findMatchingBlueprint(route *routev1.Route) *routev1.Route { + + // HAProxy 2.8 is not working well adding backend servers on an empty backend, like blueprint servers. + // Blueprint servers are being removed via https://redhat.atlassian.net/browse/NE-2663, so we're + // just anticipating its deprecation by not using it in case it is being configured. + return nil + termination := routeTerminationType(route) routeModifiers := backendModAnnotations(route) for _, candidate := range cm.blueprintRoutes { diff --git a/pkg/router/template/router.go b/pkg/router/template/router.go index 79f46983b..0368bdbc6 100644 --- a/pkg/router/template/router.go +++ b/pkg/router/template/router.go @@ -9,6 +9,7 @@ import ( "os/exec" "path/filepath" "reflect" + "slices" "strconv" "strings" "sync" @@ -794,6 +795,23 @@ func (r *templateRouter) removeServiceAliasAssociation(id ServiceUnitKey, alias } } +func (r *templateRouter) updateEndpointTable(cfg *ServiceAliasConfig) { + weights := r.calculateServiceWeights(cfg.ServiceUnits, cfg.PreferPort) + endpointTable := make(map[ServiceUnitKey][]Endpoint) + for key := range cfg.ServiceUnits { + if service, found := r.findMatchingServiceUnit(key); found { + newEndpoints := slices.Clone(endpointsForAlias(*cfg, service)) + for i := range newEndpoints { + ep := &newEndpoints[i] + ep.Weight = weights[key] + ep.VerifyHostname = cfg.VerifyServiceHostname + } + endpointTable[key] = newEndpoints + } + } + cfg.EndpointTable = endpointTable +} + // dynamicallyAddRoute attempts to dynamically add a route. // Note: The config should have been synced at least once initially and // the caller needs to acquire a lock [and release it]. @@ -805,18 +823,6 @@ func (r *templateRouter) dynamicallyAddRoute(backendKey ServiceAliasConfigKey, r log.V(4).Info("dynamically adding route backend", "backendKey", backendKey) r.dynamicConfigManager.Register(backendKey, backend, route) - // Saving the initial state of the endpoints from all the service units of the backend. - // This state is used later to calculate added and removed endpoints without the need to read proxy api. - for key := range backend.ServiceUnits { - if service, found := r.findMatchingServiceUnit(key); found { - backend.EndpointTable[key] = endpointsForAlias(*backend, service) - } - } - - // Fully skipping DCM for now when adding or changing routes, - // should be reincluded along with the fix for https://issues.redhat.com/browse/OCPBUGS-77344 - return false - // If no initial sync was done, don't try to dynamically add the // route as we will need a reload anyway. if !r.synced { @@ -829,21 +835,12 @@ func (r *templateRouter) dynamicallyAddRoute(backendKey ServiceAliasConfigKey, r return false } - // For each referenced service unit replace the route endpoints. - oldEndpoints := []Endpoint{} - - // As the endpoints have changed, recalculate the weights. - newWeights := r.calculateServiceWeights(backend.ServiceUnits, backend.PreferPort) + activeEndpoints := r.getActiveEndpoints(backend.ServiceUnits, backend.PreferPort) for key := range backend.ServiceUnits { if service, ok := r.findMatchingServiceUnit(key); ok { - newEndpoints := endpointsForAlias(*backend, service) + newEndpoints := backend.EndpointTable[key] log.V(4).Info("for new route backend, replacing endpoints for service", "backendKey", backendKey, "serviceKey", key, "newEndpoints", newEndpoints) - - weight, ok := newWeights[key] - if !ok { - weight = 0 - } - if err := r.dynamicConfigManager.ReplaceRouteEndpoints(backendKey, &service, oldEndpoints, newEndpoints, weight); err != nil { + if err := r.dynamicConfigManager.ReplaceRouteEndpoints(backendKey, &service, nil, newEndpoints, activeEndpoints); err != nil { log.Info("router will reload as the ConfigManager could not dynamically replace endpoints for route backend", "backendKey", backendKey, "serviceKey", key, "error", err) return false @@ -855,14 +852,60 @@ func (r *templateRouter) dynamicallyAddRoute(backendKey ServiceAliasConfigKey, r return true } +func (r *templateRouter) dynamicallyUpdateRoute(backendKey ServiceAliasConfigKey, oldConfig, newConfig *ServiceAliasConfig) bool { + if r.dynamicConfigManager == nil || !r.synced { + return false + } + + // We currently support changes on a few fields, checking if there are changes elsewhere. + newConfigCopy := *newConfig + + // fields that can be dynamically updated + newConfigCopy.PreferPort = oldConfig.PreferPort + newConfigCopy.VerifyServiceHostname = oldConfig.VerifyServiceHostname + newConfigCopy.ServiceUnits = oldConfig.ServiceUnits + + if !configsAreEqual(oldConfig, &newConfigCopy) { + // TODO need to take hostname+path maps and certificate changes into account when adding dynamic update from them + log.Info("router will reload due to changes outside target services", "backendKey", backendKey) + return false + } + + failed := false + activeEndpoints := r.getActiveEndpoints(newConfig.ServiceUnits, newConfig.PreferPort) + for serviceKey := range newConfig.ServiceUnits { + if service, ok := r.findMatchingServiceUnit(serviceKey); ok { + var oldEndpoints []Endpoint + if _, oldFound := oldConfig.ServiceUnits[serviceKey]; oldFound { + // Service didn't change, update the old endpoints to new values. + oldEndpoints = oldConfig.EndpointTable[serviceKey] + } + if err := r.dynamicConfigManager.ReplaceRouteEndpoints(backendKey, &service, oldEndpoints, newConfig.EndpointTable[serviceKey], activeEndpoints); err != nil { + log.Info("router will reload due to error changing endpoints", "backendKey", backendKey, "serviceKey", serviceKey, "error", err) + failed = true + } + } + } + + for serviceKey := range oldConfig.ServiceUnits { + if _, newFound := newConfig.ServiceUnits[serviceKey]; !newFound { + if _, ok := r.findMatchingServiceUnit(serviceKey); ok { + // Service was removed from route, remove the endpoints that were in the old config. + if err := r.dynamicConfigManager.RemoveRouteEndpoints(backendKey, oldConfig.EndpointTable[serviceKey]); err != nil { + log.Info("router will reload due to error removing endpoints", "backendKey", backendKey, "serviceKey", serviceKey, "error", err) + failed = true + } + } + } + } + + return !failed +} + // dynamicallyRemoveRoute attempts to dynamically remove a route. // Note: The config should have been synced at least once initially and // the caller needs to acquire a lock [and release it]. func (r *templateRouter) dynamicallyRemoveRoute(backendKey ServiceAliasConfigKey, route *routev1.Route) bool { - // Fully skipping DCM for now when adding or changing routes, - // should be reincluded along with the fix for https://issues.redhat.com/browse/OCPBUGS-77344 - return false - if r.dynamicConfigManager == nil || !r.synced { return false } @@ -896,66 +939,41 @@ func (r *templateRouter) dynamicallyReplaceEndpoints(id ServiceUnitKey, service continue } - if id != cfg.PrimaryServiceUnitKey && cfg.VerifyServiceHostname /*VerifyServiceHostname is true only if route type is reencrypt*/ { - // Reload to avoid enabing an endpoint with different FQDN in "verifyhost" setting. - // "verifyhost" is set to the primary service FQDN on dynamic servers. - log.V(4).Info("router will reload as the ConfigManager could not dynamically replace endpoints for alternate backend of reencrypt route", "service", id, "backendKey", backendKey, "primaryService", cfg.PrimaryServiceUnitKey) - return false - } - - // Synchronizing EndpointTable only after backing up the former state, which is the current state in the proxy. - // These old (current haproxy state) and new (current cluster state) are used to calculate added and removed endpoints. - oldEndpoints := cfg.EndpointTable[id] - newEndpoints := endpointsForAlias(cfg, service) - cfg.EndpointTable[id] = newEndpoints - - // If a service is idled, createRouterEndpoints returns a slice - // containing 1 endpoint (namely the idled service's ClusterIP - // address), which has NoHealthCheck set. It is crucial that - // health checks not be enabled for an idled service lest the - // health check itself unidle the service. - // - // A route can have multiple associated services, any one of - // which could be idled, so we need to check all of the service - // units. - // - // Even if there is only a single service-unit associated with - // the route, we cannot determine here whether a service was - // scaled from 1 active endpoint (which *would not* have had - // health checks enabled) to 1 idled-service endpoint, or - // whether the service was scaled from multiple active endpoints - // (which *would* have had health checks enabled) to 1 active - // endpoint and then updated to 1 idled-service endpoint, so the - // only safe thing to do is to force a reload if the route has - // *any* endpoint with NoHealthCheck set. - // - // TODO: Extend the dynamic configuration manager to use the - // "disable health" and "enable health" commands, and use those - // commands instead of forcing a reload. - // https://docs.haproxy.org/2.8/management.html#9.3-disable%20health - // https://docs.haproxy.org/2.8/management.html#9.3-enable%20health - for _, ep := range newEndpoints { - if ep.NoHealthCheck { - log.V(4).Info("router will reload to disable health check for idled service", "service", id, "backendKey", backendKey) - return false - } - } + // Backup old state + oldEndpointTable := cfg.EndpointTable + oldEndpoints := oldEndpointTable[id] + oldWeights := cfg.ServiceUnitNames - // As the endpoints have changed, recalculate the weights. - newWeights := r.calculateServiceWeights(cfg.ServiceUnits, cfg.PreferPort) - - // Get the weight for this service unit. - weight, ok := newWeights[id] - if !ok { - weight = 0 - } + // Update endpoints for the changed service + r.updateEndpointTable(&cfg) + newEndpoints := cfg.EndpointTable[id] + r.state[backendKey] = cfg // Persist updated state + r.stateChanged = true // Mark state as changed + activeEndpoints := r.getActiveEndpoints(cfg.ServiceUnits, cfg.PreferPort) + // Update the changed service's endpoints log.V(4).Info("dynamically replacing endpoints for associated backend", "backendKey", backendKey, "newEndpoints", newEndpoints) - if err := r.dynamicConfigManager.ReplaceRouteEndpoints(backendKey, &service, oldEndpoints, newEndpoints, weight); err != nil { - // Error dynamically modifying the config, so return false to cause a reload to happen. - log.Info("router will reload as the ConfigManager could not dynamically replace endpoints for service", "service", id, "backendKey", backendKey, "weight", weight, "error", err) + if err := r.dynamicConfigManager.ReplaceRouteEndpoints(backendKey, &service, oldEndpoints, newEndpoints, activeEndpoints); err != nil { + log.Info("router will reload as the ConfigManager could not dynamically replace endpoints for service", "service", id, "backendKey", backendKey, "error", err) return false } + + // Update weights for OTHER services if they changed + for otherServiceKey, newWeight := range cfg.ServiceUnits { + if otherServiceKey == id { + continue // Already updated above + } + if otherService, found := r.findMatchingServiceUnit(otherServiceKey); found { + oldWeight := oldWeights[otherServiceKey] + oldOtherEndpoints := oldEndpointTable[otherServiceKey] + newOtherEndpoints := cfg.EndpointTable[otherServiceKey] + log.V(4).Info("updating sibling service endpoints due to weight redistribution", "service", otherServiceKey, "backendKey", backendKey, "oldWeight", oldWeight, "newWeight", newWeight) + if err := r.dynamicConfigManager.ReplaceRouteEndpoints(backendKey, &otherService, oldOtherEndpoints, newOtherEndpoints, activeEndpoints); err != nil { + log.Info("router will reload as the ConfigManager could not update weight for sibling service", "service", otherServiceKey, "backendKey", backendKey, "error", err) + return false + } + } + } } return true @@ -965,7 +983,7 @@ func (r *templateRouter) dynamicallyReplaceEndpoints(id ServiceUnitKey, service // all the routes associated with a given service. // Note: The config should have been synced at least once initially and // the caller needs to acquire a lock [and release it]. -func (r *templateRouter) dynamicallyRemoveEndpoints(service ServiceUnit, endpoints []Endpoint) bool { +func (r *templateRouter) dynamicallyRemoveEndpoints(id ServiceUnitKey, service ServiceUnit) bool { if r.dynamicConfigManager == nil || !r.synced { return false } @@ -973,12 +991,13 @@ func (r *templateRouter) dynamicallyRemoveEndpoints(service ServiceUnit, endpoin log.V(4).Info("dynamically removing endpoints for service unit", "service", service.Name) for backendKey := range service.ServiceAliasAssociations { - if _, ok := r.state[backendKey]; !ok { + cfg, ok := r.state[backendKey] + if !ok { continue } log.V(4).Info("dynamically removing endpoints for associated backend", "backendKey", backendKey) - if err := r.dynamicConfigManager.RemoveRouteEndpoints(backendKey, endpoints); err != nil { + if err := r.dynamicConfigManager.RemoveRouteEndpoints(backendKey, cfg.EndpointTable[id]); err != nil { // Error dynamically modifying the config, so return false to cause a reload to happen. log.Info("router will reload as the ConfigManager could not dynamically remove endpoints for backend", "backendKey", backendKey, "error", err) return false @@ -997,7 +1016,7 @@ func (r *templateRouter) DeleteEndpoints(id ServiceUnitKey) { return } - configChanged := r.dynamicallyRemoveEndpoints(service, service.EndpointTable) + configChanged := r.dynamicallyRemoveEndpoints(id, service) service.EndpointTable = []Endpoint{} @@ -1167,21 +1186,18 @@ func (r *templateRouter) AddRoute(route *routev1.Route) { r.lock.Lock() defer r.lock.Unlock() - if existingConfig, exists := r.state[backendKey]; exists { + existingConfig, exists := r.state[backendKey] + if exists { if configsAreEqual(newConfig, &existingConfig) { return } log.V(4).Info("updating route", "namespace", route.Namespace, "name", route.Name) - - // Delete the route first, because modify is to be treated as delete+add - r.removeRouteInternal(route) - - // TODO - clean up service units that are no longer - // referenced. This may be challenging if a service unit can - // be referenced by more than one route, but the alternative - // is having stale service units accumulate with the attendant - // cost to router memory usage. + for key := range existingConfig.ServiceUnits { + r.removeServiceAliasAssociation(key, backendKey) + } + // TODO revisit this cleanup when adding dynamic update for certificates + r.cleanUpServiceAliasConfig(&existingConfig) } else { log.V(4).Info("adding route", "namespace", route.Namespace, "name", route.Name) } @@ -1195,11 +1211,26 @@ func (r *templateRouter) AddRoute(route *routev1.Route) { r.addServiceAliasAssociation(key, backendKey) } - configChanged := r.dynamicallyAddRoute(backendKey, route, newConfig) + r.updateEndpointTable(newConfig) + + var configChanged bool + if exists { + configChanged = r.dynamicallyUpdateRoute(backendKey, &existingConfig, newConfig) + } else { + configChanged = r.dynamicallyAddRoute(backendKey, route, newConfig) + } r.state[backendKey] = *newConfig r.stateChanged = true r.dynamicallyConfigured = r.dynamicallyConfigured && configChanged + + if exists { + // TODO - clean up service units that are no longer + // referenced. This may be challenging if a service unit can + // be referenced by more than one route, but the alternative + // is having stale service units accumulate with the attendant + // cost to router memory usage. + } } // RemoveRoute removes the given route @@ -1207,12 +1238,6 @@ func (r *templateRouter) RemoveRoute(route *routev1.Route) { r.lock.Lock() defer r.lock.Unlock() - r.removeRouteInternal(route) -} - -// removeRouteInternal removes the given route - internal -// lockless form, caller needs to ensure lock acquisition [and release]. -func (r *templateRouter) removeRouteInternal(route *routev1.Route) { backendKey := routeKey(route) serviceAliasConfig, ok := r.state[backendKey] if !ok { diff --git a/pkg/router/template/types.go b/pkg/router/template/types.go index 292d4d29a..33758d31a 100644 --- a/pkg/router/template/types.go +++ b/pkg/router/template/types.go @@ -117,6 +117,10 @@ type Endpoint struct { IdHash string NoHealthCheck bool AppProtocol string + + // fields with late update, when assigned to a ServiceAliasConfig + Weight int32 + VerifyHostname bool } // certificateManager provides the ability to write certificates for a ServiceAliasConfig @@ -223,7 +227,7 @@ type ConfigManager interface { // ReplaceRouteEndpoints replaces a subset (the ones associated with // a single service unit) of a route endpoints. - ReplaceRouteEndpoints(id ServiceAliasConfigKey, svc *ServiceUnit, oldEndpoints, newEndpoints []Endpoint, weight int32) error + ReplaceRouteEndpoints(id ServiceAliasConfigKey, svc *ServiceUnit, oldEndpoints, newEndpoints []Endpoint, activeEndpoints int) error // RemoveRouteEndpoints removes a set of endpoints from a route. RemoveRouteEndpoints(id ServiceAliasConfigKey, endpoints []Endpoint) error