[KWOK] retry when applying Stages fails

[caozhuozi]

What type of PR is this?

/kind feature

What this PR does / why we need it:

Which issue(s) this PR fixes:

Fixes #816

Special notes for your reviewer:

Since we have a new design based on the weight delaying queue(#902), the old implementation #904 was rejected. Will close the old one and start here.

Currently, I only modified the node controller for a pre-review. Please feel free to give comments!

Does this PR introduce a user-facing change?

Add Stage retry mechanism

Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

[netlify]

✅ Deploy Preview for k8s-kwok canceled.

Name	Link
🔨 Latest commit	8832ec1
🔍 Latest deploy log	https://app.netlify.com/sites/k8s-kwok/deploys/65b30a5b4442aa0008e7a419

[k8s-ci-robot]

Hi @caozhuozi. Thanks for your PR.

I'm waiting for a kubernetes-sigs member to verify that this patch is reasonable to test. If it is, they should reply with /ok-to-test on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should join the org to skip this step.

Once the patch is verified, the new status will be reflected by the ok-to-test label.

I understand the commands that are listed here.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

[caozhuozi]

Hi, @wzshiming.
I added a shoudRetry function to determine whether an error needs to be retried.

Currently, I only check those network errors and bypass all other error types.

The logic is slightly different from the one I proposed before: #911 (comment)
The main difference is that I gave up using the IsKwokError() method which was designed to check whether the error is generated by kwok itself such as ComputePatchError.
It is easy for me to define the error:

var ComputePatchError = errors.New("compute patch error")

But it's hard for me to integrate the newly defined error into the existing code. For example:

func (c *NodeController) computePatch(node *corev1.Node, tpl string) ([]byte, error) {
	patch, err := c.renderer.ToJSON(tpl, node)
	if err != nil {
		return nil, err
	}

	original, err := json.Marshal(node.Status)
	if err != nil {
		return nil, err
	}

	sum, err := strategicpatch.StrategicMergePatch(original, patch, node.Status)
	if err != nil {
		return nil, err
	}

	nodeStatus := corev1.NodeStatus{}
	err = json.Unmarshal(sum, &nodeStatus)
	if err != nil {
		return nil, err
	}

	dist, err := json.Marshal(nodeStatus)
	if err != nil {
		return nil, err
	}

	if bytes.Equal(original, dist) {
		return nil, nil
	}

	return json.Marshal(map[string]json.RawMessage{
		"status": patch,
	})
}

If I return the newly defined error in computePatch directly, the original err messages within the method will be lost.
For example:

func (c *NodeController) computePatch(node *corev1.Node, tpl string) ([]byte, error) {
	patch, err := c.renderer.ToJSON(tpl, node)
	if err != nil {
		return nil, ComputePatchError  // the original `err` will be lost
	}
        // omitted
}

An alternative way is that I return the newly defined error where computePatch method is called:

patch, err = c.computePatch(node, next.StatusTemplate)
if err != nil {
    logger.Debug("failed to obtain the patch of node %s: %w", node.Name, err)
    return ComputePatchError
}

But I am not sure if it is Go idiomatic.

[wzshiming]

I think like We can use errors.Join and fmt.Errorf

[caozhuozi]

Hi, @wzshiming. Do you mean to join the specific error with the newly defined error?
I have two concerns here:

1. I'm not sure if it is a Go idiomatic way

2. the other concern is that where should I put the join statement? There are 2 choices currently:

   • inside computePatch method, like:

       func (c *NodeController) computePatch(node *corev1.Node, tpl string) ([]byte, error) {
       patch, err := c.renderer.ToJSON(tpl, node)
       if err != nil {
       return errors.Join(ComputePatchError, err)  // the original `err` will be lost
       }
       // omitted
       }

   • at where computePath method is called, like

     patch, err = c.computePatch(node, next.StatusTemplate)
     if err != nil {
         return errors.Join(ComputePatchError, err)
     }

   which one do you prefer?

[wzshiming]

I think you can finish this and verify the various corner cases locally

[caozhuozi]

I made several changes in this commit:

1. define a new error type ErrComputePatch, which can cover both finalizers and status patching errors.
2. remove those resource isNotFound checks in node controller as they will be handled in shouldRetry, which is the only entry for determining errors.
3. move the patch operation out of finalizersModify and rename the method to computeFinalizerPatch. If any error occurs in computeFinalizerPatch, we should also mark the error as ErrComputePatch. The motivation is that we should put all the patch operations in playStage, resulting in clearer code logic.
4. By comparison, rename the computePatch to computeStatusPatch to indicate it is exclusively used to patch status.
5. generalize patchResource method such that it can be used by both status patching and finalizers patching.

Please feel free to give comments! 🙏

[caozhuozi]

since we only need to check the network error, I just removed the defined ErrComputePatch in the latest commit.

[wzshiming]

/ok-to-test

[caozhuozi]

In this new commit,

• Get the finalizerModify logic back
• check whether the key already exists in the map when adding a job
• Create a function in utils to obtain a default backoff setting shared by all kwok controllers for those retried jobs.
• rebase the code based on the recent changes.

If there are no problems with this version, I will start to change other controllers.

[wzshiming]

Sorry for the delay, this PR may need to be rebase due to the conflict with #920, I will take time to test this PR this week.

[caozhuozi]

Sorry for the delay, this PR may need to be rebase due to the conflict with #920, I will take time to test this PR this week.

Oh, @wzshiming! Understood! No worry! I will rebase it. 😊

[wzshiming]

Suggested change

	"node", pod.Resource.Name, "stage", pod.Stage.Name())
	"pod", pod.Key,
	"stage", pod.Stage.Name(),
	)

