Skip to content

Commit 9d816f1

Browse files
omerap12omerap12
authored
Replace PollImmediate with PollUntilContextTimeout (kubernetes#128147)
* Replace PollImmediate with PollUntilContextTimeout Signed-off-by: Omer Aplatony <[email protected]> * Add context to RetryErrorCondition function Signed-off-by: Omer Aplatony <[email protected]> * lint: fix error comparison in scale package Signed-off-by: Omer Aplatony <[email protected]> * Fix RetryErrorCondition function signature Signed-off-by: Omer Aplatony <[email protected]> * revert to if err statement Signed-off-by: Omer Aplatony <[email protected]> --------- Signed-off-by: Omer Aplatony <[email protected]>
1 parent c9024e7 commit 9d816f1

File tree

5 files changed

+34
-32
lines changed

5 files changed

+34
-32
lines changed

‎staging/src/k8s.io/kubectl/pkg/cmd/apply/patcher.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ limitations under the License.
1717
package apply
1818

1919
import (
20+
"context"
2021
"encoding/json"
2122
"fmt"
2223
"io"
@@ -387,7 +388,7 @@ func (p *Patcher) deleteAndCreate(original runtime.Object, modified []byte, name
387388
return modified, nil, err
388389
}
389390
// TODO: use wait
390-
if err := wait.PollImmediate(1*time.Second, p.Timeout, func() (bool, error) {
391+
if err := wait.PollUntilContextTimeout(context.Background(), 1*time.Second, p.Timeout, true, func(ctx context.Context) (bool, error) {
391392
if _, err := p.Helper.Get(namespace, name); !apierrors.IsNotFound(err) {
392393
return false, err
393394
}

‎staging/src/k8s.io/kubectl/pkg/cmd/replace/replace.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ limitations under the License.
1717
package replace
1818

1919
import (
20+
"context"
2021
"fmt"
2122
"net/url"
2223
"os"
@@ -358,8 +359,7 @@ func (o *ReplaceOptions) forceReplace() error {
358359
if err != nil {
359360
return err
360361
}
361-
362-
return wait.PollImmediate(1*time.Second, timeout, func() (bool, error) {
362+
return wait.PollUntilContextTimeout(context.Background(), 1*time.Second, timeout, true, func(ctx context.Context) (bool, error) {
363363
if err := info.Get(); !errors.IsNotFound(err) {
364364
return false, err
365365
}

‎staging/src/k8s.io/kubectl/pkg/scale/scale.go‎

Lines changed: 12 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,13 @@ package scale
1818

1919
import (
2020
"context"
21+
"errors"
2122
"fmt"
2223
"strconv"
2324
"time"
2425

2526
autoscalingv1 "k8s.io/api/autoscaling/v1"
26-
"k8s.io/apimachinery/pkg/api/errors"
27+
apierrors "k8s.io/apimachinery/pkg/api/errors"
2728
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2829
"k8s.io/apimachinery/pkg/runtime/schema"
2930
"k8s.io/apimachinery/pkg/types"
@@ -80,14 +81,14 @@ func NewRetryParams(interval, timeout time.Duration) *RetryParams {
8081
}
8182

8283
// ScaleCondition is a closure around Scale that facilitates retries via util.wait
83-
func ScaleCondition(r Scaler, precondition *ScalePrecondition, namespace, name string, count uint, updatedResourceVersion *string, gvr schema.GroupVersionResource, dryRun bool) wait.ConditionFunc {
84-
return func() (bool, error) {
84+
func ScaleCondition(r Scaler, precondition *ScalePrecondition, namespace, name string, count uint, updatedResourceVersion *string, gvr schema.GroupVersionResource, dryRun bool) wait.ConditionWithContextFunc {
85+
return func(context.Context) (bool, error) {
8586
rv, err := r.ScaleSimple(namespace, name, precondition, count, gvr, dryRun)
8687
if updatedResourceVersion != nil {
8788
*updatedResourceVersion = rv
8889
}
8990
// Retry only on update conflicts.
90-
if errors.IsConflict(err) {
91+
if apierrors.IsConflict(err) {
9192
return false, nil
9293
}
9394
if err != nil {
@@ -171,7 +172,7 @@ func (s *genericScaler) Scale(namespace, resourceName string, newSize uint, prec
171172
retry = &RetryParams{Interval: time.Millisecond, Timeout: time.Millisecond}
172173
}
173174
cond := ScaleCondition(s, preconditions, namespace, resourceName, newSize, nil, gvr, dryRun)
174-
if err := wait.PollImmediate(retry.Interval, retry.Timeout, cond); err != nil {
175+
if err := wait.PollUntilContextTimeout(context.Background(), retry.Interval, retry.Timeout, true, cond); err != nil {
175176
return err
176177
}
177178
if waitForReplicas != nil {
@@ -182,9 +183,9 @@ func (s *genericScaler) Scale(namespace, resourceName string, newSize uint, prec
182183

183184
// scaleHasDesiredReplicas returns a condition that will be true if and only if the desired replica
184185
// count for a scale (Spec) equals its updated replicas count (Status)
185-
func scaleHasDesiredReplicas(sClient scaleclient.ScalesGetter, gr schema.GroupResource, resourceName string, namespace string, desiredReplicas int32) wait.ConditionFunc {
186-
return func() (bool, error) {
187-
actualScale, err := sClient.Scales(namespace).Get(context.TODO(), gr, resourceName, metav1.GetOptions{})
186+
func scaleHasDesiredReplicas(sClient scaleclient.ScalesGetter, gr schema.GroupResource, resourceName string, namespace string, desiredReplicas int32) wait.ConditionWithContextFunc {
187+
return func(ctx context.Context) (bool, error) {
188+
actualScale, err := sClient.Scales(namespace).Get(ctx, gr, resourceName, metav1.GetOptions{})
188189
if err != nil {
189190
return false, err
190191
}
@@ -203,11 +204,9 @@ func WaitForScaleHasDesiredReplicas(sClient scaleclient.ScalesGetter, gr schema.
203204
if waitForReplicas == nil {
204205
return fmt.Errorf("waitForReplicas parameter cannot be nil")
205206
}
206-
err := wait.PollImmediate(
207-
waitForReplicas.Interval,
208-
waitForReplicas.Timeout,
209-
scaleHasDesiredReplicas(sClient, gr, resourceName, namespace, int32(newSize)))
210-
if err == wait.ErrWaitTimeout {
207+
err := wait.PollUntilContextTimeout(context.Background(), waitForReplicas.Interval, waitForReplicas.Timeout, true, scaleHasDesiredReplicas(sClient, gr, resourceName, namespace, int32(newSize)))
208+
209+
if errors.Is(err, context.DeadlineExceeded) {
211210
return fmt.Errorf("timed out waiting for %q to be synced", resourceName)
212211
}
213212
return err

‎staging/src/k8s.io/kubectl/pkg/scale/scale_test.go‎

Lines changed: 13 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ limitations under the License.
1717
package scale
1818

1919
import (
20+
"context"
2021
"encoding/json"
2122
"fmt"
2223
"testing"
@@ -69,7 +70,7 @@ func TestReplicationControllerScaleRetry(t *testing.T) {
6970
namespace := metav1.NamespaceDefault
7071

7172
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, rcgvr, false)
72-
pass, err := scaleFunc()
73+
pass, err := scaleFunc(context.Background())
7374
if pass {
7475
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
7576
}
@@ -78,7 +79,7 @@ func TestReplicationControllerScaleRetry(t *testing.T) {
7879
}
7980
preconditions := ScalePrecondition{3, ""}
8081
scaleFunc = ScaleCondition(scaler, &preconditions, namespace, name, count, nil, rcgvr, false)
81-
_, err = scaleFunc()
82+
_, err = scaleFunc(context.Background())
8283
if err == nil {
8384
t.Errorf("Expected error on precondition failure")
8485
}
@@ -105,7 +106,7 @@ func TestReplicationControllerScaleInvalid(t *testing.T) {
105106
namespace := "default"
106107

107108
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, rcgvr, false)
108-
pass, err := scaleFunc()
109+
pass, err := scaleFunc(context.Background())
109110
if pass {
110111
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
111112
}
@@ -179,7 +180,7 @@ func TestDeploymentScaleRetry(t *testing.T) {
179180
namespace := "default"
180181

181182
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, deploygvr, false)
182-
pass, err := scaleFunc()
183+
pass, err := scaleFunc(context.Background())
183184
if pass != false {
184185
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
185186
}
@@ -188,7 +189,7 @@ func TestDeploymentScaleRetry(t *testing.T) {
188189
}
189190
preconditions := &ScalePrecondition{3, ""}
190191
scaleFunc = ScaleCondition(scaler, preconditions, namespace, name, count, nil, deploygvr, false)
191-
_, err = scaleFunc()
192+
_, err = scaleFunc(context.Background())
192193
if err == nil {
193194
t.Error("Expected error on precondition failure")
194195
}
@@ -236,7 +237,7 @@ func TestDeploymentScaleInvalid(t *testing.T) {
236237
namespace := "default"
237238

238239
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, deploygvr, false)
239-
pass, err := scaleFunc()
240+
pass, err := scaleFunc(context.Background())
240241
if pass {
241242
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
242243
}
@@ -309,7 +310,7 @@ func TestStatefulSetScaleRetry(t *testing.T) {
309310
namespace := "default"
310311

311312
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, stsgvr, false)
312-
pass, err := scaleFunc()
313+
pass, err := scaleFunc(context.Background())
313314
if pass != false {
314315
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
315316
}
@@ -318,7 +319,7 @@ func TestStatefulSetScaleRetry(t *testing.T) {
318319
}
319320
preconditions := &ScalePrecondition{3, ""}
320321
scaleFunc = ScaleCondition(scaler, preconditions, namespace, name, count, nil, stsgvr, false)
321-
_, err = scaleFunc()
322+
_, err = scaleFunc(context.Background())
322323
if err == nil {
323324
t.Error("Expected error on precondition failure")
324325
}
@@ -345,7 +346,7 @@ func TestStatefulSetScaleInvalid(t *testing.T) {
345346
namespace := "default"
346347

347348
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, stsgvr, false)
348-
pass, err := scaleFunc()
349+
pass, err := scaleFunc(context.Background())
349350
if pass {
350351
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
351352
}
@@ -418,7 +419,7 @@ func TestReplicaSetScaleRetry(t *testing.T) {
418419
namespace := "default"
419420

420421
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, rsgvr, false)
421-
pass, err := scaleFunc()
422+
pass, err := scaleFunc(context.Background())
422423
if pass != false {
423424
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
424425
}
@@ -427,7 +428,7 @@ func TestReplicaSetScaleRetry(t *testing.T) {
427428
}
428429
preconditions := &ScalePrecondition{3, ""}
429430
scaleFunc = ScaleCondition(scaler, preconditions, namespace, name, count, nil, rsgvr, false)
430-
_, err = scaleFunc()
431+
_, err = scaleFunc(context.Background())
431432
if err == nil {
432433
t.Error("Expected error on precondition failure")
433434
}
@@ -454,7 +455,7 @@ func TestReplicaSetScaleInvalid(t *testing.T) {
454455
namespace := "default"
455456

456457
scaleFunc := ScaleCondition(scaler, nil, namespace, name, count, nil, rsgvr, false)
457-
pass, err := scaleFunc()
458+
pass, err := scaleFunc(context.Background())
458459
if pass {
459460
t.Errorf("Expected an update failure to return pass = false, got pass = %v", pass)
460461
}

‎test/utils/update_resources.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ limitations under the License.
1717
package utils
1818

1919
import (
20+
"context"
2021
"fmt"
2122
"time"
2223

@@ -35,9 +36,9 @@ const (
3536
waitRetryTimeout = 5 * time.Minute
3637
)
3738

38-
func RetryErrorCondition(condition wait.ConditionFunc) wait.ConditionFunc {
39-
return func() (bool, error) {
40-
done, err := condition()
39+
func RetryErrorCondition(condition wait.ConditionWithContextFunc) wait.ConditionWithContextFunc {
40+
return func(ctx context.Context) (bool, error) {
41+
done, err := condition(ctx)
4142
return done, err
4243
}
4344
}
@@ -50,7 +51,7 @@ func ScaleResourceWithRetries(scalesGetter scaleclient.ScalesGetter, namespace,
5051
}
5152
waitForReplicas := scale.NewRetryParams(waitRetryInterval, waitRetryTimeout)
5253
cond := RetryErrorCondition(scale.ScaleCondition(scaler, preconditions, namespace, name, size, nil, gvr, false))
53-
err := wait.PollImmediate(updateRetryInterval, updateRetryTimeout, cond)
54+
err := wait.PollUntilContextTimeout(context.Background(), updateRetryInterval, updateRetryTimeout, true, cond)
5455
if err == nil {
5556
err = scale.WaitForScaleHasDesiredReplicas(scalesGetter, gvr.GroupResource(), name, namespace, size, waitForReplicas)
5657
}

0 commit comments

Comments
 (0)