diff --git a/proto/distributed_cache.go b/proto/distributed_cache.go index fe66d26fe..9bcfb954d 100644 --- a/proto/distributed_cache.go +++ b/proto/distributed_cache.go @@ -61,9 +61,9 @@ var ( ErrorResultCodeNOKTpl = "ResultCode NOK (%v)" ErrorUnknownOpcodeTpl = "unknown Opcode:%d" ErrorTaskIDNotExistTpl = "task id(%v) not exist" - ErrorBlockAlreadyExistsTpl = "block %v already exist" + ErrorBlockAlreadyExistsTpl = "block %v already exist expireTime(%v)" ErrorCreateBlockFailedTpl = "create block(%v) error %v" - ErrorBlockAlreadyCreatedTpl = "block(%v) already created" + ErrorBlockAlreadyCreatedTpl = "block(%v) already created expireTime(%v)" ErrorPutDataLengthInvalidTpl = "put data length %v is leq 0 or gt 4M" ErrorInconsistentCRCTpl = "inconsistent CRC, expect(%v) reply(%v)" ErrorExpectedReadBytesMismatchTpl = "expected to read %d bytes, but only read %d" diff --git a/remotecache/flashgroupmanager/admin_task_manager_test.go b/remotecache/flashgroupmanager/admin_task_manager_test.go new file mode 100644 index 000000000..0b33ad334 --- /dev/null +++ b/remotecache/flashgroupmanager/admin_task_manager_test.go @@ -0,0 +1,431 @@ +package flashgroupmanager + +import ( + "encoding/json" + "fmt" + "net" + "testing" + "time" + + "github.com/cubefs/cubefs/proto" + "github.com/stretchr/testify/assert" +) + +// This file contains test cases for the syncSendAdminTask method of AdminTaskManager +// Tests cover various network scenarios, error conditions, and edge cases +// Uses real TCP connections to simulate network environment, ensuring test authenticity + +func TestAdminTaskManager_syncSendAdminTask(t *testing.T) { + // Test various scenarios of the syncSendAdminTask method + // Including successful sending, connection failure, task build failure, response errors, etc. + tests := []struct { + name string + task *proto.AdminTask + expectError bool + setupMock func() (net.Listener, string) + }{ + { + name: "successful_send", // Test successful task sending + task: &proto.AdminTask{ + ID: "test_task_1", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: "test_request", + }, + expectError: false, + setupMock: func() (net.Listener, string) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + + go func() { + // Simulate server side: accept connection and process request + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + // Read data packet sent by client + packet := proto.NewPacket() + if err := packet.ReadFromConnWithVer(conn, proto.SyncSendTaskDeadlineTime); err != nil { + return + } + + // Send success response to client + responsePacket := proto.NewPacket() + responsePacket.ResultCode = proto.OpOk + responsePacket.Data = []byte("success") + responsePacket.WriteToConn(conn) + }() + + return listener, listener.Addr().String() + }, + }, + { + name: "connection_failed", // Test connection failure: use invalid port to simulate network connection failure + task: &proto.AdminTask{ + ID: "test_task_2", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:9999", // Invalid port + Request: "test_request", + }, + expectError: true, + setupMock: func() (net.Listener, string) { + return nil, "127.0.0.1:9999" + }, + }, + { + name: "task_build_failed", // Test task build failure: use channel type to cause JSON serialization failure + task: &proto.AdminTask{ + ID: "test_task_3", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: make(chan int), // This will cause JSON marshal to fail + }, + expectError: true, + setupMock: func() (net.Listener, string) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + return listener, listener.Addr().String() + }, + }, + { + name: "response_error_code", // Test response error code: simulate server returning error response code + task: &proto.AdminTask{ + ID: "test_task_4", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: "test_request", + }, + expectError: true, + setupMock: func() (net.Listener, string) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + + go func() { + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + // Read the packet + packet := proto.NewPacket() + if err := packet.ReadFromConnWithVer(conn, proto.SyncSendTaskDeadlineTime); err != nil { + return + } + + // Send error response + responsePacket := proto.NewPacket() + responsePacket.ResultCode = proto.ErrCodeParamError + responsePacket.Data = []byte("error") + responsePacket.WriteToConn(conn) + }() + + return listener, listener.Addr().String() + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Set up mock network environment for each test case + var listener net.Listener + var targetAddr string + + if tt.setupMock != nil { + listener, targetAddr = tt.setupMock() + if listener != nil { + defer listener.Close() + } + } + + // Create task manager instance + manager := newAdminTaskManager(targetAddr, "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) // Wait for goroutine to exit + }() + // Execute synchronous task sending test + packet, err := manager.syncSendAdminTask(tt.task) + + // Verify test results based on expected results + if tt.expectError { + assert.Error(t, err) // Expect error to occur + } else { + assert.NoError(t, err) // Expect no error + assert.NotNil(t, packet) // Expect valid packet to be returned + assert.Equal(t, proto.OpOk, packet.ResultCode) // Expect operation to succeed + } + }) + } +} + +func TestAdminTaskManager_syncSendAdminTask_WriteError(t *testing.T) { + // Test network write error scenario + // Simulate connection being closed immediately after establishment, causing write failure + task := &proto.AdminTask{ + ID: "test_task_write_error", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: "test_request", + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + defer listener.Close() + + targetAddr := listener.Addr().String() + + go func() { + // Simulate server side: accept connection and close immediately to simulate network write error + conn, err := listener.Accept() + if err != nil { + return + } + // Close connection immediately to cause write error + conn.Close() + }() + + manager := newAdminTaskManager(targetAddr, "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) + }() + packet, err := manager.syncSendAdminTask(task) + + assert.Error(t, err) + assert.Nil(t, packet) +} + +func TestAdminTaskManager_syncSendAdminTask_ReadError(t *testing.T) { + // Test network read error scenario + // Simulate server sending invalid data, causing read failure + task := &proto.AdminTask{ + ID: "test_task_read_error", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: "test_request", + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + defer listener.Close() + + targetAddr := listener.Addr().String() + + go func() { + // Simulate server side: accept connection and send invalid data to simulate read error + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + // Send invalid data to cause read error + conn.Write([]byte("invalid_packet_data")) + }() + + manager := newAdminTaskManager(targetAddr, "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) + }() + + packet, err := manager.syncSendAdminTask(task) + + assert.Error(t, err) + assert.Nil(t, packet) +} + +func TestAdminTaskManager_syncSendAdminTask_Timeout(t *testing.T) { + // Test network timeout scenario + // Simulate server not responding, causing read timeout + task := &proto.AdminTask{ + ID: "test_task_timeout", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: "test_request", + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + defer listener.Close() + + targetAddr := listener.Addr().String() + + go func() { + // Simulate server side: accept connection and don't respond to simulate network timeout + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + // Don't send any response to cause timeout + time.Sleep(proto.SyncSendTaskDeadlineTime + time.Second) + }() + + manager := newAdminTaskManager(targetAddr, "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) + }() + + // Note: useConnPool is a constant, so we can't modify it in tests + // The test will use the default connection pool behavior + + packet, err := manager.syncSendAdminTask(task) + + assert.Error(t, err) + assert.Nil(t, packet) +} + +func TestAdminTaskManager_syncSendAdminTask_InvalidTask(t *testing.T) { + // Test invalid task scenario: pass nil task + // Verify method's ability to handle abnormal input + manager := newAdminTaskManager("127.0.0.1:8080", "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) + }() + + // Expected to panic or return error, as nil task cannot be processed + defer func() { + if r := recover(); r != nil { + fmt.Println("task has recover") + } + }() + + _, err := manager.syncSendAdminTask(nil) + assert.Error(t, err) +} + +func TestAdminTaskManager_syncSendAdminTask_EmptyTask(t *testing.T) { + // Test empty task scenario: all fields are zero values or empty + // Verify method's ability to handle boundary input + task := &proto.AdminTask{ + ID: "", + OpCode: 0, + OperatorAddr: "", + Request: nil, + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + defer listener.Close() + + targetAddr := listener.Addr().String() + + go func() { + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + // Read the packet + packet := proto.NewPacket() + if err := packet.ReadFromConnWithVer(conn, proto.SyncSendTaskDeadlineTime); err != nil { + return + } + + // Send success response + responsePacket := proto.NewPacket() + responsePacket.ResultCode = proto.OpOk + responsePacket.Data = []byte("success") + responsePacket.WriteToConn(conn) + }() + + manager := newAdminTaskManager(targetAddr, "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) + }() + + packet, err := manager.syncSendAdminTask(task) + + assert.NoError(t, err) + assert.NotNil(t, packet) + assert.Equal(t, proto.OpOk, packet.ResultCode) +} + +func TestAdminTaskManager_syncSendAdminTask_ComplexRequest(t *testing.T) { + // Test complex request data structure scenario + // Verify method's ability to handle complex JSON serialization and deserialization + complexRequest := map[string]interface{}{ + "string_field": "test_value", + "int_field": 123, + "bool_field": true, + "array_field": []string{"item1", "item2"}, + "nested_field": map[string]interface{}{ + "nested_key": "nested_value", + }, + } + + task := &proto.AdminTask{ + ID: "test_task_complex", + OpCode: proto.OpVersionOperation, + OperatorAddr: "127.0.0.1:8080", + Request: complexRequest, + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Failed to create listener: %v", err) + } + defer listener.Close() + + targetAddr := listener.Addr().String() + + go func() { + // Simulate server side: accept connection and validate complex data structure + conn, err := listener.Accept() + if err != nil { + return + } + defer conn.Close() + + // Read data packet sent by client + packet := proto.NewPacket() + if err := packet.ReadFromConnWithVer(conn, proto.SyncSendTaskDeadlineTime); err != nil { + return + } + + // Validate request data integrity: attempt to deserialize complex data structure + var receivedTask proto.AdminTask + if err := json.Unmarshal(packet.Data, &receivedTask); err == nil { + // Data validation successful, send success response + responsePacket := proto.NewPacket() + responsePacket.ResultCode = proto.OpOk + responsePacket.Data = []byte("success") + responsePacket.WriteToConn(conn) + } + }() + + manager := newAdminTaskManager(targetAddr, "test_cluster") + defer func() { + close(manager.exitCh) + time.Sleep(100 * time.Millisecond) + }() + + packet, err := manager.syncSendAdminTask(task) + + assert.NoError(t, err) + assert.NotNil(t, packet) + assert.Equal(t, proto.OpOk, packet.ResultCode) +} diff --git a/remotecache/flashgroupmanager/cluster.go b/remotecache/flashgroupmanager/cluster.go index d50e70f8a..5c017c90c 100644 --- a/remotecache/flashgroupmanager/cluster.go +++ b/remotecache/flashgroupmanager/cluster.go @@ -70,6 +70,9 @@ func (c *Cluster) createFlashGroup(setSlots []uint32, setWeight uint32, gradualF } func (c *Cluster) removeFlashGroup(flashGroup *FlashGroup, gradualFlag bool, step uint32) (err error) { + if log.EnableInfo() { + log.LogInfof("removeFlashGroup with fg(%v) gradualFlag(%v) step(%v)", flashGroup, gradualFlag, step) + } remainingSlotsNum := uint32(flashGroup.getSlotsCount()) - step if gradualFlag && remainingSlotsNum > 0 { err = c.flashNodeTopo.gradualRemoveFlashGroup(flashGroup, c, step) @@ -95,7 +98,9 @@ func (c *Cluster) removeAllFlashNodeFromFlashGroup(flashGroup *FlashGroup) (err } successHost = append(successHost, flashNodeHost) } - log.LogInfof("action[RemoveAllFlashNodeFromFlashGroup] flashGroup:%v successHost:%v", flashGroup.ID, successHost) + if log.EnableInfo() { + log.LogInfof("action[RemoveAllFlashNodeFromFlashGroup] flashGroup:%v successHost:%v step:%v", flashGroup.ID, successHost, flashGroup.Step) + } return } diff --git a/remotecache/flashgroupmanager/flash_group.go b/remotecache/flashgroupmanager/flash_group.go index ad3ea5fe2..39a345445 100644 --- a/remotecache/flashgroupmanager/flash_group.go +++ b/remotecache/flashgroupmanager/flash_group.go @@ -7,11 +7,14 @@ import ( "time" "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/log" ) const ( UnusedFlashNodeFlashGroupID = 0 DefaultWaitClientUpdateFgTimeSec = 65 + WaitForRecoverCount = 10 ) type flashGroupValue struct { @@ -89,42 +92,48 @@ func (fg *FlashGroup) IsLostAllFlashNode() bool { } func (fg *FlashGroup) ReduceSlot() { - reducingSlots := atomic.LoadInt32(&fg.ReducingSlots) != 0 - if reducingSlots { + if atomic.CompareAndSwapInt32(&fg.ReducingSlots, 0, 1) { return } - atomic.StoreInt32(&fg.ReducingSlots, 1) - - ticker := time.NewTicker(10 * time.Minute) - defer ticker.Stop() + if log.EnableDebug() { + log.LogDebugf("flashgroup %v is reducing slots", fg) + } go func() { + ticker := time.NewTicker(30 * time.Second) + defer func() { + atomic.StoreInt32(&fg.ReducingSlots, 0) + ticker.Stop() + }() + var i int + numToSelect := (len(fg.Slots) + 4 - 1) / 4 for { - <-ticker.C - if fg.IsLostAllFlashNode() { - fg.executeReduceSlot() - atomic.StoreInt32(&fg.SlotChanged, 1) - } else { - atomic.StoreInt32(&fg.ReducingSlots, 0) + if !fg.IsLostAllFlashNode() || len(fg.Slots) == 0 { return } + <-ticker.C + i++ + if i <= WaitForRecoverCount { + continue + } + fg.executeReduceSlot(numToSelect) + atomic.StoreInt32(&fg.SlotChanged, 1) } }() } -func (fg *FlashGroup) executeReduceSlot() { - rand.Seed(time.Now().UnixNano()) - numToSelect := len(fg.Slots) / 4 - if numToSelect == 0 { +func (fg *FlashGroup) executeReduceSlot(numToReduce int) { + if len(fg.Slots) == 0 { return } - selectedIndexes := rand.Perm(len(fg.Slots))[:numToSelect] - for i := len(selectedIndexes) - 1; i >= 0; i-- { - index := selectedIndexes[i] - // add selected slot to ReservedSlots - fg.ReservedSlots = append(fg.ReservedSlots, fg.Slots[index]) - // remove selected slot from Slots - fg.Slots = append(fg.Slots[:index], fg.Slots[index+1:]...) + rand.Seed(time.Now().UnixNano()) + fg.lock.Lock() + defer fg.lock.Unlock() + numToReduce = util.Min(numToReduce, len(fg.Slots)) + if numToReduce == 0 { + return } + fg.ReservedSlots = append(fg.ReservedSlots, fg.Slots[:numToReduce]...) + fg.Slots = fg.Slots[numToReduce:] } func (fg *FlashGroup) GetStatus() (st proto.FlashGroupStatus) { diff --git a/remotecache/flashgroupmanager/flash_group_test.go b/remotecache/flashgroupmanager/flash_group_test.go new file mode 100644 index 000000000..195be5961 --- /dev/null +++ b/remotecache/flashgroupmanager/flash_group_test.go @@ -0,0 +1,334 @@ +package flashgroupmanager + +import ( + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cubefs/cubefs/proto" + "github.com/stretchr/testify/assert" +) + +// TestNewFlashGroup tests the creation of a new FlashGroup instance +func TestNewFlashGroup(t *testing.T) { + id := uint64(123) + slots := []uint32{1, 2, 3, 4, 5} + slotStatus := proto.SlotStatus_Completed + pendingSlots := []uint32{6, 7} + step := uint32(5) + status := proto.FlashGroupStatus_Active + weight := uint32(100) + + fg := newFlashGroup(id, slots, slotStatus, pendingSlots, step, status, weight) + + assert.Equal(t, id, fg.ID) + assert.Equal(t, slots, fg.Slots) + assert.Equal(t, slotStatus, fg.SlotStatus) + assert.Equal(t, pendingSlots, fg.PendingSlots) + assert.Equal(t, step, fg.Step) + assert.Equal(t, status, fg.Status) + assert.Equal(t, weight, fg.Weight) + assert.NotNil(t, fg.flashNodes) + assert.Equal(t, 0, len(fg.flashNodes)) +} + +// TestFlashGroup_GetAdminView tests the GetAdminView method +func TestFlashGroup_GetAdminView(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + // Add some flash nodes + fn1 := &FlashNode{flashNodeValue: flashNodeValue{Addr: "node1", ZoneName: "zone1"}} + fn2 := &FlashNode{flashNodeValue: flashNodeValue{Addr: "node2", ZoneName: "zone2"}} + fg.putFlashNode(fn1) + fg.putFlashNode(fn2) + + view := fg.GetAdminView() + + assert.Equal(t, uint64(1), view.ID) + assert.Equal(t, []uint32{1, 2, 3}, view.Slots) + assert.Equal(t, proto.SlotStatus_Completed, view.SlotStatus) + assert.Equal(t, proto.FlashGroupStatus_Active, view.Status) + assert.Equal(t, uint32(100), view.Weight) + assert.Equal(t, uint32(3), view.Step) + assert.Equal(t, 2, view.FlashNodeCount) + assert.Equal(t, 2, len(view.ZoneFlashNodes)) +} + +// TestFlashGroup_ReduceSlot_AlreadyReducing tests that ReduceSlot doesn't start multiple goroutines +func TestFlashGroup_ReduceSlot_AlreadyReducing(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3, 4, 5, 6, 7, 8}, proto.SlotStatus_Completed, []uint32{}, 8, proto.FlashGroupStatus_Active, 100) + + // Set the flag to indicate reduction is already in progress + atomic.StoreInt32(&fg.ReducingSlots, 1) + + // This should return immediately without starting a new goroutine + fg.ReduceSlot() + + // Verify the flag is still set + assert.Equal(t, int32(1), atomic.LoadInt32(&fg.ReducingSlots)) +} + +// TestFlashGroup_ReduceSlot_NoSlots tests ReduceSlot behavior when there are no slots +func TestFlashGroup_ReduceSlot_NoSlots(t *testing.T) { + fg := newFlashGroup(1, []uint32{}, proto.SlotStatus_Completed, []uint32{}, 0, proto.FlashGroupStatus_Active, 100) + + // Set lost all flash nodes flag + atomic.StoreInt32(&fg.LostAllFlashNode, 1) + + // Start reduction + fg.ReduceSlot() + + // Wait a bit for the goroutine to process + time.Sleep(100 * time.Millisecond) + + // Verify no slots were processed since there were none to begin with + assert.Equal(t, 0, len(fg.Slots)) + assert.Equal(t, 0, len(fg.ReservedSlots)) +} + +// TestFlashGroup_ReduceSlot_WithSlots tests the complete ReduceSlot functionality +func TestFlashGroup_ReduceSlot_WithSlots(t *testing.T) { + // Create a flash group with 8 slots + initialSlots := []uint32{1, 2, 3, 4, 5, 6, 7, 8} + fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 8, proto.FlashGroupStatus_Active, 100) + + // Set lost all flash nodes flag to trigger reduction + atomic.StoreInt32(&fg.LostAllFlashNode, 1) + + // Start reduction + fg.ReduceSlot() + + // Wait for the first reduction cycle (20 seconds, but we'll wait less for testing) + // In a real scenario, this would take 20 seconds + time.Sleep(100 * time.Millisecond) + + // Verify that slots were moved to reserved slots + // Note: In a real scenario with 20-second intervals, this would take longer + // For testing purposes, we can verify the mechanism works by checking the structure + assert.Equal(t, 8, len(fg.Slots)+len(fg.ReservedSlots)) + + // Verify the reducing flag is set + assert.Equal(t, int32(1), atomic.LoadInt32(&fg.ReducingSlots)) +} + +// TestFlashGroup_executeReduceSlot tests the executeReduceSlot method directly +func TestFlashGroup_executeReduceSlot(t *testing.T) { + // Create a flash group with 8 slots + initialSlots := []uint32{1, 2, 3, 4, 5, 6, 7, 8} + fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 8, proto.FlashGroupStatus_Active, 100) + + initialSlotCount := len(fg.Slots) + + // Execute reduction + fg.executeReduceSlot((len(fg.Slots) + 4 - 1) / 4) + + // Verify that approximately 25% of slots were moved (8 slots -> 6 slots, 2 to reserved) + expectedRemaining := (initialSlotCount * 3) / 4 // 75% remaining + expectedReserved := initialSlotCount - expectedRemaining + + assert.Equal(t, expectedRemaining, len(fg.Slots)) + assert.Equal(t, expectedReserved, len(fg.ReservedSlots)) + + // Verify total count remains the same + assert.Equal(t, initialSlotCount, len(fg.Slots)+len(fg.ReservedSlots)) +} + +// TestFlashGroup_executeReduceSlot_EmptySlots tests executeReduceSlot with no slots +func TestFlashGroup_executeReduceSlot_EmptySlots(t *testing.T) { + fg := newFlashGroup(1, []uint32{}, proto.SlotStatus_Completed, []uint32{}, 0, proto.FlashGroupStatus_Active, 100) + + // This should not panic or cause issues + fg.executeReduceSlot((len(fg.Slots) + 4 - 1) / 4) + + assert.Equal(t, 0, len(fg.Slots)) + assert.Equal(t, 0, len(fg.ReservedSlots)) +} + +// TestFlashGroup_executeReduceSlot_SingleSlot tests executeReduceSlot with only one slot +func TestFlashGroup_executeReduceSlot_SingleSlot(t *testing.T) { + fg := newFlashGroup(1, []uint32{1}, proto.SlotStatus_Completed, []uint32{}, 1, proto.FlashGroupStatus_Active, 100) + + fg.executeReduceSlot((len(fg.Slots) + 4 - 1) / 4) + + // With 1 slot, 25% rounded up is 1, so all slots should be moved to reserved + assert.Equal(t, 0, len(fg.Slots)) + assert.Equal(t, 1, len(fg.ReservedSlots)) +} + +// TestFlashGroup_IsLostAllFlashNode tests the IsLostAllFlashNode method +func TestFlashGroup_IsLostAllFlashNode(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + // Initially should be false + assert.False(t, fg.IsLostAllFlashNode()) + + // Set the flag + atomic.StoreInt32(&fg.LostAllFlashNode, 1) + assert.True(t, fg.IsLostAllFlashNode()) + + // Clear the flag + atomic.StoreInt32(&fg.LostAllFlashNode, 0) + assert.False(t, fg.IsLostAllFlashNode()) +} + +// TestFlashGroup_GetStatus tests the GetStatus method +func TestFlashGroup_GetStatus(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + status := fg.GetStatus() + assert.Equal(t, proto.FlashGroupStatus_Active, status) + + // Change status + fg.Status = proto.FlashGroupStatus_Inactive + status = fg.GetStatus() + assert.Equal(t, proto.FlashGroupStatus_Inactive, status) +} + +// TestFlashGroup_getSlots tests the getSlots method +func TestFlashGroup_getSlots(t *testing.T) { + initialSlots := []uint32{1, 2, 3, 4, 5} + fg := newFlashGroup(1, initialSlots, proto.SlotStatus_Completed, []uint32{}, 5, proto.FlashGroupStatus_Active, 100) + + slots := fg.getSlots() + + // Should return a copy of the slots + assert.Equal(t, initialSlots, slots) + assert.NotSame(t, &initialSlots[0], &slots[0]) // Should be different memory addresses +} + +// TestFlashGroup_getSlotStatus tests the getSlotStatus method +func TestFlashGroup_getSlotStatus(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Creating, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + status := fg.getSlotStatus() + assert.Equal(t, proto.SlotStatus_Creating, status) +} + +// TestFlashGroup_getFlashNodesCount tests the getFlashNodesCount method +func TestFlashGroup_getFlashNodesCount(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + // Initially no flash nodes + assert.Equal(t, 0, fg.getFlashNodesCount()) + + // Add a flash node + fn := &FlashNode{flashNodeValue: flashNodeValue{Addr: "node1", ZoneName: "zone1"}} + fg.putFlashNode(fn) + + assert.Equal(t, 1, fg.getFlashNodesCount()) +} + +// TestFlashGroup_getSlotsCount tests the getSlotsCount method +func TestFlashGroup_getSlotsCount(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3, 4, 5}, proto.SlotStatus_Completed, []uint32{}, 5, proto.FlashGroupStatus_Active, 100) + + assert.Equal(t, 5, fg.getSlotsCount()) +} + +// TestFlashGroup_putFlashNode tests the putFlashNode method +func TestFlashGroup_putFlashNode(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + fn := &FlashNode{flashNodeValue: flashNodeValue{Addr: "node1", ZoneName: "zone1"}} + fg.putFlashNode(fn) + + assert.Equal(t, 1, len(fg.flashNodes)) + assert.Equal(t, fn, fg.flashNodes["node1"]) +} + +// TestFlashGroup_removeFlashNode tests the removeFlashNode method +func TestFlashGroup_removeFlashNode(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3}, proto.SlotStatus_Completed, []uint32{}, 3, proto.FlashGroupStatus_Active, 100) + + // Add a flash node first + fn := &FlashNode{flashNodeValue: flashNodeValue{Addr: "node1", ZoneName: "zone1"}} + fg.putFlashNode(fn) + assert.Equal(t, 1, len(fg.flashNodes)) + + // Remove it + fg.removeFlashNode("node1") + assert.Equal(t, 0, len(fg.flashNodes)) +} + +// TestFlashGroup_GetPendingSlotsCount tests the GetPendingSlotsCount method +func TestFlashGroup_GetPendingSlotsCount(t *testing.T) { + pendingSlots := []uint32{1, 2, 3} + fg := newFlashGroup(1, []uint32{4, 5, 6}, proto.SlotStatus_Completed, pendingSlots, 3, proto.FlashGroupStatus_Active, 100) + + assert.Equal(t, 3, fg.GetPendingSlotsCount()) +} + +// TestFlashGroup_argConvertFlashGroupStatus tests the argConvertFlashGroupStatus function +func TestFlashGroup_argConvertFlashGroupStatus(t *testing.T) { + // Test active = true + status := argConvertFlashGroupStatus(true) + assert.Equal(t, proto.FlashGroupStatus_Active, status) + + // Test active = false + status = argConvertFlashGroupStatus(false) + assert.Equal(t, proto.FlashGroupStatus_Inactive, status) +} + +// TestFlashGroup_NewFlashGroupFromFgv tests the NewFlashGroupFromFgv function +func TestFlashGroup_NewFlashGroupFromFgv(t *testing.T) { + fgv := flashGroupValue{ + ID: 123, + Slots: []uint32{1, 2, 3}, + SlotStatus: proto.SlotStatus_Completed, + PendingSlots: []uint32{4, 5}, + Step: 3, + Weight: 100, + Status: proto.FlashGroupStatus_Active, + } + + fg := NewFlashGroupFromFgv(fgv) + + assert.Equal(t, fgv.ID, fg.ID) + assert.Equal(t, fgv.Slots, fg.Slots) + assert.Equal(t, fgv.SlotStatus, fg.SlotStatus) + assert.Equal(t, fgv.PendingSlots, fg.PendingSlots) + assert.Equal(t, fgv.Step, fg.Step) + assert.Equal(t, fgv.Weight, fg.Weight) + assert.Equal(t, fgv.Status, fg.Status) + assert.NotNil(t, fg.flashNodes) +} + +// TestFlashGroup_ConcurrentAccess tests concurrent access to FlashGroup methods +func TestFlashGroup_ConcurrentAccess(t *testing.T) { + fg := newFlashGroup(1, []uint32{1, 2, 3, 4, 5, 6, 7, 8}, proto.SlotStatus_Completed, []uint32{}, 8, proto.FlashGroupStatus_Active, 100) + + var wg sync.WaitGroup + numGoroutines := 10 + + // Test concurrent reads + for i := 0; i < numGoroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 100; j++ { + fg.getSlots() + fg.getSlotStatus() + fg.getFlashNodesCount() + fg.getSlotsCount() + } + }() + } + + // Test concurrent writes + for i := 0; i < numGoroutines; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + fn := &FlashNode{flashNodeValue: flashNodeValue{Addr: fmt.Sprintf("node%d", id), ZoneName: fmt.Sprintf("zone%d", id)}} + fg.putFlashNode(fn) + }(i) + } + + wg.Wait() + + // Verify the structure is still consistent + assert.Equal(t, 8, len(fg.Slots)) + assert.Equal(t, numGoroutines, len(fg.flashNodes)) +} diff --git a/remotecache/flashgroupmanager/flash_node_topology.go b/remotecache/flashgroupmanager/flash_node_topology.go index f4f124306..df19897b2 100644 --- a/remotecache/flashgroupmanager/flash_node_topology.go +++ b/remotecache/flashgroupmanager/flash_node_topology.go @@ -186,11 +186,16 @@ func (t *FlashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) { if fg.GetStatus().IsActive() { hosts := fg.getFlashNodeHostsEnableAndActive() if len(hosts) == 0 { + if log.EnableInfo() { + log.LogInfof("fg(%v) lost all flashnodes", fg) + } atomic.StoreInt32(&fg.LostAllFlashNode, 1) fg.ReduceSlot() } else if len(fg.ReservedSlots) > 0 { + fg.lock.Lock() fg.Slots = append(fg.Slots, fg.ReservedSlots...) - fg.ReservedSlots = nil + fg.ReservedSlots = make([]uint32, 0) + fg.lock.Unlock() atomic.StoreInt32(&fg.LostAllFlashNode, 0) atomic.StoreInt32(&fg.SlotChanged, 1) } @@ -198,7 +203,10 @@ func (t *FlashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) { if len(fg.Slots) == 0 { disableFlashGroupNum++ if disableFlashGroupNum >= maxDisableFlashGroupCount && len(fg.ReservedSlots) > 0 { + fg.lock.Lock() fg.Slots = append(fg.Slots, fg.ReservedSlots...) + fg.ReservedSlots = make([]uint32, 0) + fg.lock.Unlock() atomic.StoreInt32(&fg.SlotChanged, 1) } else { return true diff --git a/remotecache/flashnode/cachengine/block.go b/remotecache/flashnode/cachengine/block.go index 38f65febb..3a222db74 100644 --- a/remotecache/flashnode/cachengine/block.go +++ b/remotecache/flashnode/cachengine/block.go @@ -276,7 +276,7 @@ func (cb *CacheBlock) writeCacheBlockFileHeader(file *os.File) (err error) { } func (cb *CacheBlock) GetCreateTime() time.Time { - createTime := time.Now() + createTime := time.Time{} if value, ok := cb.cacheEngine.lruCacheMap.Load(cb.rootPath); ok { cacheItem := value.(*lruCacheItem) if blockCreateTime, exist := cacheItem.lruCache.GetCreateTime(cb.blockKey); exist { @@ -286,6 +286,17 @@ func (cb *CacheBlock) GetCreateTime() time.Time { return createTime } +func (cb *CacheBlock) GetExpiredTime() time.Time { + expiredTime := time.Time{} + if value, ok := cb.cacheEngine.lruCacheMap.Load(cb.rootPath); ok { + cacheItem := value.(*lruCacheItem) + if blockExpiredTime, exist := cacheItem.lruCache.GetExpiredTime(cb.blockKey); exist { + expiredTime = blockExpiredTime + } + } + return expiredTime +} + func (cb *CacheBlock) checkCacheBlockFileHeader(file *os.File, sourceType string) (allocSize, usedSize int64, expiredTime time.Time, err error) { var stat os.FileInfo var seconds int64 diff --git a/remotecache/flashnode/cachengine/miss_cache.go b/remotecache/flashnode/cachengine/miss_cache.go index 3adc22c70..5d21439ed 100644 --- a/remotecache/flashnode/cachengine/miss_cache.go +++ b/remotecache/flashnode/cachengine/miss_cache.go @@ -117,7 +117,9 @@ func (mc *MissCache) backgroundEviction() { } func (mc *MissCache) Delete(uniKey string) { - log.LogDebugf("MissCache Delete: key(%v)", uniKey) + if log.EnableDebug() { + log.LogDebugf("MissCache Delete: key(%v)", uniKey) + } mc.Lock() element, ok := mc.cache[uniKey] if ok { diff --git a/remotecache/flashnode/flashnode.go b/remotecache/flashnode/flashnode.go index 5598fa0ff..c14e7cf5f 100644 --- a/remotecache/flashnode/flashnode.go +++ b/remotecache/flashnode/flashnode.go @@ -57,7 +57,7 @@ const ( _connPoolIdleTimeout = 60 // 60s _extentReadMaxRetry = 3 _defaultDiskWriteIOCC = 64 - _defaultDiskWriteFlow = _defaultDiskReadFlow / 2 + _defaultDiskWriteFlow = _defaultDiskReadFlow _defaultDiskWriteFactor = 8 _defaultDiskReadIOCC = 64 _defaultDiskReadFlow = 1 * util.GB @@ -294,7 +294,7 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) { if f.diskReadIocc <= 0 { f.diskReadIocc = _defaultDiskReadIOCC } - f.diskReadFlow = cfg.GetInt(cfgDiskReadFlow) + f.diskReadFlow = int(cfg.GetInt64(cfgDiskReadFlow)) if f.diskReadFlow == 0 { f.diskReadFlow = _defaultDiskReadFlow } diff --git a/remotecache/flashnode/flashnode_op.go b/remotecache/flashnode/flashnode_op.go index 40cce791f..279489313 100644 --- a/remotecache/flashnode/flashnode_op.go +++ b/remotecache/flashnode/flashnode_op.go @@ -318,15 +318,19 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) return proto.ErrorParsingCacheWriteHead } uniKey = req.UniKey + var ep time.Time pDir := cachengine.MapKeyToDirectory(uniKey) blockKey := cachengine.GenCacheBlockKeyV2(pDir, uniKey) - _, err1 := f.cacheEngine.GetCacheBlockForReadByKey(blockKey) + eb, err1 := f.cacheEngine.GetCacheBlockForReadByKey(blockKey) if err1 == nil { if log.EnableDebug() { log.LogDebug(logPrefix+" check block key:"+uniKey+" logMsg:", p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) } - err = fmt.Errorf(proto.ErrorBlockAlreadyExistsTpl, uniKey) + if eb != nil { + ep = eb.GetExpiredTime() + } + err = fmt.Errorf(proto.ErrorBlockAlreadyExistsTpl, uniKey, ep) return err } var allocSize int @@ -338,11 +342,15 @@ func (f *FlashNode) opCachePutBlock(conn net.Conn, p *proto.Packet) (err error) log.LogDebug(logPrefix+" create block key:"+uniKey+" logMsg:", p.LogMessage(p.GetOpMsg(), conn.RemoteAddr().String(), p.StartT, err)) } + defer f.missCache.Delete(uniKey) if cb, err2, created := f.cacheEngine.CreateBlockV2(pDir, uniKey, req.TTL, uint32(allocSize), conn.RemoteAddr().String()); err2 != nil || created { if err2 != nil { err = fmt.Errorf(proto.ErrorCreateBlockFailedTpl, cachengine.GenCacheBlockKeyV2(pDir, uniKey), err2.Error()) } else { - err = fmt.Errorf(proto.ErrorBlockAlreadyCreatedTpl, cachengine.GenCacheBlockKeyV2(pDir, uniKey)) + if cb != nil { + ep = cb.GetExpiredTime() + } + err = fmt.Errorf(proto.ErrorBlockAlreadyCreatedTpl, cachengine.GenCacheBlockKeyV2(pDir, uniKey), ep) } return } else { diff --git a/remotecache/flashnode/serve_http.go b/remotecache/flashnode/serve_http.go index 98844ed38..fe133c8a6 100644 --- a/remotecache/flashnode/serve_http.go +++ b/remotecache/flashnode/serve_http.go @@ -203,8 +203,10 @@ func (f *FlashNode) handleSetWriteDiskQos(w http.ResponseWriter, r *http.Request if updated { f.limitWrite.ResetIOEx(f.diskWriteIocc*len(f.disks), f.diskWriteIoFactorFlow, f.handleReadTimeout) f.limitWrite.ResetFlow(f.diskWriteFlow) + replyOK(w, r, nil) + } else { + replyErr(w, r, http.StatusBadRequest, "request param is not an update key", nil) } - replyOK(w, r, nil) } func (f *FlashNode) handleSetReadDiskQos(w http.ResponseWriter, r *http.Request) { @@ -244,8 +246,10 @@ func (f *FlashNode) handleSetReadDiskQos(w http.ResponseWriter, r *http.Request) if updated { f.limitRead.ResetIOEx(f.diskReadIocc*len(f.disks), f.diskReadIoFactorFlow, f.handleReadTimeout) f.limitRead.ResetFlow(f.diskReadFlow) + replyOK(w, r, nil) + } else { + replyErr(w, r, http.StatusBadRequest, "request param is not an update key", nil) } - replyOK(w, r, nil) } func (f *FlashNode) handleGetDiskQos(w http.ResponseWriter, r *http.Request) { diff --git a/sdk/remotecache/client.go b/sdk/remotecache/client.go index 0b51e7832..75b19ea4b 100755 --- a/sdk/remotecache/client.go +++ b/sdk/remotecache/client.go @@ -938,7 +938,9 @@ func (rc *RemoteCacheClient) Name() string { } func (rc *RemoteCacheClient) Get(ctx context.Context, reqId, key string, from, to int64) (r io.ReadCloser, length int64, shouldCache bool, err error) { - log.LogDebugf("RemoteCacheClient Get: key(%v) reqId(%v) from(%v) to(%v)", key, reqId, from, to) + if log.EnableDebug() { + log.LogDebugf("RemoteCacheClient Get: key(%v) reqId(%v) from(%v) to(%v)", key, reqId, from, to) + } if from < 0 || to-from <= 0 { return nil, 0, false, fmt.Errorf(proto.ErrorInvalidRangeTpl, from, to) } @@ -1087,6 +1089,7 @@ type RemoteCacheReader struct { needReadLen int64 alreadyReadLen int64 reqID string + flashIp string ctx context.Context buffer []byte localData []byte @@ -1095,20 +1098,23 @@ type RemoteCacheReader struct { closed bool } -func (rc *RemoteCacheClient) NewRemoteCacheReader(ctx context.Context, conn *net.TCPConn, reqID string) *RemoteCacheReader { +func (rc *RemoteCacheClient) NewRemoteCacheReader(ctx context.Context, conn *net.TCPConn, reqID, flashIp string) *RemoteCacheReader { r := &RemoteCacheReader{ - conn: conn, - rc: rc, - reqID: reqID, - ctx: ctx, - buffer: bytespool.Alloc(proto.CACHE_BLOCK_PACKET_SIZE), - offset: 0, - size: 0, + conn: conn, + rc: rc, + reqID: reqID, + ctx: ctx, + buffer: bytespool.Alloc(proto.CACHE_BLOCK_PACKET_SIZE), + offset: 0, + size: 0, + flashIp: flashIp, } return r } func (reader *RemoteCacheReader) read(p []byte) (n int, err error) { + bg := stat.BeginStat() + defer stat.EndStat(fmt.Sprintf("readCache:%v", reader.flashIp), err, bg, 1) if reader.ctx.Err() != nil { log.LogErrorf("RemoteCacheReader:reqID(%v) err(%v)", reader.reqID, reader.ctx.Err()) return 0, reader.ctx.Err() @@ -1265,6 +1271,7 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req var ( conn *net.TCPConn addr string + flashIp string reqPacket *proto.Packet logPrefix = fmt.Sprintf("reqID[%v]", reqId) ) @@ -1273,9 +1280,8 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req if err != nil && strings.Contains(err.Error(), "timeout") { err = proto.ErrorReadTimeout } - parts := strings.Split(addr, ":") - if len(parts) > 0 && addr != "" { - stat.EndStat(fmt.Sprintf("flashNode:%v", parts[0]), err, bgTime, 1) + if len(flashIp) > 0 { + stat.EndStat(fmt.Sprintf("flashNode:%v", flashIp), err, bgTime, 1) } stat.EndStat("flashNode", err, bgTime, 1) }() @@ -1289,6 +1295,7 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req log.LogWarnf("%v FlashGroup Read failed: fg(%v) err(%v)", logPrefix, fg, err) return } + flashIp = strings.Split(addr, ":")[0] reqPacket = NewRemoteCachePacket(reqId, proto.OpFlashNodeCacheReadObject) if err = reqPacket.MarshalDataPb(req); err != nil { log.LogWarnf("%v FlashGroup Read: failed to MarshalData (%+v). err(%v)", logPrefix, req, err) @@ -1328,7 +1335,7 @@ func (rc *RemoteCacheClient) ReadObject(ctx context.Context, fg *FlashGroup, req fg.moveToUnknownRank(addr, err, rc.FlashNodeTimeoutCount) return nil, 0, err } - reader = rc.NewRemoteCacheReader(ctx, conn, reqId) + reader = rc.NewRemoteCacheReader(ctx, conn, reqId, flashIp) reader.needReadLen = int64(req.Size_) break }