cri,nri: block NRI plugin sync. during event processing.

klihub · klihub · commit 11af05177545 · 2025-02-13T21:12:18.000+02:00
Block the synchronization of registering NRI plugins during
CRI events to avoid the plugin ending up in an inconsistent
starting state after initial sync (missing pods, containers
or missed events for some pods or containers).

Signed-off-by: Krisztian Litkey &lt;krisztian.litkey@intel.com&gt;
diff --git a/pkg/cri/nri/nri_api_linux.go b/pkg/cri/nri/nri_api_linux.go
@@ -361,6 +361,15 @@ func (a *API) WithContainerExit(criCtr *cstore.Container) containerd.ProcessDele
 	}
 }
 
+type PluginSyncBlock = nri.PluginSyncBlock
+
+func (a *API) BlockPluginSync() *PluginSyncBlock {
+	if a.IsDisabled() {
+		return nil
+	}
+	return a.nri.BlockPluginSync()
+}
+
 //
 // NRI-CRI 'domain' interface
 //
diff --git a/pkg/cri/nri/nri_api_other.go b/pkg/cri/nri/nri_api_other.go
@@ -108,6 +108,14 @@ func (*API) WithContainerExit(*cstore.Container) containerd.ProcessDeleteOpts {
 	}
 }
 
+type PluginSyncBlock struct{}
+
+func (*API) BlockPluginSync() *PluginSyncBlock {
+	return nil
+}
+
+func (*PluginSyncBlock) Unblock() {}
+
 //
 // NRI-CRI no-op 'domain' interface
 //
