adaptation: support concurrent RPCs via per Pod lock

For users who do not need to perform cross Pod updates, a new per Pod
lock implementation can be selected via WithPodLocking() Option func.

Signed-off-by: JP Phillips <[email protected]>

Author: jipperinbham

pkg/adaptation/adaptation.go

@@ -58,7 +58,7 @@ type UpdateFn func(context.Context, []*ContainerUpdate) ([]*ContainerUpdate, err
 
 // Adaptation is the NRI abstraction for container runtime NRI adaptation/integration.
 type Adaptation struct {
-	sync.Mutex
+	locker      locker
 	name        string
 	version     string
 	dropinPath  string
@@ -144,6 +144,16 @@ func WithDefaultValidator(cfg *validator.DefaultValidatorConfig) Option {
 	}
 }
 
+// WithPodLocking enables the per-Pod locking strategy instead of the default global lock.
+// WARNING: This is ONLY safe if the consumer guarantees that no plugin
+// will ever request cross-Pod container updates in its responses.
+func WithPodLocking() Option {
+	return func(r *Adaptation) error {
+		r.locker = newPodLocker()
+		return nil
+	}
+}
+
 // New creates a new NRI Runtime.
 func New(name, version string, syncFn SyncFn, updateFn UpdateFn, opts ...Option) (*Adaptation, error) {
 	var err error
@@ -175,6 +185,7 @@ func New(name, version string, syncFn SyncFn, updateFn UpdateFn, opts ...Option)
 	}
 
 	r := &Adaptation{
+		locker:      newGlobalLocker(),
 		name:        name,
 		version:     version,
 		syncFn:      syncFn,
@@ -192,7 +203,7 @@ func New(name, version string, syncFn SyncFn, updateFn UpdateFn, opts ...Option)
 		}
 	}
 
-	log.Infof(noCtx, "runtime interface created")
+	log.Infof(noCtx, "runtime interface created, using locker: %T", r.locker)
 
 	return r, nil
 }
