You can not select more than 25 topics
			Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
		
		
		
		
		
			
		
			
				
					
					
						
							607 lines
						
					
					
						
							17 KiB
						
					
					
				
			
		
		
		
			
			
			
		
		
	
	
							607 lines
						
					
					
						
							17 KiB
						
					
					
				| package topology | |
| 
 | |
| import ( | |
| 	"fmt" | |
| 	"testing" | |
| 	"time" | |
| 
 | |
| 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb" | |
| 	"github.com/stretchr/testify/assert" | |
| 	"github.com/stretchr/testify/require" | |
| ) | |
| 
 | |
| // Helper function to find a disk by ID for testing - reduces code duplication | |
| func findDiskByID(disks []*DiskInfo, diskID uint32) *DiskInfo { | |
| 	for _, disk := range disks { | |
| 		if disk.DiskID == diskID { | |
| 			return disk | |
| 		} | |
| 	} | |
| 	return nil | |
| } | |
| 
 | |
| // TestActiveTopologyBasicOperations tests basic topology management | |
| func TestActiveTopologyBasicOperations(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 	assert.NotNil(t, topology) | |
| 	assert.Equal(t, 10, topology.recentTaskWindowSeconds) | |
| 
 | |
| 	// Test empty topology | |
| 	assert.Equal(t, 0, len(topology.nodes)) | |
| 	assert.Equal(t, 0, len(topology.disks)) | |
| 	assert.Equal(t, 0, len(topology.pendingTasks)) | |
| } | |
| 
 | |
| // TestActiveTopologyUpdate tests topology updates from master | |
| func TestActiveTopologyUpdate(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 
 | |
| 	// Create sample topology info | |
| 	topologyInfo := createSampleTopology() | |
| 
 | |
| 	err := topology.UpdateTopology(topologyInfo) | |
| 	require.NoError(t, err) | |
| 
 | |
| 	// Verify topology structure | |
| 	assert.Equal(t, 2, len(topology.nodes)) // 2 nodes | |
| 	assert.Equal(t, 4, len(topology.disks)) // 4 disks total (2 per node) | |
|  | |
| 	// Verify node structure | |
| 	node1, exists := topology.nodes["10.0.0.1:8080"] | |
| 	require.True(t, exists) | |
| 	assert.Equal(t, "dc1", node1.dataCenter) | |
| 	assert.Equal(t, "rack1", node1.rack) | |
| 	assert.Equal(t, 2, len(node1.disks)) | |
| 
 | |
| 	// Verify disk structure | |
| 	disk1, exists := topology.disks["10.0.0.1:8080:0"] | |
| 	require.True(t, exists) | |
| 	assert.Equal(t, uint32(0), disk1.DiskID) | |
| 	assert.Equal(t, "hdd", disk1.DiskType) | |
| 	assert.Equal(t, "dc1", disk1.DataCenter) | |
| } | |
| 
 | |
| // TestTaskLifecycle tests the complete task lifecycle | |
| func TestTaskLifecycle(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	taskID := "balance-001" | |
| 
 | |
| 	// 1. Add pending task | |
| 	err := topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     taskID, | |
| 		TaskType:   TaskTypeBalance, | |
| 		VolumeID:   1001, | |
| 		VolumeSize: 1024 * 1024 * 1024, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 0}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "10.0.0.2:8080", DiskID: 1}, | |
| 		}, | |
| 	}) | |
| 	assert.NoError(t, err, "Should add pending task successfully") | |
| 
 | |
| 	// Verify pending state | |
| 	assert.Equal(t, 1, len(topology.pendingTasks)) | |
| 	assert.Equal(t, 0, len(topology.assignedTasks)) | |
| 	assert.Equal(t, 0, len(topology.recentTasks)) | |
| 
 | |
| 	task := topology.pendingTasks[taskID] | |
| 	assert.Equal(t, TaskStatusPending, task.Status) | |
| 	assert.Equal(t, uint32(1001), task.VolumeID) | |
| 
 | |
| 	// Verify task assigned to disks | |
| 	sourceDisk := topology.disks["10.0.0.1:8080:0"] | |
| 	targetDisk := topology.disks["10.0.0.2:8080:1"] | |
| 	assert.Equal(t, 1, len(sourceDisk.pendingTasks)) | |
| 	assert.Equal(t, 1, len(targetDisk.pendingTasks)) | |
| 
 | |
