From a8c5e9701f28e6419558bc6fca4600c6b074d93c Mon Sep 17 00:00:00 2001 From: Denozordec Date: Thu, 21 May 2026 10:45:39 +0700 Subject: [PATCH] perf: optimize data path, indexes and frontend virtualization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Убран latestCDNRowsBySource; expanded BIRD preview генерируется on-demand - Batch AS meta updates; миграция perf-индексов 000012 - VirtualPrefixList для preview префиксов; метрики pipeline refresh - PolitePause для RIPEstat после cache miss Co-authored-by: Cursor --- internal/pipeline/asn_cache.go | 1 + internal/pipeline/cdn_snapshot.go | 3 -- internal/pipeline/collect_parallel.go | 22 ++++++--- internal/pipeline/refresh.go | 46 ++++++++----------- internal/repository/postgres_entities.go | 32 +++++++++++++ internal/store/as_meta_batch.go | 8 ++++ internal/store/backend.go | 3 ++ internal/store/memory_crud.go | 9 ++++ .../postgres/000012_perf_indexes.down.sql | 6 +++ .../postgres/000012_perf_indexes.up.sql | 17 +++++++ .../sqlite/000012_perf_indexes.down.sql | 6 +++ migrations/sqlite/000012_perf_indexes.up.sql | 17 +++++++ .../virtual-list/virtual-prefix-list.svelte | 37 +++++++++++++++ web/src/routes/operations/+page.svelte | 11 +---- 14 files changed, 172 insertions(+), 46 deletions(-) create mode 100644 internal/store/as_meta_batch.go create mode 100644 migrations/postgres/000012_perf_indexes.down.sql create mode 100644 migrations/postgres/000012_perf_indexes.up.sql create mode 100644 migrations/sqlite/000012_perf_indexes.down.sql create mode 100644 migrations/sqlite/000012_perf_indexes.up.sql create mode 100644 web/src/lib/ui/patterns/virtual-list/virtual-prefix-list.svelte diff --git a/internal/pipeline/asn_cache.go b/internal/pipeline/asn_cache.go index 66d4b19..3115197 100644 --- a/internal/pipeline/asn_cache.go +++ b/internal/pipeline/asn_cache.go @@ -44,6 +44,7 @@ func resolveASNForEntry(ctx context.Context, st store.Backend, hc *http.Client, if err != nil { return nil, "", err } + asnresolve.PolitePause() holder, _ := asnresolve.ASHolderName(ctx, hc, asn) if st != nil { strs := make([]string, len(pfxs)) diff --git a/internal/pipeline/cdn_snapshot.go b/internal/pipeline/cdn_snapshot.go index 42055b9..c1316a4 100644 --- a/internal/pipeline/cdn_snapshot.go +++ b/internal/pipeline/cdn_snapshot.go @@ -25,9 +25,6 @@ func cachedCDNPrefixRows(st store.Backend, tenantID, moduleID string, priorSnaps return cached } } - if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 { - return cached - } } return nil } diff --git a/internal/pipeline/collect_parallel.go b/internal/pipeline/collect_parallel.go index 1d773a0..f29bf7b 100644 --- a/internal/pipeline/collect_parallel.go +++ b/internal/pipeline/collect_parallel.go @@ -97,15 +97,18 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, seenPfx := make(map[string]struct{}) var out []store.PrefixRow + var metaUpdates []store.ASEntryResolveMetaUpdate now := time.Now().UTC() for _, r := range results { if r.err != nil { return nil, r.err } if r.metaID != "" { - if err := st.UpdateASEntryResolveMeta(tenantID, moduleID, r.metaID, r.holder, r.count, now); err != nil { - return nil, fmt.Errorf("as entry meta AS%d: %w", r.asn, err) - } + metaUpdates = append(metaUpdates, store.ASEntryResolveMetaUpdate{ + EntryID: r.metaID, + ASNName: r.holder, + PrefixCount: r.count, + }) } for _, row := range r.rows { k := row.Prefix @@ -116,6 +119,11 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, out = append(out, row) } } + if len(metaUpdates) > 0 { + if err := st.UpdateASEntryResolveMetaBatch(tenantID, moduleID, metaUpdates, now); err != nil { + return nil, fmt.Errorf("as entry meta batch: %w", err) + } + } return out, nil } @@ -150,9 +158,11 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client results[idx] = srcResult{rows: cached} return } - if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 { - results[idx] = srcResult{rows: cached} - return + if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil { + if cached := prefixRowsForSource(snap.Prefixes, sourceKey); len(cached) > 0 { + results[idx] = srcResult{rows: cached} + return + } } } rows, err := fetchCDNSourceRows(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now) diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index 336bdca..3c41c49 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -36,6 +36,11 @@ const ( revisionDefaultTTL = 30 * 24 * time.Hour ) +// AuxBirdFullExpandedKey returns the preview map key for the expanded BIRD config (generated on demand). +func AuxBirdFullExpandedKey() string { + return auxBirdFullExpanded +} + // MaterializedASPrefixKey returns the revision snapshot key for an AS-only entry (not a CIDR). func MaterializedASPrefixKey(asn int64) string { return fmt.Sprintf("as:%d", asn) @@ -47,10 +52,14 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, if hc == nil { hc = http.DefaultClient } + start := time.Now() mod, err := st.GetModule(tenantID, moduleID) if err != nil { return err } + defer func() { + observability.RecordPipelineRefresh(mod.Type, time.Since(start)) + }() if !mod.Enabled { return fmt.Errorf("pipeline: module disabled") } @@ -197,33 +206,6 @@ func shouldSkipCDNSourceFetch(src *store.CDNSource, now time.Time) bool { return now.UTC().Before(nextRefreshAt) } -func latestCDNRowsBySource(st store.Backend, tenantID string) map[string][]store.PrefixRow { - out := make(map[string][]store.PrefixRow) - if st == nil { - return out - } - revs, _, _ := st.ListRevisions(tenantID, "", "", 1) - if len(revs) == 0 || strings.TrimSpace(revs[0].ID) == "" { - return out - } - revID := strings.TrimSpace(revs[0].ID) - cursor := "" - for { - page, next, more := st.ListRevisionPrefixes(tenantID, revID, cursor, 2000) - for _, row := range page { - if !strings.HasPrefix(strings.TrimSpace(row.Source), "cdn:") { - continue - } - out[row.Source] = append(out[row.Source], row) - } - if !more || strings.TrimSpace(next) == "" { - break - } - cursor = next - } - return out -} - type dohJSONAnswer struct { Type int `json:"type"` Data string `json:"data"` @@ -836,10 +818,18 @@ func buildPreviewFragments(st store.Backend, tenantID, moduleID, revisionID stri px6: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), staticV6), pPeers: peersBody, } - out[auxBirdFullExpanded] = buildExpandedBirdText(main, out) return out, nil } +// BuildExpandedBirdPreview concatenates bird.conf and deployable includes for UI preview (not persisted in revision). +func BuildExpandedBirdPreview(frags map[string]string) string { + if frags == nil { + return "" + } + main := frags["bird.conf"] + return buildExpandedBirdText(main, frags) +} + func renderStaticProtocolsByCommunity(groups []staticCommunityRoutes) (string, string) { var b4 strings.Builder var b6 strings.Builder diff --git a/internal/repository/postgres_entities.go b/internal/repository/postgres_entities.go index bc014f4..10d74e9 100644 --- a/internal/repository/postgres_entities.go +++ b/internal/repository/postgres_entities.go @@ -325,6 +325,38 @@ func (p *Postgres) UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, return nil } +func (p *Postgres) UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []store.ASEntryResolveMetaUpdate, resolvedAt time.Time) error { + if len(updates) == 0 { + return nil + } + if _, err := p.GetModule(tenantID, moduleID); err != nil { + return err + } + ctx := context.Background() + batch := &pgx.Batch{} + for _, u := range updates { + var nameArg any + sn := strings.TrimSpace(u.ASNName) + if sn == "" { + nameArg = nil + } else { + nameArg = sn + } + batch.Queue(` + UPDATE module_as_entry SET asn_name=$3, prefix_count=$4, asn_resolved_at=$5, updated_at=now() + WHERE id=$1 AND module_id=$2`, + u.EntryID, moduleID, nameArg, u.PrefixCount, resolvedAt.UTC()) + } + br := p.pool.SendBatch(ctx, batch) + defer func() { _ = br.Close() }() + for range updates { + if _, err := br.Exec(); err != nil { + return err + } + } + return nil +} + func (p *Postgres) DeleteASEntry(tenantID, moduleID, entryID string) error { if _, err := p.GetModule(tenantID, moduleID); err != nil { return err diff --git a/internal/store/as_meta_batch.go b/internal/store/as_meta_batch.go new file mode 100644 index 0000000..8d1dd89 --- /dev/null +++ b/internal/store/as_meta_batch.go @@ -0,0 +1,8 @@ +package store + +// ASEntryResolveMetaUpdate is one row for batch AS resolve metadata writes. +type ASEntryResolveMetaUpdate struct { + EntryID string + ASNName string + PrefixCount int64 +} diff --git a/internal/store/backend.go b/internal/store/backend.go index 42ca694..505a76c 100644 --- a/internal/store/backend.go +++ b/internal/store/backend.go @@ -18,6 +18,8 @@ type Backend interface { // ListModules returns all modules for a tenant (control plane may paginate in httpapi). ListModules(tenantID string) []*Module + // ListModulesPage returns one page of modules (limit capped by caller). + ListModulesPage(tenantID, cursor string, limit int) ([]*Module, string, bool) GetModule(tenantID, moduleID string) (*Module, error) CreateModule(tenantID string, in *Module) (*Module, error) UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*Module, error) @@ -34,6 +36,7 @@ type Backend interface { DeleteASEntry(tenantID, moduleID, entryID string) error // UpdateASEntryResolveMeta записывает имя AS, число объявленных префиксов и время успешного резолва (pipeline). UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, asnName string, prefixCount int64, resolvedAt time.Time) error + UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []ASEntryResolveMetaUpdate, resolvedAt time.Time) error ListDomainEntries(tenantID, moduleID string) ([]*DomainEntry, error) CreateDomainEntry(tenantID, moduleID string, in *DomainEntry) (*DomainEntry, error) diff --git a/internal/store/memory_crud.go b/internal/store/memory_crud.go index cff51c5..aa2c0ed 100644 --- a/internal/store/memory_crud.go +++ b/internal/store/memory_crud.go @@ -301,6 +301,15 @@ func (m *Memory) UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, as return nil } +func (m *Memory) UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []ASEntryResolveMetaUpdate, resolvedAt time.Time) error { + for _, u := range updates { + if err := m.UpdateASEntryResolveMeta(tenantID, moduleID, u.EntryID, u.ASNName, u.PrefixCount, resolvedAt); err != nil { + return err + } + } + return nil +} + func (m *Memory) DeleteASEntry(tenantID, moduleID, entryID string) error { m.mu.Lock() defer m.mu.Unlock() diff --git a/migrations/postgres/000012_perf_indexes.down.sql b/migrations/postgres/000012_perf_indexes.down.sql new file mode 100644 index 0000000..85b9ba9 --- /dev/null +++ b/migrations/postgres/000012_perf_indexes.down.sql @@ -0,0 +1,6 @@ +DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_id; +DROP INDEX IF EXISTS idx_config_revision_tenant_module_created; +DROP INDEX IF EXISTS idx_bgp_speaker_published; +DROP INDEX IF EXISTS idx_bgp_speaker_last_applied; +DROP INDEX IF EXISTS idx_module_default_community; +DROP INDEX IF EXISTS idx_module_doh_profile_id; diff --git a/migrations/postgres/000012_perf_indexes.up.sql b/migrations/postgres/000012_perf_indexes.up.sql new file mode 100644 index 0000000..215db8b --- /dev/null +++ b/migrations/postgres/000012_perf_indexes.up.sql @@ -0,0 +1,17 @@ +CREATE INDEX IF NOT EXISTS idx_module_doh_profile_id + ON module (doh_profile_id) WHERE deleted_at IS NULL AND doh_profile_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_module_default_community + ON module (default_community_id) WHERE deleted_at IS NULL AND default_community_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_bgp_speaker_last_applied + ON bgp_speaker (last_applied_revision_id) WHERE last_applied_revision_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_bgp_speaker_published + ON bgp_speaker (published_revision_id) WHERE published_revision_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_config_revision_tenant_module_created + ON config_revision (tenant_id, module_id, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_id + ON revision_materialized_prefix (revision_id, id); diff --git a/migrations/sqlite/000012_perf_indexes.down.sql b/migrations/sqlite/000012_perf_indexes.down.sql new file mode 100644 index 0000000..85b9ba9 --- /dev/null +++ b/migrations/sqlite/000012_perf_indexes.down.sql @@ -0,0 +1,6 @@ +DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_id; +DROP INDEX IF EXISTS idx_config_revision_tenant_module_created; +DROP INDEX IF EXISTS idx_bgp_speaker_published; +DROP INDEX IF EXISTS idx_bgp_speaker_last_applied; +DROP INDEX IF EXISTS idx_module_default_community; +DROP INDEX IF EXISTS idx_module_doh_profile_id; diff --git a/migrations/sqlite/000012_perf_indexes.up.sql b/migrations/sqlite/000012_perf_indexes.up.sql new file mode 100644 index 0000000..215db8b --- /dev/null +++ b/migrations/sqlite/000012_perf_indexes.up.sql @@ -0,0 +1,17 @@ +CREATE INDEX IF NOT EXISTS idx_module_doh_profile_id + ON module (doh_profile_id) WHERE deleted_at IS NULL AND doh_profile_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_module_default_community + ON module (default_community_id) WHERE deleted_at IS NULL AND default_community_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_bgp_speaker_last_applied + ON bgp_speaker (last_applied_revision_id) WHERE last_applied_revision_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_bgp_speaker_published + ON bgp_speaker (published_revision_id) WHERE published_revision_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_config_revision_tenant_module_created + ON config_revision (tenant_id, module_id, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_id + ON revision_materialized_prefix (revision_id, id); diff --git a/web/src/lib/ui/patterns/virtual-list/virtual-prefix-list.svelte b/web/src/lib/ui/patterns/virtual-list/virtual-prefix-list.svelte new file mode 100644 index 0000000..dfb3daf --- /dev/null +++ b/web/src/lib/ui/patterns/virtual-list/virtual-prefix-list.svelte @@ -0,0 +1,37 @@ + + +
{ + scrollTop = e.currentTarget.scrollTop; + }} +> +
+ {#each visibleItems as pfx, i (`${startIndex + i}-${pfx}`)} +

+ {pfx} +

+ {:else} +

Нет префиксов

+ {/each} +
+
diff --git a/web/src/routes/operations/+page.svelte b/web/src/routes/operations/+page.svelte index 885ce03..899bfca 100644 --- a/web/src/routes/operations/+page.svelte +++ b/web/src/routes/operations/+page.svelte @@ -60,6 +60,7 @@ ReportRow } from '$lib/components/operations/types.js'; import ScrollPreBlock from '$lib/components/app/scroll-pre-block.svelte'; + import VirtualPrefixList from '$lib/ui/patterns/virtual-list/virtual-prefix-list.svelte'; import CardSkeleton from '$lib/ui/patterns/feedback/card-skeleton.svelte'; import KpiMetricsGrid from '$lib/ui/patterns/kpi/kpi-metrics-grid.svelte'; import { confirm } from '$lib/ui/patterns/confirm/confirm-state.svelte.js'; @@ -1105,15 +1106,7 @@ Префиксов: {prefixesData.length}

-
- {#each prefixesData as pfx, idx (`${idx}-${pfx}`)} -

{pfx}

- {:else} -

Нет префиксов

- {/each} -
+