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

@@ -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 {

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))
+		p := strings.TrimSpace(row.Prefix)
-		if err != nil || !pfx.Addr().Is4() {
+		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, aggregateGroupedRows(groupedV4, aggregateIPv4Group)...)
-		out = append(out, aggregateIPv4Group(grp)...)
+	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).
+type prefixHashLine struct{ p, c, s string }
-func hashAggregatedMaterialization(tenantID string, rows []store.PrefixRow) string {
+
-	type line struct{ p, c, s string }
+func dedupeSortedPrefixLines(rows []store.PrefixRow) []prefixHashLine {
-	var lines []line
+	seen := make(map[string]struct{}, len(rows))
+	lines := make([]prefixHashLine, 0, len(rows))
 	for _, r := range rows {
-		c := ""
+		c := prefixRowCommunity(r)
-		if r.CommunityID != nil {
+		key := r.Prefix + "\x00" + c + "\x00" + r.Source
-			c = *r.CommunityID
+		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 }
+	lines := dedupeSortedPrefixLines(rows)
-	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
-	})
 	h := sha256.New()
-	h.Write([]byte(strings.TrimSpace(tenantID)))
+	writePrefixLinesHash(h, tenantID, lines)
-	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})
-	}
 	h.Write([]byte("peers"))
 	h.Write([]byte{0})
 	peers := st.ListPeers(tenantID)

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)
+	}
+}

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