| 	// 2. Assign task | |
| 	err = topology.AssignTask(taskID) | |
| 	require.NoError(t, err) | |
| 
 | |
| 	// Verify assigned state | |
| 	assert.Equal(t, 0, len(topology.pendingTasks)) | |
| 	assert.Equal(t, 1, len(topology.assignedTasks)) | |
| 	assert.Equal(t, 0, len(topology.recentTasks)) | |
| 
 | |
| 	task = topology.assignedTasks[taskID] | |
| 	assert.Equal(t, TaskStatusInProgress, task.Status) | |
| 
 | |
| 	// Verify task moved to assigned on disks | |
| 	assert.Equal(t, 0, len(sourceDisk.pendingTasks)) | |
| 	assert.Equal(t, 1, len(sourceDisk.assignedTasks)) | |
| 	assert.Equal(t, 0, len(targetDisk.pendingTasks)) | |
| 	assert.Equal(t, 1, len(targetDisk.assignedTasks)) | |
| 
 | |
| 	// 3. Complete task | |
| 	err = topology.CompleteTask(taskID) | |
| 	require.NoError(t, err) | |
| 
 | |
| 	// Verify completed state | |
| 	assert.Equal(t, 0, len(topology.pendingTasks)) | |
| 	assert.Equal(t, 0, len(topology.assignedTasks)) | |
| 	assert.Equal(t, 1, len(topology.recentTasks)) | |
| 
 | |
| 	task = topology.recentTasks[taskID] | |
| 	assert.Equal(t, TaskStatusCompleted, task.Status) | |
| 	assert.False(t, task.CompletedAt.IsZero()) | |
| } | |
| 
 | |
| // TestTaskDetectionScenarios tests various task detection scenarios | |
| func TestTaskDetectionScenarios(t *testing.T) { | |
| 	tests := []struct { | |
| 		name          string | |
| 		scenario      func() *ActiveTopology | |
| 		expectedTasks map[string]bool // taskType -> shouldDetect | |
| 	}{ | |
| 		{ | |
| 			name: "Empty cluster - no tasks needed", | |
| 			scenario: func() *ActiveTopology { | |
| 				topology := NewActiveTopology(10) | |
| 				topology.UpdateTopology(createEmptyTopology()) | |
| 				return topology | |
| 			}, | |
| 			expectedTasks: map[string]bool{ | |
| 				"balance": false, | |
| 				"vacuum":  false, | |
| 				"ec":      false, | |
| 			}, | |
| 		}, | |
| 		{ | |
| 			name: "Unbalanced cluster - balance task needed", | |
| 			scenario: func() *ActiveTopology { | |
| 				topology := NewActiveTopology(10) | |
| 				topology.UpdateTopology(createUnbalancedTopology()) | |
| 				return topology | |
| 			}, | |
| 			expectedTasks: map[string]bool{ | |
| 				"balance": true, | |
| 				"vacuum":  false, | |
| 				"ec":      false, | |
| 			}, | |
| 		}, | |
| 		{ | |
| 			name: "High garbage ratio - vacuum task needed", | |
| 			scenario: func() *ActiveTopology { | |
| 				topology := NewActiveTopology(10) | |
| 				topology.UpdateTopology(createHighGarbageTopology()) | |
| 				return topology | |
| 			}, | |
| 			expectedTasks: map[string]bool{ | |
| 				"balance": false, | |
| 				"vacuum":  true, | |
| 				"ec":      false, | |
| 			}, | |
| 		}, | |
| 		{ | |
| 			name: "Large volumes - EC task needed", | |
| 			scenario: func() *ActiveTopology { | |
| 				topology := NewActiveTopology(10) | |
| 				topology.UpdateTopology(createLargeVolumeTopology()) | |
| 				return topology | |
| 			}, | |
| 			expectedTasks: map[string]bool{ | |
| 				"balance": false, | |
| 				"vacuum":  false, | |
| 				"ec":      true, | |
| 			}, | |
| 		}, | |
| 		{ | |
| 			name: "Recent tasks - no immediate re-detection", | |
| 			scenario: func() *ActiveTopology { | |
| 				topology := NewActiveTopology(10) | |
| 				topology.UpdateTopology(createUnbalancedTopology()) | |
| 				// Add recent balance task | |
| 				topology.recentTasks["recent-balance"] = &taskState{ | |
| 					VolumeID:    1001, | |
| 					TaskType:    TaskTypeBalance, | |
| 					Status:      TaskStatusCompleted, | |
| 					CompletedAt: time.Now().Add(-5 * time.Second), // 5 seconds ago | |
| 				} | |
| 				return topology | |
| 			}, | |
| 			expectedTasks: map[string]bool{ | |
| 				"balance": false, // Should not detect due to recent task | |
| 				"vacuum":  false, | |
| 				"ec":      false, | |
| 			}, | |
| 		}, | |
| 	} | |
| 
 | |
| 	for _, tt := range tests { | |
| 		t.Run(tt.name, func(t *testing.T) { | |
| 			topology := tt.scenario() | |
| 
 | |
| 			// Test balance task detection | |
| 			shouldDetectBalance := tt.expectedTasks["balance"] | |
| 			actualDetectBalance := !topology.HasRecentTaskForVolume(1001, TaskTypeBalance) | |
| 			if shouldDetectBalance { | |
| 				assert.True(t, actualDetectBalance, "Should detect balance task") | |
| 			} else { | |
| 				// Note: In real implementation, task detection would be more sophisticated | |
| 				// This is a simplified test of the recent task prevention mechanism | |
| 			} | |
| 
 | |
| 			// Test that recent tasks prevent re-detection | |
| 			if len(topology.recentTasks) > 0 { | |
| 				for _, task := range topology.recentTasks { | |
| 					hasRecent := topology.HasRecentTaskForVolume(task.VolumeID, task.TaskType) | |
| 					assert.True(t, hasRecent, "Should find recent task for volume %d", task.VolumeID) | |
| 				} | |
| 			} | |
| 		}) | |
| 	} | |
| } | |
| 
 | |
