feat(api): add live status tracking for speakers and BGP sessions
CI / changes (push) Successful in 9s
CI / commitlint (push) Has been skipped
CI / openapi (push) Successful in 24s
CI / web (push) Successful in 29s
CI / go (push) Successful in 43s
CI / bird2 (push) Successful in 15s
CI / release (push) Successful in 3m29s
CI / changes (push) Successful in 9s
CI / commitlint (push) Has been skipped
CI / openapi (push) Successful in 24s
CI / web (push) Successful in 29s
CI / go (push) Successful in 43s
CI / bird2 (push) Successful in 15s
CI / release (push) Successful in 3m29s
- Introduced new schemas for `SpeakerLiveStatus`, `BgpSessionLive`, and `LiveSpeakerPoll` in OpenAPI documentation to support live status queries. - Enhanced the `/v1/speakers` endpoint to include a `live` query parameter, allowing retrieval of real-time speaker and BGP status. - Updated the HTTP API to collect and return live status data for speakers, improving monitoring capabilities. - Modified frontend components to display live status information, enhancing user visibility into speaker health and BGP session states. - Added a new endpoint `/v1/bird/status` for retrieving the local BIRD status, further enriching the network monitoring features.
This commit is contained in:
1 parent
b5ed47902c
commit
a1ada06a76
17 files changed
+1493
-135
No files matched your search
@@ -385,9 +385,22 @@ func (s *Server) handleListSpeakers(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
speakers := s.store.ListSpeakersForTenant(a.TenantID)
|
||||
fresh := r != nil && strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("live")), "1")
|
||||
var liveByID map[string]map[string]any
|
||||
if fresh {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 12*time.Second)
|
||||
defer cancel()
|
||||
liveByID = s.collectSpeakerLiveStatus(ctx, a.TenantID, true, speakers)
|
||||
}
|
||||
items := make([]map[string]any, 0, len(speakers))
|
||||
for _, sp := range speakers {
|
||||
items = append(items, speakerJSONFromStore(s.store, sp))
|
||||
row := speakerJSONFromStore(s.store, sp)
|
||||
if liveByID != nil {
|
||||
if live, ok := liveByID[sp.ID]; ok {
|
||||
row["live"] = live
|
||||
}
|
||||
}
|
||||
items = append(items, row)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"items": items, "next_cursor": nil, "has_more": false,
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"evobgp/internal/birdfmt"
|
||||
"evobgp/internal/nodedispatch"
|
||||
"evobgp/internal/store"
|
||||
)
|
||||
|
||||
func countBGPSessions(sessions []birdfmt.BGPSession) (total, established int) {
|
||||
total = len(sessions)
|
||||
for _, s := range sessions {
|
||||
if strings.EqualFold(strings.TrimSpace(s.State), "Established") {
|
||||
established++
|
||||
}
|
||||
}
|
||||
return total, established
|
||||
}
|
||||
|
||||
func speakerLiveStatusJSON(sp *store.Speaker, view speakerBGPLive, health *nodedispatch.AgentHealthResult) map[string]any {
|
||||
total, established := countBGPSessions(view.Sessions)
|
||||
m := map[string]any{
|
||||
"label": view.Label,
|
||||
"bgp_poll_ok": view.Error == "",
|
||||
"bgp_sessions_total": total,
|
||||
"bgp_established": established,
|
||||
}
|
||||
if view.Error != "" {
|
||||
m["bgp_poll_error"] = view.Error
|
||||
}
|
||||
if health != nil {
|
||||
m["agent_ok"] = health.OK
|
||||
if health.Error != "" {
|
||||
m["agent_error"] = health.Error
|
||||
}
|
||||
if health.LastSyncAt != "" {
|
||||
m["agent_last_sync_at"] = health.LastSyncAt
|
||||
}
|
||||
if health.LastAppliedRevisionID != "" {
|
||||
m["agent_last_applied_revision_id"] = health.LastAppliedRevisionID
|
||||
}
|
||||
} else if sp != nil && strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
|
||||
m["agent_ok"] = view.Error == ""
|
||||
if view.Error != "" {
|
||||
m["agent_error"] = view.Error
|
||||
}
|
||||
} else if sp != nil && store.SpeakerNeedsRemoteDispatch(sp.Role, store.ParseSpeakerMeta(sp.MetaJSON)) {
|
||||
m["agent_ok"] = false
|
||||
m["agent_error"] = "agent health not polled"
|
||||
}
|
||||
if len(view.Sessions) > 0 {
|
||||
sess := make([]map[string]any, 0, len(view.Sessions))
|
||||
for _, s := range view.Sessions {
|
||||
row := map[string]any{
|
||||
"name": s.Name,
|
||||
"state": s.State,
|
||||
}
|
||||
if strings.TrimSpace(s.Neighbor) != "" {
|
||||
row["neighbor"] = s.Neighbor
|
||||
}
|
||||
sess = append(sess, row)
|
||||
}
|
||||
m["sessions"] = sess
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func (s *Server) collectSpeakerLiveStatus(ctx context.Context, tenantID string, fresh bool, speakers []*store.Speaker) map[string]map[string]any {
|
||||
views := s.collectSpeakerBGPLive(ctx, tenantID, fresh)
|
||||
viewByID := make(map[string]speakerBGPLive, len(views))
|
||||
for _, v := range views {
|
||||
if v.SpeakerID != "" {
|
||||
viewByID[v.SpeakerID] = v
|
||||
}
|
||||
}
|
||||
|
||||
opts := nodedispatch.Options{Timeout: 8 * time.Second}
|
||||
type healthWrap struct {
|
||||
id string
|
||||
h nodedispatch.AgentHealthResult
|
||||
}
|
||||
healthCh := make(chan healthWrap, len(speakers))
|
||||
var wg sync.WaitGroup
|
||||
for _, sp := range speakers {
|
||||
if sp == nil {
|
||||
continue
|
||||
}
|
||||
meta := store.ParseSpeakerMeta(sp.MetaJSON)
|
||||
if !store.SpeakerNeedsRemoteDispatch(sp.Role, meta) {
|
||||
continue
|
||||
}
|
||||
wg.Add(1)
|
||||
go func(speaker *store.Speaker) {
|
||||
defer wg.Done()
|
||||
healthCh <- healthWrap{
|
||||
id: speaker.ID,
|
||||
h: nodedispatch.FetchAgentHealth(ctx, speaker, opts),
|
||||
}
|
||||
}(sp)
|
||||
}
|
||||
wg.Wait()
|
||||
close(healthCh)
|
||||
healthByID := make(map[string]nodedispatch.AgentHealthResult, len(speakers))
|
||||
for hw := range healthCh {
|
||||
healthByID[hw.id] = hw.h
|
||||
}
|
||||
|
||||
out := make(map[string]map[string]any, len(speakers))
|
||||
for _, sp := range speakers {
|
||||
if sp == nil {
|
||||
continue
|
||||
}
|
||||
view, ok := viewByID[sp.ID]
|
||||
if !ok {
|
||||
view = speakerBGPLive{SpeakerID: sp.ID, Label: speakerDisplayLabel(sp)}
|
||||
}
|
||||
var hp *nodedispatch.AgentHealthResult
|
||||
if h, ok := healthByID[sp.ID]; ok {
|
||||
hCopy := h
|
||||
hp = &hCopy
|
||||
}
|
||||
out[sp.ID] = speakerLiveStatusJSON(sp, view, hp)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"evobgp/internal/birdfmt"
|
||||
"evobgp/internal/nodedispatch"
|
||||
"evobgp/internal/store"
|
||||
)
|
||||
|
||||
func TestCountBGPSessions(t *testing.T) {
|
||||
total, est := countBGPSessions([]birdfmt.BGPSession{
|
||||
{Name: "p1", State: "Established"},
|
||||
{Name: "p2", State: "Idle"},
|
||||
{Name: "p3", State: "established"},
|
||||
})
|
||||
if total != 3 || est != 2 {
|
||||
t.Fatalf("total=%d established=%d", total, est)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpeakerLiveStatusJSON_masterUsesBirdPoll(t *testing.T) {
|
||||
sp := &store.Speaker{ID: "m1", Role: "master", Endpoint: "https://cp.example"}
|
||||
view := speakerBGPLive{
|
||||
SpeakerID: "m1",
|
||||
Label: "CP · cp.example",
|
||||
Sessions: []birdfmt.BGPSession{
|
||||
{Name: "evobgp_peer_x", State: "Established"},
|
||||
},
|
||||
}
|
||||
m := speakerLiveStatusJSON(sp, view, nil)
|
||||
if m["agent_ok"] != true || m["bgp_established"] != 1 || m["bgp_sessions_total"] != 1 {
|
||||
t.Fatalf("got %#v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpeakerLiveStatusJSON_replicaWithHealth(t *testing.T) {
|
||||
sp := &store.Speaker{
|
||||
ID: "r1",
|
||||
Role: "replica",
|
||||
Endpoint: "https://node.example",
|
||||
MetaJSON: `{"agent_domain":"node.example","agent_secret":"s"}`,
|
||||
}
|
||||
view := speakerBGPLive{
|
||||
SpeakerID: "r1",
|
||||
Label: "node.example",
|
||||
Sessions: []birdfmt.BGPSession{{Name: "p", State: "Idle"}},
|
||||
}
|
||||
health := &nodedispatch.AgentHealthResult{
|
||||
OK: true,
|
||||
LastSyncAt: "2026-05-21T12:00:00Z",
|
||||
LastAppliedRevisionID: "rev-1",
|
||||
}
|
||||
m := speakerLiveStatusJSON(sp, view, health)
|
||||
if m["agent_ok"] != true || m["agent_last_sync_at"] != "2026-05-21T12:00:00Z" {
|
||||
t.Fatalf("got %#v", m)
|
||||
}
|
||||
if m["bgp_established"] != 0 || m["bgp_poll_ok"] != true {
|
||||
t.Fatalf("bgp fields: %#v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpeakerLiveStatusJSON_pollError(t *testing.T) {
|
||||
sp := &store.Speaker{ID: "r1", Role: "replica", MetaJSON: `{"agent_domain":"x.example"}`}
|
||||
view := speakerBGPLive{SpeakerID: "r1", Label: "x.example", Error: "HTTP 503"}
|
||||
health := &nodedispatch.AgentHealthResult{OK: false, Error: "timeout"}
|
||||
m := speakerLiveStatusJSON(sp, view, health)
|
||||
if m["bgp_poll_ok"] != false || m["bgp_poll_error"] != "HTTP 503" {
|
||||
t.Fatalf("got %#v", m)
|
||||
}
|
||||
if m["agent_ok"] != false {
|
||||
t.Fatalf("agent_ok: %#v", m)
|
||||
}
|
||||
}
|
||||
@@ -156,33 +156,14 @@ func WakeReplicas(ctx context.Context, st store.Backend, tenantID, revisionID st
|
||||
return out
|
||||
}
|
||||
|
||||
// CheckHealth GETs /v1/agent/health for UI Connected/Offline status.
|
||||
// CheckHealth is deprecated; use FetchAgentHealth.
|
||||
func CheckHealth(ctx context.Context, sp *store.Speaker, opts Options) (ok bool, detail string) {
|
||||
if sp == nil {
|
||||
return false, "nil speaker"
|
||||
}
|
||||
meta := store.ParseSpeakerMeta(sp.MetaJSON)
|
||||
url := store.AgentHealthURL(meta)
|
||||
if url == "" {
|
||||
return false, "agent_domain not configured"
|
||||
}
|
||||
secret := strings.TrimSpace(meta.AgentSecret)
|
||||
if secret == "" {
|
||||
return false, "agent_secret missing"
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return false, err.Error()
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+secret)
|
||||
resp, err := opts.client().Do(req)
|
||||
if err != nil {
|
||||
return false, err.Error()
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
res := FetchAgentHealth(ctx, sp, opts)
|
||||
if res.OK {
|
||||
return true, "connected"
|
||||
}
|
||||
b, _ := io.ReadAll(resp.Body)
|
||||
return false, fmt.Sprintf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
|
||||
if res.Error != "" {
|
||||
return false, res.Error
|
||||
}
|
||||
return false, "agent unhealthy"
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package nodedispatch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"evobgp/internal/store"
|
||||
)
|
||||
|
||||
// AgentHealthResult is the parsed outcome of GET /v1/agent/health on a replica.
|
||||
type AgentHealthResult struct {
|
||||
OK bool
|
||||
Error string
|
||||
LastAppliedRevisionID string
|
||||
LastSyncAt string
|
||||
}
|
||||
|
||||
// FetchAgentHealth GETs /v1/agent/health for UI Connected/Offline status.
|
||||
func FetchAgentHealth(ctx context.Context, sp *store.Speaker, opts Options) AgentHealthResult {
|
||||
if sp == nil {
|
||||
return AgentHealthResult{Error: "nil speaker"}
|
||||
}
|
||||
meta := store.ParseSpeakerMeta(sp.MetaJSON)
|
||||
url := store.AgentHealthURL(meta)
|
||||
if url == "" {
|
||||
return AgentHealthResult{Error: "agent_domain not configured"}
|
||||
}
|
||||
secret := strings.TrimSpace(meta.AgentSecret)
|
||||
if secret == "" {
|
||||
return AgentHealthResult{Error: "agent_secret missing"}
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return AgentHealthResult{Error: err.Error()}
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+secret)
|
||||
resp, err := opts.client().Do(req)
|
||||
if err != nil {
|
||||
return AgentHealthResult{Error: err.Error()}
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
b, _ := io.ReadAll(resp.Body)
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return AgentHealthResult{
|
||||
Error: fmt.Sprintf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b))),
|
||||
}
|
||||
}
|
||||
var out struct {
|
||||
OK bool `json:"ok"`
|
||||
LastAppliedRevisionID string `json:"last_applied_revision_id"`
|
||||
LastSyncAt string `json:"last_sync_at"`
|
||||
}
|
||||
if err := json.Unmarshal(b, &out); err != nil {
|
||||
return AgentHealthResult{Error: err.Error()}
|
||||
}
|
||||
res := AgentHealthResult{
|
||||
OK: out.OK,
|
||||
LastAppliedRevisionID: strings.TrimSpace(out.LastAppliedRevisionID),
|
||||
LastSyncAt: strings.TrimSpace(out.LastSyncAt),
|
||||
}
|
||||
if !res.OK {
|
||||
res.Error = "agent reported ok=false"
|
||||
}
|
||||
return res
|
||||
}
|
||||
Reference in new issue
Block a user