- Update edge node registry with snapshot tracking - Fix mapper and store tests for registry changes - Update transport server integration tests - Add config support for new edge settings - Sync roadmap with current phase progress - Archive completed task groups
192 lines
4 KiB
Go
192 lines
4 KiB
Go
package node
|
|
|
|
import (
|
|
"fmt"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
|
|
toki "git.toki-labs.com/toki/proto-socket/go"
|
|
|
|
"iop/packages/go/config"
|
|
)
|
|
|
|
// LifecycleConnected is the default lifecycle state for a freshly registered entry.
|
|
const LifecycleConnected = "connected"
|
|
|
|
// NodeEntry represents one connected node.
|
|
type NodeEntry struct {
|
|
NodeID string
|
|
Alias string
|
|
AgentKind string
|
|
LifecycleState string
|
|
Client *toki.TcpClient
|
|
Index int
|
|
HasIndex bool
|
|
}
|
|
|
|
// Registry manages all nodes connected to edge.
|
|
type Registry struct {
|
|
mu sync.RWMutex
|
|
byID map[string]*NodeEntry
|
|
byAlias map[string]*NodeEntry
|
|
byIndex map[int]*NodeEntry
|
|
indexByID map[string]int
|
|
nextIndex int
|
|
}
|
|
|
|
func NewRegistry() *Registry {
|
|
return &Registry{
|
|
byID: make(map[string]*NodeEntry),
|
|
byAlias: make(map[string]*NodeEntry),
|
|
byIndex: make(map[int]*NodeEntry),
|
|
indexByID: make(map[string]int),
|
|
}
|
|
}
|
|
|
|
func (r *Registry) Register(entry *NodeEntry) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if entry.AgentKind == "" {
|
|
entry.AgentKind = config.AgentKindGenericNode
|
|
}
|
|
if entry.LifecycleState == "" {
|
|
entry.LifecycleState = LifecycleConnected
|
|
}
|
|
if idx, ok := r.indexByID[entry.NodeID]; ok {
|
|
entry.Index = idx
|
|
entry.HasIndex = true
|
|
} else {
|
|
if !entry.HasIndex {
|
|
entry.Index = r.nextIndex
|
|
}
|
|
entry.HasIndex = true
|
|
r.indexByID[entry.NodeID] = entry.Index
|
|
if entry.Index >= r.nextIndex {
|
|
r.nextIndex = entry.Index + 1
|
|
}
|
|
}
|
|
r.byID[entry.NodeID] = entry
|
|
r.byIndex[entry.Index] = entry
|
|
if entry.Alias != "" {
|
|
r.byAlias[entry.Alias] = entry
|
|
}
|
|
}
|
|
|
|
func (e *NodeEntry) DisplayLabel() string {
|
|
if e == nil {
|
|
return "unknown"
|
|
}
|
|
if e.HasIndex {
|
|
return fmt.Sprintf("node%d", e.Index)
|
|
}
|
|
if e.Alias != "" {
|
|
return e.Alias
|
|
}
|
|
if e.NodeID != "" {
|
|
return e.NodeID
|
|
}
|
|
return "unknown"
|
|
}
|
|
|
|
func (r *Registry) Unregister(nodeID string) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if entry, ok := r.byID[nodeID]; ok {
|
|
if entry.Alias != "" {
|
|
delete(r.byAlias, entry.Alias)
|
|
}
|
|
delete(r.byIndex, entry.Index)
|
|
delete(r.byID, nodeID)
|
|
}
|
|
}
|
|
|
|
func (r *Registry) Get(nodeID string) (*NodeEntry, bool) {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
e, ok := r.byID[nodeID]
|
|
return e, ok
|
|
}
|
|
|
|
func (r *Registry) Resolve(ref string) (*NodeEntry, error) {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
|
|
if ref != "" {
|
|
if entry, ok := r.byID[ref]; ok {
|
|
return entry, nil
|
|
}
|
|
if entry, ok := r.byAlias[ref]; ok {
|
|
return entry, nil
|
|
}
|
|
if entry, ok := r.resolveDisplayLabelLocked(ref); ok {
|
|
return entry, nil
|
|
}
|
|
return nil, fmt.Errorf("node %q not found", ref)
|
|
}
|
|
|
|
if len(r.byID) == 1 {
|
|
for _, entry := range r.byID {
|
|
return entry, nil
|
|
}
|
|
}
|
|
if len(r.byID) == 0 {
|
|
return nil, fmt.Errorf("no nodes connected")
|
|
}
|
|
return nil, fmt.Errorf("multiple nodes connected; select one with /node <id|alias>")
|
|
}
|
|
|
|
func (r *Registry) resolveDisplayLabelLocked(ref string) (*NodeEntry, bool) {
|
|
idx, ok := parseDisplayNodeIndex(ref)
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
entry, ok := r.byIndex[idx]
|
|
return entry, ok
|
|
}
|
|
|
|
func parseDisplayNodeIndex(ref string) (int, bool) {
|
|
if !strings.HasPrefix(ref, "node") {
|
|
return 0, false
|
|
}
|
|
raw := strings.TrimPrefix(ref, "node")
|
|
if raw == "" {
|
|
return 0, false
|
|
}
|
|
idx, err := strconv.Atoi(raw)
|
|
if err != nil || idx < 0 {
|
|
return 0, false
|
|
}
|
|
return idx, true
|
|
}
|
|
|
|
func (r *Registry) All() []*NodeEntry {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
out := make([]*NodeEntry, 0, len(r.byID))
|
|
for _, e := range r.byID {
|
|
out = append(out, e)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool {
|
|
if out[i].HasIndex != out[j].HasIndex {
|
|
return out[i].HasIndex
|
|
}
|
|
if out[i].HasIndex && out[i].Index != out[j].Index {
|
|
return out[i].Index < out[j].Index
|
|
}
|
|
return out[i].NodeID < out[j].NodeID
|
|
})
|
|
return out
|
|
}
|
|
|
|
func (r *Registry) Count() int {
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return len(r.byID)
|
|
}
|
|
|
|
// Pick is deprecated: use Resolve("") instead for single-node fallback, or Resolve(ref) for explicit selection.
|
|
func (r *Registry) Pick() (*NodeEntry, error) {
|
|
return r.Resolve("")
|
|
}
|