add context on socket calls

HAProxy Ingress shutdown is eventually racing during stop, leading to
the 25s controller manager timeout, taking more time for the controller
pod to be finalized. This is happening when sigint/sigterm happens just
before a new haproxy reload is issued: the signal closes the reconciler
context, which stops haproxy, making the reload check to wait forever
until haproxy is back again, leading to the timeout of the controller
manager.

This update changes the StopCh channel to a proper context, and uses
the context on all the socket calls.

Should be added to v0.15 branch.

Author: jcmoraisjr

Makefile

@@ -64,8 +64,9 @@ setup-envtest:
 test-integration: gotestsum setup-envtest
 	@echo
 	@echo "Running Kubernetes $(HAPROXY_INGRESS_ENVTEST)"
+	haproxy -v
 	KUBEBUILDER_ASSETS="$(shell $(LOCAL_SETUP_ENVTEST) use $(HAPROXY_INGRESS_ENVTEST) --bin-dir $(LOCALBIN) -i -p path)"\
-		$(LOCAL_GOTESTSUM) --format=testname -- -count=1 -tags=cgo ./tests/integration/...
+		$(LOCAL_GOTESTSUM) --format=testname --hide-summary=output -- -count=1 -tags=cgo ./tests/integration/...
 
 .PHONY: linux-build
 linux-build:

pkg/controller/legacy/controller.go

@@ -145,7 +145,7 @@ func (hc *HAProxyController) configController() {
 		ReloadStrategy:    hc.cfg.ReloadStrategy,
 		MaxOldConfigFiles: hc.cfg.MaxOldConfigFiles,
 		SortEndpointsBy:   hc.cfg.SortEndpointsBy,
-		StopCh:            hc.stopCh,
+		StopCtx:           wait.ContextForChannel(hc.stopCh),
 		TrackInstances:    hc.cfg.TrackOldInstances,
 		ValidateConfig:    hc.cfg.ValidateConfig,
 	}

pkg/controller/reconciler/reconciler.go

