From 07b97bddc1d6cbce222fa3db5e67ccbcfec88d8c Mon Sep 17 00:00:00 2001 From: Denozordec Date: Tue, 19 May 2026 15:38:33 +0700 Subject: [PATCH] feat: enhance observability with prefix aggregation metrics - Added new histograms to track prefix aggregation duration, raw count, and aggregated count during tenant rendering. - Implemented `RecordPrefixAggregation` function to record metrics for aggregation performance. - Updated `RenderTenantRevision` and `RenderTenantRevisionFromPrefixes` functions to log aggregation statistics. - Refactored `smartAggregatePrefixRows` to support both IPv4 and IPv6 aggregation, improving overall prefix handling. --- internal/observability/metrics.go | 34 ++++ internal/pipeline/refresh.go | 210 +++++++++++++++----- internal/pipeline/refresh_aggregate_test.go | 150 ++++++++++++++ internal/store/memory.go | 14 +- 4 files changed, 358 insertions(+), 50 deletions(-) diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index 87850f0..4c7460e 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -59,8 +59,42 @@ var ( Name: "build_info", Help: "Build metadata (value always 1).", }, []string{"version", "git_sha"}) + + prefixAggregationDuration = promauto.NewHistogram(prometheus.HistogramOpts{ + Namespace: namespace, + Name: "prefix_aggregation_duration_seconds", + Help: "Time spent in smartAggregatePrefixRows during tenant render.", + Buckets: prometheus.ExponentialBuckets(0.0001, 2, 16), + }) + + prefixAggregationRawCount = promauto.NewHistogram(prometheus.HistogramOpts{ + Namespace: namespace, + Name: "prefix_aggregation_raw_count", + Help: "Prefix row count before CIDR aggregation on tenant render.", + Buckets: prometheus.ExponentialBuckets(1, 2, 16), + }) + + prefixAggregationAggregatedCount = promauto.NewHistogram(prometheus.HistogramOpts{ + Namespace: namespace, + Name: "prefix_aggregation_aggregated_count", + Help: "Prefix row count after CIDR aggregation on tenant render.", + Buckets: prometheus.ExponentialBuckets(1, 2, 16), + }) ) +// RecordPrefixAggregation records tenant render CIDR aggregation stats. +func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Duration) { + if rawCount < 0 { + rawCount = 0 + } + if aggregatedCount < 0 { + aggregatedCount = 0 + } + prefixAggregationDuration.Observe(duration.Seconds()) + prefixAggregationRawCount.Observe(float64(rawCount)) + prefixAggregationAggregatedCount.Observe(float64(aggregatedCount)) +} + // RecordJobTerminal increments jobs_finished_total for terminal statuses. func RecordJobTerminal(kind, status string) { switch status { diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index 7635a32..56be6bd 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -8,6 +8,7 @@ import ( "encoding/json" "fmt" "io" + "math/big" "net" "net/http" "net/netip" @@ -18,6 +19,7 @@ import ( "time" "evobgp/internal/birdfmt" + "evobgp/internal/observability" "evobgp/internal/store" "github.com/google/uuid" @@ -79,11 +81,14 @@ func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client if err != nil { return "", err } + rawCount := len(agg) + aggStart := time.Now() + agg = smartAggregatePrefixRows(agg) + observability.RecordPrefixAggregation(rawCount, len(agg), time.Since(aggStart)) hash := hashAggregatedMaterializationWithPeers(st, tenantID, agg) if prev := latestTenantRevision(st, tenantID); prev != nil && prev.ContentHash == hash { return prev.ID, nil } - agg = smartAggregatePrefixRows(agg) revisionID = uuid.NewString() parent := parentRevision(st, tenantID, triggerModuleID) @@ -106,11 +111,14 @@ func RenderTenantRevisionFromPrefixes(ctx context.Context, st store.Backend, hc hc = http.DefaultClient } agg := append([]store.PrefixRow(nil), rows...) + rawCount := len(agg) + aggStart := time.Now() + agg = smartAggregatePrefixRows(agg) + observability.RecordPrefixAggregation(rawCount, len(agg), time.Since(aggStart)) hash := hashAggregatedMaterializationWithPeers(st, tenantID, agg) if prev := latestTenantRevision(st, tenantID); prev != nil && prev.ContentHash == hash { return prev.ID, nil } - agg = smartAggregatePrefixRows(agg) revisionID = uuid.NewString() parent := parentRevision(st, tenantID, triggerModuleID) @@ -426,14 +434,20 @@ type prefixGroupKey struct { source string } -// smartAggregatePrefixRows performs "safe" IPv4 CIDR aggregation after full tenant materialization. +// smartAggregatePrefixRows performs "safe" IPv4/IPv6 CIDR aggregation after full tenant materialization. // We aggregate only inside identical community/source groups to preserve BIRD attributes semantics. func smartAggregatePrefixRows(rows []store.PrefixRow) []store.PrefixRow { - grouped := make(map[prefixGroupKey][]store.PrefixRow) + groupedV4 := make(map[prefixGroupKey][]store.PrefixRow) + groupedV6 := make(map[prefixGroupKey][]store.PrefixRow) var passthrough []store.PrefixRow for _, row := range rows { - pfx, err := netip.ParsePrefix(strings.TrimSpace(row.Prefix)) - if err != nil || !pfx.Addr().Is4() { + p := strings.TrimSpace(row.Prefix) + if strings.HasPrefix(p, "as:") { + passthrough = append(passthrough, row) + continue + } + pfx, err := netip.ParsePrefix(p) + if err != nil { passthrough = append(passthrough, row) continue } @@ -443,17 +457,73 @@ func smartAggregatePrefixRows(rows []store.PrefixRow) []store.PrefixRow { } r := row r.Prefix = pfx.Masked().String() - grouped[k] = append(grouped[k], r) + switch { + case pfx.Addr().Is4(): + groupedV4[k] = append(groupedV4[k], r) + case pfx.Addr().Is6(): + groupedV6[k] = append(groupedV6[k], r) + default: + passthrough = append(passthrough, row) + } } out := append([]store.PrefixRow{}, passthrough...) - for _, grp := range grouped { - out = append(out, aggregateIPv4Group(grp)...) + out = append(out, aggregateGroupedRows(groupedV4, aggregateIPv4Group)...) + out = append(out, aggregateGroupedRows(groupedV6, aggregateIPv6Group)...) + sortPrefixRows(out) + return out +} + +func aggregateGroupedRows(grouped map[prefixGroupKey][]store.PrefixRow, aggregateFn func([]store.PrefixRow) []store.PrefixRow) []store.PrefixRow { + if len(grouped) == 0 { + return nil + } + keys := make([]prefixGroupKey, 0, len(grouped)) + for k := range grouped { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].community != keys[j].community { + return keys[i].community < keys[j].community + } + return keys[i].source < keys[j].source + }) + var out []store.PrefixRow + for _, k := range keys { + out = append(out, aggregateFn(grouped[k])...) } return out } +func sortPrefixRows(rows []store.PrefixRow) { + sort.Slice(rows, func(i, j int) bool { + if rows[i].Prefix != rows[j].Prefix { + return rows[i].Prefix < rows[j].Prefix + } + ci, cj := prefixRowCommunity(rows[i]), prefixRowCommunity(rows[j]) + if ci != cj { + return ci < cj + } + return rows[i].Source < rows[j].Source + }) +} + +func prefixRowCommunity(r store.PrefixRow) string { + if r.CommunityID != nil { + return *r.CommunityID + } + return "" +} + func aggregateIPv4Group(rows []store.PrefixRow) []store.PrefixRow { + return aggregateCIDRGroup(rows, mergeSiblingPrefixesIPv4) +} + +func aggregateIPv6Group(rows []store.PrefixRow) []store.PrefixRow { + return aggregateCIDRGroup(rows, mergeSiblingPrefixesIPv6) +} + +func aggregateCIDRGroup(rows []store.PrefixRow, mergeFn func(map[string]store.PrefixRow) bool) []store.PrefixRow { if len(rows) <= 1 { return rows } @@ -463,7 +533,7 @@ func aggregateIPv4Group(rows []store.PrefixRow) []store.PrefixRow { } pruneCoveredPrefixes(set) for { - if !mergeSiblingPrefixes(set) { + if !mergeFn(set) { break } pruneCoveredPrefixes(set) @@ -472,6 +542,7 @@ func aggregateIPv4Group(rows []store.PrefixRow) []store.PrefixRow { for _, row := range set { out = append(out, row) } + sortPrefixRows(out) return out } @@ -484,7 +555,7 @@ func pruneCoveredPrefixes(set map[string]store.PrefixRow) { items := make([]item, 0, len(set)) for k := range set { p, err := netip.ParsePrefix(k) - if err != nil || !p.Addr().Is4() { + if err != nil { continue } items = append(items, item{key: k, pfx: p, bits: p.Bits()}) @@ -507,7 +578,7 @@ func pruneCoveredPrefixes(set map[string]store.PrefixRow) { } } -func mergeSiblingPrefixes(set map[string]store.PrefixRow) bool { +func mergeSiblingPrefixesIPv4(set map[string]store.PrefixRow) bool { merged := false seen := make(map[string]struct{}, len(set)) for key, row := range set { @@ -555,6 +626,60 @@ func u32ToIPv4(v uint32) netip.Addr { return netip.AddrFrom4([4]byte{byte(v >> 24), byte(v >> 16), byte(v >> 8), byte(v)}) } +func mergeSiblingPrefixesIPv6(set map[string]store.PrefixRow) bool { + merged := false + seen := make(map[string]struct{}, len(set)) + for key, row := range set { + if _, done := seen[key]; done { + continue + } + pfx, err := netip.ParsePrefix(key) + if err != nil || !pfx.Addr().Is6() { + continue + } + bits := pfx.Bits() + if bits <= 16 { + continue + } + netNum := ipv6PrefixNetwork(pfx) + blockSize := new(big.Int).Lsh(big.NewInt(1), uint(128-bits)) + siblingNet := new(big.Int).Xor(netNum, blockSize) + siblingPfx := ipv6PrefixFromBigInt(siblingNet, bits).String() + if _, ok := set[siblingPfx]; !ok { + continue + } + parentBits := bits - 1 + parentBlock := new(big.Int).Lsh(big.NewInt(1), uint(128-parentBits)) + mask := new(big.Int).Sub(parentBlock, big.NewInt(1)) + mask.Not(mask) + parentNet := new(big.Int).And(netNum, mask) + parentPfx := ipv6PrefixFromBigInt(parentNet, parentBits).String() + delete(set, key) + delete(set, siblingPfx) + parentRow := row + parentRow.Prefix = parentPfx + set[parentPfx] = parentRow + seen[key] = struct{}{} + seen[siblingPfx] = struct{}{} + merged = true + } + return merged +} + +func ipv6PrefixNetwork(p netip.Prefix) *big.Int { + a := p.Masked().Addr().As16() + n := new(big.Int) + n.SetBytes(a[:]) + return n +} + +func ipv6PrefixFromBigInt(n *big.Int, bits int) netip.Prefix { + b := n.Bytes() + var a [16]byte + copy(a[16-len(b):], b) + return netip.PrefixFrom(netip.AddrFrom16(a), bits).Masked() +} + func parentRevision(st store.Backend, tenantID, moduleID string) *string { items, _, _ := st.ListRevisions(tenantID, moduleID, "", 1) if len(items) == 0 { @@ -573,16 +698,19 @@ func latestTenantRevision(st store.Backend, tenantID string) *store.Revision { return items[0] } -// 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 +type prefixHashLine struct{ p, c, s string } + +func dedupeSortedPrefixLines(rows []store.PrefixRow) []prefixHashLine { + seen := make(map[string]struct{}, len(rows)) + lines := make([]prefixHashLine, 0, len(rows)) for _, r := range rows { - c := "" - if r.CommunityID != nil { - c = *r.CommunityID + c := prefixRowCommunity(r) + key := r.Prefix + "\x00" + c + "\x00" + r.Source + if _, ok := seen[key]; ok { + continue } - lines = append(lines, line{r.Prefix, c, r.Source}) + seen[key] = struct{}{} + lines = append(lines, prefixHashLine{r.Prefix, c, r.Source}) } sort.Slice(lines, func(i, j int) bool { if lines[i].p != lines[j].p { @@ -593,7 +721,10 @@ func hashAggregatedMaterialization(tenantID string, rows []store.PrefixRow) stri } return lines[i].s < lines[j].s }) - h := sha256.New() + return lines +} + +func writePrefixLinesHash(h interface{ Write([]byte) (int, error) }, tenantID string, lines []prefixHashLine) { h.Write([]byte(strings.TrimSpace(tenantID))) h.Write([]byte{0}) for _, l := range lines { @@ -604,39 +735,20 @@ func hashAggregatedMaterialization(tenantID string, rows []store.PrefixRow) stri h.Write([]byte(l.s)) h.Write([]byte{0}) } +} + +// hashAggregatedMaterialization hashes the post-aggregation tenant-wide prefix set used for BIRD. +func hashAggregatedMaterialization(tenantID string, rows []store.PrefixRow) string { + lines := dedupeSortedPrefixLines(rows) + h := sha256.New() + writePrefixLinesHash(h, tenantID, lines) return fmt.Sprintf("sha256:%x", h.Sum(nil)) } func hashAggregatedMaterializationWithPeers(st store.Backend, tenantID string, rows []store.PrefixRow) string { - type line struct{ p, c, s string } - var lines []line - for _, r := range rows { - c := "" - if r.CommunityID != nil { - c = *r.CommunityID - } - lines = append(lines, line{r.Prefix, c, r.Source}) - } - sort.Slice(lines, func(i, j int) bool { - if lines[i].p != lines[j].p { - return lines[i].p < lines[j].p - } - if lines[i].c != lines[j].c { - return lines[i].c < lines[j].c - } - return lines[i].s < lines[j].s - }) + lines := dedupeSortedPrefixLines(rows) h := sha256.New() - h.Write([]byte(strings.TrimSpace(tenantID))) - h.Write([]byte{0}) - for _, l := range lines { - h.Write([]byte(l.p)) - h.Write([]byte{1}) - h.Write([]byte(l.c)) - h.Write([]byte{1}) - h.Write([]byte(l.s)) - h.Write([]byte{0}) - } + writePrefixLinesHash(h, tenantID, lines) h.Write([]byte("peers")) h.Write([]byte{0}) peers := st.ListPeers(tenantID) diff --git a/internal/pipeline/refresh_aggregate_test.go b/internal/pipeline/refresh_aggregate_test.go index 890d9e3..796af89 100644 --- a/internal/pipeline/refresh_aggregate_test.go +++ b/internal/pipeline/refresh_aggregate_test.go @@ -8,6 +8,15 @@ import ( "evobgp/internal/store" ) +func prefixRowSet(rows []store.PrefixRow) map[string]struct{} { + got := make(map[string]struct{}, len(rows)) + for _, r := range rows { + c := prefixRowCommunity(r) + got[r.Prefix+"|"+c+"|"+r.Source] = struct{}{} + } + return got +} + func TestRefreshModule_AggregatesAllEnabledModules(t *testing.T) { t.Setenv("EVOBGP_ASN_RESOLVE", "0") @@ -119,3 +128,144 @@ func TestSmartAggregatePrefixRows_RespectsCommunityAndSource(t *testing.T) { t.Fatalf("expected 3 resulting rows, got %d: %+v", len(out), out) } } + +func TestSmartAggregatePrefixRows_PruneCoveredPrefixes(t *testing.T) { + rows := []store.PrefixRow{ + {Prefix: "10.0.0.0/16", Source: "ip_range"}, + {Prefix: "10.0.0.0/24", Source: "ip_range"}, + {Prefix: "10.0.1.0/24", Source: "ip_range"}, + } + out := smartAggregatePrefixRows(rows) + got := prefixRowSet(out) + if len(out) != 1 { + t.Fatalf("expected 1 row after covered prune, got %d: %+v", len(out), out) + } + if _, ok := got["10.0.0.0/16||ip_range"]; !ok { + t.Fatalf("expected /16 only, got: %+v", out) + } +} + +func TestSmartAggregatePrefixRows_MergeChainFourSlash24(t *testing.T) { + rows := []store.PrefixRow{ + {Prefix: "10.0.0.0/24", Source: "ip_range"}, + {Prefix: "10.0.1.0/24", Source: "ip_range"}, + {Prefix: "10.0.2.0/24", Source: "ip_range"}, + {Prefix: "10.0.3.0/24", Source: "ip_range"}, + } + out := smartAggregatePrefixRows(rows) + got := prefixRowSet(out) + if _, ok := got["10.0.0.0/22||ip_range"]; !ok { + t.Fatalf("expected merged /22, got: %+v", out) + } + if len(out) != 1 { + t.Fatalf("expected 1 row, got %d: %+v", len(out), out) + } +} + +func TestSmartAggregatePrefixRows_DoesNotMergeAdjacentSlash8(t *testing.T) { + rows := []store.PrefixRow{ + {Prefix: "10.0.0.0/8", Source: "ip_range"}, + {Prefix: "11.0.0.0/8", Source: "ip_range"}, + } + out := smartAggregatePrefixRows(rows) + if len(out) != 2 { + t.Fatalf("expected two /8 prefixes, got %d: %+v", len(out), out) + } +} + +func TestSmartAggregatePrefixRows_MergesIPv6Slash64(t *testing.T) { + rows := []store.PrefixRow{ + {Prefix: "2001:db8:0:0::/64", Source: "ip_range"}, + {Prefix: "2001:db8:0:1::/64", Source: "ip_range"}, + } + out := smartAggregatePrefixRows(rows) + got := prefixRowSet(out) + if _, ok := got["2001:db8::/63||ip_range"]; !ok { + t.Fatalf("expected merged IPv6 /63, got: %+v", out) + } + if len(out) != 1 { + t.Fatalf("expected 1 row, got %d: %+v", len(out), out) + } +} + +func TestSmartAggregatePrefixRows_StableOutputOrder(t *testing.T) { + rows := []store.PrefixRow{ + {Prefix: "192.168.0.0/24", Source: "cdn:b"}, + {Prefix: "10.0.0.0/24", Source: "ip_range"}, + {Prefix: "10.0.1.0/24", Source: "ip_range"}, + } + out1 := smartAggregatePrefixRows(rows) + out2 := smartAggregatePrefixRows(rows) + if len(out1) != len(out2) { + t.Fatalf("length mismatch: %d vs %d", len(out1), len(out2)) + } + for i := range out1 { + if out1[i].Prefix != out2[i].Prefix || out1[i].Source != out2[i].Source { + t.Fatalf("order not stable at %d: %+v vs %+v", i, out1[i], out2[i]) + } + } +} + +func TestRenderTenantRevisionFromPrefixes_SkipsWhenRedundantRawRows(t *testing.T) { + t.Setenv("EVOBGP_ASN_RESOLVE", "0") + + m := store.NewMemory() + m.SeedDemo() + tenant, _, modIP, _, _ := m.DemoIDs() + for _, mod := range m.ListModules(tenant) { + if mod == nil || mod.ID == modIP { + continue + } + disabled := false + if _, err := m.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil { + t.Fatal(err) + } + } + + ctx := context.Background() + baseRows := []store.PrefixRow{{Prefix: "10.0.0.0/16", Source: "ip_range"}} + rev1, err := RenderTenantRevisionFromPrefixes(ctx, m, http.DefaultClient, tenant, modIP, baseRows) + if err != nil { + t.Fatal(err) + } + + rev2, err := RenderTenantRevisionFromPrefixes(ctx, m, http.DefaultClient, tenant, modIP, append(baseRows, + store.PrefixRow{Prefix: "10.0.0.0/24", Source: "ip_range"}, + )) + if err != nil { + t.Fatal(err) + } + if rev2 != rev1 { + t.Fatalf("expected hash skip for redundant covered prefix, got rev1=%s rev2=%s", rev1, rev2) + } + + before, _, _ := m.ListRevisions(tenant, "", "", 200) + rev3, err := RenderTenantRevisionFromPrefixes(ctx, m, http.DefaultClient, tenant, modIP, append(baseRows, + store.PrefixRow{Prefix: "10.0.0.0/24", Source: "ip_range"}, + store.PrefixRow{Prefix: "10.0.0.0/24", Source: "ip_range"}, + )) + if err != nil { + t.Fatal(err) + } + if rev3 != rev1 { + t.Fatalf("expected hash skip for duplicate raw rows, got rev1=%s rev3=%s", rev1, rev3) + } + after, _, _ := m.ListRevisions(tenant, "", "", 200) + if len(after) != len(before) { + t.Fatalf("duplicate raw rows should not create revision, before=%d after=%d", len(before), len(after)) + } +} + +func TestHashAggregatedMaterialization_DedupesIdenticalRows(t *testing.T) { + rows := []store.PrefixRow{ + {Prefix: "10.0.0.0/24", Source: "ip_range"}, + {Prefix: "10.0.0.0/24", Source: "ip_range"}, + } + h1 := hashAggregatedMaterialization("tenant-a", rows) + h2 := hashAggregatedMaterialization("tenant-a", []store.PrefixRow{ + {Prefix: "10.0.0.0/24", Source: "ip_range"}, + }) + if h1 != h2 { + t.Fatalf("expected deduped hash to match, got %s vs %s", h1, h2) + } +} diff --git a/internal/store/memory.go b/internal/store/memory.go index 94b41ba..de875ee 100644 --- a/internal/store/memory.go +++ b/internal/store/memory.go @@ -327,13 +327,22 @@ func (m *Memory) CreateRenderRevision(revisionID, tenantID, moduleID string, par frag[k] = v } parent := parentRevisionID + createdAt := time.Now().UTC() + for _, r := range m.revisions { + if r == nil || r.TenantID != tenantID { + continue + } + if !r.CreatedAt.Before(createdAt) { + createdAt = r.CreatedAt.Add(time.Microsecond) + } + } m.revisions[revisionID] = &Revision{ ID: revisionID, TenantID: tenantID, ModuleID: moduleID, ContentHash: contentHash, ParentRevisionID: parent, - CreatedAt: time.Now().UTC(), + CreatedAt: createdAt, MaterializedPrefixCount: len(prefixes), PreviewFragments: frag, } @@ -657,6 +666,9 @@ func (m *Memory) ListRevisions(tenantID, moduleID string, cursor string, limit i all = append(all, r) } sort.Slice(all, func(i, j int) bool { + if all[i].CreatedAt.Equal(all[j].CreatedAt) { + return all[i].ID > all[j].ID + } return all[i].CreatedAt.After(all[j].CreatedAt) }) off := 0