From 431c6a90f1c430c9e681034967a1ecd7a2da28be Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 6 Apr 2026 11:03:15 +0700 Subject: [PATCH] feat: enhance RefreshModule to aggregate prefixes from all enabled tenant modules. Update documentation to clarify the behavior of ingest and revision generation, ensuring accurate representation of data sources and prefix handling. --- docs/openapi.yaml | 3 + internal/pipeline/refresh.go | 102 ++++++++++++++------ internal/pipeline/refresh_aggregate_test.go | 51 ++++++++++ 3 files changed, 129 insertions(+), 27 deletions(-) create mode 100644 internal/pipeline/refresh_aggregate_test.go diff --git a/docs/openapi.yaml b/docs/openapi.yaml index f80abb8..4f19267 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1421,6 +1421,9 @@ paths: summary: Запустить ingest description: | Запуск обновления данных модуля (CDN / DoH / AS в зависимости от типа). + Итоговая ревизия и фрагменты BIRD строятся **по всем включённым модулям тенанта**: + обновляемый модуль пересчитывается заново, остальные — по текущему состоянию в БД/источниках, + так что префиксы ASN, CDN, IP и т.д. суммируются в одном конфиге. Для `IP_RANGES` данные обычно только в БД: сервер может вернуть **204** (no-op) или **400**, если refresh не поддерживается - поведение фиксируется в реализации. Рекомендуется передавать `Idempotency-Key`. operationId: postModuleRefresh diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index 820987d..7c0eaac 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -31,7 +31,8 @@ func MaterializedASPrefixKey(asn int64) string { return fmt.Sprintf("as:%d", asn) } -// RefreshModule runs ingest (where applicable) and creates a new rendered revision for the module. +// RefreshModule runs ingest (where applicable) for one module, then renders a new revision whose +// BIRD materialization includes prefixes from all enabled modules of the tenant (others via live collect). func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) (revisionID string, err error) { if hc == nil { hc = http.DefaultClient @@ -44,13 +45,38 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan return "", fmt.Errorf("module disabled") } - var rows []store.PrefixRow + rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod) + if err != nil { + return "", err + } + + revisionID = uuid.NewString() + parent := parentRevision(st, tenantID, moduleID) + agg, err := aggregateTenantPrefixRows(ctx, st, hc, tenantID, moduleID, rows) + if err != nil { + return "", err + } + hash := hashAggregatedMaterialization(tenantID, agg) + preview, err := buildPreviewFragments(st, tenantID, moduleID, revisionID, agg) + if err != nil { + return "", err + } + if err := st.CreateRenderRevision(revisionID, tenantID, moduleID, parent, hash, preview, agg); err != nil { + return "", err + } + return revisionID, nil +} + +// collectModulePrefixRows returns materialized prefix rows for a single module (source of truth from store / ASN resolve / CDN fetch). +func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module) ([]store.PrefixRow, error) { + moduleID := mod.ID switch mod.Type { case "IP_RANGES": list, err := st.ListIPRangeEntries(tenantID, moduleID) if err != nil { - return "", err + return nil, err } + var rows []store.PrefixRow for _, e := range list { comm := e.CommunityID if comm == nil && mod.DefaultCommunityID != nil { @@ -59,14 +85,16 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan } rows = append(rows, store.PrefixRow{Prefix: e.Prefix, CommunityID: comm, Source: "ip_range"}) } + return rows, nil case "AS_PREFIXES": list, err := st.ListASEntries(tenantID, moduleID) if err != nil { - return "", err + return nil, err } sort.Slice(list, func(i, j int) bool { return list[i].ASN < list[j].ASN }) legacy := strings.TrimSpace(os.Getenv("EVOBGP_ASN_RESOLVE")) == "0" seenPfx := make(map[string]struct{}) + var rows []store.PrefixRow for i, e := range list { if !store.ValidASN(e.ASN) { continue @@ -85,7 +113,7 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan } pfxs, err := asnresolve.AnnouncedPrefixes(ctx, hc, e.ASN) if err != nil { - return "", fmt.Errorf("resolve AS%d: %w", e.ASN, err) + return nil, fmt.Errorf("resolve AS%d: %w", e.ASN, err) } holder := "" asnresolve.PolitePause() @@ -94,7 +122,7 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan } now := time.Now().UTC() if err := st.UpdateASEntryResolveMeta(tenantID, moduleID, e.ID, holder, int64(len(pfxs)), now); err != nil { - return "", fmt.Errorf("as entry meta AS%d: %w", e.ASN, err) + return nil, fmt.Errorf("as entry meta AS%d: %w", e.ASN, err) } src := fmt.Sprintf("as:%d", e.ASN) for _, pfx := range pfxs { @@ -106,11 +134,13 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan rows = append(rows, store.PrefixRow{Prefix: k, CommunityID: comm, Source: src}) } } + return rows, nil case "CDN_CIDRS": sources, err := st.ListCDNSources(tenantID, moduleID) if err != nil { - return "", err + return nil, err } + var rows []store.PrefixRow for _, src := range sources { u := strings.TrimSpace(src.URL) if u == "" { @@ -118,14 +148,14 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan } req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) if err != nil { - return "", err + return nil, err } if strings.TrimSpace(src.Etag) != "" { req.Header.Set("If-None-Match", strings.TrimSpace(src.Etag)) } resp, err := hc.Do(req) if err != nil { - return "", fmt.Errorf("cdn fetch %s: %w", u, err) + return nil, fmt.Errorf("cdn fetch %s: %w", u, err) } if resp.StatusCode == http.StatusNotModified { _ = resp.Body.Close() @@ -134,12 +164,12 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan if resp.StatusCode != http.StatusOK { _, _ = io.Copy(io.Discard, resp.Body) _ = resp.Body.Close() - return "", fmt.Errorf("cdn url %s: %s", u, resp.Status) + 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 "", err + return nil, err } etag := strings.TrimSpace(resp.Header.Get("ETag")) if etag != "" && etag != strings.TrimSpace(src.Etag) { @@ -155,26 +185,43 @@ func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenan rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm, Source: "cdn:" + src.ID}) } } + return rows, nil case "DOMAINS": if _, err := st.ListDomainEntries(tenantID, moduleID); err != nil { - return "", err + return nil, err } - // DNS/DoH resolution not wired yet; emit empty prefix set (valid revision). + return nil, nil default: - return "", fmt.Errorf("unknown module type %q", mod.Type) + return nil, fmt.Errorf("unknown module type %q", mod.Type) } +} - revisionID = uuid.NewString() - parent := parentRevision(st, tenantID, moduleID) - hash := hashMaterialization(moduleID, rows) - preview, err := buildPreviewFragments(st, tenantID, moduleID, revisionID, rows) - if err != nil { - return "", err +// aggregateTenantPrefixRows builds the union of materialized prefixes for all enabled modules. +// The module that triggered refresh contributes freshRows; every other module is collected live from the store +// (same logic as refresh). We do not reuse other modules' saved revisions as prefix sources, because each revision +// already stores the full tenant-wide aggregate — mixing them with freshRows would duplicate prefixes. +func aggregateTenantPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID, changedModuleID string, freshRows []store.PrefixRow) ([]store.PrefixRow, error) { + mods := st.ListModules(tenantID) + var out []store.PrefixRow + for _, m := range mods { + if m == nil || !m.Enabled { + continue + } + if m.ID == changedModuleID { + out = append(out, freshRows...) + continue + } + omod, err := st.GetModule(tenantID, m.ID) + if err != nil { + return nil, err + } + rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, omod) + if err != nil { + return nil, fmt.Errorf("module %s: %w", m.ID, err) + } + out = append(out, rows...) } - if err := st.CreateRenderRevision(revisionID, tenantID, moduleID, parent, hash, preview, rows); err != nil { - return "", err - } - return revisionID, nil + return out, nil } func parentRevision(st store.Backend, tenantID, moduleID string) *string { @@ -186,7 +233,8 @@ func parentRevision(st store.Backend, tenantID, moduleID string) *string { return &id } -func hashMaterialization(moduleID string, rows []store.PrefixRow) string { +// hashAggregatedMaterialization hashes the full tenant-wide prefix set used for BIRD (all enabled modules). +func hashAggregatedMaterialization(tenantID string, rows []store.PrefixRow) string { type line struct{ p, c, s string } var lines []line for _, r := range rows { @@ -206,7 +254,7 @@ func hashMaterialization(moduleID string, rows []store.PrefixRow) string { return lines[i].s < lines[j].s }) h := sha256.New() - h.Write([]byte(moduleID)) + h.Write([]byte(strings.TrimSpace(tenantID))) h.Write([]byte{0}) for _, l := range lines { h.Write([]byte(l.p)) @@ -254,7 +302,7 @@ func buildPreviewFragments(st store.Backend, tenantID, moduleID, revisionID stri main, err := birdfmt.RenderMainBirdConf(birdfmt.MainBirdConfOptions{ RouterID: locals.routerID, Includes: birdfmt.StandardIncludeFragments(), - Preamble: fmt.Sprintf("EvoBGP module %s revision %s", moduleID, revisionID), + Preamble: fmt.Sprintf("EvoBGP tenant aggregate config (trigger module %s) revision %s", moduleID, revisionID), }) if err != nil { return nil, err diff --git a/internal/pipeline/refresh_aggregate_test.go b/internal/pipeline/refresh_aggregate_test.go new file mode 100644 index 0000000..a059d19 --- /dev/null +++ b/internal/pipeline/refresh_aggregate_test.go @@ -0,0 +1,51 @@ +package pipeline + +import ( + "context" + "net/http" + "testing" + + "evobgp/internal/store" +) + +func TestRefreshModule_AggregatesAllEnabledModules(t *testing.T) { + m := store.NewMemory() + m.SeedDemo() + tenant, _, modIP, _, _ := m.DemoIDs() + + mod2, err := m.CreateModule(tenant, &store.Module{Type: "IP_RANGES", Name: "extra-ip", Enabled: true, Priority: 30}) + if err != nil { + t.Fatal(err) + } + if _, err := m.CreateIPRangeEntry(tenant, modIP, &store.IPRangeEntry{Prefix: "10.0.0.0/24"}); err != nil { + t.Fatal(err) + } + if _, err := m.CreateIPRangeEntry(tenant, mod2.ID, &store.IPRangeEntry{Prefix: "192.168.0.0/24"}); err != nil { + t.Fatal(err) + } + + ctx := context.Background() + if _, err := RefreshModule(ctx, m, http.DefaultClient, tenant, modIP); err != nil { + t.Fatal(err) + } + revs, _, _ := m.ListRevisions(tenant, modIP, "", 1) + if len(revs) == 0 { + t.Fatal("no revision") + } + px, _, _ := m.ListRevisionPrefixes(tenant, revs[0].ID, "", 1000) + if len(px) != 2 { + t.Fatalf("first refresh: want 2 aggregated prefixes, got %d: %+v", len(px), px) + } + + if _, err := RefreshModule(ctx, m, http.DefaultClient, tenant, mod2.ID); err != nil { + t.Fatal(err) + } + revs2, _, _ := m.ListRevisions(tenant, mod2.ID, "", 1) + if len(revs2) == 0 { + t.Fatal("no revision for mod2") + } + px2, _, _ := m.ListRevisionPrefixes(tenant, revs2[0].ID, "", 1000) + if len(px2) != 2 { + t.Fatalf("second refresh: want 2 aggregated prefixes, got %d: %+v", len(px2), px2) + } +}