fix(flash): correct the log information

with: #1000275887

Signed-off-by: clinx <chenlin1@oppo.com>
(cherry picked from commit 25b958ca13)

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"

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)
+}

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
 }
 

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) {

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))
+}

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

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

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 {

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
 	}

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 {

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) {

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
 	}