Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
880d77810a | ||
|
|
8ebce28e34 | ||
|
|
dd7d43c2c2 | ||
|
|
a8c5e9701f | ||
|
|
be3d73f374 |
@@ -55,8 +55,12 @@ func main() {
|
||||
startBirdMetricsPoller(ctx)
|
||||
|
||||
httpSrv := &http.Server{
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: srv.Handler(),
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: srv.Handler(),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
ReadTimeout: 60 * time.Second,
|
||||
WriteTimeout: 120 * time.Second,
|
||||
IdleTimeout: 120 * time.Second,
|
||||
}
|
||||
go func() {
|
||||
svc := platform.ServiceName("evobgp-all")
|
||||
@@ -91,6 +95,7 @@ func startBirdMetricsPoller(ctx context.Context) {
|
||||
return birdfmt.ShowProtocols(ctx, socket, birdcBin)
|
||||
},
|
||||
birdfmt.CountEstablishedBGPSessions,
|
||||
birdfmt.ParseBGPProtocolStates,
|
||||
)
|
||||
log.Printf("birdc protocols poller enabled (socket=%s interval=%s)", sock, interval)
|
||||
}
|
||||
|
||||
@@ -42,8 +42,12 @@ func main() {
|
||||
startBirdMetricsPoller(ctx)
|
||||
|
||||
httpSrv := &http.Server{
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: srv.Handler(),
|
||||
Addr: cfg.HTTPAddr,
|
||||
Handler: srv.Handler(),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
ReadTimeout: 60 * time.Second,
|
||||
WriteTimeout: 120 * time.Second,
|
||||
IdleTimeout: 120 * time.Second,
|
||||
}
|
||||
go func() {
|
||||
svc := platform.ServiceName("evobgp-api")
|
||||
@@ -84,6 +88,7 @@ func startBirdMetricsPoller(ctx context.Context) {
|
||||
return birdfmt.ShowProtocols(ctx, socket, birdcBin)
|
||||
},
|
||||
birdfmt.CountEstablishedBGPSessions,
|
||||
birdfmt.ParseBGPProtocolStates,
|
||||
)
|
||||
log.Printf("birdc protocols poller enabled (socket=%s interval=%s)", sock, interval)
|
||||
}
|
||||
|
||||
@@ -9,6 +9,21 @@ import (
|
||||
|
||||
const maxBGPASN = 4294967295
|
||||
|
||||
const filterPrefixChunkSize = 500
|
||||
|
||||
func writePrefixSetAcceptBlocks(b *strings.Builder, keys []string) {
|
||||
for i := 0; i < len(keys); i += filterPrefixChunkSize {
|
||||
end := i + filterPrefixChunkSize
|
||||
if end > len(keys) {
|
||||
end = len(keys)
|
||||
}
|
||||
chunk := keys[i:end]
|
||||
b.WriteString(" if net ~ [ ")
|
||||
b.WriteString(strings.Join(chunk, ", "))
|
||||
b.WriteString(" ] then accept;\n")
|
||||
}
|
||||
}
|
||||
|
||||
func filterUniqueASNs(pathASNs []int64) []int64 {
|
||||
seen := make(map[int64]struct{})
|
||||
for _, a := range pathASNs {
|
||||
@@ -52,9 +67,7 @@ func RenderExportFilterIPv4(filterName string, prefixes []netip.Prefix, pathASNs
|
||||
b.WriteString(strings.TrimSpace(filterName))
|
||||
b.WriteString(" {\n")
|
||||
if len(keys) > 0 {
|
||||
b.WriteString(" if net ~ [ ")
|
||||
b.WriteString(strings.Join(keys, ", "))
|
||||
b.WriteString(" ] then accept;\n")
|
||||
writePrefixSetAcceptBlocks(&b, keys)
|
||||
}
|
||||
for _, asn := range asns {
|
||||
fmt.Fprintf(&b, " if bgp_path ~ [= * %d =] then accept;\n", asn)
|
||||
@@ -95,9 +108,7 @@ func RenderExportFilterIPv6(filterName string, prefixes []netip.Prefix, pathASNs
|
||||
b.WriteString(strings.TrimSpace(filterName))
|
||||
b.WriteString(" {\n")
|
||||
if len(keys) > 0 {
|
||||
b.WriteString(" if net ~ [ ")
|
||||
b.WriteString(strings.Join(keys, ", "))
|
||||
b.WriteString(" ] then accept;\n")
|
||||
writePrefixSetAcceptBlocks(&b, keys)
|
||||
}
|
||||
for _, asn := range asns {
|
||||
fmt.Fprintf(&b, " if bgp_path ~ [= * %d =] then accept;\n", asn)
|
||||
|
||||
@@ -50,3 +50,48 @@ func CountEstablishedBGPSessions(showProtocolsOutput string) int {
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// ParseBGPProtocolStates parses `birdc show protocols all` summary rows into protocol_name -> state.
|
||||
func ParseBGPProtocolStates(output string) map[string]string {
|
||||
out := make(map[string]string)
|
||||
for _, raw := range strings.Split(output, "\n") {
|
||||
line := strings.TrimSpace(raw)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
low := strings.ToLower(line)
|
||||
if strings.HasPrefix(low, "bird ") || strings.HasPrefix(low, "name ") || strings.HasPrefix(low, "table ") {
|
||||
continue
|
||||
}
|
||||
fields := strings.Fields(line)
|
||||
if len(fields) < 4 {
|
||||
continue
|
||||
}
|
||||
if !strings.EqualFold(fields[1], "BGP") {
|
||||
continue
|
||||
}
|
||||
state := extractBGPSessionStateLine(line)
|
||||
if state == "" {
|
||||
state = fields[3]
|
||||
}
|
||||
out[fields[0]] = state
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func extractBGPSessionStateLine(line string) string {
|
||||
known := []string{
|
||||
"Established",
|
||||
"Idle",
|
||||
"Connect",
|
||||
"Active",
|
||||
"OpenSent",
|
||||
"OpenConfirm",
|
||||
}
|
||||
for _, st := range known {
|
||||
if strings.Contains(line, st) {
|
||||
return st
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@@ -6,9 +6,12 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"evobgp/migrations"
|
||||
|
||||
@@ -22,6 +25,17 @@ func OpenPostgresPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if max := os.Getenv("EVOBGP_DB_MAX_CONNS"); max != "" {
|
||||
if n, err := strconv.Atoi(strings.TrimSpace(max)); err == nil && n > 0 {
|
||||
cfg.MaxConns = int32(n)
|
||||
}
|
||||
}
|
||||
if min := os.Getenv("EVOBGP_DB_MIN_CONNS"); min != "" {
|
||||
if n, err := strconv.Atoi(strings.TrimSpace(min)); err == nil && n >= 0 {
|
||||
cfg.MinConns = int32(n)
|
||||
}
|
||||
}
|
||||
cfg.MaxConnLifetime = 30 * time.Minute
|
||||
pool, err := pgxpool.NewWithConfig(ctx, cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -53,6 +53,21 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R
|
||||
wk := &jobs.Worker{Store: backend, HTTPClient: cdnHTTP}
|
||||
reg := jobs.NewRegistry(wk.Process)
|
||||
wk.Registry = reg
|
||||
if pool != nil {
|
||||
audit := repository.NewJobAuditWriter(pool)
|
||||
reg.SetTerminalHook(func(j *jobs.Job) {
|
||||
if j == nil {
|
||||
return
|
||||
}
|
||||
st := j.Snapshot()
|
||||
status, _ := st["status"].(string)
|
||||
var errMsg *string
|
||||
if e, ok := st["error"].(string); ok && e != "" {
|
||||
errMsg = &e
|
||||
}
|
||||
audit.MarkTerminal(context.Background(), j.TenantID, j.ID, status, errMsg, time.Now().UTC())
|
||||
})
|
||||
}
|
||||
observability.RegisterStoreBackend(backend)
|
||||
return backend, reg, pool, nil
|
||||
}
|
||||
|
||||
@@ -69,6 +69,7 @@ func (s *Server) registerV1(m *http.ServeMux) {
|
||||
m.HandleFunc("GET /bird/status", s.handleBirdStatus)
|
||||
m.HandleFunc("GET /jobs", s.handleListJobs)
|
||||
m.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
|
||||
m.HandleFunc("GET /jobs/{job_id}/report", s.handleGetJobReport)
|
||||
m.HandleFunc("POST /jobs/{job_id}/cancel", s.handleCancelJob)
|
||||
m.HandleFunc("GET /speakers/{speaker_id}/revisions/latest", s.handleNodeLatestRevision)
|
||||
m.HandleFunc("GET /speakers/{speaker_id}/bundle/{revision_id}", s.handleNodeBundle)
|
||||
@@ -201,6 +202,22 @@ func (s *Server) handleListModules(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
filtered := make([]*store.Module, 0)
|
||||
limit := parseListLimit(r)
|
||||
cursor := r.URL.Query().Get("cursor")
|
||||
if typeFilter == "" && enabledFilter == nil {
|
||||
page, next, more := s.store.ListModulesPage(a.TenantID, cursor, limit)
|
||||
for _, mod := range page {
|
||||
filtered = append(filtered, mod)
|
||||
}
|
||||
items := make([]map[string]any, 0, len(filtered))
|
||||
for _, mod := range filtered {
|
||||
items = append(items, moduleJSON(mod))
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"items": items, "next_cursor": strPtrOrNull(next), "has_more": more,
|
||||
})
|
||||
return
|
||||
}
|
||||
for _, mod := range s.store.ListModules(a.TenantID) {
|
||||
if typeFilter != "" && mod.Type != typeFilter {
|
||||
continue
|
||||
@@ -275,7 +292,7 @@ func (s *Server) handleListPeers(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
allPeers := s.store.ListPeers(a.TenantID)
|
||||
page, next, more := store.PaginateOffset(allPeers, r.URL.Query().Get("cursor"), parseListLimit(r))
|
||||
liveStates := s.liveBGPProtocolStates(r.Context())
|
||||
liveStates := s.liveBGPProtocolStates(r)
|
||||
items := make([]map[string]any, 0, len(page))
|
||||
for _, p := range page {
|
||||
row := peerJSON(p)
|
||||
@@ -289,7 +306,17 @@ func (s *Server) handleListPeers(w http.ResponseWriter, r *http.Request) {
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) liveBGPProtocolStates(ctx context.Context) map[string]string {
|
||||
func (s *Server) liveBGPProtocolStates(r *http.Request) map[string]string {
|
||||
if r != nil && strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("live")), "1") {
|
||||
return s.liveBGPProtocolStatesFresh(r.Context())
|
||||
}
|
||||
if cached, ok := observability.CachedBirdProtocolStates(90 * time.Second); ok {
|
||||
return cached
|
||||
}
|
||||
return s.liveBGPProtocolStatesFresh(r.Context())
|
||||
}
|
||||
|
||||
func (s *Server) liveBGPProtocolStatesFresh(ctx context.Context) map[string]string {
|
||||
sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET"))
|
||||
if sock == "" {
|
||||
return map[string]string{}
|
||||
@@ -298,7 +325,9 @@ func (s *Server) liveBGPProtocolStates(ctx context.Context) map[string]string {
|
||||
if err != nil {
|
||||
return map[string]string{}
|
||||
}
|
||||
return parseBGPProtocolStates(out)
|
||||
states := birdfmt.ParseBGPProtocolStates(out)
|
||||
observability.SetBirdProtocolStates(states)
|
||||
return states
|
||||
}
|
||||
|
||||
// parseBGPProtocolStates parses `birdc show protocols all` summary rows into protocol_name -> state.
|
||||
@@ -522,7 +551,8 @@ func (s *Server) enqueueModuleRefreshIfEnabled(tenantID, moduleID, trigger strin
|
||||
return
|
||||
}
|
||||
mid := moduleID
|
||||
_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, nil, &mid, map[string]any{
|
||||
key := "module_refresh:" + moduleID
|
||||
_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, &key, &mid, map[string]any{
|
||||
"module_id": moduleID,
|
||||
"trigger": trigger,
|
||||
})
|
||||
@@ -589,6 +619,9 @@ func (s *Server) handleRevisionPreview(w http.ResponseWriter, r *http.Request) {
|
||||
for k, v := range rev.PreviewFragments {
|
||||
obj[k] = v
|
||||
}
|
||||
if expanded := pipeline.BuildExpandedBirdPreview(rev.PreviewFragments); expanded != "" {
|
||||
obj[pipeline.AuxBirdFullExpandedKey()] = expanded
|
||||
}
|
||||
writeJSON(w, http.StatusOK, obj)
|
||||
}
|
||||
|
||||
@@ -843,6 +876,44 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, j.Snapshot())
|
||||
}
|
||||
|
||||
func (s *Server) handleGetJobReport(w http.ResponseWriter, r *http.Request) {
|
||||
a, ok := authFromContext(r.Context())
|
||||
if !ok {
|
||||
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
|
||||
return
|
||||
}
|
||||
if !s.requireAtLeast(w, a, "viewer") {
|
||||
return
|
||||
}
|
||||
j, err := s.jobs.Get(a.TenantID, r.PathValue("job_id"))
|
||||
if err != nil {
|
||||
writeProblem(w, http.StatusNotFound, "Not Found", "job not found")
|
||||
return
|
||||
}
|
||||
snap := j.Snapshot()
|
||||
meta, _ := snap["meta"].(map[string]any)
|
||||
out := map[string]any{
|
||||
"job_id": snap["job_id"],
|
||||
"kind": snap["kind"],
|
||||
"status": snap["status"],
|
||||
"meta": meta,
|
||||
"error": snap["error"],
|
||||
"created_at": snap["created_at"],
|
||||
}
|
||||
if meta != nil {
|
||||
if v, ok := meta["log_entries"]; ok {
|
||||
out["log_entries"] = v
|
||||
}
|
||||
if v, ok := meta["log_total"]; ok {
|
||||
out["log_total"] = v
|
||||
}
|
||||
if v, ok := meta["revision_id"]; ok {
|
||||
out["revision_id"] = v
|
||||
}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
|
||||
a, ok := authFromContext(r.Context())
|
||||
if !ok {
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultJobTimeoutModuleRefresh = 10 * time.Minute
|
||||
defaultJobTimeoutTenantRefresh = 15 * time.Minute
|
||||
defaultJobTimeoutDeployApply = 5 * time.Minute
|
||||
defaultJobTimeoutPeerReconcile = 10 * time.Minute
|
||||
defaultJobTimeoutRollback = 5 * time.Minute
|
||||
defaultJobTimeoutBirdReload = 2 * time.Minute
|
||||
)
|
||||
|
||||
func jobTimeout(kind string) time.Duration {
|
||||
envKey := map[string]string{
|
||||
KindModuleRefresh: "EVOBGP_JOB_TIMEOUT_MODULE_REFRESH",
|
||||
KindTenantRefresh: "EVOBGP_JOB_TIMEOUT_TENANT_REFRESH",
|
||||
KindDeployApply: "EVOBGP_JOB_TIMEOUT_DEPLOY_APPLY",
|
||||
KindPeerReconcile: "EVOBGP_JOB_TIMEOUT_PEER_RECONCILE",
|
||||
KindRevisionRollback: "EVOBGP_JOB_TIMEOUT_ROLLBACK",
|
||||
KindBirdReload: "EVOBGP_JOB_TIMEOUT_BIRD_RELOAD",
|
||||
}[kind]
|
||||
if envKey != "" {
|
||||
if d, err := time.ParseDuration(os.Getenv(envKey)); err == nil && d > 0 {
|
||||
return d
|
||||
}
|
||||
}
|
||||
switch kind {
|
||||
case KindModuleRefresh:
|
||||
return defaultJobTimeoutModuleRefresh
|
||||
case KindTenantRefresh:
|
||||
return defaultJobTimeoutTenantRefresh
|
||||
case KindDeployApply:
|
||||
return defaultJobTimeoutDeployApply
|
||||
case KindPeerReconcile:
|
||||
return defaultJobTimeoutPeerReconcile
|
||||
case KindRevisionRollback:
|
||||
return defaultJobTimeoutRollback
|
||||
case KindBirdReload:
|
||||
return defaultJobTimeoutBirdReload
|
||||
default:
|
||||
if n, err := strconv.Atoi(os.Getenv("EVOBGP_JOB_TIMEOUT_SEC")); err == nil && n > 0 {
|
||||
return time.Duration(n) * time.Second
|
||||
}
|
||||
return defaultJobTimeoutModuleRefresh
|
||||
}
|
||||
}
|
||||
|
||||
// workContext returns a timeout context that also cancels when the job is cancelled.
|
||||
func (j *Job) workContext() (context.Context, context.CancelFunc) {
|
||||
if j == nil {
|
||||
return context.Background(), func() {}
|
||||
}
|
||||
timeout := jobTimeout(j.Kind)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
go func() {
|
||||
ticker := time.NewTicker(500 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if j.IsCancelRequested() {
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return ctx, cancel
|
||||
}
|
||||
+51
-2
@@ -9,6 +9,8 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"evobgp/internal/observability"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
@@ -178,6 +180,8 @@ type Registry struct {
|
||||
byID map[string]*Job
|
||||
byIdempo map[idempoKey]*Job
|
||||
workerStart func(j *Job)
|
||||
workerSem chan struct{}
|
||||
onTerminal func(j *Job)
|
||||
}
|
||||
|
||||
type idempoKey struct {
|
||||
@@ -186,13 +190,44 @@ type idempoKey struct {
|
||||
}
|
||||
|
||||
func NewRegistry(workerStart func(j *Job)) *Registry {
|
||||
maxWorkers := registryMaxConcurrentJobs()
|
||||
return &Registry{
|
||||
byID: make(map[string]*Job),
|
||||
byIdempo: make(map[idempoKey]*Job),
|
||||
workerStart: workerStart,
|
||||
workerSem: make(chan struct{}, maxWorkers),
|
||||
}
|
||||
}
|
||||
|
||||
// SetTerminalHook registers a best-effort callback when jobs reach a terminal state.
|
||||
func (r *Registry) SetTerminalHook(fn func(j *Job)) {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.onTerminal = fn
|
||||
}
|
||||
|
||||
func (r *Registry) fireTerminal(j *Job) {
|
||||
if r == nil || j == nil {
|
||||
return
|
||||
}
|
||||
r.mu.RLock()
|
||||
fn := r.onTerminal
|
||||
r.mu.RUnlock()
|
||||
if fn != nil {
|
||||
fn(j)
|
||||
}
|
||||
}
|
||||
|
||||
func registryMaxConcurrentJobs() int {
|
||||
if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_MAX_CONCURRENT"))); err == nil && n > 0 {
|
||||
return n
|
||||
}
|
||||
return 8
|
||||
}
|
||||
|
||||
// pruneTerminalIfOver удаляет самые старые завершённые джобы (succeeded/failed/cancelled), пока len(byID) > maxJobs.
|
||||
func (r *Registry) pruneTerminalIfOver(maxJobs int) {
|
||||
if r == nil || maxJobs <= 0 || len(r.byID) <= maxJobs {
|
||||
@@ -244,7 +279,11 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
|
||||
if idempotencyKey != nil && *idempotencyKey != "" {
|
||||
k := idempoKey{tenant: tenantID, key: *idempotencyKey}
|
||||
if existing, ok := r.byIdempo[k]; ok {
|
||||
return existing, false, nil
|
||||
st := existing.statusLocked()
|
||||
if st == StatusQueued || st == StatusRunning {
|
||||
return existing, false, nil
|
||||
}
|
||||
delete(r.byIdempo, k)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -265,7 +304,17 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
|
||||
r.pruneTerminalIfOver(maxJobs)
|
||||
|
||||
if r.workerStart != nil {
|
||||
go r.workerStart(j)
|
||||
go func() {
|
||||
r.workerSem <- struct{}{}
|
||||
active := len(r.workerSem)
|
||||
capacity := cap(r.workerSem)
|
||||
observability.RecordJobQueueDepth(active, capacity)
|
||||
defer func() {
|
||||
<-r.workerSem
|
||||
observability.RecordJobQueueDepth(len(r.workerSem), capacity)
|
||||
}()
|
||||
r.workerStart(j)
|
||||
}()
|
||||
}
|
||||
return j, true, nil
|
||||
}
|
||||
|
||||
+56
-8
@@ -82,6 +82,9 @@ func (w *Worker) httpClient() *http.Client {
|
||||
func (w *Worker) Process(j *Job) {
|
||||
defer func() {
|
||||
observability.RecordJobTerminal(j.Kind, j.statusLocked())
|
||||
if w != nil && w.Registry != nil {
|
||||
w.Registry.fireTerminal(j)
|
||||
}
|
||||
}()
|
||||
|
||||
if w == nil || w.Store == nil {
|
||||
@@ -102,7 +105,17 @@ func (w *Worker) Process(j *Job) {
|
||||
j.Fail("missing module_id in job meta")
|
||||
return
|
||||
}
|
||||
if err := pipeline.RefreshModuleIngest(context.Background(), w.Store, w.httpClient(), j.TenantID, mid); err != nil {
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
if err := pipeline.RefreshModuleIngest(ctx, w.Store, w.httpClient(), j.TenantID, mid); err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
@@ -121,11 +134,17 @@ func (w *Worker) Process(j *Job) {
|
||||
j.Succeed()
|
||||
return
|
||||
}
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
ctl := &birdfmt.BirdCtl{
|
||||
Socket: sock,
|
||||
Birdc: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")),
|
||||
}
|
||||
if err := ctl.Configure(context.Background()); err != nil {
|
||||
if err := ctl.Configure(ctx); err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
@@ -152,9 +171,14 @@ func (w *Worker) runPeerReconcile(j *Job) {
|
||||
return
|
||||
}
|
||||
if len(latest) == 0 {
|
||||
// First run fallback: render full tenant state once if no baseline revision exists yet.
|
||||
rid, err := pipeline.RenderTenantRevision(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID)
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
rid, err := pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
@@ -171,8 +195,14 @@ func (w *Worker) runPeerReconcile(j *Job) {
|
||||
}
|
||||
cursor = next
|
||||
}
|
||||
rid, err := pipeline.RenderTenantRevisionFromPrefixes(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID, rows)
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
rid, err := pipeline.RenderTenantRevisionFromPrefixes(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID, rows)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
@@ -230,7 +260,13 @@ func (w *Worker) runTenantRefresh(j *Job) {
|
||||
j.Fail("missing module_ids in job meta")
|
||||
return
|
||||
}
|
||||
if err := pipeline.RefreshTenantModules(context.Background(), w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil {
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
if err := pipeline.RefreshTenantModules(ctx, w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
@@ -275,6 +311,8 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
|
||||
mu := w.tenantRefreshMu(j.TenantID)
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
deferDeploy := false
|
||||
if w.Registry != nil {
|
||||
deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0
|
||||
@@ -288,8 +326,12 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
|
||||
return
|
||||
}
|
||||
|
||||
rev, err := pipeline.RenderTenantRevision(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID)
|
||||
rev, err := pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
@@ -335,6 +377,8 @@ func (w *Worker) runDeployApply(j *Job) {
|
||||
j.Fail("missing revision_id in job meta")
|
||||
return
|
||||
}
|
||||
ctx, cancel := j.workContext()
|
||||
defer cancel()
|
||||
activeDir := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR"))
|
||||
if activeDir != "" {
|
||||
revObj, err := w.Store.GetRevision(j.TenantID, revID)
|
||||
@@ -354,7 +398,11 @@ func (w *Worker) runDeployApply(j *Job) {
|
||||
Socket: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")),
|
||||
}
|
||||
ctl := &birdfmt.BirdCtl{Bird: cfg.BirdBin, Birdc: cfg.BirdcBin, Socket: cfg.Socket}
|
||||
if err := birddeploy.ApplyRevision(context.Background(), ctl, revObj, cfg); err != nil {
|
||||
if err := birddeploy.ApplyRevision(ctx, ctl, revObj, cfg); err != nil {
|
||||
if ctx.Err() != nil {
|
||||
j.MarkCancelled()
|
||||
return
|
||||
}
|
||||
j.Fail(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
@@ -80,6 +80,38 @@ var (
|
||||
Help: "Prefix row count after CIDR aggregation on tenant render.",
|
||||
Buckets: prometheus.ExponentialBuckets(1, 2, 16),
|
||||
})
|
||||
|
||||
pipelineRefreshDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
|
||||
Namespace: namespace,
|
||||
Name: "pipeline_refresh_duration_seconds",
|
||||
Help: "Module refresh ingest duration by module type.",
|
||||
Buckets: prometheus.ExponentialBuckets(0.05, 2, 14),
|
||||
}, []string{"module_type"})
|
||||
|
||||
renderPrefixCount = promauto.NewHistogram(prometheus.HistogramOpts{
|
||||
Namespace: namespace,
|
||||
Name: "render_prefix_count",
|
||||
Help: "Materialized prefix count per tenant render.",
|
||||
Buckets: prometheus.ExponentialBuckets(10, 2, 16),
|
||||
})
|
||||
|
||||
jobQueueActive = promauto.NewGauge(prometheus.GaugeOpts{
|
||||
Namespace: namespace,
|
||||
Name: "job_queue_active",
|
||||
Help: "Currently running in-process async jobs.",
|
||||
})
|
||||
|
||||
jobQueueCapacity = promauto.NewGauge(prometheus.GaugeOpts{
|
||||
Namespace: namespace,
|
||||
Name: "job_queue_capacity",
|
||||
Help: "Maximum concurrent in-process async jobs.",
|
||||
})
|
||||
)
|
||||
|
||||
var (
|
||||
birdProtocolStatesMu sync.RWMutex
|
||||
birdProtocolStates map[string]string
|
||||
birdProtocolStatesAt time.Time
|
||||
)
|
||||
|
||||
// RecordPrefixAggregation records tenant render CIDR aggregation stats.
|
||||
@@ -93,6 +125,27 @@ func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Durati
|
||||
prefixAggregationDuration.Observe(duration.Seconds())
|
||||
prefixAggregationRawCount.Observe(float64(rawCount))
|
||||
prefixAggregationAggregatedCount.Observe(float64(aggregatedCount))
|
||||
renderPrefixCount.Observe(float64(aggregatedCount))
|
||||
}
|
||||
|
||||
// RecordPipelineRefresh records module ingest duration.
|
||||
func RecordPipelineRefresh(moduleType string, duration time.Duration) {
|
||||
if moduleType == "" {
|
||||
moduleType = "unknown"
|
||||
}
|
||||
pipelineRefreshDuration.WithLabelValues(moduleType).Observe(duration.Seconds())
|
||||
}
|
||||
|
||||
// RecordJobQueueDepth updates in-process job worker utilization gauges.
|
||||
func RecordJobQueueDepth(active, capacity int) {
|
||||
if active < 0 {
|
||||
active = 0
|
||||
}
|
||||
if capacity < 0 {
|
||||
capacity = 0
|
||||
}
|
||||
jobQueueActive.Set(float64(active))
|
||||
jobQueueCapacity.Set(float64(capacity))
|
||||
}
|
||||
|
||||
// RecordJobTerminal increments jobs_finished_total for terminal statuses.
|
||||
@@ -205,6 +258,35 @@ func SetBirdSessionMetrics(established int, scrapeOK bool) {
|
||||
}
|
||||
}
|
||||
|
||||
// SetBirdProtocolStates caches parsed BGP protocol states from the last birdc scrape.
|
||||
func SetBirdProtocolStates(states map[string]string) {
|
||||
birdProtocolStatesMu.Lock()
|
||||
defer birdProtocolStatesMu.Unlock()
|
||||
if states == nil {
|
||||
birdProtocolStates = map[string]string{}
|
||||
} else {
|
||||
birdProtocolStates = states
|
||||
}
|
||||
birdProtocolStatesAt = time.Now()
|
||||
}
|
||||
|
||||
// CachedBirdProtocolStates returns cached protocol states if younger than maxAge.
|
||||
func CachedBirdProtocolStates(maxAge time.Duration) (map[string]string, bool) {
|
||||
if maxAge <= 0 {
|
||||
maxAge = 60 * time.Second
|
||||
}
|
||||
birdProtocolStatesMu.RLock()
|
||||
defer birdProtocolStatesMu.RUnlock()
|
||||
if birdProtocolStates == nil || time.Since(birdProtocolStatesAt) > maxAge {
|
||||
return nil, false
|
||||
}
|
||||
out := make(map[string]string, len(birdProtocolStates))
|
||||
for k, v := range birdProtocolStates {
|
||||
out[k] = v
|
||||
}
|
||||
return out, true
|
||||
}
|
||||
|
||||
// MetricsHandler returns the Prometheus scrape handler.
|
||||
func MetricsHandler() http.Handler {
|
||||
return promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{})
|
||||
@@ -231,7 +313,7 @@ func (s *statusRecorder) WriteHeader(code int) {
|
||||
|
||||
// StartBirdProtocolsPoller runs birdc "show protocols" on interval when socket is non-empty.
|
||||
// Горутина завершается при отмене ctx (корректное завершение вместе с процессом API).
|
||||
func StartBirdProtocolsPoller(ctx context.Context, socket string, birdcPath string, interval time.Duration, showFn func(ctx context.Context, socket, birdcBin string) (string, error), countFn func(output string) int) {
|
||||
func StartBirdProtocolsPoller(ctx context.Context, socket string, birdcPath string, interval time.Duration, showFn func(ctx context.Context, socket, birdcBin string) (string, error), countFn func(output string) int, parseFn func(output string) map[string]string) {
|
||||
socket = trimSpace(socket)
|
||||
if ctx == nil || socket == "" || interval <= 0 || showFn == nil || countFn == nil {
|
||||
return
|
||||
@@ -245,6 +327,9 @@ func StartBirdProtocolsPoller(ctx context.Context, socket string, birdcPath stri
|
||||
return
|
||||
}
|
||||
SetBirdSessionMetrics(countFn(out), true)
|
||||
if parseFn != nil {
|
||||
SetBirdProtocolStates(parseFn(out))
|
||||
}
|
||||
}
|
||||
go func() {
|
||||
scrape()
|
||||
|
||||
@@ -44,6 +44,7 @@ func resolveASNForEntry(ctx context.Context, st store.Backend, hc *http.Client,
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
asnresolve.PolitePause()
|
||||
holder, _ := asnresolve.ASHolderName(ctx, hc, asn)
|
||||
if st != nil {
|
||||
strs := make([]string, len(pfxs))
|
||||
|
||||
@@ -25,9 +25,6 @@ func cachedCDNPrefixRows(st store.Backend, tenantID, moduleID string, priorSnaps
|
||||
return cached
|
||||
}
|
||||
}
|
||||
if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 {
|
||||
return cached
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -45,6 +42,35 @@ func mergeSnapshotDropSource(rows []store.PrefixRow, sourceKey string) []store.P
|
||||
return out
|
||||
}
|
||||
|
||||
// mergeSnapshotDropCDNSources removes all cdn:* rows (used before batch CDN merge).
|
||||
func mergeSnapshotDropCDNSources(rows []store.PrefixRow) []store.PrefixRow {
|
||||
if len(rows) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]store.PrefixRow, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
if !strings.HasPrefix(strings.TrimSpace(row.Source), "cdn:") {
|
||||
out = append(out, row)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// mergeAllCDNSourcesIntoModuleSnapshot replaces all CDN rows in one write (avoids parallel read-modify-write races).
|
||||
func mergeAllCDNSourcesIntoModuleSnapshot(st store.Backend, tenantID string, mod *store.Module, priorSnapshot []store.PrefixRow, cdnRows []store.PrefixRow) error {
|
||||
if st == nil || mod == nil {
|
||||
return nil
|
||||
}
|
||||
var base []store.PrefixRow
|
||||
if len(priorSnapshot) > 0 {
|
||||
base = mergeSnapshotDropCDNSources(priorSnapshot)
|
||||
} else if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, mod.ID); ok && snap != nil {
|
||||
base = mergeSnapshotDropCDNSources(snap.Prefixes)
|
||||
}
|
||||
merged := append(base, cdnRows...)
|
||||
return persistModuleSnapshot(st, tenantID, mod, merged)
|
||||
}
|
||||
|
||||
func cdnRowsFromParsed(mod *store.Module, src *store.CDNSource, pfxStrings []string) []store.PrefixRow {
|
||||
var rows []store.PrefixRow
|
||||
for _, p := range pfxStrings {
|
||||
@@ -156,3 +182,68 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
// fetchCDNSourceRows loads CDN prefixes without persisting the module snapshot (caller merges once).
|
||||
func fetchCDNSourceRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string, mod *store.Module, src *store.CDNSource, priorSnapshot []store.PrefixRow, now time.Time) ([]store.PrefixRow, error) {
|
||||
u := strings.TrimSpace(src.URL)
|
||||
if u == "" {
|
||||
return nil, nil
|
||||
}
|
||||
sourceKey := cdnSourceKey(src.ID)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if etag := strings.TrimSpace(src.Etag); etag != "" {
|
||||
req.Header.Set("If-None-Match", etag)
|
||||
}
|
||||
resp, err := hc.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
||||
}
|
||||
|
||||
if resp.StatusCode == http.StatusNotModified {
|
||||
if cached := cachedCDNPrefixRows(st, tenantID, moduleID, priorSnapshot, sourceKey); len(cached) > 0 {
|
||||
_ = resp.Body.Close()
|
||||
return cached, nil
|
||||
}
|
||||
_ = resp.Body.Close()
|
||||
req2, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err = hc.Do(req2)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
|
||||
}
|
||||
if resp.StatusCode == http.StatusNotModified {
|
||||
_ = resp.Body.Close()
|
||||
return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u)
|
||||
}
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status)
|
||||
}
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
prefixStrs, err := parseCDNBody(string(body), src)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cdn parse %s: %w", u, err)
|
||||
}
|
||||
etag := strings.TrimSpace(resp.Header.Get("ETag"))
|
||||
patch := &store.CDNSourcePatch{}
|
||||
if etag != "" && etag != strings.TrimSpace(src.Etag) {
|
||||
e := etag
|
||||
patch.Etag = &e
|
||||
}
|
||||
refreshedAt := now
|
||||
patch.LastRefreshedAt = &refreshedAt
|
||||
_, _ = st.UpdateCDNSource(tenantID, moduleID, src.ID, patch)
|
||||
|
||||
return cdnRowsFromParsed(mod, src, prefixStrs), nil
|
||||
}
|
||||
|
||||
@@ -97,15 +97,18 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client,
|
||||
|
||||
seenPfx := make(map[string]struct{})
|
||||
var out []store.PrefixRow
|
||||
var metaUpdates []store.ASEntryResolveMetaUpdate
|
||||
now := time.Now().UTC()
|
||||
for _, r := range results {
|
||||
if r.err != nil {
|
||||
return nil, r.err
|
||||
}
|
||||
if r.metaID != "" {
|
||||
if err := st.UpdateASEntryResolveMeta(tenantID, moduleID, r.metaID, r.holder, r.count, now); err != nil {
|
||||
return nil, fmt.Errorf("as entry meta AS%d: %w", r.asn, err)
|
||||
}
|
||||
metaUpdates = append(metaUpdates, store.ASEntryResolveMetaUpdate{
|
||||
EntryID: r.metaID,
|
||||
ASNName: r.holder,
|
||||
PrefixCount: r.count,
|
||||
})
|
||||
}
|
||||
for _, row := range r.rows {
|
||||
k := row.Prefix
|
||||
@@ -116,6 +119,11 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client,
|
||||
out = append(out, row)
|
||||
}
|
||||
}
|
||||
if len(metaUpdates) > 0 {
|
||||
if err := st.UpdateASEntryResolveMetaBatch(tenantID, moduleID, metaUpdates, now); err != nil {
|
||||
return nil, fmt.Errorf("as entry meta batch: %w", err)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -150,12 +158,14 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
|
||||
results[idx] = srcResult{rows: cached}
|
||||
return
|
||||
}
|
||||
if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 {
|
||||
results[idx] = srcResult{rows: cached}
|
||||
return
|
||||
if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil {
|
||||
if cached := prefixRowsForSource(snap.Prefixes, sourceKey); len(cached) > 0 {
|
||||
results[idx] = srcResult{rows: cached}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
rows, err := applyCDNSourceHTTPResult(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now)
|
||||
rows, err := fetchCDNSourceRows(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now)
|
||||
if err != nil {
|
||||
results[idx] = srcResult{err: err}
|
||||
return
|
||||
@@ -172,6 +182,11 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
|
||||
}
|
||||
out = append(out, r.rows...)
|
||||
}
|
||||
if len(valid) > 0 {
|
||||
if err := mergeAllCDNSourcesIntoModuleSnapshot(st, tenantID, mod, priorSnapshot, out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -36,6 +36,11 @@ const (
|
||||
revisionDefaultTTL = 30 * 24 * time.Hour
|
||||
)
|
||||
|
||||
// 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)
|
||||
@@ -47,10 +52,14 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client,
|
||||
if hc == nil {
|
||||
hc = http.DefaultClient
|
||||
}
|
||||
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")
|
||||
}
|
||||
@@ -197,33 +206,6 @@ func shouldSkipCDNSourceFetch(src *store.CDNSource, now time.Time) bool {
|
||||
return now.UTC().Before(nextRefreshAt)
|
||||
}
|
||||
|
||||
func latestCDNRowsBySource(st store.Backend, tenantID string) map[string][]store.PrefixRow {
|
||||
out := make(map[string][]store.PrefixRow)
|
||||
if st == nil {
|
||||
return out
|
||||
}
|
||||
revs, _, _ := st.ListRevisions(tenantID, "", "", 1)
|
||||
if len(revs) == 0 || strings.TrimSpace(revs[0].ID) == "" {
|
||||
return out
|
||||
}
|
||||
revID := strings.TrimSpace(revs[0].ID)
|
||||
cursor := ""
|
||||
for {
|
||||
page, next, more := st.ListRevisionPrefixes(tenantID, revID, cursor, 2000)
|
||||
for _, row := range page {
|
||||
if !strings.HasPrefix(strings.TrimSpace(row.Source), "cdn:") {
|
||||
continue
|
||||
}
|
||||
out[row.Source] = append(out[row.Source], row)
|
||||
}
|
||||
if !more || strings.TrimSpace(next) == "" {
|
||||
break
|
||||
}
|
||||
cursor = next
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
type dohJSONAnswer struct {
|
||||
Type int `json:"type"`
|
||||
Data string `json:"data"`
|
||||
@@ -836,10 +818,18 @@ func buildPreviewFragments(st store.Backend, tenantID, moduleID, revisionID stri
|
||||
px6: birdfmt.JoinFragments(birdfmt.ManagedBanner(revisionID), staticV6),
|
||||
pPeers: peersBody,
|
||||
}
|
||||
out[auxBirdFullExpanded] = buildExpandedBirdText(main, out)
|
||||
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
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// JobAuditWriter persists async job lifecycle rows to job_audit (optional cross-process queue foundation).
|
||||
type JobAuditWriter struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewJobAuditWriter(pool *pgxpool.Pool) *JobAuditWriter {
|
||||
if pool == nil {
|
||||
return nil
|
||||
}
|
||||
return &JobAuditWriter{pool: pool}
|
||||
}
|
||||
|
||||
// UpsertRunning inserts or updates a running job row (best-effort).
|
||||
func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, meta map[string]any) {
|
||||
if w == nil || w.pool == nil {
|
||||
return
|
||||
}
|
||||
metaJSON, _ := json.Marshal(meta)
|
||||
var idem any
|
||||
if idempotencyKey != nil && *idempotencyKey != "" {
|
||||
idem = *idempotencyKey
|
||||
}
|
||||
_, _ = w.pool.Exec(ctx, `
|
||||
INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, meta_json, created_at, started_at)
|
||||
VALUES ($1::uuid, $2::uuid, $3, 'running', $4, $5::jsonb, now(), now())
|
||||
ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL
|
||||
DO UPDATE SET status='running', started_at=now(), meta_json=EXCLUDED.meta_json`,
|
||||
jobID, tenantID, kind, idem, metaJSON)
|
||||
}
|
||||
|
||||
// MarkTerminal updates job_audit terminal state (best-effort).
|
||||
func (w *JobAuditWriter) MarkTerminal(ctx context.Context, tenantID, jobID, status string, errMsg *string, finishedAt time.Time) {
|
||||
if w == nil || w.pool == nil {
|
||||
return
|
||||
}
|
||||
_, _ = w.pool.Exec(ctx, `
|
||||
UPDATE job_audit SET status=$3, error_message=$4, finished_at=$5
|
||||
WHERE id=$1::uuid AND tenant_id=$2::uuid`,
|
||||
jobID, tenantID, status, errMsg, finishedAt.UTC())
|
||||
}
|
||||
+127
-27
@@ -126,6 +126,7 @@ func (p *Postgres) ListModules(tenantID string) []*store.Module {
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*store.Module
|
||||
moduleByID := make(map[string]*store.Module)
|
||||
for rows.Next() {
|
||||
var m store.Module
|
||||
m.TenantID = tenantID
|
||||
@@ -152,14 +153,84 @@ func (p *Postgres) ListModules(tenantID string) []*store.Module {
|
||||
t := last.UTC()
|
||||
m.LastRefreshedAt = &t
|
||||
}
|
||||
if err := p.fillModuleDohFields(ctx, &m); err != nil {
|
||||
continue
|
||||
}
|
||||
out = append(out, &m)
|
||||
moduleByID[m.ID] = &m
|
||||
}
|
||||
if err := p.batchFillModuleDohFields(ctx, moduleByID); err != nil {
|
||||
return nil
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (p *Postgres) ListModulesPage(tenantID, cursor string, limit int) ([]*store.Module, string, bool) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
off := 0
|
||||
if cursor != "" {
|
||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
||||
off = n
|
||||
}
|
||||
}
|
||||
ctx := context.Background()
|
||||
rows, err := p.pool.Query(ctx, `
|
||||
SELECT id, type, name, enabled, priority, doh_profile_id::text, doh_resolver_policy,
|
||||
refresh_interval_sec, cron_expr, default_community_id::text, last_refreshed_at
|
||||
FROM module WHERE tenant_id = $1 AND deleted_at IS NULL
|
||||
ORDER BY priority, name
|
||||
LIMIT $2 OFFSET $3`, tenantID, limit+1, off)
|
||||
if err != nil {
|
||||
return nil, "", false
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*store.Module
|
||||
moduleByID := make(map[string]*store.Module)
|
||||
for rows.Next() {
|
||||
var m store.Module
|
||||
m.TenantID = tenantID
|
||||
var doh, dc, cron *string
|
||||
var refresh *int32
|
||||
var last *time.Time
|
||||
if err := rows.Scan(&m.ID, &m.Type, &m.Name, &m.Enabled, &m.Priority, &doh, &m.DohResolverPolicy, &refresh, &cron, &dc, &last); err != nil {
|
||||
continue
|
||||
}
|
||||
m.DohResolverPolicy = store.NormalizeDohResolverPolicy(m.DohResolverPolicy)
|
||||
if refresh != nil {
|
||||
m.RefreshIntervalSec = int(*refresh)
|
||||
}
|
||||
if cron != nil {
|
||||
m.CronExpr = *cron
|
||||
}
|
||||
if doh != nil && *doh != "" {
|
||||
m.DohProfileID = doh
|
||||
}
|
||||
if dc != nil && *dc != "" {
|
||||
m.DefaultCommunityID = dc
|
||||
}
|
||||
if last != nil {
|
||||
t := last.UTC()
|
||||
m.LastRefreshedAt = &t
|
||||
}
|
||||
out = append(out, &m)
|
||||
moduleByID[m.ID] = &m
|
||||
}
|
||||
if err := p.batchFillModuleDohFields(ctx, moduleByID); err != nil {
|
||||
return nil, "", false
|
||||
}
|
||||
more := len(out) > limit
|
||||
if more {
|
||||
out = out[:limit]
|
||||
}
|
||||
next := ""
|
||||
if more {
|
||||
next = fmt.Sprintf("%d", off+limit)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil, "", false
|
||||
}
|
||||
return out, next, more
|
||||
}
|
||||
|
||||
func (p *Postgres) GetModule(tenantID, moduleID string) (*store.Module, error) {
|
||||
ctx := context.Background()
|
||||
var m store.Module
|
||||
@@ -685,7 +756,13 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
|
||||
}
|
||||
}
|
||||
ctx := context.Background()
|
||||
if _, err := p.GetRevision(tenantID, revisionID); err != nil {
|
||||
var one int
|
||||
if err := p.pool.QueryRow(ctx, `
|
||||
SELECT 1 FROM config_revision WHERE id = $1::uuid AND tenant_id = $2::uuid`,
|
||||
revisionID, tenantID).Scan(&one); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, "", false
|
||||
}
|
||||
return nil, "", false
|
||||
}
|
||||
rows, err := p.pool.Query(ctx, `
|
||||
@@ -754,6 +831,8 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st
|
||||
return newID, nil
|
||||
}
|
||||
|
||||
const maxRevisionDiffRows = 5000
|
||||
|
||||
func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, error) {
|
||||
if _, err := p.GetRevision(tenantID, aID); err != nil {
|
||||
return nil, err
|
||||
@@ -778,7 +857,7 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
||||
EXCEPT
|
||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
||||
) s ORDER BY 1`, bID, aID)
|
||||
) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -790,13 +869,18 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
||||
continue
|
||||
}
|
||||
added = append(added, s)
|
||||
if len(added) > maxRevisionDiffRows {
|
||||
added = added[:maxRevisionDiffRows]
|
||||
break
|
||||
}
|
||||
}
|
||||
addedTruncated := len(added) >= maxRevisionDiffRows
|
||||
rowsRem, err := p.pool.Query(ctx, `
|
||||
SELECT prefix::text FROM (
|
||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
|
||||
EXCEPT
|
||||
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
|
||||
) s ORDER BY 1`, aID, bID)
|
||||
) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -808,40 +892,56 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
|
||||
continue
|
||||
}
|
||||
removed = append(removed, s)
|
||||
if len(removed) > maxRevisionDiffRows {
|
||||
removed = removed[:maxRevisionDiffRows]
|
||||
break
|
||||
}
|
||||
}
|
||||
return map[string]any{
|
||||
"revision_a": aID,
|
||||
"revision_b": bID,
|
||||
"prefixes": map[string]any{
|
||||
"added": added, "removed": removed, "unchanged_count": unchanged,
|
||||
"truncated": addedTruncated || len(removed) >= maxRevisionDiffRows,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *Postgres) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
|
||||
ctx := context.Background()
|
||||
cmd, err := p.pool.Exec(ctx, `
|
||||
DELETE FROM config_revision AS cr
|
||||
WHERE cr.tenant_id = $1
|
||||
AND cr.created_at < $2
|
||||
AND cr.id <> (
|
||||
SELECT id
|
||||
FROM config_revision
|
||||
WHERE tenant_id = $1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM bgp_speaker AS sp
|
||||
WHERE sp.tenant_id = $1
|
||||
AND (sp.last_applied_revision_id = cr.id OR sp.published_revision_id = cr.id)
|
||||
)`,
|
||||
tenantID, cutoff.UTC())
|
||||
if err != nil {
|
||||
return 0, err
|
||||
total := 0
|
||||
const batchSize = 50
|
||||
for {
|
||||
cmd, err := p.pool.Exec(ctx, `
|
||||
DELETE FROM config_revision AS cr
|
||||
WHERE cr.id IN (
|
||||
SELECT id FROM config_revision
|
||||
WHERE tenant_id = $1
|
||||
AND created_at < $2
|
||||
AND id <> (
|
||||
SELECT id FROM config_revision
|
||||
WHERE tenant_id = $1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM bgp_speaker AS sp
|
||||
WHERE sp.tenant_id = $1
|
||||
AND (sp.last_applied_revision_id = config_revision.id OR sp.published_revision_id = config_revision.id)
|
||||
)
|
||||
ORDER BY created_at ASC
|
||||
LIMIT $3
|
||||
)`, tenantID, cutoff.UTC(), batchSize)
|
||||
if err != nil {
|
||||
return total, err
|
||||
}
|
||||
n := int(cmd.RowsAffected())
|
||||
total += n
|
||||
if n < batchSize {
|
||||
break
|
||||
}
|
||||
}
|
||||
return int(cmd.RowsAffected()), nil
|
||||
return total, nil
|
||||
}
|
||||
|
||||
func (p *Postgres) SetLastAppliedRevision(tenantID, speakerID, revisionID string) error {
|
||||
|
||||
@@ -325,6 +325,38 @@ func (p *Postgres) UpdateASEntryResolveMeta(tenantID, moduleID, entryID string,
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Postgres) UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []store.ASEntryResolveMetaUpdate, resolvedAt time.Time) error {
|
||||
if len(updates) == 0 {
|
||||
return nil
|
||||
}
|
||||
if _, err := p.GetModule(tenantID, moduleID); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx := context.Background()
|
||||
batch := &pgx.Batch{}
|
||||
for _, u := range updates {
|
||||
var nameArg any
|
||||
sn := strings.TrimSpace(u.ASNName)
|
||||
if sn == "" {
|
||||
nameArg = nil
|
||||
} else {
|
||||
nameArg = sn
|
||||
}
|
||||
batch.Queue(`
|
||||
UPDATE module_as_entry SET asn_name=$3, prefix_count=$4, asn_resolved_at=$5, updated_at=now()
|
||||
WHERE id=$1 AND module_id=$2`,
|
||||
u.EntryID, moduleID, nameArg, u.PrefixCount, resolvedAt.UTC())
|
||||
}
|
||||
br := p.pool.SendBatch(ctx, batch)
|
||||
defer func() { _ = br.Close() }()
|
||||
for range updates {
|
||||
if _, err := br.Exec(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Postgres) DeleteASEntry(tenantID, moduleID, entryID string) error {
|
||||
if _, err := p.GetModule(tenantID, moduleID); err != nil {
|
||||
return err
|
||||
|
||||
@@ -10,28 +10,41 @@ func (p *Postgres) fillModuleDohFields(ctx context.Context, m *store.Module) err
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
return p.batchFillModuleDohFields(ctx, map[string]*store.Module{m.ID: m})
|
||||
}
|
||||
|
||||
func (p *Postgres) batchFillModuleDohFields(ctx context.Context, modules map[string]*store.Module) error {
|
||||
if len(modules) == 0 {
|
||||
return nil
|
||||
}
|
||||
ids := make([]string, 0, len(modules))
|
||||
for id := range modules {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
rows, err := p.pool.Query(ctx, `
|
||||
SELECT doh_profile_id::text
|
||||
SELECT module_id::text, doh_profile_id::text
|
||||
FROM module_doh_profile
|
||||
WHERE module_id = $1
|
||||
ORDER BY sort_order, doh_profile_id`, m.ID)
|
||||
WHERE module_id = ANY($1::uuid[])
|
||||
ORDER BY module_id, sort_order, doh_profile_id`, ids)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
var ids []string
|
||||
byModule := make(map[string][]string, len(modules))
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
var moduleID, profileID string
|
||||
if err := rows.Scan(&moduleID, &profileID); err != nil {
|
||||
return err
|
||||
}
|
||||
ids = append(ids, id)
|
||||
byModule[moduleID] = append(byModule[moduleID], profileID)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
m.DohProfileIDs = store.NormalizeDohProfileIDList(ids)
|
||||
m.SyncLegacyDohProfileID()
|
||||
for id, m := range modules {
|
||||
m.DohProfileIDs = store.NormalizeDohProfileIDList(byModule[id])
|
||||
m.SyncLegacyDohProfileID()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
package store
|
||||
|
||||
// ASEntryResolveMetaUpdate is one row for batch AS resolve metadata writes.
|
||||
type ASEntryResolveMetaUpdate struct {
|
||||
EntryID string
|
||||
ASNName string
|
||||
PrefixCount int64
|
||||
}
|
||||
@@ -18,6 +18,8 @@ type Backend interface {
|
||||
|
||||
// ListModules returns all modules for a tenant (control plane may paginate in httpapi).
|
||||
ListModules(tenantID string) []*Module
|
||||
// ListModulesPage returns one page of modules (limit capped by caller).
|
||||
ListModulesPage(tenantID, cursor string, limit int) ([]*Module, string, bool)
|
||||
GetModule(tenantID, moduleID string) (*Module, error)
|
||||
CreateModule(tenantID string, in *Module) (*Module, error)
|
||||
UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*Module, error)
|
||||
@@ -34,6 +36,7 @@ type Backend interface {
|
||||
DeleteASEntry(tenantID, moduleID, entryID string) error
|
||||
// UpdateASEntryResolveMeta записывает имя AS, число объявленных префиксов и время успешного резолва (pipeline).
|
||||
UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, asnName string, prefixCount int64, resolvedAt time.Time) error
|
||||
UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []ASEntryResolveMetaUpdate, resolvedAt time.Time) error
|
||||
|
||||
ListDomainEntries(tenantID, moduleID string) ([]*DomainEntry, error)
|
||||
CreateDomainEntry(tenantID, moduleID string, in *DomainEntry) (*DomainEntry, error)
|
||||
|
||||
@@ -382,6 +382,11 @@ func (m *Memory) ListModules(tenantID string) []*Module {
|
||||
return out
|
||||
}
|
||||
|
||||
func (m *Memory) ListModulesPage(tenantID, cursor string, limit int) ([]*Module, string, bool) {
|
||||
all := m.ListModules(tenantID)
|
||||
return PaginateOffset(all, cursor, limit)
|
||||
}
|
||||
|
||||
// ListPeers returns BGP peers for a tenant (sorted by name).
|
||||
func (m *Memory) ListPeers(tenantID string) []*BGPPeer {
|
||||
m.mu.RLock()
|
||||
|
||||
@@ -301,6 +301,15 @@ func (m *Memory) UpdateASEntryResolveMeta(tenantID, moduleID, entryID string, as
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Memory) UpdateASEntryResolveMetaBatch(tenantID, moduleID string, updates []ASEntryResolveMetaUpdate, resolvedAt time.Time) error {
|
||||
for _, u := range updates {
|
||||
if err := m.UpdateASEntryResolveMeta(tenantID, moduleID, u.EntryID, u.ASNName, u.PrefixCount, resolvedAt); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Memory) DeleteASEntry(tenantID, moduleID, entryID string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_id;
|
||||
DROP INDEX IF EXISTS idx_config_revision_tenant_module_created;
|
||||
DROP INDEX IF EXISTS idx_bgp_speaker_published;
|
||||
DROP INDEX IF EXISTS idx_bgp_speaker_last_applied;
|
||||
DROP INDEX IF EXISTS idx_module_default_community;
|
||||
DROP INDEX IF EXISTS idx_module_doh_profile_id;
|
||||
@@ -0,0 +1,17 @@
|
||||
CREATE INDEX IF NOT EXISTS idx_module_doh_profile_id
|
||||
ON module (doh_profile_id) WHERE deleted_at IS NULL AND doh_profile_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_module_default_community
|
||||
ON module (default_community_id) WHERE deleted_at IS NULL AND default_community_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_bgp_speaker_last_applied
|
||||
ON bgp_speaker (last_applied_revision_id) WHERE last_applied_revision_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_bgp_speaker_published
|
||||
ON bgp_speaker (published_revision_id) WHERE published_revision_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_config_revision_tenant_module_created
|
||||
ON config_revision (tenant_id, module_id, created_at DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_id
|
||||
ON revision_materialized_prefix (revision_id, id);
|
||||
@@ -0,0 +1,6 @@
|
||||
DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_id;
|
||||
DROP INDEX IF EXISTS idx_config_revision_tenant_module_created;
|
||||
DROP INDEX IF EXISTS idx_bgp_speaker_published;
|
||||
DROP INDEX IF EXISTS idx_bgp_speaker_last_applied;
|
||||
DROP INDEX IF EXISTS idx_module_default_community;
|
||||
DROP INDEX IF EXISTS idx_module_doh_profile_id;
|
||||
@@ -0,0 +1,17 @@
|
||||
CREATE INDEX IF NOT EXISTS idx_module_doh_profile_id
|
||||
ON module (doh_profile_id) WHERE deleted_at IS NULL AND doh_profile_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_module_default_community
|
||||
ON module (default_community_id) WHERE deleted_at IS NULL AND default_community_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_bgp_speaker_last_applied
|
||||
ON bgp_speaker (last_applied_revision_id) WHERE last_applied_revision_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_bgp_speaker_published
|
||||
ON bgp_speaker (published_revision_id) WHERE published_revision_id IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_config_revision_tenant_module_created
|
||||
ON config_revision (tenant_id, module_id, created_at DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_id
|
||||
ON revision_materialized_prefix (revision_id, id);
|
||||
@@ -0,0 +1,38 @@
|
||||
<script lang="ts">
|
||||
type Props = {
|
||||
items: readonly string[];
|
||||
rowHeight?: number;
|
||||
viewportHeight?: number;
|
||||
class?: string;
|
||||
};
|
||||
|
||||
let { items, rowHeight = 20, viewportHeight = 320, class: className = '' }: Props = $props();
|
||||
|
||||
let scrollTop = $state(0);
|
||||
const totalHeight = $derived(items.length * rowHeight);
|
||||
const startIndex = $derived(Math.max(0, Math.floor(scrollTop / rowHeight) - 2));
|
||||
const visibleCount = $derived(Math.ceil(viewportHeight / rowHeight) + 4);
|
||||
const visibleItems = $derived(items.slice(startIndex, startIndex + visibleCount));
|
||||
</script>
|
||||
|
||||
<div
|
||||
class="overflow-auto rounded-md border {className}"
|
||||
style="height: {viewportHeight}px"
|
||||
onscroll={(e) => {
|
||||
scrollTop = e.currentTarget.scrollTop;
|
||||
}}
|
||||
>
|
||||
<div style="height: {totalHeight}px; position: relative">
|
||||
{#each visibleItems as pfx, i (`${startIndex + i}-${pfx}`)}
|
||||
<p
|
||||
class="truncate px-2 font-mono text-xs leading-5"
|
||||
style="position: absolute; top: {(startIndex + i) *
|
||||
rowHeight}px; left: 0; right: 0; height: {rowHeight}px"
|
||||
>
|
||||
{pfx}
|
||||
</p>
|
||||
{:else}
|
||||
<p class="text-muted-foreground p-2 text-sm">Нет префиксов</p>
|
||||
{/each}
|
||||
</div>
|
||||
</div>
|
||||
@@ -185,8 +185,8 @@
|
||||
apiJSON<ModulesResponse>('/v1/modules?limit=200'),
|
||||
apiJSON<PeersResponse>('/v1/peers?limit=200'),
|
||||
apiJSON<SpeakersResponse>('/v1/speakers?limit=200'),
|
||||
apiJSON<RevisionsResponse>('/v1/revisions?limit=200'),
|
||||
apiJSON<JobsResponse>('/v1/jobs?limit=20')
|
||||
apiJSON<RevisionsResponse>('/v1/revisions?limit=10'),
|
||||
apiJSON<JobsResponse>('/v1/jobs?limit=10')
|
||||
]);
|
||||
|
||||
const firstReject = [m, p, s, r, j].find((x) => x.status === 'rejected');
|
||||
|
||||
@@ -60,6 +60,7 @@
|
||||
ReportRow
|
||||
} from '$lib/components/operations/types.js';
|
||||
import ScrollPreBlock from '$lib/components/app/scroll-pre-block.svelte';
|
||||
import VirtualPrefixList from '$lib/ui/patterns/virtual-list/virtual-prefix-list.svelte';
|
||||
import CardSkeleton from '$lib/ui/patterns/feedback/card-skeleton.svelte';
|
||||
import KpiMetricsGrid from '$lib/ui/patterns/kpi/kpi-metrics-grid.svelte';
|
||||
import { confirm } from '$lib/ui/patterns/confirm/confirm-state.svelte.js';
|
||||
@@ -128,6 +129,7 @@
|
||||
let jobs = $state<JobRow[]>([]);
|
||||
let jobsLoading = $state(false);
|
||||
let jobSearchQ = $state('');
|
||||
let jobSearchDebounced = $state('');
|
||||
let jobFilterStatus = $state('');
|
||||
let jobFilterKind = $state('');
|
||||
let jobFilterModule = $state('');
|
||||
@@ -302,13 +304,23 @@
|
||||
return j.kind === 'module_refresh' && mid === jobFilterModule;
|
||||
});
|
||||
}
|
||||
const q = jobSearchQ.trim();
|
||||
const q = jobSearchDebounced.trim();
|
||||
if (q) {
|
||||
list = list.filter((j) => jobMatchesSearch(j, q));
|
||||
}
|
||||
return list;
|
||||
});
|
||||
|
||||
let jobSearchTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
$effect(() => {
|
||||
const q = jobSearchQ;
|
||||
clearTimeout(jobSearchTimer);
|
||||
jobSearchTimer = setTimeout(() => {
|
||||
jobSearchDebounced = q;
|
||||
}, 250);
|
||||
return () => clearTimeout(jobSearchTimer);
|
||||
});
|
||||
|
||||
const jobModuleOptions = $derived(
|
||||
[...moduleNameById.entries()]
|
||||
.map(([id, name]) => ({ id, name }))
|
||||
@@ -340,8 +352,18 @@
|
||||
|
||||
onMount(() => {
|
||||
activeTab = parseOpsTab(page.url.searchParams.get('tab'));
|
||||
lastLoadedTab = activeTab;
|
||||
tabSyncReady = true;
|
||||
void refreshAll(true);
|
||||
void refreshActiveTab(true);
|
||||
});
|
||||
|
||||
let lastLoadedTab = $state('');
|
||||
$effect(() => {
|
||||
if (!tabSyncReady) return;
|
||||
const tab = activeTab;
|
||||
if (tab === lastLoadedTab) return;
|
||||
lastLoadedTab = tab;
|
||||
void refreshActiveTab();
|
||||
});
|
||||
|
||||
const statAccents = [
|
||||
@@ -425,6 +447,29 @@
|
||||
syncTabToUrl(activeTab);
|
||||
});
|
||||
|
||||
async function refreshActiveTab(isInitial = false) {
|
||||
if (isInitial) initialLoading = true;
|
||||
else refreshing = true;
|
||||
switch (activeTab) {
|
||||
case 'revisions':
|
||||
await loadRevisions();
|
||||
break;
|
||||
case 'jobs':
|
||||
await Promise.all([loadJobs(), loadModules()]);
|
||||
break;
|
||||
case 'diff':
|
||||
break;
|
||||
case 'system':
|
||||
await loadBirdStatus();
|
||||
break;
|
||||
default:
|
||||
await loadRevisions();
|
||||
}
|
||||
lastUpdated = new Date();
|
||||
initialLoading = false;
|
||||
refreshing = false;
|
||||
}
|
||||
|
||||
async function refreshAll(isInitial = false) {
|
||||
if (isInitial) initialLoading = true;
|
||||
else refreshing = true;
|
||||
@@ -1061,15 +1106,7 @@
|
||||
<span class="font-medium">Префиксов:</span>
|
||||
{prefixesData.length}
|
||||
</p>
|
||||
<div
|
||||
class="flex min-h-[10rem] min-w-0 flex-1 flex-col overflow-auto overscroll-contain rounded-lg border border-border bg-muted/40 p-3 [scrollbar-gutter:stable]"
|
||||
>
|
||||
{#each prefixesData as pfx, idx (`${idx}-${pfx}`)}
|
||||
<p class="font-mono text-xs">{pfx}</p>
|
||||
{:else}
|
||||
<p class="text-muted-foreground text-xs">Нет префиксов</p>
|
||||
{/each}
|
||||
</div>
|
||||
<VirtualPrefixList items={prefixesData} viewportHeight={360} />
|
||||
</TabsContent>
|
||||
</Tabs>
|
||||
</div>
|
||||
|
||||
Reference in New Issue
Block a user