Skip to content

OCPBUGS-77344: enable dcm on route changes #778

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jcmoraisjr wants to merge 2 commits into
base: master
Choose a base branch
from
Open

OCPBUGS-77344: enable dcm on route changes #778

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9496569
Revert "reload router if route is changed on DCM"
jcmoraisjr May 27, 2026
b69f565
enable dcm on route changes
jcmoraisjr Apr 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 104 additions & 33 deletions pkg/router/template/configmanager/haproxy/backend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
}

Expand All @@ -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)
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

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
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
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 {
Expand All @@ -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.
Expand All @@ -376,15 +438,15 @@ 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:
cmd += " ssl"
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 {
Expand All @@ -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:
Expand All @@ -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"
Expand All @@ -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)
}
Expand Down Expand Up @@ -560,6 +630,7 @@ const (
apiDelServer
apiSetHealth
apiSetServerAddr
apiSetServerHealth
apiSetServerWeight
apiSetServerState
)
Expand All @@ -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
Expand Down
Loading