diff --git a/internal/pipeline/collect_parallel.go b/internal/pipeline/collect_parallel.go index f29bf7b..e89d759 100644 --- a/internal/pipeline/collect_parallel.go +++ b/internal/pipeline/collect_parallel.go @@ -25,7 +25,7 @@ func prefixRowsForSource(rows []store.PrefixRow, sourceKey string) []store.Prefi return out } -func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, list []*store.ASEntry) ([]store.PrefixRow, error) { +func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, list []*store.ASEntry, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) { moduleID := mod.ID legacy := strings.TrimSpace(os.Getenv("EVOBGP_ASN_RESOLVE")) == "0" if legacy { @@ -76,6 +76,41 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, } pfxs, holder, err := resolveASNForEntry(ctx, st, hc, entry.ASN) if err != nil { + if staleOnUpstreamError() { + if staleRows, staleHolder, ok := staleASNPrefixes(st, priorSnapshot, entry.ASN); ok { + logStaleUpstream("asn", fmt.Sprintf("AS%d: %v", entry.ASN, err)) + src := fmt.Sprintf("as:%d", entry.ASN) + rows := append([]store.PrefixRow(nil), staleRows...) + for i := range rows { + rows[i].CommunityID = comm + rows[i].Source = src + } + results[idx] = entryResult{ + rows: rows, + metaID: entry.ID, + asn: entry.ASN, + holder: staleHolder, + count: int64(len(rows)), + } + return + } + if pfxs2, holder2, ok := asnCacheExpired(st, entry.ASN); ok { + logStaleUpstream("asn", fmt.Sprintf("AS%d expired cache: %v", entry.ASN, err)) + src := fmt.Sprintf("as:%d", entry.ASN) + var rows []store.PrefixRow + for _, pfx := range pfxs2 { + rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm, Source: src}) + } + results[idx] = entryResult{ + rows: rows, + metaID: entry.ID, + asn: entry.ASN, + holder: holder2, + count: int64(len(pfxs2)), + } + return + } + } results[idx] = entryResult{err: fmt.Errorf("resolve AS%d: %w", entry.ASN, err)} return } @@ -167,6 +202,13 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client } rows, err := fetchCDNSourceRows(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now) if err != nil { + if staleOnUpstreamError() { + if cached, ok := staleCDNPrefixes(st, tenantID, moduleID, priorSnapshot, src.ID); ok { + logStaleUpstream("cdn", fmt.Sprintf("source %s: %v", src.ID, err)) + results[idx] = srcResult{rows: cached} + return + } + } results[idx] = srcResult{err: err} return } @@ -190,7 +232,7 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client return out, nil } -func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Module, profiles []*store.DohProfile, policy string, entries []*store.DomainEntry) ([]store.PrefixRow, error) { +func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Module, profiles []*store.DohProfile, policy string, entries []*store.DomainEntry, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) { var validDom []*store.DomainEntry for _, e := range entries { if e != nil { @@ -219,6 +261,19 @@ func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Mo } addrs, err := resolveDomainIPsWithPolicy(ctx, hc, profiles, policy, entry.FQDN) if err != nil { + if staleOnUpstreamError() { + if cached, ok := staleDomainPrefixes(priorSnapshot, entry.FQDN); ok { + logStaleUpstream("domain", fmt.Sprintf("%q: %v", entry.FQDN, err)) + rows := append([]store.PrefixRow(nil), cached...) + for i := range rows { + if rows[i].CommunityID == nil { + rows[i].CommunityID = comm + } + } + results[idx] = domResult{rows: rows} + return + } + } results[idx] = domResult{err: fmt.Errorf("resolve domain %q: %w", entry.FQDN, err)} return } diff --git a/internal/pipeline/collect_stale.go b/internal/pipeline/collect_stale.go new file mode 100644 index 0000000..2871be6 --- /dev/null +++ b/internal/pipeline/collect_stale.go @@ -0,0 +1,86 @@ +package pipeline + +import ( + "fmt" + "log" + "net/netip" + "os" + "strings" + + "evobgp/internal/store" +) + +// staleOnUpstreamError reports whether ingest should keep last-known prefixes when an upstream fetch fails. +// Enabled by default; set EVOBGP_STALE_ON_UPSTREAM_ERROR=0 to restore fail-fast behavior. +func staleOnUpstreamError() bool { + v := strings.TrimSpace(os.Getenv("EVOBGP_STALE_ON_UPSTREAM_ERROR")) + if v == "" || v == "1" || strings.EqualFold(v, "true") { + return true + } + return false +} + +func logStaleUpstream(kind, detail string) { + log.Printf("pipeline: stale upstream fallback (%s): %s", kind, detail) +} + +func staleASNPrefixes(st store.Backend, priorSnapshot []store.PrefixRow, asn int64) ([]store.PrefixRow, string, bool) { + sourceKey := fmt.Sprintf("as:%d", asn) + if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 { + return cached, "", true + } + if st == nil { + return nil, "", false + } + ent, ok, err := st.GetASNPrefixCache(asn) + if err != nil || !ok || ent == nil || len(ent.Prefixes) == 0 { + return nil, "", false + } + var rows []store.PrefixRow + for _, p := range ent.Prefixes { + pfx, perr := netip.ParsePrefix(strings.TrimSpace(p)) + if perr != nil { + continue + } + rows = append(rows, store.PrefixRow{Prefix: pfx.Masked().String(), Source: sourceKey}) + } + if len(rows) == 0 { + return nil, "", false + } + return rows, ent.Holder, true +} + +func staleDomainPrefixes(priorSnapshot []store.PrefixRow, fqdn string) ([]store.PrefixRow, bool) { + sourceKey := "domain:" + strings.TrimSpace(fqdn) + cached := prefixRowsForSource(priorSnapshot, sourceKey) + return cached, len(cached) > 0 +} + +func staleCDNPrefixes(st store.Backend, tenantID, moduleID string, priorSnapshot []store.PrefixRow, sourceID string) ([]store.PrefixRow, bool) { + sourceKey := cdnSourceKey(sourceID) + cached := cachedCDNPrefixRows(st, tenantID, moduleID, priorSnapshot, sourceKey) + return cached, len(cached) > 0 +} + +// asnCacheExpired returns cached ASN prefixes even past TTL (for stale fallback only). +func asnCacheExpired(st store.Backend, asn int64) ([]netip.Prefix, string, bool) { + if st == nil { + return nil, "", false + } + ent, ok, err := st.GetASNPrefixCache(asn) + if err != nil || !ok || ent == nil || len(ent.Prefixes) == 0 { + return nil, "", false + } + out := make([]netip.Prefix, 0, len(ent.Prefixes)) + for _, p := range ent.Prefixes { + pfx, perr := netip.ParsePrefix(strings.TrimSpace(p)) + if perr != nil { + continue + } + out = append(out, pfx.Masked()) + } + if len(out) == 0 { + return nil, "", false + } + return out, ent.Holder, true +} diff --git a/internal/pipeline/collect_stale_test.go b/internal/pipeline/collect_stale_test.go new file mode 100644 index 0000000..6762348 --- /dev/null +++ b/internal/pipeline/collect_stale_test.go @@ -0,0 +1,70 @@ +package pipeline + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "evobgp/internal/store" +) + +func TestCollectCDNPrefixRows_StaleOnFetchError(t *testing.T) { + t.Setenv("EVOBGP_STALE_ON_UPSTREAM_ERROR", "1") + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "upstream down", http.StatusServiceUnavailable) + })) + defer srv.Close() + + st := store.NewMemory() + st.SeedDemo() + tenant, _, _, _, _ := st.DemoIDs() + mod, err := st.CreateModule(tenant, &store.Module{Type: "CDN_CIDRS", Name: "cdn", Enabled: true}) + if err != nil { + t.Fatal(err) + } + if _, err := st.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ + ID: "s1", URL: srv.URL, SourceKind: "plain", + }); err != nil { + t.Fatal(err) + } + + prior := []store.PrefixRow{ + {Prefix: "203.0.113.0/24", Source: "cdn:s1"}, + } + rows, err := collectCDNPrefixRows(context.Background(), st, srv.Client(), tenant, mod, []*store.CDNSource{{ID: "s1", URL: srv.URL, SourceKind: "plain"}}, prior) + if err != nil { + t.Fatalf("expected stale fallback, got err: %v", err) + } + if len(rows) != 1 || rows[0].Prefix != "203.0.113.0/24" { + t.Fatalf("unexpected rows: %+v", rows) + } +} + +func TestCollectCDNPrefixRows_FailFastWhenNoStale(t *testing.T) { + t.Setenv("EVOBGP_STALE_ON_UPSTREAM_ERROR", "0") + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "upstream down", http.StatusServiceUnavailable) + })) + defer srv.Close() + + st := store.NewMemory() + st.SeedDemo() + tenant, _, _, _, _ := st.DemoIDs() + mod, err := st.CreateModule(tenant, &store.Module{Type: "CDN_CIDRS", Name: "cdn", Enabled: true}) + if err != nil { + t.Fatal(err) + } + if _, err := st.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ + ID: "s1", URL: srv.URL, SourceKind: "plain", + }); err != nil { + t.Fatal(err) + } + + _, err = collectCDNPrefixRows(context.Background(), st, srv.Client(), tenant, mod, []*store.CDNSource{{ID: "s1", URL: srv.URL, SourceKind: "plain"}}, nil) + if err == nil { + t.Fatal("expected error when stale disabled and no cache") + } +} diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index 7b716b3..ced9811 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -176,7 +176,7 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli return nil, err } sort.Slice(list, func(i, j int) bool { return list[i].ASN < list[j].ASN }) - return collectASPrefixRows(ctx, st, hc, tenantID, mod, list) + return collectASPrefixRows(ctx, st, hc, tenantID, mod, list, priorSnapshot) case "CDN_CIDRS": sources, err := st.ListCDNSources(tenantID, moduleID) if err != nil { @@ -192,7 +192,7 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli if err != nil { return nil, err } - return collectDomainPrefixRows(ctx, hc, mod, profiles, policy, entries) + return collectDomainPrefixRows(ctx, hc, mod, profiles, policy, entries, priorSnapshot) default: return nil, fmt.Errorf("pipeline: unknown module type %q", mod.Type) }