perf: quick wins for pipeline, jobs, httpapi and web ops
- CDN snapshot: batch merge после parallel fetch, без race на persist - ListRevisionPrefixes: лёгкая проверка revision вместо GetRevision - Jobs: timeout/cancel context для pipeline и deploy - HTTP server timeouts; кэш birdc для GET /peers - ListModules: batch DoH profiles одним запросом - Web: tab-scoped load на /operations, debounce job search, меньше over-fetch на dashboard Co-authored-by: Cursor <[email protected]>
This commit is contained in:
@@ -55,8 +55,12 @@ func main() {
|
|||||||
startBirdMetricsPoller(ctx)
|
startBirdMetricsPoller(ctx)
|
||||||
|
|
||||||
httpSrv := &http.Server{
|
httpSrv := &http.Server{
|
||||||
Addr: cfg.HTTPAddr,
|
Addr: cfg.HTTPAddr,
|
||||||
Handler: srv.Handler(),
|
Handler: srv.Handler(),
|
||||||
|
ReadHeaderTimeout: 10 * time.Second,
|
||||||
|
ReadTimeout: 60 * time.Second,
|
||||||
|
WriteTimeout: 120 * time.Second,
|
||||||
|
IdleTimeout: 120 * time.Second,
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
svc := platform.ServiceName("evobgp-all")
|
svc := platform.ServiceName("evobgp-all")
|
||||||
@@ -91,6 +95,7 @@ func startBirdMetricsPoller(ctx context.Context) {
|
|||||||
return birdfmt.ShowProtocols(ctx, socket, birdcBin)
|
return birdfmt.ShowProtocols(ctx, socket, birdcBin)
|
||||||
},
|
},
|
||||||
birdfmt.CountEstablishedBGPSessions,
|
birdfmt.CountEstablishedBGPSessions,
|
||||||
|
birdfmt.ParseBGPProtocolStates,
|
||||||
)
|
)
|
||||||
log.Printf("birdc protocols poller enabled (socket=%s interval=%s)", sock, interval)
|
log.Printf("birdc protocols poller enabled (socket=%s interval=%s)", sock, interval)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -42,8 +42,12 @@ func main() {
|
|||||||
startBirdMetricsPoller(ctx)
|
startBirdMetricsPoller(ctx)
|
||||||
|
|
||||||
httpSrv := &http.Server{
|
httpSrv := &http.Server{
|
||||||
Addr: cfg.HTTPAddr,
|
Addr: cfg.HTTPAddr,
|
||||||
Handler: srv.Handler(),
|
Handler: srv.Handler(),
|
||||||
|
ReadHeaderTimeout: 10 * time.Second,
|
||||||
|
ReadTimeout: 60 * time.Second,
|
||||||
|
WriteTimeout: 120 * time.Second,
|
||||||
|
IdleTimeout: 120 * time.Second,
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
svc := platform.ServiceName("evobgp-api")
|
svc := platform.ServiceName("evobgp-api")
|
||||||
@@ -84,6 +88,7 @@ func startBirdMetricsPoller(ctx context.Context) {
|
|||||||
return birdfmt.ShowProtocols(ctx, socket, birdcBin)
|
return birdfmt.ShowProtocols(ctx, socket, birdcBin)
|
||||||
},
|
},
|
||||||
birdfmt.CountEstablishedBGPSessions,
|
birdfmt.CountEstablishedBGPSessions,
|
||||||
|
birdfmt.ParseBGPProtocolStates,
|
||||||
)
|
)
|
||||||
log.Printf("birdc protocols poller enabled (socket=%s interval=%s)", sock, interval)
|
log.Printf("birdc protocols poller enabled (socket=%s interval=%s)", sock, interval)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -50,3 +50,48 @@ func CountEstablishedBGPSessions(showProtocolsOutput string) int {
|
|||||||
}
|
}
|
||||||
return n
|
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 ""
|
||||||
|
}
|
||||||
|
|||||||
@@ -275,7 +275,7 @@ func (s *Server) handleListPeers(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
allPeers := s.store.ListPeers(a.TenantID)
|
allPeers := s.store.ListPeers(a.TenantID)
|
||||||
page, next, more := store.PaginateOffset(allPeers, r.URL.Query().Get("cursor"), parseListLimit(r))
|
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))
|
items := make([]map[string]any, 0, len(page))
|
||||||
for _, p := range page {
|
for _, p := range page {
|
||||||
row := peerJSON(p)
|
row := peerJSON(p)
|
||||||
@@ -289,7 +289,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"))
|
sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET"))
|
||||||
if sock == "" {
|
if sock == "" {
|
||||||
return map[string]string{}
|
return map[string]string{}
|
||||||
@@ -298,7 +308,9 @@ func (s *Server) liveBGPProtocolStates(ctx context.Context) map[string]string {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return map[string]string{}
|
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.
|
// parseBGPProtocolStates parses `birdc show protocols all` summary rows into protocol_name -> state.
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
+53
-8
@@ -102,7 +102,17 @@ func (w *Worker) Process(j *Job) {
|
|||||||
j.Fail("missing module_id in job meta")
|
j.Fail("missing module_id in job meta")
|
||||||
return
|
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())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -121,11 +131,17 @@ func (w *Worker) Process(j *Job) {
|
|||||||
j.Succeed()
|
j.Succeed()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
ctx, cancel := j.workContext()
|
||||||
|
defer cancel()
|
||||||
ctl := &birdfmt.BirdCtl{
|
ctl := &birdfmt.BirdCtl{
|
||||||
Socket: sock,
|
Socket: sock,
|
||||||
Birdc: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")),
|
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())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -152,9 +168,14 @@ func (w *Worker) runPeerReconcile(j *Job) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if len(latest) == 0 {
|
if len(latest) == 0 {
|
||||||
// First run fallback: render full tenant state once if no baseline revision exists yet.
|
ctx, cancel := j.workContext()
|
||||||
rid, err := pipeline.RenderTenantRevision(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID)
|
defer cancel()
|
||||||
|
rid, err := pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
j.MarkCancelled()
|
||||||
|
return
|
||||||
|
}
|
||||||
j.Fail(err.Error())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -171,8 +192,14 @@ func (w *Worker) runPeerReconcile(j *Job) {
|
|||||||
}
|
}
|
||||||
cursor = next
|
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 err != nil {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
j.MarkCancelled()
|
||||||
|
return
|
||||||
|
}
|
||||||
j.Fail(err.Error())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -230,7 +257,13 @@ func (w *Worker) runTenantRefresh(j *Job) {
|
|||||||
j.Fail("missing module_ids in job meta")
|
j.Fail("missing module_ids in job meta")
|
||||||
return
|
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())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -275,6 +308,8 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
|
|||||||
mu := w.tenantRefreshMu(j.TenantID)
|
mu := w.tenantRefreshMu(j.TenantID)
|
||||||
mu.Lock()
|
mu.Lock()
|
||||||
defer mu.Unlock()
|
defer mu.Unlock()
|
||||||
|
ctx, cancel := j.workContext()
|
||||||
|
defer cancel()
|
||||||
deferDeploy := false
|
deferDeploy := false
|
||||||
if w.Registry != nil {
|
if w.Registry != nil {
|
||||||
deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0
|
deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0
|
||||||
@@ -288,8 +323,12 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
|
|||||||
return
|
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 err != nil {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
j.MarkCancelled()
|
||||||
|
return
|
||||||
|
}
|
||||||
j.Fail(err.Error())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -335,6 +374,8 @@ func (w *Worker) runDeployApply(j *Job) {
|
|||||||
j.Fail("missing revision_id in job meta")
|
j.Fail("missing revision_id in job meta")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
ctx, cancel := j.workContext()
|
||||||
|
defer cancel()
|
||||||
activeDir := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR"))
|
activeDir := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR"))
|
||||||
if activeDir != "" {
|
if activeDir != "" {
|
||||||
revObj, err := w.Store.GetRevision(j.TenantID, revID)
|
revObj, err := w.Store.GetRevision(j.TenantID, revID)
|
||||||
@@ -354,7 +395,11 @@ func (w *Worker) runDeployApply(j *Job) {
|
|||||||
Socket: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")),
|
Socket: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")),
|
||||||
}
|
}
|
||||||
ctl := &birdfmt.BirdCtl{Bird: cfg.BirdBin, Birdc: cfg.BirdcBin, Socket: cfg.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())
|
j.Fail(err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -82,6 +82,12 @@ var (
|
|||||||
})
|
})
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
birdProtocolStatesMu sync.RWMutex
|
||||||
|
birdProtocolStates map[string]string
|
||||||
|
birdProtocolStatesAt time.Time
|
||||||
|
)
|
||||||
|
|
||||||
// RecordPrefixAggregation records tenant render CIDR aggregation stats.
|
// RecordPrefixAggregation records tenant render CIDR aggregation stats.
|
||||||
func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Duration) {
|
func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Duration) {
|
||||||
if rawCount < 0 {
|
if rawCount < 0 {
|
||||||
@@ -205,6 +211,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.
|
// MetricsHandler returns the Prometheus scrape handler.
|
||||||
func MetricsHandler() http.Handler {
|
func MetricsHandler() http.Handler {
|
||||||
return promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{})
|
return promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{})
|
||||||
@@ -231,7 +266,7 @@ func (s *statusRecorder) WriteHeader(code int) {
|
|||||||
|
|
||||||
// StartBirdProtocolsPoller runs birdc "show protocols" on interval when socket is non-empty.
|
// StartBirdProtocolsPoller runs birdc "show protocols" on interval when socket is non-empty.
|
||||||
// Горутина завершается при отмене ctx (корректное завершение вместе с процессом API).
|
// Горутина завершается при отмене 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)
|
socket = trimSpace(socket)
|
||||||
if ctx == nil || socket == "" || interval <= 0 || showFn == nil || countFn == nil {
|
if ctx == nil || socket == "" || interval <= 0 || showFn == nil || countFn == nil {
|
||||||
return
|
return
|
||||||
@@ -245,6 +280,9 @@ func StartBirdProtocolsPoller(ctx context.Context, socket string, birdcPath stri
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
SetBirdSessionMetrics(countFn(out), true)
|
SetBirdSessionMetrics(countFn(out), true)
|
||||||
|
if parseFn != nil {
|
||||||
|
SetBirdProtocolStates(parseFn(out))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
scrape()
|
scrape()
|
||||||
|
|||||||
@@ -45,6 +45,35 @@ func mergeSnapshotDropSource(rows []store.PrefixRow, sourceKey string) []store.P
|
|||||||
return out
|
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 {
|
func cdnRowsFromParsed(mod *store.Module, src *store.CDNSource, pfxStrings []string) []store.PrefixRow {
|
||||||
var rows []store.PrefixRow
|
var rows []store.PrefixRow
|
||||||
for _, p := range pfxStrings {
|
for _, p := range pfxStrings {
|
||||||
@@ -156,3 +185,68 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
|
|||||||
}
|
}
|
||||||
return rows, nil
|
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
|
||||||
|
}
|
||||||
|
|||||||
@@ -155,7 +155,7 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
|
|||||||
return
|
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 {
|
if err != nil {
|
||||||
results[idx] = srcResult{err: err}
|
results[idx] = srcResult{err: err}
|
||||||
return
|
return
|
||||||
@@ -172,6 +172,11 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
|
|||||||
}
|
}
|
||||||
out = append(out, r.rows...)
|
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
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -126,6 +126,7 @@ func (p *Postgres) ListModules(tenantID string) []*store.Module {
|
|||||||
}
|
}
|
||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
var out []*store.Module
|
var out []*store.Module
|
||||||
|
moduleByID := make(map[string]*store.Module)
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var m store.Module
|
var m store.Module
|
||||||
m.TenantID = tenantID
|
m.TenantID = tenantID
|
||||||
@@ -152,10 +153,11 @@ func (p *Postgres) ListModules(tenantID string) []*store.Module {
|
|||||||
t := last.UTC()
|
t := last.UTC()
|
||||||
m.LastRefreshedAt = &t
|
m.LastRefreshedAt = &t
|
||||||
}
|
}
|
||||||
if err := p.fillModuleDohFields(ctx, &m); err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
out = append(out, &m)
|
out = append(out, &m)
|
||||||
|
moduleByID[m.ID] = &m
|
||||||
|
}
|
||||||
|
if err := p.batchFillModuleDohFields(ctx, moduleByID); err != nil {
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
@@ -685,7 +687,13 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
ctx := context.Background()
|
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
|
return nil, "", false
|
||||||
}
|
}
|
||||||
rows, err := p.pool.Query(ctx, `
|
rows, err := p.pool.Query(ctx, `
|
||||||
|
|||||||
@@ -10,28 +10,41 @@ func (p *Postgres) fillModuleDohFields(ctx context.Context, m *store.Module) err
|
|||||||
if m == nil {
|
if m == nil {
|
||||||
return 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, `
|
rows, err := p.pool.Query(ctx, `
|
||||||
SELECT doh_profile_id::text
|
SELECT module_id::text, doh_profile_id::text
|
||||||
FROM module_doh_profile
|
FROM module_doh_profile
|
||||||
WHERE module_id = $1
|
WHERE module_id = ANY($1::uuid[])
|
||||||
ORDER BY sort_order, doh_profile_id`, m.ID)
|
ORDER BY module_id, sort_order, doh_profile_id`, ids)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
var ids []string
|
byModule := make(map[string][]string, len(modules))
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var id string
|
var moduleID, profileID string
|
||||||
if err := rows.Scan(&id); err != nil {
|
if err := rows.Scan(&moduleID, &profileID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
ids = append(ids, id)
|
byModule[moduleID] = append(byModule[moduleID], profileID)
|
||||||
}
|
}
|
||||||
if err := rows.Err(); err != nil {
|
if err := rows.Err(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
m.DohProfileIDs = store.NormalizeDohProfileIDList(ids)
|
for id, m := range modules {
|
||||||
m.SyncLegacyDohProfileID()
|
m.DohProfileIDs = store.NormalizeDohProfileIDList(byModule[id])
|
||||||
|
m.SyncLegacyDohProfileID()
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -185,8 +185,8 @@
|
|||||||
apiJSON<ModulesResponse>('/v1/modules?limit=200'),
|
apiJSON<ModulesResponse>('/v1/modules?limit=200'),
|
||||||
apiJSON<PeersResponse>('/v1/peers?limit=200'),
|
apiJSON<PeersResponse>('/v1/peers?limit=200'),
|
||||||
apiJSON<SpeakersResponse>('/v1/speakers?limit=200'),
|
apiJSON<SpeakersResponse>('/v1/speakers?limit=200'),
|
||||||
apiJSON<RevisionsResponse>('/v1/revisions?limit=200'),
|
apiJSON<RevisionsResponse>('/v1/revisions?limit=10'),
|
||||||
apiJSON<JobsResponse>('/v1/jobs?limit=20')
|
apiJSON<JobsResponse>('/v1/jobs?limit=10')
|
||||||
]);
|
]);
|
||||||
|
|
||||||
const firstReject = [m, p, s, r, j].find((x) => x.status === 'rejected');
|
const firstReject = [m, p, s, r, j].find((x) => x.status === 'rejected');
|
||||||
|
|||||||
@@ -128,6 +128,7 @@
|
|||||||
let jobs = $state<JobRow[]>([]);
|
let jobs = $state<JobRow[]>([]);
|
||||||
let jobsLoading = $state(false);
|
let jobsLoading = $state(false);
|
||||||
let jobSearchQ = $state('');
|
let jobSearchQ = $state('');
|
||||||
|
let jobSearchDebounced = $state('');
|
||||||
let jobFilterStatus = $state('');
|
let jobFilterStatus = $state('');
|
||||||
let jobFilterKind = $state('');
|
let jobFilterKind = $state('');
|
||||||
let jobFilterModule = $state('');
|
let jobFilterModule = $state('');
|
||||||
@@ -302,13 +303,23 @@
|
|||||||
return j.kind === 'module_refresh' && mid === jobFilterModule;
|
return j.kind === 'module_refresh' && mid === jobFilterModule;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
const q = jobSearchQ.trim();
|
const q = jobSearchDebounced.trim();
|
||||||
if (q) {
|
if (q) {
|
||||||
list = list.filter((j) => jobMatchesSearch(j, q));
|
list = list.filter((j) => jobMatchesSearch(j, q));
|
||||||
}
|
}
|
||||||
return list;
|
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(
|
const jobModuleOptions = $derived(
|
||||||
[...moduleNameById.entries()]
|
[...moduleNameById.entries()]
|
||||||
.map(([id, name]) => ({ id, name }))
|
.map(([id, name]) => ({ id, name }))
|
||||||
@@ -340,8 +351,18 @@
|
|||||||
|
|
||||||
onMount(() => {
|
onMount(() => {
|
||||||
activeTab = parseOpsTab(page.url.searchParams.get('tab'));
|
activeTab = parseOpsTab(page.url.searchParams.get('tab'));
|
||||||
|
lastLoadedTab = activeTab;
|
||||||
tabSyncReady = true;
|
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 = [
|
const statAccents = [
|
||||||
@@ -425,6 +446,29 @@
|
|||||||
syncTabToUrl(activeTab);
|
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) {
|
async function refreshAll(isInitial = false) {
|
||||||
if (isInitial) initialLoading = true;
|
if (isInitial) initialLoading = true;
|
||||||
else refreshing = true;
|
else refreshing = true;
|
||||||
|
|||||||
Reference in New Issue
Block a user