diff --git a/internal/backup/11backup_scheduler.go b/internal/backup/11backup_scheduler.go deleted file mode 100644 index 9870395..0000000 --- a/internal/backup/11backup_scheduler.go +++ /dev/null @@ -1,715 +0,0 @@ -/* - * 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") -} \ No newline at end of file