Skip to content

Commit 2f36d16

Browse files
committed
feat(delivery): add maximum backoff duration
Signed-off-by: kahirokunn <okinakahiro@gmail.com>
1 parent 72ec4f4 commit 2f36d16

17 files changed

Lines changed: 668 additions & 5 deletions

File tree

config/core/configmaps/features.yaml

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,11 @@ data:
3131
# For more details: https://github.com/knative/eventing/issues/5811
3232
delivery-retryafter: "disabled"
3333

34+
# ALPHA feature: The delivery-backoff-max flag allows DeliverySpec.backoffMax
35+
# to cap the delay calculated from backoffDelay and backoffPolicy.
36+
# For more details: https://github.com/knative/eventing/issues/9278
37+
delivery-backoff-max: "disabled"
38+
3439
# BETA feature: The delivery-timeout allows you to use the Timeout field in DeliverySpec.
3540
# For more details: https://github.com/knative/eventing/issues/5148
3641
delivery-timeout: "enabled"

docs/eventing-api.md

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -484,6 +484,27 @@ For exponential policy, backoff delay is backoffDelay*2^<numberOfRetries>.</p>
484484
</tr>
485485
<tr>
486486
<td>
487+
<code>backoffMax</code><br/>
488+
<em>
489+
string
490+
</em>
491+
</td>
492+
<td>
493+
<em>(Optional)</em>
494+
<p>BackoffMax is the maximum delay between normal delivery attempts. It caps
495+
the delay calculated from BackoffDelay and BackoffPolicy, but does not cap
496+
delays requested by a Retry-After response header. The value must be
497+
greater than zero.</p>
498+
<p>Note: This API is EXPERIMENTAL and might be changed at any time. Cluster
499+
operators must enable the delivery-backoff-max feature before users can set
500+
this field.</p>
501+
<p>More information on Duration format:
502+
- <a href="https://www.iso.org/iso-8601-date-and-time-format.html">https://www.iso.org/iso-8601-date-and-time-format.html</a>
503+
- <a href="https://en.wikipedia.org/wiki/ISO_8601">https://en.wikipedia.org/wiki/ISO_8601</a></p>
504+
</td>
505+
</tr>
506+
<tr>
507+
<td>
487508
<code>retryAfterMax</code><br/>
488509
<em>
489510
string

pkg/apis/duck/v1/delivery_types.go

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,22 @@ type DeliverySpec struct {
6262
// +optional
6363
BackoffDelay *string `json:"backoffDelay,omitempty"`
6464

65+
// BackoffMax is the maximum delay between normal delivery attempts. It caps
66+
// the delay calculated from BackoffDelay and BackoffPolicy, but does not cap
67+
// delays requested by a Retry-After response header. The value must be
68+
// greater than zero.
69+
//
70+
// Note: This API is EXPERIMENTAL and might be changed at any time. Cluster
71+
// operators must enable the delivery-backoff-max feature before users can set
72+
// this field.
73+
//
74+
// More information on Duration format:
75+
// - https://www.iso.org/iso-8601-date-and-time-format.html
76+
// - https://en.wikipedia.org/wiki/ISO_8601
77+
//
78+
// +optional
79+
BackoffMax *string `json:"backoffMax,omitempty"`
80+
6581
// RetryAfterMax provides an optional upper bound on the duration specified in a "Retry-After" header
6682
// when calculating backoff times for retrying 429 and 503 response codes. Setting the value to
6783
// zero ("PT0S") can be used to opt-out of respecting "Retry-After" header values altogether. This
@@ -131,6 +147,17 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
131147
}
132148
}
133149

