Очередь job получила lease и fencing, телеметрия и прогресс deploy сохраняются, ревизия пишет manifest входов. Обновление префиксов не переписывает неизменённый снимок, большие наборы идут через COPY, а устаревшая ревизия не попадает в deploy. Бандл ноды кэшируется, prune и постановка job больше не держат горячий путь. Co-authored-by: Cursor <[email protected]>
364 lines
11 KiB
Go
364 lines
11 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
)
|
|
|
|
func (m *Memory) GetRenderGeneration(ctx context.Context, tenantID string) (int64, error) {
|
|
if err := ctx.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
if m.tenants[tenantID] == nil {
|
|
return 0, ErrNotFound
|
|
}
|
|
gen := m.renderGeneration[tenantID]
|
|
if gen == 0 {
|
|
gen = 1
|
|
}
|
|
return gen, nil
|
|
}
|
|
|
|
func (m *Memory) ListModuleSnapshotHeaders(ctx context.Context, tenantID string) ([]ModuleSnapshotUse, error) {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
return append([]ModuleSnapshotUse(nil), m.snapshotHeadersLocked(tenantID)...), nil
|
|
}
|
|
|
|
func (m *Memory) snapshotHeadersLocked(tenantID string) []ModuleSnapshotUse {
|
|
prefix := moduleSnapshotKey(tenantID, "")
|
|
var out []ModuleSnapshotUse
|
|
for key, snap := range m.moduleSnapshots {
|
|
if snap == nil || !strings.HasPrefix(key, prefix) {
|
|
continue
|
|
}
|
|
out = append(out, ModuleSnapshotUse{
|
|
ModuleID: strings.TrimPrefix(key, prefix),
|
|
InputHash: snap.InputHash,
|
|
SnapshotVersion: snap.Version,
|
|
})
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].ModuleID < out[j].ModuleID })
|
|
return out
|
|
}
|
|
|
|
func (m *Memory) LatestRenderManifest(ctx context.Context, tenantID string) (*RenderManifestRecord, bool, error) {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, false, err
|
|
}
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
var latest *Revision
|
|
for _, rev := range m.revisions {
|
|
if rev == nil || rev.TenantID != tenantID {
|
|
continue
|
|
}
|
|
if latest == nil || rev.CreatedAt.After(latest.CreatedAt) || (rev.CreatedAt.Equal(latest.CreatedAt) && rev.ID > latest.ID) {
|
|
latest = rev
|
|
}
|
|
}
|
|
if latest == nil {
|
|
return nil, false, nil
|
|
}
|
|
rec := m.renderManifests[latest.ID]
|
|
if rec == nil {
|
|
return nil, false, nil
|
|
}
|
|
return cloneRenderManifest(rec), true, nil
|
|
}
|
|
|
|
func (m *Memory) recordRenderManifestLocked(revisionID, tenantID string) {
|
|
if m.renderManifests == nil {
|
|
m.renderManifests = make(map[string]*RenderManifestRecord)
|
|
}
|
|
gen := m.renderGeneration[tenantID]
|
|
if gen == 0 {
|
|
gen = 1
|
|
}
|
|
m.renderManifests[revisionID] = &RenderManifestRecord{
|
|
RevisionID: revisionID,
|
|
Generation: gen,
|
|
Snapshots: m.snapshotHeadersLocked(tenantID),
|
|
}
|
|
}
|
|
|
|
func cloneRenderManifest(rec *RenderManifestRecord) *RenderManifestRecord {
|
|
if rec == nil {
|
|
return nil
|
|
}
|
|
out := *rec
|
|
out.Snapshots = append([]ModuleSnapshotUse(nil), rec.Snapshots...)
|
|
return &out
|
|
}
|
|
|
|
// bumpRenderGenerationLocked is called only while holding m.mu for a semantic write.
|
|
func (m *Memory) bumpRenderGenerationLocked(tenantID string) {
|
|
if m.renderGeneration == nil {
|
|
m.renderGeneration = make(map[string]int64)
|
|
}
|
|
if m.renderGeneration[tenantID] == 0 {
|
|
m.renderGeneration[tenantID] = 1
|
|
}
|
|
m.renderGeneration[tenantID]++
|
|
}
|
|
|
|
// RenderInputs is the complete, immutable configuration read by tenant rendering.
|
|
// Fetch observations such as CDN ETags, resolve metadata and dispatch status are
|
|
// retained in the captured rows where the ingest path needs them, but excluded
|
|
// from SemanticHash. Callers must not mutate a captured value.
|
|
type RenderInputs struct {
|
|
TenantID string `json:"tenant_id"`
|
|
Generation int64 `json:"generation"`
|
|
Modules []RenderModule `json:"modules"`
|
|
Profiles []*DohProfile `json:"profiles"`
|
|
Peers []*BGPPeer `json:"peers"`
|
|
Communities []*Community `json:"communities"`
|
|
Settings map[string]any `json:"settings"`
|
|
Speakers []RenderSpeaker `json:"speakers"`
|
|
}
|
|
|
|
// RenderModule contains all ingestion inputs for one module in one capture.
|
|
type RenderModule struct {
|
|
Module *Module `json:"module"`
|
|
CDNSources []*CDNSource `json:"cdn_sources"`
|
|
ASEntries []*ASEntry `json:"as_entries"`
|
|
DomainEntries []*DomainEntry `json:"domain_entries"`
|
|
IPRanges []*IPRangeEntry `json:"ip_ranges"`
|
|
}
|
|
|
|
// RenderSpeaker holds only values that affect speaker-specific BIRD fragments.
|
|
type RenderSpeaker struct {
|
|
ID string `json:"id"`
|
|
NodeIPv4 string `json:"node_ipv4"`
|
|
BirdBgpSourceIPv4 string `json:"bird_bgp_source_ipv4"`
|
|
BirdBgpSourceIPv6 string `json:"bird_bgp_source_ipv6"`
|
|
}
|
|
|
|
// ModuleSnapshotUse identifies the exact module materialization used by a render.
|
|
type ModuleSnapshotUse struct {
|
|
ModuleID string `json:"module_id"`
|
|
InputHash string `json:"input_hash"`
|
|
SnapshotVersion int64 `json:"snapshot_version"`
|
|
}
|
|
|
|
// RenderManifestRecord is the admission barrier stored with a config revision.
|
|
// Generation and snapshot versions let a later render return the revision
|
|
// without loading prefix rows when nothing material changed.
|
|
type RenderManifestRecord struct {
|
|
RevisionID string
|
|
Generation int64
|
|
Snapshots []ModuleSnapshotUse
|
|
}
|
|
|
|
// RenderManifest is persisted with a revision so new publish admission can
|
|
// distinguish a current render from a historical rollback or durable replay.
|
|
type RenderManifest struct {
|
|
Inputs RenderInputs `json:"inputs"`
|
|
ConfigHash string `json:"config_hash"`
|
|
Snapshots []ModuleSnapshotUse `json:"snapshots"`
|
|
OverlayBySpeaker map[string]map[string]string `json:"overlay_by_speaker"`
|
|
}
|
|
|
|
// SemanticHash fingerprints captured render settings and ingest configuration.
|
|
// It deliberately excludes fetch timestamps, ETags, health and dispatch state.
|
|
func (in *RenderInputs) SemanticHash() (string, error) {
|
|
if in == nil || strings.TrimSpace(in.TenantID) == "" {
|
|
return "", ErrInvalidInput
|
|
}
|
|
semantic := struct {
|
|
TenantID string `json:"tenant_id"`
|
|
Modules []any `json:"modules"`
|
|
Profiles []any `json:"profiles"`
|
|
Peers []any `json:"peers"`
|
|
Communities []any `json:"communities"`
|
|
Settings map[string]any `json:"settings"`
|
|
Speakers []RenderSpeaker `json:"speakers"`
|
|
}{TenantID: in.TenantID, Settings: in.Settings, Speakers: append([]RenderSpeaker(nil), in.Speakers...)}
|
|
modules := append([]RenderModule(nil), in.Modules...)
|
|
for _, item := range modules {
|
|
if item.Module == nil {
|
|
return "", fmt.Errorf("store: render inputs: nil module")
|
|
}
|
|
}
|
|
sort.Slice(modules, func(i, j int) bool { return modules[i].Module.ID < modules[j].Module.ID })
|
|
for _, item := range modules {
|
|
m := item.Module
|
|
semantic.Modules = append(semantic.Modules, struct {
|
|
ID string
|
|
Type string
|
|
Enabled bool
|
|
DefaultCommunityID *string
|
|
DohResolverPolicy string
|
|
DohProfileIDs []string
|
|
CDNSources []any
|
|
ASEntries []any
|
|
DomainEntries []any
|
|
IPRanges []any
|
|
}{ID: m.ID, Type: m.Type, Enabled: m.Enabled, DefaultCommunityID: m.DefaultCommunityID, DohResolverPolicy: m.DohResolverPolicy,
|
|
DohProfileIDs: append([]string(nil), m.EffectiveDohProfileIDs()...),
|
|
CDNSources: semanticCDNSources(item.CDNSources), ASEntries: semanticASEntries(item.ASEntries), DomainEntries: semanticDomainEntries(item.DomainEntries), IPRanges: semanticIPRanges(item.IPRanges)})
|
|
}
|
|
profiles := append([]*DohProfile(nil), in.Profiles...)
|
|
sort.Slice(profiles, func(i, j int) bool {
|
|
if profiles[i] == nil {
|
|
return false
|
|
}
|
|
if profiles[j] == nil {
|
|
return true
|
|
}
|
|
return profiles[i].ID < profiles[j].ID
|
|
})
|
|
for _, p := range profiles {
|
|
if p != nil {
|
|
semantic.Profiles = append(semantic.Profiles, struct {
|
|
ID, URL string
|
|
TimeoutMs *int
|
|
SecretRef *string
|
|
}{p.ID, p.URL, p.TimeoutMs, p.SecretRef})
|
|
}
|
|
}
|
|
peers := append([]*BGPPeer(nil), in.Peers...)
|
|
sort.Slice(peers, func(i, j int) bool {
|
|
if peers[i] == nil {
|
|
return false
|
|
}
|
|
if peers[j] == nil {
|
|
return true
|
|
}
|
|
return peers[i].ID < peers[j].ID
|
|
})
|
|
for _, p := range peers {
|
|
if p != nil {
|
|
semantic.Peers = append(semantic.Peers, struct {
|
|
ID, Neighbor, PoliciesJSON string
|
|
RemoteASN int64
|
|
Enabled bool
|
|
SpeakerID *string
|
|
}{p.ID, p.Neighbor, p.PoliciesJSON, p.RemoteASN, p.Enabled, p.SpeakerID})
|
|
}
|
|
}
|
|
communities := append([]*Community(nil), in.Communities...)
|
|
sort.Slice(communities, func(i, j int) bool {
|
|
if communities[i] == nil {
|
|
return false
|
|
}
|
|
if communities[j] == nil {
|
|
return true
|
|
}
|
|
return communities[i].ID < communities[j].ID
|
|
})
|
|
for _, c := range communities {
|
|
if c != nil {
|
|
semantic.Communities = append(semantic.Communities, struct{ ID, Community, ValueJSON string }{c.ID, c.Community, c.ValueJSON})
|
|
}
|
|
}
|
|
sort.Slice(semantic.Speakers, func(i, j int) bool { return semantic.Speakers[i].ID < semantic.Speakers[j].ID })
|
|
data, err := json.Marshal(semantic)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
sum := sha256.Sum256(data)
|
|
return "sha256:" + hex.EncodeToString(sum[:]), nil
|
|
}
|
|
|
|
func semanticCDNSources(items []*CDNSource) []any {
|
|
items = append([]*CDNSource(nil), items...)
|
|
sort.Slice(items, func(i, j int) bool {
|
|
if items[i] == nil {
|
|
return false
|
|
}
|
|
if items[j] == nil {
|
|
return true
|
|
}
|
|
return items[i].ID < items[j].ID
|
|
})
|
|
out := make([]any, 0, len(items))
|
|
for _, s := range items {
|
|
if s != nil {
|
|
out = append(out, struct {
|
|
ID, SourceKind, URL, PrefixPath string
|
|
RefreshIntervalSec *int
|
|
CommunityID *string
|
|
}{s.ID, s.SourceKind, s.URL, s.PrefixPath, s.RefreshIntervalSec, s.CommunityID})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
func semanticASEntries(items []*ASEntry) []any {
|
|
items = append([]*ASEntry(nil), items...)
|
|
sort.Slice(items, func(i, j int) bool {
|
|
if items[i] == nil {
|
|
return false
|
|
}
|
|
if items[j] == nil {
|
|
return true
|
|
}
|
|
return items[i].ID < items[j].ID
|
|
})
|
|
out := make([]any, 0, len(items))
|
|
for _, s := range items {
|
|
if s != nil {
|
|
out = append(out, struct {
|
|
ID string
|
|
ASN int64
|
|
CommunityID *string
|
|
}{s.ID, s.ASN, s.CommunityID})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
func semanticDomainEntries(items []*DomainEntry) []any {
|
|
items = append([]*DomainEntry(nil), items...)
|
|
sort.Slice(items, func(i, j int) bool {
|
|
if items[i] == nil {
|
|
return false
|
|
}
|
|
if items[j] == nil {
|
|
return true
|
|
}
|
|
return items[i].ID < items[j].ID
|
|
})
|
|
out := make([]any, 0, len(items))
|
|
for _, s := range items {
|
|
if s != nil {
|
|
out = append(out, struct {
|
|
ID, FQDN string
|
|
CommunityID *string
|
|
}{s.ID, s.FQDN, s.CommunityID})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
func semanticIPRanges(items []*IPRangeEntry) []any {
|
|
items = append([]*IPRangeEntry(nil), items...)
|
|
sort.Slice(items, func(i, j int) bool {
|
|
if items[i] == nil {
|
|
return false
|
|
}
|
|
if items[j] == nil {
|
|
return true
|
|
}
|
|
return items[i].ID < items[j].ID
|
|
})
|
|
out := make([]any, 0, len(items))
|
|
for _, s := range items {
|
|
if s != nil {
|
|
out = append(out, struct {
|
|
ID, Prefix string
|
|
CommunityID *string
|
|
}{s.ID, s.Prefix, s.CommunityID})
|
|
}
|
|
}
|
|
return out
|
|
}
|