pending_operations_test.go 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. package maintenance
  2. import (
  3. "testing"
  4. "time"
  5. "github.com/seaweedfs/seaweedfs/weed/worker/types"
  6. )
  7. func TestPendingOperations_ConflictDetection(t *testing.T) {
  8. pendingOps := NewPendingOperations()
  9. // Add a pending erasure coding operation on volume 123
  10. op := &PendingOperation{
  11. VolumeID: 123,
  12. OperationType: OpTypeErasureCoding,
  13. SourceNode: "node1",
  14. TaskID: "task-001",
  15. StartTime: time.Now(),
  16. EstimatedSize: 1024 * 1024 * 1024, // 1GB
  17. Collection: "test",
  18. Status: "assigned",
  19. }
  20. pendingOps.AddOperation(op)
  21. // Test conflict detection
  22. if !pendingOps.HasPendingOperationOnVolume(123) {
  23. t.Errorf("Expected volume 123 to have pending operation")
  24. }
  25. if !pendingOps.WouldConflictWithPending(123, OpTypeVacuum) {
  26. t.Errorf("Expected conflict when trying to add vacuum operation on volume 123")
  27. }
  28. if pendingOps.HasPendingOperationOnVolume(124) {
  29. t.Errorf("Expected volume 124 to have no pending operation")
  30. }
  31. if pendingOps.WouldConflictWithPending(124, OpTypeVacuum) {
  32. t.Errorf("Expected no conflict for volume 124")
  33. }
  34. }
  35. func TestPendingOperations_CapacityProjection(t *testing.T) {
  36. pendingOps := NewPendingOperations()
  37. // Add operation moving volume from node1 to node2
  38. op1 := &PendingOperation{
  39. VolumeID: 100,
  40. OperationType: OpTypeVolumeMove,
  41. SourceNode: "node1",
  42. DestNode: "node2",
  43. TaskID: "task-001",
  44. StartTime: time.Now(),
  45. EstimatedSize: 2 * 1024 * 1024 * 1024, // 2GB
  46. Collection: "test",
  47. Status: "in_progress",
  48. }
  49. // Add operation moving volume from node3 to node1
  50. op2 := &PendingOperation{
  51. VolumeID: 101,
  52. OperationType: OpTypeVolumeMove,
  53. SourceNode: "node3",
  54. DestNode: "node1",
  55. TaskID: "task-002",
  56. StartTime: time.Now(),
  57. EstimatedSize: 1 * 1024 * 1024 * 1024, // 1GB
  58. Collection: "test",
  59. Status: "assigned",
  60. }
  61. pendingOps.AddOperation(op1)
  62. pendingOps.AddOperation(op2)
  63. // Test capacity impact for node1
  64. incoming, outgoing := pendingOps.GetPendingCapacityImpactForNode("node1")
  65. expectedIncoming := uint64(1 * 1024 * 1024 * 1024) // 1GB incoming
  66. expectedOutgoing := uint64(2 * 1024 * 1024 * 1024) // 2GB outgoing
  67. if incoming != expectedIncoming {
  68. t.Errorf("Expected incoming capacity %d, got %d", expectedIncoming, incoming)
  69. }
  70. if outgoing != expectedOutgoing {
  71. t.Errorf("Expected outgoing capacity %d, got %d", expectedOutgoing, outgoing)
  72. }
  73. // Test projection for node1
  74. currentUsed := uint64(10 * 1024 * 1024 * 1024) // 10GB current
  75. totalCapacity := uint64(50 * 1024 * 1024 * 1024) // 50GB total
  76. projection := pendingOps.GetNodeCapacityProjection("node1", currentUsed, totalCapacity)
  77. expectedProjectedUsed := currentUsed + incoming - outgoing // 10 + 1 - 2 = 9GB
  78. expectedProjectedFree := totalCapacity - expectedProjectedUsed // 50 - 9 = 41GB
  79. if projection.ProjectedUsed != expectedProjectedUsed {
  80. t.Errorf("Expected projected used %d, got %d", expectedProjectedUsed, projection.ProjectedUsed)
  81. }
  82. if projection.ProjectedFree != expectedProjectedFree {
  83. t.Errorf("Expected projected free %d, got %d", expectedProjectedFree, projection.ProjectedFree)
  84. }
  85. }
  86. func TestPendingOperations_VolumeFiltering(t *testing.T) {
  87. pendingOps := NewPendingOperations()
  88. // Create volume metrics
  89. metrics := []*types.VolumeHealthMetrics{
  90. {VolumeID: 100, Server: "node1"},
  91. {VolumeID: 101, Server: "node2"},
  92. {VolumeID: 102, Server: "node3"},
  93. {VolumeID: 103, Server: "node1"},
  94. }
  95. // Add pending operations on volumes 101 and 103
  96. op1 := &PendingOperation{
  97. VolumeID: 101,
  98. OperationType: OpTypeVacuum,
  99. SourceNode: "node2",
  100. TaskID: "task-001",
  101. StartTime: time.Now(),
  102. EstimatedSize: 1024 * 1024 * 1024,
  103. Status: "in_progress",
  104. }
  105. op2 := &PendingOperation{
  106. VolumeID: 103,
  107. OperationType: OpTypeErasureCoding,
  108. SourceNode: "node1",
  109. TaskID: "task-002",
  110. StartTime: time.Now(),
  111. EstimatedSize: 2 * 1024 * 1024 * 1024,
  112. Status: "assigned",
  113. }
  114. pendingOps.AddOperation(op1)
  115. pendingOps.AddOperation(op2)
  116. // Filter metrics
  117. filtered := pendingOps.FilterVolumeMetricsExcludingPending(metrics)
  118. // Should only have volumes 100 and 102 (101 and 103 are filtered out)
  119. if len(filtered) != 2 {
  120. t.Errorf("Expected 2 filtered metrics, got %d", len(filtered))
  121. }
  122. // Check that correct volumes remain
  123. foundVolumes := make(map[uint32]bool)
  124. for _, metric := range filtered {
  125. foundVolumes[metric.VolumeID] = true
  126. }
  127. if !foundVolumes[100] || !foundVolumes[102] {
  128. t.Errorf("Expected volumes 100 and 102 to remain after filtering")
  129. }
  130. if foundVolumes[101] || foundVolumes[103] {
  131. t.Errorf("Expected volumes 101 and 103 to be filtered out")
  132. }
  133. }
  134. func TestPendingOperations_OperationLifecycle(t *testing.T) {
  135. pendingOps := NewPendingOperations()
  136. // Add operation
  137. op := &PendingOperation{
  138. VolumeID: 200,
  139. OperationType: OpTypeVolumeBalance,
  140. SourceNode: "node1",
  141. DestNode: "node2",
  142. TaskID: "task-balance-001",
  143. StartTime: time.Now(),
  144. EstimatedSize: 1024 * 1024 * 1024,
  145. Status: "assigned",
  146. }
  147. pendingOps.AddOperation(op)
  148. // Check it exists
  149. if !pendingOps.HasPendingOperationOnVolume(200) {
  150. t.Errorf("Expected volume 200 to have pending operation")
  151. }
  152. // Update status
  153. pendingOps.UpdateOperationStatus("task-balance-001", "in_progress")
  154. retrievedOp := pendingOps.GetPendingOperationOnVolume(200)
  155. if retrievedOp == nil {
  156. t.Errorf("Expected to retrieve pending operation for volume 200")
  157. } else if retrievedOp.Status != "in_progress" {
  158. t.Errorf("Expected operation status to be 'in_progress', got '%s'", retrievedOp.Status)
  159. }
  160. // Complete operation
  161. pendingOps.RemoveOperation("task-balance-001")
  162. if pendingOps.HasPendingOperationOnVolume(200) {
  163. t.Errorf("Expected volume 200 to have no pending operation after removal")
  164. }
  165. }
  166. func TestPendingOperations_StaleCleanup(t *testing.T) {
  167. pendingOps := NewPendingOperations()
  168. // Add recent operation
  169. recentOp := &PendingOperation{
  170. VolumeID: 300,
  171. OperationType: OpTypeVacuum,
  172. SourceNode: "node1",
  173. TaskID: "task-recent",
  174. StartTime: time.Now(),
  175. EstimatedSize: 1024 * 1024 * 1024,
  176. Status: "in_progress",
  177. }
  178. // Add stale operation (24 hours ago)
  179. staleOp := &PendingOperation{
  180. VolumeID: 301,
  181. OperationType: OpTypeErasureCoding,
  182. SourceNode: "node2",
  183. TaskID: "task-stale",
  184. StartTime: time.Now().Add(-24 * time.Hour),
  185. EstimatedSize: 2 * 1024 * 1024 * 1024,
  186. Status: "in_progress",
  187. }
  188. pendingOps.AddOperation(recentOp)
  189. pendingOps.AddOperation(staleOp)
  190. // Clean up operations older than 1 hour
  191. removedCount := pendingOps.CleanupStaleOperations(1 * time.Hour)
  192. if removedCount != 1 {
  193. t.Errorf("Expected to remove 1 stale operation, removed %d", removedCount)
  194. }
  195. // Recent operation should still exist
  196. if !pendingOps.HasPendingOperationOnVolume(300) {
  197. t.Errorf("Expected recent operation on volume 300 to still exist")
  198. }
  199. // Stale operation should be removed
  200. if pendingOps.HasPendingOperationOnVolume(301) {
  201. t.Errorf("Expected stale operation on volume 301 to be removed")
  202. }
  203. }