Support gc pause refine

Signed-off-by: wayblink <[email protected]>
zilliztech · Feb 6, 2024 · d9130ef · d9130ef
1 parent 3ef8e81
commit d9130ef
Show file tree

Hide file tree

Showing 9 changed files with 270 additions and 256 deletions.
diff --git a/build.sh b/build.sh
@@ -1,4 +1,4 @@
 go get
-./gen_proto.sh
 ./gen_swag.sh
+./gen_proto.sh
 go build
diff --git a/configs/backup.yaml b/configs/backup.yaml
@@ -18,9 +18,6 @@ milvus:
   tlsMode: 0
   user: "root"
   password: "Milvus"
-
-  useSSL: false # Optional, http API protocol, used to pause/resume GC of the Milvus cluster
-  httpPort: 9091 # Optional, http API port, used to pause/resume GC of the Milvus cluster
 
 # Related configuration of minio, which is responsible for data persistence for Milvus.
 minio:
@@ -60,5 +57,7 @@ backup:
   keepTempFiles: false
 
   # Pause GC during backup through Milvus Http API. 
-  pauseGcWhenBackup: true
-  pauseGcSeconds: 7200
+  gcPause:
+    enable: true
+    seconds: 7200
+    address: http://localhost:9091
diff --git a/core/backup_impl_create_backup.go b/core/backup_impl_create_backup.go
@@ -539,17 +539,11 @@ func (b *BackupContext) backupCollectionExecute(ctx context.Context, backupInfo
 	return nil
 }
 
