From 9496569a133cd121d302dac99e52178f1326edd3 Mon Sep 17 00:00:00 2001 From: Joao Morais Date: Wed, 27 May 2026 14:28:04 -0300 Subject: [PATCH 1/2] Revert "reload router if route is changed on DCM" This reverts commit 0e47c836e9f2e5bbbf95c35604fc89a6fa8b32ee. --- pkg/router/template/router.go | 8 -------- 1 file changed, 8 deletions(-) diff --git a/pkg/router/template/router.go b/pkg/router/template/router.go index 79f46983b..beaf9c3c2 100644 --- a/pkg/router/template/router.go +++ b/pkg/router/template/router.go @@ -813,10 +813,6 @@ func (r *templateRouter) dynamicallyAddRoute(backendKey ServiceAliasConfigKey, r } } - // 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 { @@ -859,10 +855,6 @@ func (r *templateRouter) dynamicallyAddRoute(backendKey ServiceAliasConfigKey, r // 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 } From b69f565612d05b9686f6c443f1dd7083db939837 Mon Sep 17 00:00:00 2001 From: Joao Morais Date: Mon, 20 Apr 2026 13:50:49 -0300 Subject: [PATCH 2/2] enable dcm on route changes DCM was disabled on route changes due to the way the resource changes are being handled: without DCM, the template plugin applies a route remove, followed by a route add, which works well in the fork-and-reload approach since the removal is not applied until adding the route again succeed. On DCM, each of these operations are done separately as well, which means that the route is completely disabled during some time. This update changes this behavior by saving previous and updated states of a route, identifying the differences, and dynamically applying them if possible. A fork-and-reload is made in case the dynamic update is not possible, as before. Part of the changes also include extending the Endpoint struct so that it stores the weight and the VerifyHostname properties of each endpoint in order to make comparisons easier, as well as decrease the number of parameters sent to inner methods. These fields cannot be updated during the Endpoint creation, since they depend on the backend the endpoint is assigned. The router's updateEndpointTable() method centralizes all changes to these context based fields. https://redhat.atlassian.net/browse/OCPBUGS-77344 --- .../template/configmanager/haproxy/backend.go | 137 ++++++++--- .../configmanager/haproxy/backend_test.go | 186 +++++++++++++-- .../haproxy/blueprint_plugin_test.go | 2 +- .../template/configmanager/haproxy/manager.go | 133 +++++++---- pkg/router/template/router.go | 225 ++++++++++-------- pkg/router/template/types.go | 6 +- 6 files changed, 488 insertions(+), 201 deletions(-) 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 beaf9c3c2..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,14 +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) - } - } - // If no initial sync was done, don't try to dynamically add the // route as we will need a reload anyway. if !r.synced { @@ -825,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 @@ -851,6 +852,56 @@ 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]. @@ -888,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 - } + // Backup old state + oldEndpointTable := cfg.EndpointTable + oldEndpoints := oldEndpointTable[id] + oldWeights := cfg.ServiceUnitNames - // 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 - } - } - - // 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 @@ -957,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 } @@ -965,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 @@ -989,7 +1016,7 @@ func (r *templateRouter) DeleteEndpoints(id ServiceUnitKey) { return } - configChanged := r.dynamicallyRemoveEndpoints(service, service.EndpointTable) + configChanged := r.dynamicallyRemoveEndpoints(id, service) service.EndpointTable = []Endpoint{} @@ -1159,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) } @@ -1187,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 @@ -1199,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