From ff1f157f2029ab07832c824d2a0cf7ea2ca062be Mon Sep 17 00:00:00 2001 From: zhumingze Date: Wed, 20 Aug 2025 20:13:00 +0800 Subject: [PATCH] fix(master): add some log to locate migrate mp problem. #1000295064 Signed-off-by: zhumingze (cherry picked from commit f6359bf1e86a09a50f74cfb5a574c1c3c89a2a6b) --- master/cluster_task.go | 8 +- sdk/data/stream/async_flush_sequencer_test.go | 143 ------------------ sdk/data/stream/stream_reader.go | 34 ----- sdk/data/stream/stream_writer.go | 1 - 4 files changed, 7 insertions(+), 179 deletions(-) diff --git a/master/cluster_task.go b/master/cluster_task.go index 5c3fed3da..a3326ccaf 100644 --- a/master/cluster_task.go +++ b/master/cluster_task.go @@ -22,6 +22,7 @@ import ( "time" "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/util/auditlog" "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/log" ) @@ -105,6 +106,7 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti finalHosts []string oldHosts []string zones []string + auditMsg string ) log.LogWarnf("action[migrateMetaPartition],volName[%v], migrate from src[%s] to target[%s],partitionID[%v] begin", @@ -174,6 +176,10 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti } } + auditMsg = fmt.Sprintf("volName[%v] partitionID[%v] hosts[%v] srcAddr[%v] choose targetAddr[%v]", + mp.volName, mp.PartitionID, mp.Hosts, srcAddr, newPeers) + auditlog.LogMasterOp("migrateMetaPartition", auditMsg, err) + finalHosts = make([]string, 0, len(oldHosts)) for _, host := range oldHosts { if host != srcAddr { @@ -539,7 +545,7 @@ func (c *Cluster) addMetaReplica(partition *MetaPartition, addr string) (err err partition.Lock() defer partition.Unlock() if contains(partition.Hosts, addr) { - err = fmt.Errorf("vol[%v],mp[%v] has contains host[%v]", partition.volName, partition.PartitionID, addr) + err = fmt.Errorf("vol[%v],mp[%v] hosts[%v] has contains host[%v]", partition.volName, partition.PartitionID, partition.Hosts, addr) return } metaNode, err := c.metaNode(addr) diff --git a/sdk/data/stream/async_flush_sequencer_test.go b/sdk/data/stream/async_flush_sequencer_test.go index 39b9d64a7..4a874d3ef 100644 --- a/sdk/data/stream/async_flush_sequencer_test.go +++ b/sdk/data/stream/async_flush_sequencer_test.go @@ -1,117 +1,12 @@ package stream import ( - "sync" "testing" "github.com/cubefs/cubefs/util/log" "github.com/stretchr/testify/assert" ) -func TestAsyncFlushSequencer(t *testing.T) { - // Create a test streamer to test local sequencer - client := &ExtentClient{} - streamer := NewStreamer(client, 1, true, false, "/test") - - // Test ID generation - id1 := streamer.getNextAsyncFlushID() - id2 := streamer.getNextAsyncFlushID() - id3 := streamer.getNextAsyncFlushID() - - assert.Equal(t, uint64(1), id1) - assert.Equal(t, uint64(2), id2) - assert.Equal(t, uint64(3), id3) - - // Test pending map operations - req1 := &AsyncFlushRequest{id: id1} - req2 := &AsyncFlushRequest{id: id2} - req3 := &AsyncFlushRequest{id: id3} - - streamer.addPendingAsyncFlush(1, req1) - streamer.addPendingAsyncFlush(2, req2) - streamer.addPendingAsyncFlush(3, req3) - - // Test getNextPendingAsyncFlush returns the oldest request - next := streamer.getNextPendingAsyncFlush() - assert.Equal(t, uint64(1), next.handler.id) - - // Test removal - streamer.removePendingAsyncFlush(1) - next = streamer.getNextPendingAsyncFlush() - assert.Equal(t, uint64(2), next.handler.id) - - streamer.removePendingAsyncFlush(2) - next = streamer.getNextPendingAsyncFlush() - assert.Equal(t, uint64(3), next.handler.id) - - streamer.removePendingAsyncFlush(3) - next = streamer.getNextPendingAsyncFlush() - assert.Nil(t, next) -} - -func TestAsyncFlushSequencerConcurrency(t *testing.T) { - // Create a test streamer to test local sequencer - client := &ExtentClient{} - streamer := NewStreamer(client, 1, true, false, "/test") - - // Test concurrent ID generation - var wg sync.WaitGroup - ids := make([]uint64, 100) - - for i := 0; i < 100; i++ { - wg.Add(1) - go func(index int) { - defer wg.Done() - ids[index] = streamer.getNextAsyncFlushID() - }(i) - } - - wg.Wait() - - // Verify all IDs are unique and sequential - seen := make(map[uint64]bool) - for _, id := range ids { - assert.False(t, seen[id], "Duplicate ID found: %d", id) - seen[id] = true - assert.Greater(t, id, uint64(0), "ID should be greater than 0") - } -} - -func TestAsyncFlushSequencerStats(t *testing.T) { - // Create a test streamer to test local sequencer - client := &ExtentClient{} - streamer := NewStreamer(client, 1, true, false, "/test") - - // Test stats with no requests - stats := streamer.getAsyncFlushStats() - assert.Equal(t, uint64(0), stats["current_sequencer_id"]) - assert.Equal(t, 0, stats["pending_requests_count"]) - - // Add some requests - id1 := streamer.getNextAsyncFlushID() - id2 := streamer.getNextAsyncFlushID() - id3 := streamer.getNextAsyncFlushID() - - req1 := &AsyncFlushRequest{id: id1} - req2 := &AsyncFlushRequest{id: id2} - req3 := &AsyncFlushRequest{id: id3} - - streamer.addPendingAsyncFlush(1, req1) - streamer.addPendingAsyncFlush(2, req2) - streamer.addPendingAsyncFlush(3, req3) - - // Test statistics - stats = streamer.getAsyncFlushStats() - - assert.Equal(t, uint64(3), stats["current_sequencer_id"]) - assert.Equal(t, 3, stats["pending_requests_count"]) - - // Clean up - streamer.removePendingAsyncFlush(1) - streamer.removePendingAsyncFlush(2) - streamer.removePendingAsyncFlush(3) -} - func TestChannelCloseSafety(t *testing.T) { // Test that we can safely close channels multiple times done := make(chan struct{}) @@ -140,44 +35,6 @@ func TestChannelCloseSafety(t *testing.T) { assert.True(t, true) } -func TestDuplicateHandlerFlushPrevention(t *testing.T) { - // Test the active handler tracking functions directly - client := &ExtentClient{} - streamer := NewStreamer(client, 1, true, false, "/test") - - // Initially no active handlers - assert.False(t, streamer.isHandlerFlushActive(123)) - - // Create a mock request - req := &AsyncFlushRequest{ - id: 1, - } - - // Add handler to pending map - streamer.addPendingAsyncFlush(123, req) - assert.True(t, streamer.isHandlerFlushActive(123)) - - // Get the active request - retrievedReq := streamer.getActiveHandlerFlush(123) - assert.Equal(t, req, retrievedReq) - - // Remove handler from pending map - streamer.removePendingAsyncFlush(123) - assert.False(t, streamer.isHandlerFlushActive(123)) - - // Test with multiple handlers - streamer.addPendingAsyncFlush(456, req) - streamer.addPendingAsyncFlush(789, req) - assert.True(t, streamer.isHandlerFlushActive(456)) - assert.True(t, streamer.isHandlerFlushActive(789)) - - // Clean up - streamer.removePendingAsyncFlush(456) - streamer.removePendingAsyncFlush(789) - assert.False(t, streamer.isHandlerFlushActive(456)) - assert.False(t, streamer.isHandlerFlushActive(789)) -} - func TestAsyncFlushRaceConditionFix(t *testing.T) { // Test the race condition fix logic diff --git a/sdk/data/stream/stream_reader.go b/sdk/data/stream/stream_reader.go index 4d3d474fe..d51daef31 100644 --- a/sdk/data/stream/stream_reader.go +++ b/sdk/data/stream/stream_reader.go @@ -795,40 +795,6 @@ func (s *Streamer) waitForAsyncFlush(handler *ExtentHandler, timeout time.Durati } } -// cancelAsyncFlush cancels pending async flush for a handler -// Note: With channel-based approach, we can't easily cancel specific requests -// This is a limitation of the channel approach -func (s *Streamer) cancelAsyncFlush(handler *ExtentHandler) { - // With channel-based approach, we can't easily cancel specific requests - // The request will be processed and can check if handler is still valid - log.LogDebugf("Cancel requested for handler(%v) - will be handled during processing", handler) -} - -// handleHandlerClosed handles cleanup when a handler is closed during async flush -func (s *Streamer) handleHandlerClosed(handler *ExtentHandler) { - // With channel-based approach, we can't easily cancel specific requests - // Just ensure all in-flight packets are processed - for atomic.LoadInt32(&handler.inflight) > 0 { - time.Sleep(time.Duration(asyncFlushCheckIntervalMs) * time.Millisecond) - } - - log.LogDebugf("Handler closed cleanup completed for handler(%v)", handler) -} - -// getAsyncFlushStats returns statistics about async flush operations -func (s *Streamer) getAsyncFlushStats() map[string]interface{} { - stats := make(map[string]interface{}) - stats["channel_capacity"] = cap(s.asyncFlushCh) - stats["channel_length"] = len(s.asyncFlushCh) - stats["enable_async_flush"] = s.client.enableAsyncFlush - - // Add sequencer statistics - stats["pending_requests"] = s.getPendingRequests() - stats["current_sequencer_id"] = atomic.LoadUint64(&s.asyncFlushSequencer) - - return stats -} - // getAsyncFlushSequencerStats returns detailed statistics about the async flush sequencer // Note: This function is deprecated as sequencer is now per-streamer func (s *Streamer) getAsyncFlushSequencerStats() map[string]interface{} { diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go index df7e4e6a2..614494a24 100644 --- a/sdk/data/stream/stream_writer.go +++ b/sdk/data/stream/stream_writer.go @@ -53,7 +53,6 @@ const ( asyncFlushCheckIntervalMs = 100 // Check every 100ms maxAsyncFlushRetries = 3 // Max retries for failed async flush asyncFlushQueueSize = 128 // Buffer size for async flush channel - asyncFlushWorkerCount = 4 // Number of worker goroutines for async flush asyncFlushSemaphoreSize = 4 // Capacity of semaphore to limit concurrent processAsyncFlushRequest executions )