diff --git a/internal/backup/111backup_scheduler.go b/internal/backup/111backup_scheduler.go deleted file mode 100644 index f55d29d..0000000 --- a/internal/backup/111backup_scheduler.go +++ /dev/null @@ -1,1177 +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 ( - "bytes" - "compress/gzip" - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "io" - "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"` -} - -// 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"` -} - -// WALReader интерфейс для чтения WAL -type WALReader interface { - ReadSince(index uint64) ([]WALEntry, error) - GetCurrentIndex() (uint64, error) - GetLastBackupIndex() uint64 - SetLastBackupIndex(index uint64) error - GetSegments() ([]string, error) - ReadSegment(segmentPath string) ([]WALEntry, error) -} - -// 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 - compressLevel int - mu sync.Mutex -} - -// NewFileBackupStorage создаёт новое файловое хранилище бэкапов -func NewFileBackupStorage(backupDir string, compressLevel int) (*FileBackupStorage, error) { - if compressLevel < 1 || compressLevel > 9 { - compressLevel = 6 - } - if err := os.MkdirAll(backupDir, 0755); err != nil { - return nil, fmt.Errorf("failed to create backup dir: %v", err) - } - return &FileBackupStorage{ - backupDir: backupDir, - compressLevel: compressLevel, - }, nil -} - -func (fbs *FileBackupStorage) SaveBackup(path string, data []byte) error { - fbs.mu.Lock() - defer fbs.mu.Unlock() - - fullPath := filepath.Join(fbs.backupDir, path) - if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { - return err - } - - // Сжимаем данные - var buf bytes.Buffer - gzWriter, err := gzip.NewWriterLevel(&buf, fbs.compressLevel) - if err != nil { - return err - } - if _, err := gzWriter.Write(data); err != nil { - gzWriter.Close() - return err - } - if err := gzWriter.Close(); err != nil { - return err - } - - compressedPath := fullPath + ".gz" - return os.WriteFile(compressedPath, buf.Bytes(), 0644) -} - -func (fbs *FileBackupStorage) LoadBackup(path string) ([]byte, error) { - fbs.mu.Lock() - defer fbs.mu.Unlock() - - fullPath := filepath.Join(fbs.backupDir, path) - compressedPath := fullPath + ".gz" - - var data []byte - var err error - - // Пробуем сначала сжатый файл - if _, err := os.Stat(compressedPath); err == nil { - compressedData, err := os.ReadFile(compressedPath) - if err != nil { - return nil, err - } - gzReader, err := gzip.NewReader(bytes.NewReader(compressedData)) - if err != nil { - return nil, err - } - defer gzReader.Close() - data, err = io.ReadAll(gzReader) - if err != nil { - return nil, err - } - } else { - data, err = os.ReadFile(fullPath) - if err != nil { - return nil, err - } - } - - return data, nil -} - -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 { - if e.IsDir() { - // Рекурсивно собираем файлы из поддиректорий - subEntries, err := os.ReadDir(filepath.Join(fbs.backupDir, e.Name())) - if err == nil { - for _, sub := range subEntries { - if !sub.IsDir() { - names = append(names, filepath.Join(e.Name(), sub.Name())) - } - } - } - } else { - names = append(names, e.Name()) - } - } - return names, nil -} - -func (fbs *FileBackupStorage) DeleteBackup(path string) error { - fbs.mu.Lock() - defer fbs.mu.Unlock() - - fullPath := filepath.Join(fbs.backupDir, path) - compressedPath := fullPath + ".gz" - - var err error - if _, statErr := os.Stat(compressedPath); statErr == nil { - err = os.Remove(compressedPath) - } else { - err = os.Remove(fullPath) - } - - // Пробуем удалить пустые директории - dirPath := filepath.Dir(fullPath) - os.Remove(dirPath) - - return err -} - -func (fbs *FileBackupStorage) BackupExists(path string) bool { - fullPath := filepath.Join(fbs.backupDir, path) - compressedPath := fullPath + ".gz" - _, err1 := os.Stat(fullPath) - _, err2 := os.Stat(compressedPath) - return err1 == nil || err2 == nil -} - -// WALReaderImpl реализация WALReader для SegmentedWALManager -type WALReaderImpl struct { - walManager interface{} // *SegmentedWALManager или *WALManager - lastBackupIndex atomic.Uint64 - segmentsDir string - mu sync.RWMutex - logger LoggerInterface -} - -// NewWALReaderImpl создаёт новый WALReader -func NewWALReaderImpl(walManager interface{}, segmentsDir string, logger LoggerInterface) *WALReaderImpl { - return &WALReaderImpl{ - walManager: walManager, - segmentsDir: segmentsDir, - logger: logger, - } -} - -// ReadSince читает записи WAL с указанного индекса -func (wr *WALReaderImpl) ReadSince(index uint64) ([]WALEntry, error) { - wr.mu.RLock() - defer wr.mu.RUnlock() - - entries := make([]WALEntry, 0) - - // Получаем список сегментов - segments, err := wr.GetSegments() - if err != nil { - return nil, err - } - - for _, segmentPath := range segments { - segEntries, err := wr.ReadSegment(segmentPath) - if err != nil { - if wr.logger != nil { - wr.logger.Warn(fmt.Sprintf("Failed to read segment %s: %v", segmentPath, err)) - } - continue - } - - for _, entry := range segEntries { - if entry.Index > index { - entries = append(entries, entry) - } - } - } - - return entries, nil -} - -// GetCurrentIndex возвращает текущий индекс WAL -func (wr *WALReaderImpl) GetCurrentIndex() (uint64, error) { - segments, err := wr.GetSegments() - if err != nil { - return 0, err - } - - if len(segments) == 0 { - return 0, nil - } - - // Читаем последний сегмент для получения последнего индекса - lastSegment := segments[len(segments)-1] - entries, err := wr.ReadSegment(lastSegment) - if err != nil { - return 0, err - } - - if len(entries) == 0 { - return 0, nil - } - - return entries[len(entries)-1].Index, nil -} - -// GetLastBackupIndex возвращает последний сохранённый индекс бэкапа -func (wr *WALReaderImpl) GetLastBackupIndex() uint64 { - return wr.lastBackupIndex.Load() -} - -// SetLastBackupIndex устанавливает последний индекс бэкапа -func (wr *WALReaderImpl) SetLastBackupIndex(index uint64) error { - wr.lastBackupIndex.Store(index) - - // Сохраняем в файл для персистентности - indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json") - data, err := json.Marshal(map[string]uint64{"last_index": index}) - if err != nil { - return err - } - return os.WriteFile(indexPath, data, 0644) -} - -// GetSegments возвращает список сегментов WAL -func (wr *WALReaderImpl) GetSegments() ([]string, error) { - if wr.segmentsDir == "" { - return nil, fmt.Errorf("segments directory not set") - } - - pattern := filepath.Join(wr.segmentsDir, "wal_segment_*.log") - files, err := filepath.Glob(pattern) - if err != nil { - return nil, err - } - - sort.Strings(files) - return files, nil -} - -// ReadSegment читает сегмент WAL -func (wr *WALReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) { - data, err := os.ReadFile(segmentPath) - if err != nil { - return nil, err - } - - entries := make([]WALEntry, 0) - - // Парсим WAL записи (формат: [length][data] повторяется) - pos := 0 - for pos < len(data) { - if pos+4 > len(data) { - break - } - - // Читаем длину записи (4 байта, big-endian) - length := int(data[pos])<<24 | int(data[pos+1])<<16 | int(data[pos+2])<<8 | int(data[pos+3]) - pos += 4 - - if pos+length > len(data) { - break - } - - recordData := data[pos : pos+length] - pos += length - - var record struct { - LSN uint64 `json:"lsn"` - Timestamp int64 `json:"timestamp"` - Type byte `json:"type"` - Data []byte `json:"data"` - } - - if err := json.Unmarshal(recordData, &record); err != nil { - continue - } - - entry := WALEntry{ - Index: record.LSN, - Timestamp: record.Timestamp, - Type: fmt.Sprintf("%d", record.Type), - Data: record.Data, - } - - entries = append(entries, entry) - } - - return entries, nil -} - -// loadLastBackupIndex загружает последний индекс из файла -func (wr *WALReaderImpl) loadLastBackupIndex() { - indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json") - data, err := os.ReadFile(indexPath) - if err != nil { - return - } - - var meta map[string]uint64 - if err := json.Unmarshal(data, &meta); err != nil { - return - } - - if lastIdx, ok := meta["last_index"]; ok { - wr.lastBackupIndex.Store(lastIdx) - } -} - -// 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 - config *BackupSchedulerConfig -} - -// BackupSchedulerConfig конфигурация планировщика -type BackupSchedulerConfig struct { - BackupDir string `json:"backup_dir"` - MaxConcurrent int `json:"max_concurrent"` - CompressEnabled bool `json:"compress_enabled"` - CompressionLevel int `json:"compression_level"` - MaxBackupSizeMB int64 `json:"max_backup_size_mb"` - EnableIncremental bool `json:"enable_incremental"` -} - -// DefaultBackupSchedulerConfig возвращает конфигурацию по умолчанию -func DefaultBackupSchedulerConfig() *BackupSchedulerConfig { - return &BackupSchedulerConfig{ - BackupDir: "backups", - MaxConcurrent: 1, - CompressEnabled: true, - CompressionLevel: 6, - MaxBackupSizeMB: 10240, // 10GB - EnableIncremental: true, - } -} - -// NewBackupScheduler создаёт новый планировщик бэкапов -func NewBackupScheduler(cfg *BackupSchedulerConfig, walReader WALReader, logger LoggerInterface) (*BackupScheduler, error) { - if cfg == nil { - cfg = DefaultBackupSchedulerConfig() - } - - compressLevel := cfg.CompressionLevel - if !cfg.CompressEnabled { - compressLevel = 0 - } - - storage, err := NewFileBackupStorage(cfg.BackupDir, compressLevel) - 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{}), - config: cfg, - } - - // Загружаем существующие расписания - bs.loadSchedules() - - // Загружаем историю бэкапов - bs.loadBackupHistory() - - return bs, nil -} - -// loadSchedules загружает расписания из хранилища -func (bs *BackupScheduler) loadSchedules() { - data, err := bs.storage.LoadBackup("schedules.json") - if err != nil { - if !os.IsNotExist(err) && bs.logger != nil { - 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 { - if bs.logger != 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() { - data, err := bs.storage.LoadBackup("backups.json") - if err != nil { - if !os.IsNotExist(err) && bs.logger != nil { - 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 { - if bs.logger != 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() - - // Проверяем, что расписание с таким именем не существует - for _, s := range bs.schedules { - if s.Name == name { - return nil, fmt.Errorf("schedule with name '%s' already exists", name) - } - } - - 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, fmt.Errorf("failed to save schedules: %v", err) - } - - if bs.logger != nil { - 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) - - if err := bs.saveSchedules(); err != nil { - return fmt.Errorf("failed to save schedules: %v", err) - } - - if bs.logger != nil { - bs.logger.Info(fmt.Sprintf("Removed schedule: %s", scheduleID)) - } - return nil -} - -// 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 !bs.config.EnableIncremental && schedule.Type == IncrementalBackup { - return fmt.Errorf("incremental backups are disabled in configuration") - } - - 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_%d", startTime, 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)) - } - - // Читаем все WAL записи для полного бэкапа - walEntries, err := bs.walReader.ReadSince(0) - if err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to read WAL for full backup: %v", err)) - walEntries = []WALEntry{} - } - - // Создаём структуру бэкапа - backupData := struct { - BackupID string `json:"backup_id"` - Type string `json:"type"` - StartTime int64 `json:"start_time"` - WALIndex uint64 `json:"wal_index"` - Entries []WALEntry `json:"entries,omitempty"` - Metadata interface{} `json:"metadata,omitempty"` - }{ - BackupID: backupID, - Type: "full", - StartTime: startTime, - WALIndex: walIndex, - Entries: walEntries, - Metadata: map[string]interface{}{ - "version": "1.0", - "created_by": "backup_scheduler", - "wal_entries_count": len(walEntries), - }, - } - - data, err := json.Marshal(backupData) - if err != nil { - return fmt.Errorf("failed to marshal backup data: %v", err) - } - - // Проверяем размер бэкапа - if bs.config.MaxBackupSizeMB > 0 && int64(len(data)) > bs.config.MaxBackupSizeMB*1024*1024 { - return fmt.Errorf("backup size %d exceeds maximum allowed %d MB", - len(data)/(1024*1024), bs.config.MaxBackupSizeMB) - } - - // Сохраняем бэкап - backupPath := fmt.Sprintf("full/%s/%s.backup", - time.Now().Format("2006-01-02"), backupID) - - if err := bs.storage.SaveBackup(backupPath, data); err != nil { - return fmt.Errorf("failed to save backup: %v", err) - } - - // Вычисляем контрольную сумму - hash := sha256.Sum256(data) - 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(data)), - Path: backupPath, - WALStart: 0, - WALEnd: walIndex, - Checksum: checksum, - } - - bs.mu.Lock() - bs.backups[backupID] = backupInfo - bs.mu.Unlock() - - if err := bs.saveBackupHistory(); err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err)) - } - - 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, entries=%d", - backupID, duration, backupInfo.SizeBytes, len(walEntries))) - - 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_%d", startTime, 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) - } - - // Проверяем, что родительский бэкап существует - if !bs.storage.BackupExists(lastFullBackup.Path) { - bs.logger.Warn(fmt.Sprintf("Parent backup %s file missing, performing full backup instead", lastFullBackup.ID)) - 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, err := bs.walReader.GetCurrentIndex() - if err != nil { - return fmt.Errorf("failed to get current WAL index: %v", err) - } - - // Создаём структуру инкрементального бэкапа - 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"` - Metadata interface{} `json:"metadata,omitempty"` - }{ - BackupID: backupID, - Type: "incremental", - StartTime: startTime, - WALStart: lastIndex, - WALEnd: currentIndex, - ParentID: lastFullBackup.ID, - Entries: walEntries, - Metadata: map[string]interface{}{ - "version": "1.0", - "created_by": "backup_scheduler", - "entries_count": len(walEntries), - "parent_backup": lastFullBackup.ID, - "parent_time": lastFullBackup.StartTime, - }, - } - - data, err := json.Marshal(backupData) - if err != nil { - return fmt.Errorf("failed to marshal backup data: %v", err) - } - - // Проверяем размер бэкапа - if bs.config.MaxBackupSizeMB > 0 && int64(len(data)) > bs.config.MaxBackupSizeMB*1024*1024 { - // Если инкрементальный бэкап слишком большой, делаем полный - bs.logger.Info("Incremental backup too large, performing full backup instead") - return bs.performFullBackup(scheduleID) - } - - // Сохраняем инкрементальный бэкап - backupPath := fmt.Sprintf("inc/%s/%s.backup", - time.Now().Format("2006-01-02"), backupID) - - 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() - - if err := bs.saveBackupHistory(); err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to save backup history: %v", err)) - } - - 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() - if err := bs.saveSchedules(); err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to save schedules after update: %v", err)) - } - } -} - -// 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) - // Удаляем файл бэкапа - if err := bs.storage.DeleteBackup(backup.Path); err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to delete backup file %s: %v", backup.Path, err)) - } - } - } - - for _, id := range toDelete { - delete(bs.backups, id) - } - - if len(toDelete) > 0 { - if err := bs.saveBackupHistory(); err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to save backup history after cleanup: %v", err)) - } - 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 { - if !bs.running.CompareAndSwap(false, true) { - return fmt.Errorf("restore already in progress") - } - defer bs.running.Store(false) - - 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)) - - // Проверяем существование файла бэкапа - if !bs.storage.BackupExists(backup.Path) { - return fmt.Errorf("backup file not found: %s", backup.Path) - } - - 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...") - - var fullData struct { - BackupID string `json:"backup_id"` - Type string `json:"type"` - Entries []WALEntry `json:"entries"` - Metadata interface{} `json:"metadata"` - } - - if err := json.Unmarshal(data, &fullData); err != nil { - return fmt.Errorf("failed to parse full backup data: %v", err) - } - - bs.logger.Info(fmt.Sprintf("Full backup contains %d WAL entries", len(fullData.Entries))) - // Здесь реальное восстановление данных - - } else { - // Восстановление из инкрементального бэкапа - bs.logger.Info("Restoring from incremental backup...") - - // Сначала восстанавливаем полный бэкап-родитель - if backup.ParentID == "" { - return fmt.Errorf("incremental backup has no parent ID") - } - - // Проверяем существование родительского бэкапа - parentBackup := bs.GetBackup(backup.ParentID) - if parentBackup == nil { - return fmt.Errorf("parent backup %s not found", backup.ParentID) - } - - if err := bs.Restore(backup.ParentID); err != nil { - return fmt.Errorf("failed to restore parent backup: %v", err) - } - - var incData struct { - BackupID string `json:"backup_id"` - Type string `json:"type"` - WALStart uint64 `json:"wal_start"` - WALEnd uint64 `json:"wal_end"` - Entries []WALEntry `json:"entries"` - } - - if err := json.Unmarshal(data, &incData); err != nil { - return fmt.Errorf("failed to parse incremental backup data: %v", err) - } - - bs.logger.Info(fmt.Sprintf("Applying %d incremental changes (WAL %d -> %d)", - len(incData.Entries), incData.WALStart, incData.WALEnd)) - // Здесь применяем инкрементальные изменения - } - - 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() - if bs.logger != nil { - 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() - - if err := bs.saveSchedules(); err != nil { - bs.logger.Warn(fmt.Sprintf("Failed to save schedule after update: %v", err)) - } - - // Запускаем бэкап в отдельной горутине - 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() - if bs.logger != nil { - bs.logger.Info("Backup scheduler stopped") - } -} - -// GetStats возвращает статистику планировщика -func (bs *BackupScheduler) GetStats() map[string]interface{} { - bs.mu.RLock() - defer bs.mu.RUnlock() - - fullCount := 0 - incCount := 0 - totalSize := int64(0) - - for _, b := range bs.backups { - if b.Type == FullBackup { - fullCount++ - } else { - incCount++ - } - totalSize += b.SizeBytes - } - - return map[string]interface{}{ - "total_backups": len(bs.backups), - "full_backups": fullCount, - "incremental_backups": incCount, - "total_size_bytes": totalSize, - "schedules": len(bs.schedules), - "is_running": bs.running.Load(), - "backup_dir": bs.config.BackupDir, - "compress_enabled": bs.config.CompressEnabled, - "incremental_enabled": bs.config.EnableIncremental, - } -}