| // TestTargetSelectionScenarios tests target selection for different task types | |
| func TestTargetSelectionScenarios(t *testing.T) { | |
| 	tests := []struct { | |
| 		name               string | |
| 		topology           *ActiveTopology | |
| 		taskType           TaskType | |
| 		excludeNode        string | |
| 		expectedTargets    int | |
| 		expectedBestTarget string | |
| 	}{ | |
| 		{ | |
| 			name:            "Balance task - find least loaded disk", | |
| 			topology:        createTopologyWithLoad(), | |
| 			taskType:        TaskTypeBalance, | |
| 			excludeNode:     "10.0.0.1:8080", // Exclude source node | |
| 			expectedTargets: 2,               // 2 disks on other node | |
| 		}, | |
| 		{ | |
| 			name:            "EC task - find multiple available disks", | |
| 			topology:        createTopologyForEC(), | |
| 			taskType:        TaskTypeErasureCoding, | |
| 			excludeNode:     "", // Don't exclude any nodes | |
| 			expectedTargets: 4,  // All 4 disks available | |
| 		}, | |
| 		{ | |
| 			name:            "Vacuum task - avoid conflicting disks", | |
| 			topology:        createTopologyWithConflicts(), | |
| 			taskType:        TaskTypeVacuum, | |
| 			excludeNode:     "", | |
| 			expectedTargets: 1, // Only 1 disk without conflicts (conflicts exclude more disks) | |
| 		}, | |
| 	} | |
| 
 | |
| 	for _, tt := range tests { | |
| 		t.Run(tt.name, func(t *testing.T) { | |
| 			availableDisks := tt.topology.GetAvailableDisks(tt.taskType, tt.excludeNode) | |
| 			assert.Equal(t, tt.expectedTargets, len(availableDisks), | |
| 				"Expected %d available disks, got %d", tt.expectedTargets, len(availableDisks)) | |
| 
 | |
| 			// Verify disks are actually available | |
| 			for _, disk := range availableDisks { | |
| 				assert.NotEqual(t, tt.excludeNode, disk.NodeID, | |
| 					"Available disk should not be on excluded node") | |
| 
 | |
| 				assert.Less(t, disk.LoadCount, 2, "Disk load should be less than 2") | |
| 			} | |
| 		}) | |
| 	} | |
| } | |
| 
 | |
