resmgr: retain pending updates on nri failure

cmd/plugins/topology-aware/policy/mocks_test.go

@@ -530,7 +530,10 @@ func (m *mockContainer) SetSchedulingIOPriority(int32) {
 func (m *mockContainer) GetPendingAdjustment() *nri.ContainerAdjustment {
 	panic("unimplemented")
 }
-func (m *mockContainer) GetPendingUpdate() *nri.ContainerUpdate {
+func (m *mockContainer) PeekPendingUpdate() *nri.ContainerUpdate {
+	panic("unimplemented")
+}
+func (m *mockContainer) ClearPendingUpdate() {
 	panic("unimplemented")
 }
 func (m *mockContainer) GetAffinity() ([]*cache.Affinity, error) {

pkg/resmgr/cache/cache.go

@@ -345,8 +345,10 @@ type Container interface {
 
 	// GetPendingAdjusmentn clears and returns any pending adjustment for the container.
 	GetPendingAdjustment() *nri.ContainerAdjustment
-	// GetPendingUpdate clears and returns any pending update for the container.
-	GetPendingUpdate() *nri.ContainerUpdate
+	// PeekPendingUpdate returns any pending update for the container without clearing it.
+	PeekPendingUpdate() *nri.ContainerUpdate
+	// ClearPendingUpdate discards any pending update for the container.
+	ClearPendingUpdate()
 
 	// GetAffinity returns the annotated affinity expressions for this container.
 	GetAffinity() ([]*Affinity, error)

pkg/resmgr/cache/container.go

@@ -691,22 +691,28 @@ func (c *container) GetPendingAdjustment() *nri.ContainerAdjustment {
 	return req
 }
 
-func (c *container) GetPendingUpdate() *nri.ContainerUpdate {
+func (c *container) PeekPendingUpdate() *nri.ContainerUpdate {
 	if c.request == nil {
 		return nil
 	}
 
 	req, ok := c.request.(*nri.ContainerUpdate)
 	if !ok {
+		// Drop a request we cannot deliver as an update, the same way returning
+		// one does: keeping it would only repeat this error forever.
 		log.Errorf("%s: queried pending update has mismatching type %T",
 			c.PrettyName(), c.request)
-		req = nil
+		c.request = nil
+		return nil
 	}
 
-	c.request = nil
 	return req
 }
 
+func (c *container) ClearPendingUpdate() {
+	c.request = nil
+}
+
 func (c *container) InsertMount(m *Mount) {
 	var adjust *nri.ContainerAdjustment
 

pkg/resmgr/cache/container_test.go

@@ -545,7 +545,7 @@ var _ = Describe("Container", func() {
 
 		ctrs[0].SetCPUShares(int64(shares))
 
-		pending := ctrs[0].GetPendingUpdate()
+		pending := ctrs[0].PeekPendingUpdate()
 		Expect(pending).ToNot(BeNil())
 		value := pending.GetLinux().GetResources().GetCpu().GetShares().GetValue()
 		Expect(value).To(Equal(uint64(shares)))
@@ -569,7 +569,7 @@ var _ = Describe("Container", func() {
 
 		ctrs[0].SetCPUQuota(int64(quota))
 
-		pending := ctrs[0].GetPendingUpdate()
+		pending := ctrs[0].PeekPendingUpdate()
 		Expect(pending).ToNot(BeNil())
 		value := pending.GetLinux().GetResources().GetCpu().GetQuota().GetValue()
 		Expect(value).To(Equal(int64(quota)))
@@ -593,7 +593,7 @@ var _ = Describe("Container", func() {
 
 		ctrs[0].SetCPUPeriod(int64(period))
 
-		pending := ctrs[0].GetPendingUpdate()
+		pending := ctrs[0].PeekPendingUpdate()
 		Expect(pending).ToNot(BeNil())
 		value := pending.GetLinux().GetResources().GetCpu().GetPeriod().GetValue()
 		Expect(value).To(Equal(uint64(period)))
@@ -617,7 +617,7 @@ var _ = Describe("Container", func() {
 
 		ctrs[0].SetCpusetCpus(cpus)
 
-		pending := ctrs[0].GetPendingUpdate()
+		pending := ctrs[0].PeekPendingUpdate()
 		Expect(pending).ToNot(BeNil())
 		value := pending.GetLinux().GetResources().GetCpu().GetCpus()
 		Expect(value).To(Equal(cpus))
@@ -641,7 +641,7 @@ var _ = Describe("Container", func() {
 
 		ctrs[0].SetCpusetMems(mems)
 
-		pending := ctrs[0].GetPendingUpdate()
+		pending := ctrs[0].PeekPendingUpdate()
 		Expect(pending).ToNot(BeNil())
 		value := pending.GetLinux().GetResources().GetCpu().GetMems()
 		Expect(value).To(Equal(mems))
@@ -665,7 +665,7 @@ var _ = Describe("Container", func() {
 
 		ctrs[0].SetMemoryLimit(limit)
 
-		pending := ctrs[0].GetPendingUpdate()
+		pending := ctrs[0].PeekPendingUpdate()
 		Expect(pending).ToNot(BeNil())
 		value := pending.GetLinux().GetResources().GetMemory().GetLimit().GetValue()
 		Expect(value).To(Equal(limit))

pkg/resmgr/nri.go

@@ -696,25 +696,76 @@ func (p *nriPlugin) RemoveContainer(ctx context.Context, pod *api.PodSandbox, co
 	return nil
 }
 
+// updateContainers pushes the pending container updates to the runtime. Unlike
+// the updates carried by the response to an NRI request, these are only taken
+// from the containers once the runtime has them: an update lost on the way out
+// is the only record of what the container should look like, so dropping it
+// would leave the container as it is with nothing left to retry it with.
 func (p *nriPlugin) updateContainers() (retErr error) {
 	// Notes: must be called with p.resmgr lock held.
 
-	updates := p.getPendingUpdates(nil)
+	updates := p.peekPendingUpdates(nil)
 
 	event := UpdateContainers
 	p.dump(out, event, updates)
 	defer func() {
 		p.dump(in, event, retErr)
 	}()
 
-	_, err := p.stub.UpdateContainers(updates)
+	failed, err := p.stub.UpdateContainers(updates)
 	if err != nil {
 		return fmt.Errorf("post-config container update failed: %w", err)
 	}
 
+	p.clearPendingUpdates(updates, failed)
+
 	return nil
 }
 
+func (p *nriPlugin) peekPendingUpdates(skip *api.Container) []*api.ContainerUpdate {
+	m := p.resmgr
+	updates := []*api.ContainerUpdate{}
+	for _, c := range m.cache.GetPendingContainers() {
+		if skip != nil && skip.GetId() == c.GetID() {
+			continue
+		}
+
+		if u := c.PeekPendingUpdate(); u != nil {
+			p.setDefaultClasses(c, u)
+			updates = append(updates, u)
+		}
+	}
+
+	return updates
+}
+
+// clearPendingUpdates takes the delivered updates from their containers. The
+// ones the runtime says it could not apply are left pending, for the same
+// reason a failed request leaves all of them pending.
+func (p *nriPlugin) clearPendingUpdates(delivered, failed []*api.ContainerUpdate) {
+	m := p.resmgr
+	for _, u := range delivered {
+		id := u.GetContainerId()
+		if slices.ContainsFunc(failed, func(f *api.ContainerUpdate) bool {
+			return f.GetContainerId() == id
+		}) {
+			nri.Warnf("runtime failed to update container %s, keeping the update pending", id)
+			continue
+		}
+
+		c, ok := m.cache.LookupContainer(id)
+		if !ok {
+			continue
+		}
+
+		c.ClearPendingUpdate()
+		for _, ctrl := range c.GetPending() {
+			c.ClearPending(ctrl)
+		}
+		m.policy.ExportResourceData(c)
+	}
+}
+
 func (p *nriPlugin) getPendingAdjustment(container *api.Container) *api.ContainerAdjustment {
 	if c, ok := p.resmgr.cache.LookupContainer(container.GetId()); ok {
 		adjust := c.GetPendingAdjustment()
@@ -729,24 +780,14 @@ func (p *nriPlugin) getPendingAdjustment(container *api.Container) *api.Containe
 	return nil
 }
 
+// getPendingUpdates collects the updates to carry in the response to an NRI
+// request, and takes them from their containers as it does so. Nothing tells us
+// whether the runtime applied the updates it was handed this way, so there is no
+// later point to clear them at, and keeping them would repeat them in every
+// response from here on.
 func (p *nriPlugin) getPendingUpdates(skip *api.Container) []*api.ContainerUpdate {
-	m := p.resmgr
-	updates := []*api.ContainerUpdate{}
-	for _, c := range m.cache.GetPendingContainers() {
-		if skip != nil && skip.GetId() == c.GetID() {
-			continue
-		}
-
-		if u := c.GetPendingUpdate(); u != nil {
-			p.setDefaultClasses(c, u)
-			updates = append(updates, u)
-
-			for _, ctrl := range c.GetPending() {
-				c.ClearPending(ctrl)
-			}
-			m.policy.ExportResourceData(c)
-		}
-	}
+	updates := p.peekPendingUpdates(skip)
+	p.clearPendingUpdates(updates, nil)
 
 	return updates
 }