From 13b5d18f7e95499d88f12c2a89d64e802e7be136 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Tue, 22 Sep 2026 15:24:44 +0300 Subject: [PATCH] resmgr: retain pending updates on nri failure An update pushed to the runtime outside an NRI response is the only record of what its container should look like: if we drop it and the push does not arrive, the container stays as it is with nothing left to retry it with. So the updates are now taken from their containers only once the runtime has them, and a failed request leaves them all pending. The runtime can also answer a request it accepted with the updates it could not apply. Those containers were not updated either, so their updates stay pending just the same, which needs peeking at a pending update without consuming it and clearing the delivered ones by hand. The updates carried by an NRI response are still taken as they are collected. Nothing acknowledges those, so there is no later point to clear them at, and keeping them would repeat them in every response from here on. Collecting them is now peeking at them and clearing them right away, which is all it ever did, so the destructive read they used to use is gone from the container interface. Signed-off-by: Ed Bartosh --- .../topology-aware/policy/mocks_test.go | 5 +- pkg/resmgr/cache/cache.go | 6 +- pkg/resmgr/cache/container.go | 12 +- pkg/resmgr/cache/container_test.go | 12 +- pkg/resmgr/nri.go | 79 ++++++-- pkg/resmgr/nri_test.go | 189 ++++++++++++++++++ 6 files changed, 272 insertions(+), 31 deletions(-) create mode 100644 pkg/resmgr/nri_test.go diff --git a/cmd/plugins/topology-aware/policy/mocks_test.go b/cmd/plugins/topology-aware/policy/mocks_test.go index 28b074dbf..2cde49234 100644 --- a/cmd/plugins/topology-aware/policy/mocks_test.go +++ b/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) { diff --git a/pkg/resmgr/cache/cache.go b/pkg/resmgr/cache/cache.go index f5091999d..82cc9d65a 100644 --- a/pkg/resmgr/cache/cache.go +++ b/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) diff --git a/pkg/resmgr/cache/container.go b/pkg/resmgr/cache/container.go index 62b18a28d..052543fdc 100644 --- a/pkg/resmgr/cache/container.go +++ b/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 diff --git a/pkg/resmgr/cache/container_test.go b/pkg/resmgr/cache/container_test.go index 7331f50af..df92abda2 100644 --- a/pkg/resmgr/cache/container_test.go +++ b/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)) diff --git a/pkg/resmgr/nri.go b/pkg/resmgr/nri.go index 225e433a1..7833731b7 100644 --- a/pkg/resmgr/nri.go +++ b/pkg/resmgr/nri.go @@ -696,10 +696,15 @@ 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 cleared +// 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) @@ -707,14 +712,60 @@ func (p *nriPlugin) updateContainers() (retErr error) { 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 clears 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 clears 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 } diff --git a/pkg/resmgr/nri_test.go b/pkg/resmgr/nri_test.go new file mode 100644 index 000000000..950d0068d --- /dev/null +++ b/pkg/resmgr/nri_test.go @@ -0,0 +1,189 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// 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 resmgr + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" + + nriapi "github.com/containerd/nri/pkg/api" + nrilog "github.com/containerd/nri/pkg/log" + nristub "github.com/containerd/nri/pkg/stub" + + cfgapi "github.com/containers/nri-plugins/pkg/apis/config/v1alpha1" + "github.com/containers/nri-plugins/pkg/resmgr/cache" + "github.com/containers/nri-plugins/pkg/resmgr/policy" +) + +// testPolicy is a policy which only exports resource data. Embedding the +// interface leaves the rest nil, which is fine as long as nothing calls them. +type testPolicy struct { + policy.Policy +} + +func (testPolicy) ExportResourceData(cache.Container) {} + +// testStub is an NRI stub which only serves container updates: err fails the +// request, failed is what the runtime reports it could not apply, and sent +// records what we asked it to update. +type testStub struct { + err error + failed []*nriapi.ContainerUpdate + sent []*nriapi.ContainerUpdate +} + +var _ nristub.Stub = (*testStub)(nil) + +func (s *testStub) Run(context.Context) error { return nil } +func (s *testStub) Start(context.Context) error { return nil } +func (s *testStub) Stop() {} +func (s *testStub) Wait() {} +func (s *testStub) RegistrationTimeout() time.Duration { return 0 } +func (s *testStub) RequestTimeout() time.Duration { return 0 } +func (s *testStub) Logger() nrilog.Logger { return nil } +func (s *testStub) RuntimeNRIVersion() string { return "" } +func (s *testStub) PluginNRIVersion() string { return "" } +func (s *testStub) UpdateContainers(updates []*nriapi.ContainerUpdate) ([]*nriapi.ContainerUpdate, error) { + s.sent = append(s.sent, updates...) + return s.failed, s.err +} + +// newTestUpdate returns a resource manager which updates containers through the +// given stub, and a running container with a pending update of 42 CPU shares, +// as an out of band allocation change would leave behind. +func newTestUpdate(t *testing.T, stub *testStub) (*resmgr, cache.Container) { + t.Helper() + + // The cache refuses a directory others may write to, which is what a + // temporary directory is here. + cacheDir := filepath.Join(t.TempDir(), "cache") + if err := os.Mkdir(cacheDir, 0700); err != nil { + t.Fatalf("failed to create cache directory: %v", err) + } + + cch, err := cache.NewCache(cache.Options{CacheDir: cacheDir}) + if err != nil { + t.Fatalf("failed to create cache: %v", err) + } + + pod := cch.InsertPod(&nriapi.PodSandbox{ + Id: "test-pod", + Uid: "uid-test-pod", + Name: "test-pod", + Namespace: "default", + Linux: &nriapi.LinuxPodSandbox{ + CgroupParent: "/test/pod", + }, + }, nil) + if pod == nil { + t.Fatal("failed to insert pod into cache") + } + + ctr, err := cch.InsertContainer(&nriapi.Container{ + Id: "test-container", + PodSandboxId: pod.GetID(), + Name: "test-container", + State: cache.ContainerStateRunning, + }) + if err != nil { + t.Fatalf("failed to insert container into cache: %v", err) + } + + ctr.SetCPUShares(42) + + m := &resmgr{ + cache: cch, + cfg: &cfgapi.TopologyAwarePolicy{}, + policy: testPolicy{}, + } + m.nri = &nriPlugin{ + resmgr: m, + stub: stub, + } + + return m, ctr +} + +func TestUpdateContainersRetainsPendingUpdatesOnFailure(t *testing.T) { + want := errors.New("update failed") + m, ctr := newTestUpdate(t, &testStub{err: want}) + + err := m.nri.updateContainers() + if err == nil { + t.Fatal("updateContainers() succeeded, expected failure") + } + if !errors.Is(err, want) { + t.Fatalf("updateContainers() error = %v, want %v", err, want) + } + + pending := ctr.PeekPendingUpdate() + if pending == nil { + t.Fatal("updateContainers() cleared the pending update after failure") + } + if got := pending.GetLinux().GetResources().GetCpu().GetShares().GetValue(); got != 42 { + t.Fatalf("pending CPU shares = %d, want 42", got) + } + if !ctr.HasPending(cache.NRI) { + t.Fatal("updateContainers() cleared pending container markers after failure") + } +} + +func TestUpdateContainersClearsDeliveredUpdates(t *testing.T) { + stub := &testStub{} + m, ctr := newTestUpdate(t, stub) + + if err := m.nri.updateContainers(); err != nil { + t.Fatalf("updateContainers() failed: %v", err) + } + + if len(stub.sent) != 1 { + t.Fatalf("runtime got %d update(s), want 1", len(stub.sent)) + } + if got := stub.sent[0].GetLinux().GetResources().GetCpu().GetShares().GetValue(); got != 42 { + t.Fatalf("updated CPU shares = %d, want 42", got) + } + + if ctr.PeekPendingUpdate() != nil { + t.Error("updateContainers() left the delivered update pending") + } + if ctr.HasPending(cache.NRI) { + t.Error("updateContainers() left pending container markers behind") + } +} + +// TestUpdateContainersRetainsUpdatesTheRuntimeRejected covers the updates which +// come back in the runtime's answer instead of failing the request: those +// containers were not updated either, so their updates stay pending just the +// same. +func TestUpdateContainersRetainsUpdatesTheRuntimeRejected(t *testing.T) { + stub := &testStub{} + m, ctr := newTestUpdate(t, stub) + stub.failed = []*nriapi.ContainerUpdate{{ContainerId: ctr.GetID()}} + + if err := m.nri.updateContainers(); err != nil { + t.Fatalf("updateContainers() failed: %v", err) + } + + if ctr.PeekPendingUpdate() == nil { + t.Error("updateContainers() cleared the update the runtime rejected") + } + if !ctr.HasPending(cache.NRI) { + t.Error("updateContainers() cleared the pending container markers of a rejected update") + } +}