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") + } +}