Skip to content

Commit c91bb0b

Browse files
congqixiacongqixia
authored
Unify rootPath in configs and ChunkManager (milvus-io#18808)
Signed-off-by: Congqi Xia <[email protected]> Signed-off-by: Congqi Xia <[email protected]>
1 parent 232f487 commit c91bb0b

20 files changed

+138
-97
lines changed

‎internal/common/common.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,3 +57,17 @@ const (
5757
// Endian is type alias of binary.LittleEndian.
5858
// Milvus uses little endian by default.
5959
var Endian = binary.LittleEndian
60+
61+
const (
62+
// SegmentInsertLogPath storage path const for segment insert binlog.
63+
SegmentInsertLogPath = `insert_log`
64+
65+
// SegmentDeltaLogPath storage path const for segment delta log.
66+
SegmentDeltaLogPath = `delta_log`
67+
68+
// SegmentStatslogPath storage path const for segment stats log.
69+
SegmentStatslogPath = `stats_log`
70+
71+
// SegmentIndexPath storage path const for segment index files.
72+
SegmentIndexPath = `index_files`
73+
)

‎internal/datanode/binlog_io.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,7 @@ func (b *binlogIO) genDeltaBlobs(data *DeleteData, collID, partID, segID UniqueI
231231
return "", nil, err
232232
}
233233

234-
key := path.Join(Params.DataNodeCfg.DeleteBinlogRootPath, k)
234+
key := path.Join(b.ChunkManager.RootPath(), common.SegmentDeltaLogPath, k)
235235

236236
return key, blob.GetValue(), nil
237237
}
@@ -262,7 +262,7 @@ func (b *binlogIO) genInsertBlobs(data *InsertData, partID, segID UniqueID, meta
262262
// Blob Key is generated by Serialize from int64 fieldID in collection schema, which won't raise error in ParseInt
263263
fID, _ := strconv.ParseInt(blob.GetKey(), 10, 64)
264264
k := JoinIDPath(meta.GetID(), partID, segID, fID, <-generator)
265-
key := path.Join(Params.DataNodeCfg.InsertBinlogRootPath, k)
265+
key := path.Join(b.ChunkManager.RootPath(), common.SegmentInsertLogPath, k)
266266

267267
value := blob.GetValue()
268268
fileLen := len(value)
@@ -279,7 +279,7 @@ func (b *binlogIO) genInsertBlobs(data *InsertData, partID, segID UniqueID, meta
279279
fID, _ := strconv.ParseInt(blob.GetKey(), 10, 64)
280280

281281
k := JoinIDPath(meta.GetID(), partID, segID, fID, <-generator)
282-
key := path.Join(Params.DataNodeCfg.StatsBinlogRootPath, k)
282+
key := path.Join(b.ChunkManager.RootPath(), common.SegmentStatslogPath, k)
283283

284284
value := blob.GetValue()
285285
fileLen := len(value)

‎internal/datanode/binlog_io_test.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -369,6 +369,10 @@ type mockCm struct {
369369

370370
var _ storage.ChunkManager = (*mockCm)(nil)
371371

372+
func (mk *mockCm) RootPath() string {
373+
return "mock_test"
374+
}
375+
372376
func (mk *mockCm) Write(filePath string, content []byte) error {
373377
return nil
374378
}

‎internal/datanode/data_node.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1114,7 +1114,7 @@ func importFlushReqFunc(node *DataNode, req *datapb.ImportTaskRequest, res *root
11141114
// no error raise if alloc=false
11151115
k := JoinIDPath(req.GetImportTask().GetCollectionId(), req.GetImportTask().GetPartitionId(), segmentID, fieldID, logidx)
11161116

1117-
key := path.Join(Params.DataNodeCfg.InsertBinlogRootPath, k)
1117+
key := path.Join(node.chunkManager.RootPath(), common.SegmentInsertLogPath, k)
11181118
kvs[key] = blob.Value[:]
11191119
field2Insert[fieldID] = &datapb.Binlog{
11201120
EntriesNum: data.size,
@@ -1140,7 +1140,7 @@ func importFlushReqFunc(node *DataNode, req *datapb.ImportTaskRequest, res *root
11401140
// no error raise if alloc=false
11411141
k := JoinIDPath(req.GetImportTask().GetCollectionId(), req.GetImportTask().GetPartitionId(), segmentID, fieldID, logidx)
11421142

1143-
key := path.Join(Params.DataNodeCfg.StatsBinlogRootPath, k)
1143+
key := path.Join(node.chunkManager.RootPath(), common.SegmentStatslogPath, k)
11441144
kvs[key] = blob.Value
11451145
field2Stats[fieldID] = &datapb.Binlog{
11461146
EntriesNum: 0,

‎internal/datanode/flow_graph_delete_node_test.go‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -313,7 +313,6 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
313313
testPath := "/test/datanode/root/meta"
314314
assert.NoError(t, clearEtcd(testPath))
315315
Params.EtcdCfg.MetaRootPath = testPath
316-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
317316

318317
c := &nodeConfig{
319318
replica: replica,
@@ -337,7 +336,6 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
337336
testPath := "/test/datanode/root/meta"
338337
assert.NoError(t, clearEtcd(testPath))
339338
Params.EtcdCfg.MetaRootPath = testPath
340-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
341339

342340
c := &nodeConfig{
343341
replica: replica,
@@ -367,7 +365,6 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
367365
testPath := "/test/datanode/root/meta"
368366
assert.NoError(t, clearEtcd(testPath))
369367
Params.EtcdCfg.MetaRootPath = testPath
370-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
371368

372369
c := &nodeConfig{
373370
replica: replica,
@@ -403,7 +400,6 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
403400
testPath := "/test/datanode/root/meta"
404401
assert.NoError(t, clearEtcd(testPath))
405402
Params.EtcdCfg.MetaRootPath = testPath
406-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
407403

408404
c := &nodeConfig{
409405
replica: &mockReplica{},
@@ -436,7 +432,6 @@ func TestFlowGraphDeleteNode_Operate(t *testing.T) {
436432
testPath := "/test/datanode/root/meta"
437433
assert.NoError(t, clearEtcd(testPath))
438434
Params.EtcdCfg.MetaRootPath = testPath
439-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
440435

441436
replica := &SegmentReplica{
442437
newSegments: make(map[UniqueID]*Segment),
@@ -492,7 +487,6 @@ func TestFlowGraphDeleteNode_showDelBuf(t *testing.T) {
492487
testPath := "/test/datanode/root/meta"
493488
assert.NoError(t, clearEtcd(testPath))
494489
Params.EtcdCfg.MetaRootPath = testPath
495-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
496490

497491
c := &nodeConfig{
498492
replica: &mockReplica{},
@@ -533,7 +527,6 @@ func TestFlowGraphDeleteNode_updateCompactedSegments(t *testing.T) {
533527
testPath := "/test/datanode/root/meta"
534528
assert.NoError(t, clearEtcd(testPath))
535529
Params.EtcdCfg.MetaRootPath = testPath
536-
Params.DataNodeCfg.DeleteBinlogRootPath = testPath
537530

538531
replica := SegmentReplica{
539532
newSegments: make(map[UniqueID]*Segment),

‎internal/datanode/flush_manager.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424
"strconv"
2525
"sync"
2626

27+
"github.com/milvus-io/milvus/internal/common"
2728
"github.com/milvus-io/milvus/internal/log"
2829
"github.com/milvus-io/milvus/internal/metrics"
2930
"github.com/milvus-io/milvus/internal/proto/commonpb"
@@ -384,7 +385,8 @@ func (m *rendezvousFlushManager) flushBufferData(data *BufferData, segmentID Uni
384385
// no error raise if alloc=false
385386
k := JoinIDPath(collID, partID, segmentID, fieldID, logidx)
386387

387-
key := path.Join(Params.DataNodeCfg.InsertBinlogRootPath, k)
388+
// [rootPath]/[insert_log]/key
389+
key := path.Join(m.ChunkManager.RootPath(), common.SegmentInsertLogPath, k)
388390
kvs[key] = blob.Value[:]
389391
field2Insert[fieldID] = &datapb.Binlog{
390392
EntriesNum: data.size,
@@ -410,7 +412,7 @@ func (m *rendezvousFlushManager) flushBufferData(data *BufferData, segmentID Uni
410412
// no error raise if alloc=false
411413
k := JoinIDPath(collID, partID, segmentID, fieldID, logidx)
412414

413-
key := path.Join(Params.DataNodeCfg.StatsBinlogRootPath, k)
415+
key := path.Join(m.ChunkManager.RootPath(), common.SegmentStatslogPath, k)
414416
kvs[key] = blob.Value
415417
field2Stats[fieldID] = &datapb.Binlog{
416418
EntriesNum: 0,
@@ -460,7 +462,7 @@ func (m *rendezvousFlushManager) flushDelData(data *DelDataBuf, segmentID Unique
460462
}
461463

462464
blobKey := JoinIDPath(collID, partID, segmentID, logID)
463-
blobPath := path.Join(Params.DataNodeCfg.DeleteBinlogRootPath, blobKey)
465+
blobPath := path.Join(m.ChunkManager.RootPath(), common.SegmentDeltaLogPath, blobKey)
464466
kvs := map[string][]byte{blobPath: blob.Value[:]}
465467
data.LogSize = int64(len(blob.Value))
466468
data.LogPath = blobPath

‎internal/indexcoord/garbage_collector.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,13 @@ package indexcoord
1818

1919
import (
2020
"context"
21+
"path"
2122
"sync"
2223
"time"
2324

2425
"go.uber.org/zap"
2526

27+
"github.com/milvus-io/milvus/internal/common"
2628
"github.com/milvus-io/milvus/internal/log"
2729
"github.com/milvus-io/milvus/internal/proto/commonpb"
2830
"github.com/milvus-io/milvus/internal/proto/datapb"
@@ -222,7 +224,7 @@ func (gc *garbageCollector) recycleUnusedIndexFiles() {
222224
case <-gc.ctx.Done():
223225
return
224226
case <-ticker.C:
225-
prefix := Params.IndexCoordCfg.IndexStorageRootPath + "/"
227+
prefix := path.Join(gc.chunkManager.RootPath(), common.SegmentIndexPath) + "/"
226228
// list dir first
227229
keys, _, err := gc.chunkManager.ListWithPrefix(prefix, false)
228230
if err != nil {

‎internal/indexcoord/index_builder.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,11 @@ package indexcoord
1818

1919
import (
2020
"context"
21+
"path"
2122
"sync"
2223
"time"
2324

25+
"github.com/milvus-io/milvus/internal/common"
2426
"github.com/milvus-io/milvus/internal/proto/datapb"
2527

2628
"go.uber.org/zap"
@@ -251,7 +253,7 @@ func (ib *indexBuilder) process(buildID UniqueID) {
251253
UseIAM: Params.MinioCfg.UseIAM,
252254
IAMEndpoint: Params.MinioCfg.IAMEndpoint,
253255
},
254-
IndexFilePrefix: Params.IndexCoordCfg.IndexStorageRootPath,
256+
IndexFilePrefix: path.Join(ib.ic.chunkManager.RootPath(), common.SegmentIndexPath),
255257
BuildID: buildID,
256258
DataPaths: binLogs,
257259
IndexVersion: meta.IndexVersion + 1,

‎internal/indexcoord/index_coord_mock.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -547,6 +547,10 @@ type chunkManagerMock struct {
547547
remove func(string) error
548548
}
549549

550+
func (cmm *chunkManagerMock) RootPath() string {
551+
return ""
552+
}
553+
550554
func (cmm *chunkManagerMock) RemoveWithPrefix(prefix string) error {
551555
return cmm.removeWithPrefix(prefix)
552556
}

‎internal/indexnode/chunkmgr_mock.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,10 @@ var _ storage.ChunkManager = &mockChunkmgr{}
9191

9292
// var _ dependency.Factory = &mockFactory{}
9393

94+
func (c *mockChunkmgr) RootPath() string {
95+
return ""
96+
}
97+
9498
func (c *mockChunkmgr) Path(filePath string) (string, error) {
9599
// TODO
96100
return filePath, errNotImplErr

0 commit comments

Comments
 (0)