|
1 | 1 | package core |
2 | 2 |
|
3 | 3 | import ( |
| 4 | + "bytes" |
4 | 5 | "fmt" |
| 6 | + "io" |
| 7 | + "math/rand/v2" |
5 | 8 | "net/http" |
| 9 | + "strconv" |
| 10 | + "time" |
6 | 11 |
|
7 | 12 | "github.com/stackitcloud/stackit-sdk-go/core/config" |
8 | 13 | alb "github.com/stackitcloud/stackit-sdk-go/services/alb/v2api" |
@@ -313,9 +318,116 @@ func (f *DefaultClientFactory) newMongoDbFlexV2Client() (mongodbflex.DefaultAPI, |
313 | 318 | return apiClient.DefaultAPI, nil |
314 | 319 | } |
315 | 320 |
|
| 321 | +// RetryTransport wraps an underlying RoundTripper to handle HTTP 429s with jitter. |
| 322 | +type RetryTransport struct { |
| 323 | + Base http.RoundTripper |
| 324 | + MaxRetries int |
| 325 | + BaseBackoff time.Duration |
| 326 | + MaxJitter time.Duration |
| 327 | +} |
| 328 | + |
| 329 | +func (t *RetryTransport) RoundTrip(req *http.Request) (*http.Response, error) { |
| 330 | + base := t.Base |
| 331 | + if base == nil { |
| 332 | + base = http.DefaultTransport |
| 333 | + } |
| 334 | + |
| 335 | + // Preserve request body for retries if present |
| 336 | + var bodyBytes []byte |
| 337 | + if req.Body != nil && req.Body != http.NoBody { |
| 338 | + var err error |
| 339 | + bodyBytes, err = io.ReadAll(req.Body) |
| 340 | + if err != nil { |
| 341 | + return nil, err |
| 342 | + } |
| 343 | + |
| 344 | + err = req.Body.Close() |
| 345 | + if err != nil { |
| 346 | + return nil, err |
| 347 | + } |
| 348 | + } |
| 349 | + |
| 350 | + var resp *http.Response |
| 351 | + var err error |
| 352 | + |
| 353 | + for attempt := 0; attempt <= t.MaxRetries; attempt++ { |
| 354 | + // Re-hydrate the request body on each attempt |
| 355 | + if bodyBytes != nil { |
| 356 | + req.Body = io.NopCloser(bytes.NewReader(bodyBytes)) |
| 357 | + } |
| 358 | + |
| 359 | + resp, err = base.RoundTrip(req) |
| 360 | + |
| 361 | + // If success or non-429 error, return immediately |
| 362 | + if err != nil || resp.StatusCode != http.StatusTooManyRequests { |
| 363 | + return resp, err |
| 364 | + } |
| 365 | + |
| 366 | + // Stop if max retries reached |
| 367 | + if attempt == t.MaxRetries { |
| 368 | + break |
| 369 | + } |
| 370 | + |
| 371 | + // Calculate base sleep duration (Retry-After or Exponential Backoff) |
| 372 | + wait := t.getWaitDuration(resp, attempt) |
| 373 | + |
| 374 | + // Always add random jitter regardless of Retry-After header presence |
| 375 | + jitter := time.Duration(rand.Int64N(int64(t.MaxJitter))) //nolint:gosec // only used for jitter |
| 376 | + totalWait := wait + jitter |
| 377 | + |
| 378 | + // Drain and close response body before retrying to reuse TCP connections |
| 379 | + _, err = io.Copy(io.Discard, resp.Body) |
| 380 | + if err != nil { |
| 381 | + return nil, err |
| 382 | + } |
| 383 | + |
| 384 | + err = resp.Body.Close() |
| 385 | + if err != nil { |
| 386 | + return nil, err |
| 387 | + } |
| 388 | + |
| 389 | + select { |
| 390 | + case <-req.Context().Done(): |
| 391 | + return nil, req.Context().Err() |
| 392 | + case <-time.After(totalWait): |
| 393 | + } |
| 394 | + } |
| 395 | + |
| 396 | + return resp, err |
| 397 | +} |
| 398 | + |
| 399 | +func (t *RetryTransport) getWaitDuration(resp *http.Response, attempt int) time.Duration { |
| 400 | + if retryAfter := resp.Header.Get("Retry-After"); retryAfter != "" { |
| 401 | + // Try parsing as integer seconds |
| 402 | + if seconds, err := strconv.Atoi(retryAfter); err == nil { |
| 403 | + return time.Duration(seconds) * time.Second |
| 404 | + } |
| 405 | + // Try parsing as HTTP-Date string |
| 406 | + if date, err := http.ParseTime(retryAfter); err == nil { |
| 407 | + if d := time.Until(date); d > 0 { |
| 408 | + return d |
| 409 | + } |
| 410 | + } |
| 411 | + } |
| 412 | + |
| 413 | + // Fallback to exponential backoff |
| 414 | + return t.BaseBackoff * (1 << attempt) |
| 415 | +} |
| 416 | + |
316 | 417 | func (f *DefaultClientFactory) newObjectStorageV2Client() (objectstorage.DefaultAPI, error) { |
317 | 418 | apiClientConfigOptions := f.defaultConfigOptions(f.CustomEndpoints.ObjectStorageCustomEndpoint) |
318 | 419 |
|
| 420 | + mdlw := func(rt http.RoundTripper) http.RoundTripper { |
| 421 | + return &RetryTransport{ |
| 422 | + Base: rt, |
| 423 | + MaxRetries: 3, |
| 424 | + BaseBackoff: 1 * time.Second, |
| 425 | + MaxJitter: 500 * time.Millisecond, // Always added to wait time |
| 426 | + } |
| 427 | + } |
| 428 | + |
| 429 | + apiClientConfigOptions = append(apiClientConfigOptions, config.WithMiddleware(mdlw)) |
| 430 | + |
319 | 431 | apiClient, err := objectstorage.NewAPIClient(apiClientConfigOptions...) |
320 | 432 | if err != nil { |
321 | 433 | return nil, fmt.Errorf("configuring client: %w. This is an error related to the provider configuration, not to the resource configuration", err) |
|
0 commit comments