Merge pull request #1097 from signal18/1095-stop-maintenance-when-repman-disk-is-full

1095 stop maintenance when repman disk is full

Author: caffeinated92

cluster/cluster.go

@@ -242,6 +242,7 @@ type Cluster struct {
 	ConfigManager             *manager.ConfigManager      `json:"-"`
 	failSendCount             int                         `json:"-"`
 	MeetUserID                string                      `json:"-"` //To store meet user id
+	DiskStatManager           *misc.DiskStatManager       `json:"diskStat"`
 	LastDelayStatPrint        time.Time
 	sync.Mutex
 	crcTable               *crc64.Table
@@ -695,8 +696,9 @@ func (cluster *Cluster) Run() {
 							go cluster.CheckCredentialRotation()
 							cluster.CheckCanSaveDynamicConfig()
 							cluster.CheckIsOverwrite()
+							cluster.CheckAllBackupFreeSpace()
 						} else {
-							cluster.StateMachine.PreserveState("WARN0093", "WARN0084", "WARN0095", "WARN0101", "WARN0111", "WARN0112", "ERR00090", "WARN0102", "WARN0134")
+							cluster.StateMachine.PreserveState("WARN0093", "WARN0084", "WARN0095", "WARN0101", "WARN0111", "WARN0112", "ERR00090", "WARN0102", "WARN0134", "WARN0139", "WARN0140", "WARN0141", "WARN0142", "WARN0143")
 						}
 						if !cluster.CanInitNodes {
 							cluster.SetState("ERR00082", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(clusterError["ERR00082"], cluster.errorInitNodes), ErrFrom: "OPENSVC"})

cluster/cluster_bck.go

@@ -9,9 +9,13 @@ package cluster
 import (
 	"fmt"
 	"os"
+	"sync"
 
+	"github.com/dustin/go-humanize"
+	"github.com/shirou/gopsutil/disk"
 	"github.com/signal18/replication-manager/config"
 	"github.com/signal18/replication-manager/utils/archiver"
+	"github.com/signal18/replication-manager/utils/dbhelper"
 	"github.com/signal18/replication-manager/utils/state"
 	"github.com/sirupsen/logrus"
 )
@@ -196,3 +200,107 @@ func (cluster *Cluster) ResticResetQueue() error {
 
 	return nil
 }
+
+func (cluster *Cluster) CheckBackupFreeSpace(backtype string, backup bool) error {
+	var isWarning bool
+	bcksrv := cluster.GetBackupServer()
+	if bcksrv == nil {
+		bcksrv = cluster.master
+	}
+
+	parentDir := cluster.Conf.WorkingDir + "/" + config.ConstStreamingSubDir + "/" + cluster.Name
+	diskstat, err := disk.Usage(parentDir)
+	if err != nil {
+		cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModTask, config.LvlErr, "Error getting disk usage: %s", err)
+		return err
+	}
+
+	cluster.DiskStatManager.UpdateStat(parentDir, diskstat)
+	if diskstat.UsedPercent > float64(cluster.Conf.BackupDiskTresholdCrit) {
+		cluster.SetState("WARN0140", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(cluster.GetErrorList()["WARN0140"], diskstat.Path, diskstat.UsedPercent, cluster.Conf.BackupDiskTresholdCrit), ErrFrom: "JOB", ServerUrl: bcksrv.URL})
+		return fmt.Errorf("Disk usage is over %d%% on %s. Used: %s", cluster.Conf.BackupDiskTresholdCrit, diskstat.Path, humanize.Bytes(diskstat.Used))
+	} else if diskstat.UsedPercent > float64(cluster.Conf.BackupDiskTresholdWarn) {
+		isWarning = true
+		cluster.SetState("WARN0139", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(cluster.GetErrorList()["WARN0139"], diskstat.Path, diskstat.UsedPercent, cluster.Conf.BackupDiskTresholdWarn), ErrFrom: "JOB", ServerUrl: bcksrv.URL})
+	}
+
+	// Estimate size if disk usage is over treshold and estimate size is enabled. For binlog we will always estimate size to 2GB
+	if (isWarning && cluster.Conf.BackupEstimateSize) || backtype == "binlog" {
+		free := diskstat.Free
+		required := uint64(0)
+
+		switch backtype {
+		case "logical", "physical":
+			_, prev := bcksrv.GetLatestMeta(backtype)
+			if prev != nil && prev.Completed {
+				required = uint64(prev.Size * int64(100+cluster.Conf.BackupGrowthPercentage) / 100)
+
+				// If not keep until valid, we need to add the size of the previous backup to the free space
+				if !cluster.Conf.BackupKeepUntilValid {
+					free = free + uint64(prev.Size)
+				}
+
+			} else {
+				cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModTask, config.LvlInfo, "No previous backup found for %s. Estimating backup size.", bcksrv.URL)
+				estimatedSize, err := dbhelper.GetBackupSizeEstimation(bcksrv.Conn, bcksrv.DBVersion)
+				if err != nil {
+					return fmt.Errorf("Error estimating backup size: %s", err)
+				}
+
+				required = estimatedSize * uint64(cluster.Conf.BackupEstimateSizePercentage) / 100
+			}
+		case "binlog":
+			// Max binlog size per file is 1GB, additional 1GB for unexpected growth
+			required = 2 * 1024 * 1024 * 1024
+		case "restic":
+			// Restic backup size is not known until the backup is done
+		}
+
+		if free < required {
+			if backtype == "logical" {
+				cluster.SetState("WARN0141", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(cluster.GetErrorList()["WARN0139"], cluster.Conf.BackupLogicalType, bcksrv.URL, diskstat.Path, humanize.Bytes(diskstat.Free), humanize.Bytes(required)), ErrFrom: "JOB", ServerUrl: bcksrv.URL})
+			} else if backtype == "physical" {
+				cluster.SetState("WARN0142", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(cluster.GetErrorList()["WARN0140"], cluster.Conf.BackupPhysicalType, bcksrv.URL, diskstat.Path, humanize.Bytes(diskstat.Free), humanize.Bytes(required)), ErrFrom: "JOB", ServerUrl: bcksrv.URL})
+			} else if backtype == "binlog" {
+				cluster.SetState("WARN0143", state.State{ErrType: "WARNING", ErrDesc: fmt.Sprintf(cluster.GetErrorList()["WARN0141"], bcksrv.URL, diskstat.Path, humanize.Bytes(diskstat.Free), humanize.Bytes(required)), ErrFrom: "JOB", ServerUrl: bcksrv.URL})
+			}
+
+			return fmt.Errorf("Not enough free space on %s for backup. Free: %s", diskstat.Path, humanize.Bytes(diskstat.Free))
+		}
+
+		if backup {
+			cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModTask, config.LvlInfo, "Free space is enough on %s: %s. Required: %s", diskstat.Path, humanize.Bytes(diskstat.Free), humanize.Bytes(required))
+		}
+	}
+
+	return nil
+}
+
+func (cluster *Cluster) CheckAllBackupFreeSpace() {
+	if !cluster.Conf.BackupCheckFreeSpace {
+		return
+	}
+
+	// Check based on treshold
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		cluster.CheckBackupFreeSpace("logical", false)
+		wg.Done()
+	}()
+
+	// if estimate size is enabled, check the free space for physical and binlog backups too
+	if cluster.Conf.BackupEstimateSize {
+		wg.Add(2)
+		go func() {
+			cluster.CheckBackupFreeSpace("physical", false)
+			wg.Done()
+		}()
+		go func() {
+			cluster.CheckBackupFreeSpace("binlog", false)
+			wg.Done()
+		}()
+	}
+
+	wg.Wait()
+}

