diff --git a/internal/httpapi/routes.go b/internal/httpapi/routes.go index f092a03..2310a77 100644 --- a/internal/httpapi/routes.go +++ b/internal/httpapi/routes.go @@ -18,6 +18,7 @@ import ( "evobgp/internal/bundle" "evobgp/internal/jobs" "evobgp/internal/observability" + "evobgp/internal/pipeline" "evobgp/internal/store" ) @@ -52,6 +53,7 @@ func (s *Server) registerV1(m *http.ServeMux) { m.HandleFunc("GET /peers", s.handleListPeers) m.HandleFunc("GET /speakers", s.handleListSpeakers) m.HandleFunc("POST /modules/{module_id}/refresh", s.handleModuleRefresh) + m.HandleFunc("POST /tenant/refresh", s.handleTenantRefresh) m.HandleFunc("GET /revisions", s.handleListRevisions) m.HandleFunc("GET /revisions/{revision_id}", s.handleGetRevision) m.HandleFunc("GET /revisions/{revision_id}/preview", s.handleRevisionPreview) @@ -462,6 +464,60 @@ func (s *Server) handleModuleRefresh(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]}) } +func (s *Server) handleTenantRefresh(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + if !s.requireAtLeast(w, a, "editor") { + return + } + var body struct { + ModuleIDs []string `json:"module_ids"` + } + if r.Body != nil && r.ContentLength != 0 { + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&body); err != nil && err != io.EOF { + writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid JSON body") + return + } + } + moduleIDs := body.ModuleIDs + if len(moduleIDs) == 0 { + now := time.Now().UTC() + for _, mod := range s.store.ListModules(a.TenantID) { + if pipeline.ModuleDueForScheduler(mod, now) { + moduleIDs = append(moduleIDs, mod.ID) + } + } + } + if len(moduleIDs) == 0 { + writeJSON(w, http.StatusNoContent, map[string]any{"message": "no modules due for refresh"}) + return + } + idem := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + var idemPtr *string + if idem != "" { + idemPtr = &idem + } + j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindTenantRefresh, idemPtr, nil, map[string]any{ + "module_ids": moduleIDs, + "trigger": "api", + }) + if err != nil { + writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + return + } + w.Header().Set("Location", "/v1/jobs/"+j.ID) + snap := j.Snapshot() + writeJSON(w, http.StatusAccepted, map[string]any{ + "job_id": snap["job_id"], + "status": snap["status"], + "module_ids": moduleIDs, + "modules_due": len(moduleIDs), + }) +} + func (s *Server) handleListRevisions(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { diff --git a/internal/jobs/job.go b/internal/jobs/job.go index fafce3d..d2cffd4 100644 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -346,14 +346,21 @@ func (r *Registry) List(tenantID, statusFilter, kindFilter, cursor string, limit // CountOtherActiveModuleRefresh returns how many module_refresh jobs for the tenant are still // queued or running, excluding excludeJobID (the current job). Used to batch deploy_apply. func (r *Registry) CountOtherActiveModuleRefresh(tenantID, excludeJobID string) int { + return r.CountOtherActiveRefresh(tenantID, excludeJobID) +} + +// CountOtherActiveRefresh counts queued/running module_refresh and tenant_refresh jobs for the tenant. +func (r *Registry) CountOtherActiveRefresh(tenantID, excludeJobID string) int { if r == nil { return 0 } - // Two-phase: snapshot job pointers under r.mu, then read j.Status under Job.mu. r.mu.RLock() candidates := make([]*Job, 0, 8) for _, j := range r.byID { - if j.TenantID != tenantID || j.Kind != KindModuleRefresh { + if j.TenantID != tenantID { + continue + } + if j.Kind != KindModuleRefresh && j.Kind != KindTenantRefresh { continue } if j.ID == excludeJobID { diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index ab3f6a4..2cce850 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -42,6 +42,7 @@ func mergeBirdPostApplyMeta(j *Job) { const ( KindModuleRefresh = "module_refresh" + KindTenantRefresh = "tenant_refresh" KindPeerReconcile = "peer_reconcile" KindDeployApply = "deploy_apply" KindRevisionRollback = "revision_rollback" @@ -106,6 +107,8 @@ func (w *Worker) Process(j *Job) { return } w.finishModuleRefreshSuccess(j, mid) + case KindTenantRefresh: + w.runTenantRefresh(j) case KindPeerReconcile: w.runPeerReconcile(j) case KindDeployApply: @@ -221,6 +224,49 @@ func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex { // finishModuleRefreshSuccess marks the refresh job and, for the last active refresh in tenant, // creates one aggregate revision and enqueues a single deploy_apply. +func (w *Worker) runTenantRefresh(j *Job) { + moduleIDs := moduleIDsFromJobMeta(j.Meta) + if len(moduleIDs) == 0 { + j.Fail("missing module_ids in job meta") + return + } + if err := pipeline.RefreshTenantModules(context.Background(), w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil { + j.Fail(err.Error()) + return + } + trigger, err := pipeline.PickTenantRefreshTriggerModule(w.Store, j.TenantID, moduleIDs) + if err != nil { + j.Fail(err.Error()) + return + } + j.mergeMeta(map[string]any{"module_ids": moduleIDs, "modules_refreshed": len(moduleIDs)}) + w.finishModuleRefreshSuccess(j, trigger) +} + +func moduleIDsFromJobMeta(meta map[string]any) []string { + if meta == nil { + return nil + } + raw, ok := meta["module_ids"] + if !ok { + return nil + } + switch v := raw.(type) { + case []string: + return v + case []any: + var out []string + for _, x := range v { + if s, ok := x.(string); ok && strings.TrimSpace(s) != "" { + out = append(out, strings.TrimSpace(s)) + } + } + return out + default: + return nil + } +} + func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) { if w == nil || w.Store == nil { j.Succeed() @@ -231,7 +277,7 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) { defer mu.Unlock() deferDeploy := false if w.Registry != nil { - deferDeploy = w.Registry.CountOtherActiveModuleRefresh(j.TenantID, j.ID) > 0 + deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0 } if deferDeploy { j.mergeMeta(map[string]any{ diff --git a/internal/pipeline/asn_cache.go b/internal/pipeline/asn_cache.go new file mode 100644 index 0000000..66d4b19 --- /dev/null +++ b/internal/pipeline/asn_cache.go @@ -0,0 +1,58 @@ +package pipeline + +import ( + "context" + "fmt" + "net/http" + "net/netip" + "os" + "strconv" + "strings" + "time" + + "evobgp/internal/asnresolve" + "evobgp/internal/store" +) + +func asnCacheTTL() time.Duration { + sec := 1800 + if s := strings.TrimSpace(os.Getenv("EVOBGP_ASN_CACHE_TTL_SEC")); s != "" { + if v, err := strconv.Atoi(s); err == nil && v > 0 { + sec = v + } + } + return time.Duration(sec) * time.Second +} + +// resolveASNForEntry fetches prefixes and holder with shared TTL cache (asn_prefix_cache). +func resolveASNForEntry(ctx context.Context, st store.Backend, hc *http.Client, asn int64) ([]netip.Prefix, string, error) { + ttl := asnCacheTTL() + if st != nil { + if ent, ok, err := st.GetASNPrefixCache(asn); err == nil && ok && ent != nil && time.Since(ent.FetchedAt) < ttl { + 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()) + } + return out, ent.Holder, nil + } + } + pfxs, err := asnresolve.AnnouncedPrefixes(ctx, hc, asn) + if err != nil { + return nil, "", err + } + holder, _ := asnresolve.ASHolderName(ctx, hc, asn) + if st != nil { + strs := make([]string, len(pfxs)) + for i, p := range pfxs { + strs[i] = p.String() + } + if err := st.SetASNPrefixCache(asn, holder, strs); err != nil { + return nil, "", fmt.Errorf("asn cache AS%d: %w", asn, err) + } + } + return pfxs, holder, nil +} diff --git a/internal/pipeline/asn_cache_test.go b/internal/pipeline/asn_cache_test.go new file mode 100644 index 0000000..aa1ebd1 --- /dev/null +++ b/internal/pipeline/asn_cache_test.go @@ -0,0 +1,64 @@ +package pipeline + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "evobgp/internal/store" +) + +func TestResolveASNForEntry_UsesTTLCache(t *testing.T) { + m := store.NewMemory() + t.Setenv("EVOBGP_ASN_CACHE_TTL_SEC", "3600") + + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + if strings.Contains(r.URL.Path, "/announced") { + _, _ = w.Write([]byte(`{"status":"ok","data":{"prefixes":[{"prefix":"203.0.113.0/24"}]}}`)) + return + } + _, _ = w.Write([]byte(`{"status":"ok","data":{"holder":"Test AS"}}`)) + })) + defer srv.Close() + t.Setenv("EVOBGP_RIPESTAT_ANNOUNCED_PREFIXES_URL", srv.URL+"/announced") + t.Setenv("EVOBGP_RIPESTAT_AS_OVERVIEW_URL", srv.URL+"/overview") + + ctx := context.Background() + hc := srv.Client() + p1, h1, err := resolveASNForEntry(ctx, m, hc, 64512) + if err != nil { + t.Fatal(err) + } + if len(p1) != 1 || h1 != "Test AS" { + t.Fatalf("unexpected first resolve: %+v holder=%q", p1, h1) + } + if calls.Load() < 2 { + t.Fatalf("expected ripestat calls on first resolve, got %d", calls.Load()) + } + firstCalls := calls.Load() + _, _, err = resolveASNForEntry(ctx, m, hc, 64512) + if err != nil { + t.Fatal(err) + } + if calls.Load() != firstCalls { + t.Fatalf("expected cache hit (no new HTTP), calls went from %d to %d", firstCalls, calls.Load()) + } +} + +func TestModuleDueForScheduler_BucketRollover(t *testing.T) { + mod := &store.Module{Enabled: true, Type: "CDN_CIDRS", RefreshIntervalSec: 300} + boundary := time.Unix(300, 0) + if !ModuleDueForScheduler(mod, boundary) { + t.Fatal("expected due when refresh bucket rolls") + } + mid := time.Unix(330, 0) + if ModuleDueForScheduler(mod, mid) { + t.Fatal("expected not due within same bucket") + } +} diff --git a/internal/pipeline/cdn_prefetch_test.go b/internal/pipeline/cdn_prefetch_test.go new file mode 100644 index 0000000..6bcbfa9 --- /dev/null +++ b/internal/pipeline/cdn_prefetch_test.go @@ -0,0 +1,99 @@ +package pipeline + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "evobgp/internal/store" +) + +func TestCollectModulePrefixRows_CDNSendsIfNoneMatch(t *testing.T) { + m := store.NewMemory() + m.SeedDemo() + tenant, _, _, _, _ := m.DemoIDs() + + mod, err := m.CreateModule(tenant, &store.Module{ + Type: "CDN_CIDRS", + Name: "cdn-conditional", + Enabled: true, + }) + if err != nil { + t.Fatal(err) + } + + var gotIfNoneMatch string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotIfNoneMatch = strings.TrimSpace(r.Header.Get("If-None-Match")) + w.Header().Set("ETag", "etag-new") + _, _ = w.Write([]byte("198.51.100.0/24\n")) + })) + defer srv.Close() + + etag := "etag-old" + if _, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ + SourceKind: "txt", + URL: srv.URL, + Etag: etag, + }); err != nil { + t.Fatal(err) + } + + collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil) + if err != nil { + t.Fatal(err) + } + if gotIfNoneMatch != etag { + t.Fatalf("want If-None-Match %q, got %q", etag, gotIfNoneMatch) + } + if len(collected) != 1 || collected[0].Prefix != "198.51.100.0/24" { + t.Fatalf("unexpected collected rows: %+v", collected) + } +} + +func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) { + m := store.NewMemory() + m.SeedDemo() + tenant, _, _, _, _ := m.DemoIDs() + + mod, err := m.CreateModule(tenant, &store.Module{ + Type: "CDN_CIDRS", + Name: "cdn-304", + Enabled: true, + }) + if err != nil { + t.Fatal(err) + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotModified) + })) + defer srv.Close() + + etag := "etag-stable" + src, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ + SourceKind: "txt", + URL: srv.URL, + Etag: etag, + }) + if err != nil { + t.Fatal(err) + } + prior := []store.PrefixRow{{ + Prefix: "203.0.113.0/24", + Source: cdnSourceKey(src.ID), + }} + if err := mergeCDNSourceIntoModuleSnapshot(m, tenant, mod, src.ID, prior); err != nil { + t.Fatal(err) + } + + collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, prior) + if err != nil { + t.Fatal(err) + } + if len(collected) != 1 || collected[0].Prefix != "203.0.113.0/24" { + t.Fatalf("want cached prefix on 304, got %+v", collected) + } +} diff --git a/internal/pipeline/cdn_snapshot.go b/internal/pipeline/cdn_snapshot.go new file mode 100644 index 0000000..0f0e587 --- /dev/null +++ b/internal/pipeline/cdn_snapshot.go @@ -0,0 +1,126 @@ +package pipeline + +import ( + "context" + "fmt" + "io" + "net/http" + "strings" + "time" + + "evobgp/internal/store" +) + +func cdnSourceKey(sourceID string) string { + return "cdn:" + strings.TrimSpace(sourceID) +} + +func mergeSnapshotDropSource(rows []store.PrefixRow, sourceKey string) []store.PrefixRow { + if len(rows) == 0 { + return nil + } + out := make([]store.PrefixRow, 0, len(rows)) + for _, row := range rows { + if row.Source != sourceKey { + out = append(out, row) + } + } + return out +} + +func cdnRowsFromParsed(mod *store.Module, src *store.CDNSource, pfxStrings []string) []store.PrefixRow { + var rows []store.PrefixRow + for _, p := range pfxStrings { + p = strings.TrimSpace(p) + if p == "" { + continue + } + comm := src.CommunityID + if comm == nil && mod != nil && mod.DefaultCommunityID != nil { + c := *mod.DefaultCommunityID + comm = &c + } + rows = append(rows, store.PrefixRow{Prefix: p, CommunityID: comm, Source: cdnSourceKey(src.ID)}) + } + return rows +} + +// mergeCDNSourceIntoModuleSnapshot replaces rows for one CDN source in the module snapshot. +func mergeCDNSourceIntoModuleSnapshot(st store.Backend, tenantID string, mod *store.Module, sourceID string, newRows []store.PrefixRow) error { + if st == nil || mod == nil { + return nil + } + sourceKey := cdnSourceKey(sourceID) + var base []store.PrefixRow + if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, mod.ID); ok && snap != nil { + base = mergeSnapshotDropSource(snap.Prefixes, sourceKey) + } + merged := append(base, newRows...) + return persistModuleSnapshot(st, tenantID, mod, merged) +} + +func parseCDNBody(body string, src *store.CDNSource) ([]string, error) { + pfxs, err := ExtractCIDRs(body, src.SourceKind, src.PrefixPath) + if err != nil { + return nil, err + } + out := make([]string, 0, len(pfxs)) + for _, pfx := range pfxs { + out = append(out, pfx.String()) + } + return out, nil +} + +func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string, mod *store.Module, src *store.CDNSource, priorSnapshot []store.PrefixRow, now time.Time) ([]store.PrefixRow, error) { + u := strings.TrimSpace(src.URL) + if u == "" { + return nil, nil + } + sourceKey := cdnSourceKey(src.ID) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return nil, err + } + if etag := strings.TrimSpace(src.Etag); etag != "" { + req.Header.Set("If-None-Match", etag) + } + resp, err := hc.Do(req) + if err != nil { + return nil, fmt.Errorf("cdn fetch %s: %w", u, err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotModified { + if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 { + return cached, nil + } + return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u) + } + if resp.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, resp.Body) + return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if err != nil { + return nil, err + } + prefixStrs, err := parseCDNBody(string(body), src) + if err != nil { + return nil, fmt.Errorf("cdn parse %s: %w", u, err) + } + etag := strings.TrimSpace(resp.Header.Get("ETag")) + patch := &store.CDNSourcePatch{} + if etag != "" && etag != strings.TrimSpace(src.Etag) { + e := etag + patch.Etag = &e + } + refreshedAt := now + patch.LastRefreshedAt = &refreshedAt + _, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, patch) + + rows := cdnRowsFromParsed(mod, src, prefixStrs) + if err := mergeCDNSourceIntoModuleSnapshot(st, tenantID, mod, src.ID, rows); err != nil { + return nil, err + } + return rows, nil +} diff --git a/internal/pipeline/collect_parallel.go b/internal/pipeline/collect_parallel.go index e74d69c..5379e44 100644 --- a/internal/pipeline/collect_parallel.go +++ b/internal/pipeline/collect_parallel.go @@ -3,14 +3,12 @@ package pipeline import ( "context" "fmt" - "io" "net/http" "os" "strings" "sync" "time" - "evobgp/internal/asnresolve" "evobgp/internal/store" ) @@ -76,15 +74,11 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, c := *mod.DefaultCommunityID comm = &c } - pfxs, err := asnresolve.AnnouncedPrefixes(ctx, hc, entry.ASN) + pfxs, holder, err := resolveASNForEntry(ctx, st, hc, entry.ASN) if err != nil { results[idx] = entryResult{err: fmt.Errorf("resolve AS%d: %w", entry.ASN, err)} return } - holder := "" - if h, err := asnresolve.ASHolderName(ctx, hc, entry.ASN); err == nil { - holder = h - } src := fmt.Sprintf("as:%d", entry.ASN) var rows []store.PrefixRow for _, pfx := range pfxs { @@ -161,18 +155,11 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client return } } - u := strings.TrimSpace(src.URL) - if u == "" { - return - } - rows, err := fetchAndParseCDNSource(ctx, hc, st, tenantID, moduleID, mod, src, now) + rows, err := applyCDNSourceHTTPResult(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now) if err != nil { results[idx] = srcResult{err: err} return } - for i := range rows { - rows[i].Source = sourceKey - } results[idx] = srcResult{rows: rows} }(i, src) } @@ -188,51 +175,6 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client return out, nil } -func fetchAndParseCDNSource(ctx context.Context, hc *http.Client, st store.Backend, tenantID, moduleID string, mod *store.Module, src *store.CDNSource, now time.Time) ([]store.PrefixRow, error) { - u := strings.TrimSpace(src.URL) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) - if err != nil { - return nil, err - } - resp, err := hc.Do(req) - if err != nil { - return nil, fmt.Errorf("cdn fetch %s: %w", u, err) - } - if resp.StatusCode != http.StatusOK { - _, _ = io.Copy(io.Discard, resp.Body) - _ = resp.Body.Close() - return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status) - } - body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) - _ = resp.Body.Close() - if err != nil { - return nil, err - } - etag := strings.TrimSpace(resp.Header.Get("ETag")) - patch := &store.CDNSourcePatch{} - if etag != "" && etag != strings.TrimSpace(src.Etag) { - e := etag - patch.Etag = &e - } - refreshedAt := now - patch.LastRefreshedAt = &refreshedAt - _, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, patch) - pfxs, err := ExtractCIDRs(string(body), src.SourceKind, src.PrefixPath) - if err != nil { - return nil, fmt.Errorf("cdn parse %s: %w", u, err) - } - var rows []store.PrefixRow - for _, pfx := range pfxs { - comm := src.CommunityID - if comm == nil && mod.DefaultCommunityID != nil { - c := *mod.DefaultCommunityID - comm = &c - } - rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm}) - } - return rows, nil -} - func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Module, profile *store.DohProfile, entries []*store.DomainEntry) ([]store.PrefixRow, error) { var validDom []*store.DomainEntry for _, e := range entries { diff --git a/internal/pipeline/materialize_regression_test.go b/internal/pipeline/materialize_regression_test.go index a59eea8..20fb2a2 100644 --- a/internal/pipeline/materialize_regression_test.go +++ b/internal/pipeline/materialize_regression_test.go @@ -1,9 +1,6 @@ package pipeline import ( - "context" - "net/http" - "net/http/httptest" "strings" "testing" @@ -50,45 +47,3 @@ func TestBuildPreviewFragments_SamePrefixDifferentCommunity(t *testing.T) { } } -func TestCollectModulePrefixRows_CDNRefreshForcesFullGet(t *testing.T) { - m := store.NewMemory() - m.SeedDemo() - tenant, _, _, _, _ := m.DemoIDs() - - mod, err := m.CreateModule(tenant, &store.Module{ - Type: "CDN_CIDRS", - Name: "cdn-refresh-force-full", - Enabled: true, - }) - if err != nil { - t.Fatal(err) - } - - var gotIfNoneMatch string - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotIfNoneMatch = strings.TrimSpace(r.Header.Get("If-None-Match")) - w.Header().Set("ETag", "etag-new") - _, _ = w.Write([]byte("198.51.100.0/24\n")) - })) - defer srv.Close() - - etag := "etag-old" - if _, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ - SourceKind: "txt", - URL: srv.URL, - Etag: etag, - }); err != nil { - t.Fatal(err) - } - - collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil) - if err != nil { - t.Fatal(err) - } - if gotIfNoneMatch != "" { - t.Fatalf("refresh path must not send If-None-Match, got %q", gotIfNoneMatch) - } - if len(collected) != 1 || collected[0].Prefix != "198.51.100.0/24" { - t.Fatalf("unexpected collected rows: %+v", collected) - } -} diff --git a/internal/pipeline/prefetch.go b/internal/pipeline/prefetch.go index ad4be87..d7fbb9d 100644 --- a/internal/pipeline/prefetch.go +++ b/internal/pipeline/prefetch.go @@ -5,11 +5,12 @@ import ( "io" "net/http" "strings" + "time" "evobgp/internal/store" ) -// PrefetchCDNSourceETags performs conditional GETs for CDN module sources and updates stored ETags when the origin responds 200. +// 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 @@ -18,42 +19,70 @@ func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Clie if err != nil { return err } + now := time.Now().UTC() for _, tid := range tenants { for _, mod := range st.ListModules(tid) { - if !mod.Enabled || mod.Type != "CDN_CIDRS" { + if mod == nil || !mod.Enabled || mod.Type != "CDN_CIDRS" { + continue + } + omod, err := st.GetModule(tid, mod.ID) + if err != nil { continue } sources, err := st.ListCDNSources(tid, mod.ID) if err != nil { continue } + var prior []store.PrefixRow + if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil { + prior = snap.Prefixes + } for _, src := range sources { - u := strings.TrimSpace(src.URL) - if u == "" { + if src == nil || strings.TrimSpace(src.URL) == "" { continue } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSpace(src.URL), nil) if err != nil { continue } - if strings.TrimSpace(src.Etag) != "" { - req.Header.Set("If-None-Match", strings.TrimSpace(src.Etag)) + if etag := strings.TrimSpace(src.Etag); etag != "" { + req.Header.Set("If-None-Match", etag) } resp, err := hc.Do(req) if err != nil { continue } - _, _ = io.Copy(io.Discard, resp.Body) - _ = resp.Body.Close() + if resp.StatusCode == http.StatusNotModified { + _ = resp.Body.Close() + continue + } if resp.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() continue } - etag := strings.TrimSpace(resp.Header.Get("ETag")) - if etag == "" || etag == strings.TrimSpace(src.Etag) { + body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + _ = resp.Body.Close() + if err != nil { continue } - e := etag - _, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, &store.CDNSourcePatch{Etag: &e}) + prefixStrs, err := parseCDNBody(string(body), src) + if err != nil { + continue + } + newEtag := strings.TrimSpace(resp.Header.Get("ETag")) + patch := &store.CDNSourcePatch{LastRefreshedAt: &now} + if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) { + e := newEtag + patch.Etag = &e + } + _, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch) + rows := cdnRowsFromParsed(omod, src, prefixStrs) + _ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows) + _ = prior // prior may be stale after merge; refresh for next source in loop + if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil { + prior = snap.Prefixes + } } } } diff --git a/internal/pipeline/scheduler_due.go b/internal/pipeline/scheduler_due.go new file mode 100644 index 0000000..5b11001 --- /dev/null +++ b/internal/pipeline/scheduler_due.go @@ -0,0 +1,24 @@ +package pipeline + +import ( + "time" + + "evobgp/internal/store" +) + +// SchedulerTickSec matches the scheduler ticker interval (internal/scheduler). +const SchedulerTickSec = 30 + +// ModuleDueForScheduler reports whether a module's refresh interval bucket rolled since the last scheduler tick. +func ModuleDueForScheduler(mod *store.Module, now time.Time) bool { + if mod == nil || !mod.Enabled || mod.Type == "IP_RANGES" || mod.RefreshIntervalSec <= 0 { + return false + } + win := int64(mod.RefreshIntervalSec) + if win < 60 { + win = 60 + } + cur := now.Unix() / win + prev := (now.Unix() - SchedulerTickSec) / win + return cur != prev +} diff --git a/internal/pipeline/tenant_refresh.go b/internal/pipeline/tenant_refresh.go new file mode 100644 index 0000000..35aab6c --- /dev/null +++ b/internal/pipeline/tenant_refresh.go @@ -0,0 +1,80 @@ +package pipeline + +import ( + "context" + "fmt" + "net/http" + "strings" + "sync" + + "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 + } + var ids []string + seen := make(map[string]struct{}) + for _, id := range moduleIDs { + id = strings.TrimSpace(id) + if id == "" { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + ids = append(ids, id) + } + if len(ids) == 0 { + return nil + } + if len(ids) == 1 { + return RefreshModuleIngest(ctx, st, hc, tenantID, ids[0]) + } + + sem := make(chan struct{}, collectConcurrency()) + errs := make([]error, len(ids)) + var wg sync.WaitGroup + for i, mid := range ids { + wg.Add(1) + go func(idx int, moduleID string) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + errs[idx] = RefreshModuleIngest(ctx, st, hc, tenantID, moduleID) + }(i, mid) + } + wg.Wait() + for _, err := range errs { + if err != nil { + return err + } + } + return nil +} + +// PickTenantRefreshTriggerModule returns a module id for revision metadata (first enabled due module). +func PickTenantRefreshTriggerModule(st store.Backend, tenantID string, moduleIDs []string) (string, error) { + for _, id := range moduleIDs { + id = strings.TrimSpace(id) + if id == "" { + continue + } + mod, err := st.GetModule(tenantID, id) + if err != nil { + return "", err + } + if mod.Enabled { + return mod.ID, nil + } + } + for _, mod := range st.ListModules(tenantID) { + if mod != nil && mod.Enabled { + return mod.ID, nil + } + } + return "", fmt.Errorf("no enabled module for tenant refresh") +} diff --git a/internal/repository/asn_cache.go b/internal/repository/asn_cache.go new file mode 100644 index 0000000..f6d106f --- /dev/null +++ b/internal/repository/asn_cache.go @@ -0,0 +1,55 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "time" + + "evobgp/internal/store" + + "github.com/jackc/pgx/v5" +) + +func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, bool, error) { + ctx := context.Background() + var holder string + var fetchedAt time.Time + var raw []byte + err := p.pool.QueryRow(ctx, ` + SELECT holder, fetched_at, prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn). + Scan(&holder, &fetchedAt, &raw) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, false, nil + } + return nil, false, err + } + var prefixes []string + if len(raw) > 0 { + _ = json.Unmarshal(raw, &prefixes) + } + return &store.ASNPrefixCacheEntry{ + ASN: asn, + Holder: holder, + Prefixes: prefixes, + FetchedAt: fetchedAt.UTC(), + }, true, nil +} + +func (p *Postgres) SetASNPrefixCache(asn int64, holder string, prefixes []string) error { + raw, err := json.Marshal(prefixes) + if err != nil { + return err + } + ctx := context.Background() + _, err = p.pool.Exec(ctx, ` + INSERT INTO asn_prefix_cache (asn, holder, prefixes_json, fetched_at) + VALUES ($1, $2, $3::jsonb, now()) + ON CONFLICT (asn) DO UPDATE SET + holder = EXCLUDED.holder, + prefixes_json = EXCLUDED.prefixes_json, + fetched_at = EXCLUDED.fetched_at`, + asn, holder, string(raw)) + return err +} diff --git a/internal/scheduler/run.go b/internal/scheduler/run.go index 9381e24..7487f08 100644 --- a/internal/scheduler/run.go +++ b/internal/scheduler/run.go @@ -1,8 +1,10 @@ -// Package scheduler drives module refresh intervals and enqueues module_refresh jobs on the shared Registry. +// Package scheduler drives module refresh intervals and enqueues tenant_refresh jobs on the shared Registry. package scheduler import ( + "bytes" "context" + "encoding/json" "fmt" "io" "log" @@ -13,6 +15,7 @@ import ( "evobgp/internal/broker" "evobgp/internal/config" "evobgp/internal/jobs" + "evobgp/internal/pipeline" "evobgp/internal/store" ) @@ -39,9 +42,9 @@ func Run(ctx context.Context, deps *Deps) { t := time.NewTicker(30 * time.Second) defer t.Stop() if deps.Jobs != nil { - log.Printf("evobgp-scheduler: active (in-process enqueue module_refresh)") + log.Printf("evobgp-scheduler: active (in-process enqueue tenant_refresh)") } else { - log.Printf("evobgp-scheduler: active (HTTP POST .../modules/{id}/refresh → %s)", strings.TrimSpace(deps.APIBase)) + log.Printf("evobgp-scheduler: active (HTTP POST .../tenant/refresh → %s)", strings.TrimSpace(deps.APIBase)) } for { select { @@ -60,51 +63,53 @@ func tick(ctx context.Context, deps *Deps) { log.Printf("evobgp-scheduler: list tenants: %v", err) return } + now := time.Now().UTC() + tickBucket := now.Unix() / pipeline.SchedulerTickSec for _, tid := range tenants { + var due []string for _, mod := range deps.Store.ListModules(tid) { - if !mod.Enabled || mod.Type == "IP_RANGES" { + if pipeline.ModuleDueForScheduler(mod, now) { + due = append(due, mod.ID) + } + } + if len(due) == 0 { + continue + } + key := fmt.Sprintf("sched-tenant-%s-%d", tid, tickBucket) + if deps.Jobs != nil { + ids := append([]string(nil), due...) + _, created, err := deps.Jobs.Enqueue(tid, jobs.KindTenantRefresh, &key, nil, map[string]any{ + "module_ids": ids, + "trigger": "scheduler", + }) + if err != nil { + log.Printf("evobgp-scheduler: enqueue tenant %s: %v", tid, err) continue } - interval := mod.RefreshIntervalSec - if interval <= 0 { - continue - } - win := interval - if win < 60 { - win = 60 - } - bucket := time.Now().Unix() / int64(win) - key := fmt.Sprintf("sched-%s-%d", mod.ID, bucket) - if deps.Jobs != nil { - mid := mod.ID - _, created, err := deps.Jobs.Enqueue(tid, jobs.KindModuleRefresh, &key, &mid, map[string]any{ - "module_id": mod.ID, - "trigger": "scheduler", - }) - if err != nil { - log.Printf("evobgp-scheduler: enqueue module %s: %v", mod.ID, err) - continue - } - if created { - log.Printf("evobgp-scheduler: queued refresh for module %s (%s)", mod.ID, mod.Type) - } - continue - } - if err := postModuleRefresh(ctx, deps, mod.ID, key); err != nil { - log.Printf("evobgp-scheduler: http refresh module %s: %v", mod.ID, err) + if created { + log.Printf("evobgp-scheduler: queued tenant refresh for %d module(s) in tenant %s", len(due), tid) } + continue + } + if err := postTenantRefresh(ctx, deps, due, key); err != nil { + log.Printf("evobgp-scheduler: http tenant refresh %s: %v", tid, err) } } } -func postModuleRefresh(ctx context.Context, deps *Deps, moduleID, idempotencyKey string) error { +func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idempotencyKey string) error { base := strings.TrimRight(strings.TrimSpace(deps.APIBase), "/") - u := base + "/v1/modules/" + moduleID + "/refresh" - req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, nil) + u := base + "/v1/tenant/refresh" + body, err := json.Marshal(map[string]any{"module_ids": moduleIDs}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(body)) if err != nil { return err } req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(deps.APIToken)) + req.Header.Set("Content-Type", "application/json") if idempotencyKey != "" { req.Header.Set("Idempotency-Key", idempotencyKey) } diff --git a/internal/store/backend.go b/internal/store/backend.go index b9342d2..e25d45c 100644 --- a/internal/store/backend.go +++ b/internal/store/backend.go @@ -84,6 +84,18 @@ type Backend interface { GetModulePrefixSnapshot(tenantID, moduleID string) (*ModulePrefixSnapshot, bool, error) SetModulePrefixSnapshot(tenantID, moduleID, inputHash string, prefixes []PrefixRow) error DeleteModulePrefixSnapshot(tenantID, moduleID string) error + + // ASNPrefixCache stores RIPEstat announced-prefixes per ASN (global TTL cache). + GetASNPrefixCache(asn int64) (*ASNPrefixCacheEntry, bool, error) + SetASNPrefixCache(asn int64, holder string, prefixes []string) error +} + +// ASNPrefixCacheEntry is a cached RIPEstat response for one ASN. +type ASNPrefixCacheEntry struct { + ASN int64 + Holder string + Prefixes []string + FetchedAt time.Time } // ModulePrefixSnapshot is the cached materialization for one module between refreshes. diff --git a/internal/store/memory.go b/internal/store/memory.go index 447dcbc..1f22519 100644 --- a/internal/store/memory.go +++ b/internal/store/memory.go @@ -41,6 +41,7 @@ type Memory struct { settings map[string]map[string]any // tenantID -> key -> JSON-compatible value revPrefixes map[string][]PrefixRow moduleSnapshots map[string]*moduleSnapshotRec + asnPrefixCache map[int64]*ASNPrefixCacheEntry // DemoIDs valid after SeedDemo() demoTenantID string @@ -128,6 +129,7 @@ func NewMemory() *Memory { settings: make(map[string]map[string]any), revPrefixes: make(map[string][]PrefixRow), moduleSnapshots: make(map[string]*moduleSnapshotRec), + asnPrefixCache: make(map[int64]*ASNPrefixCacheEntry), } } diff --git a/internal/store/memory_asn_cache.go b/internal/store/memory_asn_cache.go new file mode 100644 index 0000000..f77a67a --- /dev/null +++ b/internal/store/memory_asn_cache.go @@ -0,0 +1,40 @@ +package store + +import ( + "time" +) + +func (m *Memory) GetASNPrefixCache(asn int64) (*ASNPrefixCacheEntry, bool, error) { + m.mu.RLock() + defer m.mu.RUnlock() + if m.asnPrefixCache == nil { + return nil, false, nil + } + e, ok := m.asnPrefixCache[asn] + if !ok || e == nil { + return nil, false, nil + } + pfx := append([]string(nil), e.Prefixes...) + return &ASNPrefixCacheEntry{ + ASN: asn, + Holder: e.Holder, + Prefixes: pfx, + FetchedAt: e.FetchedAt, + }, true, nil +} + +func (m *Memory) SetASNPrefixCache(asn int64, holder string, prefixes []string) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.asnPrefixCache == nil { + m.asnPrefixCache = make(map[int64]*ASNPrefixCacheEntry) + } + cp := append([]string(nil), prefixes...) + m.asnPrefixCache[asn] = &ASNPrefixCacheEntry{ + ASN: asn, + Holder: holder, + Prefixes: cp, + FetchedAt: time.Now().UTC(), + } + return nil +} diff --git a/migrations/postgres/000010_asn_prefix_cache.down.sql b/migrations/postgres/000010_asn_prefix_cache.down.sql new file mode 100644 index 0000000..019c381 --- /dev/null +++ b/migrations/postgres/000010_asn_prefix_cache.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS asn_prefix_cache; diff --git a/migrations/postgres/000010_asn_prefix_cache.up.sql b/migrations/postgres/000010_asn_prefix_cache.up.sql new file mode 100644 index 0000000..dd3e58d --- /dev/null +++ b/migrations/postgres/000010_asn_prefix_cache.up.sql @@ -0,0 +1,9 @@ +-- TTL cache for RIPEstat announced-prefixes per ASN (pipeline ingest). +CREATE TABLE asn_prefix_cache ( + asn BIGINT PRIMARY KEY, + holder TEXT NOT NULL DEFAULT '', + prefixes_json JSONB NOT NULL DEFAULT '[]', + fetched_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX idx_asn_prefix_cache_fetched ON asn_prefix_cache (fetched_at); diff --git a/migrations/sqlite/000010_asn_prefix_cache.down.sql b/migrations/sqlite/000010_asn_prefix_cache.down.sql new file mode 100644 index 0000000..019c381 --- /dev/null +++ b/migrations/sqlite/000010_asn_prefix_cache.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS asn_prefix_cache; diff --git a/migrations/sqlite/000010_asn_prefix_cache.up.sql b/migrations/sqlite/000010_asn_prefix_cache.up.sql new file mode 100644 index 0000000..61ff926 --- /dev/null +++ b/migrations/sqlite/000010_asn_prefix_cache.up.sql @@ -0,0 +1,8 @@ +CREATE TABLE asn_prefix_cache ( + asn INTEGER PRIMARY KEY, + holder TEXT NOT NULL DEFAULT '', + prefixes_json TEXT NOT NULL DEFAULT '[]', + fetched_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) +); + +CREATE INDEX idx_asn_prefix_cache_fetched ON asn_prefix_cache (fetched_at);