Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
443425d6e3 | ||
|
|
70eddbc533 | ||
|
|
9970c99509 |
+97
-8
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/nats-io/nats.go"
|
"github.com/nats-io/nats.go"
|
||||||
)
|
)
|
||||||
@@ -137,11 +138,43 @@ func (dvm *DistributedVM) LoadModel(model RuntimeModel) error {
|
|||||||
|
|
||||||
// SendMessage routes messages across the distributed cluster
|
// SendMessage routes messages across the distributed cluster
|
||||||
func (dvm *DistributedVM) SendMessage(message RuntimeMessage) error {
|
func (dvm *DistributedVM) SendMessage(message RuntimeMessage) error {
|
||||||
// This is a simplified implementation
|
actorID := message.GetTargetActorID()
|
||||||
// In practice, this would determine the target node based on sharding
|
targetNode := dvm.GetActorNode(actorID)
|
||||||
// and route the message appropriately
|
|
||||||
|
|
||||||
|
if targetNode == dvm.nodeID {
|
||||||
return dvm.localRuntime.SendMessage(message)
|
return dvm.localRuntime.SendMessage(message)
|
||||||
|
}
|
||||||
|
|
||||||
|
return dvm.routeMessageToNode(actorID, message)
|
||||||
|
}
|
||||||
|
|
||||||
|
// routeMessageToNode sends a message to another node for delivery to the target actor
|
||||||
|
func (dvm *DistributedVM) routeMessageToNode(actorID string, message RuntimeMessage) error {
|
||||||
|
hops := 0
|
||||||
|
var body map[string]interface{}
|
||||||
|
if mp, ok := message.(*MessagePayload); ok {
|
||||||
|
hops = mp.Hops
|
||||||
|
body = mp.Body
|
||||||
|
}
|
||||||
|
if hops >= MaxRouteHops {
|
||||||
|
dvm.cluster.logger.Printf("Dropping message for actor %s: exceeded max hops (%d)", actorID, MaxRouteHops)
|
||||||
|
return fmt.Errorf("message exceeded max hops")
|
||||||
|
}
|
||||||
|
|
||||||
|
msg := ClusterMessage{
|
||||||
|
Type: "route_message",
|
||||||
|
From: dvm.nodeID,
|
||||||
|
To: actorID,
|
||||||
|
Payload: MessagePayload{
|
||||||
|
TargetActorID: actorID,
|
||||||
|
Type: message.GetType(),
|
||||||
|
Hops: hops + 1,
|
||||||
|
Body: body,
|
||||||
|
},
|
||||||
|
Timestamp: time.Now(),
|
||||||
|
}
|
||||||
|
|
||||||
|
return dvm.publishClusterMessage(msg)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetActorNode determines which node should handle a specific actor
|
// GetActorNode determines which node should handle a specific actor
|
||||||
@@ -189,8 +222,10 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) {
|
|||||||
dvm.localRuntime.LoadModel(&model)
|
dvm.localRuntime.LoadModel(&model)
|
||||||
|
|
||||||
case "route_message":
|
case "route_message":
|
||||||
// Handle message routing from other nodes
|
if clusterMsg.From == dvm.nodeID {
|
||||||
// Re-marshal and unmarshal to convert map[string]interface{} to concrete type
|
return
|
||||||
|
}
|
||||||
|
|
||||||
payloadBytes, err := json.Marshal(clusterMsg.Payload)
|
payloadBytes, err := json.Marshal(clusterMsg.Payload)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return
|
||||||
@@ -199,7 +234,24 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) {
|
|||||||
if err := json.Unmarshal(payloadBytes, &message); err != nil {
|
if err := json.Unmarshal(payloadBytes, &message); err != nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
dvm.localRuntime.SendMessage(&message)
|
|
||||||
|
if message.Hops >= MaxRouteHops {
|
||||||
|
dvm.cluster.logger.Printf("Dropping message for actor %s: exceeded max hops (%d)", message.TargetActorID, MaxRouteHops)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
targetActor := message.TargetActorID
|
||||||
|
msg := &MessagePayload{
|
||||||
|
TargetActorID: targetActor,
|
||||||
|
Type: message.Type,
|
||||||
|
Hops: message.Hops,
|
||||||
|
Body: message.Body,
|
||||||
|
}
|
||||||
|
if dvm.IsLocalActor(targetActor) {
|
||||||
|
dvm.localRuntime.SendMessage(msg)
|
||||||
|
} else {
|
||||||
|
dvm.routeMessageToNode(targetActor, msg)
|
||||||
|
}
|
||||||
|
|
||||||
case "rebalance":
|
case "rebalance":
|
||||||
// Handle shard rebalancing requests
|
// Handle shard rebalancing requests
|
||||||
@@ -209,8 +261,45 @@ func (dvm *DistributedVM) handleClusterMessage(msg *nats.Msg) {
|
|||||||
|
|
||||||
// handleRebalanceRequest processes shard rebalancing requests
|
// handleRebalanceRequest processes shard rebalancing requests
|
||||||
func (dvm *DistributedVM) handleRebalanceRequest(msg ClusterMessage) {
|
func (dvm *DistributedVM) handleRebalanceRequest(msg ClusterMessage) {
|
||||||
// Simplified rebalancing logic
|
if msg.From == dvm.nodeID {
|
||||||
// In practice, this would implement complex actor migration
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if !dvm.cluster.IsLeader() {
|
||||||
|
dvm.cluster.logger.Printf("Ignoring rebalance request: not the leader")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if dvm.cluster.shardMap == nil {
|
||||||
|
dvm.cluster.logger.Printf("Shard map is nil, skipping rebalance")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
payloadBytes, err := json.Marshal(msg.Payload)
|
||||||
|
if err != nil {
|
||||||
|
dvm.cluster.logger.Printf("Failed to marshal rebalance payload: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var newShardMap ShardMap
|
||||||
|
if err := json.Unmarshal(payloadBytes, &newShardMap); err != nil {
|
||||||
|
dvm.cluster.logger.Printf("Failed to unmarshal shard map: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
dvm.cluster.mutex.Lock()
|
||||||
|
if newShardMap.Version > dvm.cluster.shardMap.Version {
|
||||||
|
dvm.cluster.shardMap = &newShardMap
|
||||||
|
dvm.cluster.hashRing = NewConsistentHashRing()
|
||||||
|
for nodeID := range newShardMap.Nodes {
|
||||||
|
dvm.cluster.hashRing.AddNode(nodeID)
|
||||||
|
}
|
||||||
|
dvm.cluster.logger.Printf("Applied new shard map (version %d) from rebalance", newShardMap.Version)
|
||||||
|
} else {
|
||||||
|
dvm.cluster.logger.Printf("Ignoring stale shard map (got version %d, current %d)",
|
||||||
|
newShardMap.Version, dvm.cluster.shardMap.Version)
|
||||||
|
}
|
||||||
|
dvm.cluster.mutex.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
// publishClusterMessage sends a message to other cluster nodes
|
// publishClusterMessage sends a message to other cluster nodes
|
||||||
|
|||||||
+188
-12
@@ -154,6 +154,10 @@ func (cm *ClusterManager) handleClusterMessage(msg *nats.Msg) {
|
|||||||
if update, ok := clusterMsg.Payload.(NodeUpdate); ok {
|
if update, ok := clusterMsg.Payload.(NodeUpdate); ok {
|
||||||
cm.handleNodeUpdate(update)
|
cm.handleNodeUpdate(update)
|
||||||
}
|
}
|
||||||
|
case "shard_map":
|
||||||
|
cm.handleShardMapUpdate(clusterMsg)
|
||||||
|
case "migration_update":
|
||||||
|
cm.handleMigrationUpdate(clusterMsg)
|
||||||
default:
|
default:
|
||||||
cm.logger.Printf("Unknown cluster message type: %s", clusterMsg.Type)
|
cm.logger.Printf("Unknown cluster message type: %s", clusterMsg.Type)
|
||||||
}
|
}
|
||||||
@@ -217,16 +221,91 @@ func (cm *ClusterManager) handleNodeUpdate(update NodeUpdate) {
|
|||||||
func (cm *ClusterManager) handleRebalanceRequest(msg ClusterMessage) {
|
func (cm *ClusterManager) handleRebalanceRequest(msg ClusterMessage) {
|
||||||
cm.logger.Printf("Handling rebalance request from %s", msg.From)
|
cm.logger.Printf("Handling rebalance request from %s", msg.From)
|
||||||
|
|
||||||
// Implementation would handle the specific rebalancing logic
|
if !cm.IsLeader() {
|
||||||
// This is a simplified version
|
cm.logger.Printf("Ignoring rebalance request: not the leader")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.mutex.RLock()
|
||||||
|
activeNodes := make(map[string]*NodeInfo)
|
||||||
|
for nodeID, nodeInfo := range cm.nodes {
|
||||||
|
if nodeInfo.Status == NodeStatusActive {
|
||||||
|
activeNodes[nodeID] = nodeInfo
|
||||||
|
}
|
||||||
|
}
|
||||||
|
cm.mutex.RUnlock()
|
||||||
|
|
||||||
|
if len(activeNodes) == 0 {
|
||||||
|
cm.logger.Printf("No active nodes for rebalancing")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
placement := &ConsistentHashPlacement{}
|
||||||
|
newShardMap, err := placement.RebalanceShards(cm.shardMap, activeNodes)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to compute new shard map: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.mutex.Lock()
|
||||||
|
cm.shardMap = newShardMap
|
||||||
|
cm.mutex.Unlock()
|
||||||
|
|
||||||
|
cm.hashRing = NewConsistentHashRing()
|
||||||
|
for nodeID := range activeNodes {
|
||||||
|
cm.hashRing.AddNode(nodeID)
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.broadcastShardMap(newShardMap)
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleMigrationRequest processes actor migration requests
|
// handleMigrationRequest processes actor migration requests
|
||||||
func (cm *ClusterManager) handleMigrationRequest(msg ClusterMessage) {
|
func (cm *ClusterManager) handleMigrationRequest(msg ClusterMessage) {
|
||||||
cm.logger.Printf("Handling migration request from %s", msg.From)
|
cm.logger.Printf("Handling migration request from %s", msg.From)
|
||||||
|
|
||||||
// Implementation would handle the specific migration logic
|
var migration ActorMigration
|
||||||
// This is a simplified version
|
payloadBytes, err := json.Marshal(msg.Payload)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to marshal migration payload: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(payloadBytes, &migration); err != nil {
|
||||||
|
cm.logger.Printf("Failed to unmarshal migration request: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.logger.Printf("Actor %s migrating from %s to %s (shard %d)",
|
||||||
|
migration.ActorID, migration.FromNode, migration.ToNode, migration.ShardID)
|
||||||
|
|
||||||
|
if migration.FromNode == cm.nodeID {
|
||||||
|
cm.logger.Printf("Initiating local actor state export for %s", migration.ActorID)
|
||||||
|
migration.Status = string(MigrationInProgress)
|
||||||
|
cm.broadcastMigrationUpdate(migration)
|
||||||
|
}
|
||||||
|
|
||||||
|
if migration.ToNode == cm.nodeID {
|
||||||
|
cm.logger.Printf("Actor %s assigned to this node, waiting for state import", migration.ActorID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// broadcastMigrationUpdate propagates migration status updates to the cluster
|
||||||
|
func (cm *ClusterManager) broadcastMigrationUpdate(migration ActorMigration) {
|
||||||
|
msg := ClusterMessage{
|
||||||
|
Type: "migration_update",
|
||||||
|
From: cm.nodeID,
|
||||||
|
To: "broadcast",
|
||||||
|
Payload: migration,
|
||||||
|
}
|
||||||
|
|
||||||
|
data, err := json.Marshal(msg)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to marshal migration update: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := cm.natsConn.Publish("aether.cluster.migration_update", data); err != nil {
|
||||||
|
cm.logger.Printf("Failed to publish migration update: %v", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// triggerShardRebalancing initiates shard rebalancing across the cluster
|
// triggerShardRebalancing initiates shard rebalancing across the cluster
|
||||||
@@ -237,12 +316,11 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
|
|||||||
|
|
||||||
cm.logger.Printf("Triggering shard rebalancing: %s", reason)
|
cm.logger.Printf("Triggering shard rebalancing: %s", reason)
|
||||||
|
|
||||||
// Get active nodes
|
|
||||||
var activeNodes []*NodeInfo
|
|
||||||
cm.mutex.RLock()
|
cm.mutex.RLock()
|
||||||
for _, node := range cm.nodes {
|
activeNodes := make(map[string]*NodeInfo)
|
||||||
if node.Status == NodeStatusActive {
|
for nodeID, nodeInfo := range cm.nodes {
|
||||||
activeNodes = append(activeNodes, node)
|
if nodeInfo.Status == NodeStatusActive {
|
||||||
|
activeNodes[nodeID] = nodeInfo
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
cm.mutex.RUnlock()
|
cm.mutex.RUnlock()
|
||||||
@@ -252,8 +330,23 @@ func (cm *ClusterManager) triggerShardRebalancing(reason string) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// This would implement the actual rebalancing logic
|
placement := &ConsistentHashPlacement{}
|
||||||
cm.logger.Printf("Would rebalance across %d active nodes", len(activeNodes))
|
newShardMap, err := placement.RebalanceShards(cm.shardMap, activeNodes)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to compute new shard map: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.mutex.Lock()
|
||||||
|
cm.shardMap = newShardMap
|
||||||
|
cm.mutex.Unlock()
|
||||||
|
|
||||||
|
cm.hashRing = NewConsistentHashRing()
|
||||||
|
for nodeID := range activeNodes {
|
||||||
|
cm.hashRing.AddNode(nodeID)
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.broadcastShardMap(newShardMap)
|
||||||
}
|
}
|
||||||
|
|
||||||
// monitorNodes periodically checks node health and updates
|
// monitorNodes periodically checks node health and updates
|
||||||
@@ -319,16 +412,99 @@ func (cm *ClusterManager) GetNodes() map[string]*NodeInfo {
|
|||||||
return nodes
|
return nodes
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// handleShardMapUpdate applies a new shard map received from the leader
|
||||||
|
func (cm *ClusterManager) handleShardMapUpdate(msg ClusterMessage) {
|
||||||
|
if msg.From == cm.nodeID {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
payloadBytes, err := json.Marshal(msg.Payload)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to marshal shard map payload: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var newShardMap ShardMap
|
||||||
|
if err := json.Unmarshal(payloadBytes, &newShardMap); err != nil {
|
||||||
|
cm.logger.Printf("Failed to unmarshal shard map: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.mutex.Lock()
|
||||||
|
if newShardMap.Version > cm.shardMap.Version {
|
||||||
|
cm.shardMap = &newShardMap
|
||||||
|
cm.hashRing = NewConsistentHashRing()
|
||||||
|
for nodeID := range newShardMap.Nodes {
|
||||||
|
cm.hashRing.AddNode(nodeID)
|
||||||
|
}
|
||||||
|
cm.logger.Printf("Applied new shard map (version %d)", newShardMap.Version)
|
||||||
|
} else {
|
||||||
|
cm.logger.Printf("Ignoring stale shard map (got version %d, current %d)",
|
||||||
|
newShardMap.Version, cm.shardMap.Version)
|
||||||
|
}
|
||||||
|
cm.mutex.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
// GetShardMap returns the current shard mapping
|
// GetShardMap returns the current shard mapping
|
||||||
func (cm *ClusterManager) GetShardMap() *ShardMap {
|
func (cm *ClusterManager) GetShardMap() *ShardMap {
|
||||||
cm.mutex.RLock()
|
cm.mutex.RLock()
|
||||||
defer cm.mutex.RUnlock()
|
defer cm.mutex.RUnlock()
|
||||||
|
|
||||||
// Return a copy to prevent external mutation
|
// Return a copy to prevent external mutation
|
||||||
return &ShardMap{
|
copy := &ShardMap{
|
||||||
Version: cm.shardMap.Version,
|
Version: cm.shardMap.Version,
|
||||||
Shards: make(map[int][]string),
|
Shards: make(map[int][]string),
|
||||||
Nodes: make(map[string]NodeInfo),
|
Nodes: make(map[string]NodeInfo),
|
||||||
UpdateTime: cm.shardMap.UpdateTime,
|
UpdateTime: cm.shardMap.UpdateTime,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for shardID, nodes := range cm.shardMap.Shards {
|
||||||
|
copy.Shards[shardID] = append([]string(nil), nodes...)
|
||||||
|
}
|
||||||
|
|
||||||
|
for nodeID, nodeInfo := range cm.shardMap.Nodes {
|
||||||
|
copy.Nodes[nodeID] = nodeInfo
|
||||||
|
}
|
||||||
|
|
||||||
|
return copy
|
||||||
|
}
|
||||||
|
|
||||||
|
// broadcastShardMap propagates a new shard map to all cluster nodes via NATS
|
||||||
|
func (cm *ClusterManager) broadcastShardMap(newShardMap *ShardMap) {
|
||||||
|
msg := ClusterMessage{
|
||||||
|
Type: "shard_map",
|
||||||
|
From: cm.nodeID,
|
||||||
|
To: "broadcast",
|
||||||
|
Payload: newShardMap,
|
||||||
|
Timestamp: time.Now(),
|
||||||
|
}
|
||||||
|
|
||||||
|
data, err := json.Marshal(msg)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to marshal shard map broadcast: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := cm.natsConn.Publish("aether.cluster.shard_map", data); err != nil {
|
||||||
|
cm.logger.Printf("Failed to publish shard map broadcast: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.logger.Printf("Broadcast new shard map (version %d) to cluster", newShardMap.Version)
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleMigrationUpdate processes migration status update messages from other nodes
|
||||||
|
func (cm *ClusterManager) handleMigrationUpdate(msg ClusterMessage) {
|
||||||
|
var migration ActorMigration
|
||||||
|
payloadBytes, err := json.Marshal(msg.Payload)
|
||||||
|
if err != nil {
|
||||||
|
cm.logger.Printf("Failed to marshal migration update payload: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(payloadBytes, &migration); err != nil {
|
||||||
|
cm.logger.Printf("Failed to unmarshal migration update: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
cm.logger.Printf("Migration update for actor %s: status=%s (from %s)",
|
||||||
|
migration.ActorID, migration.Status, msg.From)
|
||||||
}
|
}
|
||||||
|
|||||||
+100
-18
@@ -6,6 +6,8 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"hash"
|
"hash"
|
||||||
"hash/fnv"
|
"hash/fnv"
|
||||||
|
"sort"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// MigrationStatus tracks actor migration progress
|
// MigrationStatus tracks actor migration progress
|
||||||
@@ -180,36 +182,116 @@ func (sm *ShardManager) GetReplicationFactor() int {
|
|||||||
// ConsistentHashPlacement implements PlacementStrategy using consistent hashing
|
// ConsistentHashPlacement implements PlacementStrategy using consistent hashing
|
||||||
type ConsistentHashPlacement struct{}
|
type ConsistentHashPlacement struct{}
|
||||||
|
|
||||||
// PlaceActor places an actor using consistent hashing
|
// PlaceActor places an actor using the consistent hash ring
|
||||||
func (chp *ConsistentHashPlacement) PlaceActor(actorID string, shardMap *ShardMap, nodes map[string]*NodeInfo) (string, error) {
|
func (chp *ConsistentHashPlacement) PlaceActor(actorID string, shardMap *ShardMap, nodes map[string]*NodeInfo) (string, error) {
|
||||||
if len(nodes) == 0 {
|
if len(nodes) == 0 {
|
||||||
return "", fmt.Errorf("no nodes available for placement")
|
return "", fmt.Errorf("no nodes available for placement")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Simple consistent hash placement - in a real implementation,
|
ring := NewConsistentHashRing()
|
||||||
// this would use the consistent hash ring
|
|
||||||
h := sha256.Sum256([]byte(actorID))
|
|
||||||
nodeIndex := binary.BigEndian.Uint32(h[:4]) % uint32(len(nodes))
|
|
||||||
|
|
||||||
i := 0
|
|
||||||
for nodeID := range nodes {
|
for nodeID := range nodes {
|
||||||
if i == int(nodeIndex) {
|
ring.AddNode(nodeID)
|
||||||
return nodeID, nil
|
|
||||||
}
|
|
||||||
i++
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fallback to first node
|
node := ring.GetNode(actorID)
|
||||||
|
if node == "" {
|
||||||
|
sortedNodeIDs := make([]string, 0, len(nodes))
|
||||||
for nodeID := range nodes {
|
for nodeID := range nodes {
|
||||||
return nodeID, nil
|
sortedNodeIDs = append(sortedNodeIDs, nodeID)
|
||||||
|
}
|
||||||
|
sort.Strings(sortedNodeIDs)
|
||||||
|
return sortedNodeIDs[0], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
return "", fmt.Errorf("failed to place actor")
|
return node, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// RebalanceShards rebalances shards across nodes
|
// RebalanceShards redistributes shards across nodes using consistent hashing
|
||||||
func (chp *ConsistentHashPlacement) RebalanceShards(currentMap *ShardMap, nodes map[string]*NodeInfo) (*ShardMap, error) {
|
func (chp *ConsistentHashPlacement) RebalanceShards(currentMap *ShardMap, nodes map[string]*NodeInfo) (*ShardMap, error) {
|
||||||
// This is a simplified implementation
|
if len(nodes) == 0 {
|
||||||
// In practice, this would implement sophisticated rebalancing logic
|
return nil, fmt.Errorf("no nodes available for rebalancing")
|
||||||
return currentMap, nil
|
}
|
||||||
|
|
||||||
|
ring := NewConsistentHashRing()
|
||||||
|
for nodeID := range nodes {
|
||||||
|
ring.AddNode(nodeID)
|
||||||
|
}
|
||||||
|
|
||||||
|
replicaCount := chp.deriveReplicaCount(currentMap)
|
||||||
|
|
||||||
|
newMap := &ShardMap{
|
||||||
|
Version: currentMap.Version + 1,
|
||||||
|
Shards: make(map[int][]string),
|
||||||
|
Nodes: make(map[string]NodeInfo),
|
||||||
|
UpdateTime: time.Now(),
|
||||||
|
}
|
||||||
|
|
||||||
|
for nodeID, nodeInfo := range nodes {
|
||||||
|
newMap.Nodes[nodeID] = *nodeInfo
|
||||||
|
}
|
||||||
|
|
||||||
|
for shardID := range currentMap.Shards {
|
||||||
|
primaryNode := ring.GetNode(fmt.Sprintf("shard-%d", shardID))
|
||||||
|
if primaryNode == "" {
|
||||||
|
sortedNodeIDs := make([]string, 0, len(nodes))
|
||||||
|
for nodeID := range nodes {
|
||||||
|
sortedNodeIDs = append(sortedNodeIDs, nodeID)
|
||||||
|
}
|
||||||
|
sort.Strings(sortedNodeIDs)
|
||||||
|
primaryNode = sortedNodeIDs[0]
|
||||||
|
}
|
||||||
|
|
||||||
|
var replicaNodes []string
|
||||||
|
candidates := make([]string, 0, len(nodes))
|
||||||
|
for nodeID := range nodes {
|
||||||
|
if nodeID != primaryNode {
|
||||||
|
candidates = append(candidates, nodeID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sort.Strings(candidates)
|
||||||
|
|
||||||
|
for i := 0; i < replicaCount && len(replicaNodes) < replicaCount; i++ {
|
||||||
|
node := ring.GetNode(fmt.Sprintf("shard-%d-replica-%d", shardID, i))
|
||||||
|
if node != "" && node != primaryNode {
|
||||||
|
found := false
|
||||||
|
for _, existing := range replicaNodes {
|
||||||
|
if existing == node {
|
||||||
|
found = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if !found {
|
||||||
|
replicaNodes = append(replicaNodes, node)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(replicaNodes) == 0 && len(candidates) > 0 {
|
||||||
|
replicaNodes = append(replicaNodes, candidates[0])
|
||||||
|
}
|
||||||
|
|
||||||
|
if len(replicaNodes) > replicaCount {
|
||||||
|
replicaNodes = replicaNodes[:replicaCount]
|
||||||
|
}
|
||||||
|
|
||||||
|
shardNodes := []string{primaryNode}
|
||||||
|
shardNodes = append(shardNodes, replicaNodes...)
|
||||||
|
newMap.Shards[shardID] = shardNodes
|
||||||
|
}
|
||||||
|
|
||||||
|
return newMap, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// deriveReplicaCount extracts the replication factor from the current shard map
|
||||||
|
func (chp *ConsistentHashPlacement) deriveReplicaCount(currentMap *ShardMap) int {
|
||||||
|
maxNodes := 0
|
||||||
|
for _, nodes := range currentMap.Shards {
|
||||||
|
if len(nodes) > maxNodes {
|
||||||
|
maxNodes = len(nodes)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if maxNodes <= 1 {
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
return maxNodes - 1
|
||||||
}
|
}
|
||||||
|
|||||||
+20
-4
@@ -650,7 +650,8 @@ func TestConsistentHashPlacement_RebalanceShards(t *testing.T) {
|
|||||||
placement := &ConsistentHashPlacement{}
|
placement := &ConsistentHashPlacement{}
|
||||||
currentMap := &ShardMap{
|
currentMap := &ShardMap{
|
||||||
Version: 1,
|
Version: 1,
|
||||||
Shards: map[int][]string{0: {"node-1"}},
|
Shards: map[int][]string{0: {"node-1"}, 1: {"node-1"}, 2: {"node-2"}},
|
||||||
|
Nodes: map[string]NodeInfo{},
|
||||||
}
|
}
|
||||||
nodes := map[string]*NodeInfo{
|
nodes := map[string]*NodeInfo{
|
||||||
"node-1": {ID: "node-1"},
|
"node-1": {ID: "node-1"},
|
||||||
@@ -662,9 +663,24 @@ func TestConsistentHashPlacement_RebalanceShards(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("unexpected error: %v", err)
|
t.Errorf("unexpected error: %v", err)
|
||||||
}
|
}
|
||||||
// Current implementation returns unchanged map
|
if result == nil {
|
||||||
if result != currentMap {
|
t.Fatal("rebalance returned nil")
|
||||||
t.Error("expected same map returned (simplified implementation)")
|
}
|
||||||
|
if result.Version != currentMap.Version+1 {
|
||||||
|
t.Errorf("expected version %d, got %d", currentMap.Version+1, result.Version)
|
||||||
|
}
|
||||||
|
if len(result.Shards) != len(currentMap.Shards) {
|
||||||
|
t.Errorf("expected %d shards, got %d", len(currentMap.Shards), len(result.Shards))
|
||||||
|
}
|
||||||
|
for shardID, shardNodes := range result.Shards {
|
||||||
|
if len(shardNodes) == 0 {
|
||||||
|
t.Errorf("shard %d has no nodes assigned", shardID)
|
||||||
|
}
|
||||||
|
for _, node := range shardNodes {
|
||||||
|
if _, exists := nodes[node]; !exists {
|
||||||
|
t.Errorf("shard %d assigned to unknown node %s", shardID, node)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -191,11 +191,16 @@ func (m *ModelPayload) GetID() string { return m.ID }
|
|||||||
// GetName implements RuntimeModel
|
// GetName implements RuntimeModel
|
||||||
func (m *ModelPayload) GetName() string { return m.Name }
|
func (m *ModelPayload) GetName() string { return m.Name }
|
||||||
|
|
||||||
|
// MaxRouteHops is the maximum number of hops a routed message can take before being dropped
|
||||||
|
const MaxRouteHops = 10
|
||||||
|
|
||||||
// MessagePayload is a concrete type for JSON-unmarshaling RuntimeMessage payloads.
|
// MessagePayload is a concrete type for JSON-unmarshaling RuntimeMessage payloads.
|
||||||
// Use this when receiving message data over the network.
|
// Use this when receiving message data over the network.
|
||||||
type MessagePayload struct {
|
type MessagePayload struct {
|
||||||
TargetActorID string `json:"targetActorId"`
|
TargetActorID string `json:"targetActorId"`
|
||||||
Type string `json:"type"`
|
Type string `json:"type"`
|
||||||
|
Hops int `json:"hops,omitempty"`
|
||||||
|
Body map[string]interface{} `json:"body,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetTargetActorID implements RuntimeMessage
|
// GetTargetActorID implements RuntimeMessage
|
||||||
|
|||||||
@@ -1,26 +1,26 @@
|
|||||||
module git.flowmade.one/flowmade-one/aether
|
module git.flowmade.one/flowmade-one/aether
|
||||||
|
|
||||||
go 1.25.0
|
go 1.23.0
|
||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/google/uuid v1.6.0
|
github.com/google/uuid v1.6.0
|
||||||
github.com/nats-io/nats.go v1.52.0
|
github.com/nats-io/nats.go v1.37.0
|
||||||
github.com/prometheus/client_golang v1.23.2
|
github.com/prometheus/client_golang v1.23.2
|
||||||
)
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
github.com/beorn7/perks v1.0.1 // indirect
|
github.com/beorn7/perks v1.0.1 // indirect
|
||||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||||
github.com/klauspost/compress v1.18.5 // indirect
|
github.com/klauspost/compress v1.18.0 // indirect
|
||||||
github.com/kr/text v0.2.0 // indirect
|
github.com/kr/text v0.2.0 // indirect
|
||||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||||
github.com/nats-io/nkeys v0.4.15 // indirect
|
github.com/nats-io/nkeys v0.4.7 // indirect
|
||||||
github.com/nats-io/nuid v1.0.1 // indirect
|
github.com/nats-io/nuid v1.0.1 // indirect
|
||||||
github.com/prometheus/client_model v0.6.2 // indirect
|
github.com/prometheus/client_model v0.6.2 // indirect
|
||||||
github.com/prometheus/common v0.66.1 // indirect
|
github.com/prometheus/common v0.66.1 // indirect
|
||||||
github.com/prometheus/procfs v0.16.1 // indirect
|
github.com/prometheus/procfs v0.16.1 // indirect
|
||||||
go.yaml.in/yaml/v2 v2.4.2 // indirect
|
go.yaml.in/yaml/v2 v2.4.2 // indirect
|
||||||
golang.org/x/crypto v0.49.0 // indirect
|
golang.org/x/crypto v0.18.0 // indirect
|
||||||
golang.org/x/sys v0.42.0 // indirect
|
golang.org/x/sys v0.35.0 // indirect
|
||||||
google.golang.org/protobuf v1.36.8 // indirect
|
google.golang.org/protobuf v1.36.8 // indirect
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -11,8 +11,6 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
|||||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||||
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
|
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
|
||||||
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
|
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
|
||||||
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
|
|
||||||
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
|
||||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||||
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
||||||
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||||
@@ -21,12 +19,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq
|
|||||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||||
github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE=
|
github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE=
|
||||||
github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
|
github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
|
||||||
github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc=
|
|
||||||
github.com/nats-io/nats.go v1.52.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
|
|
||||||
github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
|
github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
|
||||||
github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
|
github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
|
||||||
github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
|
|
||||||
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
|
|
||||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||||
@@ -49,12 +43,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
|
|||||||
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
|
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
|
||||||
golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
|
golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
|
||||||
golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
|
golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
|
||||||
golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
|
|
||||||
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
|
|
||||||
golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
|
golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
|
||||||
golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
|
golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
|
||||||
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
|
|
||||||
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
|
||||||
google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
|
google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
|
||||||
google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
|
google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
|
||||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||||
|
|||||||
@@ -1,6 +0,0 @@
|
|||||||
{
|
|
||||||
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
|
|
||||||
"extends": [
|
|
||||||
"config:recommended"
|
|
||||||
]
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user