From 2289107911d5ee28d03650fa62942fcaae2bed1a Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 25 May 2026 10:15:32 +0700 Subject: [PATCH] feat(httpclient): add circuit breaker for CDN and RIPEstat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-host circuit breaker с retry для CDN fetch и RIPEstat; порог 5 ошибок, cooldown 30s. Co-authored-by: Cursor --- internal/asnresolve/ripestat.go | 4 +-- internal/httpclient/circuit.go | 55 +++++++++++++++++++++++++++++ internal/httpclient/circuit_test.go | 34 ++++++++++++++++++ internal/httpclient/httpclient.go | 25 +++++++++++-- internal/pipeline/cdn_snapshot.go | 8 ++--- internal/pipeline/upstream_http.go | 23 ++++++++++++ 6 files changed, 140 insertions(+), 9 deletions(-) create mode 100644 internal/httpclient/circuit.go create mode 100644 internal/httpclient/circuit_test.go create mode 100644 internal/pipeline/upstream_http.go diff --git a/internal/asnresolve/ripestat.go b/internal/asnresolve/ripestat.go index f7566a4..9debe2f 100644 --- a/internal/asnresolve/ripestat.go +++ b/internal/asnresolve/ripestat.go @@ -40,7 +40,7 @@ func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip req.Header.Set("Accept", "application/json") 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 { 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("User-Agent", "evobgp-asnresolve/1.0") - resp, err := hc.Do(req) + resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3) if err != nil { return "", fmt.Errorf("ripestat as-overview AS%d: %w", asn, err) } diff --git a/internal/httpclient/circuit.go b/internal/httpclient/circuit.go new file mode 100644 index 0000000..949d8ba --- /dev/null +++ b/internal/httpclient/circuit.go @@ -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{} +} diff --git a/internal/httpclient/circuit_test.go b/internal/httpclient/circuit_test.go new file mode 100644 index 0000000..c0db5eb --- /dev/null +++ b/internal/httpclient/circuit_test.go @@ -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") + } +} diff --git a/internal/httpclient/httpclient.go b/internal/httpclient/httpclient.go index da02a58..1edfa81 100644 --- a/internal/httpclient/httpclient.go +++ b/internal/httpclient/httpclient.go @@ -24,9 +24,6 @@ func New(timeout time.Duration) *http.Client { // 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) { - if hc == nil { - hc = New(0) - } if maxAttempts <= 0 { 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) } + +// 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 +} diff --git a/internal/pipeline/cdn_snapshot.go b/internal/pipeline/cdn_snapshot.go index 5fe731d..61a2774 100644 --- a/internal/pipeline/cdn_snapshot.go +++ b/internal/pipeline/cdn_snapshot.go @@ -133,7 +133,7 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl if etag := strings.TrimSpace(src.Etag); etag != "" { req.Header.Set("If-None-Match", etag) } - resp, err := hc.Do(req) + resp, err := upstreamHTTPDo(ctx, hc, req) if err != nil { 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 { return nil, err } - resp, err = hc.Do(req2) + resp, err = upstreamHTTPDo(ctx, hc, req2) if err != nil { 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 != "" { req.Header.Set("If-None-Match", etag) } - resp, err := hc.Do(req) + resp, err := upstreamHTTPDo(ctx, hc, req) if err != nil { 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 { return nil, err } - resp, err = hc.Do(req2) + resp, err = upstreamHTTPDo(ctx, hc, req2) if err != nil { return nil, fmt.Errorf("cdn fetch %s: %w", u, err) } diff --git a/internal/pipeline/upstream_http.go b/internal/pipeline/upstream_http.go new file mode 100644 index 0000000..2345487 --- /dev/null +++ b/internal/pipeline/upstream_http.go @@ -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 +}