From f5d90636bbe8ad8e80cc7c8bfc7c8652d5c37696 Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Wed, 28 Jan 2026 21:50:48 -0800 Subject: [PATCH] Move protocol data onto nodes and simplify API response Co-Authored-By: Claude Opus 4.5 --- errors.go | 25 +++-- http.go | 271 +++++++++++++++++++++++++++++++++------------- sacn.go | 97 ----------------- static/index.html | 119 ++++++++++---------- types.go | 14 +++ 5 files changed, 285 insertions(+), 241 deletions(-) delete mode 100644 sacn.go diff --git a/errors.go b/errors.go index 72149ef..757de3b 100644 --- a/errors.go +++ b/errors.go @@ -239,26 +239,37 @@ func (e *ErrorTracker) clearAllErrorsLocked() bool { return had } -func (e *ErrorTracker) GetErrors() []*PortError { +func (e *ErrorTracker) GetErrors() []*Error { e.mu.RLock() defer e.mu.RUnlock() - errors := make([]*PortError, 0, len(e.errors)) + errors := make([]*Error, 0, len(e.errors)) for _, err := range e.errors { - errors = append(errors, err) + errors = append(errors, &Error{ + ID: err.ID, + NodeTypeID: err.NodeTypeID, + NodeName: err.NodeName, + Type: string(err.ErrorType), + Port: err.PortName, + InErrors: err.InErrors, + OutErrors: err.OutErrors, + InDelta: err.InDelta, + OutDelta: err.OutDelta, + Utilization: err.Utilization, + }) } return errors } -func (e *ErrorTracker) GetUnreachableNodes() []string { +func (e *ErrorTracker) GetUnreachableNodeSet() map[string]bool { e.mu.RLock() defer e.mu.RUnlock() - nodes := make([]string, 0, len(e.unreachableNodes)) + result := map[string]bool{} for nodeTypeID := range e.unreachableNodes { - nodes = append(nodes, nodeTypeID) + result[nodeTypeID] = true } - return nodes + return result } func (e *ErrorTracker) SetUnreachable(node *Node) bool { diff --git a/http.go b/http.go index 47b0699..e96322b 100644 --- a/http.go +++ b/http.go @@ -25,16 +25,24 @@ const ( ) type StatusResponse struct { - Config *Config `json:"config"` - Nodes []*Node `json:"nodes"` - Links []*Link `json:"links"` - MulticastGroups []*MulticastGroupMembers `json:"multicast_groups"` - ArtNetNodes []*ArtNetNode `json:"artnet_nodes"` - SACNNodes []*SACNNode `json:"sacn_nodes"` - DanteFlows []*DanteFlow `json:"dante_flows"` - PortErrors []*PortError `json:"port_errors"` - UnreachableNodes []string `json:"unreachable_nodes"` - BroadcastStats *BroadcastStatsResponse `json:"broadcast_stats,omitempty"` + Config *Config `json:"config"` + Nodes []*Node `json:"nodes"` + Links []*Link `json:"links"` + Errors []*Error `json:"errors,omitempty"` + BroadcastStats *BroadcastStatsResponse `json:"broadcast_stats,omitempty"` +} + +type Error struct { + ID string `json:"id"` + NodeTypeID string `json:"node_typeid"` + NodeName string `json:"node_name"` + Type string `json:"type"` + Port string `json:"port,omitempty"` + InErrors uint64 `json:"in_errors,omitempty"` + OutErrors uint64 `json:"out_errors,omitempty"` + InDelta uint64 `json:"in_delta,omitempty"` + OutDelta uint64 `json:"out_delta,omitempty"` + Utilization float64 `json:"utilization,omitempty"` } func (t *Tendrils) startHTTPServer() { @@ -135,16 +143,11 @@ func (t *Tendrils) GetStatus() *StatusResponse { config = &Config{} } return &StatusResponse{ - Config: config, - Nodes: t.getNodes(), - Links: t.getLinks(), - MulticastGroups: t.getMulticastGroups(), - ArtNetNodes: t.getArtNetNodes(), - SACNNodes: t.getSACNNodes(), - DanteFlows: t.getDanteFlows(), - PortErrors: t.errors.GetErrors(), - UnreachableNodes: t.errors.GetUnreachableNodes(), - BroadcastStats: broadcastStats, + Config: config, + Nodes: t.getNodes(), + Links: t.getLinks(), + Errors: t.errors.GetErrors(), + BroadcastStats: broadcastStats, } } @@ -220,13 +223,41 @@ func (t *Tendrils) handleAPIStatusStream(w http.ResponseWriter, r *http.Request) } func (t *Tendrils) getNodes() []*Node { + t.nodes.mu.Lock() + t.nodes.expireMulticastMemberships() + t.nodes.mu.Unlock() + + t.artnet.Expire() + t.sacnSources.Expire() + t.danteFlows.Expire() + t.nodes.mu.RLock() defer t.nodes.mu.RUnlock() + multicastByNode := t.buildMulticastByNode() + artnetByNode := t.buildArtNetByNode() + sacnByNode := t.buildSACNByNode() + danteTxByNode, danteRxByNode := t.buildDanteByNode() + unreachableNodes := t.errors.GetUnreachableNodeSet() + nodes := make([]*Node, 0, len(t.nodes.nodes)) for _, node := range t.nodes.nodes { - nodes = append(nodes, node) + nodeCopy := *node + nodeCopy.MulticastGroups = multicastByNode[node] + if artnet := artnetByNode[node]; artnet != nil { + nodeCopy.ArtNetInputs = artnet.Inputs + nodeCopy.ArtNetOutputs = artnet.Outputs + } + if sacn := sacnByNode[node]; sacn != nil { + nodeCopy.SACNInputs = sacn.Inputs + nodeCopy.SACNOutputs = sacn.Outputs + } + nodeCopy.DanteTx = danteTxByNode[node] + nodeCopy.DanteRx = danteRxByNode[node] + nodeCopy.Unreachable = unreachableNodes[node.TypeID] + nodes = append(nodes, &nodeCopy) } + sort.Slice(nodes, func(i, j int) bool { if nodes[i].DisplayName() != nodes[j].DisplayName() { return sortorder.NaturalLess(nodes[i].DisplayName(), nodes[j].DisplayName()) @@ -240,6 +271,151 @@ func (t *Tendrils) getNodes() []*Node { return nodes } +func (t *Tendrils) buildMulticastByNode() map[*Node][]string { + result := map[*Node][]string{} + for _, gm := range t.nodes.multicastGroups { + for _, membership := range gm.Members { + if membership.Node != nil { + result[membership.Node] = append(result[membership.Node], gm.Group.Name) + } + } + } + for node, groups := range result { + sort.Strings(groups) + result[node] = groups + } + return result +} + +type artnetNodeData struct { + Inputs []int + Outputs []int +} + +func (t *Tendrils) buildArtNetByNode() map[*Node]*artnetNodeData { + t.artnet.mu.RLock() + defer t.artnet.mu.RUnlock() + + result := map[*Node]*artnetNodeData{} + for _, an := range t.artnet.nodes { + inputs := make([]int, len(an.Inputs)) + for i, u := range an.Inputs { + inputs[i] = int(u) + } + outputs := make([]int, len(an.Outputs)) + for i, u := range an.Outputs { + outputs[i] = int(u) + } + sort.Ints(inputs) + sort.Ints(outputs) + result[an.Node] = &artnetNodeData{Inputs: inputs, Outputs: outputs} + } + return result +} + +type sacnNodeData struct { + Inputs []int + Outputs []int +} + +func (t *Tendrils) buildSACNByNode() map[*Node]*sacnNodeData { + result := map[*Node]*sacnNodeData{} + + for _, gm := range t.nodes.multicastGroups { + if len(gm.Group.Name) < 5 || gm.Group.Name[:5] != "sacn:" { + continue + } + var universe int + if _, err := fmt.Sscanf(gm.Group.Name, "sacn:%d", &universe); err != nil { + continue + } + for _, membership := range gm.Members { + if membership.Node == nil { + continue + } + data := result[membership.Node] + if data == nil { + data = &sacnNodeData{} + result[membership.Node] = data + } + if !containsInt(data.Inputs, universe) { + data.Inputs = append(data.Inputs, universe) + } + } + } + + t.sacnSources.mu.RLock() + for _, source := range t.sacnSources.sources { + if source.SrcIP == nil { + continue + } + node := t.nodes.getByIPLocked(source.SrcIP) + if node == nil { + continue + } + data := result[node] + if data == nil { + data = &sacnNodeData{} + result[node] = data + } + for _, u := range source.Universes { + if !containsInt(data.Outputs, u) { + data.Outputs = append(data.Outputs, u) + } + } + } + t.sacnSources.mu.RUnlock() + + for _, data := range result { + sort.Ints(data.Inputs) + sort.Ints(data.Outputs) + } + + return result +} + +func (t *Tendrils) buildDanteByNode() (map[*Node][]*DantePeer, map[*Node][]*DantePeer) { + t.danteFlows.mu.RLock() + defer t.danteFlows.mu.RUnlock() + + txByNode := map[*Node][]*DantePeer{} + rxByNode := map[*Node][]*DantePeer{} + + for source, flow := range t.danteFlows.flows { + for subNode, sub := range flow.Subscribers { + status := map[string]string{} + for ch, st := range sub.ChannelStatus { + status[ch] = st.String() + } + txByNode[source] = append(txByNode[source], &DantePeer{ + Node: subNode, + Channels: sub.Channels, + Status: status, + }) + rxByNode[subNode] = append(rxByNode[subNode], &DantePeer{ + Node: source, + Channels: sub.Channels, + Status: status, + }) + } + } + + for node, peers := range txByNode { + sort.Slice(peers, func(i, j int) bool { + return sortorder.NaturalLess(peers[i].Node.DisplayName(), peers[j].Node.DisplayName()) + }) + txByNode[node] = peers + } + for node, peers := range rxByNode { + sort.Slice(peers, func(i, j int) bool { + return sortorder.NaturalLess(peers[i].Node.DisplayName(), peers[j].Node.DisplayName()) + }) + rxByNode[node] = peers + } + + return txByNode, rxByNode +} + func (t *Tendrils) getLinks() []*Link { t.nodes.mu.RLock() defer t.nodes.mu.RUnlock() @@ -260,56 +436,3 @@ func (t *Tendrils) getLinks() []*Link { return links } - -func (t *Tendrils) getMulticastGroups() []*MulticastGroupMembers { - t.nodes.mu.Lock() - t.nodes.expireMulticastMemberships() - t.nodes.mu.Unlock() - - t.nodes.mu.RLock() - defer t.nodes.mu.RUnlock() - - groups := make([]*MulticastGroupMembers, 0, len(t.nodes.multicastGroups)) - for _, gm := range t.nodes.multicastGroups { - groups = append(groups, gm) - } - sort.Slice(groups, func(i, j int) bool { - return sortorder.NaturalLess(groups[i].Group.Name, groups[j].Group.Name) - }) - - return groups -} - -func (t *Tendrils) getArtNetNodes() []*ArtNetNode { - t.artnet.Expire() - - t.artnet.mu.RLock() - defer t.artnet.mu.RUnlock() - - nodes := make([]*ArtNetNode, 0, len(t.artnet.nodes)) - for _, node := range t.artnet.nodes { - nodes = append(nodes, node) - } - sort.Slice(nodes, func(i, j int) bool { - return sortorder.NaturalLess(nodes[i].Node.DisplayName(), nodes[j].Node.DisplayName()) - }) - - return nodes -} - -func (t *Tendrils) getDanteFlows() []*DanteFlow { - t.danteFlows.Expire() - - t.danteFlows.mu.RLock() - defer t.danteFlows.mu.RUnlock() - - flows := make([]*DanteFlow, 0, len(t.danteFlows.flows)) - for _, flow := range t.danteFlows.flows { - flows = append(flows, flow) - } - sort.Slice(flows, func(i, j int) bool { - return sortorder.NaturalLess(flows[i].Source.DisplayName(), flows[j].Source.DisplayName()) - }) - - return flows -} diff --git a/sacn.go b/sacn.go deleted file mode 100644 index bfedeb4..0000000 --- a/sacn.go +++ /dev/null @@ -1,97 +0,0 @@ -package tendrils - -import ( - "fmt" - "sort" - "strings" - - "github.com/fvbommel/sortorder" -) - -type SACNNode struct { - TypeID string `json:"typeid"` - Node *Node `json:"node"` - Inputs []int `json:"inputs,omitempty"` - Outputs []int `json:"outputs,omitempty"` -} - -func (t *Tendrils) getSACNNodes() []*SACNNode { - t.nodes.mu.Lock() - t.nodes.expireMulticastMemberships() - t.nodes.mu.Unlock() - - t.sacnSources.Expire() - - t.nodes.mu.RLock() - defer t.nodes.mu.RUnlock() - - nodeInputs := map[*Node][]int{} - nodeOutputs := map[*Node][]int{} - - for _, gm := range t.nodes.multicastGroups { - if !strings.HasPrefix(gm.Group.Name, "sacn:") { - continue - } - - var universe int - if _, err := fmt.Sscanf(gm.Group.Name, "sacn:%d", &universe); err != nil { - continue - } - - for _, membership := range gm.Members { - if membership.Node == nil { - continue - } - inputs := nodeInputs[membership.Node] - if !containsInt(inputs, universe) { - nodeInputs[membership.Node] = append(inputs, universe) - } - } - } - - t.sacnSources.mu.RLock() - for _, source := range t.sacnSources.sources { - if source.SrcIP == nil { - continue - } - node := t.nodes.getByIPLocked(source.SrcIP) - if node == nil { - continue - } - for _, u := range source.Universes { - outputs := nodeOutputs[node] - if !containsInt(outputs, u) { - nodeOutputs[node] = append(outputs, u) - } - } - } - t.sacnSources.mu.RUnlock() - - allNodes := map[*Node]bool{} - for node := range nodeInputs { - allNodes[node] = true - } - for node := range nodeOutputs { - allNodes[node] = true - } - - result := make([]*SACNNode, 0, len(allNodes)) - for node := range allNodes { - inputs := nodeInputs[node] - outputs := nodeOutputs[node] - sort.Ints(inputs) - sort.Ints(outputs) - result = append(result, &SACNNode{ - TypeID: newTypeID("sacnnode"), - Node: node, - Inputs: inputs, - Outputs: outputs, - }) - } - - sort.Slice(result, func(i, j int) bool { - return sortorder.NaturalLess(result[i].Node.DisplayName(), result[j].Node.DisplayName()) - }) - - return result -} diff --git a/static/index.html b/static/index.html index 96447d2..20afa75 100644 --- a/static/index.html +++ b/static/index.html @@ -1690,15 +1690,15 @@ nodeEl.addEventListener('click', () => scrollToNode(err.node_typeid)); item.appendChild(nodeEl); - if (err.error_type === 'unreachable') { + if (err.type === 'unreachable') { const typeEl = document.createElement('div'); typeEl.className = 'error-type'; typeEl.textContent = 'Unreachable'; item.appendChild(typeEl); - } else if (err.error_type === 'high_utilization') { + } else if (err.type === 'high_utilization') { const portEl = document.createElement('div'); portEl.className = 'error-port'; - portEl.textContent = 'Port: ' + err.port_name; + portEl.textContent = 'Port: ' + err.port; item.appendChild(portEl); const countsEl = document.createElement('div'); @@ -1713,7 +1713,7 @@ } else { const portEl = document.createElement('div'); portEl.className = 'error-port'; - portEl.textContent = 'Port: ' + err.port_name; + portEl.textContent = 'Port: ' + err.port; item.appendChild(portEl); const countsEl = document.createElement('div'); @@ -1723,7 +1723,7 @@ const typeEl = document.createElement('div'); typeEl.className = 'error-type'; - typeEl.textContent = err.error_type === 'startup' ? 'Present at startup' : 'New errors detected'; + typeEl.textContent = err.type === 'startup' ? 'Present at startup' : 'New errors detected'; item.appendChild(typeEl); } @@ -1806,9 +1806,9 @@ const nodes = data.nodes || []; const links = data.links || []; - portErrors = data.port_errors || []; - const unreachableNodeIds = new Set(data.unreachable_nodes || []); - const errorNodeIds = new Set(portErrors.filter(e => e.error_type !== 'unreachable').map(e => e.node_typeid)); + portErrors = data.errors || []; + const unreachableNodeIds = new Set(nodes.filter(n => n.unreachable).map(n => n.typeid)); + const errorNodeIds = new Set(portErrors.filter(e => e.type !== 'unreachable').map(e => e.node_typeid)); const locationTree = buildLocationTree(config.locations || [], null); @@ -1891,52 +1891,40 @@ } }); - const danteFlows = data.dante_flows || []; const danteNodes = new Map(); - danteFlows.forEach(flow => { - const sourceId = flow.source?.typeid; - if (!sourceId) return; + nodes.forEach(node => { + const nodeId = node.typeid; + const danteTx = node.dante_tx || []; + const danteRx = node.dante_rx || []; - if (!danteNodes.has(sourceId)) { - danteNodes.set(sourceId, { isTx: false, isRx: false, txTo: [], rxFrom: [] }); - } - const sourceInfo = danteNodes.get(sourceId); - sourceInfo.isTx = true; + if (danteTx.length === 0 && danteRx.length === 0) return; - (flow.subscribers || []).forEach(sub => { - const subId = sub.node?.typeid; - if (!subId) return; - - const subName = getShortLabel(sub.node); - const channels = sub.channels || []; + const txTo = danteTx.map(peer => { + const peerName = getShortLabel(peer.node); + const channels = peer.channels || []; const channelSummary = channels.length > 0 ? '\n ' + channels.join('\n ') : ''; - const txEntry = subName + channelSummary; - if (!sourceInfo.txTo.some(e => e.startsWith(subName))) { - sourceInfo.txTo.push(txEntry); - } + return peerName + channelSummary; + }); - if (!danteNodes.has(subId)) { - danteNodes.set(subId, { isTx: false, isRx: false, txTo: [], rxFrom: [] }); - } - const subInfo = danteNodes.get(subId); - subInfo.isRx = true; + const rxFrom = danteRx.map(peer => { + const peerName = getShortLabel(peer.node); + const channels = peer.channels || []; + const channelSummary = channels.length > 0 ? '\n ' + channels.join('\n ') : ''; + return peerName + channelSummary; + }); - const sourceName = getShortLabel(flow.source); - const rxChannelSummary = channels.length > 0 ? '\n ' + channels.join('\n ') : ''; - const rxEntry = sourceName + rxChannelSummary; - if (!subInfo.rxFrom.some(e => e.startsWith(sourceName))) { - subInfo.rxFrom.push(rxEntry); - } + txTo.sort((a, b) => a.split('\n')[0].localeCompare(b.split('\n')[0])); + rxFrom.sort((a, b) => a.split('\n')[0].localeCompare(b.split('\n')[0])); + + danteNodes.set(nodeId, { + isTx: danteTx.length > 0, + isRx: danteRx.length > 0, + txTo: txTo, + rxFrom: rxFrom }); }); - danteNodes.forEach(info => { - info.txTo.sort((a, b) => a.split('\n')[0].localeCompare(b.split('\n')[0])); - info.rxFrom.sort((a, b) => a.split('\n')[0].localeCompare(b.split('\n')[0])); - }); - - const artnetData = data.artnet_nodes || []; const artnetNodes = new Map(); const formatUniverse = (u) => { @@ -1949,13 +1937,13 @@ const universeInputs = new Map(); const universeOutputs = new Map(); - artnetData.forEach(an => { - const name = getShortLabel(an.node); - (an.inputs || []).forEach(u => { + nodes.forEach(node => { + const name = getShortLabel(node); + (node.artnet_inputs || []).forEach(u => { if (!universeInputs.has(u)) universeInputs.set(u, []); universeInputs.get(u).push(name); }); - (an.outputs || []).forEach(u => { + (node.artnet_outputs || []).forEach(u => { if (!universeOutputs.has(u)) universeOutputs.set(u, []); universeOutputs.get(u).push(name); }); @@ -1967,11 +1955,14 @@ return Object.entries(counts).map(([name, count]) => count > 1 ? name + ' x' + count : name); }; - artnetData.forEach(an => { - const nodeId = an.node?.typeid; - if (!nodeId) return; + nodes.forEach(node => { + const nodeId = node.typeid; + const artnetInputs = node.artnet_inputs || []; + const artnetOutputs = node.artnet_outputs || []; - const inputs = (an.inputs || []).slice().sort((a, b) => a - b).map(u => { + if (artnetInputs.length === 0 && artnetOutputs.length === 0) return; + + const inputs = artnetInputs.slice().sort((a, b) => a - b).map(u => { const sources = collapseNames(universeOutputs.get(u) || []); const uniStr = formatUniverse(u); if (sources.length > 0) { @@ -1979,7 +1970,7 @@ } return { display: uniStr, firstTarget: null }; }); - const outputs = (an.outputs || []).slice().sort((a, b) => a - b).map(u => { + const outputs = artnetOutputs.slice().sort((a, b) => a - b).map(u => { const dests = collapseNames(universeInputs.get(u) || []); const uniStr = formatUniverse(u); if (dests.length > 0) { @@ -1996,19 +1987,18 @@ }); }); - const sacnData = data.sacn_nodes || []; const sacnNodes = new Map(); const sacnUniverseInputs = new Map(); const sacnUniverseOutputs = new Map(); - sacnData.forEach(sn => { - const name = getShortLabel(sn.node); - (sn.inputs || []).forEach(u => { + nodes.forEach(node => { + const name = getShortLabel(node); + (node.sacn_inputs || []).forEach(u => { if (!sacnUniverseInputs.has(u)) sacnUniverseInputs.set(u, []); sacnUniverseInputs.get(u).push(name); }); - (sn.outputs || []).forEach(u => { + (node.sacn_outputs || []).forEach(u => { if (!sacnUniverseOutputs.has(u)) sacnUniverseOutputs.set(u, []); sacnUniverseOutputs.get(u).push(name); }); @@ -2020,18 +2010,21 @@ return Object.entries(counts).map(([name, count]) => count > 1 ? name + ' x' + count : name); }; - sacnData.forEach(sn => { - const nodeId = sn.node?.typeid; - if (!nodeId) return; + nodes.forEach(node => { + const nodeId = node.typeid; + const sacnInputs = node.sacn_inputs || []; + const sacnOutputs = node.sacn_outputs || []; - const inputs = (sn.inputs || []).slice().sort((a, b) => a - b).map(u => { + if (sacnInputs.length === 0 && sacnOutputs.length === 0) return; + + const inputs = sacnInputs.slice().sort((a, b) => a - b).map(u => { const sources = sacnCollapseNames(sacnUniverseOutputs.get(u) || []); if (sources.length > 0) { return { display: sources[0] + ' [' + u + ']', firstTarget: sources[0] }; } return { display: String(u), firstTarget: null }; }); - const outputs = (sn.outputs || []).slice().sort((a, b) => a - b).map(u => { + const outputs = sacnOutputs.slice().sort((a, b) => a - b).map(u => { const dests = sacnCollapseNames(sacnUniverseInputs.get(u) || []); if (dests.length > 0) { return { display: dests[0] + ' [' + u + ']', firstTarget: dests[0] }; diff --git a/types.go b/types.go index d67bba2..562a345 100644 --- a/types.go +++ b/types.go @@ -140,9 +140,23 @@ type Node struct { PoEBudget *PoEBudget `json:"poe_budget,omitempty"` IsDanteClockMaster bool `json:"is_dante_clock_master,omitempty"` DanteTxChannels string `json:"dante_tx_channels,omitempty"` + MulticastGroups []string `json:"multicast_groups,omitempty"` + ArtNetInputs []int `json:"artnet_inputs,omitempty"` + ArtNetOutputs []int `json:"artnet_outputs,omitempty"` + SACNInputs []int `json:"sacn_inputs,omitempty"` + SACNOutputs []int `json:"sacn_outputs,omitempty"` + DanteTx []*DantePeer `json:"dante_tx,omitempty"` + DanteRx []*DantePeer `json:"dante_rx,omitempty"` + Unreachable bool `json:"unreachable,omitempty"` pollTrigger chan struct{} } +type DantePeer struct { + Node *Node `json:"node"` + Channels []string `json:"channels,omitempty"` + Status map[string]string `json:"status,omitempty"` +} + func (n *Node) WithInterface(ifaceKey string) *Node { if ifaceKey == "" { return n