@@ -51,6 +51,9 @@ type rparam struct {
 
 // Reconcile ...
 func (r *IngressReconciler) Reconcile(ctx context.Context, req rparam) (ctrl.Result, error) {
+	if err := ctx.Err(); err != nil {
+		return ctrl.Result{}, err
+	}
 	changed := r.watchers.getChangedObjects()
 	changed.NeedFullSync = req.fullsync
 	err := r.Services.ReconcileIngress(ctx, changed)

pkg/controller/services/services.go

@@ -142,7 +142,7 @@ func (s *Services) setup(ctx context.Context) error {
 		ReloadStrategy:    cfg.ReloadStrategy,
 		MaxOldConfigFiles: cfg.MaxOldConfigFiles,
 		SortEndpointsBy:   cfg.SortEndpointsBy,
-		StopCh:            ctx.Done(),
+		StopCtx:           ctx,
 		TrackInstances:    cfg.TrackOldInstances,
 		ValidateConfig:    cfg.ValidateConfig,
 		AcmeSigner:        acmeSigner,

pkg/haproxy/connections.go

@@ -17,22 +17,25 @@ limitations under the License.
 package haproxy
 
 import (
+	"context"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/jcmoraisjr/haproxy-ingress/pkg/haproxy/socket"
 )
 
-func newConnections(masterSock, adminSock string) *connections {
+func newConnections(ctx context.Context, masterSock, adminSock string) *connections {
 	return &connections{
+		ctx:        ctx,
 		mutex:      sync.Mutex{},
 		masterSock: masterSock,
 		adminSock:  adminSock,
 	}
 }
 
 type connections struct {
+	ctx          context.Context
 	mutex        sync.Mutex
 	masterSock   string
 	adminSock    string
@@ -47,7 +50,7 @@ func (c *connections) TrackCurrentInstance(timeoutStopDur, closeSessDur time.Dur
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	c.shrinkConns()
-	sock := socket.NewSocketConcurrent(c.adminSock, true)
+	sock := socket.NewSocketConcurrent(c.ctx, c.adminSock, true)
 	if err := sock.Unlistening(); err != nil {
 		return err
 	}
@@ -132,14 +135,14 @@ func shutdownSessionsSync(sock socket.HAProxySocket, duration time.Duration) {
 
 func (c *connections) Admin() socket.HAProxySocket {
 	if c.admin == nil {
-		c.admin = socket.NewSocket(c.adminSock, false)
+		c.admin = socket.NewSocket(c.ctx, c.adminSock, false)
 	}
 	return c.admin
 }
 
 func (c *connections) Master() socket.HAProxySocket {
 	if c.master == nil {
-		c.master = socket.NewSocket(c.masterSock, false)
+		c.master = socket.NewSocket(c.ctx, c.masterSock, false)
 	}
 	return c.master
 }
@@ -148,14 +151,14 @@ func (c *connections) DynUpdate() socket.HAProxySocket {
 	if c.dynUpdate == nil {
 		// using a non persistent connection (keep alive false)
 		// to ensure that the current instance will be used
-		c.dynUpdate = socket.NewSocket(c.adminSock, false)
+		c.dynUpdate = socket.NewSocket(c.ctx, c.adminSock, false)
 	}
 	return c.dynUpdate
 }
 
 func (c *connections) IdleChk() socket.HAProxySocket {
 	if c.idleChk == nil {
-		c.idleChk = socket.NewSocket(c.adminSock, false)
+		c.idleChk = socket.NewSocket(c.ctx, c.adminSock, false)
 	}
 	return c.idleChk
 }

pkg/haproxy/instance.go

@@ -17,6 +17,7 @@ limitations under the License.
 package haproxy
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"os/exec"
@@ -58,7 +59,7 @@ type InstanceOptions struct {
 	ReloadQueue       utils.QueueFacade
 	ReloadStrategy    string
 	SortEndpointsBy   string
-	StopCh            <-chan struct{}
+	StopCtx           context.Context
 	TrackInstances    bool
 	ValidateConfig    bool
 	// TODO Fake is used to skip real haproxy calls. Use a mock instead.
@@ -83,7 +84,7 @@ func CreateInstance(logger types.Logger, options InstanceOptions) Instance {
 		waitProc: make(chan struct{}),
 		logger:   logger,
 		options:  &options,
-		conns:    newConnections(options.MasterSocket, options.AdminSocket),
+		conns:    newConnections(options.StopCtx, options.MasterSocket, options.AdminSocket),
 		metrics:  options.Metrics,
 		//
 		haproxyTmpl:     template.CreateConfig(),
@@ -648,7 +649,7 @@ func (i *instance) reloadEmbeddedDaemon() error {
 func (i *instance) reloadEmbeddedMasterWorker() error {
 	i.embdStart.Do(func() {
 		go func() {
-			wait.Until(i.startHAProxySync, 4*time.Second, i.options.StopCh)
+			wait.UntilWithContext(i.options.StopCtx, i.startHAProxySync, 4*time.Second)
 			close(i.waitProc)
 		}()
 	})
@@ -665,7 +666,7 @@ func (i *instance) reloadEmbeddedMasterWorker() error {
 	return i.waitWorker(prevReloads)
 }
 
-func (i *instance) startHAProxySync() {
+func (i *instance) startHAProxySync(ctx context.Context) {
 	cmd := exec.Command(
 		"haproxy",
 		"-W",
@@ -692,7 +693,7 @@ func (i *instance) startHAProxySync() {
 		close(wait)
 	}()
 	select {
-	case <-i.options.StopCh:
+	case <-ctx.Done():
 		i.logger.Info("stopping haproxy master process (pid: %d)", cmd.Process.Pid)
 		if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
 			i.logger.Error("error stopping haproxy process: %v", err)
@@ -726,14 +727,14 @@ func (i *instance) waitMaster() error {
 	errCh := make(chan error)
 	masterSock := i.conns.Master()
 	go func() {
-		_, err := socket.HAProxyProcs(masterSock)
+		_, err := socket.HAProxyProcs(i.options.StopCtx, masterSock)
 		errCh <- err
 	}()
 	for {
 		select {
 		case err := <-errCh:
 			return err
-		case <-i.options.StopCh:
+		case <-i.options.StopCtx.Done():
 			return fmt.Errorf("received sigterm")
 		case <-time.After(10 * time.Second):
 			i.logger.Info("... still waiting for the master socket '%s'", masterSock.Address())
@@ -747,7 +748,7 @@ func (i *instance) reloadWorker() (int, error) {
 			i.logger.Warn("failed to persist servers state before worker reload: %w", err)
 		}
 	}
-	procs, err := socket.HAProxyProcs(i.conns.Master())
+	procs, err := socket.HAProxyProcs(i.options.StopCtx, i.conns.Master())
 	if err != nil {
 		return 0, fmt.Errorf("error reading haproxy procs: %w", err)
 	}
@@ -758,12 +759,17 @@ func (i *instance) reloadWorker() (int, error) {
 }
 
 func (i *instance) waitWorker(prevReloads int) error {
-	out, err := socket.HAProxyProcs(i.conns.Master())
+	out, err := socket.HAProxyProcs(i.options.StopCtx, i.conns.Master())
 	for err == nil && out.Master.Reloads <= prevReloads {
-		// continues to wait until we can see `reloads` greater than the previous one.
-		time.Sleep(time.Second)
-		i.logger.Info("reconnecting to master socket. current-reloads=%d prev-reloads=%d", out.Master.Reloads, prevReloads)
-		out, err = socket.HAProxyProcs(i.conns.Master())
+		// Continues to wait until we can see `Reloads` greater than the previous one.
+		// This is needed on 2.6 and older, whose `reload` command runs async.
+		select {
+		case <-time.After(time.Second):
+		case <-i.options.StopCtx.Done():
+			return fmt.Errorf("received sigterm")
+		}
+		i.logger.Info("reconnecting master socket. current-reloads=%d prev-reloads=%d", out.Master.Reloads, prevReloads)
+		out, err = socket.HAProxyProcs(i.options.StopCtx, i.conns.Master())
 	}
 	if err != nil {
 		return fmt.Errorf("error reading procs from master socket: %w", err)

pkg/haproxy/socket/socket.go

@@ -17,6 +17,7 @@ limitations under the License.
 package socket
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -33,19 +34,20 @@ import (
 )
 
 // NewSocket ...
-func NewSocket(address string, keepalive bool) HAProxySocket {
-	return newSocket(address, keepalive)
+func NewSocket(ctx context.Context, address string, keepalive bool) HAProxySocket {
+	return newSocket(ctx, address, keepalive)
 }
 
 // NewSocketConcurrent ...
-func NewSocketConcurrent(address string, keepalive bool) HAProxySocket {
-	s := newSocket(address, keepalive)
+func NewSocketConcurrent(ctx context.Context, address string, keepalive bool) HAProxySocket {
+	s := newSocket(ctx, address, keepalive)
 	s.mutex = &sync.Mutex{}
 	return s
 }
 
-func newSocket(address string, keepalive bool) *sock {
+func newSocket(ctx context.Context, address string, keepalive bool) *sock {
 	return &sock{
+		ctx:       ctx,
 		address:   address,
 		listening: true,
 		keepalive: keepalive,
@@ -63,6 +65,7 @@ type HAProxySocket interface {
 }
 
 type sock struct {
+	ctx       context.Context
 	mutex     *sync.Mutex
 	address   string
 	listening bool
@@ -200,7 +203,8 @@ func (s *sock) acquireConn() (net.Conn, error) {
 		if !s.listening {
 			return nil, fmt.Errorf("cannot connect to '%s': listening is down", s.address)
 		}
-		c, err := net.Dial("unix", s.address)
+		var d net.Dialer
+		c, err := d.DialContext(s.ctx, "unix", s.address)
 		if err != nil {
 			return nil, err
 		}
@@ -244,17 +248,20 @@ type Proc struct {
 // instance. Waits for the reload to complete while master CLI is down and the
 // attempt to connect leads to a connection refused. Some context:
 // https://www.mail-archive.com/[email protected]/msg38415.html
-// The amount of time between attempts increases exponentially between 1ms and 64ms,
-// and arithmetically between 128ms and 1s in order to save CPU on long reload events
+// The amount of time between attempts increases exponentially between 1ms and 256ms,
+// and arithmetically between 256ms and 1s in order to save CPU on long reload events
 // and quit fast on the fastest ones. The whole processing time can be calculated by
 // the caller as the haproxy reload time.
-func HAProxyProcs(masterSocket HAProxySocket) (*ProcTable, error) {
-	maxLogWait := 64 * time.Millisecond
+func HAProxyProcs(ctx context.Context, masterSocket HAProxySocket) (*ProcTable, error) {
+	maxLogWait := 128 * time.Millisecond
 	logFactor := 2
 	maxArithWait := 1024 * time.Millisecond
-	arithFactor := 32 * time.Millisecond
+	arithFactor := 64 * time.Millisecond
 	wait := time.Millisecond
 	for {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
 		time.Sleep(wait)
 		out, err := masterSocket.Send(nil, "show proc")
 		if !waitHAProxy(masterSocket, err) {

pkg/haproxy/socket/socket_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package socket
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"regexp"
@@ -129,13 +130,14 @@ func testSocket(t *testing.T, keepalive bool) {
 			},
 		},
 	}
+	ctx := context.Background()
 	clientSocket := make([]HAProxySocket, len(testCases))
 	clientSocketPos := make([]HAProxySocket, len(testCases))
 	masterSocket := make([]HAProxySocket, len(testCases))
 	for i, test := range testCases {
-		clientSocket[i] = NewSocket(clisock, keepalive)
-		clientSocketPos[i] = NewSocket(clisock, false)
-		masterSocket[i] = NewSocket(mastersock, keepalive)
+		clientSocket[i] = NewSocket(ctx, clisock, keepalive)
+		clientSocketPos[i] = NewSocket(ctx, clisock, false)
+		masterSocket[i] = NewSocket(ctx, mastersock, keepalive)
 		time.Sleep(test.waitBefore)
 		var sock, sockPos HAProxySocket
 		if test.master {
@@ -406,14 +408,15 @@ func TestHAProxyProcs(t *testing.T) {
 			},
 		},
 	}
+	ctx := context.Background()
 	for i, test := range testCases {
 		c := setup(t)
 		cli := &clientMock{
 			cmdOutput: test.cmdOutput,
 			cmdError:  test.cmdError,
 			hasSock:   test.hasSock,
 		}
-		out, err := HAProxyProcs(cli)
+		out, err := HAProxyProcs(ctx, cli)
 		if !reflect.DeepEqual(out, test.expOutput) {
 			t.Errorf("output differs on %d - expected: %+v, actual: %+v", i, test.expOutput, out)
 		}
@@ -433,7 +436,7 @@ func TestHAProxyProcsLoop(t *testing.T) {
 		// 0
 		{
 			reload:   0,
-			minDelay: 1 * time.Millisecond,
+			minDelay: 0 * time.Millisecond,
 			maxCnt:   1,
 		},
 		// 1
@@ -444,19 +447,20 @@ func TestHAProxyProcsLoop(t *testing.T) {
 		},
 		// 2
 		{
-			reload:   450 * time.Millisecond,
-			minDelay: (1 + 2 + 4 + 8 + 16 + 32 + 64 + 96 + 128 + 160) * time.Millisecond,
+			reload:   650 * time.Millisecond,
+			minDelay: (1 + 2 + 4 + 8 + 16 + 32 + 64 + 128 + 192 + 256) * time.Millisecond,
 			maxCnt:   10,
 		},
 	}
+	ctx := context.Background()
 	for i, test := range testCases {
 		c := setup(t)
 		cli := &clientMock{
 			cmdError: syscall.ECONNREFUSED,
 		}
 		time.AfterFunc(test.reload, func() { cli.cmdError = nil })
 		start := time.Now()
-		_, err := HAProxyProcs(cli)
+		_, err := HAProxyProcs(ctx, cli)
 		if err != nil {
 			t.Errorf("%d should not return an error: %v", i, err)
 		}