feat(httpclient): add circuit breaker for CDN and RIPEstat
Per-host circuit breaker с retry для CDN fetch и RIPEstat; порог 5 ошибок, cooldown 30s. Co-authored-by: Cursor <[email protected]>
This commit is contained in:
@@ -40,7 +40,7 @@ func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip
|
|||||||
req.Header.Set("Accept", "application/json")
|
req.Header.Set("Accept", "application/json")
|
||||||
req.Header.Set("User-Agent", "evobgp-asnresolve/1.0")
|
req.Header.Set("User-Agent", "evobgp-asnresolve/1.0")
|
||||||
|
|
||||||
resp, err := hc.Do(req)
|
resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("ripestat fetch AS%d: %w", asn, err)
|
return nil, fmt.Errorf("ripestat fetch AS%d: %w", asn, err)
|
||||||
}
|
}
|
||||||
@@ -102,7 +102,7 @@ func ASHolderName(ctx context.Context, hc *http.Client, asn int64) (string, erro
|
|||||||
req.Header.Set("Accept", "application/json")
|
req.Header.Set("Accept", "application/json")
|
||||||
req.Header.Set("User-Agent", "evobgp-asnresolve/1.0")
|
req.Header.Set("User-Agent", "evobgp-asnresolve/1.0")
|
||||||
|
|
||||||
resp, err := hc.Do(req)
|
resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Errorf("ripestat as-overview AS%d: %w", asn, err)
|
return "", fmt.Errorf("ripestat as-overview AS%d: %w", asn, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,55 @@
|
|||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
defaultBreakerThreshold = 5
|
||||||
|
defaultBreakerCooldown = 30 * time.Second
|
||||||
|
)
|
||||||
|
|
||||||
|
type hostBreaker struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
failures int
|
||||||
|
openUntil time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
var hostBreakers sync.Map // string -> *hostBreaker
|
||||||
|
|
||||||
|
func breakerForHost(host string) *hostBreaker {
|
||||||
|
if host == "" {
|
||||||
|
host = "_"
|
||||||
|
}
|
||||||
|
v, _ := hostBreakers.LoadOrStore(host, &hostBreaker{})
|
||||||
|
return v.(*hostBreaker)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *hostBreaker) allow() bool {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
return time.Now().After(b.openUntil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *hostBreaker) recordSuccess() {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
b.failures = 0
|
||||||
|
b.openUntil = time.Time{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (b *hostBreaker) recordFailure() {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
b.failures++
|
||||||
|
if b.failures >= defaultBreakerThreshold {
|
||||||
|
b.openUntil = time.Now().Add(defaultBreakerCooldown)
|
||||||
|
b.failures = 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResetHostBreakers clears all circuit breakers (tests only).
|
||||||
|
func ResetHostBreakers() {
|
||||||
|
hostBreakers = sync.Map{}
|
||||||
|
}
|
||||||
@@ -0,0 +1,34 @@
|
|||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestDoWithBreaker_opensAfterFailures(t *testing.T) {
|
||||||
|
ResetHostBreakers()
|
||||||
|
var calls atomic.Int32
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
calls.Add(1)
|
||||||
|
http.Error(w, "fail", http.StatusBadGateway)
|
||||||
|
}))
|
||||||
|
defer srv.Close()
|
||||||
|
|
||||||
|
hc := New(5 * time.Second)
|
||||||
|
for i := 0; i < defaultBreakerThreshold*3; i++ {
|
||||||
|
req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
|
||||||
|
_, _ = DoWithBreaker(context.Background(), hc, req, 1)
|
||||||
|
}
|
||||||
|
req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
|
||||||
|
_, err := DoWithBreaker(context.Background(), hc, req, 1)
|
||||||
|
if err == nil || err.Error() == "" {
|
||||||
|
t.Fatal("expected circuit open error")
|
||||||
|
}
|
||||||
|
if got := calls.Load(); got == 0 {
|
||||||
|
t.Fatal("expected at least one upstream call")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -24,9 +24,6 @@ func New(timeout time.Duration) *http.Client {
|
|||||||
|
|
||||||
// DoWithRetry executes hc.Do(req) up to maxAttempts times with linear backoff.
|
// DoWithRetry executes hc.Do(req) up to maxAttempts times with linear backoff.
|
||||||
func DoWithRetry(ctx context.Context, hc *http.Client, req *http.Request, maxAttempts int) (*http.Response, error) {
|
func DoWithRetry(ctx context.Context, hc *http.Client, req *http.Request, maxAttempts int) (*http.Response, error) {
|
||||||
if hc == nil {
|
|
||||||
hc = New(0)
|
|
||||||
}
|
|
||||||
if maxAttempts <= 0 {
|
if maxAttempts <= 0 {
|
||||||
maxAttempts = 3
|
maxAttempts = 3
|
||||||
}
|
}
|
||||||
@@ -66,3 +63,25 @@ func DoWithRetry(ctx context.Context, hc *http.Client, req *http.Request, maxAtt
|
|||||||
}
|
}
|
||||||
return nil, fmt.Errorf("httpclient: request failed after %d attempts", maxAttempts)
|
return nil, fmt.Errorf("httpclient: request failed after %d attempts", maxAttempts)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DoWithBreaker applies per-host circuit breaking then retries transient failures.
|
||||||
|
func DoWithBreaker(ctx context.Context, hc *http.Client, req *http.Request, maxAttempts int) (*http.Response, error) {
|
||||||
|
if req == nil || req.URL == nil {
|
||||||
|
return nil, fmt.Errorf("httpclient: nil request")
|
||||||
|
}
|
||||||
|
br := breakerForHost(req.URL.Hostname())
|
||||||
|
if !br.allow() {
|
||||||
|
return nil, fmt.Errorf("httpclient: circuit open for %s", req.URL.Hostname())
|
||||||
|
}
|
||||||
|
resp, err := DoWithRetry(ctx, hc, req, maxAttempts)
|
||||||
|
if err != nil {
|
||||||
|
br.recordFailure()
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if resp.StatusCode >= 500 {
|
||||||
|
br.recordFailure()
|
||||||
|
return resp, nil
|
||||||
|
}
|
||||||
|
br.recordSuccess()
|
||||||
|
return resp, nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -133,7 +133,7 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
|
|||||||
if etag := strings.TrimSpace(src.Etag); etag != "" {
|
if etag := strings.TrimSpace(src.Etag); etag != "" {
|
||||||
req.Header.Set("If-None-Match", etag)
|
req.Header.Set("If-None-Match", etag)
|
||||||
}
|
}
|
||||||
resp, err := hc.Do(req)
|
resp, err := upstreamHTTPDo(ctx, hc, req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
||||||
}
|
}
|
||||||
@@ -149,7 +149,7 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
resp, err = hc.Do(req2)
|
resp, err = upstreamHTTPDo(ctx, hc, req2)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
||||||
}
|
}
|
||||||
@@ -209,7 +209,7 @@ func fetchCDNSourceRows(ctx context.Context, st store.Backend, hc *http.Client,
|
|||||||
if etag := strings.TrimSpace(src.Etag); etag != "" {
|
if etag := strings.TrimSpace(src.Etag); etag != "" {
|
||||||
req.Header.Set("If-None-Match", etag)
|
req.Header.Set("If-None-Match", etag)
|
||||||
}
|
}
|
||||||
resp, err := hc.Do(req)
|
resp, err := upstreamHTTPDo(ctx, hc, req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
||||||
}
|
}
|
||||||
@@ -224,7 +224,7 @@ func fetchCDNSourceRows(ctx context.Context, st store.Backend, hc *http.Client,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
resp, err = hc.Do(req2)
|
resp, err = upstreamHTTPDo(ctx, hc, req2)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,23 @@
|
|||||||
|
package pipeline
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
|
)
|
||||||
|
|
||||||
|
func upstreamHTTPDo(ctx context.Context, hc *http.Client, req *http.Request) (*http.Response, error) {
|
||||||
|
if hc == nil {
|
||||||
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
|
}
|
||||||
|
resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3)
|
||||||
|
if err != nil {
|
||||||
|
if req.URL != nil {
|
||||||
|
return nil, fmt.Errorf("cdn fetch %s: %w", req.URL.String(), err)
|
||||||
|
}
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return resp, nil
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user