| // TestDiskLoadCalculation tests disk load calculation | |
| func TestDiskLoadCalculation(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	// Initially no load | |
| 	disks := topology.GetNodeDisks("10.0.0.1:8080") | |
| 	targetDisk := findDiskByID(disks, 0) | |
| 	require.NotNil(t, targetDisk, "Should find disk with ID 0") | |
| 	assert.Equal(t, 0, targetDisk.LoadCount) | |
| 
 | |
| 	// Add pending task | |
| 	err := topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     "task1", | |
| 		TaskType:   TaskTypeBalance, | |
| 		VolumeID:   1001, | |
| 		VolumeSize: 1024 * 1024 * 1024, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 0}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "10.0.0.2:8080", DiskID: 1}, | |
| 		}, | |
| 	}) | |
| 	assert.NoError(t, err, "Should add pending task successfully") | |
| 
 | |
| 	// Check load increased | |
| 	disks = topology.GetNodeDisks("10.0.0.1:8080") | |
| 	targetDisk = findDiskByID(disks, 0) | |
| 	assert.Equal(t, 1, targetDisk.LoadCount) | |
| 
 | |
| 	// Add another task to same disk | |
| 	err = topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     "task2", | |
| 		TaskType:   TaskTypeVacuum, | |
| 		VolumeID:   1002, | |
| 		VolumeSize: 0, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 0}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "", DiskID: 0}, // Vacuum doesn't have a destination | |
| 		}, | |
| 	}) | |
| 	assert.NoError(t, err, "Should add vacuum task successfully") | |
| 
 | |
| 	disks = topology.GetNodeDisks("10.0.0.1:8080") | |
| 	targetDisk = findDiskByID(disks, 0) | |
| 	assert.Equal(t, 2, targetDisk.LoadCount) | |
| 
 | |
| 	// Move one task to assigned | |
| 	topology.AssignTask("task1") | |
| 
 | |
| 	// Load should still be 2 (1 pending + 1 assigned) | |
| 	disks = topology.GetNodeDisks("10.0.0.1:8080") | |
| 	targetDisk = findDiskByID(disks, 0) | |
| 	assert.Equal(t, 2, targetDisk.LoadCount) | |
| 
 | |
| 	// Complete one task | |
| 	topology.CompleteTask("task1") | |
| 
 | |
| 	// Load should decrease to 1 | |
| 	disks = topology.GetNodeDisks("10.0.0.1:8080") | |
| 	targetDisk = findDiskByID(disks, 0) | |
| 	assert.Equal(t, 1, targetDisk.LoadCount) | |
| } | |
| 
 | |
| // TestTaskConflictDetection tests task conflict detection | |
| func TestTaskConflictDetection(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	// Add a balance task | |
| 	err := topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     "balance1", | |
| 		TaskType:   TaskTypeBalance, | |
| 		VolumeID:   1001, | |
| 		VolumeSize: 1024 * 1024 * 1024, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 0}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "10.0.0.2:8080", DiskID: 1}, | |
| 		}, | |
| 	}) | |
| 	assert.NoError(t, err, "Should add balance task successfully") | |
| 	topology.AssignTask("balance1") | |
| 
 | |
| 	// Try to get available disks for vacuum (conflicts with balance) | |
| 	availableDisks := topology.GetAvailableDisks(TaskTypeVacuum, "") | |
| 
 | |
| 	// Source disk should not be available due to conflict | |
| 	sourceDiskAvailable := false | |
| 	for _, disk := range availableDisks { | |
| 		if disk.NodeID == "10.0.0.1:8080" && disk.DiskID == 0 { | |
| 			sourceDiskAvailable = true | |
| 			break | |
| 		} | |
| 	} | |
| 	assert.False(t, sourceDiskAvailable, "Source disk should not be available due to task conflict") | |
| } | |
| 
 | |
