From 16c661c722fc8f0c96b6f63ae9bcd7ab2a1325dc Mon Sep 17 00:00:00 2001 From: congqixia Date: Tue, 19 Mar 2024 11:27:06 +0800 Subject: [PATCH] enhance: Use different interval for gc scan (#31363) See also #31362 This PR make datacoord garbage collection scan operation using differet interval than other opeartion. This interval is a newly added param item, which default value is 7*24 hours. Signed-off-by: Congqi Xia --- configs/milvus.yaml | 1 + internal/datacoord/garbage_collector.go | 15 ++++++++++++--- internal/datacoord/garbage_collector_test.go | 12 ++++++++++++ internal/datacoord/server.go | 1 + pkg/util/paramtable/component_param.go | 10 ++++++++++ 5 files changed, 36 insertions(+), 3 deletions(-) diff --git a/configs/milvus.yaml b/configs/milvus.yaml index f5d68a8266..b17b40e244 100644 --- a/configs/milvus.yaml +++ b/configs/milvus.yaml @@ -442,6 +442,7 @@ dataCoord: enableGarbageCollection: true gc: interval: 3600 # gc interval in seconds + scanInterval: 168 #gc residual file scan interval in hours missingTolerance: 3600 # file meta missing tolerance duration in seconds, 3600 dropTolerance: 10800 # file belongs to dropped entity tolerance duration in seconds. 10800 enableActiveStandby: false diff --git a/internal/datacoord/garbage_collector.go b/internal/datacoord/garbage_collector.go index 3c34cbb206..ea598dfd0b 100644 --- a/internal/datacoord/garbage_collector.go +++ b/internal/datacoord/garbage_collector.go @@ -51,6 +51,7 @@ type GcOption struct { checkInterval time.Duration // each interval missingTolerance time.Duration // key missing in meta tolerance time dropTolerance time.Duration // dropped segment related key tolerance time + scanInterval time.Duration // interval for scan residue for interupted log wrttien removeLogPool *conc.Pool[struct{}] } @@ -77,8 +78,12 @@ type gcCmd struct { // newGarbageCollector create garbage collector with meta and option func newGarbageCollector(meta *meta, handler Handler, opt GcOption) *garbageCollector { - log.Info("GC with option", zap.Bool("enabled", opt.enabled), zap.Duration("interval", opt.checkInterval), - zap.Duration("missingTolerance", opt.missingTolerance), zap.Duration("dropTolerance", opt.dropTolerance)) + log.Info("GC with option", + zap.Bool("enabled", opt.enabled), + zap.Duration("interval", opt.checkInterval), + zap.Duration("scanInterval", opt.scanInterval), + zap.Duration("missingTolerance", opt.missingTolerance), + zap.Duration("dropTolerance", opt.dropTolerance)) opt.removeLogPool = conc.NewPool[struct{}](Params.DataCoordCfg.GCRemoveConcurrent.GetAsInt(), conc.WithExpiryDuration(time.Minute)) return &garbageCollector{ meta: meta, @@ -145,6 +150,8 @@ func (gc *garbageCollector) work() { defer gc.wg.Done() ticker := time.NewTicker(gc.option.checkInterval) defer ticker.Stop() + scanTicker := time.NewTicker(gc.option.scanInterval) + defer ticker.Stop() for { select { case <-ticker.C: @@ -155,8 +162,10 @@ func (gc *garbageCollector) work() { gc.clearEtcd() gc.recycleUnusedIndexes() gc.recycleUnusedSegIndexes() - gc.scan() gc.recycleUnusedIndexFiles() + case <-scanTicker.C: + log.Info("Garbage collector start to scan interrupted write residue") + gc.scan() case cmd := <-gc.cmdCh: switch cmd.cmdType { case datapb.GcCommand_Pause: diff --git a/internal/datacoord/garbage_collector_test.go b/internal/datacoord/garbage_collector_test.go index 38e31853cc..c556bae34e 100644 --- a/internal/datacoord/garbage_collector_test.go +++ b/internal/datacoord/garbage_collector_test.go @@ -68,6 +68,7 @@ func Test_garbageCollector_basic(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Millisecond * 10, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -84,6 +85,7 @@ func Test_garbageCollector_basic(t *testing.T) { cli: nil, enabled: true, checkInterval: time.Millisecond * 10, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -120,6 +122,7 @@ func Test_garbageCollector_scan(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Minute * 30, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -137,6 +140,7 @@ func Test_garbageCollector_scan(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Minute * 30, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -162,6 +166,7 @@ func Test_garbageCollector_scan(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Minute * 30, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -190,6 +195,7 @@ func Test_garbageCollector_scan(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Minute * 30, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: 0, }) @@ -206,6 +212,7 @@ func Test_garbageCollector_scan(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Minute * 30, + scanInterval: time.Hour * 7 * 24, missingTolerance: 0, dropTolerance: 0, }) @@ -227,6 +234,7 @@ func Test_garbageCollector_scan(t *testing.T) { cli: cli, enabled: true, checkInterval: time.Minute * 30, + scanInterval: time.Hour * 7 * 24, missingTolerance: 0, dropTolerance: 0, }) @@ -1473,6 +1481,7 @@ func (s *GarbageCollectorSuite) TestPauseResume() { cli: s.cli, enabled: false, checkInterval: time.Millisecond * 10, + scanInterval: time.Hour * 24 * 7, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -1494,6 +1503,7 @@ func (s *GarbageCollectorSuite) TestPauseResume() { cli: s.cli, enabled: true, checkInterval: time.Millisecond * 10, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -1518,6 +1528,7 @@ func (s *GarbageCollectorSuite) TestPauseResume() { cli: s.cli, enabled: true, checkInterval: time.Millisecond * 10, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) @@ -1545,6 +1556,7 @@ func (s *GarbageCollectorSuite) TestPauseResume() { cli: s.cli, enabled: true, checkInterval: time.Millisecond * 10, + scanInterval: time.Hour * 7 * 24, missingTolerance: time.Hour * 24, dropTolerance: time.Hour * 24, }) diff --git a/internal/datacoord/server.go b/internal/datacoord/server.go index 6a8503258e..9b5fb7993c 100644 --- a/internal/datacoord/server.go +++ b/internal/datacoord/server.go @@ -531,6 +531,7 @@ func (s *Server) initGarbageCollection(cli storage.ChunkManager) { cli: cli, enabled: Params.DataCoordCfg.EnableGarbageCollection.GetAsBool(), checkInterval: Params.DataCoordCfg.GCInterval.GetAsDuration(time.Second), + scanInterval: Params.DataCoordCfg.GCScanIntervalInHour.GetAsDuration(time.Hour), missingTolerance: Params.DataCoordCfg.GCMissingTolerance.GetAsDuration(time.Second), dropTolerance: Params.DataCoordCfg.GCDropTolerance.GetAsDuration(time.Second), }) diff --git a/pkg/util/paramtable/component_param.go b/pkg/util/paramtable/component_param.go index 1c588e5706..c724dfd6b3 100644 --- a/pkg/util/paramtable/component_param.go +++ b/pkg/util/paramtable/component_param.go @@ -2558,6 +2558,7 @@ type dataCoordConfig struct { GCMissingTolerance ParamItem `refreshable:"false"` GCDropTolerance ParamItem `refreshable:"false"` GCRemoveConcurrent ParamItem `refreshable:"false"` + GCScanIntervalInHour ParamItem `refreshable:"false"` EnableActiveStandby ParamItem `refreshable:"false"` BindIndexNodeMode ParamItem `refreshable:"false"` @@ -2922,6 +2923,15 @@ During compaction, the size of segment # of rows is able to exceed segment max # } p.GCInterval.Init(base.mgr) + p.GCScanIntervalInHour = ParamItem{ + Key: "dataCoord.gc.scanInterval", + Version: "2.4.0", + DefaultValue: "168", // hours, default 7 * 24 hours + Doc: "garbage collection scan residue interval in hours", + Export: true, + } + p.GCScanIntervalInHour.Init(base.mgr) + // Do not set this to incredible small value, make sure this to be more than 10 minutes at least p.GCMissingTolerance = ParamItem{ Key: "dataCoord.gc.missingTolerance",