/release-note-edit

kube-apiserver: fix a memory leak when deleting multiple objects with a deletecollection.

from what I can see, this should only be possible in unexpected panic cases... do you see panics logged?

from what I can see, this should only be possible in unexpected panic cases... do you see panics logged?

I only found the timeout filter related panic log. I think the goroutine leak here will not trigger panic. Please correct me if i am wrong

/assign @CatherineF-dev @caesarxuchao
/triage accepted

@fedebongio: GitHub didn't allow me to assign the following users: CatherineF-dev.

Note that only kubernetes members, repo collaborators and people who have commented on this issue/PR can be assigned. Additionally, issues/PRs can only have 10 assignees at the same time.
For more information please see the contributor guide

/ping @liggitt

yeah, it looks like passing deleteoptions with a precondition (like a uid precondition) that causes the delete to return an error will exit workers early

this looks reasonable, but needs to be exercised via a test

/lgtm
/approve

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: sxllwx, wojtek-t

The full list of commands accepted by this bot can be found here.

The pull request process is described here

/lgtm cancel

Heh... - I didn't make the cancelling on time.

@aojea - #105606 (comment) is a valid comment
We should send a follow up to it.

@aojea - #105606 (comment) is a valid comment
We should send a follow up to it.

Though - I don't see any reason why that goroutine may crash - so maybe not super critical...

yeah, me neither, I'm fine with it,

hmm @sxllwx this test pass with and without this patch 🤔

hmm @sxllwx this test pass with and without this patch

Yeah - it's somewhat expected. Because the changes in the code are actually two things:

• change the blocking mechanism
• fix the issue

This test wouldn't pass only if we had a PR with the first one, but without the second.

hmm @sxllwx this test pass with and without this patch 🤔

Without this patch, the new UT TestStoreDeleteCollectionWorkDistributorExited cannot pass.

hmm @sxllwx this test pass with and without this patch thinking

Without this patch, the new UT TestStoreDeleteCollectionWorkDistributorExited cannot pass.

it does, I run the test only without the patch and passes

hmm @sxllwx this test pass with and without this patch thinking

Without this patch, the new UT TestStoreDeleteCollectionWorkDistributorExited cannot pass.

it does, I run the test only without the patch and passes

Humm...Ignore my answer. Although the unit test can pass without this patch, there is still a leak of goroutine. (this is due to the concurrency mechanism of the previous version.)

I think the code can be simplified as follows.

tokenBucket := make(chan struct{}, workersNumber)
errs := make(chan error, workersNumber)
wg := &sync.WaitGroup{}

for _, item := range items {
	wg.Add(1)
	tokenBucket <- struct{}{}

	go func(i runtime.Object) {
		// panics don't cross goroutine boundaries
		defer utilruntime.HandleCrash(func(panicReason interface{}) {
			errs <- fmt.Errorf("DeleteCollection goroutine panicked: %v", panicReason)
		})
		defer wg.Done()
		defer func() {
			<-tokenBucket
		}()
		accessor, err := meta.Accessor(i)
		if err != nil {
			errs <- err
			return
		}

		// DeepCopy the deletion options because individual graceful deleters communicate changes via a mutating
		// function in the delete strategy called in the delete method.  While that is always ugly, it works
		// when making a single call.  When making multiple calls via delete collection, the mutation applied to
		// pod/A can change the option ultimately used for pod/B.
		if _, _, err := e.Delete(ctx, accessor.GetName(), deleteValidation, options.DeepCopy()); err != nil && !apierrors.IsNotFound(err) {
			klog.V(4).InfoS("Delete object in DeleteCollection failed", "object", klog.KObj(accessor), "err", err)
			errs <- err
			return
		}
	}(item)
}
wg.Wait()

select {
case err := <-errs:
	return nil, err
default:
	return listObj, nil
}

Whatever wojtek or Jordan say :)

I think the code can be simplified as follows.

tokenBucket := make(chan struct{}, workersNumber)
errs := make(chan error, workersNumber)
wg := &sync.WaitGroup{}

for _, item := range items {
	wg.Add(1)
	tokenBucket <- struct{}{}

	go func(i runtime.Object) {
		// panics don't cross goroutine boundaries
		defer utilruntime.HandleCrash(func(panicReason interface{}) {
			errs <- fmt.Errorf("DeleteCollection goroutine panicked: %v", panicReason)
		})
		defer wg.Done()
		defer func() {
			<-tokenBucket
		}()
		accessor, err := meta.Accessor(i)
		if err != nil {
			errs <- err
			return
		}

		// DeepCopy the deletion options because individual graceful deleters communicate changes via a mutating
		// function in the delete strategy called in the delete method.  While that is always ugly, it works
		// when making a single call.  When making multiple calls via delete collection, the mutation applied to
		// pod/A can change the option ultimately used for pod/B.
		if _, _, err := e.Delete(ctx, accessor.GetName(), deleteValidation, options.DeepCopy()); err != nil && !apierrors.IsNotFound(err) {
			klog.V(4).InfoS("Delete object in DeleteCollection failed", "object", klog.KObj(accessor), "err", err)
			errs <- err
			return
		}
	}(item)
}
wg.Wait()

select {
case err := <-errs:
	return nil, err
default:
	return listObj, nil
}

I think this code is simpler and faster. But the buffer size of the errs channel needs to be adjusted to len(items).

I'm against the above - this potentially may start deletion of all items at once, which may overload etcd.
We need a control over number of inflight requests to etcd - which the workersNumber is achieving.

@wojtek-t The above is essentially the same as the existed. The goroutine can be executed only on the premise of obtaining a token, thereby ensuring that at most workersNumber requests can be executed each time.

OK sorry - I misread it. Yes - that looks fine