| // TestPublicInterfaces tests the public interface methods | |
| func TestPublicInterfaces(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	// Test GetAllNodes | |
| 	nodes := topology.GetAllNodes() | |
| 	assert.Equal(t, 2, len(nodes)) | |
| 	assert.Contains(t, nodes, "10.0.0.1:8080") | |
| 	assert.Contains(t, nodes, "10.0.0.2:8080") | |
| 
 | |
| 	// Test GetNodeDisks | |
| 	disks := topology.GetNodeDisks("10.0.0.1:8080") | |
| 	assert.Equal(t, 2, len(disks)) | |
| 
 | |
| 	// Test with non-existent node | |
| 	disks = topology.GetNodeDisks("non-existent") | |
| 	assert.Nil(t, disks) | |
| } | |
| 
 | |
| // Helper functions to create test topologies | |
|  | |
| func createSampleTopology() *master_pb.TopologyInfo { | |
| 	return &master_pb.TopologyInfo{ | |
| 		DataCenterInfos: []*master_pb.DataCenterInfo{ | |
| 			{ | |
| 				Id: "dc1", | |
| 				RackInfos: []*master_pb.RackInfo{ | |
| 					{ | |
| 						Id: "rack1", | |
| 						DataNodeInfos: []*master_pb.DataNodeInfo{ | |
| 							{ | |
| 								Id: "10.0.0.1:8080", | |
| 								DiskInfos: map[string]*master_pb.DiskInfo{ | |
| 									"hdd": {DiskId: 0, VolumeCount: 10, MaxVolumeCount: 100}, | |
| 									"ssd": {DiskId: 1, VolumeCount: 5, MaxVolumeCount: 50}, | |
| 								}, | |
| 							}, | |
| 							{ | |
| 								Id: "10.0.0.2:8080", | |
| 								DiskInfos: map[string]*master_pb.DiskInfo{ | |
| 									"hdd": {DiskId: 0, VolumeCount: 8, MaxVolumeCount: 100}, | |
| 									"ssd": {DiskId: 1, VolumeCount: 3, MaxVolumeCount: 50}, | |
| 								}, | |
| 							}, | |
| 						}, | |
| 					}, | |
| 				}, | |
| 			}, | |
| 		}, | |
| 	} | |
| } | |
| 
 | |
| func createEmptyTopology() *master_pb.TopologyInfo { | |
| 	return &master_pb.TopologyInfo{ | |
| 		DataCenterInfos: []*master_pb.DataCenterInfo{ | |
| 			{ | |
| 				Id: "dc1", | |
| 				RackInfos: []*master_pb.RackInfo{ | |
| 					{ | |
| 						Id: "rack1", | |
| 						DataNodeInfos: []*master_pb.DataNodeInfo{ | |
| 							{ | |
| 								Id: "10.0.0.1:8080", | |
| 								DiskInfos: map[string]*master_pb.DiskInfo{ | |
| 									"hdd": {DiskId: 0, VolumeCount: 0, MaxVolumeCount: 100}, | |
| 								}, | |
| 							}, | |
| 						}, | |
| 					}, | |
| 				}, | |
| 			}, | |
| 		}, | |
| 	} | |
| } | |
| 
 | |
| func createUnbalancedTopology() *master_pb.TopologyInfo { | |
| 	return &master_pb.TopologyInfo{ | |
| 		DataCenterInfos: []*master_pb.DataCenterInfo{ | |
| 			{ | |
| 				Id: "dc1", | |
| 				RackInfos: []*master_pb.RackInfo{ | |
| 					{ | |
| 						Id: "rack1", | |
| 						DataNodeInfos: []*master_pb.DataNodeInfo{ | |
| 							{ | |
| 								Id: "10.0.0.1:8080", | |
| 								DiskInfos: map[string]*master_pb.DiskInfo{ | |
| 									"hdd": {DiskId: 0, VolumeCount: 90, MaxVolumeCount: 100}, // Very loaded | |
| 								}, | |
| 							}, | |
| 							{ | |
| 								Id: "10.0.0.2:8080", | |
| 								DiskInfos: map[string]*master_pb.DiskInfo{ | |
| 									"hdd": {DiskId: 0, VolumeCount: 10, MaxVolumeCount: 100}, // Lightly loaded | |
| 								}, | |
| 							}, | |
| 						}, | |
| 					}, | |
| 				}, | |
| 			}, | |
| 		}, | |
| 	} | |
| } | |
| 
 | |