-func (b *BackupContext) pauseMilvusGC(ctx context.Context) {
-	httpAddress := b.params.MilvusCfg.Address + ":" + b.params.MilvusCfg.HttpPort
-	if b.params.MilvusCfg.EnableSSL {
-		httpAddress = "https://" + httpAddress
-	} else {
-		httpAddress = "http://" + httpAddress
-	}
+func (b *BackupContext) pauseMilvusGC(ctx context.Context, gcAddress string, pauseSeconds int) {
 	pauseAPI := "/management/datacoord/garbage_collection/pause"
 	params := url.Values{}
-	params.Add("pause_seconds", strconv.Itoa(b.params.BackupCfg.PauseGcSeconds))
-	fullURL := fmt.Sprintf("%s?%s", httpAddress+pauseAPI, params.Encode())
+	params.Add("pause_seconds", strconv.Itoa(pauseSeconds))
+	fullURL := fmt.Sprintf("%s?%s", gcAddress+pauseAPI, params.Encode())
 	response, err := http.Get(fullURL)
 	if err != nil {
 		log.Error("Pause Milvus GC Error:", zap.Error(err))
@@ -562,18 +556,12 @@ func (b *BackupContext) pauseMilvusGC(ctx context.Context) {
 		log.Error("Read response Error:", zap.Error(err))
 		return
 	}
-	log.Info("Pause Milvus GC response", zap.String("response", string(body)))
+	log.Info("Pause Milvus GC response", zap.String("response", string(body)), zap.String("address", gcAddress), zap.Int("pauseSeconds", pauseSeconds))
 }
 
-func (b *BackupContext) resumeMilvusGC(ctx context.Context) {
-	httpAddress := b.params.MilvusCfg.Address + ":" + b.params.MilvusCfg.HttpPort
-	if b.params.MilvusCfg.EnableSSL {
-		httpAddress = "https://" + httpAddress
-	} else {
-		httpAddress = "http://" + httpAddress
-	}
+func (b *BackupContext) resumeMilvusGC(ctx context.Context, gcAddress string) {
 	pauseAPI := "/management/datacoord/garbage_collection/resume"
-	fullURL := httpAddress + pauseAPI
+	fullURL := gcAddress + pauseAPI
 	response, err := http.Get(fullURL)
 	if err != nil {
 		log.Error("Resume Milvus GC Error:", zap.Error(err))
@@ -585,17 +573,29 @@ func (b *BackupContext) resumeMilvusGC(ctx context.Context) {
 		log.Error("Read response Error:", zap.Error(err))
 		return
 	}
-	log.Info("Resume Milvus GC response", zap.String("response", string(body)))
+	log.Info("Resume Milvus GC response", zap.String("response", string(body)), zap.String("address", gcAddress))
 }
 
 func (b *BackupContext) executeCreateBackup(ctx context.Context, request *backuppb.CreateBackupRequest, backupInfo *backuppb.BackupInfo) (*backuppb.BackupInfo, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
 	// pause GC
-	if b.params.BackupCfg.PauseGcWhenBackup {
-		b.pauseMilvusGC(ctx)
-		defer b.resumeMilvusGC(ctx)
+	if request.GetGcPauseEnable() || b.params.BackupCfg.GcPauseEnable {
+		var pause = 0
+		if request.GetGcPauseSeconds() == 0 {
+			pause = b.params.BackupCfg.GcPauseSeconds
+		} else {
+			pause = int(request.GetGcPauseSeconds())
+		}
+		var gcAddress string = ""
+		if request.GetGcPauseAddress() == "" {
+			gcAddress = b.params.BackupCfg.GcPauseAddress
+		} else {
+			gcAddress = request.GetGcPauseAddress()
+		}
+		b.pauseMilvusGC(ctx, gcAddress, pause)
+		defer b.resumeMilvusGC(ctx, gcAddress)
 	}
 
 	backupInfo.BackupTimestamp = uint64(time.Now().UnixNano() / int64(time.Millisecond))

diff --git a/core/paramtable/params.go b/core/paramtable/params.go
@@ -38,9 +38,11 @@ type BackupConfig struct {
 	BackupCopyDataParallelism   int
 	RestoreParallelism          int
 
-	KeepTempFiles     bool
-	PauseGcWhenBackup bool
-	PauseGcSeconds    int
+	KeepTempFiles bool
+
+	GcPauseEnable  bool
+	GcPauseSeconds int
+	GcPauseAddress string
 }
 
 func (p *BackupConfig) init(base *BaseTable) {
@@ -51,8 +53,9 @@ func (p *BackupConfig) init(base *BaseTable) {
 	p.initRestoreParallelism()
 	p.initBackupCopyDataParallelism()
 	p.initKeepTempFiles()
-	p.initPauseGcWhenBackup()
-	p.initPauseGcSeconds()
+	p.initGcPauseEnable()
+	p.initGcPauseSeconds()
+	p.initGcPauseAddress()
 }
 
 func (p *BackupConfig) initMaxSegmentGroupSize() {
@@ -83,14 +86,19 @@ func (p *BackupConfig) initKeepTempFiles() {
 	p.KeepTempFiles, _ = strconv.ParseBool(keepTempFiles)
 }
 
-func (p *BackupConfig) initPauseGcWhenBackup() {
-	pauseGcWhenBackup := p.Base.LoadWithDefault("backup.pauseGcWhenBackup", "false")
-	p.PauseGcWhenBackup, _ = strconv.ParseBool(pauseGcWhenBackup)
+func (p *BackupConfig) initGcPauseEnable() {
+	enable := p.Base.LoadWithDefault("backup.gcPause.enable", "false")
+	p.GcPauseEnable, _ = strconv.ParseBool(enable)
+}
+
+func (p *BackupConfig) initGcPauseSeconds() {
+	seconds := p.Base.ParseIntWithDefault("backup.gcPause.seconds", 7200)
+	p.GcPauseSeconds = seconds
 }
 
-func (p *BackupConfig) initPauseGcSeconds() {
-	size := p.Base.ParseIntWithDefault("backup.pauseGcSeconds", 7200)
-	p.PauseGcSeconds = size
+func (p *BackupConfig) initGcPauseAddress() {
+	address := p.Base.LoadWithDefault("backup.gcPause.address", "http://localhost:9091")
+	p.GcPauseAddress = address
 }
 
 type MilvusConfig struct {
@@ -102,8 +110,6 @@ type MilvusConfig struct {
 	Password             string
 	AuthorizationEnabled bool
 	TLSMode              int
-	HttpPort             string
-	EnableSSL            bool
 }
 
 func (p *MilvusConfig) init(base *BaseTable) {
@@ -115,8 +121,6 @@ func (p *MilvusConfig) init(base *BaseTable) {
 	p.initPassword()
 	p.initAuthorizationEnabled()
 	p.initTLSMode()
-	p.initHttpPort()
-	p.initSSLEnabled()
 }
 
 func (p *MilvusConfig) initAddress() {
@@ -159,18 +163,6 @@ func (p *MilvusConfig) initTLSMode() {
 	p.TLSMode = p.Base.ParseIntWithDefault("milvus.tlsMode", 0)
 }
 
-func (p *MilvusConfig) initHttpPort() {
-	port, err := p.Base.Load("milvus.httpPort")
-	if err != nil {
-		panic(err)
-	}
-	p.HttpPort = port
-}
-
-func (p *MilvusConfig) initSSLEnabled() {
-	p.EnableSSL = p.Base.ParseBool("milvus.useSSL", false)
-}
-
 // /////////////////////////////////////////////////////////////////////////////
 // --- minio ---
 const (

diff --git a/core/proto/backup.proto b/core/proto/backup.proto
@@ -152,9 +152,11 @@ message CreateBackupRequest {
   // only backup meta, including collection schema and index info
   bool meta_only = 7;
   // if true, stop GC to avoid the data compacted and GCed during backup, use it when the data to backup is very large.
-  bool stop_gc = 8;
+  bool gc_pause_enable = 8;
   // gc pause seconds, set it larger than the time cost of backup
-  int64 gc_pause = 9;
+  int32 gc_pause_seconds = 9;
+  // gc pause API address
+  string gc_pause_address = 10;
 }
 
 /**