quality / commitlint (push) Skipped
quality / changes (push) Successful in 9s
quality / docker-check (push) Skipped
quality / openapi (push) Successful in 46s
quality / web (push) Successful in 1m16s
quality / go (push) Successful in 2m42s
quality / bird2 (push) Successful in 16s
CD / quality (push) Successful in 5m19s
CD / publish (push) Successful in 7m19s
- Deleted unused components: `DashboardActivityTimeline`, `DashboardFramePanel`, `DashboardModulesGrid`, `DashboardRecentJobsGrid`, and `DashboardRecentRevisionsGrid` to streamline the dashboard. - Updated `DashboardKpiGrid` to improve KPI display logic, including progress indicators and enhanced badge functionality. - Refactored `DashboardNetworkHealth` to provide better status representation based on loading states and network conditions. - Introduced new properties for KPI cards to support progress tracking and improved visual feedback. This cleanup aims to enhance performance and maintainability of the dashboard while providing a better user experience.
985 lines
28 KiB
Go
985 lines
28 KiB
Go
package pipeline
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
"net/netip"
|
||
"net/url"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"evobgp/internal/birdfmt"
|
||
"evobgp/internal/httpclient"
|
||
"evobgp/internal/observability"
|
||
"evobgp/internal/store"
|
||
|
||
"github.com/google/uuid"
|
||
"github.com/miekg/dns"
|
||
)
|
||
|
||
const (
|
||
birdFilterNameV4 = "evobgp_export_v4"
|
||
birdFilterNameV6 = "evobgp_export_v6"
|
||
auxBirdFullExpanded = "_bird_full_expanded.conf"
|
||
revisionTTLKey = RevisionRetentionKey
|
||
)
|
||
|
||
// 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)
|
||
}
|
||
|
||
// RefreshModuleIngest runs ingest for one module and persists side-effects (ASN metadata, CDN etags, etc).
|
||
// It does not create a new config revision.
|
||
func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) error {
|
||
if hc == nil {
|
||
hc = httpclient.New(httpclient.DefaultTimeout)
|
||
}
|
||
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")
|
||
}
|
||
|
||
var prior []store.PrefixRow
|
||
if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil {
|
||
prior = snap.Prefixes
|
||
}
|
||
rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, prior)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := persistModuleSnapshot(st, tenantID, mod, rows); err != nil {
|
||
return err
|
||
}
|
||
refreshedAt := time.Now().UTC()
|
||
_, _ = st.UpdateModule(tenantID, moduleID, &store.ModulePatch{LastRefreshedAt: &refreshedAt})
|
||
return nil
|
||
}
|
||
|
||
// RenderTenantRevision renders one tenant-wide revision using current data from all enabled modules.
|
||
// If materialized prefixes are unchanged, returns latest revision id without creating a duplicate.
|
||
func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string) (revisionID string, err error) {
|
||
if hc == nil {
|
||
hc = httpclient.New(httpclient.DefaultTimeout)
|
||
}
|
||
agg, err := aggregateTenantPrefixRowsAll(ctx, st, hc, tenantID)
|
||
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
|
||
}
|
||
|
||
revisionID = uuid.NewString()
|
||
parent := parentRevision(st, tenantID, triggerModuleID)
|
||
preview, err := buildPreviewFragments(st, tenantID, triggerModuleID, revisionID, agg)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if err := st.CreateRenderRevision(revisionID, tenantID, triggerModuleID, parent, hash, preview, agg); err != nil {
|
||
return "", err
|
||
}
|
||
applyRevisionRetention(st, tenantID)
|
||
return revisionID, nil
|
||
}
|
||
|
||
// RenderTenantRevisionFromPrefixes renders one tenant-wide revision from already materialized prefixes.
|
||
// This is used for fast paths (e.g. peer-only changes) to avoid ingest/external fetches.
|
||
func RenderTenantRevisionFromPrefixes(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string, rows []store.PrefixRow) (revisionID string, err error) {
|
||
_ = ctx
|
||
if hc == nil {
|
||
hc = httpclient.New(httpclient.DefaultTimeout)
|
||
}
|
||
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
|
||
}
|
||
|
||
revisionID = uuid.NewString()
|
||
parent := parentRevision(st, tenantID, triggerModuleID)
|
||
preview, err := buildPreviewFragments(st, tenantID, triggerModuleID, revisionID, agg)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if err := st.CreateRenderRevision(revisionID, tenantID, triggerModuleID, parent, hash, preview, agg); err != nil {
|
||
return "", err
|
||
}
|
||
applyRevisionRetention(st, tenantID)
|
||
return revisionID, nil
|
||
}
|
||
|
||
// RefreshModule keeps backwards-compatible behavior: module ingest + immediate tenant render.
|
||
func RefreshModule(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) (revisionID string, err error) {
|
||
if err := RefreshModuleIngest(ctx, st, hc, tenantID, moduleID); err != nil {
|
||
return "", err
|
||
}
|
||
return RenderTenantRevision(ctx, st, hc, tenantID, moduleID)
|
||
}
|
||
|
||
// collectModulePrefixRows returns materialized prefix rows for a single module (source of truth from store / ASN resolve / CDN fetch).
|
||
// priorSnapshot is the last stored module snapshot (used to skip CDN fetches when refresh interval has not elapsed).
|
||
func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) {
|
||
moduleID := mod.ID
|
||
switch mod.Type {
|
||
case "IP_RANGES":
|
||
list, err := st.ListIPRangeEntries(tenantID, moduleID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var rows []store.PrefixRow
|
||
for _, e := range list {
|
||
comm := e.CommunityID
|
||
if comm == nil && mod.DefaultCommunityID != nil {
|
||
c := *mod.DefaultCommunityID
|
||
comm = &c
|
||
}
|
||
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 nil, err
|
||
}
|
||
sort.Slice(list, func(i, j int) bool { return list[i].ASN < list[j].ASN })
|
||
return collectASPrefixRows(ctx, st, hc, tenantID, mod, list, priorSnapshot)
|
||
case "CDN_CIDRS":
|
||
sources, err := st.ListCDNSources(tenantID, moduleID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return collectCDNPrefixRows(ctx, st, hc, tenantID, mod, sources, priorSnapshot)
|
||
case "DOMAINS":
|
||
entries, err := st.ListDomainEntries(tenantID, moduleID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
profiles, policy, err := loadModuleDohProfiles(st, tenantID, mod)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return collectDomainPrefixRows(ctx, st, hc, mod, profiles, policy, entries, priorSnapshot)
|
||
default:
|
||
return nil, fmt.Errorf("pipeline: unknown module type %q", mod.Type)
|
||
}
|
||
}
|
||
|
||
func shouldSkipCDNSourceFetch(src *store.CDNSource, now time.Time) bool {
|
||
if src == nil || src.RefreshIntervalSec == nil || *src.RefreshIntervalSec <= 0 || src.LastRefreshedAt == nil {
|
||
return false
|
||
}
|
||
nextRefreshAt := src.LastRefreshedAt.UTC().Add(time.Duration(*src.RefreshIntervalSec) * time.Second)
|
||
return now.UTC().Before(nextRefreshAt)
|
||
}
|
||
|
||
type dohJSONAnswer struct {
|
||
Type int `json:"type"`
|
||
Data string `json:"data"`
|
||
}
|
||
|
||
type dohJSONResponse struct {
|
||
Answer []dohJSONAnswer `json:"Answer"`
|
||
}
|
||
|
||
func resolveDomainIPs(ctx context.Context, hc *http.Client, profile *store.DohProfile, fqdn string) ([]netip.Addr, error) {
|
||
host := strings.TrimSpace(strings.TrimSuffix(fqdn, "."))
|
||
if host == "" {
|
||
return nil, nil
|
||
}
|
||
if profile == nil || strings.TrimSpace(profile.URL) == "" {
|
||
return resolveDomainWithSystemDNS(ctx, host)
|
||
}
|
||
|
||
timeout := 10 * time.Second
|
||
if profile.TimeoutMs != nil && *profile.TimeoutMs > 0 {
|
||
timeout = time.Duration(*profile.TimeoutMs) * time.Millisecond
|
||
}
|
||
dctx, cancel := context.WithTimeout(ctx, timeout)
|
||
defer cancel()
|
||
|
||
baseURL := strings.TrimSpace(profile.URL)
|
||
// A and AAAA queries run concurrently: per-domain latency drops from ~2×RTT to ~1×RTT.
|
||
v4, v6, err4, err6 := resolveDOHMessagePair(dctx, hc, baseURL, host)
|
||
if err4 != nil {
|
||
// Fallback to JSON mode for providers that only expose dns-json.
|
||
v4, err4 = resolveDomainWithDOHJSON(dctx, hc, baseURL, host, "A")
|
||
}
|
||
if err6 != nil {
|
||
v6, err6 = resolveDomainWithDOHJSON(dctx, hc, baseURL, host, "AAAA")
|
||
}
|
||
if err4 != nil && err6 != nil {
|
||
// Some DoH providers return non-JSON responses (RFC8484 dns-message, HTML error pages, etc.).
|
||
// Fall back to system resolver to avoid failing the whole module refresh.
|
||
ips, fallbackErr := resolveDomainWithSystemDNS(dctx, host)
|
||
if fallbackErr != nil {
|
||
return nil, fmt.Errorf("doh failed for A and AAAA: %v; %v; fallback dns failed: %w", err4, err6, fallbackErr)
|
||
}
|
||
return ips, nil
|
||
}
|
||
out := uniqAddrs(append(v4, v6...))
|
||
if len(out) > 0 {
|
||
return out, nil
|
||
}
|
||
// If DoH succeeds but returns no A/AAAA records, attempt system resolver as best-effort fallback.
|
||
ips, err := resolveDomainWithSystemDNS(dctx, host)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return ips, nil
|
||
}
|
||
|
||
func resolveDomainWithDOHMessage(ctx context.Context, hc *http.Client, baseURL, host string, qtype uint16) ([]netip.Addr, error) {
|
||
msg := new(dns.Msg)
|
||
msg.SetQuestion(dns.Fqdn(host), qtype)
|
||
wire, err := msg.Pack()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
u, err := url.Parse(baseURL)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
q := u.Query()
|
||
q.Set("dns", base64.RawURLEncoding.EncodeToString(wire))
|
||
u.RawQuery = q.Encode()
|
||
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
req.Header.Set("Accept", "application/dns-message")
|
||
resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer func() { _ = resp.Body.Close() }()
|
||
if resp.StatusCode != http.StatusOK {
|
||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
||
return nil, fmt.Errorf("doh dns-message status %s: %s", resp.Status, strings.TrimSpace(string(body)))
|
||
}
|
||
raw, err := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
parsed := new(dns.Msg)
|
||
if err := parsed.Unpack(raw); err != nil {
|
||
return nil, err
|
||
}
|
||
if parsed.Rcode != dns.RcodeSuccess {
|
||
return nil, fmt.Errorf("doh rcode=%s", dns.RcodeToString[parsed.Rcode])
|
||
}
|
||
var out []netip.Addr
|
||
for _, rr := range parsed.Answer {
|
||
switch x := rr.(type) {
|
||
case *dns.A:
|
||
if qtype == dns.TypeA {
|
||
if ip, ok := netip.AddrFromSlice(x.A.To4()); ok {
|
||
out = append(out, ip.Unmap())
|
||
}
|
||
}
|
||
case *dns.AAAA:
|
||
if qtype == dns.TypeAAAA {
|
||
if ip, ok := netip.AddrFromSlice(x.AAAA.To16()); ok {
|
||
out = append(out, ip.Unmap())
|
||
}
|
||
}
|
||
}
|
||
}
|
||
// Some providers may return JSON even on dns-message request.
|
||
if len(out) == 0 && bytes.Contains(bytes.ToLower(raw), []byte(`"answer"`)) {
|
||
qs := "A"
|
||
if qtype == dns.TypeAAAA {
|
||
qs = "AAAA"
|
||
}
|
||
return resolveDomainWithDOHJSON(ctx, hc, baseURL, host, qs)
|
||
}
|
||
return uniqAddrs(out), nil
|
||
}
|
||
|
||
func resolveDomainWithSystemDNS(ctx context.Context, host string) ([]netip.Addr, error) {
|
||
ips, err := net.DefaultResolver.LookupNetIP(ctx, "ip", host)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return uniqAddrs(ips), nil
|
||
}
|
||
|
||
func resolveDomainWithDOHJSON(ctx context.Context, hc *http.Client, baseURL, host, qtype string) ([]netip.Addr, error) {
|
||
u, err := url.Parse(baseURL)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
q := u.Query()
|
||
q.Set("name", host)
|
||
q.Set("type", qtype)
|
||
u.RawQuery = q.Encode()
|
||
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
req.Header.Set("Accept", "application/dns-json")
|
||
|
||
resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer func() { _ = resp.Body.Close() }()
|
||
if resp.StatusCode != http.StatusOK {
|
||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
||
return nil, fmt.Errorf("doh status %s: %s", resp.Status, strings.TrimSpace(string(body)))
|
||
}
|
||
var payload dohJSONResponse
|
||
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&payload); err != nil {
|
||
return nil, err
|
||
}
|
||
var out []netip.Addr
|
||
for _, ans := range payload.Answer {
|
||
if (qtype == "A" && ans.Type != 1) || (qtype == "AAAA" && ans.Type != 28) {
|
||
continue
|
||
}
|
||
ip, err := netip.ParseAddr(strings.TrimSpace(ans.Data))
|
||
if err != nil {
|
||
continue
|
||
}
|
||
out = append(out, ip.Unmap())
|
||
}
|
||
return uniqAddrs(out), nil
|
||
}
|
||
|
||
func uniqAddrs(in []netip.Addr) []netip.Addr {
|
||
seen := make(map[string]struct{}, len(in))
|
||
out := make([]netip.Addr, 0, len(in))
|
||
for _, a := range in {
|
||
if !a.IsValid() {
|
||
continue
|
||
}
|
||
k := a.String()
|
||
if _, ok := seen[k]; ok {
|
||
continue
|
||
}
|
||
seen[k] = struct{}{}
|
||
out = append(out, a)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func ipToHostPrefix(ip netip.Addr) string {
|
||
if !ip.IsValid() {
|
||
return ""
|
||
}
|
||
bits := 128
|
||
if ip.Is4() {
|
||
bits = 32
|
||
}
|
||
return netip.PrefixFrom(ip, bits).Masked().String()
|
||
}
|
||
|
||
type prefixGroupKey struct {
|
||
community string
|
||
source string
|
||
}
|
||
|
||
// 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 {
|
||
groupedV4 := make(map[prefixGroupKey][]store.PrefixRow)
|
||
groupedV6 := make(map[prefixGroupKey][]store.PrefixRow)
|
||
var passthrough []store.PrefixRow
|
||
for _, row := range rows {
|
||
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
|
||
}
|
||
k := prefixGroupKey{source: row.Source}
|
||
if row.CommunityID != nil {
|
||
k.community = *row.CommunityID
|
||
}
|
||
r := row
|
||
r.Prefix = pfx.Masked().String()
|
||
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...)
|
||
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 collapsePrefixGroup(rows, true)
|
||
}
|
||
|
||
func aggregateIPv6Group(rows []store.PrefixRow) []store.PrefixRow {
|
||
return collapsePrefixGroup(rows, false)
|
||
}
|
||
|
||
func parentRevision(st store.Backend, tenantID, moduleID string) *string {
|
||
items, _, _ := st.ListRevisions(tenantID, moduleID, "", 1)
|
||
if len(items) == 0 {
|
||
return nil
|
||
}
|
||
id := items[0].ID
|
||
return &id
|
||
}
|
||
|
||
// latestTenantRevision is the newest config_revision for the tenant (any module), or nil.
|
||
func latestTenantRevision(st store.Backend, tenantID string) *store.Revision {
|
||
items, _, _ := st.ListRevisions(tenantID, "", "", 1)
|
||
if len(items) == 0 {
|
||
return nil
|
||
}
|
||
return items[0]
|
||
}
|
||
|
||
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 := prefixRowCommunity(r)
|
||
key := r.Prefix + "\x00" + c + "\x00" + r.Source
|
||
if _, ok := seen[key]; ok {
|
||
continue
|
||
}
|
||
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 {
|
||
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
|
||
})
|
||
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 {
|
||
_, _ = 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})
|
||
}
|
||
}
|
||
|
||
// 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 {
|
||
lines := dedupeSortedPrefixLines(rows)
|
||
h := sha256.New()
|
||
writePrefixLinesHash(h, tenantID, lines)
|
||
h.Write([]byte("peers"))
|
||
h.Write([]byte{0})
|
||
peers := st.ListPeers(tenantID)
|
||
sort.Slice(peers, func(i, j int) bool {
|
||
if peers[i] == nil || peers[j] == nil {
|
||
return i < j
|
||
}
|
||
return peers[i].ID < peers[j].ID
|
||
})
|
||
for _, p := range peers {
|
||
if p == nil {
|
||
continue
|
||
}
|
||
speakerID := ""
|
||
if p.SpeakerID != nil {
|
||
speakerID = strings.TrimSpace(*p.SpeakerID)
|
||
}
|
||
h.Write([]byte(strings.TrimSpace(p.ID)))
|
||
h.Write([]byte{1})
|
||
h.Write([]byte(strings.TrimSpace(p.Neighbor)))
|
||
h.Write([]byte{1})
|
||
h.Write([]byte(strconv.FormatInt(p.RemoteASN, 10)))
|
||
h.Write([]byte{1})
|
||
h.Write([]byte(strconv.FormatBool(p.Enabled)))
|
||
h.Write([]byte{1})
|
||
h.Write([]byte(strings.TrimSpace(p.PoliciesJSON)))
|
||
h.Write([]byte{1})
|
||
h.Write([]byte(speakerID))
|
||
h.Write([]byte{0})
|
||
}
|
||
return fmt.Sprintf("sha256:%x", h.Sum(nil))
|
||
}
|
||
|
||
func buildPreviewFragments(st store.Backend, tenantID, moduleID, revisionID string, rows []store.PrefixRow) (map[string]string, error) {
|
||
v4, v6, pathASNs, staticGroups, err := materializeRowsForBird(st, tenantID, rows)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
f4, err := birdfmt.RenderExportFilterIPv4(birdFilterNameV4, v4, pathASNs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
f6, err := birdfmt.RenderExportFilterIPv6(birdFilterNameV6, v6, pathASNs)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
staticV4, staticV6 := renderStaticProtocolsByCommunity(staticGroups)
|
||
|
||
locals := birdLocalsFromStore(st, tenantID)
|
||
tplBody, err := birdfmt.RenderBGPTemplates(birdfmt.BGPTemplatesOptions{
|
||
LocalASN: locals.localASN,
|
||
ExportFilterV4: birdFilterNameV4,
|
||
ExportFilterV6: birdFilterNameV6,
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
peersBody, err := renderPeersBirdFragment(st, tenantID, locals)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
main, err := birdfmt.RenderMainBirdConf(birdfmt.MainBirdConfOptions{
|
||
RouterID: locals.routerID,
|
||
Includes: birdfmt.StandardIncludeFragments(),
|
||
Preamble: fmt.Sprintf("EvoBGP tenant aggregate config (trigger module %s) revision %s", moduleID, revisionID),
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
p4 := birdfmt.FragmentIncludePath(birdfmt.FragmentFiltersV4)
|
||
p6 := birdfmt.FragmentIncludePath(birdfmt.FragmentFiltersV6)
|
||
pTpl := birdfmt.FragmentIncludePath(birdfmt.FragmentBGPTemplate)
|
||
px4 := birdfmt.FragmentIncludePath(birdfmt.FragmentPrefixesV4)
|
||
px6 := birdfmt.FragmentIncludePath(birdfmt.FragmentPrefixesV6)
|
||
pPeers := birdfmt.FragmentIncludePath(birdfmt.FragmentPeers)
|
||
|
||
out := map[string]string{
|
||
"bird.conf": main,
|
||
p4: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), f4),
|
||
p6: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), f6),
|
||
pTpl: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), tplBody),
|
||
px4: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), staticV4),
|
||
px6: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), staticV6),
|
||
pPeers: peersBody,
|
||
}
|
||
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
|
||
for _, grp := range groups {
|
||
nameSuffix := communityProtocolSuffix(grp.CommunityID)
|
||
if len(grp.RoutesV4) > 0 {
|
||
b4.WriteString(birdfmt.RenderStaticIPv4Routes("evobgp_prefixes_v4_"+nameSuffix, grp.RoutesV4))
|
||
}
|
||
if len(grp.RoutesV6) > 0 {
|
||
b6.WriteString(birdfmt.RenderStaticIPv6Routes("evobgp_prefixes_v6_"+nameSuffix, grp.RoutesV6))
|
||
}
|
||
}
|
||
return b4.String(), b6.String()
|
||
}
|
||
|
||
func communityProtocolSuffix(communityID string) string {
|
||
raw := strings.TrimSpace(communityID)
|
||
if raw == "" {
|
||
return "default"
|
||
}
|
||
var b strings.Builder
|
||
b.Grow(len(raw))
|
||
for _, r := range raw {
|
||
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
|
||
b.WriteRune(r)
|
||
continue
|
||
}
|
||
b.WriteByte('_')
|
||
}
|
||
out := strings.Trim(b.String(), "_")
|
||
if out == "" {
|
||
return "default"
|
||
}
|
||
return "c_" + out
|
||
}
|
||
|
||
type birdLocals struct {
|
||
routerID string
|
||
localV4 string
|
||
localV6 string
|
||
localASN uint32
|
||
}
|
||
|
||
func birdLocalsFromStore(st store.Backend, tenantID string) birdLocals {
|
||
def := birdLocals{
|
||
routerID: "192.0.2.1",
|
||
localV4: "192.0.2.1",
|
||
localV6: "2001:db8::1",
|
||
localASN: 65001,
|
||
}
|
||
settings, err := st.ListGlobalSettings(tenantID)
|
||
if err != nil {
|
||
return def
|
||
}
|
||
loc := def
|
||
if s := stringFromSettingsMap(settings, "bird_router_id"); s != "" {
|
||
loc.routerID = s
|
||
}
|
||
if s := stringFromSettingsMap(settings, "bird_local_ipv4"); s != "" {
|
||
loc.localV4 = s
|
||
}
|
||
if s := stringFromSettingsMap(settings, "bird_local_ipv6"); s != "" {
|
||
loc.localV6 = s
|
||
}
|
||
if n := uint32FromSettingsMap(settings, "bird_local_asn"); n != 0 {
|
||
loc.localASN = n
|
||
}
|
||
if s := stringFromSettingsMap(settings, "bird_bgp_source_ipv4"); s != "" {
|
||
// BIRD router id must be an IPv4 address; historically aligned with optional BGP source setting.
|
||
loc.routerID = strings.TrimSpace(s)
|
||
}
|
||
return loc
|
||
}
|
||
|
||
func stringFromSettingsMap(m map[string]any, key string) string {
|
||
v, ok := m[key]
|
||
if !ok || v == nil {
|
||
return ""
|
||
}
|
||
s, ok := v.(string)
|
||
if !ok {
|
||
return ""
|
||
}
|
||
return strings.TrimSpace(s)
|
||
}
|
||
|
||
func uint32FromSettingsMap(m map[string]any, key string) uint32 {
|
||
v, ok := m[key]
|
||
if !ok || v == nil {
|
||
return 0
|
||
}
|
||
switch x := v.(type) {
|
||
case float64:
|
||
if x >= 1 && x <= 4294967295 {
|
||
return uint32(x)
|
||
}
|
||
case int:
|
||
if x >= 1 && x <= 4294967295 {
|
||
return uint32(x)
|
||
}
|
||
case int64:
|
||
if x >= 1 && x <= 4294967295 {
|
||
return uint32(x)
|
||
}
|
||
case string:
|
||
if n, err := strconv.ParseUint(strings.TrimSpace(x), 10, 32); err == nil && n >= 1 {
|
||
return uint32(n)
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func boolFromSettingsMap(m map[string]any, key string, defaultVal bool) bool {
|
||
v, ok := m[key]
|
||
if !ok || v == nil {
|
||
return defaultVal
|
||
}
|
||
switch x := v.(type) {
|
||
case bool:
|
||
return x
|
||
case float64:
|
||
return x != 0
|
||
case int:
|
||
return x != 0
|
||
case string:
|
||
s := strings.ToLower(strings.TrimSpace(x))
|
||
if s == "true" || s == "1" || s == "yes" {
|
||
return true
|
||
}
|
||
if s == "false" || s == "0" || s == "no" {
|
||
return false
|
||
}
|
||
}
|
||
return defaultVal
|
||
}
|
||
|
||
func renderPeerDiscoveryBirdFragment(st store.Backend, tenantID string) (string, error) {
|
||
settings, err := st.ListGlobalSettings(tenantID)
|
||
if err != nil || settings == nil {
|
||
return "", nil
|
||
}
|
||
if !boolFromSettingsMap(settings, "peer_discovery_enabled", false) {
|
||
return "", nil
|
||
}
|
||
rangesV4 := birdfmt.ParseDiscoveryRanges(stringFromSettingsMap(settings, "peer_discovery_ranges_v4"))
|
||
rangesV6 := birdfmt.ParseDiscoveryRanges(stringFromSettingsMap(settings, "peer_discovery_ranges_v6"))
|
||
if len(rangesV4) == 0 && len(rangesV6) == 0 {
|
||
return "", nil
|
||
}
|
||
return birdfmt.RenderDynamicBGPDiscovery(birdfmt.DynamicBGPDiscoveryOptions{
|
||
RangesV4: rangesV4,
|
||
RangesV6: rangesV6,
|
||
RequireExternal: boolFromSettingsMap(settings, "peer_discovery_require_external", true),
|
||
})
|
||
}
|
||
|
||
func intFromSettingsMap(m map[string]any, key string) int {
|
||
v, ok := m[key]
|
||
if !ok || v == nil {
|
||
return 0
|
||
}
|
||
switch x := v.(type) {
|
||
case float64:
|
||
return int(x)
|
||
case int:
|
||
return x
|
||
case int64:
|
||
return int(x)
|
||
case string:
|
||
n, err := strconv.Atoi(strings.TrimSpace(x))
|
||
if err == nil {
|
||
return n
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func applyRevisionRetention(st store.Backend, tenantID string) {
|
||
settings, err := st.ListGlobalSettings(tenantID)
|
||
if err != nil {
|
||
return
|
||
}
|
||
cutoff := RevisionRetentionCutoff(settings)
|
||
_, _ = st.PruneRevisionsBefore(tenantID, cutoff)
|
||
}
|
||
|
||
type peerPolicyJSON struct {
|
||
LocalIPv4 string `json:"local_ipv4"`
|
||
LocalIPv6 string `json:"local_ipv6"`
|
||
LocalASN float64 `json:"local_asn"`
|
||
}
|
||
|
||
func effectivePeerLocals(loc birdLocals, pol peerPolicyJSON) (v4, v6 string, asn uint32) {
|
||
v4 = strings.TrimSpace(loc.localV4)
|
||
v6 = strings.TrimSpace(loc.localV6)
|
||
if s := strings.TrimSpace(pol.LocalIPv4); s != "" {
|
||
v4 = s
|
||
}
|
||
if s := strings.TrimSpace(pol.LocalIPv6); s != "" {
|
||
v6 = s
|
||
}
|
||
asn = loc.localASN
|
||
if pol.LocalASN >= 1 && pol.LocalASN <= 4294967295 {
|
||
asn = uint32(pol.LocalASN)
|
||
}
|
||
return v4, v6, asn
|
||
}
|
||
|
||
// peerNeedsLocalOverride is true when the peer's effective local IP or ASN should override template "local as …" (add explicit "local <addr> as …" on the peer).
|
||
func peerNeedsLocalOverride(loc birdLocals, effLocal string, effASN uint32, ipv4 bool) bool {
|
||
if ipv4 {
|
||
return strings.TrimSpace(effLocal) != strings.TrimSpace(loc.localV4) || effASN != loc.localASN
|
||
}
|
||
return strings.TrimSpace(effLocal) != strings.TrimSpace(loc.localV6) || effASN != loc.localASN
|
||
}
|
||
|
||
func renderPeersBirdFragment(st store.Backend, tenantID string, loc birdLocals) (string, error) {
|
||
peers := st.ListPeers(tenantID)
|
||
var parts []string
|
||
parts = append(parts, birdfmt.ManagedBanner("peers"))
|
||
if disc, err := renderPeerDiscoveryBirdFragment(st, tenantID); err != nil {
|
||
return "", err
|
||
} else if disc != "" {
|
||
parts = append(parts, disc)
|
||
}
|
||
for _, p := range peers {
|
||
if p == nil || !p.Enabled {
|
||
continue
|
||
}
|
||
addr, ok := store.ParsePeerNeighbor(p.Neighbor)
|
||
if !ok {
|
||
continue
|
||
}
|
||
if !store.ValidASN(p.RemoteASN) {
|
||
continue
|
||
}
|
||
pol := parsePeerPolicies(p.PoliciesJSON)
|
||
lv4, lv6, asn := effectivePeerLocals(loc, pol)
|
||
proto := birdfmt.PeerProtocolName(p.ID)
|
||
ra := uint32(p.RemoteASN)
|
||
if addr.Is4() {
|
||
opts := birdfmt.BGPPeerFromTemplateOptions{
|
||
ProtocolName: proto,
|
||
TemplateName: birdfmt.BGPTemplateNameV4,
|
||
NeighborIP: addr.String(),
|
||
NeighborASN: ra,
|
||
}
|
||
if peerNeedsLocalOverride(loc, lv4, asn, true) {
|
||
opts.OverrideLocalIP = lv4
|
||
opts.OverrideLocalASN = asn
|
||
}
|
||
s, err := birdfmt.RenderProtocolBGPFromTemplate(opts)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
parts = append(parts, s)
|
||
continue
|
||
}
|
||
if addr.Is6() {
|
||
opts := birdfmt.BGPPeerFromTemplateOptions{
|
||
ProtocolName: proto,
|
||
TemplateName: birdfmt.BGPTemplateNameV6,
|
||
NeighborIP: addr.String(),
|
||
NeighborASN: ra,
|
||
}
|
||
if peerNeedsLocalOverride(loc, lv6, asn, false) {
|
||
opts.OverrideLocalIP = lv6
|
||
opts.OverrideLocalASN = asn
|
||
}
|
||
s, err := birdfmt.RenderProtocolBGPFromTemplate(opts)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
parts = append(parts, s)
|
||
}
|
||
}
|
||
if len(parts) == 1 {
|
||
parts = append(parts, "# (no enabled BGP peers with valid neighbor addresses)\n")
|
||
}
|
||
return birdfmt.JoinFragments(parts...), nil
|
||
}
|
||
|
||
func parsePeerPolicies(raw string) peerPolicyJSON {
|
||
raw = strings.TrimSpace(raw)
|
||
if raw == "" || raw == "{}" {
|
||
return peerPolicyJSON{}
|
||
}
|
||
var pol peerPolicyJSON
|
||
_ = json.Unmarshal([]byte(raw), &pol)
|
||
return pol
|
||
}
|
||
|
||
// buildExpandedBirdText concatenates bird.conf and the contents of each standard include (for UI / preview).
|
||
func buildExpandedBirdText(main string, frags map[string]string) string {
|
||
var b strings.Builder
|
||
b.WriteString(strings.TrimSpace(main))
|
||
b.WriteString("\n")
|
||
for _, inc := range birdfmt.StandardIncludeFragments() {
|
||
b.WriteString("\n# ---------- include \"")
|
||
b.WriteString(inc)
|
||
b.WriteString("\" ----------\n")
|
||
body := strings.TrimSpace(frags[inc])
|
||
if body == "" {
|
||
b.WriteString("# (empty)\n")
|
||
continue
|
||
}
|
||
b.WriteString(body)
|
||
if !strings.HasSuffix(body, "\n") {
|
||
b.WriteByte('\n')
|
||
}
|
||
}
|
||
return b.String()
|
||
}
|