diff --git a/cmd/plugins/topology-aware/policy/mocks_test.go b/cmd/plugins/topology-aware/policy/mocks_test.go index 9bf161044..f8cd0d63b 100644 --- a/cmd/plugins/topology-aware/policy/mocks_test.go +++ b/cmd/plugins/topology-aware/policy/mocks_test.go @@ -599,6 +599,9 @@ func (m *mockContainer) GetMemorySwap() int64 { func (m *mockContainer) GetCtime() time.Time { panic("unimplemented") } +func (m *mockContainer) GetCreatedAt() int64 { + return 0 +} func (m *mockContainer) PreserveCpuResources() bool { return false } diff --git a/pkg/resmgr/cache/cache.go b/pkg/resmgr/cache/cache.go index d9974971a..5b92df723 100644 --- a/pkg/resmgr/cache/cache.go +++ b/pkg/resmgr/cache/cache.go @@ -217,6 +217,8 @@ type Container interface { GetNamespace() string // GetCtime returns the creation time of the container cache object. GetCtime() time.Time + // GetCreatedAt returns the CreatedAt timestamp of the container. + GetCreatedAt() int64 // UpdateState updates the state of the container. UpdateState(ContainerState) // GetState returns the ContainerState of the container. diff --git a/pkg/resmgr/cache/container.go b/pkg/resmgr/cache/container.go index 8436ec2e5..be06a21f3 100644 --- a/pkg/resmgr/cache/container.go +++ b/pkg/resmgr/cache/container.go @@ -423,6 +423,10 @@ func (c *container) GetCtime() time.Time { return c.ctime } +func (c *container) GetCreatedAt() int64 { + return c.Ctr.CreatedAt +} + func (c *container) UpdateState(state ContainerState) { c.Ctr.State = state } diff --git a/pkg/resmgr/nri.go b/pkg/resmgr/nri.go index efb14ff81..afd9698b5 100644 --- a/pkg/resmgr/nri.go +++ b/pkg/resmgr/nri.go @@ -123,13 +123,13 @@ func (p *nriPlugin) onClose() { os.Exit(1) } -func (p *nriPlugin) syncNamesToContainers(containers []cache.Container) []cache.Container { - unmapped := make([]cache.Container, 0, len(p.byname)) +func (p *nriPlugin) syncNamesToContainers(containers []cache.Container) map[string]cache.Container { + unmapped := map[string]cache.Container{} for _, ctr := range containers { old := p.mapNameToContainer(ctr) - if old != nil && old.GetID() != ctr.GetID() { - unmapped = append(unmapped, old) + if old != nil { + unmapped[old.GetID()] = old } } @@ -140,13 +140,43 @@ func (p *nriPlugin) mapNameToContainer(ctr cache.Container) cache.Container { name := ctr.PrettyName() old, ok := p.byname[name] - p.byname[name] = ctr if ok { - nri.Infof("%s: remapped container from %s to %s", name, old.GetID(), ctr.GetID()) - return old + switch { + case ctr.GetState() == cache.ContainerStateCreating: + p.byname[name] = ctr + nri.Infof("%s: remapped container from %s to %s (container creation)", + name, old.GetID(), ctr.GetID()) + return old + case ctr.GetCreatedAt() > old.GetCreatedAt(): + p.byname[name] = ctr + nri.Infof("%s: remapped container from %s to %s (by creation time)", + name, old.GetID(), ctr.GetID()) + return old + case ctr.GetCreatedAt() < old.GetCreatedAt(): + nri.Infof("%s: keeping container mapped to %s (by creation time)", + name, old.GetID()) + return ctr + case ctr.GetState() == cache.ContainerStateCreated && + old.GetState() == cache.ContainerStateRunning: + p.byname[name] = ctr + nri.Infof("%s: remapped container from %s to %s (by state)", + name, old.GetID(), ctr.GetID()) + return old + case ctr.GetState() == cache.ContainerStateRunning && + old.GetState() == cache.ContainerStateCreated: + nri.Infof("%s: keeping container mapped to %s (by state)", name, old.GetID()) + return ctr + default: + nri.Errorf("%s: remapped container from %s to %s (fallback by order might be wrong)", + name, old.GetID(), ctr.GetID()) + p.byname[name] = ctr + return old + } } + p.byname[name] = ctr nri.Infof("%s: mapped container to %s", name, ctr.GetID()) + return nil } @@ -279,7 +309,24 @@ func (p *nriPlugin) Synchronize(ctx context.Context, pods []*api.PodSandbox, con } unmapped := p.syncNamesToContainers(allocated) - if err := m.policy.Sync(allocated, append(released, unmapped...)); err != nil { + allocate := []cache.Container{} + + for _, old := range unmapped { + old.UpdateState(cache.ContainerStateExited) + nri.Infof("marked unmapped container %s (%s) as exited...", + old.PrettyName(), old.GetID()) + } + + for _, c := range allocated { + if _, ok := unmapped[c.GetID()]; ok { + nri.Infof("filtering unmapped container %s (%s) from being (re)allocated...", + c.PrettyName(), c.GetID()) + continue + } + allocate = append(allocate, c) + } + + if err := m.policy.Sync(allocate, released); err != nil { return nil, fmt.Errorf("failed to sync policy %s: %w", m.policy.ActivePolicy(), err) } diff --git a/test/e2e/policies.test-suite/topology-aware/helm-config.yaml.in b/test/e2e/policies.test-suite/topology-aware/helm-config.yaml.in index 06323f4da..41182fe73 100644 --- a/test/e2e/policies.test-suite/topology-aware/helm-config.yaml.in +++ b/test/e2e/policies.test-suite/topology-aware/helm-config.yaml.in @@ -31,11 +31,9 @@ config: httpEndpoint: ":8891" log: debug: - - nri-resource-policy - - resource-manager - - cache - - libmem - - policy + $(for logger in ${DEBUG_LOGGERS:-resource-manager cache libmem policy}; do echo " + - $logger + "; done ) source: true klog: skip_headers: true diff --git a/test/e2e/policies.test-suite/topology-aware/n4c16/test31-duplicate-disambiguation/code.var.sh b/test/e2e/policies.test-suite/topology-aware/n4c16/test31-duplicate-disambiguation/code.var.sh new file mode 100644 index 000000000..7aed30b3e --- /dev/null +++ b/test/e2e/policies.test-suite/topology-aware/n4c16/test31-duplicate-disambiguation/code.var.sh @@ -0,0 +1,155 @@ +TESTNS=repro-dup +# Triggering transient duplicates is timing sensitive. +PODS=16 +CONTAINERS=3 + +setup() { + vm-command "kubectl create namespace $TESTNS" + # Disable debug logging, otherwise log rotation may prevent finding remap lines. + helm_config=$(DEBUG_LOGGERS="none" instantiate helm-config.yaml) helm-launch topology-aware +} + +cleanup() { + vm-command "kubectl delete pods -n $TESTNS --all --now || :" + vm-command 'pkill -9 -f "sleep inf"' + vm-command 'pkill -9 -f "echo pod"' + vm-command "kubectl delete namespace $TESTNS --now || :" + helm-terminate +} + +create-containers() { + n=$PODS CONTCOUNT=$CONTAINERS CPUREQ=10m MEMREQ=50M CPULIM=10m namespace=$TESTNS wait='' create burstable + # Wait at least a quarter of containers are running to have + # many ocontainers in many different stages. + vm-command "nc=0; lnc=-1 + while (( nc < $(( PODS * CONTAINERS / 4 )) )); do + sleep 0.1 + nc=\$(pgrep -f pod[0-9]+c[0-9]+ | wc -l) + [ \$lnc = \$nc ] || echo \$nc podXcY processes running + lnc=\$nc + done" +} + +kill-containers() { + case ${k8scri:-containerd} in + containerd) + vm-command 'kill -STOP $(pidof containerd) + pkill -9 -f "sleep inf" + sleep 1 + kill -9 $(pidof containerd)' + + ;; + cri-o) + vm-command 'kill -STOP $(pidof crio) + pkill -9 -f "sleep inf" + sleep 1 + kill -9 $(pidof crio)' + ;; + *) + error "Unknown runtime: $runtime" + ;; + esac +} + +wait-containers-restart() { + local statuses="" + + while ! [[ "$statuses" == "Running" ]]; do + sleep 5 + if vm-command "kubectl logs -n kube-system ds/nri-resource-policy-topology-aware 2>&1 | grep -iE 'remap|keeping|duplicate'"; then + break + fi + vm-command "kubectl get pods -A --no-headers=true | tr -s '\t' ' '| cut -d ' ' -f4 | sort -u" + statuses=$COMMAND_OUTPUT + done +} + +check-transient-duplicates-present() { + # Check that we managed to trigger transient duplicate containers. + if ! grep -q -E '(remap)|(keeping.*mapped)' <<< $COMMAND_OUTPUT; then + echo "No remapping of containers found in the logs..." + return 1 + fi + return 0 +} + +check-remap-disambiguation() { + # Verify that each duplicate was disambiguated by creation time. + if grep -E '(remap)|(keeping.*mapped)' <<< $COMMAND_OUTPUT | \ + grep -v 'by creation time'; then + grep -E '(remap)|(keeping.*mapped)' <<< $COMMAND_OUTPUT | \ + grep -v 'by creation time' | sed 's/^/INCORRECT: /g' + error "Found some incorrectly remapped containers in the logs" + fi + + echo "Only found correctly remapped containers in the logs..." + grep -E '(remap)|(keeping.*mapped)' <<< $COMMAND_OUTPUT | sed 's/^/CORRECT: /g' +} + +check-no-duplicate-allocations() { + # Verify that we did not end up with duplicate allocation entries. + if grep -q -i 'duplicate allocation entries' <<< $COMMAND_OUTPUT; then + grep -i 'duplicate allocation entries' <<< $COMMAND_OUTPUT | sed 's/^/DUPLICATE: /g' + error "Found duplicate allocation entries in the logs" + fi +} + +pull-logs() { + local cnt=0 + while [ $cnt -lt 5 ]; do + vm-command "kubectl logs -n kube-system ds/nri-resource-policy-topology-aware 2>&1 | grep -iE 'remap|keeping|duplicate|unable'" + if grep -q 'unable to retrieve container logs for' <<< $COMMAND_OUTPUT; then + echo "Unable to retrieve policy logs, retrying..." + sleep 3 + let cnt=$ctn+1 + else + return 0 + fi + done + return 1 +} + +check-logs() { + if ! pull-logs; then + echo "Failed to pull policy logs..." + return 1 + fi + + if ! check-transient-duplicates-present; then + return 1 + fi + + check-remap-disambiguation + check-no-duplicate-allocations + + return 0 +} + +cleanup +setup +create-containers + +retries=0 +while true; do + kill-containers + wait-containers-restart + + if check-logs; then + break + fi + + echo "Retrying..." + let retries=$retries+1 + + if [ $retries -ge 5 ]; then + echo "Max retries ($retries) reached, could not reproduce the issue, giving up" + break + fi + + # Re-create all pods, otherwise CrashLoopBackup may prevents finding remaps. + cleanup + setup + create-containers +done + +cleanup