From 7e53a1781a744427dd79ae3229364bdcd28727b8 Mon Sep 17 00:00:00 2001 From: lukemakeit <2302063437@qq.com> Date: Thu, 14 Dec 2023 15:30:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(redis):=20redis=E5=A4=87=E4=BB=BD=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E4=B8=8A=E6=8A=A5=E6=97=B6=E9=97=B4=20#2522?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../go-pubpkg/backupclient/backupclient.go | 2 +- .../dbactuator/models/mysqlite/mysqlite.go | 48 +++ .../pkg/atomjobs/atomredis/bkdbmon_install.go | 7 +- .../pkg/atomjobs/atomredis/redis_backup.go | 228 ++++++++----- .../dbactuator/pkg/common/media_pkg.go | 2 +- .../db-tools/dbactuator/pkg/consts/consts.go | 4 - .../redis/db-tools/dbmon/config/config.go | 5 +- .../dbmon/models/mysqlite/mysqlite.go | 56 ++++ .../dbmon/pkg/backupsys/backupclient.go | 2 +- .../redis/db-tools/dbmon/pkg/consts/consts.go | 6 - .../dbmon/pkg/redisbinlogbackup/job.go | 304 +++++++---------- .../dbmon/pkg/redisbinlogbackup/task.go | 187 ++++++----- .../dbmon/pkg/redisfullbackup/backupjob.go | 309 +++++++----------- .../dbmon/pkg/redisfullbackup/checkjob.go | 8 +- .../dbmon/pkg/redisfullbackup/task.go | 200 +++++++----- .../dbmon/pkg/redismonitor/redis_task.go | 60 +++- .../dbmon/pkg/redisnodesreport/job.go | 27 +- .../flow/utils/redis/redis_act_playload.py | 6 +- 18 files changed, 791 insertions(+), 670 deletions(-) create mode 100644 dbm-services/redis/db-tools/dbactuator/models/mysqlite/mysqlite.go create mode 100644 dbm-services/redis/db-tools/dbmon/models/mysqlite/mysqlite.go diff --git a/dbm-services/common/go-pubpkg/backupclient/backupclient.go b/dbm-services/common/go-pubpkg/backupclient/backupclient.go index bb71323c7a..d521b4d673 100644 --- a/dbm-services/common/go-pubpkg/backupclient/backupclient.go +++ b/dbm-services/common/go-pubpkg/backupclient/backupclient.go @@ -140,7 +140,7 @@ func (b *BackupClient) Query2(backupTaskId string) (status int, statusMsg string } resp := []NewResp{} if err := json.Unmarshal(stdout, &resp); err != nil { - return 0, "", errors.Wrapf(err, "BackupClient parse query response %s fail", string(stdout)) + return 0, "", errors.Wrapf(err, "BackupClient parse query response %s fail,queryArgs:%v", string(stdout), queryArgs) } return resp[0].Status, resp[0].Message, err } diff --git a/dbm-services/redis/db-tools/dbactuator/models/mysqlite/mysqlite.go b/dbm-services/redis/db-tools/dbactuator/models/mysqlite/mysqlite.go new file mode 100644 index 0000000000..dcce4d2b8e --- /dev/null +++ b/dbm-services/redis/db-tools/dbactuator/models/mysqlite/mysqlite.go @@ -0,0 +1,48 @@ +// Package mysqlite TODO +package mysqlite + +import ( + "fmt" + "path/filepath" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + + "dbm-services/redis/db-tools/dbactuator/mylog" + "dbm-services/redis/db-tools/dbactuator/pkg/consts" + "dbm-services/redis/db-tools/dbactuator/pkg/util" +) + +func getLocalDbName() (dbname string, err error) { + dbname = filepath.Join(consts.BkDbmonPath, "db", "lucky_boy.db") + return +} + +// GetLocalSqDB TODO +func GetLocalSqDB() (sqDB *gorm.DB, err error) { + dbName, err := getLocalDbName() + if err != nil { + return + } + dbDir := filepath.Dir(dbName) + err = util.MkDirsIfNotExists([]string{dbDir}) + if err != nil { + return + } + util.LocalDirChownMysql(dbDir) + sqDB, err = gorm.Open(sqlite.Open(dbName), &gorm.Config{}) + if err != nil { + err = fmt.Errorf("gorm.Open failed,err:%v,dbname:%s", err, dbName) + mylog.Logger.Info(err.Error()) + return + } + return +} + +// CloseDB TODO +func CloseDB(sqDB *gorm.DB) { + if sqDB != nil { + dbInstance, _ := sqDB.DB() + _ = dbInstance.Close() + } +} diff --git a/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/bkdbmon_install.go b/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/bkdbmon_install.go index 0a1997fa86..c06f9af36f 100644 --- a/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/bkdbmon_install.go +++ b/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/bkdbmon_install.go @@ -204,8 +204,8 @@ func (job *BkDbmonInstall) UntarMedia() (err error) { var remoteVersion, localVersion string err = job.params.BkDbmonPkg.Check() if err != nil { - job.runtime.Logger.Error("UntarMedia failed,err:%v", err) - return + job.runtime.Logger.Error("UntarMedia check failed,err:%v,skip...", err) + // return } defer util.LocalDirChownMysql(consts.BkDbmonPath + "/") verReg := regexp.MustCompile(`bk-dbmon-(v\d+.\d+).tar.gz`) @@ -346,9 +346,6 @@ func (job *BkDbmonInstall) GenerateConfigFile() (err error) { var yamlData []byte var confMd5, tempMd5 string var notUpdateConf bool = false - if job.params.BackupClientStrorageType == "" { - job.params.BackupClientStrorageType = consts.BackupClientStrorageTypeCOS - } confData := &bkDbmonConf{ ReportSaveDir: consts.DbaReportSaveDir, ReportLeftDay: consts.RedisReportLeftDay, diff --git a/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/redis_backup.go b/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/redis_backup.go index 3217afcb6f..1792fe91b1 100644 --- a/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/redis_backup.go +++ b/dbm-services/redis/db-tools/dbactuator/pkg/atomjobs/atomredis/redis_backup.go @@ -11,10 +11,10 @@ import ( "time" "dbm-services/redis/db-tools/dbactuator/models/myredis" + "dbm-services/redis/db-tools/dbactuator/models/mysqlite" "dbm-services/redis/db-tools/dbactuator/mylog" "dbm-services/redis/db-tools/dbactuator/pkg/backupsys" "dbm-services/redis/db-tools/dbactuator/pkg/consts" - "dbm-services/redis/db-tools/dbactuator/pkg/customtime" "dbm-services/redis/db-tools/dbactuator/pkg/jobruntime" "dbm-services/redis/db-tools/dbactuator/pkg/report" "dbm-services/redis/db-tools/dbactuator/pkg/util" @@ -22,6 +22,8 @@ import ( "github.com/go-playground/validator/v10" "github.com/gofrs/flock" "gopkg.in/yaml.v2" + "gorm.io/gorm" + "gorm.io/gorm/clause" ) // TendisSSDSetLogCount tendisSSD设置log参数 @@ -51,6 +53,7 @@ type RedisBackup struct { params RedisBackupParams Reporter report.Reporter `json:"-"` backupClient backupsys.BackupClient + sqdb *gorm.DB } // 无实际作用,仅确保实现了 jobruntime.JobRunner 接口 @@ -106,9 +109,6 @@ func (job *RedisBackup) Init(m *jobruntime.JobGenericRuntime) error { job.runtime.Logger.Error(err.Error()) return err } - if job.params.BackupClientStrorageType == "" { - job.params.BackupClientStrorageType = consts.BackupClientStrorageTypeCOS - } return nil } @@ -133,6 +133,14 @@ func (job *RedisBackup) Run() (err error) { if err != nil { return } + defer job.Reporter.Close() + + err = job.getSqlDB() + if err != nil { + return + } + defer job.closeDB() + err = job.GetBackupClient() if err != nil { return @@ -149,7 +157,8 @@ func (job *RedisBackup) Run() (err error) { } task := NewFullBackupTask(job.params.BkBizID, job.params.Domain, job.params.IP, port, password, toBackSys, job.params.BackupType, bakDir, - true, consts.BackupTarSplitSize, job.params.SSDLogCount, job.Reporter, job.backupClient) + true, consts.BackupTarSplitSize, job.params.SSDLogCount, + job.Reporter, job.backupClient, job.sqdb) bakTasks = append(bakTasks, task) } @@ -224,6 +233,24 @@ func (job *RedisBackup) GetReporter() (err error) { return } +func (job *RedisBackup) getSqlDB() (err error) { + job.sqdb, err = mysqlite.GetLocalSqDB() + if err != nil { + return + } + err = job.sqdb.AutoMigrate(&RedisFullbackupHistorySchema{}) + if err != nil { + err = fmt.Errorf("RedisFullbackupHistorySchema AutoMigrate fail,err:%v", err) + mylog.Logger.Info(err.Error()) + return + } + return +} + +func (job *RedisBackup) closeDB() { + mysqlite.CloseDB(job.sqdb) +} + // GetBackupClient 获取备份系统client func (job *RedisBackup) GetBackupClient() (err error) { bkTag := consts.RedisFullBackupTAG @@ -240,64 +267,115 @@ func (job *RedisBackup) GetBackupClient() (err error) { return } +// RedisFullbackupHistorySchema TODO +type RedisFullbackupHistorySchema struct { + ID int64 `json:"-" gorm:"primaryKey;column:id;not null` + ReportType string `json:"report_type" gorm:"column:report_type;not null;default:''"` + BkBizID string `json:"bk_biz_id" gorm:"column:bk_biz_id;not null;default:''"` + BkCloudID int64 `json:"bk_cloud_id" gorm:"column:bk_cloud_id;not null;default:0"` + ServerIP string `json:"server_ip" gorm:"column:server_ip;not null;default:''"` + ServerPort int `json:"server_port" gorm:"column:server_port;not null;default:0"` + Domain string `json:"domain" gorm:"column:domain;not null;default:'';index"` + // RedisInstance or TendisplusInstance or TendisSSDInstance + DbType string `json:"db_type" gorm:"column:db_type;not null;default:''"` + Role string `json:"role" gorm:"column:role;not null;default:''"` + BackupDir string `json:"backup_dir" gorm:"column:backup_dir;not null;default:''"` + // 备份的目标文件 + BackupFile string `json:"backup_file" gorm:"column:backup_file;not null;default:''"` + // 备份文件大小(已切割 or 已压缩 or 已打包) + BackupFileSize int64 `json:"backup_file_size" gorm:"column:backup_file_size;not null;default:0"` + BackupTaskID string `json:"backup_taskid" gorm:"column:backup_taskid;not null;default:''"` + // 目前为空 + BackupMD5 string `json:"backup_md5" gorm:"column:backup_md5;not null;default:''"` + // REDIS_FULL + BackupTag string `json:"backup_tag" gorm:"column:backup_tag;not null;default:''"` + // shard值 + ShardValue string `json:"shard_value" gorm:"column:shard_value;not null;default:''"` + // 生成全备的起始时间 + StartTime time.Time `json:"start_time" gorm:"column:start_time;not null;default:'';index"` + // 生成全备的结束时间 + EndTime time.Time `json:"end_time" gorm:"column:end_time;not null;default:'';index"` + TimeZone string `json:"time_zone" gorm:"column:time_zone;not null;default:''"` + Status string `json:"status" gorm:"column:status;not null;default:''"` + Message string `json:"message" gorm:"column:message;not null;default:''"` + // 本地文件是否已删除,未被删除为0,已被删除为1 + LocalFileRemoved int `json:"-" gorm:"column:local_file_removed;not null;default:0"` +} + +// TableName TODO +func (r *RedisFullbackupHistorySchema) TableName() string { + return "redis_fullbackup_history" +} + +type redisFullBackupReport struct { + RedisFullbackupHistorySchema + StartTime string `json:"start_time"` + EndTime string `json:"end_time"` +} + +// BackupRecordReport 备份记录上报 +func (r *RedisFullbackupHistorySchema) BackupRecordReport(reporter report.Reporter) { + if reporter == nil { + return + } + reportRow := redisFullBackupReport{ + RedisFullbackupHistorySchema: *r, + StartTime: r.StartTime.Local().Format(time.RFC3339), + EndTime: r.EndTime.Local().Format(time.RFC3339), + } + tmpBytes, _ := json.Marshal(reportRow) + reporter.AddRecord(string(tmpBytes)+"\n", true) +} + // BackupTask redis备份task type BackupTask struct { - ReportType string `json:"report_type"` - BkBizID string `json:"bk_biz_id"` - ServerIP string `json:"server_ip"` - ServerPort int `json:"server_port"` - Domain string `json:"domain"` - Password string `json:"-"` - ToBackupSystem string `json:"-"` - DbType string `json:"db_type"` // RedisInstance or TendisplusInstance or TendisSSDInstance - BackupType string `json:"-"` // 常规备份、下线备份 - Role string `json:"role"` - DataSize uint64 `json:"-"` // redis实例数据大小 - DataDir string `json:"-"` - BackupDir string `json:"backup_dir"` - TarSplit bool `json:"-"` // 是否对tar文件做split - TarSplitPartSize string `json:"-"` - BackupFile string `json:"backup_file"` // 备份的目标文件,如果文件过大会切割成多个 - BackupFileSize int64 `json:"backup_file_size"` // 备份文件大小(已切割 or 已压缩 or 已打包) - BackupTaskID string `json:"backup_taskid"` - BackupMD5 string `json:"backup_md5"` // 目前为空 - BackupTag string `json:"backup_tag"` // REDIS_FULL or REDIS_BINLOG - ShardValue string `json:"shard_value"` // shard值 - // 全备尽管会切成多个文件,但其生成的起始时间、结束时间一样 - StartTime customtime.CustomTime `json:"start_time"` // 生成全备的起始时间 - EndTime customtime.CustomTime `json:"end_time"` // //生成全备的结束时间 - Status string `json:"status"` - Message string `json:"message"` - Cli *myredis.RedisClient `json:"-"` - SSDLogCount TendisSSDSetLogCount `json:"-"` - reporter report.Reporter - backupClient backupsys.BackupClient - Err error `json:"-"` + RedisFullbackupHistorySchema + Password string `json:"-"` + ToBackupSystem string `json:"-"` + BackupType string `json:"-"` // 常规备份、下线备份 + DataSize uint64 `json:"-"` // redis实例数据大小 + DataDir string `json:"-"` + TarSplit bool `json:"-"` // 是否对tar文件做split + TarSplitPartSize string `json:"-"` + Cli *myredis.RedisClient `json:"-"` + SSDLogCount TendisSSDSetLogCount `json:"-"` + reporter report.Reporter + backupClient backupsys.BackupClient + sqdb *gorm.DB + Err error `json:"-"` } // NewFullBackupTask new backup task func NewFullBackupTask(bkBizID, domain, ip string, port int, password, toBackupSys, backupType, backupDir string, tarSplit bool, tarSplitSize string, - ssdLogCount TendisSSDSetLogCount, reporter report.Reporter, bakCli backupsys.BackupClient) *BackupTask { - return &BackupTask{ - ReportType: consts.RedisFullBackupReportType, - BkBizID: bkBizID, - Domain: domain, - ServerIP: ip, - ServerPort: port, + ssdLogCount TendisSSDSetLogCount, reporter report.Reporter, + bakCli backupsys.BackupClient, + sqDB *gorm.DB) *BackupTask { + timeZone, _ := time.Now().Local().Zone() + ret := &BackupTask{ Password: password, ToBackupSystem: toBackupSys, BackupType: backupType, - BackupDir: backupDir, TarSplit: tarSplit, TarSplitPartSize: tarSplitSize, - BackupTaskID: "", - BackupMD5: "", - BackupTag: consts.RedisFullBackupTAG, SSDLogCount: ssdLogCount, reporter: reporter, backupClient: bakCli, - } + sqdb: sqDB, + } + ret.RedisFullbackupHistorySchema = RedisFullbackupHistorySchema{ + ReportType: consts.RedisFullBackupReportType, + BkBizID: bkBizID, + Domain: domain, + ServerIP: ip, + ServerPort: port, + BackupDir: backupDir, + BackupTaskID: "", + BackupMD5: "", + BackupTag: consts.RedisFullBackupTAG, + TimeZone: timeZone, + } + return ret } // Addr string @@ -357,12 +435,12 @@ func (task *BackupTask) GoFullBakcup() { task.Message = task.Err.Error() task.Status = consts.BackupStatusFailed } - task.BackupRecordReport() + task.BackupRecordReport(task.reporter) }() task.Status = consts.BackupStatusRunning task.Message = "start backup..." - task.BackupRecordReport() + task.BackupRecordReport(task.reporter) mylog.Logger.Info(fmt.Sprintf("redis(%s) dbType:%s start backup...", task.Addr(), task.DbType)) @@ -394,7 +472,7 @@ func (task *BackupTask) GoFullBakcup() { if task.Err != nil { return } - defer task.BackupRecordSaveToDoingFile() + defer task.BackupRecordSaveToLocalDB() // 备份上传备份系统 if strings.ToLower(task.ToBackupSystem) != "yes" { task.Status = consts.BackupStatusLocalSuccess @@ -548,7 +626,7 @@ func (task *BackupTask) RedisInstanceBackup() { var confMap map[string]string var fileSize int64 nowtime := time.Now().Local().Format(consts.FilenameTimeLayout) - task.StartTime.Time = time.Now().Local() + task.StartTime = time.Now().Local() if task.Role == consts.RedisMasterRole { // redis master backup rdb confMap, task.Err = task.Cli.ConfigGet("dbfilename") @@ -571,7 +649,7 @@ func (task *BackupTask) RedisInstanceBackup() { if task.Err != nil { return } - task.EndTime.Time = time.Now().Local() + task.EndTime = time.Now().Local() cpCmd := fmt.Sprintf("cp %s %s", srcFile, targetFile) mylog.Logger.Info(cpCmd) _, task.Err = util.RunBashCmd(cpCmd, "", nil, 10*time.Minute) @@ -611,12 +689,12 @@ func (task *BackupTask) TendisplusInstanceBackup() { return } util.LocalDirChownMysql(task.BackupDir) - task.StartTime.Time = time.Now().Local() + task.StartTime = time.Now().Local() task.Err = task.Cli.TendisplusBackupAndWaitForDone(backupFullDir) if task.Err != nil { return } - task.EndTime.Time = time.Now().Local() + task.EndTime = time.Now().Local() tarFile, task.Err = util.TarADir(backupFullDir, task.BackupDir, true) if task.Err != nil { mylog.Logger.Error(task.Err.Error()) @@ -669,12 +747,12 @@ func (task *BackupTask) TendisSSDInstanceBackup() { return } util.LocalDirChownMysql(task.BackupDir) - task.StartTime.Time = time.Now().Local() + task.StartTime = time.Now().Local() binlogsizeRet, _, task.Err = task.Cli.TendisSSDBackupAndWaitForDone(backupFullDir) if task.Err != nil { return } - task.EndTime.Time = time.Now().Local() + task.EndTime = time.Now().Local() task.tendisSSDBackupVerify(backupFullDir) if task.Err != nil { @@ -782,35 +860,17 @@ func (task *BackupTask) TransferToBackupSystem() { return } -// BackupRecordReport 备份记录上报 -func (task *BackupTask) BackupRecordReport() { - if task.reporter == nil { - return - } - tmpBytes, _ := json.Marshal(task) - // task.Err=task.reporter.AddRecord(string(tmpBytes),true) - task.reporter.AddRecord(string(tmpBytes)+"\n", true) -} - -// BackupRecordSaveToDoingFile 备份记录保存到本地 redis_backup_file_list_${port}_doing 文件中 -func (task *BackupTask) BackupRecordSaveToDoingFile() { - backupDir := filepath.Dir(task.BackupFile) - // 例如: /data/dbbak/backup/redis_backup_file_list_30000_doing - backupDir = filepath.Join(backupDir, "backup") - util.MkDirsIfNotExists([]string{}) - util.LocalDirChownMysql(backupDir) - doingFile := filepath.Join(backupDir, fmt.Sprintf(consts.DoingRedisFullBackFileList, task.ServerPort)) - f, err := os.OpenFile(doingFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if err != nil { - task.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doingFile, err) - mylog.Logger.Error(task.Err.Error()) +// BackupRecordSaveToLocalDB 备份信息记录到本地sqlite中 +func (task *BackupTask) BackupRecordSaveToLocalDB() { + if task.sqdb == nil { return } - defer f.Close() - tmpBytes, _ := json.Marshal(task) - - if _, err = f.WriteString(string(tmpBytes) + "\n"); err != nil { - task.Err = fmt.Errorf("f.WriteString failed,err:%v,file:%s,line:%s", err, doingFile, string(tmpBytes)) + task.RedisFullbackupHistorySchema.ID = 0 // 重置为0,以便gorm自增 + task.Err = task.sqdb.Clauses(clause.OnConflict{ + UpdateAll: true, + }).Create(&task.RedisFullbackupHistorySchema).Error + if task.Err != nil { + task.Err = fmt.Errorf("BackupRecordSaveToLocalDB sqdb.Create fail,err:%v", task.Err) mylog.Logger.Error(task.Err.Error()) return } diff --git a/dbm-services/redis/db-tools/dbactuator/pkg/common/media_pkg.go b/dbm-services/redis/db-tools/dbactuator/pkg/common/media_pkg.go index fe5930098a..d9fd73ea4f 100644 --- a/dbm-services/redis/db-tools/dbactuator/pkg/common/media_pkg.go +++ b/dbm-services/redis/db-tools/dbactuator/pkg/common/media_pkg.go @@ -44,7 +44,7 @@ func (m *MediaPkg) Check() (err error) { } // 校验md5 if fileMd5 != m.PkgMd5 { - return fmt.Errorf("安装包的md5不匹配,[%s]文件的md5[%s]不正确", fileMd5, m.PkgMd5) + return fmt.Errorf("安装包的md5不匹配,%s文件的md5[%s],不满足预期%s", pkgAbPath, fileMd5, m.PkgMd5) } return } diff --git a/dbm-services/redis/db-tools/dbactuator/pkg/consts/consts.go b/dbm-services/redis/db-tools/dbactuator/pkg/consts/consts.go index 3494c012db..6932b97d73 100644 --- a/dbm-services/redis/db-tools/dbactuator/pkg/consts/consts.go +++ b/dbm-services/redis/db-tools/dbactuator/pkg/consts/consts.go @@ -164,10 +164,6 @@ const ( RedisForeverBackupTAG = "DBFILE" RedisFullBackupReportType = "redis_fullbackup" RedisBinlogBackupReportType = "redis_binlogbackup" - DoingRedisFullBackFileList = "redis_backup_file_list_%d_doing" - DoneRedisFullBackFileList = "redis_backup_file_list_%d_done" - DoingRedisBinlogFileList = "redis_binlog_file_list_%d_doing" - DoneRedisBinlogFileList = "redis_binlog_file_list_%d_done" RedisFullbackupRepoter = "redis_fullbackup_%s.log" RedisBinlogRepoter = "redis_binlog_%s.log" BackupStatusStart = "start" diff --git a/dbm-services/redis/db-tools/dbmon/config/config.go b/dbm-services/redis/db-tools/dbmon/config/config.go index cabde950b2..87cc91af69 100644 --- a/dbm-services/redis/db-tools/dbmon/config/config.go +++ b/dbm-services/redis/db-tools/dbmon/config/config.go @@ -112,9 +112,8 @@ func loadConfigFile() { if conf.BeatPath == "" { conf.BeatPath = "/usr/local/gse_bkte/plugins/bin/bkmonitorbeat" } - if conf.BackupClientStrorageType == "" { - conf.BackupClientStrorageType = "cos" - } + // backup_client_storage_type 设置为空,用系统默认的 + conf.BackupClientStrorageType = "" fmt.Println(conf.String()) GlobalConf = &conf } diff --git a/dbm-services/redis/db-tools/dbmon/models/mysqlite/mysqlite.go b/dbm-services/redis/db-tools/dbmon/models/mysqlite/mysqlite.go new file mode 100644 index 0000000000..c93dd7fae7 --- /dev/null +++ b/dbm-services/redis/db-tools/dbmon/models/mysqlite/mysqlite.go @@ -0,0 +1,56 @@ +// Package mysqlite TODO +package mysqlite + +import ( + "fmt" + "os" + "path/filepath" + + "github.com/glebarez/sqlite" + "gorm.io/gorm" + + "dbm-services/redis/db-tools/dbmon/mylog" + "dbm-services/redis/db-tools/dbmon/util" +) + +func getLocalDbName() (dbname string, err error) { + var homeDir string + homeDir, err = os.Executable() + if err != nil { + err = fmt.Errorf("os.Executable failed,err:%v", err) + mylog.Logger.Info(err.Error()) + return + } + homeDir = filepath.Dir(homeDir) + dbname = filepath.Join(homeDir, "db", "lucky_boy.db") + return +} + +// GetLocalSqDB TODO +func GetLocalSqDB() (sqDB *gorm.DB, err error) { + dbName, err := getLocalDbName() + if err != nil { + return + } + dbDir := filepath.Dir(dbName) + err = util.MkDirsIfNotExists([]string{dbDir}) + if err != nil { + return + } + util.LocalDirChownMysql(dbDir) + sqDB, err = gorm.Open(sqlite.Open(dbName), &gorm.Config{}) + if err != nil { + err = fmt.Errorf("gorm.Open failed,err:%v,dbname:%s", err, dbName) + mylog.Logger.Info(err.Error()) + return + } + return +} + +// CloseDB TODO +func CloseDB(sqDB *gorm.DB) { + if sqDB != nil { + dbInstance, _ := sqDB.DB() + _ = dbInstance.Close() + } +} diff --git a/dbm-services/redis/db-tools/dbmon/pkg/backupsys/backupclient.go b/dbm-services/redis/db-tools/dbmon/pkg/backupsys/backupclient.go index 463e3b3f8a..34de09716a 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/backupsys/backupclient.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/backupsys/backupclient.go @@ -49,7 +49,7 @@ func (o *COSBackupClient) Upload(fileName string) (taskId string, err error) { func (o *COSBackupClient) TaskStatus(taskId string) (status int, statusMsg string, err error) { status, statusMsg, err = o.backupClient.Query2(taskId) if err != nil { - mylog.Logger.Error(fmt.Sprintf("CosBackupClient Query2 failed,err:%v", err)) + mylog.Logger.Error(fmt.Sprintf("CosBackupClient Query2 failed,err:%v,taskId:%s", err, taskId)) return } return diff --git a/dbm-services/redis/db-tools/dbmon/pkg/consts/consts.go b/dbm-services/redis/db-tools/dbmon/pkg/consts/consts.go index 8d76e3473c..fad1a1655a 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/consts/consts.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/consts/consts.go @@ -147,12 +147,6 @@ const ( RedisFullBackupReportType = "redis_fullbackup" RedisBinlogBackupReportType = "redis_binlogbackup" - DoingRedisFullBackFileList = "redis_backup_file_list_%d_doing" - DoneRedisFullBackFileList = "redis_backup_file_list_%d_done" - - DoingRedisBinlogFileList = "redis_binlog_file_list_%d_doing" - DoneRedisBinlogFileList = "redis_binlog_file_list_%d_done" - RedisFullbackupRepoter = "redis_fullbackup_%s.log" RedisBinlogRepoter = "redis_binlog_%s.log" RedisClusterNodesRepoter = "redis_cluster_nodes_%s.log" diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/job.go b/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/job.go index 6f373e34d5..8febc9e039 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/job.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/job.go @@ -1,8 +1,6 @@ package redisbinlogbackup import ( - "bufio" - "encoding/json" "fmt" "os" "path/filepath" @@ -11,8 +9,11 @@ import ( "sync" "time" + "gorm.io/gorm" + "dbm-services/redis/db-tools/dbmon/config" "dbm-services/redis/db-tools/dbmon/models/myredis" + "dbm-services/redis/db-tools/dbmon/models/mysqlite" "dbm-services/redis/db-tools/dbmon/mylog" "dbm-services/redis/db-tools/dbmon/pkg/backupsys" "dbm-services/redis/db-tools/dbmon/pkg/consts" @@ -31,6 +32,7 @@ type Job struct { // NOCC:golint/naming(其他:设计如此) RealBackupDir string `json:"real_backup_dir"` // 如 /data/dbbak Reporter report.Reporter `json:"-"` backupClient backupsys.BackupClient + sqdb *gorm.DB Err error `json:"-"` } @@ -65,13 +67,23 @@ func (job *Job) Run() { } defer job.Reporter.Close() + job.getSqlDB() + if job.Err != nil { + return + } + defer job.closeDB() + // job.backupClient = backupsys.NewIBSBackupClient(consts.IBSBackupClient, consts.RedisBinlogTAG) job.backupClient, job.Err = backupsys.NewCosBackupClient(consts.COSBackupClient, consts.COSInfoFile, consts.RedisBinlogTAG, job.Conf.BackupClientStrorageType) - if job.Err != nil && !strings.HasPrefix(job.Err.Error(), "backup_client path not found") { - return + if job.Err != nil { + if strings.HasPrefix(job.Err.Error(), "backup_client path not found") { + mylog.Logger.Debug(fmt.Sprintf("backup_client path:%s not found", consts.COSBackupClient)) + job.Err = nil + } else { + return + } } - job.Err = nil job.createTasks() if job.Err != nil { return @@ -92,8 +104,8 @@ func (job *Job) Run() { continue } for _, port := range svrItem.ServerPorts { - job.CheckOldBinlogBackupStatus(port) job.DeleteTooOldBinlogbackup(port) + job.CheckOldBinlogBackupStatus(port) } } } @@ -114,6 +126,23 @@ func (job *Job) GetReporter() { job.Reporter, job.Err = report.NewFileReport(filepath.Join(reportDir, reportFile)) } +func (job *Job) getSqlDB() { + job.sqdb, job.Err = mysqlite.GetLocalSqDB() + if job.Err != nil { + return + } + job.Err = job.sqdb.AutoMigrate(&RedisBinlogHistorySchema{}) + if job.Err != nil { + job.Err = fmt.Errorf("RedisFullbackupHistorySchema AutoMigrate fail,err:%v", job.Err) + mylog.Logger.Info(job.Err.Error()) + return + } +} + +func (job *Job) closeDB() { + mysqlite.CloseDB(job.sqdb) +} + func (job *Job) createTasks() { var task *Task var password string @@ -140,7 +169,8 @@ func (job *Job) createTasks() { taskBackupDir, svrItem.ServerShards[instStr], job.Conf.RedisBinlogBackup.OldFileLeftDay, job.Reporter, - job.Conf.BackupClientStrorageType) + job.Conf.BackupClientStrorageType, + job.sqdb) if job.Err != nil { return } @@ -149,218 +179,126 @@ func (job *Job) createTasks() { } } -// CheckOldBinlogBackupStatus 检查历史binlog备份任务状态 -// 1. 遍历 redis_binlog_file_list_${port}_doing 文件 -// 2. 已超过时间的任务,删除本地文件,从 redis_binlog_file_list_${port}_doing 中剔除 -// 3. 上传备份系统 运行中 or 失败的任务 记录到 redis_binlog_file_list_${port}_doing_temp -// 4. 已成功的任务,记录到 redis_binlog_file_list_${port}_done -// 5. rename redis_binlog_file_list_${port}_doing_temp to redis_binlog_file_list_${port}_doing +// CheckOldBinlogBackupStatus 重试备份系统上传失败的,检查备份系统上传中的是否成功 func (job *Job) CheckOldBinlogBackupStatus(port int) { - var doingHandler, tempHandler, doneHandler *os.File - var line string - var err error + mylog.Logger.Debug(fmt.Sprintf("port:%d start CheckOldBinlogBackupStatus", port)) + toCheckRows := []RedisBinlogHistorySchema{} var taskStatus int var statusMsg string - task := Task{} - oldFileLeftSec := job.Conf.RedisBinlogBackup.OldFileLeftDay * 24 * 3600 - nowTime := time.Now().Local() - // 示例: /data/dbbak/binlog/30000/redis_binlog_file_list_30000_doing - doingFile := filepath.Join(job.RealBackupDir, "binlog", strconv.Itoa(port), - fmt.Sprintf(consts.DoingRedisBinlogFileList, port)) - if !util.FileExists(doingFile) { - return - } - // 示例: /data/dbbak/binlog/30000/redis_binlog_file_list_30000_doing_temp - tempDoingFile := doingFile + "_temp" - // 示例: /data/dbbak/binlog/30000/redis_binlog_file_list_30000_done - doneFile := filepath.Join(job.RealBackupDir, "binlog", strconv.Itoa(port), - fmt.Sprintf(consts.DoneRedisBinlogFileList, port)) - - defer func() { - if job.Err == nil { - mylog.Logger.Info(fmt.Sprintf("rename %s to %s", tempDoingFile, doingFile)) - os.Rename(tempDoingFile, doingFile) // rename - } - }() - - doingHandler, job.Err = os.Open(doingFile) - if job.Err != nil { - job.Err = fmt.Errorf("os.Open file:%s fail,err:%v", doingFile, job.Err) - mylog.Logger.Error(job.Err.Error()) - return - } - defer doingHandler.Close() - - doneHandler, job.Err = os.OpenFile(doneFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if job.Err != nil { - job.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doneFile, job.Err) + job.Err = job.sqdb.Where("status not in (?,?,?)", + consts.BackupStatusToBakSysSuccess, + consts.BackupStatusFailed, + consts.BackupStatusLocalSuccess, + ).Find(&toCheckRows).Error + if job.Err != nil && job.Err != gorm.ErrRecordNotFound { + job.Err = fmt.Errorf("CheckOldBinlogBackupStatus gorm find fail,err:%v", job.Err) mylog.Logger.Error(job.Err.Error()) return } - defer doneHandler.Close() - - tempHandler, job.Err = os.OpenFile(tempDoingFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if job.Err != nil { - job.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", tempDoingFile, job.Err) - mylog.Logger.Error(job.Err.Error()) - return - } - defer tempHandler.Close() - - scanner := bufio.NewScanner(doingHandler) - scanner.Split(bufio.ScanLines) - for scanner.Scan() { - line = scanner.Text() - err = json.Unmarshal([]byte(line), &task) - if err != nil { - // json.Unmarshal failed,skip ... - err = fmt.Errorf("json.Unmarshal fail,err:%s,data:%s,skip it", err, line) - mylog.Logger.Error(err.Error()) - continue - } - task.reporter = job.Reporter - task.backupClient = job.backupClient - // 删除旧文件 - if nowTime.Sub(task.EndTime.Time).Seconds() > float64(oldFileLeftSec) { - mylog.Logger.Info(fmt.Sprintf("%s start removing...", task.BackupFile)) - if util.FileExists(task.BackupFile) { - err = os.Remove(task.BackupFile) - if err != nil { - err = fmt.Errorf("os.Remove fail,err:%s,file:%s", err, task.BackupFile) - mylog.Logger.Error(err.Error()) - tempHandler.WriteString(line + "\n") // 删除失败的,记录到temp文件,下次继续重试 - } - fmt.Printf("remove %s\n", task.BackupFile) + job.Err = nil + mylog.Logger.Debug(fmt.Sprintf("port:%d CheckOldBinlogBackupStatus toCheckRowsCnt:%d", port, len(toCheckRows))) + for _, row := range toCheckRows { + if row.Status == consts.BackupStatusToBakSystemFailed && job.backupClient != nil { + // 重试备份系统上传失败的 + if !util.FileExists(row.BackupFile) { + continue } - continue - } - // 无需上传备份系统,本地已备份成功的情况 - if task.Status == consts.BackupStatusLocalSuccess { - doneHandler.WriteString(line + "\n") - continue - } - // 上传备份系统失败的情况,重试上传并写入temp文件中 - if task.Status == consts.BackupStatusToBakSystemFailed { - task.TransferToBackupSystem() - if task.Err != nil { - task.Message = task.Err.Error() + mylog.Logger.Info(fmt.Sprintf("redis(%s) binlog:%+v retry upload backupSystem", row.Addr(), row.BackupFile)) + row.BackupTaskID, job.Err = job.backupClient.Upload(row.BackupFile) + if job.Err != nil { + row.Message = job.Err.Error() } else { - task.Status = consts.BackupStatusToBakSystemStart - task.Message = "上传备份系统中" + row.Status = consts.BackupStatusToBakSystemStart + row.Message = "上传备份系统中" + } + // 更新记录 status 和 message + job.Err = job.sqdb.Save(&row).Error + if job.Err != nil { + job.Err = fmt.Errorf("gorm save fail,err:%v", job.Err) + mylog.Logger.Error(job.Err.Error()) } - tempHandler.WriteString(task.ToString() + "\n") continue } - // 判断是否上传成功 - if task.BackupTaskID != "" { - taskStatus, statusMsg, job.Err = task.backupClient.TaskStatus(task.BackupTaskID) + if row.BackupTaskID != "" && job.backupClient != nil { + taskStatus, statusMsg, job.Err = job.backupClient.TaskStatus(row.BackupTaskID) if job.Err != nil { - tempHandler.WriteString(line + "\n") // 获取tasks状态失败,下次重试 + // 依然是失败的,下次继续重试 continue } // taskStatus>4,上传失败; // taskStatus==4,上传成功; - // taskStatus<4,上传中 + // taskStatus<4,上传中; if taskStatus > 4 { - if task.Status != consts.BackupStatusFailed { // 失败状态不重复上报 - task.Status = consts.BackupStatusFailed - task.Message = fmt.Sprintf("上传失败,err:%s", statusMsg) - task.BackupRecordReport() - line = task.ToString() + if row.Status != consts.BackupStatusToBakSystemFailed { // 失败状态不重复上报 + row.Status = consts.BackupStatusToBakSystemFailed + row.Message = fmt.Sprintf("上传失败,err:%s", statusMsg) + row.BackupRecordReport(job.Reporter) } - tempHandler.WriteString(line + "\n") // 上传失败,下次继续重试 } else if taskStatus < 4 { - tempHandler.WriteString(line + "\n") // 上传中,下次继续探测 + // 上传中,下次继续探测 + continue } else if taskStatus == 4 { // 上传成功 - task.Status = consts.BackupStatusToBakSysSuccess - task.Message = "上传备份系统成功" - task.BackupRecordReport() - doneHandler.WriteString(task.ToString() + "\n") + row.Status = consts.BackupStatusToBakSysSuccess + row.Message = "上传备份系统成功" + row.BackupRecordReport(job.Reporter) + } + // 更新记录 status 和 message + job.Err = job.sqdb.Save(&row).Error + if job.Err != nil { + job.Err = fmt.Errorf("gorm save fail,err:%v", job.Err) + mylog.Logger.Error(job.Err.Error()) } - } - // 其他失败的情况,写到done文件中 - if task.Status == consts.BackupStatusFailed { - doneHandler.WriteString(line + "\n") - continue } } - if job.Err = scanner.Err(); job.Err != nil { - job.Err = fmt.Errorf("scanner.Scan fail,err:%v,file:%v", job.Err, doingFile) - mylog.Logger.Error(job.Err.Error()) - return - } } -// DeleteTooOldBinlogbackup 根据 redis_binlog_file_list_{port}_done 删除太旧的本地文件 -// 将删除失败 or 不到OldFileLeftDay天数的task继续回写到 redis_binlog_file_list_{port}_done 文件中 +// DeleteTooOldBinlogbackup 删除 OldFileLeftDay天前的本地文件,15天前的记录 func (job *Job) DeleteTooOldBinlogbackup(port int) { - var doneHandler *os.File - task := Task{} - var line string + var toDoRows []RedisBinlogHistorySchema var err error - keepTasks := []string{} - oldFileLeftSec := job.Conf.RedisBinlogBackup.OldFileLeftDay * 24 * 3600 - nowTime := time.Now().Local() + var removeOK bool + NDaysAgo := time.Now().Local().AddDate(0, 0, -job.Conf.RedisFullBackup.OldFileLeftDay) + Days15Ago := time.Now().Local().AddDate(0, 0, -15) + mylog.Logger.Debug(fmt.Sprintf("port:%d start DeleteTooOldBinlogbackup", port)) - // 示例: /data/dbbak/binlog/30000/redis_binlog_file_list_30000_done - doneFile := filepath.Join(job.RealBackupDir, "binlog", - strconv.Itoa(port), fmt.Sprintf(consts.DoneRedisBinlogFileList, port)) - if !util.FileExists(doneFile) { + // 15 天以前的,本地文件已删除的,记录直接删除 + job.Err = job.sqdb.Where("start_time<=? and local_file_removed=?", Days15Ago, 1). + Delete(&RedisBinlogHistorySchema{}).Error + if job.Err != nil { + job.Err = fmt.Errorf( + "DeleteTooOldBinlogbackup gorm delete fail,err:%v,start_time:(%s) local_file_removed:%d", + job.Err, Days15Ago, 1) + mylog.Logger.Error(job.Err.Error()) return } - - defer func() { - if len(keepTasks) > 0 { - // 回写到 doneFile中 - done02, err01 := os.OpenFile(doneFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0755) - if err01 != nil { - job.Err = fmt.Errorf("os.Openfile fail,err:%v,file:%s", err01, doneFile) - mylog.Logger.Error(job.Err.Error()) - return - } - defer done02.Close() - for _, line := range keepTasks { - done02.WriteString(line + "\n") - } - } - }() - doneHandler, job.Err = os.Open(doneFile) - if job.Err != nil { - job.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doneFile, job.Err) + // OldFileLeftDay天以以前的,本地文件未删除的,remove本地文件,并记录下该行为 + job.Err = job.sqdb.Where("start_time<=? and local_file_removed=?", NDaysAgo, 0).Find(&toDoRows).Error + if job.Err != nil && job.Err != gorm.ErrRecordNotFound { + job.Err = fmt.Errorf("DeleteTooOldBinlogbackup gorm find fail,err:%v", job.Err) mylog.Logger.Error(job.Err.Error()) return } - defer doneHandler.Close() - - scanner := bufio.NewScanner(doneHandler) - scanner.Split(bufio.ScanLines) - for scanner.Scan() { - line = scanner.Text() - err = json.Unmarshal([]byte(line), &task) - if err != nil { - // json.Unmarshal failed,skip ... - err = fmt.Errorf("json.Unmarshal fail,err:%v,data:%s,file:%s", job.Err, line, doneFile) - mylog.Logger.Warn(err.Error()) - continue - } - if nowTime.Sub(task.EndTime.Time).Seconds() > float64(oldFileLeftSec) { - if util.FileExists(task.BackupFile) { - err = os.Remove(task.BackupFile) - if err != nil { - err = fmt.Errorf("os.Remove fail,err:%v,file:%s", err, task.BackupFile) - mylog.Logger.Warn(err.Error()) - keepTasks = append(keepTasks, line) // 删除失败的,下次继续重试 - } + job.Err = nil + for _, row := range toDoRows { + removeOK = true + if util.FileExists(row.BackupFile) { + err = os.Remove(row.BackupFile) + if err != nil { + err = fmt.Errorf("os.Remove fail,err:%v,file:%s", err, row.BackupFile) + mylog.Logger.Warn(err.Error()) + removeOK = false } - } else { - keepTasks = append(keepTasks, line) } - } - if err = scanner.Err(); err != nil { - job.Err = fmt.Errorf("scanner.Scan fail,err:%v,file:%v", err, doneFile) - mylog.Logger.Error(job.Err.Error()) - return + if !removeOK { + continue // 删除失败的,下次继续重试 + } + row.LocalFileRemoved = 1 + job.Err = job.sqdb.Save(&row).Error + if job.Err != nil { + job.Err = fmt.Errorf("gorm save fail,err:%v", job.Err) + mylog.Logger.Error(job.Err.Error()) + } } } diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/task.go b/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/task.go index 4c24874165..0752165c3b 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/task.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redisbinlogbackup/task.go @@ -15,11 +15,12 @@ import ( "dbm-services/redis/db-tools/dbmon/mylog" "dbm-services/redis/db-tools/dbmon/pkg/backupsys" "dbm-services/redis/db-tools/dbmon/pkg/consts" - "dbm-services/redis/db-tools/dbmon/pkg/customtime" "dbm-services/redis/db-tools/dbmon/pkg/report" "dbm-services/redis/db-tools/dbmon/util" "github.com/gofrs/flock" + "gorm.io/gorm" + "gorm.io/gorm/clause" ) // tendisssd binlog文件正则 @@ -28,36 +29,86 @@ import ( // 例如: binlog-0-0000887-20221107123830.log var tendisBinlogReg = regexp.MustCompile(`binlog-(\d+)-(\d+)-(\d+).log`) +// RedisBinlogHistorySchema TODO +type RedisBinlogHistorySchema struct { + ID int64 `json:"-" gorm:"primaryKey;column:id;not null` + ReportType string `json:"report_type" gorm:"column:report_type;not null;default:''"` + BkBizID string `json:"bk_biz_id" gorm:"column:bk_biz_id;not null;default:''"` + BkCloudID int64 `json:"bk_cloud_id" gorm:"column:bk_cloud_id;not null;default:0"` + ServerIP string `json:"server_ip" gorm:"column:server_ip;not null;default:''"` + ServerPort int `json:"server_port" gorm:"column:server_port;not null;default:0"` + Domain string `json:"domain" gorm:"column:domain;not null;default:'';index"` + // TendisplusInstance or TendisSSDInstance + DbType string `json:"db_type" gorm:"column:db_type;not null;default:''"` + RealRole string `json:"role" gorm:"column:role;not null;default:''"` + // 备份路径,如 /data/dbbak/binlog/30000 + BackupDir string `json:"backup_dir" gorm:"column:backup_dir;not null;default:''"` + // 备份的目标文件(已压缩) + BackupFile string `json:"backup_file" gorm:"column:backup_file;not null;default:''"` + // binlog对应的 kvstoreidx + KvstoreIdx int `json:"kvstoreidx" gorm:"column:kvstoreidx;not null;default:0"` + // 备份文件大小(已压缩) + BackupFileSize int64 `json:"backup_file_size" gorm:"column:backup_file_size;not null;default:0"` + // binlog文件生成时间(非压缩) + StartTime time.Time `json:"start_time" gorm:"column:start_time;not null;default:'';index"` + // binlog文件最后修改时间(非压缩) + EndTime time.Time `json:"end_time" gorm:"column:end_time;not null;default:'';index"` + TimeZone string `json:"time_zone" gorm:"column:time_zone;not null;default:''"` + BackupTaskID string `json:"backup_taskid" gorm:"column:backup_taskid;not null;default:''"` + // 目前为空 + BackupMD5 string `json:"backup_md5" gorm:"column:backup_md5;not null;default:''"` + // REDIS_BINLOG + BackupTag string `json:"backup_tag" gorm:"column:backup_tag;not null;default:''"` + // shard值 + ShardValue string `json:"shard_value" gorm:"column:shard_value;not null;default:''"` + Status string `json:"status" gorm:"column:status;not null;default:''"` + Message string `json:"message" gorm:"column:message;not null;default:''"` + // 本地文件是否已删除,未被删除为0,已被删除为1 + LocalFileRemoved int `json:"-" gorm:"column:local_file_removed;not null;default:0"` +} + +// TableName TODO +func (r *RedisBinlogHistorySchema) TableName() string { + return "redis_binlog_history" +} + +// Addr string +func (r *RedisBinlogHistorySchema) Addr() string { + return r.ServerIP + ":" + strconv.Itoa(r.ServerPort) +} + +type redisBinlogReport struct { + RedisBinlogHistorySchema + StartTime string `json:"start_time"` + EndTime string `json:"end_time"` +} + +// BackupRecordReport 备份记录上报 +func (r *RedisBinlogHistorySchema) BackupRecordReport(reporter report.Reporter) { + if reporter == nil { + return + } + reportRow := redisBinlogReport{ + RedisBinlogHistorySchema: *r, + StartTime: r.StartTime.Local().Format(time.RFC3339), + EndTime: r.EndTime.Local().Format(time.RFC3339), + } + tmpBytes, _ := json.Marshal(reportRow) + reporter.AddRecord(string(tmpBytes)+"\n", true) +} + // Task redis binlog备份task type Task struct { - ReportType string `json:"report_type"` - BkBizID string `json:"bk_biz_id"` - BkCloudID int64 `json:"bk_cloud_id"` - ServerIP string `json:"server_ip"` - ServerPort int `json:"server_port"` - Domain string `json:"domain"` - Password string `json:"-"` - ToBackupSystem string `json:"-"` - OldFileLeftDay int `json:"-"` - DbType string `json:"db_type"` // TendisplusInstance or TendisSSDInstance - RealRole string `json:"role"` - DumpDir string `json:"-"` - KvStoreCount int `json:"-"` - BackupDir string `json:"backup_dir"` // 备份路径,如 /data/dbbak/binlog/30000 - BackupFile string `json:"backup_file"` // 备份的目标文件(已压缩) - KvstoreIdx int `json:"kvstoreidx"` // binlog对应的 kvstoreidx - BackupFileSize int64 `json:"backup_file_size"` // 备份文件大小(已压缩) - StartTime customtime.CustomTime `json:"start_time"` // binlog文件生成时间(非压缩) - EndTime customtime.CustomTime `json:"end_time"` // binlog文件最后修改时间(非压缩) - BackupTaskID string `json:"backup_taskid"` - BackupMD5 string `json:"backup_md5"` // 目前为空 - BackupTag string `json:"backup_tag"` // REDIS_BINLOG - ShardValue string `json:"shard_value"` // shard值 - Status string `json:"status"` - Message string `json:"message"` - Cli *myredis.RedisClient `json:"-"` + RedisBinlogHistorySchema + Password string `json:"-"` + ToBackupSystem string `json:"-"` + OldFileLeftDay int `json:"-"` + DumpDir string `json:"-"` + KvStoreCount int `json:"-"` + Cli *myredis.RedisClient `json:"-"` reporter report.Reporter backupClient backupsys.BackupClient + sqdb *gorm.DB lockFile string `json:"-"` Err error `json:"-"` } @@ -65,22 +116,28 @@ type Task struct { // NewBinlogBackupTask new binlog backup task func NewBinlogBackupTask(bkBizID string, bkCloudID int64, domain, ip string, port int, password, toBackupSys, backupDir, shardValue string, oldFileLeftDay int, - reporter report.Reporter, storageType string) (ret *Task, err error) { + reporter report.Reporter, storageType string, + sqdb *gorm.DB) (ret *Task, err error) { + timeZone, _ := time.Now().Local().Zone() ret = &Task{ - ReportType: consts.RedisBinlogBackupReportType, - BkBizID: bkBizID, - BkCloudID: bkCloudID, - Domain: domain, - ServerIP: ip, - ServerPort: port, Password: password, ToBackupSystem: toBackupSys, OldFileLeftDay: oldFileLeftDay, - BackupDir: backupDir, - BackupTag: consts.RedisBinlogTAG, reporter: reporter, - ShardValue: shardValue, + sqdb: sqdb, + } + ret.RedisBinlogHistorySchema = RedisBinlogHistorySchema{ + ReportType: consts.RedisBinlogBackupReportType, + BkBizID: bkBizID, + BkCloudID: bkCloudID, + Domain: domain, + ServerIP: ip, + ServerPort: port, + BackupDir: backupDir, + BackupTag: consts.RedisBinlogTAG, + ShardValue: shardValue, + TimeZone: timeZone, } // ret.backupClient = backupsys.NewIBSBackupClient(consts.IBSBackupClient, consts.RedisBinlogTAG) ret.backupClient, err = backupsys.NewCosBackupClient(consts.COSBackupClient, @@ -92,11 +149,6 @@ func NewBinlogBackupTask(bkBizID string, bkCloudID int64, domain, ip string, por return ret, err } -// Addr string -func (task *Task) Addr() string { - return task.ServerIP + ":" + strconv.Itoa(task.ServerPort) -} - // ToString .. func (task *Task) ToString() string { tmpBytes, _ := json.Marshal(task) @@ -168,8 +220,8 @@ func (task *Task) BackupLocalBinlogs() { } task.BackupFile = item.File task.KvstoreIdx = item.KvStoreIdx - task.StartTime.Time = item.StartTime - task.EndTime.Time = item.FileMtime + task.StartTime = item.StartTime + task.EndTime = item.FileMtime task.compressAndUpload() // 无论成功还是失败,都继续下一个binlog file } } @@ -182,14 +234,16 @@ func (task *Task) newConnect() { if task.Err != nil { return } - task.DumpDir, task.Err = task.Cli.GetDumpDir() - if task.Err != nil { - return - } task.DbType, task.Err = task.Cli.GetTendisType() if task.Err != nil { return } + if task.DbType != consts.TendisTypeRedisInstance { + task.DumpDir, task.Err = task.Cli.GetDumpDir() + if task.Err != nil { + return + } + } // 除tendisplus外,其余db类型, kvstorecount=1 if task.DbType != consts.TendisTypeTendisplusInsance { task.KvStoreCount = 1 @@ -356,8 +410,8 @@ func (task *Task) mvBinlogToBackupDir() { } func (task *Task) compressAndUpload() { defer func() { - task.BackupRecordReport() - task.BackupRecordSaveToDoingFile() + task.BackupRecordReport(task.reporter) + task.BackupRecordSaveToLocalDB() }() if strings.HasSuffix(task.BackupFile, ".log") { task.mvBinlogToBackupDir() @@ -422,32 +476,17 @@ func (task *Task) TransferToBackupSystem() { return } -// BackupRecordReport 备份记录上报 -func (task *Task) BackupRecordReport() { - if task.reporter == nil { - return - } - tmpBytes, _ := json.Marshal(task) - // task.Err=task.reporter.AddRecord(string(tmpBytes),true) - task.reporter.AddRecord(string(tmpBytes)+"\n", true) -} - -// BackupRecordSaveToDoingFile 备份记录保存到本地 redis_binlog_file_list_${port}_doing 文件中 -func (task *Task) BackupRecordSaveToDoingFile() { - backupDir := filepath.Dir(task.BackupFile) - // 例如: /data/dbbak/binlog/30000/redis_binlog_file_list_30000_doing - doingFile := filepath.Join(backupDir, fmt.Sprintf(consts.DoingRedisBinlogFileList, task.ServerPort)) - f, err := os.OpenFile(doingFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if err != nil { - task.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doingFile, err) - mylog.Logger.Error(task.Err.Error()) +// BackupRecordSaveToLocalDB 备份记录保存到本地sqlite中 +func (task *Task) BackupRecordSaveToLocalDB() { + if task.sqdb == nil { return } - defer f.Close() - tmpBytes, _ := json.Marshal(task) - - if _, err = f.WriteString(string(tmpBytes) + "\n"); err != nil { - task.Err = fmt.Errorf("f.WriteString failed,err:%v,file:%s,line:%s", err, doingFile, string(tmpBytes)) + task.RedisBinlogHistorySchema.ID = 0 // 重置为0,以便gorm自增 + task.Err = task.sqdb.Clauses(clause.OnConflict{ + UpdateAll: true, + }).Create(&task.RedisBinlogHistorySchema).Error + if task.Err != nil { + task.Err = fmt.Errorf("BackupRecordSaveToLocalDB sqdb.Create fail,err:%v", task.Err) mylog.Logger.Error(task.Err.Error()) return } diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/backupjob.go b/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/backupjob.go index ba81ab7359..2f037204f1 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/backupjob.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/backupjob.go @@ -2,8 +2,6 @@ package redisfullbackup import ( - "bufio" - "encoding/json" "fmt" "os" "path/filepath" @@ -13,13 +11,14 @@ import ( "dbm-services/redis/db-tools/dbmon/config" "dbm-services/redis/db-tools/dbmon/models/myredis" + "dbm-services/redis/db-tools/dbmon/models/mysqlite" "dbm-services/redis/db-tools/dbmon/mylog" "dbm-services/redis/db-tools/dbmon/pkg/backupsys" "dbm-services/redis/db-tools/dbmon/pkg/consts" "dbm-services/redis/db-tools/dbmon/pkg/report" "dbm-services/redis/db-tools/dbmon/util" - "go.uber.org/zap" + "gorm.io/gorm" ) // globRedisFullBackupJob global var @@ -33,6 +32,7 @@ type Job struct { // NOCC:golint/naming(其他:设计如此) RealBackupDir string `json:"real_backup_dir"` Reporter report.Reporter `json:"-"` backupClient backupsys.BackupClient + sqdb *gorm.DB Err error `json:"-"` } @@ -48,7 +48,7 @@ func GetGlobRedisFullBackupJob(conf *config.Configuration) *Job { // Run 执行例行备份 func (job *Job) Run() { - mylog.Logger.Info("redisfullbackup wakeup,start running...", zap.String("conf", util.ToString(job.Conf))) + mylog.Logger.Info("redisfullbackup wakeup,start running...") defer func() { if job.Err != nil { mylog.Logger.Info(fmt.Sprintf("redisfullbackup end fail,err:%v", job.Err)) @@ -67,6 +67,12 @@ func (job *Job) Run() { } defer job.Reporter.Close() + job.getSqlDB() + if job.Err != nil { + return + } + defer job.closeDB() + // job.backupClient = backupsys.NewIBSBackupClient(consts.IBSBackupClient, consts.RedisFullBackupTAG) job.backupClient, job.Err = backupsys.NewCosBackupClient(consts.COSBackupClient, consts.COSInfoFile, consts.RedisFullBackupTAG, job.Conf.BackupClientStrorageType) @@ -94,8 +100,8 @@ func (job *Job) Run() { continue } for _, port := range svrItem.ServerPorts { + job.DeleteTooOldFullBackup(port) job.CheckOldFullbackupStatus(port) - job.DeleteTooOldFullbackup(port) } } } @@ -108,7 +114,7 @@ func (job *Job) GetRealBackupDir() { util.MkDirsIfNotExists([]string{ filepath.Join(job.RealBackupDir, "backup"), }) - util.LocalDirChownMysql(job.RealBackupDir) + // util.LocalDirChownMysql(job.RealBackupDir) } // GetReporter 上报者 @@ -120,6 +126,23 @@ func (job *Job) GetReporter() { job.Reporter, job.Err = report.NewFileReport(filepath.Join(reportDir, reportFile)) } +func (job *Job) getSqlDB() { + job.sqdb, job.Err = mysqlite.GetLocalSqDB() + if job.Err != nil { + return + } + job.Err = job.sqdb.AutoMigrate(&RedisFullbackupHistorySchema{}) + if job.Err != nil { + job.Err = fmt.Errorf("RedisFullbackupHistorySchema AutoMigrate fail,err:%v", job.Err) + mylog.Logger.Info(job.Err.Error()) + return + } +} + +func (job *Job) closeDB() { + mysqlite.CloseDB(job.sqdb) +} + func (job *Job) createTasks() { var task *BackupTask var password string @@ -144,244 +167,136 @@ func (job *Job) createTasks() { consts.NormalBackupType, svrItem.CacheBackupMode, job.RealBackupDir, job.Conf.RedisFullBackup.TarSplit, job.Conf.RedisFullBackup.TarSplitPartSize, svrItem.ServerShards[instStr], job.Reporter, - job.Conf.BackupClientStrorageType) + job.Conf.BackupClientStrorageType, + job.sqdb) if job.Err != nil { return } job.Tasks = append(job.Tasks, task) } } - mylog.Logger.Info(fmt.Sprintf("redisfullbackup createTasks tasks:%s", util.ToString(job.Tasks))) + mylog.Logger.Debug(fmt.Sprintf("redisfullbackup createTasks tasks:%s", util.ToString(job.Tasks))) } -// CheckOldFullbackupStatus 检查历史全备任务状态 -// 1. 遍历 redis_backup_file_list_${port}_doing 文件 -// 2. 已超过时间的任务,删除本地文件,从 redis_backup_file_list_${port}_doing 中剔除 -// 3. 上传备份系统 运行中 or 失败的任务 记录到 redis_backup_file_list_${port}_doing_temp -// 4. 已成功的任务,记录到 redis_backup_file_list_${port}_done -// 5. rename redis_backup_file_list_${port}_doing_temp to redis_backup_file_list_${port}_doing +// CheckOldFullbackupStatus 重试备份系统上传失败的,检查备份系统上传中的是否成功 func (job *Job) CheckOldFullbackupStatus(port int) { mylog.Logger.Info(fmt.Sprintf("port:%d start CheckOldFullbackupStatus", port)) - var doingHandler, tempHandler, doneHandler *os.File - var line string - task := BackupTask{} - var err error + toCheckRows := []RedisFullbackupHistorySchema{} var taskStatus int var statusMsg string - oldFileLeftSec := job.Conf.RedisFullBackup.OldFileLeftDay * 24 * 3600 - nowTime := time.Now().Local() - // 示例: /data/dbbak/backup/redis_backup_file_list_30000_doing - doingFile := filepath.Join(job.RealBackupDir, "backup", fmt.Sprintf(consts.DoingRedisFullBackFileList, port)) - if !util.FileExists(doingFile) { - return - } - // 示例: /data/dbbak/backup/redis_backup_file_list_30000_doing_temp - tempDoingFile := doingFile + "_temp" - // 示例: /data/dbbak/backup/redis_backup_file_list_30000_done - doneFile := filepath.Join(job.RealBackupDir, "backup", fmt.Sprintf(consts.DoneRedisFullBackFileList, port)) - - defer func() { - if job.Err == nil { - mylog.Logger.Info(fmt.Sprintf("rename %s to %s", tempDoingFile, doingFile)) - os.Rename(tempDoingFile, doingFile) // rename - } - }() - - doingHandler, job.Err = os.Open(doingFile) - if job.Err != nil { - job.Err = fmt.Errorf("os.Open file:%s fail,err:%v", doingFile, job.Err) - mylog.Logger.Error(job.Err.Error()) - return - } - defer doingHandler.Close() - - tempHandler, job.Err = os.OpenFile(tempDoingFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if job.Err != nil { - job.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", tempDoingFile, job.Err) - mylog.Logger.Error(job.Err.Error()) - return - } - defer tempHandler.Close() - - doneHandler, job.Err = os.OpenFile(doneFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if job.Err != nil { - job.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doneFile, job.Err) + job.Err = job.sqdb.Where("status not in (?,?,?)", + consts.BackupStatusToBakSysSuccess, + consts.BackupStatusFailed, + consts.BackupStatusLocalSuccess, + ).Find(&toCheckRows).Error + if job.Err != nil && job.Err != gorm.ErrRecordNotFound { + job.Err = fmt.Errorf("CheckOldFullbackupStatus gorm find fail,err:%v", job.Err) mylog.Logger.Error(job.Err.Error()) return } - defer doneHandler.Close() - - scanner := bufio.NewScanner(doingHandler) - scanner.Split(bufio.ScanLines) - for scanner.Scan() { - line = scanner.Text() - err = json.Unmarshal([]byte(line), &task) - if err != nil { - // json.Unmarshal failed,skip ... - err = fmt.Errorf("json.Unmarshal fail,err:%s,data:%s,skip it", err, line) - mylog.Logger.Error(err.Error()) - continue - } - task.reporter = job.Reporter - task.backupClient = job.backupClient - // 删除旧文件 - if nowTime.Sub(task.EndTime.Time).Seconds() > float64(oldFileLeftSec) { - mylog.Logger.Info(fmt.Sprintf("%+v start removing...", task.BackupFile)) - removeOK := true - if util.FileExists(task.BackupFile) { - err = os.Remove(task.BackupFile) - if err != nil { - err = fmt.Errorf("os.Remove fail,err:%s,file:%s", err, task.BackupFile) - mylog.Logger.Error(err.Error()) - removeOK = false - } + job.Err = nil + for _, row := range toCheckRows { + if row.Status == consts.BackupStatusToBakSystemFailed && job.backupClient != nil { + // 重试备份系统上传失败的 + if !util.FileExists(row.BackupFile) { + continue } - if !removeOK { - mylog.Logger.Info(fmt.Sprintf("%+v remove fail,continue add tempFile", task.BackupFile)) - _, job.Err = tempHandler.WriteString(line + "\n") // 删除失败的,记录到temp文件,下次继续重试 - if job.Err != nil { - job.Err = fmt.Errorf("%s WriteString fail,err:%v", tempDoingFile, job.Err) - mylog.Logger.Error(job.Err.Error()) - return - } + mylog.Logger.Info(fmt.Sprintf("redis(%s) backupFiles:%+v retry upload backupSystem", row.Addr(), row.BackupFile)) + row.BackupTaskID, job.Err = job.backupClient.Upload(row.BackupFile) + if job.Err != nil { + row.Message = job.Err.Error() + } else { + row.Status = consts.BackupStatusToBakSystemStart + row.Message = "上传备份系统中" } - continue - } - // 无需上传备份系统,本地已备份成功的情况 - if task.Status == consts.BackupStatusLocalSuccess { - _, job.Err = doneHandler.WriteString(line + "\n") + // 更新记录 status 和 message + job.Err = job.sqdb.Save(&row).Error if job.Err != nil { - job.Err = fmt.Errorf("%s WriteString fail,err:%v", doneFile, job.Err) + job.Err = fmt.Errorf("gorm save fail,err:%v", job.Err) mylog.Logger.Error(job.Err.Error()) - return } continue } - // 上传备份系统失败的情况,重试上传并写入temp文件中 - if task.Status == consts.BackupStatusToBakSystemFailed { - task.TransferToBackupSystem() - if task.Err != nil { - task.Message = task.Err.Error() - } else { - task.Status = consts.BackupStatusToBakSystemStart - task.Message = "上传备份系统中" - } - tempHandler.WriteString(task.ToString() + "\n") - continue - } // 判断是否上传成功 - if task.BackupTaskID != "" { - taskStatus, statusMsg, job.Err = task.backupClient.TaskStatus(task.BackupTaskID) + if row.BackupTaskID != "" && job.backupClient != nil { + taskStatus, statusMsg, job.Err = job.backupClient.TaskStatus(row.BackupTaskID) if job.Err != nil { - tempHandler.WriteString(line + "\n") // 获取tasks状态失败,下次重试 + // 依然是失败的,下次继续重试 continue } // taskStatus>4,上传失败; // taskStatus==4,上传成功; // taskStatus<4,上传中; if taskStatus > 4 { - if task.Status != consts.BackupStatusFailed { // 失败状态不重复上报 - task.Status = consts.BackupStatusFailed - task.Message = fmt.Sprintf("上传失败,err:%s", statusMsg) - task.BackupRecordReport() - line = task.ToString() + if row.Status != consts.BackupStatusToBakSystemFailed { // 失败状态不重复上报 + row.Status = consts.BackupStatusToBakSystemFailed + row.Message = fmt.Sprintf("上传失败,err:%s", statusMsg) + row.BackupRecordReport(job.Reporter) } - tempHandler.WriteString(line + "\n") // 上传失败,下次继续重试 } else if taskStatus < 4 { - tempHandler.WriteString(line + "\n") // 上传中,下次继续探测 + // 上传中,下次继续探测 + continue } else if taskStatus == 4 { // 上传成功 - task.Status = consts.BackupStatusToBakSysSuccess - task.Message = "上传备份系统成功" - task.BackupRecordReport() - doneHandler.WriteString(task.ToString() + "\n") + row.Status = consts.BackupStatusToBakSysSuccess + row.Message = "上传备份系统成功" + row.BackupRecordReport(job.Reporter) + } + // 更新记录 status 和 message + job.Err = job.sqdb.Save(&row).Error + if job.Err != nil { + job.Err = fmt.Errorf("gorm save fail,err:%v", job.Err) + mylog.Logger.Error(job.Err.Error()) } } - // 其他失败的情况,写到done文件中 - if task.Status == consts.BackupStatusFailed { - doneHandler.WriteString(line + "\n") - continue - } - } - if job.Err = scanner.Err(); job.Err != nil { - job.Err = fmt.Errorf("scanner.Scan fail,err:%v,file:%v", job.Err, doingFile) - mylog.Logger.Error(job.Err.Error()) - return } } -// DeleteTooOldFullbackup 根据 redis_backup_file_list_{port}_done 删除太旧的本地文件 -// 将删除失败 or 不到OldFileLeftDay天数的task继续回写到 redis_backup_file_list_{port}_done 文件中 -func (job *Job) DeleteTooOldFullbackup(port int) { - var doneHandler *os.File - task := BackupTask{} - var line string +// DeleteTooOldFullBackup 删除 OldFileLeftDay天前的本地文件,15天前的记录 +func (job *Job) DeleteTooOldFullBackup(port int) { + var toDoRows []RedisFullbackupHistorySchema var err error - keepTasks := []string{} - oldFileLeftSec := job.Conf.RedisFullBackup.OldFileLeftDay * 24 * 3600 - nowTime := time.Now().Local() + var removeOK bool + NDaysAgo := time.Now().Local().AddDate(0, 0, -job.Conf.RedisFullBackup.OldFileLeftDay) + Days15Ago := time.Now().Local().AddDate(0, 0, -15) + mylog.Logger.Info(fmt.Sprintf("port:%d start DeleteTooOldFullBackup", port)) - mylog.Logger.Info(fmt.Sprintf("port:%d start DeleteTooOldFullbackup", port)) - // 示例: /data/dbbak/backup/redis_backup_file_list_30000_done - doneFile := filepath.Join(job.RealBackupDir, "backup", fmt.Sprintf(consts.DoneRedisFullBackFileList, port)) - if !util.FileExists(doneFile) { + // 15 天以前的,本地文件已删除的,记录直接删除 + job.Err = job.sqdb.Where("start_time<=? and local_file_removed=?", Days15Ago, 1). + Delete(&RedisFullbackupHistorySchema{}).Error + if job.Err != nil { + job.Err = fmt.Errorf( + "DeleteTooOldFullBackup gorm delete fail,err:%v,start_time:(%s) local_file_removed:%d", + job.Err, Days15Ago, 1) + mylog.Logger.Error(job.Err.Error()) return } - - defer func() { - if len(keepTasks) > 0 { - // 回写到 doneFile中 - done02, err01 := os.OpenFile(doneFile, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0755) - if err01 != nil { - job.Err = fmt.Errorf("os.Openfile fail,err:%v,file:%s", err01, doneFile) - mylog.Logger.Error(job.Err.Error()) - return - } - defer done02.Close() - for _, line := range keepTasks { - done02.WriteString(line + "\n") - } - } - }() - doneHandler, job.Err = os.Open(doneFile) - if job.Err != nil { - job.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doneFile, job.Err) + // OldFileLeftDay天以以前的,本地文件未删除的,remove本地文件,并记录下该行为 + job.Err = job.sqdb.Where("start_time<=? and local_file_removed=?", NDaysAgo, 0).Find(&toDoRows).Error + if job.Err != nil && job.Err != gorm.ErrRecordNotFound { + job.Err = fmt.Errorf("DeleteTooOldFullBackup gorm find fail,err:%v", job.Err) mylog.Logger.Error(job.Err.Error()) return } - defer doneHandler.Close() - - scanner := bufio.NewScanner(doneHandler) - scanner.Split(bufio.ScanLines) - for scanner.Scan() { - line = scanner.Text() - err = json.Unmarshal([]byte(line), &task) - if err != nil { - // json.Unmarshal failed,skip ... - err = fmt.Errorf("json.Unmarshal fail,err:%v,data:%s,file:%s", err, line, doneFile) - mylog.Logger.Warn(err.Error()) - continue - } - if nowTime.Sub(task.EndTime.Time).Seconds() > float64(oldFileLeftSec) { - removeOK := true - if util.FileExists(task.BackupFile) { - err = os.Remove(task.BackupFile) - if err != nil { - err = fmt.Errorf("os.Remove fail,err:%v,file:%s", err, task.BackupFile) - mylog.Logger.Warn(err.Error()) - removeOK = false - } - } - if !removeOK { - keepTasks = append(keepTasks, line) // 删除失败的,下次继续重试 + job.Err = nil + for _, row := range toDoRows { + removeOK = true + if util.FileExists(row.BackupFile) { + err = os.Remove(row.BackupFile) + if err != nil { + err = fmt.Errorf("os.Remove fail,err:%v,file:%s", err, row.BackupFile) + mylog.Logger.Warn(err.Error()) + removeOK = false } - } else { - keepTasks = append(keepTasks, line) } - } - if err = scanner.Err(); err != nil { - job.Err = fmt.Errorf("scanner.Scan fail,err:%v,file:%v", err, doneFile) - mylog.Logger.Error(job.Err.Error()) - return + if !removeOK { + continue // 删除失败的,下次继续重试 + } + row.LocalFileRemoved = 1 + job.Err = job.sqdb.Save(&row).Error + if job.Err != nil { + job.Err = fmt.Errorf("gorm save fail,err:%v", job.Err) + mylog.Logger.Error(job.Err.Error()) + } } } diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/checkjob.go b/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/checkjob.go index d641ef08c9..3ea6fcb906 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/checkjob.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/checkjob.go @@ -58,6 +58,12 @@ func (job *CheckJob) Run() { } defer job.Reporter.Close() + job.getSqlDB() + if job.Err != nil { + return + } + defer job.closeDB() + // job.backupClient = backupsys.NewIBSBackupClient(consts.IBSBackupClient, consts.RedisFullBackupTAG) job.backupClient, job.Err = backupsys.NewCosBackupClient(consts.COSBackupClient, consts.COSInfoFile, consts.RedisFullBackupTAG, job.Conf.BackupClientStrorageType) @@ -72,8 +78,8 @@ func (job *CheckJob) Run() { continue } for _, port := range svrItem.ServerPorts { + job.DeleteTooOldFullBackup(port) job.CheckOldFullbackupStatus(port) - job.DeleteTooOldFullbackup(port) } } } diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/task.go b/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/task.go index 9752e1db69..56f19bbb69 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/task.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redisfullbackup/task.go @@ -14,13 +14,79 @@ import ( "dbm-services/redis/db-tools/dbmon/mylog" "dbm-services/redis/db-tools/dbmon/pkg/backupsys" "dbm-services/redis/db-tools/dbmon/pkg/consts" - "dbm-services/redis/db-tools/dbmon/pkg/customtime" "dbm-services/redis/db-tools/dbmon/pkg/report" "dbm-services/redis/db-tools/dbmon/util" "github.com/gofrs/flock" + "gorm.io/gorm" + "gorm.io/gorm/clause" ) +// RedisFullbackupHistorySchema TODO +type RedisFullbackupHistorySchema struct { + ID int64 `json:"-" gorm:"primaryKey;column:id;not null` + ReportType string `json:"report_type" gorm:"column:report_type;not null;default:''"` + BkBizID string `json:"bk_biz_id" gorm:"column:bk_biz_id;not null;default:''"` + BkCloudID int64 `json:"bk_cloud_id" gorm:"column:bk_cloud_id;not null;default:0"` + ServerIP string `json:"server_ip" gorm:"column:server_ip;not null;default:''"` + ServerPort int `json:"server_port" gorm:"column:server_port;not null;default:0"` + Domain string `json:"domain" gorm:"column:domain;not null;default:'';index"` + // RedisInstance or TendisplusInstance or TendisSSDInstance + DbType string `json:"db_type" gorm:"column:db_type;not null;default:''"` + RealRole string `json:"role" gorm:"column:role;not null;default:''"` + BackupDir string `json:"backup_dir" gorm:"column:backup_dir;not null;default:''"` + // 备份的目标文件 + BackupFile string `json:"backup_file" gorm:"column:backup_file;not null;default:''"` + // 备份文件大小(已切割 or 已压缩 or 已打包) + BackupFileSize int64 `json:"backup_file_size" gorm:"column:backup_file_size;not null;default:0"` + BackupTaskID string `json:"backup_taskid" gorm:"column:backup_taskid;not null;default:''"` + // 目前为空 + BackupMD5 string `json:"backup_md5" gorm:"column:backup_md5;not null;default:''"` + // REDIS_FULL + BackupTag string `json:"backup_tag" gorm:"column:backup_tag;not null;default:''"` + // shard值 + ShardValue string `json:"shard_value" gorm:"column:shard_value;not null;default:''"` + // 生成全备的起始时间 + StartTime time.Time `json:"start_time" gorm:"column:start_time;not null;default:'';index"` + // 生成全备的结束时间 + EndTime time.Time `json:"end_time" gorm:"column:end_time;not null;default:'';index"` + TimeZone string `json:"time_zone" gorm:"column:time_zone;not null;default:''"` + Status string `json:"status" gorm:"column:status;not null;default:''"` + Message string `json:"message" gorm:"column:message;not null;default:''"` + // 本地文件是否已删除,未被删除为0,已被删除为1 + LocalFileRemoved int `json:"-" gorm:"column:local_file_removed;not null;default:0"` +} + +// TableName TODO +func (r *RedisFullbackupHistorySchema) TableName() string { + return "redis_fullbackup_history" +} + +// Addr string +func (r *RedisFullbackupHistorySchema) Addr() string { + return r.ServerIP + ":" + strconv.Itoa(r.ServerPort) +} + +type redisFullBackupReport struct { + RedisFullbackupHistorySchema + StartTime string `json:"start_time"` + EndTime string `json:"end_time"` +} + +// BackupRecordReport 备份记录上报 +func (r *RedisFullbackupHistorySchema) BackupRecordReport(reporter report.Reporter) { + if reporter == nil { + return + } + reportRow := redisFullBackupReport{ + RedisFullbackupHistorySchema: *r, + StartTime: r.StartTime.Local().Format(time.RFC3339), + EndTime: r.EndTime.Local().Format(time.RFC3339), + } + tmpBytes, _ := json.Marshal(reportRow) + reporter.AddRecord(string(tmpBytes)+"\n", true) +} + // TendisSSDSetLogCount tendisSSD设置log参数 type TendisSSDSetLogCount struct { LogCount int64 `json:"log-count"` @@ -29,63 +95,53 @@ type TendisSSDSetLogCount struct { // BackupTask redis备份task type BackupTask struct { - ReportType string `json:"report_type"` - BkBizID string `json:"bk_biz_id"` - BkCloudID int64 `json:"bk_cloud_id"` - ServerIP string `json:"server_ip"` - ServerPort int `json:"server_port"` - Domain string `json:"domain"` - Password string `json:"-"` - ToBackupSystem string `json:"-"` - DbType string `json:"db_type"` // RedisInstance or TendisplusInstance or TendisSSDInstance - BackupType string `json:"-"` // 常规备份、下线备份 - CacheBackupMode string `json:"-"` // aof or rdb - RealRole string `json:"role"` - DataSize uint64 `json:"-"` // redis实例数据大小 - DataDir string `json:"-"` - BackupDir string `json:"backup_dir"` - TarSplit bool `json:"-"` // 是否对tar文件做split - TarSplitPartSize string `json:"-"` - BackupFile string `json:"backup_file"` // 备份的目标文件,如果文件过大会切割成多个 - BackupFileSize int64 `json:"backup_file_size"` // 备份文件大小(已切割 or 已压缩 or 已打包) - BackupTaskID string `json:"backup_taskid"` - BackupMD5 string `json:"backup_md5"` // 目前为空 - BackupTag string `json:"backup_tag"` // REDIS_FULL - ShardValue string `json:"shard_value"` // shard值 - StartTime customtime.CustomTime `json:"start_time"` // 生成全备的起始时间 - EndTime customtime.CustomTime `json:"end_time"` // //生成全备的结束时间 - Status string `json:"status"` - Message string `json:"message"` - Cli *myredis.RedisClient `json:"-"` - SSDLogCount TendisSSDSetLogCount `json:"-"` + RedisFullbackupHistorySchema + Password string `json:"-"` + ToBackupSystem string `json:"-"` + BackupType string `json:"-"` // 常规备份、下线备份 + CacheBackupMode string `json:"-"` // aof or rdb + DataSize uint64 `json:"-"` // redis实例数据大小 + DataDir string `json:"-"` + TarSplit bool `json:"-"` // 是否对tar文件做split + TarSplitPartSize string `json:"-"` + Cli *myredis.RedisClient `json:"-"` + SSDLogCount TendisSSDSetLogCount `json:"-"` reporter report.Reporter backupClient backupsys.BackupClient + sqdb *gorm.DB Err error `json:"-"` } // NewFullBackupTask new backup task func NewFullBackupTask(bkBizID string, bkCloudID int64, domain, ip string, port int, password, - toBackupSys, backupType, cacheBackupMode, backupDir string, tarSplit bool, tarSplitSize, shardValue string, - reporter report.Reporter, storageType string) (ret *BackupTask, err error) { + toBackupSys, backupType, cacheBackupMode, backupDir string, tarSplit bool, + tarSplitSize, shardValue string, + reporter report.Reporter, storageType string, + sqdb *gorm.DB) (ret *BackupTask, err error) { ret = &BackupTask{ - ReportType: consts.RedisFullBackupReportType, - BkBizID: bkBizID, - BkCloudID: bkCloudID, - Domain: domain, - ServerIP: ip, - ServerPort: port, Password: password, ToBackupSystem: toBackupSys, BackupType: backupType, CacheBackupMode: cacheBackupMode, - BackupDir: backupDir, TarSplit: tarSplit, TarSplitPartSize: tarSplitSize, - BackupTaskID: "", - BackupMD5: "", - BackupTag: consts.RedisFullBackupTAG, - ShardValue: shardValue, reporter: reporter, + sqdb: sqdb, + } + timeZone, _ := time.Now().Local().Zone() + ret.RedisFullbackupHistorySchema = RedisFullbackupHistorySchema{ + ReportType: consts.RedisFullBackupReportType, + BkBizID: bkBizID, + BkCloudID: bkCloudID, + Domain: domain, + ServerIP: ip, + ServerPort: port, + BackupDir: backupDir, + BackupTaskID: "", + BackupMD5: "", + BackupTag: consts.RedisFullBackupTAG, + TimeZone: timeZone, + ShardValue: shardValue, } // ret.backupClient = backupsys.NewIBSBackupClient(consts.IBSBackupClient, consts.RedisFullBackupTAG) ret.backupClient, err = backupsys.NewCosBackupClient(consts.COSBackupClient, @@ -97,11 +153,6 @@ func NewFullBackupTask(bkBizID string, bkCloudID int64, domain, ip string, port return } -// Addr string -func (task *BackupTask) Addr() string { - return task.ServerIP + ":" + strconv.Itoa(task.ServerPort) -} - // ToString .. func (task *BackupTask) ToString() string { tmpBytes, _ := json.Marshal(task) @@ -124,7 +175,7 @@ func (task *BackupTask) BakcupToLocal() { mylog.Logger.Info(fmt.Sprintf("redis(%s) is master and has slaves,skip backup", task.Addr())) return } - // 如果是 tendisSSD 或 tenidsplus,早上12点后不做备份 + // 如果是 tendisSSD 或 tenidsplus,早上12点后不做全备 if task.DbType == consts.TendisTypeTendisplusInsance || task.DbType == consts.TendisTypeTendisSSDInsance { if time.Now().Local().Hour() >= 12 { @@ -163,12 +214,12 @@ func (task *BackupTask) BakcupToLocal() { task.Message = task.Err.Error() task.Status = consts.BackupStatusFailed } - task.BackupRecordReport() + task.BackupRecordReport(task.reporter) }() task.Status = consts.BackupStatusRunning task.Message = "start backup..." - task.BackupRecordReport() + task.BackupRecordReport(task.reporter) mylog.Logger.Info(fmt.Sprintf("redis(%s) dbType:%s start backup...", task.Addr(), task.DbType)) @@ -196,7 +247,7 @@ func (task *BackupTask) BakcupToLocal() { if task.Err != nil { return } - defer task.BackupRecordSaveToDoingFile() + defer task.BackupRecordSaveToLocalDB() // 备份上传备份系统 if strings.ToLower(task.ToBackupSystem) != "yes" { task.Status = consts.BackupStatusLocalSuccess @@ -316,7 +367,7 @@ func (task *BackupTask) RedisInstanceBackup() { var confMap map[string]string var fileSize int64 nowtime := time.Now().Local().Format(consts.FilenameTimeLayout) - task.StartTime.Time = time.Now().Local() + task.StartTime = time.Now().Local() if task.RealRole == consts.RedisMasterRole || task.CacheBackupMode == consts.CacheBackupModeRdb { // redis master backup rdb @@ -340,7 +391,7 @@ func (task *BackupTask) RedisInstanceBackup() { if task.Err != nil { return } - task.EndTime.Time = time.Now().Local() + task.EndTime = time.Now().Local() cpCmd := fmt.Sprintf("cp %s %s", srcFile, targetFile) mylog.Logger.Info(cpCmd) _, task.Err = util.RunBashCmd(cpCmd, "", nil, 10*time.Minute) @@ -379,12 +430,12 @@ func (task *BackupTask) TendisplusInstanceBackup() { return } util.LocalDirChownMysql(task.BackupDir) - task.StartTime.Time = time.Now().Local() + task.StartTime = time.Now().Local() task.Err = task.Cli.TendisplusBackupAndWaitForDone(backupFullDir) if task.Err != nil { return } - task.EndTime.Time = time.Now().Local() + task.EndTime = time.Now().Local() tarFile, task.Err = util.TarADir(backupFullDir, task.BackupDir, true) task.BackupFile = tarFile if task.Err != nil { @@ -437,12 +488,12 @@ func (task *BackupTask) TendisSSDInstanceBackup() { return } util.LocalDirChownMysql(task.BackupDir) - task.StartTime.Time = time.Now().Local() + task.StartTime = time.Now().Local() binlogsizeRet, _, task.Err = task.Cli.TendisSSDBackupAndWaitForDone(backupFullDir) if task.Err != nil { return } - task.EndTime.Time = time.Now().Local() + task.EndTime = time.Now().Local() task.tendisSSDBackupVerify(backupFullDir) if task.Err != nil { @@ -528,32 +579,17 @@ func (task *BackupTask) TransferToBackupSystem() { return } -// BackupRecordReport 备份记录上报 -func (task *BackupTask) BackupRecordReport() { - if task.reporter == nil { - return - } - tmpBytes, _ := json.Marshal(task) - // task.Err=task.reporter.AddRecord(string(tmpBytes),true) - task.reporter.AddRecord(string(tmpBytes)+"\n", true) -} - -// BackupRecordSaveToDoingFile 备份记录保存到本地 redis_backup_file_list_${port}_doing 文件中 -func (task *BackupTask) BackupRecordSaveToDoingFile() { - backupDir := filepath.Dir(task.BackupFile) - // 例如: /data/dbbak/backup/redis_backup_file_list_30000_doing - doingFile := filepath.Join(backupDir, "backup", fmt.Sprintf(consts.DoingRedisFullBackFileList, task.ServerPort)) - f, err := os.OpenFile(doingFile, os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0744) - if err != nil { - task.Err = fmt.Errorf("os.OpenFile %s failed,err:%v", doingFile, err) - mylog.Logger.Error(task.Err.Error()) +// BackupRecordSaveToLocalDB 备份信息记录到本地sqlite中 +func (task *BackupTask) BackupRecordSaveToLocalDB() { + if task.sqdb == nil { return } - defer f.Close() - tmpBytes, _ := json.Marshal(task) - - if _, err = f.WriteString(string(tmpBytes) + "\n"); err != nil { - task.Err = fmt.Errorf("f.WriteString failed,err:%v,file:%s,line:%s", err, doingFile, string(tmpBytes)) + task.RedisFullbackupHistorySchema.ID = 0 // 重置为0,以便gorm自增 + task.Err = task.sqdb.Clauses(clause.OnConflict{ + UpdateAll: true, + }).Create(&task.RedisFullbackupHistorySchema).Error + if task.Err != nil { + task.Err = fmt.Errorf("BackupRecordSaveToLocalDB sqdb.Create fail,err:%v", task.Err) mylog.Logger.Error(task.Err.Error()) return } diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redismonitor/redis_task.go b/dbm-services/redis/db-tools/dbmon/pkg/redismonitor/redis_task.go index 7fbb600dee..9e33e7ee6c 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redismonitor/redis_task.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redismonitor/redis_task.go @@ -53,6 +53,10 @@ func (task *RedisMonitorTask) RunMonitor() { if task.Err != nil { return } + task.TendisplusMasterSetHeartbeat() + if task.Err != nil { + return + } task.CheckSyncOnSlave() if task.Err != nil { return @@ -134,11 +138,44 @@ func (task *RedisMonitorTask) SetDbmonKeyOnMaster() { } } +// TendisplusMasterSetHeartbeat tendisplus写入心跳失败则告警 +func (task *RedisMonitorTask) TendisplusMasterSetHeartbeat() { + var dbtype, role, beatKey string + var nowVal int64 + for idx, cli01 := range task.redisClis { + cliItem := cli01 + task.eventSender.SetInstance(cliItem.Addr) + dbtype, task.Err = cliItem.GetTendisType() + if task.Err != nil { + continue + } + role, task.Err = cliItem.GetRole() + if task.Err != nil { + continue + } + if dbtype != consts.TendisTypeTendisplusInsance || role != consts.RedisMasterRole { + continue + } + beatKey = fmt.Sprintf("%s_%d:heartbeat", + task.ServerConf.ServerIP, + task.ServerConf.ServerPorts[idx]) + nowVal = time.Now().Local().Unix() + _, task.Err = cliItem.AdminSet(beatKey, strconv.FormatInt(nowVal, 10)) + if task.Err != nil { + task.eventSender.SendWarning(consts.EventRedisLogin, task.Err.Error(), + consts.WarnLevelError, task.ServerConf.ServerIP, + ) + return + } + } +} + // setSlaveDbmonKey 从slave上连接master,并执行set dbmon:$slaveIP:$slaveport $time func (task *RedisMonitorTask) setSlaveDbmonKey(selfAddr, masterIP, masterPort string) { var masterCli *myredis.RedisClient = nil var msg, dbmonKey string var clusterEnabled bool + var masterRole string masterAddr := masterIP + ":" + masterPort masterCli, task.Err = myredis.NewRedisClientWithTimeout(masterAddr, task.Password, 0, consts.TendisTypeRedisInstance, 5*time.Second) @@ -159,6 +196,20 @@ func (task *RedisMonitorTask) setSlaveDbmonKey(selfAddr, masterIP, masterPort st return } + // 如果'我'的master,他的role依然是slave,说明是链式关系(m->s->s),则不写入dbmon:* keys + masterRole, task.Err = masterCli.GetRole() + if task.Err != nil { + msg = fmt.Sprintf("redis(%s) master(%s) get role fail", selfAddr, masterAddr) + mylog.Logger.Error(msg) + task.eventSender.SendWarning(consts.EventRedisSync, msg, consts.WarnLevelError, task.ServerConf.ServerIP) + return + } + if masterRole == consts.RedisSlaveRole { + msg = fmt.Sprintf("redis(%s) master(%s) role is slave", selfAddr, masterAddr) + mylog.Logger.Debug(msg) + return + } + task.Err = masterCli.SelectDB1WhenClusterDisabled() if task.Err != nil { return @@ -297,6 +348,11 @@ func (task *RedisMonitorTask) CheckPersist() { if dbtype != consts.TendisTypeRedisInstance || role != consts.RedisSlaveRole { continue } + if role == consts.RedisSlaveRole && connectedSlaves > 0 { + // 如果'我'是slave,我还有自己的slave + // 说明是链式关系(m->s->s),估计在扩容等,跳过检查 + continue + } // 检查 cache redis slave, aof 是否开启 // TODO 无法知道远程配置的情况下,如果是人为关闭的aof,如何不告警 confmap, task.Err = cliItem.ConfigGet("appendonly") @@ -306,8 +362,8 @@ func (task *RedisMonitorTask) CheckPersist() { appendonly, _ = confmap["appendonly"] if task.ServerConf.CacheBackupMode == consts.CacheBackupModeRdb { if strings.ToLower(appendonly) == "yes" { - // 如果集群是rdb备份,但是aof开启,则关闭aof - _, task.Err = cliItem.ConfigSet("appendonly", "yes") + // 如果集群配置是rdb备份,但是aof开启,则关闭aof + _, task.Err = cliItem.ConfigSet("appendonly", "no") if task.Err != nil { continue } diff --git a/dbm-services/redis/db-tools/dbmon/pkg/redisnodesreport/job.go b/dbm-services/redis/db-tools/dbmon/pkg/redisnodesreport/job.go index bf900a3e05..6bf4f81992 100644 --- a/dbm-services/redis/db-tools/dbmon/pkg/redisnodesreport/job.go +++ b/dbm-services/redis/db-tools/dbmon/pkg/redisnodesreport/job.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "fmt" - "os" "path/filepath" "sort" "strconv" @@ -12,12 +11,12 @@ import ( "sync" "time" - "github.com/glebarez/sqlite" "gorm.io/gorm" "gorm.io/gorm/clause" "dbm-services/redis/db-tools/dbmon/config" "dbm-services/redis/db-tools/dbmon/models/myredis" + "dbm-services/redis/db-tools/dbmon/models/mysqlite" "dbm-services/redis/db-tools/dbmon/mylog" "dbm-services/redis/db-tools/dbmon/pkg/consts" "dbm-services/redis/db-tools/dbmon/pkg/report" @@ -47,37 +46,19 @@ func GetGlobRedisNodesReportJob(conf *config.Configuration) *Job { } func (job *Job) getSqDB() { - var homeDir string - homeDir, job.Err = os.Executable() + job.sqdb, job.Err = mysqlite.GetLocalSqDB() if job.Err != nil { - job.Err = fmt.Errorf("os.Executable failed,err:%v", job.Err) - mylog.Logger.Info(job.Err.Error()) - return - } - homeDir = filepath.Dir(homeDir) - dbname := filepath.Join(homeDir, "db", "redis_nodes.db") - job.Err = util.MkDirsIfNotExists([]string{filepath.Join(homeDir, "db")}) - if job.Err != nil { - mylog.Logger.Error(job.Err.Error()) - return - } - util.LocalDirChownMysql(filepath.Join(homeDir, "db")) - job.sqdb, job.Err = gorm.Open(sqlite.Open(dbname), &gorm.Config{}) - if job.Err != nil { - job.Err = fmt.Errorf("gorm.Open failed,err:%v,dbname:%s", job.Err, dbname) - mylog.Logger.Info(job.Err.Error()) return } job.Err = job.sqdb.AutoMigrate(&ClusterNodesSchema{}) if job.Err != nil { - job.Err = fmt.Errorf("AutoMigrate failed,err:%v", job.Err) + job.Err = fmt.Errorf("ClusterNodesSchema AutoMigrate failed,err:%v", job.Err) mylog.Logger.Info(job.Err.Error()) return } } func (job *Job) closeDB() { - dbInstance, _ := job.sqdb.DB() - _ = dbInstance.Close() + mysqlite.CloseDB(job.sqdb) } // getReporter 上报者 diff --git a/dbm-ui/backend/flow/utils/redis/redis_act_playload.py b/dbm-ui/backend/flow/utils/redis/redis_act_playload.py index dc02286506..c19eb1e86d 100644 --- a/dbm-ui/backend/flow/utils/redis/redis_act_playload.py +++ b/dbm-ui/backend/flow/utils/redis/redis_act_playload.py @@ -783,7 +783,7 @@ def redis_cluster_backup_payload(self, **kwargs) -> dict: "backup_type": self.cluster["backup_type"], "domain": self.cluster["domain_name"], "without_to_backup_sys": not BACKUP_SYS_STATUS, - "backup_client_storage_type": "cos", + "backup_client_storage_type": "", # 留空,使用系统默认 }, } @@ -913,7 +913,7 @@ def bkdbmon_install(self, **kwargs) -> dict: "dbtoolspkg": {"pkg": self.tools_pkg.name, "pkg_md5": self.tools_pkg.md5}, "agent_address": env.MYSQL_CROND_AGENT_ADDRESS, "beat_path": env.MYSQL_CROND_BEAT_PATH, - "backup_client_storage_type": "cos", + "backup_client_storage_type": "", # 留空,使用系统默认 "redis_fullbackup": fullbackup_config, "redis_binlogbackup": binlogbackup_config, "redis_heartbeat": heartbeat_config, @@ -1065,7 +1065,7 @@ def redis_cluster_backup_4_scene(self, **kwargs) -> dict: # "inst_num":10, "backup_type": "normal_backup", "without_to_backup_sys": True, # // 是否上传到备份系统,默认false - "backup_client_storage_type": "cos", + "backup_client_storage_type": "", # 留空,使用系统默认 "ssd_log_count": params["ssd_log_count"], }, }