diff --git a/pkg/cri/sbserver/container_create.go b/pkg/cri/sbserver/container_create.go
@@ -285,6 +285,8 @@ func (c *criService) CreateContainer(ctx context.Context, r *runtime.CreateConta
 		}
 	}()
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	var cntr containerd.Container
 	if cntr, err = c.client.NewContainer(ctx, id, opts...); err != nil {
 		return nil, fmt.Errorf("failed to create containerd container: %w", err)
diff --git a/pkg/cri/sbserver/container_remove.go b/pkg/cri/sbserver/container_remove.go
@@ -44,6 +44,9 @@ func (c *criService) RemoveContainer(ctx context.Context, r *runtime.RemoveConta
 		log.G(ctx).Tracef("RemoveContainer called for container %q that does not exist", ctrID)
 		return &runtime.RemoveContainerResponse{}, nil
 	}
+
+	defer c.nri.BlockPluginSync().Unblock()
+
 	id := container.ID
 	i, err := container.Container.Info(ctx)
 	if err != nil {
diff --git a/pkg/cri/sbserver/container_start.go b/pkg/cri/sbserver/container_start.go
@@ -146,6 +146,8 @@ func (c *criService) StartContainer(ctx context.Context, r *runtime.StartContain
 		return nil, fmt.Errorf("failed to wait for containerd task: %w", err)
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	defer func() {
 		if retErr != nil {
 			deferCtx, deferCancel := ctrdutil.DeferContext()
diff --git a/pkg/cri/sbserver/container_stop.go b/pkg/cri/sbserver/container_stop.go
@@ -50,6 +50,8 @@ func (c *criService) StopContainer(ctx context.Context, r *runtime.StopContainer
 		return &runtime.StopContainerResponse{}, nil
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	if err := c.stopContainer(ctx, container, time.Duration(r.GetTimeout())*time.Second); err != nil {
 		return nil, err
 	}
diff --git a/pkg/cri/sbserver/container_update_resources.go b/pkg/cri/sbserver/container_update_resources.go
@@ -47,6 +47,8 @@ func (c *criService) UpdateContainerResources(ctx context.Context, r *runtime.Up
 		return nil, err
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	resources := r.GetLinux()
 	updated, err := c.nri.UpdateContainerResources(ctx, &sandbox, &container, resources)
 	if err != nil {
diff --git a/pkg/cri/sbserver/sandbox_remove.go b/pkg/cri/sbserver/sandbox_remove.go
@@ -44,6 +44,9 @@ func (c *criService) RemovePodSandbox(ctx context.Context, r *runtime.RemovePodS
 			r.GetPodSandboxId())
 		return &runtime.RemovePodSandboxResponse{}, nil
 	}
+
+	defer c.nri.BlockPluginSync().Unblock()
+
 	// Use the full sandbox id.
 	id := sandbox.ID
 
diff --git a/pkg/cri/sbserver/sandbox_run.go b/pkg/cri/sbserver/sandbox_run.go
@@ -263,6 +263,8 @@ func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandbox
 
 	sandbox.ProcessLabel = labels["selinux_label"]
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	err = c.nri.RunPodSandbox(ctx, &sandbox)
 	if err != nil {
 		return nil, fmt.Errorf("NRI RunPodSandbox failed: %w", err)
diff --git a/pkg/cri/sbserver/sandbox_stop.go b/pkg/cri/sbserver/sandbox_stop.go
@@ -45,6 +45,8 @@ func (c *criService) StopPodSandbox(ctx context.Context, r *runtime.StopPodSandb
 		return &runtime.StopPodSandboxResponse{}, nil
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	if err := c.stopPodSandbox(ctx, sandbox); err != nil {
 		return nil, err
 	}
diff --git a/pkg/cri/server/container_create.go b/pkg/cri/server/container_create.go
@@ -259,6 +259,8 @@ func (c *criService) CreateContainer(ctx context.Context, r *runtime.CreateConta
 		}
 	}()
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	var cntr containerd.Container
 	if cntr, err = c.client.NewContainer(ctx, id, opts...); err != nil {
 		return nil, fmt.Errorf("failed to create containerd container: %w", err)
diff --git a/pkg/cri/server/container_remove.go b/pkg/cri/server/container_remove.go
@@ -44,6 +44,9 @@ func (c *criService) RemoveContainer(ctx context.Context, r *runtime.RemoveConta
 		log.G(ctx).Tracef("RemoveContainer called for container %q that does not exist", ctrID)
 		return &runtime.RemoveContainerResponse{}, nil
 	}
+
+	defer c.nri.BlockPluginSync().Unblock()
+
 	id := container.ID
 	i, err := container.Container.Info(ctx)
 	if err != nil {
diff --git a/pkg/cri/server/container_start.go b/pkg/cri/server/container_start.go
@@ -146,6 +146,8 @@ func (c *criService) StartContainer(ctx context.Context, r *runtime.StartContain
 		return nil, fmt.Errorf("failed to wait for containerd task: %w", err)
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	defer func() {
 		if retErr != nil {
 			deferCtx, deferCancel := ctrdutil.DeferContext()
diff --git a/pkg/cri/server/container_stop.go b/pkg/cri/server/container_stop.go
@@ -50,6 +50,8 @@ func (c *criService) StopContainer(ctx context.Context, r *runtime.StopContainer
 		return &runtime.StopContainerResponse{}, nil
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	if err := c.stopContainer(ctx, container, time.Duration(r.GetTimeout())*time.Second); err != nil {
 		return nil, err
 	}
diff --git a/pkg/cri/server/container_update_resources.go b/pkg/cri/server/container_update_resources.go
@@ -47,6 +47,8 @@ func (c *criService) UpdateContainerResources(ctx context.Context, r *runtime.Up
 		return nil, err
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	resources := r.GetLinux()
 	updated, err := c.nri.UpdateContainerResources(ctx, &sandbox, &container, resources)
 	if err != nil {
diff --git a/pkg/cri/server/sandbox_remove.go b/pkg/cri/server/sandbox_remove.go
@@ -45,6 +45,9 @@ func (c *criService) RemovePodSandbox(ctx context.Context, r *runtime.RemovePodS
 			r.GetPodSandboxId())
 		return &runtime.RemovePodSandboxResponse{}, nil
 	}
+
+	defer c.nri.BlockPluginSync().Unblock()
+
 	// Use the full sandbox id.
 	id := sandbox.ID
 
diff --git a/pkg/cri/server/sandbox_run.go b/pkg/cri/server/sandbox_run.go
@@ -463,6 +463,8 @@ func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandbox
 		sandboxCreateNetworkTimer.UpdateSince(netStart)
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	err = c.nri.RunPodSandbox(ctx, &sandbox)
 	if err != nil {
 		return nil, fmt.Errorf("NRI RunPodSandbox failed: %w", err)
diff --git a/pkg/cri/server/sandbox_stop.go b/pkg/cri/server/sandbox_stop.go
@@ -50,6 +50,8 @@ func (c *criService) StopPodSandbox(ctx context.Context, r *runtime.StopPodSandb
 		return &runtime.StopPodSandboxResponse{}, nil
 	}
 
+	defer c.nri.BlockPluginSync().Unblock()
+
 	if err := c.stopPodSandbox(ctx, sandbox); err != nil {
 		return nil, err
 	}
diff --git a/pkg/nri/nri.go b/pkg/nri/nri.go
@@ -82,6 +82,9 @@ type API interface {
 
 	// StopContainer relays container removal events to NRI.
 	RemoveContainer(context.Context, PodSandbox, Container) error
+
+	// BlockPluginSync blocks plugin synchronization until it is Unblock()ed.
+	BlockPluginSync() *PluginSyncBlock
 }
 
 type State int
@@ -436,6 +439,15 @@ func (l *local) RemoveContainer(ctx context.Context, pod PodSandbox, ctr Contain
 	return err
 }
 
+type PluginSyncBlock = nri.PluginSyncBlock
+
+func (l *local) BlockPluginSync() *PluginSyncBlock {
+	if !l.IsEnabled() {
+		return nil
+	}
+	return l.nri.BlockPluginSync()
+}
+
 func (l *local) syncPlugin(ctx context.Context, syncFn nri.SyncCB) error {
 	l.Lock()
 	defer l.Unlock()

Original file line number	Diff line number	Diff line change
`@@ -108,6 +108,14 @@ func (API) WithContainerExit(cstore.Container) containerd.ProcessDeleteOpts {`
`108`	`108`	`}`
`109`	`109`	`}`
`110`	`110`
	`111`	`+type PluginSyncBlock struct{}`
	`112`	`+`
	`113`	`+func (API) BlockPluginSync() PluginSyncBlock {`
	`114`	`+ return nil`
	`115`	`+}`
	`116`	`+`
	`117`	`+func (*PluginSyncBlock) Unblock() {}`
	`118`	`+`
`111`	`119`	`//`
`112`	`120`	`// NRI-CRI no-op 'domain' interface`
`113`	`121`	`//`
Original file line number	Diff line number	Diff line change
`@@ -285,6 +285,8 @@ func (c criService) CreateContainer(ctx context.Context, r runtime.CreateConta`
`285`	`285`	`}`
`286`	`286`	`}()`
`287`	`287`
	`288`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`289`	`+`
`288`	`290`	`var cntr containerd.Container`
`289`	`291`	`if cntr, err = c.client.NewContainer(ctx, id, opts...); err != nil {`
`290`	`292`	`return nil, fmt.Errorf("failed to create containerd container: %w", err)`
Original file line number	Diff line number	Diff line change
`@@ -146,6 +146,8 @@ func (c criService) StartContainer(ctx context.Context, r runtime.StartContain`
`146`	`146`	`return nil, fmt.Errorf("failed to wait for containerd task: %w", err)`
`147`	`147`	`}`
`148`	`148`
	`149`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`150`	`+`
`149`	`151`	`defer func() {`
`150`	`152`	`if retErr != nil {`
`151`	`153`	`deferCtx, deferCancel := ctrdutil.DeferContext()`
Original file line number	Diff line number	Diff line change
`@@ -50,6 +50,8 @@ func (c criService) StopContainer(ctx context.Context, r runtime.StopContainer`
`50`	`50`	`return &runtime.StopContainerResponse{}, nil`
`51`	`51`	`}`
`52`	`52`
	`53`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`54`	`+`
`53`	`55`	`if err := c.stopContainer(ctx, container, time.Duration(r.GetTimeout())*time.Second); err != nil {`
`54`	`56`	`return nil, err`
`55`	`57`	`}`
Original file line number	Diff line number	Diff line change
`@@ -47,6 +47,8 @@ func (c criService) UpdateContainerResources(ctx context.Context, r runtime.Up`
`47`	`47`	`return nil, err`
`48`	`48`	`}`
`49`	`49`
	`50`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`51`	`+`
`50`	`52`	`resources := r.GetLinux()`
`51`	`53`	`updated, err := c.nri.UpdateContainerResources(ctx, &sandbox, &container, resources)`
`52`	`54`	`if err != nil {`
Original file line number	Diff line number	Diff line change
`@@ -44,6 +44,9 @@ func (c criService) RemovePodSandbox(ctx context.Context, r runtime.RemovePodS`
`44`	`44`	`r.GetPodSandboxId())`
`45`	`45`	`return &runtime.RemovePodSandboxResponse{}, nil`
`46`	`46`	`}`
	`47`	`+`
	`48`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`49`	`+`
`47`	`50`	`// Use the full sandbox id.`
`48`	`51`	`id := sandbox.ID`
`49`	`52`
Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,8 @@ func (c criService) StopPodSandbox(ctx context.Context, r runtime.StopPodSandb`
`45`	`45`	`return &runtime.StopPodSandboxResponse{}, nil`
`46`	`46`	`}`
`47`	`47`
	`48`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`49`	`+`
`48`	`50`	`if err := c.stopPodSandbox(ctx, sandbox); err != nil {`
`49`	`51`	`return nil, err`
`50`	`52`	`}`
Original file line number	Diff line number	Diff line change
`@@ -259,6 +259,8 @@ func (c criService) CreateContainer(ctx context.Context, r runtime.CreateConta`
`259`	`259`	`}`
`260`	`260`	`}()`
`261`	`261`
	`262`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`263`	`+`
`262`	`264`	`var cntr containerd.Container`
`263`	`265`	`if cntr, err = c.client.NewContainer(ctx, id, opts...); err != nil {`
`264`	`266`	`return nil, fmt.Errorf("failed to create containerd container: %w", err)`
Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,9 @@ func (c criService) RemovePodSandbox(ctx context.Context, r runtime.RemovePodS`
`45`	`45`	`r.GetPodSandboxId())`
`46`	`46`	`return &runtime.RemovePodSandboxResponse{}, nil`
`47`	`47`	`}`
	`48`	`+`
	`49`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`50`	`+`
`48`	`51`	`// Use the full sandbox id.`
`49`	`52`	`id := sandbox.ID`
`50`	`53`
Original file line number	Diff line number	Diff line change
`@@ -463,6 +463,8 @@ func (c criService) RunPodSandbox(ctx context.Context, r runtime.RunPodSandbox`
`463`	`463`	`sandboxCreateNetworkTimer.UpdateSince(netStart)`
`464`	`464`	`}`
`465`	`465`
	`466`	`+ defer c.nri.BlockPluginSync().Unblock()`
	`467`	`+`
`466`	`468`	`err = c.nri.RunPodSandbox(ctx, &sandbox)`
`467`	`469`	`if err != nil {`
`468`	`470`	`return nil, fmt.Errorf("NRI RunPodSandbox failed: %w", err)`