From 7a2b015b126748063113280f91b3e5e13cab00e6 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Tue, 19 May 2026 10:35:20 +0700 Subject: [PATCH] feat: add tenant refresh functionality to HTTP API and job processing Implemented a new endpoint for tenant refresh in the HTTP API, allowing for the refresh of modules associated with a tenant. Enhanced job processing to handle tenant refresh jobs, including logic for managing module IDs and job status updates. Updated the scheduler to enqueue tenant refresh jobs based on module due dates, improving the overall efficiency of module management. Additionally, introduced caching for ASN prefix data to optimize performance during refresh operations. --- internal/httpapi/routes.go | 56 ++++++++ internal/jobs/job.go | 11 +- internal/jobs/worker.go | 48 ++++++- internal/pipeline/asn_cache.go | 58 ++++++++ internal/pipeline/asn_cache_test.go | 64 +++++++++ internal/pipeline/cdn_prefetch_test.go | 99 ++++++++++++++ internal/pipeline/cdn_snapshot.go | 126 ++++++++++++++++++ internal/pipeline/collect_parallel.go | 62 +-------- .../pipeline/materialize_regression_test.go | 45 ------- internal/pipeline/prefetch.go | 55 ++++++-- internal/pipeline/scheduler_due.go | 24 ++++ internal/pipeline/tenant_refresh.go | 80 +++++++++++ internal/repository/asn_cache.go | 55 ++++++++ internal/scheduler/run.go | 73 +++++----- internal/store/backend.go | 12 ++ internal/store/memory.go | 2 + internal/store/memory_asn_cache.go | 40 ++++++ .../postgres/000010_asn_prefix_cache.down.sql | 1 + .../postgres/000010_asn_prefix_cache.up.sql | 9 ++ .../sqlite/000010_asn_prefix_cache.down.sql | 1 + .../sqlite/000010_asn_prefix_cache.up.sql | 8 ++ 21 files changed, 774 insertions(+), 155 deletions(-) create mode 100644 internal/pipeline/asn_cache.go create mode 100644 internal/pipeline/asn_cache_test.go create mode 100644 internal/pipeline/cdn_prefetch_test.go create mode 100644 internal/pipeline/cdn_snapshot.go create mode 100644 internal/pipeline/scheduler_due.go create mode 100644 internal/pipeline/tenant_refresh.go create mode 100644 internal/repository/asn_cache.go create mode 100644 internal/store/memory_asn_cache.go create mode 100644 migrations/postgres/000010_asn_prefix_cache.down.sql create mode 100644 migrations/postgres/000010_asn_prefix_cache.up.sql create mode 100644 migrations/sqlite/000010_asn_prefix_cache.down.sql create mode 100644 migrations/sqlite/000010_asn_prefix_cache.up.sql 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);