fix(httpclient): replace DefaultClient with timed clients and retry
Пакет httpclient: timeout 45s, idle pool, DoWithRetry. Scheduler и nodecli используют retry; pipeline/asnresolve/jobs — httpclient.New вместо DefaultClient. Co-authored-by: Cursor <[email protected]>
This commit is contained in:
@@ -13,6 +13,8 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
)
|
)
|
||||||
|
|
||||||
// DefaultRIPEStatURL is the RIPEstat announced-prefixes data call (no API key).
|
// DefaultRIPEStatURL is the RIPEstat announced-prefixes data call (no API key).
|
||||||
@@ -24,7 +26,7 @@ const DefaultASOverviewURL = "https://stat.ripe.net/data/as-overview/data.json"
|
|||||||
// AnnouncedPrefixes returns currently announced IPv4/IPv6 prefixes for the ASN (best-effort via RIPEstat).
|
// AnnouncedPrefixes returns currently announced IPv4/IPv6 prefixes for the ASN (best-effort via RIPEstat).
|
||||||
func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip.Prefix, error) {
|
func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip.Prefix, error) {
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_ANNOUNCED_PREFIXES_URL"))
|
base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_ANNOUNCED_PREFIXES_URL"))
|
||||||
if base == "" {
|
if base == "" {
|
||||||
@@ -86,7 +88,7 @@ func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip
|
|||||||
// ASHolderName returns the holder / organization label for the ASN from RIPEstat as-overview (best-effort).
|
// ASHolderName returns the holder / organization label for the ASN from RIPEstat as-overview (best-effort).
|
||||||
func ASHolderName(ctx context.Context, hc *http.Client, asn int64) (string, error) {
|
func ASHolderName(ctx context.Context, hc *http.Client, asn int64) (string, error) {
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_AS_OVERVIEW_URL"))
|
base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_AS_OVERVIEW_URL"))
|
||||||
if base == "" {
|
if base == "" {
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"evobgp/internal/db"
|
"evobgp/internal/db"
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/jobs"
|
"evobgp/internal/jobs"
|
||||||
"evobgp/internal/observability"
|
"evobgp/internal/observability"
|
||||||
"evobgp/internal/repository"
|
"evobgp/internal/repository"
|
||||||
@@ -17,7 +18,7 @@ import (
|
|||||||
|
|
||||||
// NewCDNHTTPClient returns the shared HTTP client for CDN and preview fetches (PERF-02 / ERR-03).
|
// NewCDNHTTPClient returns the shared HTTP client for CDN and preview fetches (PERF-02 / ERR-03).
|
||||||
func NewCDNHTTPClient() *http.Client {
|
func NewCDNHTTPClient() *http.Client {
|
||||||
return &http.Client{Timeout: 45 * time.Second}
|
return httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
|
|
||||||
// BootstrapWorkers opens the same store.Backend and jobs.Registry as New (without HTTP or bundle keys).
|
// BootstrapWorkers opens the same store.Backend and jobs.Registry as New (without HTTP or bundle keys).
|
||||||
|
|||||||
@@ -0,0 +1,68 @@
|
|||||||
|
// Package httpclient provides shared HTTP clients and retry helpers for outbound calls.
|
||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
const DefaultTimeout = 45 * time.Second
|
||||||
|
|
||||||
|
// New returns an HTTP client with timeout and tuned idle connection pooling.
|
||||||
|
func New(timeout time.Duration) *http.Client {
|
||||||
|
if timeout <= 0 {
|
||||||
|
timeout = DefaultTimeout
|
||||||
|
}
|
||||||
|
tr := http.DefaultTransport.(*http.Transport).Clone()
|
||||||
|
tr.MaxIdleConns = 100
|
||||||
|
tr.MaxIdleConnsPerHost = 10
|
||||||
|
return &http.Client{Timeout: timeout, Transport: tr}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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
|
||||||
|
}
|
||||||
|
var lastErr error
|
||||||
|
for attempt := 0; attempt < maxAttempts; attempt++ {
|
||||||
|
if attempt > 0 {
|
||||||
|
wait := time.Duration(attempt) * 2 * time.Second
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil, ctx.Err()
|
||||||
|
case <-time.After(wait):
|
||||||
|
}
|
||||||
|
if req.GetBody != nil {
|
||||||
|
body, err := req.GetBody()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
req.Body = body
|
||||||
|
}
|
||||||
|
}
|
||||||
|
reqClone := req.Clone(ctx)
|
||||||
|
resp, err := hc.Do(reqClone)
|
||||||
|
if err != nil {
|
||||||
|
lastErr = err
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if resp.StatusCode >= 500 {
|
||||||
|
_, _ = io.Copy(io.Discard, resp.Body)
|
||||||
|
_ = resp.Body.Close()
|
||||||
|
lastErr = fmt.Errorf("httpclient: upstream %s", resp.Status)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return resp, nil
|
||||||
|
}
|
||||||
|
if lastErr != nil {
|
||||||
|
return nil, lastErr
|
||||||
|
}
|
||||||
|
return nil, fmt.Errorf("httpclient: request failed after %d attempts", maxAttempts)
|
||||||
|
}
|
||||||
@@ -0,0 +1,39 @@
|
|||||||
|
package httpclient
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestDoWithRetry_retriesOn500(t *testing.T) {
|
||||||
|
var calls int
|
||||||
|
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
calls++
|
||||||
|
if calls < 3 {
|
||||||
|
http.Error(w, "fail", http.StatusBadGateway)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write([]byte("ok"))
|
||||||
|
}))
|
||||||
|
defer srv.Close()
|
||||||
|
|
||||||
|
req, err := http.NewRequest(http.MethodGet, srv.URL, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
resp, err := DoWithRetry(context.Background(), New(5*time.Second), req, 3)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer func() { _ = resp.Body.Close() }()
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("status %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
if calls != 3 {
|
||||||
|
t.Fatalf("want 3 calls, got %d", calls)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
|
|
||||||
"evobgp/internal/birddeploy"
|
"evobgp/internal/birddeploy"
|
||||||
"evobgp/internal/birdfmt"
|
"evobgp/internal/birdfmt"
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/nodedispatch"
|
"evobgp/internal/nodedispatch"
|
||||||
"evobgp/internal/observability"
|
"evobgp/internal/observability"
|
||||||
"evobgp/internal/pipeline"
|
"evobgp/internal/pipeline"
|
||||||
@@ -70,7 +71,7 @@ type revisionLogEntry struct {
|
|||||||
Message string `json:"message"`
|
Message string `json:"message"`
|
||||||
}
|
}
|
||||||
|
|
||||||
var defaultWorkerHTTP = &http.Client{Timeout: 45 * time.Second}
|
var defaultWorkerHTTP = httpclient.New(httpclient.DefaultTimeout)
|
||||||
|
|
||||||
func (w *Worker) httpClient() *http.Client {
|
func (w *Worker) httpClient() *http.Client {
|
||||||
if w != nil && w.HTTPClient != nil {
|
if w != nil && w.HTTPClient != nil {
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import (
|
|||||||
|
|
||||||
"evobgp/internal/birdfmt"
|
"evobgp/internal/birdfmt"
|
||||||
"evobgp/internal/bundle"
|
"evobgp/internal/bundle"
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/signing"
|
"evobgp/internal/signing"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -55,6 +56,10 @@ func CmdPullBundle(args []string) int {
|
|||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func nodeHTTPClient() *http.Client {
|
||||||
|
return httpclient.New(60 * time.Second)
|
||||||
|
}
|
||||||
|
|
||||||
func fetchLatestRevision(base, token, speaker string) (string, error) {
|
func fetchLatestRevision(base, token, speaker string) (string, error) {
|
||||||
u := strings.TrimRight(base, "/") + "/v1/speakers/" + speaker + "/revisions/latest"
|
u := strings.TrimRight(base, "/") + "/v1/speakers/" + speaker + "/revisions/latest"
|
||||||
req, err := http.NewRequest(http.MethodGet, u, nil)
|
req, err := http.NewRequest(http.MethodGet, u, nil)
|
||||||
@@ -62,7 +67,9 @@ func fetchLatestRevision(base, token, speaker string) (string, error) {
|
|||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
req.Header.Set("Authorization", "Bearer "+token)
|
req.Header.Set("Authorization", "Bearer "+token)
|
||||||
resp, err := http.DefaultClient.Do(req)
|
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
resp, err := httpclient.DoWithRetry(ctx, nodeHTTPClient(), req, 3)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
}
|
}
|
||||||
@@ -90,7 +97,9 @@ func fetchBundle(base, token, speaker, revision string) ([]byte, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
req.Header.Set("Authorization", "Bearer "+token)
|
req.Header.Set("Authorization", "Bearer "+token)
|
||||||
resp, err := http.DefaultClient.Do(req)
|
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
resp, err := httpclient.DoWithRetry(ctx, nodeHTTPClient(), req, 3)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,13 +7,14 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/store"
|
"evobgp/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
// PrefetchCDNSourceETags performs conditional GETs for CDN sources; on 200 parses CIDRs into module_prefix_snapshot.
|
// PrefetchCDNSourceETags performs conditional GETs for CDN sources; on 200 parses CIDRs into module_prefix_snapshot.
|
||||||
func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Client) error {
|
func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Client) error {
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
tenants, err := st.ListTenantIDs()
|
tenants, err := st.ListTenantIDs()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"evobgp/internal/birdfmt"
|
"evobgp/internal/birdfmt"
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/observability"
|
"evobgp/internal/observability"
|
||||||
"evobgp/internal/store"
|
"evobgp/internal/store"
|
||||||
|
|
||||||
@@ -50,7 +51,7 @@ func MaterializedASPrefixKey(asn int64) string {
|
|||||||
// It does not create a new config revision.
|
// It does not create a new config revision.
|
||||||
func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) error {
|
func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) error {
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
mod, err := st.GetModule(tenantID, moduleID)
|
mod, err := st.GetModule(tenantID, moduleID)
|
||||||
@@ -84,7 +85,7 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client,
|
|||||||
// If materialized prefixes are unchanged, returns latest revision id without creating a duplicate.
|
// If materialized prefixes are unchanged, returns latest revision id without creating a duplicate.
|
||||||
func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string) (revisionID string, err error) {
|
func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string) (revisionID string, err error) {
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
agg, err := aggregateTenantPrefixRowsAll(ctx, st, hc, tenantID)
|
agg, err := aggregateTenantPrefixRowsAll(ctx, st, hc, tenantID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -117,7 +118,7 @@ func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client
|
|||||||
func RenderTenantRevisionFromPrefixes(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string, rows []store.PrefixRow) (revisionID string, err error) {
|
func RenderTenantRevisionFromPrefixes(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string, rows []store.PrefixRow) (revisionID string, err error) {
|
||||||
_ = ctx
|
_ = ctx
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
agg := append([]store.PrefixRow(nil), rows...)
|
agg := append([]store.PrefixRow(nil), rows...)
|
||||||
rawCount := len(agg)
|
rawCount := len(agg)
|
||||||
|
|||||||
@@ -7,13 +7,14 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/store"
|
"evobgp/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
// RefreshTenantModules ingests all listed modules in parallel and updates per-module snapshots.
|
// RefreshTenantModules ingests all listed modules in parallel and updates per-module snapshots.
|
||||||
func RefreshTenantModules(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, moduleIDs []string) error {
|
func RefreshTenantModules(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, moduleIDs []string) error {
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
var ids []string
|
var ids []string
|
||||||
seen := make(map[string]struct{})
|
seen := make(map[string]struct{})
|
||||||
|
|||||||
@@ -14,6 +14,7 @@ import (
|
|||||||
|
|
||||||
"evobgp/internal/broker"
|
"evobgp/internal/broker"
|
||||||
"evobgp/internal/config"
|
"evobgp/internal/config"
|
||||||
|
"evobgp/internal/httpclient"
|
||||||
"evobgp/internal/jobs"
|
"evobgp/internal/jobs"
|
||||||
"evobgp/internal/pipeline"
|
"evobgp/internal/pipeline"
|
||||||
"evobgp/internal/store"
|
"evobgp/internal/store"
|
||||||
@@ -108,6 +109,9 @@ func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idem
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
req.GetBody = func() (io.ReadCloser, error) {
|
||||||
|
return io.NopCloser(bytes.NewReader(body)), nil
|
||||||
|
}
|
||||||
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(deps.APIToken))
|
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(deps.APIToken))
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
if idempotencyKey != "" {
|
if idempotencyKey != "" {
|
||||||
@@ -115,9 +119,9 @@ func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idem
|
|||||||
}
|
}
|
||||||
hc := deps.HTTP
|
hc := deps.HTTP
|
||||||
if hc == nil {
|
if hc == nil {
|
||||||
hc = http.DefaultClient
|
hc = httpclient.New(httpclient.DefaultTimeout)
|
||||||
}
|
}
|
||||||
resp, err := hc.Do(req)
|
resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user