| func createHighGarbageTopology() *master_pb.TopologyInfo { | |
| 	// In a real implementation, this would include volume-level garbage metrics | |
| 	return createSampleTopology() | |
| } | |
| 
 | |
| func createLargeVolumeTopology() *master_pb.TopologyInfo { | |
| 	// In a real implementation, this would include volume-level size metrics | |
| 	return createSampleTopology() | |
| } | |
| 
 | |
| func createTopologyWithLoad() *ActiveTopology { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	// Add some existing tasks to create load | |
| 	err := topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     "existing1", | |
| 		TaskType:   TaskTypeVacuum, | |
| 		VolumeID:   2001, | |
| 		VolumeSize: 0, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 0}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "", DiskID: 0}, // Vacuum doesn't have a destination | |
| 		}, | |
| 	}) | |
| 	if err != nil { | |
| 		// In test helper function, just log error instead of failing | |
| 		fmt.Printf("Warning: Failed to add existing task: %v\n", err) | |
| 	} | |
| 	topology.AssignTask("existing1") | |
| 
 | |
| 	return topology | |
| } | |
| 
 | |
| func createTopologyForEC() *ActiveTopology { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 	return topology | |
| } | |
| 
 | |
| func createTopologyWithConflicts() *ActiveTopology { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	// Add conflicting tasks | |
| 	err := topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     "balance1", | |
| 		TaskType:   TaskTypeBalance, | |
| 		VolumeID:   3001, | |
| 		VolumeSize: 1024 * 1024 * 1024, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 0}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "10.0.0.2:8080", DiskID: 0}, | |
| 		}, | |
| 	}) | |
| 	if err != nil { | |
| 		fmt.Printf("Warning: Failed to add balance task: %v\n", err) | |
| 	} | |
| 	topology.AssignTask("balance1") | |
| 
 | |
| 	err = topology.AddPendingTask(TaskSpec{ | |
| 		TaskID:     "ec1", | |
| 		TaskType:   TaskTypeErasureCoding, | |
| 		VolumeID:   3002, | |
| 		VolumeSize: 1024 * 1024 * 1024, | |
| 		Sources: []TaskSourceSpec{ | |
| 			{ServerID: "10.0.0.1:8080", DiskID: 1}, | |
| 		}, | |
| 		Destinations: []TaskDestinationSpec{ | |
| 			{ServerID: "", DiskID: 0}, // EC doesn't have single destination | |
| 		}, | |
| 	}) | |
| 	if err != nil { | |
| 		fmt.Printf("Warning: Failed to add EC task: %v\n", err) | |
| 	} | |
| 	topology.AssignTask("ec1") | |
| 
 | |
| 	return topology | |
| } | |
| 
 | |
| // TestDestinationPlanning tests that the public interface works correctly | |
| // NOTE: Destination planning is now done in task detection phase, not in ActiveTopology | |
| func TestDestinationPlanning(t *testing.T) { | |
| 	topology := NewActiveTopology(10) | |
| 	topology.UpdateTopology(createSampleTopology()) | |
| 
 | |
| 	// Test that GetAvailableDisks works for destination planning | |
| 	t.Run("GetAvailableDisks functionality", func(t *testing.T) { | |
| 		availableDisks := topology.GetAvailableDisks(TaskTypeBalance, "10.0.0.1:8080") | |
| 		assert.Greater(t, len(availableDisks), 0) | |
| 
 | |
| 		// Should exclude the source node | |
| 		for _, disk := range availableDisks { | |
| 			assert.NotEqual(t, "10.0.0.1:8080", disk.NodeID) | |
| 		} | |
| 	}) | |
| 
 | |
| 	// Test that topology state can be used for planning | |
| 	t.Run("Topology provides planning information", func(t *testing.T) { | |
| 		topologyInfo := topology.GetTopologyInfo() | |
| 		assert.NotNil(t, topologyInfo) | |
| 		assert.Greater(t, len(topologyInfo.DataCenterInfos), 0) | |
| 
 | |
| 		// Test getting node disks | |
| 		disks := topology.GetNodeDisks("10.0.0.1:8080") | |
| 		assert.Greater(t, len(disks), 0) | |
| 	}) | |
| }
 |