feat(api): enhance peer session management and documentation
CI / changes (push) Successful in 8s
CI / commitlint (push) Has been skipped
CI / openapi (push) Successful in 27s
CI / web (push) Failing after 35s
CI / go (push) Successful in 55s
CI / bird2 (push) Successful in 15s
CI / release (push) Has been skipped

- Added new fields to the API for tracking connected speakers and session states across multiple nodes, including `connected_speaker_id`, `connected_speaker_label`, `session_on_speakers`, `established_on_speakers`, and `session_mismatch`.
- Implemented a new endpoint for retrieving bird protocol sessions, enhancing the agent server functionality.
- Updated the OpenAPI documentation to reflect the new fields and query parameters, improving clarity for API consumers.
- Modified the frontend to display connected speaker information and session states, providing better visibility into peer connections.
- Updated deployment documentation to clarify the configuration requirements for enabling IP forwarding on VPS.
This commit is contained in:
Denozordec
2026-05-21 15:18:21 +07:00
parent e0a912a693
commit 1c0d78b552
16 changed files with 659 additions and 38 deletions
+26
View File
@@ -10,6 +10,7 @@ import (
"strings"
"time"
"evobgp/internal/birdfmt"
"evobgp/internal/nodecli"
)
@@ -41,6 +42,7 @@ type Server struct {
func New(cfg Config) *Server {
s := &Server{cfg: cfg, mux: http.NewServeMux()}
s.mux.HandleFunc("GET /v1/agent/health", s.handleHealth)
s.mux.HandleFunc("GET /v1/agent/bird/protocols", s.handleBirdProtocols)
s.mux.HandleFunc("POST /v1/agent/sync", s.handleSync)
return s
}
@@ -68,6 +70,30 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, body)
}
func (s *Server) handleBirdProtocols(w http.ResponseWriter, r *http.Request) {
if !s.authorize(r) {
writeProblem(w, http.StatusUnauthorized, "missing or invalid Authorization")
return
}
sock := strings.TrimSpace(s.cfg.Socket)
if sock == "" {
writeProblem(w, http.StatusServiceUnavailable, "EVOBGP_BIRDC_SOCKET not configured")
return
}
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
out, err := birdfmt.ShowProtocols(ctx, sock, strings.TrimSpace(s.cfg.BirdcBin))
if err != nil {
log.Printf("agentserver: bird protocols: %v", err)
writeProblem(w, http.StatusBadGateway, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{
"ok": true,
"sessions": birdfmt.ParseBGPSessions(out),
})
}
func (s *Server) handleSync(w http.ResponseWriter, r *http.Request) {
if !s.authorize(r) {
writeProblem(w, http.StatusUnauthorized, "missing or invalid Authorization")
+48
View File
@@ -0,0 +1,48 @@
package birdfmt
import (
"strings"
)
// BGPSession is one BGP protocol block from `birdc show protocols all`.
type BGPSession struct {
Name string `json:"name"`
Neighbor string `json:"neighbor,omitempty"`
State string `json:"state"`
}
// ParseBGPSessions extracts BGP protocol name, state, and neighbor (if present) from birdc output.
func ParseBGPSessions(output string) []BGPSession {
var out []BGPSession
var cur *BGPSession
for _, raw := range strings.Split(output, "\n") {
line := strings.TrimRight(raw, "\r")
trim := strings.TrimSpace(line)
if trim == "" {
continue
}
low := strings.ToLower(trim)
if strings.HasPrefix(low, "bird ") || strings.HasPrefix(low, "name ") || strings.HasPrefix(low, "table ") {
continue
}
if !strings.HasPrefix(line, " ") && !strings.HasPrefix(line, "\t") && isBGPProtocolSummaryRow(trim) {
fields := strings.Fields(trim)
state := extractBGPSessionStateLine(trim)
if state == "" && len(fields) >= 4 {
state = fields[3]
}
cur = &BGPSession{Name: fields[0], State: state}
out = append(out, *cur)
cur = &out[len(out)-1]
continue
}
if cur == nil {
continue
}
const neighborPrefix = "Neighbor address:"
if idx := strings.Index(trim, neighborPrefix); idx >= 0 {
cur.Neighbor = strings.TrimSpace(trim[idx+len(neighborPrefix):])
}
}
return out
}
+26
View File
@@ -0,0 +1,26 @@
package birdfmt
import "testing"
func TestParseBGPSessions_neighborAndState(t *testing.T) {
sample := `
BIRD 2.14 ready.
Name Proto Table State Since Info
device1 Device --- up 10:00:00
evobgp_p_abc123 BGP master4 up 10:00:05 Established
Neighbor address: 198.51.100.2
Neighbor AS: 65001
evobgp_p_def456 BGP master4 up 10:00:06 Active
Neighbor address: 2001:db8::2
`
sessions := ParseBGPSessions(sample)
if len(sessions) != 2 {
t.Fatalf("got %d sessions want 2", len(sessions))
}
if sessions[0].Name != "evobgp_p_abc123" || sessions[0].State != "Established" || sessions[0].Neighbor != "198.51.100.2" {
t.Fatalf("session0: %+v", sessions[0])
}
if sessions[1].Neighbor != "2001:db8::2" || sessions[1].State != "Active" {
t.Fatalf("session1: %+v", sessions[1])
}
}
+16
View File
@@ -0,0 +1,16 @@
package birdfmt
import "strings"
// PeerProtocolName returns the BIRD protocol name for a control-plane peer UUID.
// Must stay in sync with pipeline peer rendering.
func PeerProtocolName(peerID string) string {
s := strings.ReplaceAll(strings.TrimSpace(peerID), "-", "")
if len(s) > 16 {
s = s[:16]
}
if s == "" {
s = "x"
}
return "evobgp_p_" + s
}
+263
View File
@@ -0,0 +1,263 @@
package httpapi
import (
"context"
"net/netip"
"os"
"strings"
"sync"
"time"
"evobgp/internal/birdfmt"
"evobgp/internal/nodedispatch"
"evobgp/internal/store"
)
const peerLiveCacheTTL = 15 * time.Second
type speakerBGPLive struct {
SpeakerID string
Label string
Sessions []birdfmt.BGPSession
Error string
}
type peerLiveCacheEntry struct {
at time.Time
views []speakerBGPLive
}
var peerLiveCache sync.Map // tenantID -> peerLiveCacheEntry
type peerSessionOnSpeaker struct {
SpeakerID string `json:"speaker_id"`
Label string `json:"label"`
State string `json:"state"`
}
func speakerDisplayLabel(sp *store.Speaker) string {
if sp == nil {
return ""
}
meta := store.ParseSpeakerMeta(sp.MetaJSON)
host := strings.TrimSpace(meta.AgentDomain)
if host == "" {
host = strings.TrimSpace(sp.Endpoint)
}
if strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
if host != "" {
return "CP · " + host
}
return "CP (master)"
}
if host != "" {
return host
}
return sp.ID
}
func masterSpeakerID(speakers []*store.Speaker) string {
for _, sp := range speakers {
if sp != nil && strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
return sp.ID
}
}
return ""
}
func (s *Server) collectSpeakerBGPLive(ctx context.Context, tenantID string, fresh bool) []speakerBGPLive {
if !fresh {
if v, ok := peerLiveCache.Load(tenantID); ok {
ent := v.(peerLiveCacheEntry)
if time.Since(ent.at) < peerLiveCacheTTL {
return ent.views
}
}
}
speakers := s.store.ListSpeakersForTenant(tenantID)
views := make([]speakerBGPLive, 0, len(speakers)+1)
if sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")); sock != "" {
v := speakerBGPLive{Label: "CP (local BIRD)"}
if mid := masterSpeakerID(speakers); mid != "" {
v.SpeakerID = mid
for _, sp := range speakers {
if sp != nil && sp.ID == mid {
v.Label = speakerDisplayLabel(sp)
break
}
}
}
out, err := birdfmt.ShowProtocols(ctx, sock, strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")))
if err != nil {
v.Error = err.Error()
} else {
v.Sessions = birdfmt.ParseBGPSessions(out)
}
views = append(views, v)
}
opts := nodedispatch.Options{Timeout: 8 * time.Second}
type resWrap struct {
sp *store.Speaker
res nodedispatch.BirdProtocolsResult
}
ch := make(chan resWrap, 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()
ch <- resWrap{
sp: speaker,
res: nodedispatch.FetchBirdProtocols(ctx, speaker, opts),
}
}(sp)
}
wg.Wait()
close(ch)
for rw := range ch {
views = append(views, speakerBGPLive{
SpeakerID: rw.sp.ID,
Label: speakerDisplayLabel(rw.sp),
Sessions: rw.res.Sessions,
Error: rw.res.Error,
})
}
peerLiveCache.Store(tenantID, peerLiveCacheEntry{at: time.Now(), views: views})
return views
}
func matchPeerOnSpeakers(peer *store.BGPPeer, views []speakerBGPLive) (
bestState string,
connectedID string,
connectedLabel string,
establishedOn []peerSessionOnSpeaker,
on []peerSessionOnSpeaker,
mismatch bool,
) {
if peer == nil {
return "", "", "", nil, nil, false
}
neighbor, hasNeighbor := store.ParsePeerNeighbor(peer.Neighbor)
protoName := birdfmt.PeerProtocolName(peer.ID)
for _, v := range views {
if v.Error != "" && len(v.Sessions) == 0 {
continue
}
for _, sess := range v.Sessions {
if !peerSessionMatches(sess, protoName, neighbor, hasNeighbor) {
continue
}
hit := peerSessionOnSpeaker{
SpeakerID: v.SpeakerID,
Label: v.Label,
State: sess.State,
}
on = append(on, hit)
if strings.EqualFold(strings.TrimSpace(sess.State), "Established") {
establishedOn = append(establishedOn, hit)
}
if bestState == "" || sessionStateRank(sess.State) > sessionStateRank(bestState) {
bestState = sess.State
}
}
}
if len(establishedOn) > 0 {
bestState = "Established"
labels := make([]string, 0, len(establishedOn))
for _, e := range establishedOn {
labels = append(labels, e.Label)
}
connectedLabel = strings.Join(labels, ", ")
if len(establishedOn) == 1 {
connectedID = establishedOn[0].SpeakerID
}
} else if len(on) == 1 {
connectedID = on[0].SpeakerID
connectedLabel = on[0].Label
}
if peer.SpeakerID != nil && strings.TrimSpace(*peer.SpeakerID) != "" && len(establishedOn) > 0 {
want := strings.TrimSpace(*peer.SpeakerID)
found := false
for _, e := range establishedOn {
if strings.EqualFold(strings.TrimSpace(e.SpeakerID), want) {
found = true
break
}
}
mismatch = !found
}
return bestState, connectedID, connectedLabel, establishedOn, on, mismatch
}
func peerSessionMatches(sess birdfmt.BGPSession, protoName string, neighbor netip.Addr, hasNeighbor bool) bool {
if strings.EqualFold(strings.TrimSpace(sess.Name), protoName) {
return true
}
if !hasNeighbor || strings.TrimSpace(sess.Neighbor) == "" {
return false
}
addr, err := netip.ParseAddr(strings.TrimSpace(sess.Neighbor))
if err != nil {
return false
}
return addr == neighbor
}
func sessionStateRank(state string) int {
switch strings.ToLower(strings.TrimSpace(state)) {
case "established":
return 100
case "openconfirm", "opensent":
return 80
case "active", "connect":
return 60
case "idle":
return 20
default:
return 10
}
}
func applyPeerLiveFields(row map[string]any, peer *store.BGPPeer, views []speakerBGPLive) {
state, connID, connLabel, establishedOn, on, mismatch := matchPeerOnSpeakers(peer, views)
if len(on) > 0 {
if state != "" {
row["session_state"] = state
}
row["connected_speaker_id"] = peerLiveSpeakerIDOrNull(connID)
row["connected_speaker_label"] = connLabel
row["session_on_speakers"] = on
if len(establishedOn) > 0 {
row["established_on_speakers"] = establishedOn
} else {
row["established_on_speakers"] = []peerSessionOnSpeaker{}
}
row["session_conflict"] = false
row["session_mismatch"] = mismatch
return
}
row["session_on_speakers"] = []peerSessionOnSpeaker{}
row["established_on_speakers"] = []peerSessionOnSpeaker{}
row["session_conflict"] = false
row["session_mismatch"] = false
}
func peerLiveSpeakerIDOrNull(id string) any {
if strings.TrimSpace(id) == "" {
return nil
}
return id
}
+67
View File
@@ -0,0 +1,67 @@
package httpapi
import (
"testing"
"evobgp/internal/birdfmt"
"evobgp/internal/store"
)
func TestMatchPeerOnSpeakers_establishedOnReplica(t *testing.T) {
peer := &store.BGPPeer{
ID: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
Neighbor: "198.51.100.2",
}
views := []speakerBGPLive{
{
SpeakerID: "master-id",
Label: "CP · bgp.shz.su",
Sessions: []birdfmt.BGPSession{
{Name: birdfmt.PeerProtocolName(peer.ID), Neighbor: "198.51.100.2", State: "Idle"},
},
},
{
SpeakerID: "replica-id",
Label: "bgp2.shz.su",
Sessions: []birdfmt.BGPSession{
{Name: birdfmt.PeerProtocolName(peer.ID), Neighbor: "198.51.100.2", State: "Established"},
},
},
}
state, connID, connLabel, established, on, mismatch := matchPeerOnSpeakers(peer, views)
if state != "Established" || connID != "replica-id" || connLabel != "bgp2.shz.su" {
t.Fatalf("got state=%q conn=%q label=%q", state, connID, connLabel)
}
if mismatch || len(on) != 2 || len(established) != 1 {
t.Fatalf("on=%+v established=%+v mismatch=%v", on, established, mismatch)
}
}
func TestMatchPeerOnSpeakers_multipleEstablished(t *testing.T) {
peer := &store.BGPPeer{ID: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", Neighbor: "198.51.100.2"}
views := []speakerBGPLive{
{SpeakerID: "a", Label: "n1", Sessions: []birdfmt.BGPSession{{Name: birdfmt.PeerProtocolName(peer.ID), State: "Established"}}},
{SpeakerID: "b", Label: "n2", Sessions: []birdfmt.BGPSession{{Name: birdfmt.PeerProtocolName(peer.ID), State: "Established"}}},
}
_, connID, label, established, _, mismatch := matchPeerOnSpeakers(peer, views)
if mismatch || connID != "" || label != "n1, n2" || len(established) != 2 {
t.Fatalf("connID=%q label=%q established=%+v mismatch=%v", connID, label, established, mismatch)
}
}
func TestMatchPeerOnSpeakers_mismatchConfiguredSpeaker(t *testing.T) {
replica := "replica-id"
peer := &store.BGPPeer{
ID: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
Neighbor: "198.51.100.2",
SpeakerID: &replica,
}
views := []speakerBGPLive{
{SpeakerID: "master-id", Label: "CP", Sessions: []birdfmt.BGPSession{{Name: birdfmt.PeerProtocolName(peer.ID), State: "Established"}}},
{SpeakerID: replica, Label: "bgp2", Sessions: []birdfmt.BGPSession{{Name: birdfmt.PeerProtocolName(peer.ID), State: "Idle"}}},
}
_, _, _, _, _, mismatch := matchPeerOnSpeakers(peer, views)
if !mismatch {
t.Fatal("expected mismatch when configured replica has no Established")
}
}
+7 -13
View File
@@ -283,13 +283,14 @@ 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)
fresh := r != nil && strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("live")), "1")
ctx, cancel := context.WithTimeout(r.Context(), 12*time.Second)
defer cancel()
liveViews := s.collectSpeakerBGPLive(ctx, a.TenantID, fresh)
items := make([]map[string]any, 0, len(page))
for _, p := range page {
row := peerJSON(p)
if st, ok := liveStates[peerProtocolNameForID(p.ID)]; ok && strings.TrimSpace(st) != "" {
row["session_state"] = strings.TrimSpace(st)
}
applyPeerLiveFields(row, p, liveViews)
items = append(items, row)
}
writeJSON(w, http.StatusOK, map[string]any{
@@ -366,16 +367,9 @@ func extractBGPSessionState(line string) string {
return ""
}
// peerProtocolNameForID must stay in sync with pipeline peer protocol naming.
// peerProtocolNameForID forwards to birdfmt for tests and legacy callers.
func peerProtocolNameForID(peerID string) string {
s := strings.ReplaceAll(strings.TrimSpace(peerID), "-", "")
if len(s) > 16 {
s = s[:16]
}
if s == "" {
s = "x"
}
return "evobgp_p_" + s
return birdfmt.PeerProtocolName(peerID)
}
func (s *Server) handleListSpeakers(w http.ResponseWriter, r *http.Request) {
+69
View File
@@ -0,0 +1,69 @@
package nodedispatch
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"evobgp/internal/birdfmt"
"evobgp/internal/store"
)
// BirdProtocolsResult is agent birdc scrape outcome.
type BirdProtocolsResult struct {
SpeakerID string `json:"speaker_id,omitempty"`
Sessions []birdfmt.BGPSession `json:"sessions"`
Error string `json:"error,omitempty"`
}
// FetchBirdProtocols GETs /v1/agent/bird/protocols on a replica agent.
func FetchBirdProtocols(ctx context.Context, sp *store.Speaker, opts Options) BirdProtocolsResult {
res := BirdProtocolsResult{}
if sp != nil {
res.SpeakerID = sp.ID
}
if sp == nil {
res.Error = "nil speaker"
return res
}
meta := store.ParseSpeakerMeta(sp.MetaJSON)
url := store.AgentBirdProtocolsURL(meta)
if url == "" {
res.Error = "agent_domain not configured"
return res
}
secret := strings.TrimSpace(meta.AgentSecret)
if secret == "" {
res.Error = "agent_secret missing"
return res
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
res.Error = err.Error()
return res
}
req.Header.Set("Authorization", "Bearer "+secret)
resp, err := opts.client().Do(req)
if err != nil {
res.Error = err.Error()
return res
}
defer func() { _ = resp.Body.Close() }()
b, _ := io.ReadAll(resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
res.Error = fmt.Sprintf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
return res
}
var out struct {
Sessions []birdfmt.BGPSession `json:"sessions"`
}
if err := json.Unmarshal(b, &out); err != nil {
res.Error = err.Error()
return res
}
res.Sessions = out.Sessions
return res
}
+1 -12
View File
@@ -1029,7 +1029,7 @@ func renderPeersBirdFragment(st store.Backend, tenantID string, loc birdLocals)
}
pol := parsePeerPolicies(p.PoliciesJSON)
lv4, lv6, asn := effectivePeerLocals(loc, pol)
proto := peerProtocolName(p.ID)
proto := birdfmt.PeerProtocolName(p.ID)
ra := uint32(p.RemoteASN)
if addr.Is4() {
opts := birdfmt.BGPPeerFromTemplateOptions{
@@ -1083,17 +1083,6 @@ func parsePeerPolicies(raw string) peerPolicyJSON {
return pol
}
func peerProtocolName(peerID string) string {
s := strings.ReplaceAll(strings.TrimSpace(peerID), "-", "")
if len(s) > 16 {
s = s[:16]
}
if s == "" {
s = "x"
}
return "evobgp_p_" + s
}
// buildExpandedBirdText concatenates bird.conf and the contents of each standard include (for UI / preview).
func buildExpandedBirdText(main string, frags map[string]string) string {
var b strings.Builder
+10 -1
View File
@@ -124,11 +124,20 @@ func AgentSyncURL(meta SpeakerMeta) string {
// AgentHealthURL returns HTTPS health URL for agent_domain.
func AgentHealthURL(meta SpeakerMeta) string {
return agentHTTPSURL(meta, "/v1/agent/health")
}
// AgentBirdProtocolsURL returns HTTPS bird protocols URL for agent_domain.
func AgentBirdProtocolsURL(meta SpeakerMeta) string {
return agentHTTPSURL(meta, "/v1/agent/bird/protocols")
}
func agentHTTPSURL(meta SpeakerMeta, path string) string {
domain := strings.TrimSpace(meta.AgentDomain)
if domain == "" {
return ""
}
return "https://" + strings.TrimSuffix(domain, "/") + "/v1/agent/health"
return "https://" + strings.TrimSuffix(domain, "/") + path
}
// SpeakerNeedsRemoteDispatch reports whether deploy_apply should wake this speaker via agent HTTP.