cluster/cluster_set.go

@@ -2461,3 +2461,7 @@ func (cluster *Cluster) RenameCluster(newClusterName string) error {
 	return nil
 
 }
+
+func (cluster *Cluster) SetLogStatsLevel(value int) {
+	cluster.Conf.LogStatsLevel = value
+}

cluster/srv_job.go

@@ -245,14 +245,14 @@ func (server *ServerMonitor) JobInsertTask(task string, port string, repmanhost
 	return res.LastInsertId()
 }
 
-func (server *ServerMonitor) JobBackupPhysical() (int64, error) {
+func (server *ServerMonitor) JobBackupPhysical() error {
 	//server can be nil as no dicovered master
 	if server == nil {
-		return 0, nil
+		return nil
 	}
 
 	if server.IsDown() {
-		return 0, nil
+		return nil
 	}
 
 	cluster := server.ClusterGroup
@@ -297,7 +297,7 @@ func (server *ServerMonitor) JobBackupPhysical() (int64, error) {
 
 	if err != nil {
 		cluster.SetInPhysicalBackupState(false)
-		return 0, nil
+		return nil
 	}
 
 	now := time.Now()
@@ -308,6 +308,14 @@ func (server *ServerMonitor) JobBackupPhysical() (int64, error) {
 		prevId = prev.Id
 	}
 
+	// Check for previous backup size
+	if cluster.Conf.BackupCheckFreeSpace {
+		err = cluster.CheckBackupFreeSpace("physical", true)
+		if err != nil {
+			return err
+		}
+	}
+
 	// Remove from backup list, since the file will be replaced
 	if !cluster.Conf.BackupKeepUntilValid {
 		cluster.BackupMetaMap.Delete(prevId)
@@ -327,12 +335,12 @@ func (server *ServerMonitor) JobBackupPhysical() (int64, error) {
 
 	cluster.BackupMetaMap.Set(server.LastBackupMeta.Physical.Id, server.LastBackupMeta.Physical)
 
-	jobid, err := server.JobInsertTask(cluster.Conf.BackupPhysicalType, port, cluster.Conf.MonitorAddress)
+	_, err = server.JobInsertTask(cluster.Conf.BackupPhysicalType, port, cluster.Conf.MonitorAddress)
 	if err != nil {
 		cluster.SetInPhysicalBackupState(false)
 	}
 
-	return jobid, err
+	return err
 }
 
 func (server *ServerMonitor) JobReseedPhysicalBackup(backtype string) error {
@@ -2040,6 +2048,14 @@ func (server *ServerMonitor) JobBackupLogical() error {
 		prevId = prev.Id
 	}
 
+	// Check for previous backup size
+	if cluster.Conf.BackupCheckFreeSpace {
+		err = cluster.CheckBackupFreeSpace("logical", true)
+		if err != nil {
+			return err
+		}
+	}
+
 	// Remove from backup list, since the file will be replaced
 	if !cluster.Conf.BackupKeepUntilValid {
 		cluster.BackupMetaMap.Delete(prevId)
@@ -2437,6 +2453,13 @@ func (server *ServerMonitor) JobBackupBinlog(binlogfile string, isPurge bool) er
 		defer cluster.SetInBinlogBackupState(false)
 	}
 
+	if cluster.Conf.BackupCheckFreeSpace {
+		err = cluster.CheckBackupFreeSpace("binlog", true)
+		if err != nil {
+			return err
+		}
+	}
+
 	server.SetBackingUpBinaryLog(true)
 	defer server.SetBackingUpBinaryLog(false)
 

config/config.go

@@ -148,6 +148,7 @@ type Config struct {
 	LogSupportLevel                           int                    `scope:"server" mapstructure:"log-support-level" toml:"log-support-level" json:"logSupportLevel"`
 	LogExternalScript                         bool                   `mapstructure:"log-external-script" toml:"log-external-script" json:"ExternalScript"`
 	LogExternalScriptLevel                    int                    `mapstructure:"log-external-script-level" toml:"log-external-script-level" json:"logExternalScriptLevel"`
+	LogStatsLevel                             int                    `scope:"server" mapstructure:"log-stats-level" toml:"log-stats-level" json:"logStatsLevel"`
 	User                                      string                 `mapstructure:"db-servers-credential" toml:"db-servers-credential" json:"dbServersCredential"`
 	Hosts                                     string                 `mapstructure:"db-servers-hosts" toml:"db-servers-hosts" json:"dbServersHosts"`
 	DbServersChangeStateScript                string                 `mapstructure:"db-servers-state-change-script" toml:"db-servers-state-change-script" json:"dbServersStateChangeScript"`
@@ -642,6 +643,12 @@ type Config struct {
 	BackupSaveScript                          string                 `mapstructure:"backup-save-script" toml:"backup-save-script" json:"backupSaveScript"`
 	BackupLoadScript                          string                 `mapstructure:"backup-load-script" toml:"backup-load-script" json:"backupLoadScript"`
 	CompressBackups                           bool                   `mapstructure:"compress-backups" toml:"compress-backups" json:"compressBackups"`
+	BackupCheckFreeSpace                      bool                   `mapstructure:"backup-check-free-space" toml:"backup-check-free-space" json:"backupCheckFreeSpace"`
+	BackupDiskTresholdWarn                    int                    `mapstructure:"backup-disk-treshold-warn" toml:"backup-disk-treshold-warn" json:"backupDiskTresholdWarn"`
+	BackupDiskTresholdCrit                    int                    `mapstructure:"backup-disk-treshold-crit" toml:"backup-disk-treshold-crit" json:"backupDiskTresholdCrit"`
+	BackupEstimateSize                        bool                   `mapstructure:"backup-estimate-size" toml:"backup-estimate-size" json:"backupEstimateSize"`
+	BackupEstimateSizePercentage              int                    `mapstructure:"backup-estimate-size-percentage" toml:"backup-estimate-size-percentage" json:"backupEstimateSizePercentage"`
+	BackupGrowthPercentage                    int                    `mapstructure:"backup-growth-percentage" toml:"backup-growth-percentage" json:"backupGrowthPercentage"`
 	SchedulerDatabaseLogsTableRotate          bool                   `mapstructure:"scheduler-db-servers-logs-table-rotate" toml:"scheduler-db-servers-logs-table-rotate" json:"schedulerDbServersLogsTableRotate"`
 	SchedulerDatabaseLogsTableRotateCron      string                 `mapstructure:"scheduler-db-servers-logs-table-rotate-cron" toml:"scheduler-db-servers-logs-table-rotate-cron" json:"schedulerDbServersLogsTableRotateCron"`
 	SchedulerMaintenanceDatabaseLogsTableKeep int                    `mapstructure:"scheduler-db-servers-logs-table-keep" toml:"scheduler-db-servers-logs-table-keep" json:"schedulerDatabaseLogsTableKeep"`
@@ -1162,6 +1169,7 @@ const (
 	ConstLogModMailer         = 19
 	ConstLogModSupport        = 20
 	ConstLogModExternalScript = 21
+	ConstLogModStats          = 22
 )
 
 /*
@@ -2989,6 +2997,8 @@ func (conf *Config) IsEligibleForPrinting(module int, level string) bool {
 			return conf.LogMailerLevel >= lvl
 		case module == ConstLogModSupport:
 			return conf.LogSupportLevel >= lvl
+		case module == ConstLogModStats:
+			return conf.LogStatsLevel >= lvl
 		}
 	}
 
@@ -3154,6 +3164,8 @@ func GetTagsForLog(module int) string {
 		return "job"
 	case ConstLogModExternalScript:
 		return "externalscript"
+	case ConstLogModStats:
+		return "stats"
 	}
 	return ""
 }

config/error.go

@@ -201,6 +201,11 @@ var ClusterError = map[string]string{
 	"WARN0136":  "Enforce GTID replication is not yet available in MySQL. Skipping server %s",
 	"WARN0137":  "Unable to open certificate file %s. Err: %s",
 	"WARN0138":  "Unable to send mail alert. Please check your mailer configuration!",
+	"WARN0139":  "Backup partition used size exceed warning treshold. Mount: %s Used: %s Treshold: %s",
+	"WARN0140":  "Backup partition used size exceed critical treshold. Mount: %s Used: %s Treshold: %s",
+	"WARN0141":  "Not enough free space estimated for %s logical backup on %s. Mount: %s Free: %s Required: %s",
+	"WARN0142":  "Not enough free space estimated for %s physical backup on %s. Mount: %s Free: %s Required: %s",
+	"WARN0143":  "Not enough free space estimated for binary log backup on %s. Mount: %s Free: %s Required: %s",
 	"MDEV20821": "MariaDB version has replication issue https://jira.mariadb.org/browse/MDEV-20821",
 	"MDEV28310": "MariaDB version has replication issue for non row format https://jira.mariadb.org/browse/MDEV-28310",
 	"MDEV19577": "MariaDB version has replication issue for non row format https://jira.mariadb.org/browse/MDEV-19577",

server/api_cluster.go

@@ -2153,6 +2153,10 @@ func (repman *ReplicationManager) switchClusterSettings(mycluster *cluster.Clust
 		mycluster.SwitchBackupBinlogs()
 	case "compress-backups":
 		mycluster.SwitchCompressBackups()
+	case "backup-check-free-space":
+		mycluster.Conf.BackupCheckFreeSpace = !mycluster.Conf.BackupCheckFreeSpace
+	case "backup-estimate-size":
+		mycluster.Conf.BackupEstimateSize = !mycluster.Conf.BackupEstimateSize
 	case "monitoring-pause":
 		mycluster.SwitchMonitoringPause()
 	case "monitoring-save-config":
@@ -2523,6 +2527,18 @@ func (repman *ReplicationManager) setClusterSetting(mycluster *cluster.Cluster,
 		if err != nil {
 			return err
 		}
+	case "backup-disk-treshold-warn":
+		val, _ := strconv.Atoi(value)
+		mycluster.Conf.BackupDiskTresholdWarn = val
+	case "backup-disk-treshold-crit":
+		val, _ := strconv.Atoi(value)
+		mycluster.Conf.BackupDiskTresholdCrit = val
+	case "backup-estimate-size-percentage":
+		val, _ := strconv.Atoi(value)
+		mycluster.Conf.BackupEstimateSizePercentage = val
+	case "backup-growth-percentage":
+		val, _ := strconv.Atoi(value)
+		mycluster.Conf.BackupGrowthPercentage = val
 	case "backup-logical-type":
 		mycluster.SetBackupLogicalType(value)
 	case "backup-physical-type":
@@ -2718,6 +2734,9 @@ func (repman *ReplicationManager) setClusterSetting(mycluster *cluster.Cluster,
 	case "log-external-script-level":
 		val, _ := strconv.Atoi(value)
 		mycluster.SetLogExternalScriptLevel(val)
+	case "log-stats-level":
+		val, _ := strconv.Atoi(value)
+		mycluster.Conf.LogStatsLevel = val
 	case "monitoring-ignore-errors":
 		mycluster.SetMonitorIgnoreErrors(value)
 	case "monitoring-capture-trigger":
@@ -3135,6 +3154,10 @@ func (repman *ReplicationManager) setClusterSetting(mycluster *cluster.Cluster,
 		}
 	case "compress-backups":
 		mycluster.Conf.CompressBackups = isactive
+	case "backup-check-free-space":
+		mycluster.Conf.BackupCheckFreeSpace = isactive
+	case "backup-estimate-size":
+		mycluster.Conf.BackupEstimateSize = isactive
 	case "monitoring-pause":
 		mycluster.Conf.MonitorPause = isactive
 	case "monitoring-save-config":
@@ -3532,6 +3555,9 @@ func (repman *ReplicationManager) setRepmanSetting(name string, value string) er
 	case "log-support-level":
 		val, _ := strconv.Atoi(value)
 		repman.Conf.SetLogSupportLevel(val)
+	case "log-stats-level":
+		val, _ := strconv.Atoi(value)
+		repman.Conf.LogStatsLevel = val
 	case "mail-smtp-addr":
 		repman.Conf.SetMailSmtpAddr(value)
 		repman.Mailer.UpdateAddress(value)

server/repmanv3.go

@@ -408,7 +408,7 @@ func (s *ReplicationManager) MasterPhysicalBackup(ctx context.Context, in *v3.Cl
 	if m == nil {
 		return nil, v3.NewErrorResource(codes.InvalidArgument, v3.ErrClusterMasterNotSet, "cluster", in.Name).Err()
 	}
-	_, err = m.JobBackupPhysical()
+	err = m.JobBackupPhysical()
 	return &emptypb.Empty{}, err
 }
 
@@ -608,7 +608,7 @@ func (s *ReplicationManager) PerformClusterAction(ctx context.Context, in *v3.Cl
 		if m == nil {
 			return nil, v3.NewErrorResource(codes.InvalidArgument, v3.ErrClusterMasterNotSet, "cluster", in.Cluster.Name).Err()
 		}
-		_, err = m.JobBackupPhysical()
+		err = m.JobBackupPhysical()
 	case v3.ClusterAction_OPTIMIZE:
 		mycluster.RollingOptimize()
 	case v3.ClusterAction_RESET_FAILOVER_CONTROL: