Fix Parse binlog path failure (#18584)

Signed-off-by: yun.zhang <yun.zhang@zilliz.com>

internal/datacoord/garbage_collector.go

@@ -21,6 +21,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/milvus-io/milvus/internal/common"
+
 	"github.com/milvus-io/milvus/internal/log"
 	"github.com/milvus-io/milvus/internal/proto/commonpb"
 	"github.com/milvus-io/milvus/internal/proto/datapb"
@@ -139,7 +141,7 @@ func (gc *garbageCollector) scan() {
 			}
 
 			segmentID, err := storage.ParseSegmentIDByBinlog(gc.option.rootPath, infoKey)
-			if err != nil {
+			if err != nil && !common.IsIgnorableError(err) {
 				log.Error("parse segment id error", zap.String("infoKey", infoKey), zap.Error(err))
 				continue
 			}

internal/storage/binlog_util.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+
+	"github.com/milvus-io/milvus/internal/common"
 )
 
 // ParseSegmentIDByBinlog parse segment id from binlog paths
@@ -22,8 +24,12 @@ func ParseSegmentIDByBinlog(rootPath, path string) (UniqueID, error) {
 
 	// binlog path should consist of "[log_type]/collID/partID/segID/fieldID/fileName"
 	keyStr := strings.Split(p, "/")
-	if len(keyStr) != 6 {
-		return 0, fmt.Errorf("%s is not a valid binlog path", path)
+	if len(keyStr) == 5 {
+		return 0, common.NewIgnorableError(fmt.Errorf("%s does not contains a file name", path))
 	}
-	return strconv.ParseInt(keyStr[len(keyStr)-3], 10, 64)
+
+	if len(keyStr) == 6 {
+		return strconv.ParseInt(keyStr[len(keyStr)-3], 10, 64)
+	}
+	return 0, fmt.Errorf("%s is not a valid binlog path", path)
 }

internal/storage/binlog_util_test.go

@@ -3,17 +3,20 @@ package storage
 import (
 	"testing"
 
+	"github.com/milvus-io/milvus/internal/common"
+
 	"github.com/stretchr/testify/assert"
 )
 
 func TestParseSegmentIDByBinlog(t *testing.T) {
 
 	type testCase struct {
-		name        string
-		input       string
-		rootPath    string
-		expectError bool
-		expectID    UniqueID
+		name             string
+		input            string
+		rootPath         string
+		expectError      bool
+		expectID         UniqueID
+		isIgnorableError bool
 	}
 
 	cases := []testCase{
@@ -49,6 +52,12 @@ func TestParseSegmentIDByBinlog(t *testing.T) {
 			rootPath:    "files",
 			expectError: true,
 		},
+		{
+			name:        "file name doesn't exists",
+			input:       "tenant1/files/delta_log/609/610/457/793",
+			rootPath:    "tenant1/files",
+			expectError: true,
+		},
 	}
 
 	for _, tc := range cases {
@@ -56,6 +65,9 @@ func TestParseSegmentIDByBinlog(t *testing.T) {
 			id, err := ParseSegmentIDByBinlog(tc.rootPath, tc.input)
 			if tc.expectError {
 				assert.Error(t, err)
+				if tc.isIgnorableError {
+					assert.True(t, common.IsIgnorableError(err))
+				}
 			} else {
 				assert.NoError(t, err)
 				assert.Equal(t, tc.expectID, id)