Files
EvoBGP/internal/store/memory.go
T
Denozordec 4f50350990
CI / changes (push) Successful in 6s
CI / openapi (push) Has been skipped
CI / go (push) Successful in 44s
CI / docker-web (deploy/docker/evobgp-web/Dockerfile, , evobgp-web) (push) Successful in 1m5s
CI / docker-web (deploy/docker/evobgp-web/Dockerfile, evobgp-all, evobgp-web-all) (push) Successful in 1m1s
CI / docker-bird (push) Has been skipped
CI / bird2 (push) Successful in 15s
CI / docker-go-prime (push) Successful in 24s
CI / docker-go (deploy/docker/evobgp-agent/Dockerfile, , evobgp-agent) (push) Successful in 1m1s
CI / docker-go (evobgp-all, 1, deploy/docker/gobinary/Dockerfile, , evobgp-all) (push) Successful in 2m12s
CI / docker-go (evobgp-api, 1, deploy/docker/gobinary/Dockerfile, , evobgp-api) (push) Successful in 1m24s
CI / docker-go (evobgp-deploy, 0, deploy/docker/gobinary/Dockerfile, , evobgp-deploy) (push) Successful in 1m19s
CI / docker-go (evobgp-ingest, 0, deploy/docker/gobinary/Dockerfile, , evobgp-ingest) (push) Successful in 1m21s
CI / docker-go (evobgp-node, 0, deploy/docker/gobinary/Dockerfile, , evobgp-node) (push) Successful in 1m6s
CI / docker-go (evobgp-render, 0, deploy/docker/gobinary/Dockerfile, , evobgp-render) (push) Successful in 1m21s
CI / docker-go (evobgp-scheduler, 0, deploy/docker/gobinary/Dockerfile, , evobgp-scheduler) (push) Successful in 1m21s
feat: implement revision retention settings and pruning logic
Added functionality to handle revision retention minutes in global settings, including validation for input values. Introduced a new method to prune old revisions based on the specified retention period, ensuring that only the most recent revisions are kept. Updated the frontend to allow users to configure the revision retention setting, enhancing the management of revision lifecycles.
2026-04-08 14:08:23 +07:00

672 lines
18 KiB
Go

