Merge pull request #1267 from jcmoraisjr/jm-reload-ctx

add context on socket calls

Author: jcmoraisjr

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

@@ -139,7 +139,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"
@@ -59,7 +60,7 @@ type InstanceOptions struct {
 	ReloadQueue       *workqueue.WorkQueue[any]
 	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.
@@ -84,7 +85,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(),
@@ -638,7 +639,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)
 		}()
 	})
@@ -655,7 +656,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",
@@ -682,7 +683,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)
@@ -716,14 +717,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())
@@ -737,7 +738,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)
 	}
@@ -756,17 +757,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.
 		// This is needed on 2.6 and older, whose `reload` command runs async.
 		select {
 		case <-time.After(time.Second):
-		case <-i.options.StopCh:
+		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.conns.Master())
+		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)
 		}