diff --git a/pkg/swarm/config.go b/pkg/swarm/config.go index 6ec6c6c8e..6c12a3258 100644 --- a/pkg/swarm/config.go +++ b/pkg/swarm/config.go @@ -209,13 +209,13 @@ func DefaultConfig() *Config { BindAddr: "0.0.0.0", BindPort: DefaultBindPort, Discovery: DiscoveryConfig{ - JoinAddrs: nil, - GossipInterval: Duration{DefaultGossipInterval}, - PushPullInterval: Duration{DefaultPushPullInterval}, - NodeTimeout: Duration{DefaultNodeTimeout}, - DeadNodeTimeout: Duration{DefaultDeadNodeTimeout}, - AuthSecret: "", - RequireAuth: false, + JoinAddrs: nil, + GossipInterval: Duration{DefaultGossipInterval}, + PushPullInterval: Duration{DefaultPushPullInterval}, + NodeTimeout: Duration{DefaultNodeTimeout}, + DeadNodeTimeout: Duration{DefaultDeadNodeTimeout}, + AuthSecret: "", + RequireAuth: false, EnableMessageSigning: false, }, Handoff: HandoffConfig{ @@ -230,16 +230,16 @@ func DefaultConfig() *Config { Timeout: Duration{10 * time.Second}, }, LoadMonitor: LoadMonitorConfig{ - Enabled: true, - Interval: Duration{DefaultLoadSampleInterval}, - SampleSize: DefaultLoadSampleSize, - CPUWeight: DefaultCPUWeight, - MemoryWeight: DefaultMemoryWeight, - SessionWeight: DefaultSessionWeight, + Enabled: true, + Interval: Duration{DefaultLoadSampleInterval}, + SampleSize: DefaultLoadSampleSize, + CPUWeight: DefaultCPUWeight, + MemoryWeight: DefaultMemoryWeight, + SessionWeight: DefaultSessionWeight, OffloadThreshold: DefaultOffloadThreshold, - MaxMemoryBytes: DefaultMaxMemoryBytes, - MaxGoroutines: DefaultMaxGoroutines, - MaxSessions: DefaultMaxSessions, + MaxMemoryBytes: DefaultMaxMemoryBytes, + MaxGoroutines: DefaultMaxGoroutines, + MaxSessions: DefaultMaxSessions, }, LeaderElection: LeaderElectionConfig{ Enabled: false, @@ -247,9 +247,9 @@ func DefaultConfig() *Config { LeaderHeartbeatTimeout: Duration{10 * time.Second}, }, Metrics: MetricsConfig{ - Enabled: false, - ExportInterval: Duration{10 * time.Second}, - PrometheusEnabled: false, + Enabled: false, + ExportInterval: Duration{10 * time.Second}, + PrometheusEnabled: false, PrometheusEndpoint: "/metrics", }, } diff --git a/pkg/swarm/discovery.go b/pkg/swarm/discovery.go index 841fd3301..e44f937b1 100644 --- a/pkg/swarm/discovery.go +++ b/pkg/swarm/discovery.go @@ -31,10 +31,10 @@ type DiscoveryService struct { rpcConn net.Listener auth *AuthProvider - mu sync.RWMutex - running bool - stopChan chan struct{} - once sync.Once + mu sync.RWMutex + running bool + stopChan chan struct{} + once sync.Once // Sequence number for updates seqNum uint64 @@ -281,13 +281,13 @@ func (ds *DiscoveryService) gossipLoop() { // GossipMessage represents a gossip message. type GossipMessage struct { - Type string `json:"type"` // "ping", "pong", "join", "update" - FromNode string `json:"from_node"` - SeqNum uint64 `json:"seq_num"` - Timestamp int64 `json:"timestamp"` - Payload []byte `json:"payload,omitempty"` + Type string `json:"type"` // "ping", "pong", "join", "update" + FromNode string `json:"from_node"` + SeqNum uint64 `json:"seq_num"` + Timestamp int64 `json:"timestamp"` + Payload []byte `json:"payload,omitempty"` Nodes []*NodeInfo `json:"nodes,omitempty"` // For memberlist exchange - AuthToken *AuthToken `json:"auth_token,omitempty"` + AuthToken *AuthToken `json:"auth_token,omitempty"` } // handleGossip handles an incoming gossip message. diff --git a/pkg/swarm/handoff.go b/pkg/swarm/handoff.go index ccfa0163d..3cb35ffca 100644 --- a/pkg/swarm/handoff.go +++ b/pkg/swarm/handoff.go @@ -21,11 +21,11 @@ import ( type HandoffReason string const ( - ReasonOverloaded HandoffReason = "overloaded" // Load is too high - ReasonNoCapability HandoffReason = "no_capability" // Missing capability - ReasonUserRequest HandoffReason = "user_request" // User explicitly requested - ReasonNodeLeave HandoffReason = "node_leave" // Node is leaving - ReasonShutdown HandoffReason = "shutdown" // Graceful shutdown + ReasonOverloaded HandoffReason = "overloaded" // Load is too high + ReasonNoCapability HandoffReason = "no_capability" // Missing capability + ReasonUserRequest HandoffReason = "user_request" // User explicitly requested + ReasonNodeLeave HandoffReason = "node_leave" // Node is leaving + ReasonShutdown HandoffReason = "shutdown" // Graceful shutdown ) // HandoffState represents the state of a handoff operation. @@ -42,27 +42,27 @@ const ( // HandoffRequest represents a request to hand off a session. type HandoffRequest struct { - RequestID string `json:"request_id"` - Reason HandoffReason `json:"reason"` - SessionKey string `json:"session_key"` - SessionMessages []SessionMessage `json:"session_messages,omitempty"` - Context map[string]any `json:"context,omitempty"` - RequiredCap string `json:"required_cap,omitempty"` - Metadata map[string]string `json:"metadata,omitempty"` - FromNodeID string `json:"from_node_id"` - FromNodeAddr string `json:"from_node_addr"` - Timestamp int64 `json:"timestamp"` + RequestID string `json:"request_id"` + Reason HandoffReason `json:"reason"` + SessionKey string `json:"session_key"` + SessionMessages []SessionMessage `json:"session_messages,omitempty"` + Context map[string]any `json:"context,omitempty"` + RequiredCap string `json:"required_cap,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` + FromNodeID string `json:"from_node_id"` + FromNodeAddr string `json:"from_node_addr"` + Timestamp int64 `json:"timestamp"` } // HandoffResponse represents the response to a handoff request. type HandoffResponse struct { - RequestID string `json:"request_id"` - Accepted bool `json:"accepted"` - NodeID string `json:"node_id"` - Reason string `json:"reason,omitempty"` - SessionKey string `json:"session_key,omitempty"` // New session key on target - Timestamp int64 `json:"timestamp"` - State HandoffState `json:"state"` + RequestID string `json:"request_id"` + Accepted bool `json:"accepted"` + NodeID string `json:"node_id"` + Reason string `json:"reason,omitempty"` + SessionKey string `json:"session_key,omitempty"` // New session key on target + Timestamp int64 `json:"timestamp"` + State HandoffState `json:"state"` } // HandoffCoordinator coordinates handoff operations between nodes. @@ -71,9 +71,9 @@ type HandoffCoordinator struct { membership *MembershipManager config HandoffConfig - pending map[string]*HandoffOperation // request_id -> operation - mu sync.RWMutex - conn *net.UDPConn + pending map[string]*HandoffOperation // request_id -> operation + mu sync.RWMutex + conn *net.UDPConn // Accept/reject callbacks onHandoffRequest func(*HandoffRequest) *HandoffResponse @@ -82,22 +82,22 @@ type HandoffCoordinator struct { // HandoffOperation represents an ongoing handoff operation. type HandoffOperation struct { - Request *HandoffRequest - Response *HandoffResponse - State HandoffState - StartTime time.Time - LastUpdate time.Time - RetryCount int - TargetNode *NodeWithState + Request *HandoffRequest + Response *HandoffResponse + State HandoffState + StartTime time.Time + LastUpdate time.Time + RetryCount int + TargetNode *NodeWithState } // NewHandoffCoordinator creates a new handoff coordinator. func NewHandoffCoordinator(ds *DiscoveryService, config HandoffConfig) (*HandoffCoordinator, error) { hc := &HandoffCoordinator{ - discovery: ds, + discovery: ds, membership: ds.membership, - config: config, - pending: make(map[string]*HandoffOperation), + config: config, + pending: make(map[string]*HandoffOperation), } // Bind UDP socket for handoff messages diff --git a/pkg/swarm/leader_election.go b/pkg/swarm/leader_election.go index 7146bd9fe..66a384c70 100644 --- a/pkg/swarm/leader_election.go +++ b/pkg/swarm/leader_election.go @@ -178,7 +178,8 @@ func (le *LeaderElection) monitorLeader() { // Check if leader is still in the membership if _, exists := le.membership.GetNode(leaderID); !exists { - logger.WarnCF("swarm", "Leader no longer in membership, triggering reelection", map[string]any{"leader_id": leaderID}) + logger.WarnCF("swarm", "Leader no longer in membership, triggering reelection", + map[string]any{"leader_id": leaderID}) // Trigger reelection by clearing current leader le.mu.Lock() le.currentLeader = "" diff --git a/pkg/swarm/load_monitor.go b/pkg/swarm/load_monitor.go index fe3280ead..fdf130faf 100644 --- a/pkg/swarm/load_monitor.go +++ b/pkg/swarm/load_monitor.go @@ -14,9 +14,9 @@ import ( // LoadMonitor monitors system load and calculates a load score. type LoadMonitor struct { - config *LoadMonitorConfig - samples []float64 - mu sync.RWMutex + config *LoadMonitorConfig + samples []float64 + mu sync.RWMutex sessionCount int ticker *time.Ticker stopChan chan struct{} @@ -33,9 +33,9 @@ func NewLoadMonitor(config *LoadMonitorConfig) *LoadMonitor { } lm := &LoadMonitor{ - config: config, - samples: make([]float64, 0, config.SampleSize), - stopChan: make(chan struct{}), + config: config, + samples: make([]float64, 0, config.SampleSize), + stopChan: make(chan struct{}), onThreshold: make([]func(float64), 0), } return lm @@ -89,20 +89,20 @@ func (lm *LoadMonitor) run() { // LoadMetrics represents current load metrics. type LoadMetrics struct { - CPUUsage float64 `json:"cpu_usage"` - MemoryUsage float64 `json:"memory_usage"` - ActiveSessions int `json:"active_sessions"` - Goroutines int `json:"goroutines"` - Score float64 `json:"score"` - Timestamp int64 `json:"timestamp"` + CPUUsage float64 `json:"cpu_usage"` + MemoryUsage float64 `json:"memory_usage"` + ActiveSessions int `json:"active_sessions"` + Goroutines int `json:"goroutines"` + Score float64 `json:"score"` + Timestamp int64 `json:"timestamp"` } // GetCurrentLoad returns the current load metrics. func (lm *LoadMonitor) GetCurrentLoad() *LoadMetrics { metrics := &LoadMetrics{ ActiveSessions: lm.GetSessionCount(), - Goroutines: runtime.NumGoroutine(), - Timestamp: time.Now().UnixNano(), + Goroutines: runtime.NumGoroutine(), + Timestamp: time.Now().UnixNano(), } // Get memory usage @@ -245,7 +245,7 @@ func (lm *LoadMonitor) GetTrend() string { // Simple linear regression to detect trend n := float64(len(lm.samples)) - sumX := n*(n-1)/2 + sumX := n * (n - 1) / 2 sumY := 0.0 sumXY := 0.0 @@ -255,7 +255,7 @@ func (lm *LoadMonitor) GetTrend() string { sumXY += x * s } - slope := (n*sumXY - sumX*sumY) / (n*(n-1)*(2*n-1)/6) + slope := (n*sumXY - sumX*sumY) / (n * (n - 1) * (2*n - 1) / 6) if slope > TrendIncreasingThreshold { return "increasing" diff --git a/pkg/swarm/membership.go b/pkg/swarm/membership.go index 46dc48ef5..ffa04c23c 100644 --- a/pkg/swarm/membership.go +++ b/pkg/swarm/membership.go @@ -20,9 +20,9 @@ type MembershipManager struct { mu sync.RWMutex // Event callbacks - onJoin []func(*NodeInfo) - onLeave []func(*NodeInfo) - onUpdate []func(*NodeInfo) + onJoin []func(*NodeInfo) + onLeave []func(*NodeInfo) + onUpdate []func(*NodeInfo) } // NewMembershipManager creates a new membership manager. diff --git a/pkg/swarm/metrics.go b/pkg/swarm/metrics.go index 7d3eb8ebb..f795a553f 100644 --- a/pkg/swarm/metrics.go +++ b/pkg/swarm/metrics.go @@ -19,13 +19,13 @@ type MetricsCollector struct { mu sync.RWMutex // Counters (atomic for performance) - messagesSent atomic.Int64 - messagesReceived atomic.Int64 + messagesSent atomic.Int64 + messagesReceived atomic.Int64 handoffsInitiated atomic.Int64 - handoffsAccepted atomic.Int64 - handoffsRejected atomic.Int64 - handoffsFailed atomic.Int64 - electionsWon atomic.Int64 + handoffsAccepted atomic.Int64 + handoffsRejected atomic.Int64 + handoffsFailed atomic.Int64 + electionsWon atomic.Int64 // Gauges (use atomic.Value for float64) currentLoadScore atomic.Value // float64 @@ -40,9 +40,9 @@ type MetricsCollector struct { // LatencyBucket tracks latency distribution. type LatencyBucket struct { - mu sync.RWMutex - count int64 - sum int64 + mu sync.RWMutex + count int64 + sum int64 buckets [12]int64 // 0-1ms, 1-2ms, 2-5ms, 5-10ms, 10-20ms, 20-50ms, 50-100ms, 100-200ms, 200-500ms, 500ms-1s, 1-2s, 2s+ } @@ -118,7 +118,7 @@ func (m *MetricsCollector) RecordLatency(operation string, latency time.Duration bucket := m.latencyBuckets[operation] m.mu.Unlock() - ms := int64(latency.Milliseconds()) + ms := latency.Milliseconds() bucket.mu.Lock() bucket.count++ @@ -174,24 +174,24 @@ func (m *MetricsCollector) GetMetrics() map[string]any { return map[string]any{ // Counters - "messages_sent": m.messagesSent.Load(), - "messages_received": m.messagesReceived.Load(), - "handoffs_initiated": m.handoffsInitiated.Load(), - "handoffs_accepted": m.handoffsAccepted.Load(), - "handoffs_rejected": m.handoffsRejected.Load(), - "handoffs_failed": m.handoffsFailed.Load(), - "elections_won": m.electionsWon.Load(), + "messages_sent": m.messagesSent.Load(), + "messages_received": m.messagesReceived.Load(), + "handoffs_initiated": m.handoffsInitiated.Load(), + "handoffs_accepted": m.handoffsAccepted.Load(), + "handoffs_rejected": m.handoffsRejected.Load(), + "handoffs_failed": m.handoffsFailed.Load(), + "elections_won": m.electionsWon.Load(), // Gauges - "load_score": m.currentLoadScore.Load(), - "active_sessions": m.activeSessions.Load(), - "member_count": m.memberCount.Load(), + "load_score": m.currentLoadScore.Load(), + "active_sessions": m.activeSessions.Load(), + "member_count": m.memberCount.Load(), // System info - "uptime_seconds": time.Since(m.startTime).Seconds(), + "uptime_seconds": time.Since(m.startTime).Seconds(), // Latency histograms - "latency_ms": latency, + "latency_ms": latency, } } @@ -205,7 +205,7 @@ func (m *MetricsCollector) percentile(bucket *LatencyBucket, p float64) float64 cumulative := int64(0) // Upper bounds for each bucket in ms - upperBounds := []int64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 1<<62} + upperBounds := []int64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 1 << 62} for i, count := range bucket.buckets { cumulative += count diff --git a/pkg/swarm/node.go b/pkg/swarm/node.go index 5c12f6d7c..d2f084ad2 100644 --- a/pkg/swarm/node.go +++ b/pkg/swarm/node.go @@ -63,13 +63,13 @@ const ( // NodeState represents the state of a node in the membership view. type NodeState struct { - Node *NodeInfo `json:"node"` - Status NodeStatus `json:"status"` - StatusSince int64 `json:"status_since"` // Unix nano when status was set - LastSeen int64 `json:"last_seen"` // Unix nano of last sighting - LastPing int64 `json:"last_ping"` // Unix nano of last successful ping - PingSuccess int `json:"ping_success"` // Consecutive successful pings - PingFailure int `json:"ping_failure"` // Consecutive failed pings + Node *NodeInfo `json:"node"` + Status NodeStatus `json:"status"` + StatusSince int64 `json:"status_since"` // Unix nano when status was set + LastSeen int64 `json:"last_seen"` // Unix nano of last sighting + LastPing int64 `json:"last_ping"` // Unix nano of last successful ping + PingSuccess int `json:"ping_success"` // Consecutive successful pings + PingFailure int `json:"ping_failure"` // Consecutive failed pings } // IsAvailable returns true if the node is available for handoff. @@ -85,9 +85,9 @@ func (ns *NodeState) UpdateStatus(status NodeStatus) { // NodeEvent represents a node state change event. type NodeEvent struct { - Node *NodeInfo `json:"node"` - Event EventType `json:"event"` - Time int64 `json:"time"` + Node *NodeInfo `json:"node"` + Event EventType `json:"event"` + Time int64 `json:"time"` } // EventType represents the type of node event. @@ -195,13 +195,13 @@ func (ed *EventDispatcher) DispatchContext(event *NodeEvent, ctx context.Context // NodeStats tracks statistics about a node. type NodeStats struct { - MessagesSent int64 `json:"messages_sent"` - MessagesReceived int64 `json:"messages_received"` - HandoffsAccepted int `json:"handoffs_accepted"` - HandoffsInitiated int `json:"handoffs_initiated"` - LastError string `json:"last_error,omitempty"` - LastErrorTime time.Time `json:"last_error_time,omitempty"` - UptimeStart time.Time `json:"uptime_start"` + MessagesSent int64 `json:"messages_sent"` + MessagesReceived int64 `json:"messages_received"` + HandoffsAccepted int `json:"handoffs_accepted"` + HandoffsInitiated int `json:"handoffs_initiated"` + LastError string `json:"last_error,omitempty"` + LastErrorTime time.Time `json:"last_error_time,omitempty"` + UptimeStart time.Time `json:"uptime_start"` } // NodeWithState combines a node with its state and stats. @@ -222,9 +222,9 @@ func (nws *NodeWithState) IsAvailable() bool { // ClusterView represents the current view of the cluster. type ClusterView struct { Nodes map[string]*NodeWithState `json:"nodes"` - LocalNodeID string `json:"local_node_id"` - Size int `json:"size"` - Version int64 `json:"version"` // View version for conflict detection + LocalNodeID string `json:"local_node_id"` + Size int `json:"size"` + Version int64 `json:"version"` // View version for conflict detection mu sync.RWMutex } diff --git a/pkg/swarm/session_transfer.go b/pkg/swarm/session_transfer.go index 59cf817f6..f89417320 100644 --- a/pkg/swarm/session_transfer.go +++ b/pkg/swarm/session_transfer.go @@ -17,12 +17,12 @@ import ( // SessionTransfer handles session migration between nodes. type SessionTransfer struct { - config RPCConfig - localNode *NodeInfo - transfers map[string]*TransferOperation // session_key -> operation - mu sync.RWMutex - conn *net.UDPConn - onReceive func(*TransferPayload) + config RPCConfig + localNode *NodeInfo + transfers map[string]*TransferOperation // session_key -> operation + mu sync.RWMutex + conn *net.UDPConn + onReceive func(*TransferPayload) } // TransferOperation represents an ongoing transfer operation. @@ -49,15 +49,15 @@ const ( // TransferPayload represents the session data being transferred. type TransferPayload struct { - SessionKey string `json:"session_key"` - SourceNodeID string `json:"source_node_id"` - TargetNodeID string `json:"target_node_id"` - Messages []SessionMessage `json:"messages"` - Summary string `json:"summary,omitempty"` - Context map[string]any `json:"context,omitempty"` - Metadata map[string]string `json:"metadata,omitempty"` - Timestamp int64 `json:"timestamp"` - TransferID string `json:"transfer_id"` + SessionKey string `json:"session_key"` + SourceNodeID string `json:"source_node_id"` + TargetNodeID string `json:"target_node_id"` + Messages []SessionMessage `json:"messages"` + Summary string `json:"summary,omitempty"` + Context map[string]any `json:"context,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` + Timestamp int64 `json:"timestamp"` + TransferID string `json:"transfer_id"` } // NewSessionTransfer creates a new session transfer handler. diff --git a/pkg/swarm/swarm_test.go b/pkg/swarm/swarm_test.go index 3385d4554..3708458d1 100644 --- a/pkg/swarm/swarm_test.go +++ b/pkg/swarm/swarm_test.go @@ -85,11 +85,11 @@ func TestClusterView(t *testing.T) { func TestLoadMonitor(t *testing.T) { config := &LoadMonitorConfig{ - Enabled: true, - Interval: Duration{time.Second}, - SampleSize: 10, - CPUWeight: 0.3, - MemoryWeight: 0.3, + Enabled: true, + Interval: Duration{time.Second}, + SampleSize: 10, + CPUWeight: 0.3, + MemoryWeight: 0.3, SessionWeight: 0.4, } diff --git a/pkg/swarm/types.go b/pkg/swarm/types.go index da3fabb78..3c451aefd 100644 --- a/pkg/swarm/types.go +++ b/pkg/swarm/types.go @@ -9,17 +9,17 @@ package swarm // SessionMessage represents a message in a session. // This is shared across handoff and session transfer. type SessionMessage struct { - Role string `json:"role"` - Content string `json:"content"` - Timestamp int64 `json:"timestamp,omitempty"` + Role string `json:"role"` + Content string `json:"content"` + Timestamp int64 `json:"timestamp,omitempty"` ToolCalls []ToolCallData `json:"tool_calls,omitempty"` } // ToolCallData represents tool call information in a message. type ToolCallData struct { - ID string `json:"id"` - Name string `json:"name"` - Arguments map[string]any `json:"arguments"` - Result string `json:"result,omitempty"` - Extra map[string]any `json:"extra,omitempty"` + ID string `json:"id"` + Name string `json:"name"` + Arguments map[string]any `json:"arguments"` + Result string `json:"result,omitempty"` + Extra map[string]any `json:"extra,omitempty"` }