diff --git a/internal/asnresolve/ripestat.go b/internal/asnresolve/ripestat.go index 4baec1f..f7566a4 100644 --- a/internal/asnresolve/ripestat.go +++ b/internal/asnresolve/ripestat.go @@ -13,6 +13,8 @@ import ( "strconv" "strings" "time" + + "evobgp/internal/httpclient" ) // 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). func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip.Prefix, error) { if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_ANNOUNCED_PREFIXES_URL")) 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). func ASHolderName(ctx context.Context, hc *http.Client, asn int64) (string, error) { if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_AS_OVERVIEW_URL")) if base == "" { diff --git a/internal/httpapi/bootstrap.go b/internal/httpapi/bootstrap.go index 6359130..ce6ea1e 100644 --- a/internal/httpapi/bootstrap.go +++ b/internal/httpapi/bootstrap.go @@ -7,6 +7,7 @@ import ( "time" "evobgp/internal/db" + "evobgp/internal/httpclient" "evobgp/internal/jobs" "evobgp/internal/observability" "evobgp/internal/repository" @@ -17,7 +18,7 @@ import ( // NewCDNHTTPClient returns the shared HTTP client for CDN and preview fetches (PERF-02 / ERR-03). 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). diff --git a/internal/httpclient/httpclient.go b/internal/httpclient/httpclient.go new file mode 100644 index 0000000..da02a58 --- /dev/null +++ b/internal/httpclient/httpclient.go @@ -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) +} diff --git a/internal/httpclient/httpclient_test.go b/internal/httpclient/httpclient_test.go new file mode 100644 index 0000000..201a839 --- /dev/null +++ b/internal/httpclient/httpclient_test.go @@ -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) + } +} diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index 38872ba..a580a06 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -13,6 +13,7 @@ import ( "evobgp/internal/birddeploy" "evobgp/internal/birdfmt" + "evobgp/internal/httpclient" "evobgp/internal/nodedispatch" "evobgp/internal/observability" "evobgp/internal/pipeline" @@ -70,7 +71,7 @@ type revisionLogEntry struct { Message string `json:"message"` } -var defaultWorkerHTTP = &http.Client{Timeout: 45 * time.Second} +var defaultWorkerHTTP = httpclient.New(httpclient.DefaultTimeout) func (w *Worker) httpClient() *http.Client { if w != nil && w.HTTPClient != nil { diff --git a/internal/nodecli/commands.go b/internal/nodecli/commands.go index 5d7079e..7f819d2 100644 --- a/internal/nodecli/commands.go +++ b/internal/nodecli/commands.go @@ -16,6 +16,7 @@ import ( "evobgp/internal/birdfmt" "evobgp/internal/bundle" + "evobgp/internal/httpclient" "evobgp/internal/signing" ) @@ -55,6 +56,10 @@ func CmdPullBundle(args []string) int { return 0 } +func nodeHTTPClient() *http.Client { + return httpclient.New(60 * time.Second) +} + func fetchLatestRevision(base, token, speaker string) (string, error) { u := strings.TrimRight(base, "/") + "/v1/speakers/" + speaker + "/revisions/latest" req, err := http.NewRequest(http.MethodGet, u, nil) @@ -62,7 +67,9 @@ func fetchLatestRevision(base, token, speaker string) (string, error) { return "", err } 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 { return "", err } @@ -90,7 +97,9 @@ func fetchBundle(base, token, speaker, revision string) ([]byte, error) { return nil, err } 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 { return nil, err } diff --git a/internal/pipeline/prefetch.go b/internal/pipeline/prefetch.go index d7fbb9d..025069d 100644 --- a/internal/pipeline/prefetch.go +++ b/internal/pipeline/prefetch.go @@ -7,13 +7,14 @@ import ( "strings" "time" + "evobgp/internal/httpclient" "evobgp/internal/store" ) // 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 { if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } tenants, err := st.ListTenantIDs() if err != nil { diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index ced9811..66c7ea9 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -19,6 +19,7 @@ import ( "time" "evobgp/internal/birdfmt" + "evobgp/internal/httpclient" "evobgp/internal/observability" "evobgp/internal/store" @@ -50,7 +51,7 @@ func MaterializedASPrefixKey(asn int64) string { // It does not create a new config revision. func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) error { if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } start := time.Now() 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. func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string) (revisionID string, err error) { if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } agg, err := aggregateTenantPrefixRowsAll(ctx, st, hc, tenantID) 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) { _ = ctx if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } agg := append([]store.PrefixRow(nil), rows...) rawCount := len(agg) diff --git a/internal/pipeline/tenant_refresh.go b/internal/pipeline/tenant_refresh.go index 45fdad7..4fb3157 100644 --- a/internal/pipeline/tenant_refresh.go +++ b/internal/pipeline/tenant_refresh.go @@ -7,13 +7,14 @@ import ( "strings" "sync" + "evobgp/internal/httpclient" "evobgp/internal/store" ) // 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 { if hc == nil { - hc = http.DefaultClient + hc = httpclient.New(httpclient.DefaultTimeout) } var ids []string seen := make(map[string]struct{}) diff --git a/internal/scheduler/run.go b/internal/scheduler/run.go index 10deb9f..238c84f 100644 --- a/internal/scheduler/run.go +++ b/internal/scheduler/run.go @@ -14,6 +14,7 @@ import ( "evobgp/internal/broker" "evobgp/internal/config" + "evobgp/internal/httpclient" "evobgp/internal/jobs" "evobgp/internal/pipeline" "evobgp/internal/store" @@ -108,6 +109,9 @@ func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idem if err != nil { 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("Content-Type", "application/json") if idempotencyKey != "" { @@ -115,9 +119,9 @@ func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idem } hc := deps.HTTP 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 { return err }