Fix locking issues by using two locks for different purposes

sjberman · sjberman · commit 863f51558fbb · 2025-02-20T10:38:33.000-07:00
diff --git a/internal/mode/static/handler.go b/internal/mode/static/handler.go
@@ -251,13 +251,13 @@ func (h *eventHandlerImpl) processStateAndBuildConfig(
 
 		h.setLatestConfiguration(&cfg)
 
-		deployment.Lock.Lock()
+		deployment.FileLock.Lock()
 		if h.cfg.plus {
 			configApplied = h.cfg.nginxUpdater.UpdateUpstreamServers(deployment, cfg)
 		} else {
 			configApplied = h.updateNginxConf(deployment, cfg)
 		}
-		deployment.Lock.Unlock()
+		deployment.FileLock.Unlock()
 	case state.ClusterStateChange:
 		h.version++
 		cfg := dataplane.BuildConfiguration(ctx, gr, h.cfg.serviceResolver, h.version, h.cfg.plus)
@@ -269,9 +269,9 @@ func (h *eventHandlerImpl) processStateAndBuildConfig(
 
 		h.setLatestConfiguration(&cfg)
 
-		deployment.Lock.Lock()
+		deployment.FileLock.Lock()
 		configApplied = h.updateNginxConf(deployment, cfg)
-		deployment.Lock.Unlock()
+		deployment.FileLock.Unlock()
 	}
 
 	return configApplied
diff --git a/internal/mode/static/nginx/agent/command.go b/internal/mode/static/nginx/agent/command.go
@@ -143,13 +143,9 @@ func (cs *commandService) Subscribe(in pb.CommandService_SubscribeServer) error
 	go msgr.Run(ctx)
 
 	// apply current config before starting event loop
-	deployment.Lock.RLock()
 	if err := cs.setInitialConfig(ctx, deployment, conn, msgr); err != nil {
-		deployment.Lock.RUnlock()
-
 		return err
 	}
-	deployment.Lock.RUnlock()
 
 	// subscribe to the deployment broadcaster to get file updates
 	broadcaster := deployment.GetBroadcaster()
@@ -254,13 +250,15 @@ func (cs *commandService) waitForConnection(
 }
 
 // setInitialConfig gets the initial configuration for this connection and applies it.
-// The caller MUST lock the deployment before calling this.
 func (cs *commandService) setInitialConfig(
 	ctx context.Context,
 	deployment *Deployment,
 	conn *agentgrpc.Connection,
 	msgr messenger.Messenger,
 ) error {
+	deployment.FileLock.Lock()
+	defer deployment.FileLock.Unlock()
+
 	fileOverviews, configVersion := deployment.GetFileOverviews()
 	if err := msgr.Send(ctx, buildRequest(fileOverviews, conn.InstanceID, configVersion)); err != nil {
 		cs.logAndSendErrorStatus(deployment, conn, err)
@@ -419,7 +417,7 @@ func buildPlusAPIRequest(action *pb.NGINXPlusAction, instanceID string) *pb.Mana
 }
 
 func (cs *commandService) getPodOwner(podName string) (types.NamespacedName, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
 	var pods v1.PodList
@@ -450,12 +448,25 @@ func (cs *commandService) getPodOwner(podName string) (types.NamespacedName, err
 	}
 
 	var replicaSet appsv1.ReplicaSet
-	if err := cs.k8sReader.Get(
+	var replicaSetErr error
+	if err := wait.PollUntilContextCancel(
 		ctx,
-		types.NamespacedName{Namespace: pod.Namespace, Name: podOwnerRefs[0].Name},
-		&replicaSet,
+		500*time.Millisecond,
+		true, /* poll immediately */
+		func(ctx context.Context) (bool, error) {
+			if err := cs.k8sReader.Get(
+				ctx,
+				types.NamespacedName{Namespace: pod.Namespace, Name: podOwnerRefs[0].Name},
+				&replicaSet,
+			); err != nil {
+				replicaSetErr = err
+				return false, nil //nolint:nilerr // error is returned at the end
+			}
+
+			return true, nil
+		},
 	); err != nil {
-		return types.NamespacedName{}, fmt.Errorf("failed to get nginx Pod's ReplicaSet: %w", err)
+		return types.NamespacedName{}, fmt.Errorf("failed to get nginx Pod's ReplicaSet: %w", replicaSetErr)
 	}
 
 	replicaOwnerRefs := replicaSet.GetOwnerReferences()
diff --git a/internal/mode/static/nginx/agent/deployment.go b/internal/mode/static/nginx/agent/deployment.go
@@ -56,7 +56,8 @@ type Deployment struct {
 	fileOverviews    []*pb.File
 	files            []File
 
-	Lock sync.RWMutex
+	FileLock sync.RWMutex
+	errLock  sync.RWMutex
 }
 
 // newDeployment returns a new Deployment object.
@@ -72,56 +73,94 @@ func (d *Deployment) GetBroadcaster() broadcast.Broadcaster {
 	return d.broadcaster
 }
 
-// GetFileOverviews returns the current list of fileOverviews and configVersion for the deployment.
-func (d *Deployment) GetFileOverviews() ([]*pb.File, string) {
-	d.Lock.RLock()
-	defer d.Lock.RUnlock()
+// SetLatestConfigError sets the latest config apply error for the deployment.
+func (d *Deployment) SetLatestConfigError(err error) {
+	d.errLock.Lock()
+	defer d.errLock.Unlock()
 
-	return d.fileOverviews, d.configVersion
+	d.latestConfigError = err
 }
 
-// GetNGINXPlusActions returns the current NGINX Plus API Actions for the deployment.
-func (d *Deployment) GetNGINXPlusActions() []*pb.NGINXPlusAction {
-	d.Lock.RLock()
-	defer d.Lock.RUnlock()
+// SetLatestUpstreamError sets the latest upstream update error for the deployment.
+func (d *Deployment) SetLatestUpstreamError(err error) {
+	d.errLock.Lock()
+	defer d.errLock.Unlock()
 
-	return d.nginxPlusActions
+	d.latestUpstreamError = err
 }
 
 // GetLatestConfigError gets the latest config apply error for the deployment.
 func (d *Deployment) GetLatestConfigError() error {
-	d.Lock.RLock()
-	defer d.Lock.RUnlock()
+	d.errLock.RLock()
+	defer d.errLock.RUnlock()
 
 	return d.latestConfigError
 }
 
 // GetLatestUpstreamError gets the latest upstream update error for the deployment.
 func (d *Deployment) GetLatestUpstreamError() error {
-	d.Lock.RLock()
-	defer d.Lock.RUnlock()
+	d.errLock.RLock()
+	defer d.errLock.RUnlock()
 
 	return d.latestUpstreamError
 }
 
+// SetPodErrorStatus sets the error status of a Pod in this Deployment if applying the config failed.
+func (d *Deployment) SetPodErrorStatus(pod string, err error) {
+	d.errLock.Lock()
+	defer d.errLock.Unlock()
+
+	d.podStatuses[pod] = err
+}
+
 // RemovePodStatus deletes a pod from the pod status map.
 func (d *Deployment) RemovePodStatus(podName string) {
-	d.Lock.Lock()
-	defer d.Lock.Unlock()
+	d.errLock.Lock()
+	defer d.errLock.Unlock()
 
 	delete(d.podStatuses, podName)
 }
 
+// GetConfigurationStatus returns the current config status for this Deployment. It combines
+// the most recent errors (if they exist) for all Pods in the Deployment into a single error.
+func (d *Deployment) GetConfigurationStatus() error {
+	d.errLock.RLock()
+	defer d.errLock.RUnlock()
+
+	errs := make([]error, 0, len(d.podStatuses))
+	for _, err := range d.podStatuses {
+		errs = append(errs, err)
+	}
+
+	if len(errs) == 1 {
+		return errs[0]
+	}
+
+	return errors.Join(errs...)
+}
+
 /*
 The following functions for the Deployment object are UNLOCKED, meaning that they are unsafe.
-Callers of these functions MUST ensure the lock is set before calling.
+Callers of these functions MUST ensure the FileLock is set before calling.
 
 These functions are called as part of the ConfigApply or APIRequest processes. These entire processes
 are locked by the caller, hence why the functions themselves do not set the locks.
 */
 
+// GetFileOverviews returns the current list of fileOverviews and configVersion for the deployment.
+// The deployment FileLock MUST already be locked before calling this function.
+func (d *Deployment) GetFileOverviews() ([]*pb.File, string) {
+	return d.fileOverviews, d.configVersion
+}
+
+// GetNGINXPlusActions returns the current NGINX Plus API Actions for the deployment.
+// The deployment FileLock MUST already be locked before calling this function.
+func (d *Deployment) GetNGINXPlusActions() []*pb.NGINXPlusAction {
+	return d.nginxPlusActions
+}
+
 // GetFile gets the requested file for the deployment and returns its contents.
-// The deployment MUST already be locked before calling this function.
+// The deployment FileLock MUST already be locked before calling this function.
 func (d *Deployment) GetFile(name, hash string) []byte {
 	for _, file := range d.files {
 		if name == file.Meta.GetName() && hash == file.Meta.GetHash() {
@@ -133,7 +172,7 @@ func (d *Deployment) GetFile(name, hash string) []byte {
 }
 
 // SetFiles updates the nginx files and fileOverviews for the deployment and returns the message to send.
-// The deployment MUST already be locked before calling this function.
+// The deployment FileLock MUST already be locked before calling this function.
 func (d *Deployment) SetFiles(files []File) broadcast.NginxAgentMessage {
 	d.files = files
 
@@ -167,45 +206,11 @@ func (d *Deployment) SetFiles(files []File) broadcast.NginxAgentMessage {
 
 // SetNGINXPlusActions updates the deployment's latest NGINX Plus Actions to perform if using NGINX Plus.
 // Used by a Subscriber when it first connects.
-// The deployment MUST already be locked before calling this function.
+// The deployment FileLock MUST already be locked before calling this function.
 func (d *Deployment) SetNGINXPlusActions(actions []*pb.NGINXPlusAction) {
 	d.nginxPlusActions = actions
 }
 
-// SetPodErrorStatus sets the error status of a Pod in this Deployment if applying the config failed.
-// The deployment MUST already be locked before calling this function.
-func (d *Deployment) SetPodErrorStatus(pod string, err error) {
-	d.podStatuses[pod] = err
-}
-
-// SetLatestConfigError sets the latest config apply error for the deployment.
-// The deployment MUST already be locked before calling this function.
-func (d *Deployment) SetLatestConfigError(err error) {
-	d.latestConfigError = err
-}
-
-// SetLatestUpstreamError sets the latest upstream update error for the deployment.
-// The deployment MUST already be locked before calling this function.
-func (d *Deployment) SetLatestUpstreamError(err error) {
-	d.latestUpstreamError = err
-}
-
-// GetConfigurationStatus returns the current config status for this Deployment. It combines
-// the most recent errors (if they exist) for all Pods in the Deployment into a single error.
-// The deployment MUST already be locked before calling this function.
-func (d *Deployment) GetConfigurationStatus() error {
-	errs := make([]error, 0, len(d.podStatuses))
-	for _, err := range d.podStatuses {
-		errs = append(errs, err)
-	}
-
-	if len(errs) == 1 {
-		return errs[0]
-	}
-
-	return errors.Join(errs...)
-}
-
 //counterfeiter:generate . DeploymentStorer
 
 // DeploymentStorer is an interface to store Deployments.