@@ -201,8 +212,8 @@ func New(name, version string, syncFn SyncFn, updateFn UpdateFn, opts ...Option)
 func (r *Adaptation) Start() error {
 	log.Infof(noCtx, "runtime interface starting up...")
 
-	r.Lock()
-	defer r.Unlock()
+	unlock := r.locker.Lock(noCtx)
+	defer unlock()
 
 	if err := r.startPlugins(); err != nil {
 		return err
@@ -219,8 +230,8 @@ func (r *Adaptation) Start() error {
 func (r *Adaptation) Stop() {
 	log.Infof(noCtx, "runtime interface shutting down...")
 
-	r.Lock()
-	defer r.Unlock()
+	unlock := r.locker.Lock(noCtx)
+	defer unlock()
 
 	r.stopListener()
 	r.stopPlugins()
@@ -234,8 +245,9 @@ func (r *Adaptation) RunPodSandbox(ctx context.Context, evt *StateChangeEvent) e
 
 // UpdatePodSandbox relays the corresponding CRI request to plugins.
 func (r *Adaptation) UpdatePodSandbox(ctx context.Context, req *UpdatePodSandboxRequest) (*UpdatePodSandboxResponse, error) {
-	r.Lock()
-	defer r.Unlock()
+	podUID := req.GetPod().GetId()
+	unlock := r.locker.LockPod(ctx, podUID)
+	defer unlock()
 	defer r.removeClosedPlugins()
 
 	for _, plugin := range r.plugins {
@@ -268,8 +280,9 @@ func (r *Adaptation) RemovePodSandbox(ctx context.Context, evt *StateChangeEvent
 
 // CreateContainer relays the corresponding CRI request to plugins.
 func (r *Adaptation) CreateContainer(ctx context.Context, req *CreateContainerRequest) (*CreateContainerResponse, error) {
-	r.Lock()
-	defer r.Unlock()
+	podUID := req.GetPod().GetId()
+	unlock := r.locker.LockPod(ctx, podUID)
+	defer unlock()
 	defer r.removeClosedPlugins()
 
 	var (
@@ -321,8 +334,9 @@ func (r *Adaptation) PostStartContainer(ctx context.Context, evt *StateChangeEve
 
 // UpdateContainer relays the corresponding CRI request to plugins.
 func (r *Adaptation) UpdateContainer(ctx context.Context, req *UpdateContainerRequest) (*UpdateContainerResponse, error) {
-	r.Lock()
-	defer r.Unlock()
+	podUID := req.GetPod().GetId()
+	unlock := r.locker.LockPod(ctx, podUID)
+	defer unlock()
 	defer r.removeClosedPlugins()
 
 	result := collectUpdateContainerResult(req)
@@ -348,8 +362,9 @@ func (r *Adaptation) PostUpdateContainer(ctx context.Context, evt *StateChangeEv
 
 // StopContainer relays the corresponding CRI request to plugins.
 func (r *Adaptation) StopContainer(ctx context.Context, req *StopContainerRequest) (*StopContainerResponse, error) {
-	r.Lock()
-	defer r.Unlock()
+	podUID := req.GetPod().GetId()
+	unlock := r.locker.LockPod(ctx, podUID)
+	defer unlock()
 	defer r.removeClosedPlugins()
 
 	result := collectStopContainerResult()
@@ -379,24 +394,39 @@ func (r *Adaptation) StateChange(ctx context.Context, evt *StateChangeEvent) err
 		return errors.New("invalid (unset) event in state change notification")
 	}
 
-	r.Lock()
-	defer r.Unlock()
-	defer r.removeClosedPlugins()
+	defer func() {
+		unlock := r.locker.Lock(ctx)
+		r.removeClosedPlugins()
+		unlock()
+	}()
 
-	for _, plugin := range r.plugins {
-		err := plugin.StateChange(ctx, evt)
-		if err != nil {
-			return err
+	podUID := evt.GetPod().GetId()
+	if err := func() error {
+		unlock := r.locker.LockPod(ctx, podUID)
+		defer unlock()
+		for _, plugin := range r.plugins {
+			err := plugin.StateChange(ctx, evt)
+			if err != nil {
+				return err
+			}
 		}
+		return nil
+	}(); err != nil {
+		return err
+	}
+
+	// Cleanup pod lock resources if this is the removal event
+	if evt.Event == Event_REMOVE_POD_SANDBOX {
+		r.locker.CleanupPod(ctx, podUID)
 	}
 
 	return nil
 }
 
 // Perform a set of unsolicited container updates requested by a plugin.
 func (r *Adaptation) updateContainers(ctx context.Context, req []*ContainerUpdate) ([]*ContainerUpdate, error) {
-	r.Lock()
-	defer r.Unlock()
+	unlock := r.locker.Lock(ctx)
+	defer unlock()
 
 	return r.updateFn(ctx, req)
 }
@@ -608,13 +638,13 @@ func (r *Adaptation) acceptPluginConnections(l net.Listener) error {
 			if err != nil {
 				log.Infof(ctx, "failed to synchronize plugin: %v", err)
 			} else {
-				r.Lock()
+				unlock := r.locker.Lock(ctx)
 				r.plugins = append(r.plugins, p)
 				if p.isContainerAdjustmentValidator() {
 					r.validators = append(r.validators, p)
 				}
 				r.sortPlugins()
-				r.Unlock()
+				unlock()
 				log.Infof(ctx, "plugin %q connected and synchronized", p.name())
 			}
 

pkg/adaptation/lock.go

@@ -0,0 +1,105 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package adaptation
+
+import (
+	"context"
+	"sync"
+)
+
+// UnlockFunc is a function returned by Lock methods to release the acquired lock.
+type UnlockFunc func()
+
+// locker defines the internal interface for locking strategies.
+type locker interface {
+	// Lock acquires the global lock.
+	Lock(ctx context.Context) UnlockFunc
+	// LockPod acquires the lock for a specific pod.
+	LockPod(ctx context.Context, podUID string) UnlockFunc
+	// CleanupPod performs cleanup for a pod lock.
+	CleanupPod(ctx context.Context, podUID string)
+}
+
+var _ locker = &globalLocker{}
+
+// globalLocker implements locker using a single global mutex.
+type globalLocker struct {
+	mu sync.Mutex
+}
+
+// newGlobalLocker creates the default global locker.
+func newGlobalLocker() locker {
+	return &globalLocker{}
+}
+
+func (m *globalLocker) Lock(context.Context) UnlockFunc {
+	m.mu.Lock()
+	return m.mu.Unlock
+}
+
+func (m *globalLocker) LockPod(context.Context, string) UnlockFunc {
+	// Ignores podUID, uses the single global lock
+	m.mu.Lock()
+	return m.mu.Unlock
+}
+
+func (m *globalLocker) CleanupPod(context.Context, string) {
+	// No-op
+}
+
+var _ locker = &podLocker{}
+
+// podLocker implements locker using a separate mutex for each Pod UID
+// and using its own main mutex 'mu' as the global lock.
+type podLocker struct {
+	mu    sync.Mutex             // Protects access to the locks map AND acts as the global lock.
+	locks map[string]*sync.Mutex // Map from Pod UID to its mutex
+}
+
+// newPodLocker creates the per-pod locker.
+func newPodLocker() locker {
+	return &podLocker{
+		locks: make(map[string]*sync.Mutex),
+	}
+}
+
+// Lock acquires the main mutex, acting as the global lock.
+func (m *podLocker) Lock(context.Context) UnlockFunc {
+	m.mu.Lock()
+	return m.mu.Unlock
+}
+
+// LockPod acquires the lock for a specific Pod UID.
+func (m *podLocker) LockPod(_ context.Context, podUID string) UnlockFunc {
+	m.mu.Lock()
+	podMu, ok := m.locks[podUID]
+	if !ok {
+		podMu = &sync.Mutex{}
+		m.locks[podUID] = podMu
+	}
+	m.mu.Unlock()
+
+	podMu.Lock()
+	return podMu.Unlock
+}
+
+// CleanupPod removes the lock for a specific Pod UID from the map.
+func (m *podLocker) CleanupPod(_ context.Context, podUID string) {
+	m.mu.Lock()
+	delete(m.locks, podUID)
+	m.mu.Unlock()
+}