go.{mod,sum}: update NRI deps and re-vendor.

Update NRI dependencies to point to the latest main/HEAD.

Signed-off-by: Krisztian Litkey <krisztian.litkey@intel.com>

vendor/github.com/containerd/nri/.codespellrc

@@ -0,0 +1,3 @@
+[codespell]
+skip = .git,*.pdf,*.svg,go.sum,go.mod
+ignore-words-list = clos

vendor/github.com/containerd/nri/.golangci.yml

@@ -36,6 +36,9 @@ issues:
       linters:
         - govet
       text: "copylocks: .*protobuf/internal/impl.MessageState.*"
+    # We dot-import ginkgo and gomega in some tests. Silence any related errors.
+    - path: 'pkg/adaptation|pkg/runtime-tools/generate|pkg/net/multiplex'
+      text: "dot-imports:"
 
 run:
   timeout: 2m

vendor/github.com/containerd/nri/Makefile

@@ -124,7 +124,7 @@ $(BIN_PATH)/template: $(wildcard plugins/template/*.go)
 
 test-gopkgs: ginkgo-tests test-ulimits
 
-SKIPPED_PKGS="ulimit-adjuster"
+SKIPPED_PKGS="ulimit-adjuster,device-injector"
 
 ginkgo-tests:
 	$(Q)$(GINKGO) run \
@@ -137,12 +137,15 @@ ginkgo-tests:
 	    --coverprofile coverprofile \
 	    --succinct \
 	    --skip-package $(SKIPPED_PKGS) \
-	    -r .; \
+	    -r && \
 	$(GO_CMD) tool cover -html=$(COVERAGE_PATH)/coverprofile -o $(COVERAGE_PATH)/coverage.html
 
 test-ulimits:
 	$(Q)cd ./plugins/ulimit-adjuster && $(GO_TEST) -v
 
+test-device-injector:
+	$(Q)cd ./plugins/device-injector && $(GO_TEST) -v
+
 codecov: SHELL := $(shell which bash)
 codecov:
 	bash <(curl -s https://codecov.io/bash) -f $(COVERAGE_PATH)/coverprofile
@@ -163,6 +166,13 @@ vet:
 golangci-lint:
 	$(Q)$(GOLANG_CILINT) run
 
+validate-repo-no-changes:
+	$(Q)test -z "$$(git status --short | tee /dev/stderr)" || { \
+		echo "Repository has changes."; \
+		echo "Please make sure to commit all changes, including generated files."; \
+		exit 1; \
+	}
+
 #
 # proto generation targets
 #

vendor/github.com/containerd/nri/README.md

@@ -297,7 +297,7 @@ apply NRI responses to containers.
 
 The plugin stub hides many of the low-level details of implementing an NRI
 plugin. It takes care of connection establishment, plugin registration,
-configuration, and event subscription. All [sample plugins](pkg/plugins)
+configuration, and event subscription. All [sample plugins](plugins)
 are implemented using the stub. Any of these can be used as a tutorial on
 how the stub library should be used.
 
@@ -308,6 +308,7 @@ The following sample plugins exist for NRI:
   - [logger](plugins/logger)
   - [differ](plugins/differ)
   - [device injector](plugins/device-injector)
+  - [network device injector](plugins/network-device-injector)
   - [OCI hook injector](plugins/hook-injector)
   - [ulimit adjuster](plugins/ulimit-adjuster)
   - [NRI v0.1.0 plugin adapter](plugins/v010-adapter)

vendor/github.com/containerd/nri/pkg/adaptation/adaptation.go

@@ -65,6 +65,7 @@ type Adaptation struct {
 	serverOpts []ttrpc.ServerOpt
 	listener   net.Listener
 	plugins    []*plugin
+	syncLock   sync.RWMutex
 }
 
 var (
@@ -135,6 +136,7 @@ func New(name, version string, syncFn SyncFn, updateFn UpdateFn, opts ...Option)
 		pluginPath: DefaultPluginPath,
 		dropinPath: DefaultPluginConfigPath,
 		socketPath: DefaultSocketPath,
+		syncLock:   sync.RWMutex{},
 	}
 
 	for _, o := range opts {
@@ -336,25 +338,48 @@ func (r *Adaptation) startPlugins() (retErr error) {
 	}()
 
 	for i, name := range names {
-		log.Infof(noCtx, "starting plugin %q...", name)
+		log.Infof(noCtx, "starting pre-installed NRI plugin %q...", name)
 
 		id := ids[i]
-
 		p, err := r.newLaunchedPlugin(r.pluginPath, id, name, configs[i])
 		if err != nil {
-			return fmt.Errorf("failed to start NRI plugin %q: %w", name, err)
+			log.Warnf(noCtx, "failed to initialize pre-installed NRI plugin %q: %v", name, err)
+			continue
 		}
 
 		if err := p.start(r.name, r.version); err != nil {
-			return err
+			log.Warnf(noCtx, "failed to start pre-installed NRI plugin %q: %v", name, err)
+			continue
 		}
 
 		plugins = append(plugins, p)
 	}
 
+	// Although the error returned by syncPlugins may not be nil, r.syncFn could still ignores this error and returns a nil error.
+	// We need to make sure that the plugins are successfully synchronized in the `plugins`
+	syncPlugins := func(ctx context.Context, pods []*PodSandbox, containers []*Container) (updates []*ContainerUpdate, err error) {
+		startedPlugins := plugins
+		plugins = make([]*plugin, 0, len(plugins))
+		for _, plugin := range startedPlugins {
+			us, err := plugin.synchronize(ctx, pods, containers)
+			if err != nil {
+				plugin.stop()
+				log.Warnf(noCtx, "failed to synchronize pre-installed NRI plugin %q: %v", plugin.name(), err)
+				continue
+			}
+
+			plugins = append(plugins, plugin)
+			updates = append(updates, us...)
+			log.Infof(noCtx, "pre-installed NRI plugin %q synchronization success", plugin.name())
+		}
+		return updates, nil
+	}
+	if err := r.syncFn(noCtx, syncPlugins); err != nil {
+		return fmt.Errorf("failed to synchronize pre-installed NRI Plugins: %w", err)
+	}
+
 	r.plugins = plugins
 	r.sortPlugins()
-
 	return nil
 }
 
@@ -369,12 +394,22 @@ func (r *Adaptation) stopPlugins() {
 }
 
 func (r *Adaptation) removeClosedPlugins() {
-	active := []*plugin{}
+	var active, closed []*plugin
 	for _, p := range r.plugins {
-		if !p.isClosed() {
+		if p.isClosed() {
+			closed = append(closed, p)
+		} else {
 			active = append(active, p)
 		}
 	}
+
+	if len(closed) != 0 {
+		go func() {
+			for _, plugin := range closed {
+				plugin.stop()
+			}
+		}()
+	}
 	r.plugins = active
 }
 
@@ -431,6 +466,8 @@ func (r *Adaptation) acceptPluginConnections(l net.Listener) error {
 				continue
 			}
 
+			r.requestPluginSync()
+
 			err = r.syncFn(ctx, p.synchronize)
 			if err != nil {
 				log.Infof(ctx, "failed to synchronize plugin: %v", err)
@@ -439,9 +476,10 @@ func (r *Adaptation) acceptPluginConnections(l net.Listener) error {
 				r.plugins = append(r.plugins, p)
 				r.sortPlugins()
 				r.Unlock()
+				log.Infof(ctx, "plugin %q connected and synchronized", p.name())
 			}
 
-			log.Infof(ctx, "plugin %q connected", p.name())
+			r.finishedPluginSync()
 		}
 	}()
 
@@ -512,3 +550,30 @@ func (r *Adaptation) sortPlugins() {
 		}
 	}
 }
+
+func (r *Adaptation) requestPluginSync() {
+	r.syncLock.Lock()
+}
+
+func (r *Adaptation) finishedPluginSync() {
+	r.syncLock.Unlock()
+}
+
+type PluginSyncBlock struct {
+	r *Adaptation
+}
+
+// BlockPluginSync blocks plugins from being synchronized/fully registered.
+func (r *Adaptation) BlockPluginSync() *PluginSyncBlock {
+	r.syncLock.RLock()
+	return &PluginSyncBlock{r: r}
+}
+
+// Unblock a plugin sync. block put in place by BlockPluginSync. Safe to call
+// multiple times but only from a single goroutine.
+func (b *PluginSyncBlock) Unblock() {
+	if b != nil && b.r != nil {
+		b.r.syncLock.RUnlock()
+		b.r = nil
+	}
+}

vendor/github.com/containerd/nri/pkg/adaptation/api.go

@@ -78,6 +78,7 @@ type (
 	LinuxMemory              = api.LinuxMemory
 	LinuxDevice              = api.LinuxDevice
 	LinuxDeviceCgroup        = api.LinuxDeviceCgroup
+	CDIDevice                = api.CDIDevice
 	HugepageLimit            = api.HugepageLimit
 	Hooks                    = api.Hooks
 	Hook                     = api.Hook

vendor/github.com/containerd/nri/pkg/adaptation/plugin.go

@@ -33,13 +33,15 @@ import (
 	"github.com/containerd/nri/pkg/net"
 	"github.com/containerd/nri/pkg/net/multiplex"
 	"github.com/containerd/ttrpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 const (
 	// DefaultPluginRegistrationTimeout is the default timeout for plugin registration.
-	DefaultPluginRegistrationTimeout = 5 * time.Second
+	DefaultPluginRegistrationTimeout = api.DefaultPluginRegistrationTimeout
 	// DefaultPluginRequestTimeout is the default timeout for plugins to handle a request.
-	DefaultPluginRequestTimeout = 2 * time.Second
+	DefaultPluginRequestTimeout = api.DefaultPluginRequestTimeout
 )
 
 var (
@@ -384,9 +386,11 @@ func (p *plugin) configure(ctx context.Context, name, version, config string) er
 	defer cancel()
 
 	rpl, err := p.stub.Configure(ctx, &ConfigureRequest{
-		Config:         config,
-		RuntimeName:    name,
-		RuntimeVersion: version,
+		Config:              config,
+		RuntimeName:         name,
+		RuntimeVersion:      version,
+		RegistrationTimeout: getPluginRegistrationTimeout().Milliseconds(),
+		RequestTimeout:      getPluginRequestTimeout().Milliseconds(),
 	})
 	if err != nil {
 		return fmt.Errorf("failed to configure plugin: %w", err)
@@ -412,18 +416,101 @@ func (p *plugin) synchronize(ctx context.Context, pods []*PodSandbox, containers
 	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
-	req := &SynchronizeRequest{
-		Pods:       pods,
-		Containers: containers,
-	}
-	rpl, err := p.stub.Synchronize(ctx, req)
-	if err != nil {
-		return nil, err
+	var (
+		podsToSend = pods
+		ctrsToSend = containers
+		podsPerMsg = len(pods)
+		ctrsPerMsg = len(containers)
+
+		rpl *SynchronizeResponse
+		err error
+	)
+
+	for {
+		req := &SynchronizeRequest{
+			Pods:       podsToSend[:podsPerMsg],
+			Containers: ctrsToSend[:ctrsPerMsg],
+			More:       len(podsToSend) > podsPerMsg || len(ctrsToSend) > ctrsPerMsg,
+		}
+
+		log.Debugf(ctx, "sending sync message, %d/%d, %d/%d (more: %v)",
+			len(req.Pods), len(podsToSend), len(req.Containers), len(ctrsToSend), req.More)
+
+		rpl, err = p.stub.Synchronize(ctx, req)
+		if err == nil {
+			if !req.More {
+				break
+			}
+
+			if len(rpl.Update) > 0 || rpl.More != req.More {
+				p.close()
+				return nil, fmt.Errorf("plugin does not handle split sync requests")
+			}
+
+			podsToSend = podsToSend[podsPerMsg:]
+			ctrsToSend = ctrsToSend[ctrsPerMsg:]
+
+			if podsPerMsg > len(podsToSend) {
+				podsPerMsg = len(podsToSend)
+			}
+			if ctrsPerMsg > len(ctrsToSend) {
+				ctrsPerMsg = len(ctrsToSend)
+			}
+		} else {
+			podsPerMsg, ctrsPerMsg, err = recalcObjsPerSyncMsg(podsPerMsg, ctrsPerMsg, err)
+			if err != nil {
+				p.close()
+				return nil, err
+			}
+
+			log.Debugf(ctx, "oversized message, retrying in smaller chunks")
+		}
 	}
 
 	return rpl.Update, nil
 }
 
+func recalcObjsPerSyncMsg(pods, ctrs int, err error) (int, int, error) {
+	const (
+		minObjsPerMsg = 8
+	)
+
+	if status.Code(err) != codes.ResourceExhausted {
+		return pods, ctrs, err
+	}
+
+	if pods+ctrs <= minObjsPerMsg {
+		return pods, ctrs, fmt.Errorf("failed to synchronize plugin with split messages")
+	}
+
+	var e *ttrpc.OversizedMessageErr
+	if !errors.As(err, &e) {
+		return pods, ctrs, fmt.Errorf("failed to synchronize plugin with split messages")
+	}
+
+	maxLen := e.MaximumLength()
+	msgLen := e.RejectedLength()
+
+	if msgLen == 0 || maxLen == 0 || msgLen <= maxLen {
+		return pods, ctrs, fmt.Errorf("failed to synchronize plugin with split messages")
+	}
+
+	factor := float64(maxLen) / float64(msgLen)
+	if factor > 0.9 {
+		factor = 0.9
+	}
+
+	pods = int(float64(pods) * factor)
+	ctrs = int(float64(ctrs) * factor)
+
+	if pods+ctrs < minObjsPerMsg {
+		pods = minObjsPerMsg / 2
+		ctrs = minObjsPerMsg / 2
+	}
+
+	return pods, ctrs, nil
+}
+
 // Relay CreateContainer request to plugin.
 func (p *plugin) createContainer(ctx context.Context, req *CreateContainerRequest) (*CreateContainerResponse, error) {
 	if !p.events.IsSet(Event_CREATE_CONTAINER) {
@@ -516,7 +603,7 @@ func (p *plugin) StateChange(ctx context.Context, evt *StateChangeEvent) error {
 	return nil
 }
 
-// isFatalError returns true if the error is fatal and the plugin connection shoudld be closed.
+// isFatalError returns true if the error is fatal and the plugin connection should be closed.
 func isFatalError(err error) bool {
 	switch {
 	case errors.Is(err, ttrpc.ErrClosed):

vendor/github.com/containerd/nri/pkg/adaptation/result.go

@@ -19,6 +19,8 @@ package adaptation
 import (
 	"fmt"
 	"strings"
+
+	"github.com/containerd/nri/pkg/api"
 )
 
 type result struct {
@@ -89,6 +91,7 @@ func collectCreateContainerResult(request *CreateContainerRequest) *result {
 				Env:         []*KeyValue{},
 				Hooks:       &Hooks{},
 				Rlimits:     []*POSIXRlimit{},
+				CDIDevices:  []*CDIDevice{},
 				Linux: &LinuxContainerAdjustment{
 					Devices: []*LinuxDevice{},
 					Resources: &LinuxResources{
@@ -200,7 +203,7 @@ func (r *result) adjust(rpl *ContainerAdjustment, plugin string) error {
 	if err := r.adjustEnv(rpl.Env, plugin); err != nil {
 		return err
 	}
-	if err := r.adjustHooks(rpl.Hooks, plugin); err != nil {
+	if err := r.adjustHooks(rpl.Hooks); err != nil {
 		return err
 	}
 	if rpl.Linux != nil {
@@ -213,10 +216,16 @@ func (r *result) adjust(rpl *ContainerAdjustment, plugin string) error {
 		if err := r.adjustCgroupsPath(rpl.Linux.CgroupsPath, plugin); err != nil {
 			return err
 		}
+		if err := r.adjustOomScoreAdj(rpl.Linux.OomScoreAdj, plugin); err != nil {
+			return err
+		}
 	}
 	if err := r.adjustRlimits(rpl.Rlimits, plugin); err != nil {
 		return err
 	}
+	if err := r.adjustCDIDevices(rpl.CDIDevices, plugin); err != nil {
+		return err
+	}
 
 	return nil
 }
@@ -322,6 +331,13 @@ func (r *result) adjustMounts(mounts []*Mount, plugin string) error {
 		r.reply.adjust.Mounts = append(r.reply.adjust.Mounts, m)
 	}
 
+	// next, apply deletions with no corresponding additions
+	for _, m := range del {
+		if _, ok := mod[api.ClearRemovalMarker(m.Destination)]; !ok {
+			r.reply.adjust.Mounts = append(r.reply.adjust.Mounts, m)
+		}
+	}
+
 	// finally, apply additions/modifications to plugin container creation request
 	create.Container.Mounts = append(create.Container.Mounts, add...)
 
@@ -386,6 +402,36 @@ func (r *result) adjustDevices(devices []*LinuxDevice, plugin string) error {
 	return nil
 }
 
+func (r *result) adjustCDIDevices(devices []*CDIDevice, plugin string) error {
+	if len(devices) == 0 {
+		return nil
+	}
+
+	// Notes:
+	//   CDI devices are opaque references, typically to vendor specific
+	//   devices. They get resolved to actual devices and potential related
+	//   mounts, environment variables, etc. in the runtime. Unlike with
+	//   devices, we only allow CDI device references to be added to a
+	//   container, not removed. We pass them unresolved to the runtime and
+	//   have them resolved there. Also unlike with devices, we don't include
+	//   CDI device references in creation requests. However, since there
+	//   is typically a strong ownership and a single related management entity
+	//   per vendor/class for these devices we do treat multiple injection of
+	//   the same CDI device reference as an error here.
+
+	id := r.request.create.Container.Id
+
+	// apply additions to collected adjustments
+	for _, d := range devices {
+		if err := r.owners.claimCDIDevice(id, d.Name, plugin); err != nil {
+			return err
+		}
+		r.reply.adjust.CDIDevices = append(r.reply.adjust.CDIDevices, d)
+	}
+
+	return nil
+}
+
 func (r *result) adjustEnv(env []*KeyValue, plugin string) error {
 	if len(env) == 0 {
 		return nil
@@ -458,7 +504,7 @@ func splitEnvVar(s string) (string, string) {
 	return split[0], split[1]
 }
 
-func (r *result) adjustHooks(hooks *Hooks, plugin string) error {
+func (r *result) adjustHooks(hooks *Hooks) error {
 	if hooks == nil {
 		return nil
 	}
@@ -645,7 +691,16 @@ func (r *result) adjustResources(resources *LinuxResources, plugin string) error
 		container.RdtClass = String(v.GetValue())
 		reply.RdtClass = String(v.GetValue())
 	}
-
+	if v := resources.GetPids(); v != nil {
+		if err := r.owners.claimPidsLimit(id, plugin); err != nil {
+			return err
+		}
+		pidv := &api.LinuxPids{
+			Limit: v.GetLimit(),
+		}
+		container.Pids = pidv
+		reply.Pids = pidv
+	}
 	return nil
 }
 
@@ -666,6 +721,23 @@ func (r *result) adjustCgroupsPath(path, plugin string) error {
 	return nil
 }
 
+func (r *result) adjustOomScoreAdj(OomScoreAdj *OptionalInt, plugin string) error {
+	if OomScoreAdj == nil {
+		return nil
+	}
+
+	create, id := r.request.create, r.request.create.Container.Id
+
+	if err := r.owners.claimOomScoreAdj(id, plugin); err != nil {
+		return err
+	}
+
+	create.Container.Linux.OomScoreAdj = OomScoreAdj
+	r.reply.adjust.Linux.OomScoreAdj = OomScoreAdj
+
+	return nil
+}
+
 func (r *result) adjustRlimits(rlimits []*POSIXRlimit, plugin string) error {
 	create, id, adjust := r.request.create, r.request.create.Container.Id, r.reply.adjust
 	for _, l := range rlimits {
@@ -820,6 +892,14 @@ func (r *result) updateResources(reply, u *ContainerUpdate, plugin string) error
 		}
 		resources.RdtClass = String(v.GetValue())
 	}
+	if v := resources.GetPids(); v != nil {
+		if err := r.owners.claimPidsLimit(id, plugin); err != nil {
+			return err
+		}
+		resources.Pids = &api.LinuxPids{
+			Limit: v.GetLimit(),
+		}
+	}
 
 	// update request/reply from copy on success
 	reply.Linux.Resources = resources.Copy()
@@ -872,6 +952,7 @@ type owners struct {
 	annotations         map[string]string
 	mounts              map[string]string
 	devices             map[string]string
+	cdiDevices          map[string]string
 	env                 map[string]string
 	memLimit            string
 	memReservation      string
@@ -888,11 +969,13 @@ type owners struct {
 	cpuRealtimePeriod   string
 	cpusetCpus          string
 	cpusetMems          string
+	pidsLimit           string
 	hugepageLimits      map[string]string
 	blockioClass        string
 	rdtClass            string
 	unified             map[string]string
 	cgroupsPath         string
+	oomScoreAdj         string
 	rlimits             map[string]string
 }
 
@@ -917,6 +1000,10 @@ func (ro resultOwners) claimDevice(id, path, plugin string) error {
 	return ro.ownersFor(id).claimDevice(path, plugin)
 }
 
+func (ro resultOwners) claimCDIDevice(id, path, plugin string) error {
+	return ro.ownersFor(id).claimCDIDevice(path, plugin)
+}
+
 func (ro resultOwners) claimEnv(id, name, plugin string) error {
 	return ro.ownersFor(id).claimEnv(name, plugin)
 }
@@ -981,6 +1068,10 @@ func (ro resultOwners) claimCpusetMems(id, plugin string) error {
 	return ro.ownersFor(id).claimCpusetMems(plugin)
 }
 
+func (ro resultOwners) claimPidsLimit(id, plugin string) error {
+	return ro.ownersFor(id).claimPidsLimit(plugin)
+}
+
 func (ro resultOwners) claimHugepageLimit(id, size, plugin string) error {
 	return ro.ownersFor(id).claimHugepageLimit(size, plugin)
 }
@@ -1001,6 +1092,10 @@ func (ro resultOwners) claimCgroupsPath(id, plugin string) error {
 	return ro.ownersFor(id).claimCgroupsPath(plugin)
 }
 
+func (ro resultOwners) claimOomScoreAdj(id, plugin string) error {
+	return ro.ownersFor(id).claimOomScoreAdj(plugin)
+}
+
 func (ro resultOwners) claimRlimits(id, typ, plugin string) error {
 	return ro.ownersFor(id).claimRlimit(typ, plugin)
 }
@@ -1038,6 +1133,17 @@ func (o *owners) claimDevice(path, plugin string) error {
 	return nil
 }
 
+func (o *owners) claimCDIDevice(name, plugin string) error {
+	if o.cdiDevices == nil {
+		o.cdiDevices = make(map[string]string)
+	}
+	if other, taken := o.cdiDevices[name]; taken {
+		return conflict(plugin, other, "CDI device", name)
+	}
+	o.cdiDevices[name] = plugin
+	return nil
+}
+
 func (o *owners) claimEnv(name, plugin string) error {
 	if o.env == nil {
 		o.env = make(map[string]string)
@@ -1169,6 +1275,14 @@ func (o *owners) claimCpusetMems(plugin string) error {
 	return nil
 }
 
+func (o *owners) claimPidsLimit(plugin string) error {
+	if other := o.pidsLimit; other != "" {
+		return conflict(plugin, other, "pids pinning")
+	}
+	o.pidsLimit = plugin
+	return nil
+}
+
 func (o *owners) claimHugepageLimit(size, plugin string) error {
 	if o.hugepageLimits == nil {
 		o.hugepageLimits = make(map[string]string)
@@ -1227,6 +1341,14 @@ func (o *owners) claimCgroupsPath(plugin string) error {
 	return nil
 }
 
+func (o *owners) claimOomScoreAdj(plugin string) error {
+	if other := o.oomScoreAdj; other != "" {
+		return conflict(plugin, other, "oom score adj")
+	}
+	o.oomScoreAdj = plugin
+	return nil
+}
+
 func (ro resultOwners) clearAnnotation(id, key string) {
 	ro.ownersFor(id).clearAnnotation(key)
 }

vendor/github.com/containerd/nri/pkg/api/adjustment.go

@@ -129,6 +129,11 @@ func (a *ContainerAdjustment) RemoveDevice(path string) {
 	})
 }
 
+// AddCDIDevice records the addition of the given CDI device to a container.
+func (a *ContainerAdjustment) AddCDIDevice(d *CDIDevice) {
+	a.CDIDevices = append(a.CDIDevices, d) // TODO: should we dup d here ?
+}
+
 // SetLinuxMemoryLimit records setting the memory limit for a container.
 func (a *ContainerAdjustment) SetLinuxMemoryLimit(value int64) {
 	a.initLinuxResourcesMemory()
@@ -219,6 +224,12 @@ func (a *ContainerAdjustment) SetLinuxCPUSetMems(value string) {
 	a.Linux.Resources.Cpu.Mems = value
 }
 
+// SetLinuxPidLimits records setting the pid max number for a container.
+func (a *ContainerAdjustment) SetLinuxPidLimits(value int64) {
+	a.initLinuxResourcesPids()
+	a.Linux.Resources.Pids.Limit = value
+}
+
 // AddLinuxHugepageLimit records adding a hugepage limit for a container.
 func (a *ContainerAdjustment) AddLinuxHugepageLimit(pageSize string, value uint64) {
 	a.initLinuxResources()
@@ -253,6 +264,12 @@ func (a *ContainerAdjustment) SetLinuxCgroupsPath(value string) {
 	a.Linux.CgroupsPath = value
 }
 
+// SetLinuxOomScoreAdj records setting the kernel's Out-Of-Memory (OOM) killer score for a container.
+func (a *ContainerAdjustment) SetLinuxOomScoreAdj(value *int) {
+	a.initLinux()
+	a.Linux.OomScoreAdj = Int(value) // using Int(value) from ./options.go to optionally allocate a pointer to normalized copy of value
+}
+
 //
 // Initializing a container adjustment and container update.
 //
@@ -302,6 +319,13 @@ func (a *ContainerAdjustment) initLinuxResourcesCPU() {
 	}
 }
 
+func (a *ContainerAdjustment) initLinuxResourcesPids() {
+	a.initLinuxResources()
+	if a.Linux.Resources.Pids == nil {
+		a.Linux.Resources.Pids = &LinuxPids{}
+	}
+}
+
 func (a *ContainerAdjustment) initLinuxResourcesUnified() {
 	a.initLinuxResources()
 	if a.Linux.Resources.Unified == nil {

vendor/github.com/containerd/nri/pkg/api/api.proto

@@ -46,7 +46,7 @@ message UpdateContainersRequest {
 }
 
 message UpdateContainersResponse {
-  // Containers that the runtime failed to udpate.
+  // Containers that the runtime failed to update.
   repeated ContainerUpdate failed = 1;
 }
 
@@ -114,6 +114,10 @@ message ConfigureRequest {
   string runtime_name = 2;
   // Version of the runtime NRI is running in.
   string runtime_version = 3;
+  // Configured registration timeout in milliseconds.
+  int64 registration_timeout = 4;
+  // Configured request processing timeout in milliseconds.
+  int64 request_timeout = 5;
 }
 
 message ConfigureResponse {
@@ -127,11 +131,15 @@ message SynchronizeRequest {
   repeated PodSandbox pods = 1;
   // Containers known to the runtime.
   repeated Container containers = 2;
+  // Whether there are more pods and containers to follow.
+  bool more = 3;
 }
 
 message SynchronizeResponse {
   // Updates to containers requested by the plugin.
   repeated ContainerUpdate update = 1;
+  // Whether the client is able to handle more advertised pods and containers.
+  bool more = 2;
 }
 
 message CreateContainerRequest {
@@ -319,6 +327,11 @@ message LinuxDeviceCgroup {
   string access = 5;
 }
 
+// A CDI device reference.
+message CDIDevice {
+  string name = 1;
+}
+
 // Container (linux) resources.
 message LinuxResources {
   LinuxMemory memory = 1;
@@ -328,6 +341,7 @@ message LinuxResources {
   OptionalString rdt_class = 5;
   map<string, string> unified = 6;
   repeated LinuxDeviceCgroup devices = 7; // for NRI v1 emulation
+  LinuxPids pids = 8;
 }
 
 // Memory-related parts of (linux) resources.
@@ -366,6 +380,11 @@ message POSIXRlimit {
   uint64 soft = 3;
 }
 
+// Pids-related parts of (linux) resources.
+message LinuxPids {
+  int64 limit = 1;
+}
+
 // Requested adjustments to a container being created.
 message ContainerAdjustment {
   map<string, string> annotations = 2;
@@ -374,6 +393,7 @@ message ContainerAdjustment {
   Hooks hooks = 5;
   LinuxContainerAdjustment linux = 6;
   repeated POSIXRlimit rlimits = 7;
+  repeated CDIDevice CDI_devices = 8;
 }
 
 // Adjustments to (linux) resources.
@@ -381,6 +401,7 @@ message LinuxContainerAdjustment {
   repeated LinuxDevice devices = 1;
   LinuxResources resources = 2;
   string cgroups_path = 3;
+  OptionalInt oom_score_adj = 4;
 }
 
 // Requested update to an already created container.

vendor/github.com/containerd/nri/pkg/api/helpers.go

@@ -57,3 +57,14 @@ func IsMarkedForRemoval(key string) (string, bool) {
 func MarkForRemoval(key string) string {
 	return "-" + key
 }
+
+// ClearRemovalMarker returns a key cleared from any removal marker.
+func ClearRemovalMarker(key string) string {
+	if key == "" {
+		return ""
+	}
+	if key[0] == '-' {
+		return key[1:]
+	}
+	return key
+}

vendor/github.com/containerd/nri/pkg/api/resources.go

@@ -22,7 +22,7 @@ import (
 )
 
 // FromOCILinuxResources returns resources from an OCI runtime Spec.
-func FromOCILinuxResources(o *rspec.LinuxResources, ann map[string]string) *LinuxResources {
+func FromOCILinuxResources(o *rspec.LinuxResources, _ map[string]string) *LinuxResources {
 	if o == nil {
 		return nil
 	}
@@ -65,6 +65,11 @@ func FromOCILinuxResources(o *rspec.LinuxResources, ann map[string]string) *Linu
 			Access: d.Access,
 		})
 	}
+	if p := o.Pids; p != nil {
+		l.Pids = &LinuxPids{
+			Limit: p.Limit,
+		}
+	}
 	return l
 }
 
@@ -148,7 +153,11 @@ func (r *LinuxResources) ToOCI() *rspec.LinuxResources {
 			Access: d.Access,
 		})
 	}
-
+	if r.Pids != nil {
+		o.Pids = &rspec.LinuxPids{
+			Limit: r.Pids.Limit,
+		}
+	}
 	return o
 }
 
@@ -226,6 +235,11 @@ func (r *LinuxResources) Copy() *LinuxResources {
 			o.Unified[k] = v
 		}
 	}
+	if r.Pids != nil {
+		o.Pids = &LinuxPids{
+			Limit: r.Pids.Limit,
+		}
+	}
 	o.BlockioClass = String(r.BlockioClass)
 	o.RdtClass = String(r.RdtClass)
 

vendor/github.com/containerd/nri/pkg/api/timeouts.go

@@ -0,0 +1,28 @@
+/*
+   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 api
+
+import (
+	"time"
+)
+
+const (
+	// DefaultPluginRegistrationTimeout is the default timeout for plugin registration.
+	DefaultPluginRegistrationTimeout = 5 * time.Second
+	// DefaultPluginRequestTimeout is the default timeout for plugins to handle a request.
+	DefaultPluginRequestTimeout = 2 * time.Second
+)

vendor/github.com/containerd/nri/pkg/api/update.go

@@ -112,6 +112,12 @@ func (u *ContainerUpdate) SetLinuxCPUSetMems(value string) {
 	u.Linux.Resources.Cpu.Mems = value
 }
 
+// SetLinuxPidLimits records setting the pid max number for a container.
+func (u *ContainerUpdate) SetLinuxPidLimits(value int64) {
+	u.initLinuxResourcesPids()
+	u.Linux.Resources.Pids.Limit = value
+}
+
 // AddLinuxHugepageLimit records adding a hugepage limit for a container.
 func (u *ContainerUpdate) AddLinuxHugepageLimit(pageSize string, value uint64) {
 	u.initLinuxResources()
@@ -184,3 +190,10 @@ func (u *ContainerUpdate) initLinuxResourcesUnified() {
 		u.Linux.Resources.Unified = make(map[string]string)
 	}
 }
+
+func (u *ContainerUpdate) initLinuxResourcesPids() {
+	u.initLinuxResources()
+	if u.Linux.Resources.Pids == nil {
+		u.Linux.Resources.Pids = &LinuxPids{}
+	}
+}

vendor/github.com/containerd/nri/pkg/net/multiplex/mux.go

@@ -139,7 +139,7 @@ const (
 	// length of frame header: 4-byte ConnID, 4-byte payload length
 	headerLen = 8
 	// max. allowed payload size
-	maxPayloadSize = 1 << 24
+	maxPayloadSize = ttrpcMessageHeaderLength + ttrpcMessageLengthMax
 )
 
 // conn represents a single multiplexed connection.
@@ -234,38 +234,54 @@ func (m *mux) Listen(id ConnID) (net.Listener, error) {
 }
 
 func (m *mux) write(id ConnID, buf []byte) (int, error) {
-	var hdr [headerLen]byte
-
-	if len(buf) > maxPayloadSize {
-		return 0, syscall.EMSGSIZE
-	}
-
-	binary.BigEndian.PutUint32(hdr[0:4], uint32(id))
-	binary.BigEndian.PutUint32(hdr[4:8], uint32(len(buf)))
+	var (
+		hdr  [headerLen]byte
+		data = buf[:]
+		size = len(data)
+	)
 
 	m.writeLock.Lock()
 	defer m.writeLock.Unlock()
 
-	n, err := m.trunk.Write(hdr[:])
-	if err != nil {
-		err = fmt.Errorf("failed to write header to trunk: %w", err)
-		if n != 0 {
-			m.setError(err)
-			m.Close()
+	for {
+		if size > maxPayloadSize {
+			size = maxPayloadSize
 		}
-		return 0, err
-	}
 
-	n, err = m.trunk.Write(buf)
-	if err != nil {
-		err = fmt.Errorf("failed to write payload to trunk: %w", err)
-		if n != 0 {
-			m.setError(err)
-			m.Close()
+		binary.BigEndian.PutUint32(hdr[0:4], uint32(id))
+		binary.BigEndian.PutUint32(hdr[4:8], uint32(size))
+
+		n, err := m.trunk.Write(hdr[:])
+		if err != nil {
+			err = fmt.Errorf("failed to write header to trunk: %w", err)
+			if n != 0 {
+				m.setError(err)
+				m.Close()
+			}
+			return 0, err
+		}
+
+		n, err = m.trunk.Write(data[:size])
+		if err != nil {
+			err = fmt.Errorf("failed to write payload to trunk: %w", err)
+			if n != 0 {
+				m.setError(err)
+				m.Close()
+			}
+			return 0, err
+		}
+
+		data = data[size:]
+		if size > len(data) {
+			size = len(data)
+		}
+
+		if size == 0 {
+			break
 		}
 	}
 
-	return n, err
+	return len(buf), nil
 }
 
 func (m *mux) reader() {
@@ -429,16 +445,16 @@ func (c *conn) RemoteAddr() net.Addr {
 }
 
 // SetDeadline is the unimplemented stub for the corresponding net.Conn function.
-func (c *conn) SetDeadline(t time.Time) error {
+func (c *conn) SetDeadline(_ time.Time) error {
 	return nil
 }
 
 // SetReadDeadline is the unimplemented stub for the corresponding net.Conn function.
-func (c *conn) SetReadDeadline(t time.Time) error {
+func (c *conn) SetReadDeadline(_ time.Time) error {
 	return nil
 }
 
 // SetWriteDeadline is the unimplemented stub for the corresponding net.Conn function.
-func (c *conn) SetWriteDeadline(t time.Time) error {
+func (c *conn) SetWriteDeadline(_ time.Time) error {
 	return nil
 }

vendor/github.com/containerd/nri/pkg/net/multiplex/ttrpc.go

@@ -22,3 +22,8 @@ const (
 	// RuntimeServiceConn is the mux connection ID for NRI runtime services.
 	RuntimeServiceConn
 )
+
+const (
+	ttrpcMessageHeaderLength = 10
+	ttrpcMessageLengthMax    = 4 << 20
+)

vendor/github.com/containerd/nri/pkg/runtime-tools/generate/generate.go

@@ -40,6 +40,7 @@ type Generator struct {
 	filterAnnotations func(map[string]string) (map[string]string, error)
 	resolveBlockIO    func(string) (*rspec.LinuxBlockIO, error)
 	resolveRdt        func(string) (*rspec.LinuxIntelRdt, error)
+	injectCDIDevices  func(*rspec.Spec, []string) error
 	checkResources    func(*rspec.LinuxResources) error
 }
 
@@ -91,6 +92,14 @@ func WithResourceChecker(fn func(*rspec.LinuxResources) error) GeneratorOption {
 	}
 }
 
+// WithCDIDeviceInjector specifies a runtime-specific function to use for CDI
+// device resolution and injection into an OCI Spec.
+func WithCDIDeviceInjector(fn func(*rspec.Spec, []string) error) GeneratorOption {
+	return func(g *Generator) {
+		g.injectCDIDevices = fn
+	}
+}
+
 // Adjust adjusts all aspects of the OCI Spec that NRI knows/cares about.
 func (g *Generator) Adjust(adjust *nri.ContainerAdjustment) error {
 	if adjust == nil {
@@ -102,8 +111,12 @@ func (g *Generator) Adjust(adjust *nri.ContainerAdjustment) error {
 	}
 	g.AdjustEnv(adjust.GetEnv())
 	g.AdjustHooks(adjust.GetHooks())
+	if err := g.InjectCDIDevices(adjust.GetCDIDevices()); err != nil {
+		return err
+	}
 	g.AdjustDevices(adjust.GetLinux().GetDevices())
 	g.AdjustCgroupsPath(adjust.GetLinux().GetCgroupsPath())
+	g.AdjustOomScoreAdj(adjust.GetLinux().GetOomScoreAdj())
 
 	resources := adjust.GetLinux().GetResources()
 	if err := g.AdjustResources(resources); err != nil {
@@ -252,7 +265,9 @@ func (g *Generator) AdjustResources(r *nri.LinuxResources) error {
 	for k, v := range r.Unified {
 		g.AddLinuxResourcesUnified(k, v)
 	}
-
+	if v := r.GetPids(); v != nil {
+		g.SetLinuxResourcesPidsLimit(v.GetLimit())
+	}
 	if g.checkResources != nil {
 		if err := g.checkResources(g.Config.Linux.Resources); err != nil {
 			return fmt.Errorf("failed to adjust resources in OCI Spec: %w", err)
@@ -309,6 +324,17 @@ func (g *Generator) AdjustCgroupsPath(path string) {
 	}
 }
 
+// AdjustOomScoreAdj adjusts the kernel's Out-Of-Memory (OOM) killer score for the container.
+// This may override kubelet's settings for OOM score.
+func (g *Generator) AdjustOomScoreAdj(score *nri.OptionalInt) {
+	if score != nil {
+		g.SetProcessOOMScoreAdj(int(score.Value))
+	} else {
+		g.SetProcessOOMScoreAdj(0)
+		g.Config.Process.OOMScoreAdj = nil
+	}
+}
+
 // AdjustDevices adjusts the (Linux) devices in the OCI Spec.
 func (g *Generator) AdjustDevices(devices []*nri.LinuxDevice) {
 	for _, d := range devices {
@@ -323,6 +349,23 @@ func (g *Generator) AdjustDevices(devices []*nri.LinuxDevice) {
 	}
 }
 
+// InjectCDIDevices injects the requested CDI devices into the OCI Spec.
+// Devices are given by their fully qualified CDI device names. The
+// actual device injection is done using a runtime-specific CDI
+// injection function, set using the WithCDIDeviceInjector option.
+func (g *Generator) InjectCDIDevices(devices []*nri.CDIDevice) error {
+	if len(devices) == 0 || g.injectCDIDevices == nil {
+		return nil
+	}
+
+	names := []string{}
+	for _, d := range devices {
+		names = append(names, d.Name)
+	}
+
+	return g.injectCDIDevices(g.Config, names)
+}
+
 func (g *Generator) AdjustRlimits(rlimits []*nri.POSIXRlimit) error {
 	for _, l := range rlimits {
 		if l == nil {

vendor/github.com/containerd/nri/pkg/stub/stub.go

@@ -35,7 +35,7 @@ import (
 )
 
 // Plugin can implement a number of interfaces related to Pod and Container
-// lifecycle events. No any single such inteface is mandatory, therefore the
+// lifecycle events. No any single such interface is mandatory, therefore the
 // Plugin interface itself is empty. Plugins are required to implement at
 // least one of these interfaces and this is verified during stub creation.
 // Trying to create a stub for a plugin violating this requirement will fail
@@ -137,7 +137,9 @@ type PostUpdateContainerInterface interface {
 
 // Stub is the interface the stub provides for the plugin implementation.
 type Stub interface {
-	// Run the plugin. Starts the plugin then waits for an error or the plugin to stop
+	// Run starts the plugin then waits for the plugin service to exit, either due to a
+	// critical error or an explicit call to Stop(). Once Run() returns, the plugin can be
+	// restarted by calling Run() or Start() again.
 	Run(context.Context) error
 	// Start the plugin.
 	Start(context.Context) error
@@ -148,11 +150,23 @@ type Stub interface {
 
 	// UpdateContainer requests unsolicited updates to containers.
 	UpdateContainers([]*api.ContainerUpdate) ([]*api.ContainerUpdate, error)
+
+	// RegistrationTimeout returns the registration timeout for the stub.
+	// This is the default timeout if the plugin has not been started or
+	// the timeout received in the Configure request otherwise.
+	RegistrationTimeout() time.Duration
+
+	// RequestTimeout returns the request timeout for the stub.
+	// This is the default timeout if the plugin has not been started or
+	// the timeout received in the Configure request otherwise.
+	RequestTimeout() time.Duration
 }
 
 const (
-	// Plugin registration timeout.
-	registrationTimeout = 2 * time.Second
+	// DefaultRegistrationTimeout is the default plugin registration timeout.
+	DefaultRegistrationTimeout = api.DefaultPluginRegistrationTimeout
+	// DefaultRequestTimeout is the default plugin request processing timeout.
+	DefaultRequestTimeout = api.DefaultPluginRequestTimeout
 )
 
 var (
@@ -255,11 +269,14 @@ type stub struct {
 	rpcs       *ttrpc.Server
 	rpcc       *ttrpc.Client
 	runtime    api.RuntimeService
-	closeOnce  sync.Once
 	started    bool
 	doneC      chan struct{}
 	srvErrC    chan error
 	cfgErrC    chan error
+	syncReq    *api.SynchronizeRequest
+
+	registrationTimeout time.Duration
+	requestTimeout      time.Duration
 }
 
 // Handlers for NRI plugin event and request.
@@ -288,7 +305,9 @@ func New(p interface{}, opts ...Option) (Stub, error) {
 		idx:        os.Getenv(api.PluginIdxEnvVar),
 		socketPath: api.DefaultSocketPath,
 		dialer:     func(p string) (stdnet.Conn, error) { return stdnet.Dial("unix", p) },
-		doneC:      make(chan struct{}),
+
+		registrationTimeout: DefaultRegistrationTimeout,
+		requestTimeout:      DefaultRequestTimeout,
 	}
 
 	for _, o := range opts {
@@ -316,10 +335,10 @@ func (stub *stub) Start(ctx context.Context) (retErr error) {
 	stub.Lock()
 	defer stub.Unlock()
 
-	if stub.started {
+	if stub.isStarted() {
 		return fmt.Errorf("stub already started")
 	}
-	stub.started = true
+	stub.doneC = make(chan struct{})
 
 	err := stub.connect()
 	if err != nil {
@@ -378,10 +397,11 @@ func (stub *stub) Start(ctx context.Context) (retErr error) {
 
 	stub.srvErrC = make(chan error, 1)
 	stub.cfgErrC = make(chan error, 1)
-	go func() {
-		stub.srvErrC <- rpcs.Serve(ctx, rpcl)
-		close(stub.doneC)
-	}()
+
+	go func(l stdnet.Listener, doneC chan struct{}, srvErrC chan error) {
+		srvErrC <- rpcs.Serve(ctx, l)
+		close(doneC)
+	}(rpcl, stub.doneC, stub.srvErrC)
 
 	stub.rpcm = rpcm
 	stub.rpcl = rpcl
@@ -401,6 +421,7 @@ func (stub *stub) Start(ctx context.Context) (retErr error) {
 
 	log.Infof(ctx, "Started plugin %s...", stub.Name())
 
+	stub.started = true
 	return nil
 }
 
@@ -413,24 +434,43 @@ func (stub *stub) Stop() {
 	stub.close()
 }
 
+// IsStarted returns true if the plugin has been started either by Start() or by Run().
+func (stub *stub) IsStarted() bool {
+	stub.Lock()
+	defer stub.Unlock()
+	return stub.isStarted()
+}
+
+func (stub *stub) isStarted() bool {
+	return stub.started
+}
+
+// reset stub to the status that can initiate a new
+// NRI connection, the caller must hold lock.
 func (stub *stub) close() {
-	stub.closeOnce.Do(func() {
-		if stub.rpcl != nil {
-			stub.rpcl.Close()
-		}
-		if stub.rpcs != nil {
-			stub.rpcs.Close()
-		}
-		if stub.rpcc != nil {
-			stub.rpcc.Close()
-		}
-		if stub.rpcm != nil {
-			stub.rpcm.Close()
-		}
-		if stub.srvErrC != nil {
-			<-stub.doneC
-		}
-	})
+	if !stub.isStarted() {
+		return
+	}
+
+	if stub.rpcl != nil {
+		stub.rpcl.Close()
+	}
+	if stub.rpcs != nil {
+		stub.rpcs.Close()
+	}
+	if stub.rpcc != nil {
+		stub.rpcc.Close()
+	}
+	if stub.rpcm != nil {
+		stub.rpcm.Close()
+	}
+	if stub.srvErrC != nil {
+		<-stub.doneC
+	}
+
+	stub.started = false
+	stub.conn = nil
+	stub.syncReq = nil
 }
 
 // Run the plugin. Start event processing then wait for an error or getting stopped.
@@ -449,14 +489,11 @@ func (stub *stub) Run(ctx context.Context) error {
 	return err
 }
 
-// Wait for the plugin to stop.
+// Wait for the plugin to stop, should be called after Start() or Run().
 func (stub *stub) Wait() {
-	stub.Lock()
-	if stub.srvErrC == nil {
-		return
+	if stub.IsStarted() {
+		<-stub.doneC
 	}
-	stub.Unlock()
-	<-stub.doneC
 }
 
 // Name returns the full indexed name of the plugin.
@@ -464,6 +501,14 @@ func (stub *stub) Name() string {
 	return stub.idx + "-" + stub.name
 }
 
+func (stub *stub) RegistrationTimeout() time.Duration {
+	return stub.registrationTimeout
+}
+
+func (stub *stub) RequestTimeout() time.Duration {
+	return stub.requestTimeout
+}
+
 // Connect the plugin to NRI.
 func (stub *stub) connect() error {
 	if stub.conn != nil {
@@ -502,7 +547,7 @@ func (stub *stub) connect() error {
 func (stub *stub) register(ctx context.Context) error {
 	log.Infof(ctx, "Registering plugin %s...", stub.Name())
 
-	ctx, cancel := context.WithTimeout(ctx, registrationTimeout)
+	ctx, cancel := context.WithTimeout(ctx, stub.registrationTimeout)
 	defer cancel()
 
 	req := &api.RegisterPluginRequest{
@@ -518,7 +563,9 @@ func (stub *stub) register(ctx context.Context) error {
 
 // Handle a lost connection.
 func (stub *stub) connClosed() {
+	stub.Lock()
 	stub.close()
+	stub.Unlock()
 	if stub.onClose != nil {
 		stub.onClose()
 		return
@@ -558,6 +605,9 @@ func (stub *stub) Configure(ctx context.Context, req *api.ConfigureRequest) (rpl
 	log.Infof(ctx, "Configuring plugin %s for runtime %s/%s...", stub.Name(),
 		req.RuntimeName, req.RuntimeVersion)
 
+	stub.registrationTimeout = time.Duration(req.RegistrationTimeout * int64(time.Millisecond))
+	stub.requestTimeout = time.Duration(req.RequestTimeout * int64(time.Millisecond))
+
 	defer func() {
 		stub.cfgErrC <- retErr
 	}()
@@ -596,16 +646,55 @@ func (stub *stub) Configure(ctx context.Context, req *api.ConfigureRequest) (rpl
 func (stub *stub) Synchronize(ctx context.Context, req *api.SynchronizeRequest) (*api.SynchronizeResponse, error) {
 	handler := stub.handlers.Synchronize
 	if handler == nil {
-		return &api.SynchronizeResponse{}, nil
+		return &api.SynchronizeResponse{More: req.More}, nil
 	}
-	update, err := handler(ctx, req.Pods, req.Containers)
+
+	if req.More {
+		return stub.collectSync(req)
+	}
+
+	return stub.deliverSync(ctx, req)
+}
+
+func (stub *stub) collectSync(req *api.SynchronizeRequest) (*api.SynchronizeResponse, error) {
+	stub.Lock()
+	defer stub.Unlock()
+
+	log.Debugf(noCtx, "collecting sync req with %d pods, %d containers...",
+		len(req.Pods), len(req.Containers))
+
+	if stub.syncReq == nil {
+		stub.syncReq = req
+	} else {
+		stub.syncReq.Pods = append(stub.syncReq.Pods, req.Pods...)
+		stub.syncReq.Containers = append(stub.syncReq.Containers, req.Containers...)
+	}
+
+	return &api.SynchronizeResponse{More: req.More}, nil
+}
+
+func (stub *stub) deliverSync(ctx context.Context, req *api.SynchronizeRequest) (*api.SynchronizeResponse, error) {
+	stub.Lock()
+	syncReq := stub.syncReq
+	stub.syncReq = nil
+	stub.Unlock()
+
+	if syncReq == nil {
+		syncReq = req
+	} else {
+		syncReq.Pods = append(syncReq.Pods, req.Pods...)
+		syncReq.Containers = append(syncReq.Containers, req.Containers...)
+	}
+
+	update, err := stub.handlers.Synchronize(ctx, syncReq.Pods, syncReq.Containers)
 	return &api.SynchronizeResponse{
 		Update: update,
+		More:   false,
 	}, err
 }
 
 // Shutdown the plugin.
-func (stub *stub) Shutdown(ctx context.Context, req *api.ShutdownRequest) (*api.ShutdownResponse, error) {
+func (stub *stub) Shutdown(ctx context.Context, _ *api.ShutdownRequest) (*api.ShutdownResponse, error) {
 	handler := stub.handlers.Shutdown
 	if handler != nil {
 		handler(ctx)

vendor/github.com/containerd/nri/types/v1/types.go

@@ -33,7 +33,7 @@ type Plugin struct {
 //
 // Normally located at /etc/nri/conf.json
 type ConfigList struct {
-	// Verion of the list
+	// Version of the list
 	Version string `json:"version"`
 	// Plugins
 	Plugins []*Plugin `json:"plugins"`

vendor/github.com/containerd/ttrpc/channel.go

@@ -143,10 +143,10 @@ func (ch *channel) recv() (messageHeader, []byte, error) {
 }
 
 func (ch *channel) send(streamID uint32, t messageType, flags uint8, p []byte) error {
-	// TODO: Error on send rather than on recv
-	//if len(p) > messageLengthMax {
-	//	return status.Errorf(codes.InvalidArgument, "refusing to send, message length %v exceed maximum message size of %v", len(p), messageLengthMax)
-	//}
+	if len(p) > messageLengthMax {
+		return OversizedMessageError(len(p))
+	}
+
 	if err := writeMessageHeader(ch.bw, ch.hwbuf[:], messageHeader{Length: uint32(len(p)), StreamID: streamID, Type: t, Flags: flags}); err != nil {
 		return err
 	}

vendor/github.com/containerd/ttrpc/errors.go

@@ -16,7 +16,12 @@
 
 package ttrpc
 
-import "errors"
+import (
+	"errors"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
 
 var (
 	// ErrProtocol is a general error in the handling the protocol.
@@ -32,3 +37,44 @@ var (
 	// ErrStreamClosed is when the streaming connection is closed.
 	ErrStreamClosed = errors.New("ttrpc: stream closed")
 )
+
+// OversizedMessageErr is used to indicate refusal to send an oversized message.
+// It wraps a ResourceExhausted grpc Status together with the offending message
+// length.
+type OversizedMessageErr struct {
+	messageLength int
+	err           error
+}
+
+// OversizedMessageError returns an OversizedMessageErr error for the given message
+// length if it exceeds the allowed maximum. Otherwise a nil error is returned.
+func OversizedMessageError(messageLength int) error {
+	if messageLength <= messageLengthMax {
+		return nil
+	}
+
+	return &OversizedMessageErr{
+		messageLength: messageLength,
+		err:           status.Errorf(codes.ResourceExhausted, "message length %v exceed maximum message size of %v", messageLength, messageLengthMax),
+	}
+}
+
+// Error returns the error message for the corresponding grpc Status for the error.
+func (e *OversizedMessageErr) Error() string {
+	return e.err.Error()
+}
+
+// Unwrap returns the corresponding error with our grpc status code.
+func (e *OversizedMessageErr) Unwrap() error {
+	return e.err
+}
+
+// RejectedLength retrieves the rejected message length which triggered the error.
+func (e *OversizedMessageErr) RejectedLength() int {
+	return e.messageLength
+}
+
+// MaximumLength retrieves the maximum allowed message length that triggered the error.
+func (*OversizedMessageErr) MaximumLength() int {
+	return messageLengthMax
+}