150+
if ds.BackoffMax != nil {
151+
if feature.FromContext(ctx).IsEnabled(feature.DeliveryBackoffMax) {
152+
p, pe := period.Parse(*ds.BackoffMax)
153+
if pe != nil || p.IsZero() || p.IsNegative() {
154+
errs = errs.Also(apis.ErrInvalidValue(*ds.BackoffMax, "backoffMax"))
155+
}
156+
} else {
157+
errs = errs.Also(apis.ErrDisallowedFields("backoffMax"))
158+
}
159+
}
160+
134161
if ds.Format != nil {
135162
switch *ds.Format {
136163
case DeliveryFormatBinary, DeliveryFormatJson:

pkg/apis/duck/v1/delivery_types_test.go

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,9 @@ func TestDeliverySpecValidation(t *testing.T) {
3535
deliveryRetryAfterEnabledCtx := feature.ToContext(context.TODO(), feature.Flags{
3636
feature.DeliveryRetryAfter: feature.Enabled,
3737
})
38+
deliveryBackoffMaxEnabledCtx := feature.ToContext(context.TODO(), feature.Flags{
39+
feature.DeliveryBackoffMax: feature.Enabled,
40+
})
3841

3942
invalidString := "invalid time"
4043
bop := BackoffPolicyExponential
@@ -142,6 +145,30 @@ func TestDeliverySpecValidation(t *testing.T) {
142145
want: func() *apis.FieldError {
143146
return apis.ErrDisallowedFields("retryAfterMax")
144147
}(),
148+
}, {
149+
name: "valid backoffMax",
150+
ctx: deliveryBackoffMaxEnabledCtx,
151+
spec: &DeliverySpec{BackoffMax: &validDuration},
152+
want: nil,
153+
}, {
154+
name: "zero backoffMax",
155+
ctx: deliveryBackoffMaxEnabledCtx,
156+
spec: &DeliverySpec{BackoffMax: pointer.String("PT0S")},
157+
want: apis.ErrInvalidValue("PT0S", "backoffMax"),
158+
}, {
159+
name: "negative backoffMax",
160+
ctx: deliveryBackoffMaxEnabledCtx,
161+
spec: &DeliverySpec{BackoffMax: pointer.String("-PT1S")},
162+
want: apis.ErrInvalidValue("-PT1S", "backoffMax"),
163+
}, {
164+
name: "invalid backoffMax",
165+
ctx: deliveryBackoffMaxEnabledCtx,
166+
spec: &DeliverySpec{BackoffMax: &invalidDuration},
167+
want: apis.ErrInvalidValue(invalidDuration, "backoffMax"),
168+
}, {
169+
name: "disabled feature with backoffMax",
170+
spec: &DeliverySpec{BackoffMax: &validDuration},
171+
want: apis.ErrDisallowedFields("backoffMax"),
145172
},
146173
{
147174
name: "valid format JSON",

pkg/apis/duck/v1/zz_generated.deepcopy.go

Lines changed: 5 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

pkg/apis/feature/features.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ func newDefaults() Flags {
8181
return map[string]Flag{
8282
KReferenceGroup: Disabled,
8383
DeliveryRetryAfter: Disabled,
84+
DeliveryBackoffMax: Disabled,
8485
DeliveryTimeout: Enabled,
8586
KReferenceMapping: Disabled,
8687
TransportEncryption: Disabled,

pkg/apis/feature/flag_names.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package feature
1919
const (
2020
KReferenceGroup = "kreference-group"
2121
DeliveryRetryAfter = "delivery-retryafter"
22+
DeliveryBackoffMax = "delivery-backoff-max"
2223
DeliveryTimeout = "delivery-timeout"
2324
KReferenceMapping = "kreference-mapping"
2425
TransportEncryption = "transport-encryption"

pkg/kncloudevents/retries.go

Lines changed: 68 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ package kncloudevents
1919
import (
2020
"context"
2121
"fmt"
22-
"math"
2322
"net/http"
2423
"strconv"
2524
"time"
@@ -58,11 +57,12 @@ type Backoff func(attemptNum int, resp *http.Response) time.Duration
5857
type RetryConfig struct {
5958
// Maximum number of retries
6059
RetryMax int
61-
// These next two variables are just copied from the original DeliverySpec so
60+
// These next three variables are just copied from the original DeliverySpec so
6261
// we can detect if anything has changed. We can not do that with the CheckRetry
6362
// Backoff (at least not easily).
6463
BackoffDelay *string
6564
BackoffPolicy *v1.BackoffPolicyType
65+
BackoffMax *string
6666

6767
CheckRetry CheckRetry
6868
Backoff Backoff
@@ -92,6 +92,20 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
9292
}
9393
retryConfig.BackoffPolicy = spec.BackoffPolicy
9494
retryConfig.BackoffDelay = spec.BackoffDelay
95+
retryConfig.BackoffMax = spec.BackoffMax
96+
97+
var backoffMaxDuration *time.Duration
98+
if spec.BackoffMax != nil {
99+
maxPeriod, err := period.Parse(*spec.BackoffMax)
100+
if err != nil || maxPeriod.IsZero() || maxPeriod.IsNegative() {
101+
if err != nil {
102+
return retryConfig, fmt.Errorf("failed to parse Spec.BackoffMax: %w", err)
103+
}
104+
return retryConfig, fmt.Errorf("Spec.BackoffMax must be greater than zero")
105+
}
106+
maxDelay := saturatingPeriodDuration(maxPeriod)
107+
backoffMaxDuration = &maxDelay
108+
}
95109

96110
if spec.BackoffPolicy != nil && spec.BackoffDelay != nil {
97111

@@ -100,15 +114,15 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
100114
return retryConfig, fmt.Errorf("failed to parse Spec.BackoffDelay: %w", err)
101115
}
102116

103-
delayDuration, _ := delay.Duration()
117+
delayDuration := saturatingPeriodDuration(delay)
104118
switch *spec.BackoffPolicy {
105119
case v1.BackoffPolicyExponential:
106120
retryConfig.Backoff = func(attemptNum int, resp *http.Response) time.Duration {
107-
return delayDuration * time.Duration(math.Exp2(float64(attemptNum)))
121+
return exponentialBackoff(delayDuration, attemptNum, backoffMaxDuration)
108122
}
109123
case v1.BackoffPolicyLinear:
110124
retryConfig.Backoff = func(attemptNum int, resp *http.Response) time.Duration {
111-
return delayDuration * time.Duration(attemptNum)
125+
return linearBackoff(delayDuration, attemptNum, backoffMaxDuration)
112126
}
113127
}
114128
}
@@ -133,6 +147,55 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
133147
return retryConfig, nil
134148
}
135149

150+
const maxBackoffDuration = time.Duration(1<<63 - 1)
151+
152+
func saturatingPeriodDuration(p period.Period) time.Duration {
153+
if p.IsPositive() && p.TotalDaysApprox() > int(maxBackoffDuration/(24*time.Hour)) {
154+
return maxBackoffDuration
155+
}
156+
157+
d, _ := p.Duration()
158+
if p.IsPositive() && d < 0 {
159+
return maxBackoffDuration
160+
}
161+
return d
162+
}
163+
164+
func linearBackoff(delay time.Duration, attemptNum int, configuredMax *time.Duration) time.Duration {
165+
if attemptNum <= 0 || delay <= 0 {
166+
return 0
167+
}
168+
limit := maxBackoffDuration
169+
if configuredMax != nil {
170+
limit = *configuredMax
171+
}
172+
if delay >= limit || time.Duration(attemptNum) > limit/delay {
173+
return limit
174+
}
175+
return delay * time.Duration(attemptNum)
176+
}
177+
178+
func exponentialBackoff(delay time.Duration, attemptNum int, configuredMax *time.Duration) time.Duration {
179+
if attemptNum < 0 || delay <= 0 {
180+
return 0
181+
}
182+
limit := maxBackoffDuration
183+
if configuredMax != nil {
184+
limit = *configuredMax
185+
}
186+
if delay >= limit {
187+
return limit
188+
}
189+
result := delay
190+
for range attemptNum {
191+
if result > limit/2 {
192+
return limit
193+
}
194+
result *= 2
195+
}
196+
return result
197+
}
198+
136199
// SelectiveRetry is an alternative function to determine whether to retry based on response
137200
//
138201
// Note - Returning true indicates a retry should occur. Returning an error will result in that

0 commit comments

Comments
 (0)