package store
import (
"errors"
"fmt"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/google/uuid"
)
var (
ErrNotFound = errors.New("store: not found")
ErrTenantScope = errors.New("store: tenant mismatch")
ErrInvalidInput = errors.New("store: invalid input")
)
// Memory is a development-oriented in-memory backend for control-plane API handlers.
type Memory struct {
mu sync.RWMutex
tenants map[string]*Tenant
modules map[string]*Module
revisions map[string]*Revision
speakers map[string]*Speaker
// speakerID -> latest published revision for evobgp-node pulls
publishedRevision map[string]publishedInfo
peers map[string]*BGPPeer
dohProfiles map[string]*DohProfile
communities map[string]*Community
cdnSources map[string]*CDNSource
asEntries map[string]*ASEntry
domainEnt map[string]*DomainEntry
ipRanges map[string]*IPRangeEntry
settings map[string]map[string]any // tenantID -> key -> JSON-compatible value
revPrefixes map[string][]PrefixRow
// DemoIDs valid after SeedDemo()
demoTenantID string
demoModuleCDN string
demoModuleIP string
demoRevisionID string
demoSpeakerID string
}
type publishedInfo struct {
RevisionID string
PublishedAt time.Time
}
type Tenant struct {
ID string
Name string
Slug string
}
type Module struct {
ID string
TenantID string
Type string // AS_PREFIXES, CDN_CIDRS, DOMAINS, IP_RANGES
Name string
Enabled bool
RefreshIntervalSec int // 0 = unset
CronExpr string // optional cron for scheduler (display / future use)
Priority int
DefaultCommunityID *string
DohProfileID *string
DeletedAt *time.Time
}
type Revision struct {
ID string
TenantID string
ModuleID string
ContentHash string
ParentRevisionID *string
CreatedAt time.Time
// MaterializedPrefixCount is the size of the prefix set for this revision (control-plane / render output).
MaterializedPrefixCount int
// PreviewFragments maps logical paths (e.g. bird.conf) to generated text.
PreviewFragments map[string]string
}
// BGPPeer maps to bgp_peer (+ display fields in meta).
type BGPPeer struct {
ID string `json:"id,omitempty"`
TenantID string `json:"tenant_id,omitempty"`
SpeakerID *string `json:"bgp_speaker_id"`
Name string `json:"name"`
Neighbor string `json:"neighbor"`
RemoteASN int64 `json:"remote_asn"`
Enabled bool `json:"enabled"`
SessionState string `json:"session_state"`
PoliciesJSON string `json:"policies_json"`
}
type Speaker struct {
ID string `json:"id,omitempty"`
TenantID string `json:"tenant_id,omitempty"`
Role string `json:"role"`
Endpoint string `json:"endpoint"`
LastAppliedRevisionID *string `json:"last_applied_revision_id"`
MetaJSON string `json:"meta_json"`
}
func NewMemory() *Memory {
return &Memory{
tenants: make(map[string]*Tenant),
modules: make(map[string]*Module),
revisions: make(map[string]*Revision),
speakers: make(map[string]*Speaker),
publishedRevision: make(map[string]publishedInfo),
peers: make(map[string]*BGPPeer),
dohProfiles: make(map[string]*DohProfile),
communities: make(map[string]*Community),
cdnSources: make(map[string]*CDNSource),
asEntries: make(map[string]*ASEntry),
domainEnt: make(map[string]*DomainEntry),
ipRanges: make(map[string]*IPRangeEntry),
settings: make(map[string]map[string]any),
revPrefixes: make(map[string][]PrefixRow),
}
}
// SeedDemo installs a minimal tenant, modules, revision, and speaker for local testing.
func (m *Memory) SeedDemo() {
m.mu.Lock()
defer m.mu.Unlock()
tid := uuid.NewString()
m.tenants[tid] = &Tenant{ID: tid, Name: "Demo", Slug: "demo"}
mCDN := uuid.NewString()
m.modules[mCDN] = &Module{
ID: mCDN,
TenantID: tid,
Type: "CDN_CIDRS",
Name: "demo-cdn",
Enabled: true,
RefreshIntervalSec: 3600,
CronExpr: "",
Priority: 10,
}
mIP := uuid.NewString()
m.modules[mIP] = &Module{
ID: mIP,
TenantID: tid,
Type: "IP_RANGES",
Name: "demo-static",
Enabled: true,
RefreshIntervalSec: 0,
CronExpr: "",
Priority: 20,
}
parent := uuid.NewString()
m.revisions[parent] = &Revision{
ID: parent,
TenantID: tid,
ModuleID: "",
ContentHash: "sha256:parent",
ParentRevisionID: nil,
CreatedAt: time.Now().UTC().Add(-time.Hour),
MaterializedPrefixCount: 0,
PreviewFragments: map[string]string{"bird.conf": "# parent revision\n"},
}
rid := uuid.NewString()
m.revisions[rid] = &Revision{
ID: rid,
TenantID: tid,
ModuleID: "",
ContentHash: "sha256:demo-rev-1",
ParentRevisionID: &parent,
CreatedAt: time.Now().UTC(),
MaterializedPrefixCount: 128,
PreviewFragments: map[string]string{
// Valid minimal BIRD 2 skeleton (see internal/birdfmt/testdata/scenarios/minimal/bird.conf).
"bird.conf": `# EvoBGP demo bundle
router id 192.0.2.1;
protocol device {
}
protocol direct {
ipv4;
ipv6;
}
`,
"bird.d/evobgp_demo.conf": "# static demo fragment\n",
},
}
sid := uuid.NewString()
m.speakers[sid] = &Speaker{
ID: sid,
TenantID: tid,
Role: "replica",
Endpoint: "10.0.0.2:179",
LastAppliedRevisionID: nil,
}
m.publishedRevision[sid] = publishedInfo{RevisionID: rid, PublishedAt: time.Now().UTC()}
p1 := uuid.NewString()
m.peers[p1] = &BGPPeer{
ID: p1,
TenantID: tid,
SpeakerID: &sid,
Name: "demo-upstream-4",
Neighbor: "198.51.100.2",
RemoteASN: 65001,
Enabled: true,
SessionState: "Established",
}
p2 := uuid.NewString()
m.peers[p2] = &BGPPeer{
ID: p2,
TenantID: tid,
SpeakerID: &sid,
Name: "demo-upstream-6",
Neighbor: "2001:db8::2",
RemoteASN: 65002,
Enabled: true,
SessionState: "Idle",
}
m.demoTenantID, m.demoModuleCDN, m.demoModuleIP, m.demoRevisionID, m.demoSpeakerID = tid, mCDN, mIP, rid, sid
// Demo materialized prefixes for /revisions/{id}/prefixes
cid := uuid.NewString()
m.communities[cid] = &Community{ID: cid, TenantID: tid, Community: "demo-comm", Title: "Demo", ValueJSON: "{}"}
m.revPrefixes[rid] = []PrefixRow{
{Prefix: "203.0.113.0/24", CommunityID: &cid, Source: "demo"},
{Prefix: "2001:db8::/32", CommunityID: &cid, Source: "demo"},
}
}
// MaterializedPrefixStats returns max and sum of MaterializedPrefixCount across revisions.
func (m *Memory) MaterializedPrefixStats() (max int, sum int) {
m.mu.RLock()
defer m.mu.RUnlock()
for _, r := range m.revisions {
sum += r.MaterializedPrefixCount
if r.MaterializedPrefixCount > max {
max = r.MaterializedPrefixCount
}
}
return max, sum
}
// PeerCount returns the number of configured BGP peers.
func (m *Memory) PeerCount() int {
m.mu.RLock()
defer m.mu.RUnlock()
return len(m.peers)
}
// PeerSessionCountsByState returns peer counts grouped by SessionState.
func (m *Memory) PeerSessionCountsByState() map[string]int {
m.mu.RLock()
defer m.mu.RUnlock()
out := make(map[string]int)
for _, p := range m.peers {
st := strings.TrimSpace(p.SessionState)
if st == "" {
st = "unknown"
}
out[st]++
}
return out
}
// DemoIDs returns IDs from SeedDemo; empty strings if SeedDemo was not called.
func (m *Memory) DemoIDs() (tenant, moduleCDN, moduleIP, revision, speaker string) {
m.mu.RLock()
defer m.mu.RUnlock()
return m.demoTenantID, m.demoModuleCDN, m.demoModuleIP, m.demoRevisionID, m.demoSpeakerID
}
// ListTenantIDs returns tenant ids sorted lexicographically.
func (m *Memory) ListTenantIDs() ([]string, error) {
m.mu.RLock()
defer m.mu.RUnlock()
out := make([]string, 0, len(m.tenants))
for id := range m.tenants {
out = append(out, id)
}
sort.Strings(out)
return out, nil
}
// CreateRenderRevision stores a new revision and its materialized prefixes.
func (m *Memory) CreateRenderRevision(revisionID, tenantID, moduleID string, parentRevisionID *string, contentHash string, previewFragments map[string]string, prefixes []PrefixRow) error {
if strings.TrimSpace(revisionID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(contentHash) == "" {
return ErrInvalidInput
}
m.mu.Lock()
defer m.mu.Unlock()
if _, exists := m.revisions[revisionID]; exists {
return ErrInvalidInput
}
mod, ok := m.modules[moduleID]
if !ok || mod.DeletedAt != nil || mod.TenantID != tenantID {
return ErrNotFound
}
if parentRevisionID != nil && *parentRevisionID != "" {
if pr, ok := m.revisions[*parentRevisionID]; !ok || pr.TenantID != tenantID {
return ErrNotFound
}
}
frag := make(map[string]string, len(previewFragments))
for k, v := range previewFragments {
frag[k] = v
}
parent := parentRevisionID
m.revisions[revisionID] = &Revision{
ID: revisionID,
TenantID: tenantID,
ModuleID: moduleID,
ContentHash: contentHash,
ParentRevisionID: parent,
CreatedAt: time.Now().UTC(),
MaterializedPrefixCount: len(prefixes),
PreviewFragments: frag,
}
if len(prefixes) > 0 {
cp := make([]PrefixRow, len(prefixes))
copy(cp, prefixes)
m.revPrefixes[revisionID] = cp
} else {
m.revPrefixes[revisionID] = nil
}
return nil
}
// ListModules returns modules for a tenant (sorted by priority, then name).
func (m *Memory) ListModules(tenantID string) []*Module {
m.mu.RLock()
defer m.mu.RUnlock()
var out []*Module
for _, mod := range m.modules {
if mod.TenantID == tenantID && mod.DeletedAt == nil {
out = append(out, mod)
}
}
sort.Slice(out, func(i, j int) bool {
if out[i].Priority != out[j].Priority {
return out[i].Priority < out[j].Priority
}
return out[i].Name < out[j].Name
})
return out
}
// ListPeers returns BGP peers for a tenant (sorted by name).
func (m *Memory) ListPeers(tenantID string) []*BGPPeer {
m.mu.RLock()
defer m.mu.RUnlock()
var out []*BGPPeer
for _, p := range m.peers {
if p.TenantID == tenantID {
out = append(out, p)
}
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out
}
func (m *Memory) GetModule(tenantID, moduleID string) (*Module, error) {
m.mu.RLock()
defer m.mu.RUnlock()
mod, ok := m.modules[moduleID]
if !ok || mod.DeletedAt != nil {
return nil, ErrNotFound
}
if mod.TenantID != tenantID {
return nil, ErrTenantScope
}
return mod, nil
}
func (m *Memory) GetRevision(tenantID, revisionID string) (*Revision, error) {
m.mu.RLock()
defer m.mu.RUnlock()
rev, ok := m.revisions[revisionID]
if !ok {
return nil, ErrNotFound
}
if rev.TenantID != tenantID {
return nil, ErrTenantScope
}
return rev, nil
}
func (m *Memory) GetSpeaker(tenantID, speakerID string) (*Speaker, error) {
m.mu.RLock()
defer m.mu.RUnlock()
sp, ok := m.speakers[speakerID]
if !ok {
return nil, ErrNotFound
}
if sp.TenantID != tenantID {
return nil, ErrTenantScope
}
return sp, nil
}
// ListSpeakersForTenant returns speakers for a tenant (IDs sorted) for deploy-all and similar operations.
func (m *Memory) ListSpeakersForTenant(tenantID string) []*Speaker {
m.mu.RLock()
defer m.mu.RUnlock()
var ids []string
for id, sp := range m.speakers {
if sp.TenantID == tenantID {
ids = append(ids, id)
}
}
sort.Strings(ids)
out := make([]*Speaker, 0, len(ids))
for _, id := range ids {
out = append(out, m.speakers[id])
}
return out
}
// GetSpeakerAnyTenant resolves a speaker without tenant check (node API uses speaker id in path).
func (m *Memory) GetSpeakerAnyTenant(speakerID string) (*Speaker, error) {
m.mu.RLock()
defer m.mu.RUnlock()
sp, ok := m.speakers[speakerID]
if !ok {
return nil, ErrNotFound
}
return sp, nil
}
func (m *Memory) LatestPublishedRevision(speakerID string) (revisionID string, publishedAt time.Time, err error) {
m.mu.RLock()
defer m.mu.RUnlock()
info, ok := m.publishedRevision[speakerID]
if !ok {
return "", time.Time{}, ErrNotFound
}
return info.RevisionID, info.PublishedAt, nil
}
// CreateRollbackRevision adds a new revision that reuses content from sourceRevisionID (same hash & preview copy).
func (m *Memory) CreateRollbackRevision(tenantID, sourceRevisionID string) (newID string, err error) {
m.mu.Lock()
defer m.mu.Unlock()
src, ok := m.revisions[sourceRevisionID]
if !ok {
return "", ErrNotFound
}
if src.TenantID != tenantID {
return "", ErrTenantScope
}
newID = uuid.NewString()
frag := make(map[string]string, len(src.PreviewFragments))
for k, v := range src.PreviewFragments {
frag[k] = v
}
parent := sourceRevisionID
m.revisions[newID] = &Revision{
ID: newID,
TenantID: tenantID,
ModuleID: src.ModuleID,
ContentHash: src.ContentHash + ":rollback",
ParentRevisionID: &parent,
CreatedAt: time.Now().UTC(),
MaterializedPrefixCount: src.MaterializedPrefixCount,
PreviewFragments: frag,
}
if px, ok := m.revPrefixes[sourceRevisionID]; ok {
cp := make([]PrefixRow, len(px))
copy(cp, px)
m.revPrefixes[newID] = cp
}
return newID, nil
}
// SetLastAppliedRevision updates speaker state after deploy job (demo).
func (m *Memory) SetLastAppliedRevision(tenantID, speakerID, revisionID string) error {
m.mu.Lock()
defer m.mu.Unlock()
sp, ok := m.speakers[speakerID]
if !ok {
return ErrNotFound
}
if sp.TenantID != tenantID {
return ErrTenantScope
}
if _, ok := m.revisions[revisionID]; !ok {
return fmt.Errorf("%w: revision", ErrNotFound)
}
sp.LastAppliedRevisionID = &revisionID
return nil
}
// PublishRevisionForSpeaker marks latest bundle pointer for a replica speaker.
func (m *Memory) PublishRevisionForSpeaker(speakerID, revisionID string) error {
m.mu.Lock()
defer m.mu.Unlock()
if _, ok := m.speakers[speakerID]; !ok {
return ErrNotFound
}
if _, ok := m.revisions[revisionID]; !ok {
return fmt.Errorf("%w: revision", ErrNotFound)
}
m.publishedRevision[speakerID] = publishedInfo{RevisionID: revisionID, PublishedAt: time.Now().UTC()}
return nil
}
// RevisionDiff returns prefix set diff from materialized snapshots when present.
func (m *Memory) RevisionDiff(tenantID, aID, bID string) (map[string]any, error) {
m.mu.RLock()
defer m.mu.RUnlock()
_, err := m.getRevisionLocked(tenantID, aID)
if err != nil {
return nil, err
}
_, err = m.getRevisionLocked(tenantID, bID)
if err != nil {
return nil, err
}
setA := make(map[string]struct{})
setB := make(map[string]struct{})
for _, p := range m.revPrefixes[aID] {
setA[p.Prefix] = struct{}{}
}
for _, p := range m.revPrefixes[bID] {
setB[p.Prefix] = struct{}{}
}
var added, removed []string
unchanged := 0
for p := range setB {
if _, ok := setA[p]; !ok {
added = append(added, p)
} else {
unchanged++
}
}
for p := range setA {
if _, ok := setB[p]; !ok {
removed = append(removed, p)
}
}
sort.Strings(added)
sort.Strings(removed)
return map[string]any{
"revision_a": aID,
"revision_b": bID,
"prefixes": map[string]any{
"added": added,
"removed": removed,
"unchanged_count": unchanged,
},
}, nil
}
func (m *Memory) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
m.mu.Lock()
defer m.mu.Unlock()
var newest *Revision
for _, rev := range m.revisions {
if rev.TenantID != tenantID {
continue
}
if newest == nil || rev.CreatedAt.After(newest.CreatedAt) {
newest = rev
}
}
if newest == nil {
return 0, nil
}
protected := map[string]struct{}{
newest.ID: {},
}
for _, sp := range m.speakers {
if sp == nil || sp.TenantID != tenantID {
continue
}
if sp.LastAppliedRevisionID != nil && strings.TrimSpace(*sp.LastAppliedRevisionID) != "" {
protected[strings.TrimSpace(*sp.LastAppliedRevisionID)] = struct{}{}
}
}
for speakerID, info := range m.publishedRevision {
sp, ok := m.speakers[speakerID]
if !ok || sp == nil || sp.TenantID != tenantID {
continue
}
if strings.TrimSpace(info.RevisionID) != "" {
protected[strings.TrimSpace(info.RevisionID)] = struct{}{}
}
}
deleted := 0
for id, rev := range m.revisions {
if rev == nil || rev.TenantID != tenantID {
continue
}
if !rev.CreatedAt.Before(cutoff) {
continue
}
if _, keep := protected[id]; keep {
continue
}
delete(m.revisions, id)
delete(m.revPrefixes, id)
deleted++
}
return deleted, nil
}
func (m *Memory) getRevisionLocked(tenantID, revisionID string) (*Revision, error) {
rev, ok := m.revisions[revisionID]
if !ok {
return nil, ErrNotFound
}
if rev.TenantID != tenantID {
return nil, ErrTenantScope
}
return rev, nil
}
// ListRevisions returns recent revisions for tenant (newest first), cursor is opaque offset string.
func (m *Memory) ListRevisions(tenantID, moduleID string, cursor string, limit int) (items []*Revision, nextCursor string, hasMore bool) {
if limit <= 0 {
limit = 50
}
m.mu.RLock()
defer m.mu.RUnlock()
var all []*Revision
for _, r := range m.revisions {
if r.TenantID != tenantID {
continue
}
if moduleID != "" && r.ModuleID != moduleID {
continue
}
all = append(all, r)
}
sort.Slice(all, func(i, j int) bool {
return all[i].CreatedAt.After(all[j].CreatedAt)
})
off := 0
if cursor != "" {
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
off = n
}
}
end := off + limit
if end > len(all) {
end = len(all)
}
page := all[off:end]
if end < len(all) {
nextCursor = fmt.Sprintf("%d", end)
hasMore = true
}
return page, nextCursor, hasMore
}