[wzshiming]

Suggested change

	"resource", resource.Resource.GetName(), "stage", resource.Stage.Name(), "retry", *resource.RetryCount)
	"resource", resource.Key,
	"stage", resource.Stage.Name(),
	"retry", *resource.RetryCount,
	)

[wzshiming]

Suggested change

	logger.Debug("Skip modify status",
	"reason", "do not need to modify status",
	logger.Debug("Skip node",
	"reason", "do not need to modify",

[wzshiming]

Suggested change

	if steps <= 0 {
	if steps < 0 {

The parameter for the first failure should be 0, which is the expected 1s delay

[caozhuozi]

since uint64 has no value less than zero after the newest change, I just removed the if clause.

[wzshiming]

Suggested change

	*pod.RetryCount++
	logger.Info("retrying for failed job",
	"pod", pod.Key,
	"stage", pod.Stage.Name(),
	"retry", *pod.RetryCount,
	)
	// for failed jobs, we re-push them into the queue with a lower weight
	// and a backoff period to avoid blocking normal tasks
	retryDelay := backoffDelayByStep(*pod.RetryCount, c.backoff)
	c.addStageJob(pod, retryDelay, 1)
	retryCount := atomic.AddUint64(resource.RetryCount, 1) - 1
	logger.Info("retrying for failed job",
	"pod", pod.Key,
	"stage", pod.Stage.Name(),
	"retry", retryCount,
	)
	// for failed jobs, we re-push them into the queue with a lower weight
	// and a backoff period to avoid blocking normal tasks
	retryDelay := backoffDelayByStep(retryCount, c.backoff)
	c.addStageJob(pod, retryDelay, 1)

In some corner case, there might be a data race, using atomic add to avoid panic.

[caozhuozi]

Thanks!!

[wzshiming]

ditto

[wzshiming]

ditto

[wzshiming]

/approve
/lgtm
/label tide/merge-method-squash

[k8s-ci-robot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: caozhuozi, wzshiming

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

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [wzshiming]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[caozhuozi]

@wzshiming

I made many mistakes and misunderstood a lot of things along the way (and I also learned a lot.) Thanks for the great patience and for taking the trouble to guide me during the PR.