Safeguard concurrent access to agentpf.client with RWMutex.

Signed-off-by: Thomas Hallgren <[email protected]>

pkg/client/agentpf/clients.go

@@ -31,7 +31,7 @@ type client struct {
 	//   cancelClient
 	//   cancelDialWatch
 	// cli and cancelClient are both safe to use without a mutex once the ready channel is closed.
-	sync.Mutex
+	sync.RWMutex
 	cli             agent.AgentClient
 	session         *manager.SessionInfo
 	info            *manager.AgentPodInfo
@@ -81,7 +81,14 @@ func (ac *client) Tunnel(ctx context.Context, opts ...grpc.CallOption) (tunnel.C
 	if err := ac.ensureConnect(ctx); err != nil {
 		return nil, err
 	}
-	tc, err := ac.cli.Tunnel(ctx, opts...)
+	ac.RLock()
+	cli := ac.cli
+	ac.RUnlock()
+	if cli == nil {
+		// Client was closed.
+		return nil, io.EOF
+	}
+	tc, err := cli.Tunnel(ctx, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -138,24 +145,24 @@ func (ac *client) connect(ctx context.Context, deleteMe func()) {
 }
 
 func (ac *client) dormant() bool {
-	ac.Lock()
+	ac.RLock()
 	dormant := !(ac.infant || ac.cli == nil || ac.info.Intercepted) && atomic.LoadInt32(&ac.tunnelCount) == 0
-	ac.Unlock()
+	ac.RUnlock()
 	return dormant
 }
 
 func (ac *client) intercepted() bool {
-	ac.Lock()
+	ac.RLock()
 	ret := ac.info.Intercepted
-	ac.Unlock()
+	ac.RUnlock()
 	return ret
 }
 
 func (ac *client) cancel() bool {
-	ac.Lock()
+	ac.RLock()
 	cc := ac.cancelClient
 	cdw := ac.cancelDialWatch
-	ac.Unlock()
+	ac.RUnlock()
 	didCancel := false
 	if cc != nil {
 		didCancel = true
@@ -169,9 +176,9 @@ func (ac *client) cancel() bool {
 }
 
 func (ac *client) setIntercepted(ctx context.Context, k string, status bool) {
-	ac.Lock()
+	ac.RLock()
 	aci := ac.info.Intercepted
-	ac.Unlock()
+	ac.RUnlock()
 	if status {
 		if aci {
 			return
@@ -188,9 +195,9 @@ func (ac *client) setIntercepted(ctx context.Context, k string, status bool) {
 
 		// This agent is no longer intercepting. Stop the dial watcher
 		dlog.Debugf(ctx, "Agent %s changed to not intercepted", k)
-		ac.Lock()
+		ac.RLock()
 		cdw := ac.cancelDialWatch
-		ac.Unlock()
+		ac.RUnlock()
 		if cdw != nil {
 			cdw()
 		}
@@ -206,6 +213,12 @@ func (ac *client) startDialWatcher(ctx context.Context) error {
 }
 
 func (ac *client) startDialWatcherReady(ctx context.Context) error {
+	ac.RLock()
+	cli := ac.cli
+	ac.RUnlock()
+	if cli == nil {
+		return fmt.Errorf("agent connection closed")
+	}
 	ctx, cancel := context.WithCancel(ctx)
 
 	// Create the dial watcher