715 lines
23 KiB
Go
715 lines
23 KiB
Go
/*
|
||
* Copyright 2026 Safronov Grigorii
|
||
*
|
||
* Licensed under the CDDL, Version 1.0 (the "License");
|
||
* you may not use this file except in compliance with the License.
|
||
*
|
||
* You may obtain a copy of the License at
|
||
* https://opensource.org/licenses/CDDL-1.0
|
||
*/
|
||
|
||
// Файл: internal/backup/backup_scheduler.go
|
||
// Назначение: Планировщик бэкапов и инкрементальные бэкапы на основе WAL
|
||
|
||
package backup
|
||
|
||
import (
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
// BackupType тип бэкапа
|
||
type BackupType string
|
||
|
||
const (
|
||
FullBackup BackupType = "full"
|
||
IncrementalBackup BackupType = "incremental"
|
||
)
|
||
|
||
// BackupSchedule расписание бэкапов
|
||
type BackupSchedule struct {
|
||
ID string `json:"id"`
|
||
Name string `json:"name"`
|
||
CronExpr string `json:"cron_expr"`
|
||
Type BackupType `json:"type"`
|
||
RetentionDays int `json:"retention_days"`
|
||
Enabled bool `json:"enabled"`
|
||
LastRun int64 `json:"last_run"`
|
||
NextRun int64 `json:"next_run"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
UpdatedAt int64 `json:"updated_at"`
|
||
}
|
||
|
||
// BackupInfo информация о бэкапе
|
||
type BackupInfo struct {
|
||
ID string `json:"id"`
|
||
ScheduleID string `json:"schedule_id"`
|
||
Type BackupType `json:"type"`
|
||
StartTime int64 `json:"start_time"`
|
||
EndTime int64 `json:"end_time"`
|
||
Status string `json:"status"`
|
||
SizeBytes int64 `json:"size_bytes"`
|
||
Path string `json:"path"`
|
||
WALStart uint64 `json:"wal_start"`
|
||
WALEnd uint64 `json:"wal_end"`
|
||
ParentID string `json:"parent_id,omitempty"`
|
||
Checksum string `json:"checksum"`
|
||
Error string `json:"error,omitempty"`
|
||
}
|
||
|
||
// BackupScheduler планировщик бэкапов
|
||
type BackupScheduler struct {
|
||
schedules map[string]*BackupSchedule
|
||
backups map[string]*BackupInfo
|
||
walReader WALReader
|
||
storage BackupStorage
|
||
logger LoggerInterface
|
||
mu sync.RWMutex
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
running atomic.Bool
|
||
backupCounter atomic.Uint64
|
||
}
|
||
|
||
// WALReader интерфейс для чтения WAL
|
||
type WALReader interface {
|
||
ReadSince(index uint64) ([]WALEntry, error)
|
||
GetCurrentIndex() (uint64, error)
|
||
GetLastBackupIndex() uint64
|
||
SetLastBackupIndex(index uint64) error
|
||
}
|
||
|
||
// WALEntry запись WAL
|
||
type WALEntry struct {
|
||
Index uint64 `json:"index"`
|
||
Term uint64 `json:"term"`
|
||
Type string `json:"type"`
|
||
Data []byte `json:"data"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
// BackupStorage интерфейс для хранения бэкапов
|
||
type BackupStorage interface {
|
||
SaveBackup(path string, data []byte) error
|
||
LoadBackup(path string) ([]byte, error)
|
||
ListBackups() ([]string, error)
|
||
DeleteBackup(path string) error
|
||
BackupExists(path string) bool
|
||
}
|
||
|
||
// LoggerInterface интерфейс для логирования
|
||
type LoggerInterface interface {
|
||
Info(msg string)
|
||
Warn(msg string)
|
||
Error(msg string)
|
||
Debug(msg string)
|
||
}
|
||
|
||
// FileBackupStorage реализация BackupStorage для файловой системы
|
||
type FileBackupStorage struct {
|
||
backupDir string
|
||
}
|
||
|
||
// NewFileBackupStorage создаёт новое файловое хранилище бэкапов
|
||
func NewFileBackupStorage(backupDir string) (*FileBackupStorage, error) {
|
||
if err := os.MkdirAll(backupDir, 0755); err != nil {
|
||
return nil, fmt.Errorf("failed to create backup dir: %v", err)
|
||
}
|
||
return &FileBackupStorage{backupDir: backupDir}, nil
|
||
}
|
||
|
||
func (fbs *FileBackupStorage) SaveBackup(path string, data []byte) error {
|
||
fullPath := filepath.Join(fbs.backupDir, path)
|
||
if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil {
|
||
return err
|
||
}
|
||
return os.WriteFile(fullPath, data, 0644)
|
||
}
|
||
|
||
func (fbs *FileBackupStorage) LoadBackup(path string) ([]byte, error) {
|
||
fullPath := filepath.Join(fbs.backupDir, path)
|
||
return os.ReadFile(fullPath)
|
||
}
|
||
|
||
func (fbs *FileBackupStorage) ListBackups() ([]string, error) {
|
||
entries, err := os.ReadDir(fbs.backupDir)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
names := make([]string, 0, len(entries))
|
||
for _, e := range entries {
|
||
names = append(names, e.Name())
|
||
}
|
||
return names, nil
|
||
}
|
||
|
||
func (fbs *FileBackupStorage) DeleteBackup(path string) error {
|
||
fullPath := filepath.Join(fbs.backupDir, path)
|
||
return os.RemoveAll(fullPath)
|
||
}
|
||
|
||
func (fbs *FileBackupStorage) BackupExists(path string) bool {
|
||
fullPath := filepath.Join(fbs.backupDir, path)
|
||
_, err := os.Stat(fullPath)
|
||
return err == nil
|
||
}
|
||
|
||
// BackupSchedulerConfig конфигурация планировщика
|
||
type BackupSchedulerConfig struct {
|
||
BackupDir string `json:"backup_dir"`
|
||
MaxConcurrent int `json:"max_concurrent"`
|
||
CompressEnabled bool `json:"compress_enabled"`
|
||
CompressionLevel int `json:"compression_level"`
|
||
}
|
||
|
||
// DefaultBackupSchedulerConfig возвращает конфигурацию по умолчанию
|
||
func DefaultBackupSchedulerConfig() *BackupSchedulerConfig {
|
||
return &BackupSchedulerConfig{
|
||
BackupDir: "backups",
|
||
MaxConcurrent: 1,
|
||
CompressEnabled: true,
|
||
CompressionLevel: 6,
|
||
}
|
||
}
|
||
|
||
// NewBackupScheduler создаёт новый планировщик бэкапов
|
||
func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger LoggerInterface) (*BackupScheduler, error) {
|
||
if cfg == nil {
|
||
cfg = DefaultBackupSchedulerConfig()
|
||
}
|
||
|
||
storage, err := NewFileBackupStorage(cfg.BackupDir)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
bs := &BackupScheduler{
|
||
schedules: make(map[string]*BackupSchedule),
|
||
backups: make(map[string]*BackupInfo),
|
||
walReader: walReader,
|
||
storage: storage,
|
||
logger: logger,
|
||
stopChan: make(chan struct{}),
|
||
}
|
||
|
||
// Загружаем существующие расписания
|
||
bs.loadSchedules()
|
||
|
||
// Загружаем историю бэкапов
|
||
bs.loadBackupHistory()
|
||
|
||
return bs, nil
|
||
}
|
||
|
||
// loadSchedules загружает расписания из хранилища
|
||
func (bs *BackupScheduler) loadSchedules() {
|
||
schedulesPath := filepath.Join(bs.storage.(*FileBackupStorage).backupDir, "schedules.json")
|
||
data, err := bs.storage.LoadBackup("schedules.json")
|
||
if err != nil {
|
||
if !os.IsNotExist(err) {
|
||
bs.logger.Warn(fmt.Sprintf("Failed to load schedules: %v", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
var schedules map[string]*BackupSchedule
|
||
if err := json.Unmarshal(data, &schedules); err != nil {
|
||
bs.logger.Warn(fmt.Sprintf("Failed to parse schedules: %v", err))
|
||
return
|
||
}
|
||
|
||
bs.mu.Lock()
|
||
bs.schedules = schedules
|
||
bs.mu.Unlock()
|
||
}
|
||
|
||
// saveSchedules сохраняет расписания
|
||
func (bs *BackupScheduler) saveSchedules() error {
|
||
bs.mu.RLock()
|
||
data, err := json.MarshalIndent(bs.schedules, "", " ")
|
||
bs.mu.RUnlock()
|
||
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
return bs.storage.SaveBackup("schedules.json", data)
|
||
}
|
||
|
||
// loadBackupHistory загружает историю бэкапов
|
||
func (bs *BackupScheduler) loadBackupHistory() {
|
||
historyPath := filepath.Join(bs.storage.(*FileBackupStorage).backupDir, "backups.json")
|
||
data, err := bs.storage.LoadBackup("backups.json")
|
||
if err != nil {
|
||
if !os.IsNotExist(err) {
|
||
bs.logger.Warn(fmt.Sprintf("Failed to load backup history: %v", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
var backups map[string]*BackupInfo
|
||
if err := json.Unmarshal(data, &backups); err != nil {
|
||
bs.logger.Warn(fmt.Sprintf("Failed to parse backup history: %v", err))
|
||
return
|
||
}
|
||
|
||
bs.mu.Lock()
|
||
bs.backups = backups
|
||
bs.mu.Unlock()
|
||
}
|
||
|
||
// saveBackupHistory сохраняет историю бэкапов
|
||
func (bs *BackupScheduler) saveBackupHistory() error {
|
||
bs.mu.RLock()
|
||
data, err := json.MarshalIndent(bs.backups, "", " ")
|
||
bs.mu.RUnlock()
|
||
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
return bs.storage.SaveBackup("backups.json", data)
|
||
}
|
||
|
||
// AddSchedule добавляет расписание бэкапов
|
||
func (bs *BackupScheduler) AddSchedule(name, cronExpr string, backupType BackupType, retentionDays int) (*BackupSchedule, error) {
|
||
bs.mu.Lock()
|
||
defer bs.mu.Unlock()
|
||
|
||
schedule := &BackupSchedule{
|
||
ID: fmt.Sprintf("sch_%d", time.Now().UnixNano()),
|
||
Name: name,
|
||
CronExpr: cronExpr,
|
||
Type: backupType,
|
||
RetentionDays: retentionDays,
|
||
Enabled: true,
|
||
CreatedAt: time.Now().UnixMilli(),
|
||
UpdatedAt: time.Now().UnixMilli(),
|
||
}
|
||
|
||
bs.schedules[schedule.ID] = schedule
|
||
|
||
if err := bs.saveSchedules(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
bs.logger.Info(fmt.Sprintf("Added backup schedule: %s (%s) with cron %s", name, backupType, cronExpr))
|
||
return schedule, nil
|
||
}
|
||
|
||
// RemoveSchedule удаляет расписание
|
||
func (bs *BackupScheduler) RemoveSchedule(scheduleID string) error {
|
||
bs.mu.Lock()
|
||
defer bs.mu.Unlock()
|
||
|
||
if _, exists := bs.schedules[scheduleID]; !exists {
|
||
return fmt.Errorf("schedule not found: %s", scheduleID)
|
||
}
|
||
|
||
delete(bs.schedules, scheduleID)
|
||
return bs.saveSchedules()
|
||
}
|
||
|
||
// RunNow запускает бэкап немедленно
|
||
func (bs *BackupScheduler) RunNow(scheduleID string) error {
|
||
bs.mu.RLock()
|
||
schedule, exists := bs.schedules[scheduleID]
|
||
bs.mu.RUnlock()
|
||
|
||
if !exists {
|
||
return fmt.Errorf("schedule not found: %s", scheduleID)
|
||
}
|
||
|
||
if schedule.Type == FullBackup {
|
||
return bs.performFullBackup(scheduleID)
|
||
}
|
||
return bs.performIncrementalBackup(scheduleID)
|
||
}
|
||
|
||
// performFullBackup выполняет полный бэкап
|
||
func (bs *BackupScheduler) performFullBackup(scheduleID string) error {
|
||
if !bs.running.CompareAndSwap(false, true) {
|
||
return fmt.Errorf("backup already in progress")
|
||
}
|
||
defer bs.running.Store(false)
|
||
|
||
startTime := time.Now().UnixMilli()
|
||
backupID := fmt.Sprintf("full_%d", bs.backupCounter.Add(1))
|
||
|
||
bs.logger.Info(fmt.Sprintf("Starting full backup %s", backupID))
|
||
|
||
// Получаем текущую позицию WAL
|
||
walIndex, err := bs.walReader.GetCurrentIndex()
|
||
if err != nil {
|
||
return fmt.Errorf("failed to get WAL index: %v", err)
|
||
}
|
||
|
||
// Сохраняем последний индекс для инкрементальных бэкапов
|
||
if err := bs.walReader.SetLastBackupIndex(walIndex); err != nil {
|
||
bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
|
||
}
|
||
|
||
// Здесь выполняется реальный бэкап данных
|
||
backupPath := fmt.Sprintf("full/%s/%s.backup",
|
||
time.Now().Format("2006-01-02"), backupID)
|
||
|
||
// Симуляция бэкапа (в реальности здесь будет копирование данных)
|
||
backupData := []byte(fmt.Sprintf(`{"backup_id":"%s","type":"full","timestamp":%d,"wal_index":%d}`,
|
||
backupID, startTime, walIndex))
|
||
|
||
if err := bs.storage.SaveBackup(backupPath, backupData); err != nil {
|
||
return fmt.Errorf("failed to save backup: %v", err)
|
||
}
|
||
|
||
// Вычисляем контрольную сумму
|
||
hash := sha256.Sum256(backupData)
|
||
checksum := hex.EncodeToString(hash[:])
|
||
|
||
endTime := time.Now().UnixMilli()
|
||
|
||
backupInfo := &BackupInfo{
|
||
ID: backupID,
|
||
ScheduleID: scheduleID,
|
||
Type: FullBackup,
|
||
StartTime: startTime,
|
||
EndTime: endTime,
|
||
Status: "completed",
|
||
SizeBytes: int64(len(backupData)),
|
||
Path: backupPath,
|
||
WALStart: 0,
|
||
WALEnd: walIndex,
|
||
Checksum: checksum,
|
||
}
|
||
|
||
bs.mu.Lock()
|
||
bs.backups[backupID] = backupInfo
|
||
bs.mu.Unlock()
|
||
|
||
bs.saveBackupHistory()
|
||
bs.updateScheduleLastRun(scheduleID, startTime)
|
||
bs.cleanupOldBackups(scheduleID)
|
||
|
||
duration := (endTime - startTime) / 1000
|
||
bs.logger.Info(fmt.Sprintf("Full backup %s completed in %d seconds, size=%d bytes",
|
||
backupID, duration, backupInfo.SizeBytes))
|
||
|
||
return nil
|
||
}
|
||
|
||
// performIncrementalBackup выполняет инкрементальный бэкап на основе WAL
|
||
func (bs *BackupScheduler) performIncrementalBackup(scheduleID string) error {
|
||
if !bs.running.CompareAndSwap(false, true) {
|
||
return fmt.Errorf("backup already in progress")
|
||
}
|
||
defer bs.running.Store(false)
|
||
|
||
startTime := time.Now().UnixMilli()
|
||
backupID := fmt.Sprintf("inc_%d", bs.backupCounter.Add(1))
|
||
|
||
// Находим последний полный бэкап
|
||
lastFullBackup := bs.findLastFullBackup()
|
||
if lastFullBackup == nil {
|
||
bs.logger.Info("No full backup found, performing full backup instead")
|
||
return bs.performFullBackup(scheduleID)
|
||
}
|
||
|
||
// Получаем WAL записи с момента последнего бэкапа
|
||
lastIndex := bs.walReader.GetLastBackupIndex()
|
||
if lastIndex == 0 {
|
||
lastIndex = lastFullBackup.WALEnd
|
||
}
|
||
|
||
walEntries, err := bs.walReader.ReadSince(lastIndex)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to read WAL: %v", err)
|
||
}
|
||
|
||
if len(walEntries) == 0 {
|
||
bs.logger.Info("No new WAL entries since last backup")
|
||
return nil
|
||
}
|
||
|
||
currentIndex, _ := bs.walReader.GetCurrentIndex()
|
||
|
||
// Сохраняем инкрементальный бэкап
|
||
backupPath := fmt.Sprintf("inc/%s/%s.backup",
|
||
time.Now().Format("2006-01-02"), backupID)
|
||
|
||
backupData := struct {
|
||
BackupID string `json:"backup_id"`
|
||
Type string `json:"type"`
|
||
StartTime int64 `json:"start_time"`
|
||
WALStart uint64 `json:"wal_start"`
|
||
WALEnd uint64 `json:"wal_end"`
|
||
ParentID string `json:"parent_id"`
|
||
Entries []WALEntry `json:"entries"`
|
||
}{
|
||
BackupID: backupID,
|
||
Type: "incremental",
|
||
StartTime: startTime,
|
||
WALStart: lastIndex,
|
||
WALEnd: currentIndex,
|
||
ParentID: lastFullBackup.ID,
|
||
Entries: walEntries,
|
||
}
|
||
|
||
data, err := json.Marshal(backupData)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to marshal backup data: %v", err)
|
||
}
|
||
|
||
if err := bs.storage.SaveBackup(backupPath, data); err != nil {
|
||
return fmt.Errorf("failed to save incremental backup: %v", err)
|
||
}
|
||
|
||
// Обновляем последний индекс
|
||
if err := bs.walReader.SetLastBackupIndex(currentIndex); err != nil {
|
||
bs.logger.Warn(fmt.Sprintf("Failed to set last backup index: %v", err))
|
||
}
|
||
|
||
hash := sha256.Sum256(data)
|
||
checksum := hex.EncodeToString(hash[:])
|
||
|
||
endTime := time.Now().UnixMilli()
|
||
|
||
backupInfo := &BackupInfo{
|
||
ID: backupID,
|
||
ScheduleID: scheduleID,
|
||
Type: IncrementalBackup,
|
||
StartTime: startTime,
|
||
EndTime: endTime,
|
||
Status: "completed",
|
||
SizeBytes: int64(len(data)),
|
||
Path: backupPath,
|
||
WALStart: lastIndex,
|
||
WALEnd: currentIndex,
|
||
ParentID: lastFullBackup.ID,
|
||
Checksum: checksum,
|
||
}
|
||
|
||
bs.mu.Lock()
|
||
bs.backups[backupID] = backupInfo
|
||
bs.mu.Unlock()
|
||
|
||
bs.saveBackupHistory()
|
||
bs.updateScheduleLastRun(scheduleID, startTime)
|
||
|
||
duration := (endTime - startTime) / 1000
|
||
bs.logger.Info(fmt.Sprintf("Incremental backup %s completed in %d seconds, %d entries, size=%d bytes",
|
||
backupID, duration, len(walEntries), backupInfo.SizeBytes))
|
||
|
||
return nil
|
||
}
|
||
|
||
// findLastFullBackup находит последний успешный полный бэкап
|
||
func (bs *BackupScheduler) findLastFullBackup() *BackupInfo {
|
||
bs.mu.RLock()
|
||
defer bs.mu.RUnlock()
|
||
|
||
var lastFull *BackupInfo
|
||
for _, backup := range bs.backups {
|
||
if backup.Type == FullBackup && backup.Status == "completed" {
|
||
if lastFull == nil || backup.EndTime > lastFull.EndTime {
|
||
lastFull = backup
|
||
}
|
||
}
|
||
}
|
||
return lastFull
|
||
}
|
||
|
||
// updateScheduleLastRun обновляет время последнего запуска расписания
|
||
func (bs *BackupScheduler) updateScheduleLastRun(scheduleID string, timestamp int64) {
|
||
bs.mu.Lock()
|
||
defer bs.mu.Unlock()
|
||
|
||
if schedule, exists := bs.schedules[scheduleID]; exists {
|
||
schedule.LastRun = timestamp
|
||
schedule.UpdatedAt = time.Now().UnixMilli()
|
||
bs.saveSchedules()
|
||
}
|
||
}
|
||
|
||
// cleanupOldBackups удаляет старые бэкапы
|
||
func (bs *BackupScheduler) cleanupOldBackups(scheduleID string) {
|
||
bs.mu.Lock()
|
||
defer bs.mu.Unlock()
|
||
|
||
schedule, exists := bs.schedules[scheduleID]
|
||
if !exists {
|
||
return
|
||
}
|
||
|
||
cutoffTime := time.Now().AddDate(0, 0, -schedule.RetentionDays).UnixMilli()
|
||
|
||
toDelete := make([]string, 0)
|
||
for id, backup := range bs.backups {
|
||
if backup.ScheduleID == scheduleID && backup.StartTime < cutoffTime {
|
||
toDelete = append(toDelete, id)
|
||
// Удаляем файл бэкапа
|
||
bs.storage.DeleteBackup(backup.Path)
|
||
}
|
||
}
|
||
|
||
for _, id := range toDelete {
|
||
delete(bs.backups, id)
|
||
}
|
||
|
||
if len(toDelete) > 0 {
|
||
bs.saveBackupHistory()
|
||
bs.logger.Info(fmt.Sprintf("Cleaned up %d old backups for schedule %s", len(toDelete), scheduleID))
|
||
}
|
||
}
|
||
|
||
// ListBackups возвращает список бэкапов
|
||
func (bs *BackupScheduler) ListBackups() []*BackupInfo {
|
||
bs.mu.RLock()
|
||
defer bs.mu.RUnlock()
|
||
|
||
backups := make([]*BackupInfo, 0, len(bs.backups))
|
||
for _, b := range bs.backups {
|
||
backups = append(backups, b)
|
||
}
|
||
|
||
// Сортируем по времени
|
||
sort.Slice(backups, func(i, j int) bool {
|
||
return backups[i].StartTime > backups[j].StartTime
|
||
})
|
||
|
||
return backups
|
||
}
|
||
|
||
// ListSchedules возвращает список расписаний
|
||
func (bs *BackupScheduler) ListSchedules() []*BackupSchedule {
|
||
bs.mu.RLock()
|
||
defer bs.mu.RUnlock()
|
||
|
||
schedules := make([]*BackupSchedule, 0, len(bs.schedules))
|
||
for _, s := range bs.schedules {
|
||
schedules = append(schedules, s)
|
||
}
|
||
return schedules
|
||
}
|
||
|
||
// GetBackup возвращает информацию о бэкапе
|
||
func (bs *BackupScheduler) GetBackup(backupID string) *BackupInfo {
|
||
bs.mu.RLock()
|
||
defer bs.mu.RUnlock()
|
||
return bs.backups[backupID]
|
||
}
|
||
|
||
// Restore восстанавливает данные из бэкапа
|
||
func (bs *BackupScheduler) Restore(backupID string) error {
|
||
backup := bs.GetBackup(backupID)
|
||
if backup == nil {
|
||
return fmt.Errorf("backup not found: %s", backupID)
|
||
}
|
||
|
||
bs.logger.Info(fmt.Sprintf("Starting restore from backup %s", backupID))
|
||
|
||
data, err := bs.storage.LoadBackup(backup.Path)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to load backup: %v", err)
|
||
}
|
||
|
||
// Проверяем контрольную сумму
|
||
hash := sha256.Sum256(data)
|
||
checksum := hex.EncodeToString(hash[:])
|
||
if checksum != backup.Checksum {
|
||
return fmt.Errorf("backup checksum mismatch, data may be corrupted")
|
||
}
|
||
|
||
if backup.Type == FullBackup {
|
||
// Восстановление из полного бэкапа
|
||
bs.logger.Info("Restoring from full backup...")
|
||
// Здесь реальное восстановление
|
||
} else {
|
||
// Восстановление из инкрементального бэкапа
|
||
// Сначала восстанавливаем полный бэкап-родитель
|
||
if backup.ParentID != "" {
|
||
if err := bs.Restore(backup.ParentID); err != nil {
|
||
return fmt.Errorf("failed to restore parent backup: %v", err)
|
||
}
|
||
}
|
||
bs.logger.Info("Applying incremental changes...")
|
||
// Здесь применяем инкрементальные изменения
|
||
}
|
||
|
||
bs.logger.Info(fmt.Sprintf("Restore from backup %s completed", backupID))
|
||
return nil
|
||
}
|
||
|
||
// Start запускает планировщик
|
||
func (bs *BackupScheduler) Start() {
|
||
bs.wg.Add(1)
|
||
go bs.schedulerLoop()
|
||
bs.logger.Info("Backup scheduler started")
|
||
}
|
||
|
||
// schedulerLoop основной цикл планировщика
|
||
func (bs *BackupScheduler) schedulerLoop() {
|
||
defer bs.wg.Done()
|
||
|
||
ticker := time.NewTicker(60 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-bs.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
bs.checkSchedules()
|
||
}
|
||
}
|
||
}
|
||
|
||
// checkSchedules проверяет расписания и запускает бэкапы
|
||
func (bs *BackupScheduler) checkSchedules() {
|
||
bs.mu.RLock()
|
||
schedules := make([]*BackupSchedule, 0, len(bs.schedules))
|
||
for _, s := range bs.schedules {
|
||
if s.Enabled {
|
||
schedules = append(schedules, s)
|
||
}
|
||
}
|
||
bs.mu.RUnlock()
|
||
|
||
now := time.Now()
|
||
for _, schedule := range schedules {
|
||
// Простая проверка расписания (в реальности используется cron-парсер)
|
||
if schedule.NextRun == 0 || schedule.NextRun <= now.UnixMilli() {
|
||
// Вычисляем следующее время (упрощённо - через 24 часа)
|
||
schedule.NextRun = now.Add(24 * time.Hour).UnixMilli()
|
||
|
||
bs.mu.Lock()
|
||
bs.schedules[schedule.ID] = schedule
|
||
bs.mu.Unlock()
|
||
bs.saveSchedules()
|
||
|
||
// Запускаем бэкап в отдельной горутине
|
||
scheduleID := schedule.ID
|
||
backupType := schedule.Type
|
||
go func() {
|
||
if backupType == FullBackup {
|
||
bs.performFullBackup(scheduleID)
|
||
} else {
|
||
bs.performIncrementalBackup(scheduleID)
|
||
}
|
||
}()
|
||
}
|
||
}
|
||
}
|
||
|
||
// Stop останавливает планировщик
|
||
func (bs *BackupScheduler) Stop() {
|
||
close(bs.stopChan)
|
||
bs.wg.Wait()
|
||
bs.logger.Info("Backup scheduler stopped")
|
||
} |