1610 lines
51 KiB
Go
1610 lines
51 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/cluster/crossdc_migrator.go
|
||
// Назначение: Кросс-датацентровая миграция данных
|
||
// Алгоритм: Асинхронная репликация с CDC (Change Data Capture) и очередью изменений
|
||
|
||
package cluster
|
||
|
||
import (
|
||
"bytes"
|
||
"compress/gzip"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"sort"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"futriis/internal/config"
|
||
"futriis/internal/log"
|
||
"futriis/internal/storage"
|
||
)
|
||
|
||
// =============================================================================
|
||
// ТИПЫ ДАННЫХ ДЛЯ МИГРАЦИИ
|
||
// =============================================================================
|
||
|
||
// MigrationStatus представляет статус миграции
|
||
type MigrationStatus string
|
||
|
||
const (
|
||
MigrationStatusIdle MigrationStatus = "idle"
|
||
MigrationStatusPreparing MigrationStatus = "preparing"
|
||
MigrationStatusMigrating MigrationStatus = "migrating"
|
||
MigrationStatusDeltaSync MigrationStatus = "delta_sync"
|
||
MigrationStatusValidating MigrationStatus = "validating"
|
||
MigrationStatusCompleted MigrationStatus = "completed"
|
||
MigrationStatusFailed MigrationStatus = "failed"
|
||
MigrationStatusPaused MigrationStatus = "paused"
|
||
)
|
||
|
||
// MigrationMode представляет режим миграции
|
||
type MigrationMode string
|
||
|
||
const (
|
||
MigrationModeManual MigrationMode = "manual" // Полностью ручной
|
||
MigrationModeSemiAuto MigrationMode = "semi_auto" // Полуавтоматический
|
||
MigrationModeAuto MigrationMode = "auto" // Полностью автоматический
|
||
)
|
||
|
||
// ChangeType представляет тип изменения
|
||
type ChangeType string
|
||
|
||
const (
|
||
ChangeInsert ChangeType = "insert"
|
||
ChangeUpdate ChangeType = "update"
|
||
ChangeDelete ChangeType = "delete"
|
||
)
|
||
|
||
// ChangeRecord представляет запись об изменении
|
||
type ChangeRecord struct {
|
||
ID string `json:"id"`
|
||
Database string `json:"database"`
|
||
Collection string `json:"collection"`
|
||
DocumentID string `json:"document_id"`
|
||
ChangeType ChangeType `json:"change_type"`
|
||
Document map[string]interface{} `json:"document,omitempty"`
|
||
PreviousDoc map[string]interface{} `json:"previous_doc,omitempty"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
Version uint64 `json:"version"`
|
||
Checksum string `json:"checksum"`
|
||
Applied bool `json:"applied"`
|
||
AppliedAt int64 `json:"applied_at,omitempty"`
|
||
LSN uint64 `json:"lsn,omitempty"`
|
||
}
|
||
|
||
// MigrationTask представляет задачу миграции
|
||
type MigrationTask struct {
|
||
ID string `json:"id"`
|
||
SourceDC string `json:"source_dc"`
|
||
TargetDC string `json:"target_dc"`
|
||
Databases []string `json:"databases"`
|
||
Collections map[string][]string `json:"collections"` // database -> collections
|
||
Status MigrationStatus `json:"status"`
|
||
ProgressPercent float64 `json:"progress_percent"`
|
||
TotalDocuments int64 `json:"total_documents"`
|
||
MigratedDocs int64 `json:"migrated_docs"`
|
||
FailedDocs int64 `json:"failed_docs"`
|
||
SkippedDocs int64 `json:"skipped_docs"`
|
||
StartTime int64 `json:"start_time"`
|
||
EndTime int64 `json:"end_time"`
|
||
LastCheckpoint int64 `json:"last_checkpoint"`
|
||
CheckpointData map[string]interface{} `json:"checkpoint_data"`
|
||
Error string `json:"error,omitempty"`
|
||
CreatedAt int64 `json:"created_at"`
|
||
UpdatedAt int64 `json:"updated_at"`
|
||
mu sync.RWMutex `json:"-"`
|
||
}
|
||
|
||
// MigrationStats представляет статистику миграции
|
||
type MigrationStats struct {
|
||
TotalChanges int64 `json:"total_changes"`
|
||
AppliedChanges int64 `json:"applied_changes"`
|
||
FailedChanges int64 `json:"failed_changes"`
|
||
SkippedChanges int64 `json:"skipped_changes"`
|
||
LatencyAvg int64 `json:"latency_avg_ms"`
|
||
LatencyMax int64 `json:"latency_max_ms"`
|
||
Throughput int64 `json:"throughput_docs_per_sec"`
|
||
mu sync.RWMutex
|
||
}
|
||
|
||
// MigrationCheckpoint представляет чекпоинт миграции
|
||
type MigrationCheckpoint struct {
|
||
TaskID string `json:"task_id"`
|
||
LastLSN uint64 `json:"last_lsn"`
|
||
ProcessedDocs map[string]bool `json:"processed_docs"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
Version uint64 `json:"version"`
|
||
}
|
||
|
||
// =============================================================================
|
||
// CHANGE QUEUE - ОЧЕРЕДЬ ИЗМЕНЕНИЙ С ПЕРСИСТЕНТНОСТЬЮ
|
||
// =============================================================================
|
||
|
||
// ChangeQueue представляет очередь изменений для миграции
|
||
type ChangeQueue struct {
|
||
changes []*ChangeRecord
|
||
mu sync.RWMutex
|
||
maxSize int
|
||
persisted bool
|
||
filePath string
|
||
lastLSN atomic.Uint64
|
||
}
|
||
|
||
// NewChangeQueue создаёт новую очередь изменений
|
||
func NewChangeQueue(maxSize int, filePath string) *ChangeQueue {
|
||
q := &ChangeQueue{
|
||
changes: make([]*ChangeRecord, 0, maxSize),
|
||
maxSize: maxSize,
|
||
filePath: filePath,
|
||
persisted: false,
|
||
}
|
||
|
||
// Загружаем сохранённую очередь
|
||
if filePath != "" {
|
||
q.load()
|
||
}
|
||
|
||
return q
|
||
}
|
||
|
||
// load загружает очередь из файла
|
||
func (cq *ChangeQueue) load() {
|
||
data, err := os.ReadFile(cq.filePath)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
var records []*ChangeRecord
|
||
if err := json.Unmarshal(data, &records); err != nil {
|
||
return
|
||
}
|
||
|
||
cq.mu.Lock()
|
||
defer cq.mu.Unlock()
|
||
cq.changes = records
|
||
|
||
// Восстанавливаем последний LSN
|
||
for _, r := range records {
|
||
if r.LSN > cq.lastLSN.Load() {
|
||
cq.lastLSN.Store(r.LSN)
|
||
}
|
||
}
|
||
}
|
||
|
||
// save сохраняет очередь в файл
|
||
func (cq *ChangeQueue) save() {
|
||
if cq.filePath == "" {
|
||
return
|
||
}
|
||
|
||
cq.mu.RLock()
|
||
data, err := json.Marshal(cq.changes)
|
||
cq.mu.RUnlock()
|
||
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
os.WriteFile(cq.filePath, data, 0644)
|
||
}
|
||
|
||
// Add добавляет изменение в очередь
|
||
func (cq *ChangeQueue) Add(change *ChangeRecord) bool {
|
||
cq.mu.Lock()
|
||
defer cq.mu.Unlock()
|
||
|
||
if len(cq.changes) >= cq.maxSize {
|
||
// Удаляем половину старых записей
|
||
cq.changes = cq.changes[len(cq.changes)/2:]
|
||
}
|
||
|
||
// Устанавливаем LSN
|
||
change.LSN = cq.lastLSN.Add(1)
|
||
|
||
cq.changes = append(cq.changes, change)
|
||
|
||
// Асинхронное сохранение
|
||
if cq.filePath != "" {
|
||
go cq.save()
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
// GetAll возвращает все изменения и очищает очередь
|
||
func (cq *ChangeQueue) GetAll() []*ChangeRecord {
|
||
cq.mu.Lock()
|
||
defer cq.mu.Unlock()
|
||
|
||
changes := cq.changes
|
||
cq.changes = make([]*ChangeRecord, 0, cq.maxSize)
|
||
|
||
if cq.filePath != "" {
|
||
os.WriteFile(cq.filePath, []byte("[]"), 0644)
|
||
}
|
||
|
||
return changes
|
||
}
|
||
|
||
// GetChangesSince возвращает изменения после указанного LSN
|
||
func (cq *ChangeQueue) GetChangesSince(lsn uint64) []*ChangeRecord {
|
||
cq.mu.RLock()
|
||
defer cq.mu.RUnlock()
|
||
|
||
result := make([]*ChangeRecord, 0)
|
||
for _, change := range cq.changes {
|
||
if change.LSN > lsn {
|
||
result = append(result, change)
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// GetChangesAfter возвращает изменения после указанного времени
|
||
func (cq *ChangeQueue) GetChangesAfter(timestamp int64) []*ChangeRecord {
|
||
cq.mu.RLock()
|
||
defer cq.mu.RUnlock()
|
||
|
||
result := make([]*ChangeRecord, 0)
|
||
for _, change := range cq.changes {
|
||
if change.Timestamp > timestamp {
|
||
result = append(result, change)
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// GetLastLSN возвращает последний LSN
|
||
func (cq *ChangeQueue) GetLastLSN() uint64 {
|
||
return cq.lastLSN.Load()
|
||
}
|
||
|
||
// Clear очищает очередь
|
||
func (cq *ChangeQueue) Clear() {
|
||
cq.mu.Lock()
|
||
defer cq.mu.Unlock()
|
||
|
||
cq.changes = make([]*ChangeRecord, 0, cq.maxSize)
|
||
if cq.filePath != "" {
|
||
os.WriteFile(cq.filePath, []byte("[]"), 0644)
|
||
}
|
||
}
|
||
|
||
// Size возвращает размер очереди
|
||
func (cq *ChangeQueue) Size() int {
|
||
cq.mu.RLock()
|
||
defer cq.mu.RUnlock()
|
||
return len(cq.changes)
|
||
}
|
||
|
||
// =============================================================================
|
||
// ОСНОВНАЯ СТРУКТУРА МИГРАТОРА
|
||
// =============================================================================
|
||
|
||
// CrossDCMigrator управляет кросс-датацентровой миграцией данных
|
||
type CrossDCMigrator struct {
|
||
config *config.MigrationConfig
|
||
store *storage.Storage
|
||
logger *log.Logger
|
||
httpClient *http.Client
|
||
mu sync.RWMutex
|
||
tasks map[string]*MigrationTask
|
||
stats *MigrationStats
|
||
changeQueue *ChangeQueue
|
||
active atomic.Bool
|
||
stopChan chan struct{}
|
||
wg sync.WaitGroup
|
||
mode MigrationMode
|
||
currentTaskID string
|
||
checkpointPath string
|
||
resumedFrom *MigrationTask
|
||
deltaTicker *time.Ticker
|
||
changeSubscribers []chan *ChangeRecord
|
||
subscriberMu sync.RWMutex
|
||
}
|
||
|
||
// NewCrossDCMigrator создаёт новый экземпляр мигратора
|
||
func NewCrossDCMigrator(cfg *config.MigrationConfig, store *storage.Storage, logger *log.Logger) *CrossDCMigrator {
|
||
if cfg == nil {
|
||
cfg = &config.MigrationConfig{
|
||
Enabled: false,
|
||
Mode: "semi_auto",
|
||
Source: &config.DatacenterConfig{
|
||
Name: "dc-primary",
|
||
Endpoint: "localhost:8080",
|
||
TimeoutSec: 30,
|
||
},
|
||
Target: &config.DatacenterConfig{
|
||
Name: "dc-secondary",
|
||
Endpoint: "localhost:8081",
|
||
TimeoutSec: 30,
|
||
},
|
||
Settings: &config.MigrationSettings{
|
||
BatchSize: 1000,
|
||
Workers: 4,
|
||
Compression: "snappy",
|
||
ResumeEnabled: true,
|
||
CheckpointIntervalSec: 30,
|
||
MaxRetries: 3,
|
||
RetryBackoffSec: 5,
|
||
},
|
||
Delta: &config.DeltaSyncConfig{
|
||
Enabled: true,
|
||
IntervalSec: 60,
|
||
MaxLagSec: 300,
|
||
},
|
||
Validation: &config.ValidationConfig{
|
||
Enabled: true,
|
||
SamplePercent: 10,
|
||
MaxErrors: 100,
|
||
},
|
||
}
|
||
}
|
||
|
||
cdm := &CrossDCMigrator{
|
||
config: cfg,
|
||
store: store,
|
||
logger: logger,
|
||
tasks: make(map[string]*MigrationTask),
|
||
stats: &MigrationStats{},
|
||
changeQueue: NewChangeQueue(100000, "futriis/migration/change_queue.json"),
|
||
stopChan: make(chan struct{}),
|
||
mode: MigrationMode(cfg.Mode),
|
||
checkpointPath: "futriis/migration/checkpoints",
|
||
changeSubscribers: make([]chan *ChangeRecord, 0),
|
||
httpClient: &http.Client{
|
||
Timeout: time.Duration(cfg.Source.TimeoutSec) * time.Second,
|
||
},
|
||
}
|
||
|
||
// Создаём директорию для чекпоинтов
|
||
os.MkdirAll(cdm.checkpointPath, 0755)
|
||
os.MkdirAll("futriis/migration", 0755)
|
||
|
||
// Загружаем сохранённые задачи
|
||
cdm.loadTasks()
|
||
|
||
return cdm
|
||
}
|
||
|
||
// Start запускает мигратор
|
||
func (cdm *CrossDCMigrator) Start() {
|
||
if !cdm.active.CompareAndSwap(false, true) {
|
||
return
|
||
}
|
||
|
||
cdm.logger.Info("Cross-datacenter migrator started")
|
||
|
||
// Восстанавливаем незавершённые миграции
|
||
cdm.resumePendingMigrations()
|
||
|
||
// Запускаем обработку изменений
|
||
cdm.wg.Add(1)
|
||
go cdm.processChanges()
|
||
|
||
// Запускаем дельта-синхронизацию если включена
|
||
if cdm.config.Delta.Enabled {
|
||
cdm.wg.Add(1)
|
||
go cdm.deltaSyncLoop()
|
||
}
|
||
|
||
// Запускаем сохранение чекпоинтов
|
||
cdm.wg.Add(1)
|
||
go cdm.checkpointLoop()
|
||
}
|
||
|
||
// Stop останавливает мигратор
|
||
func (cdm *CrossDCMigrator) Stop() {
|
||
if !cdm.active.Load() {
|
||
return
|
||
}
|
||
|
||
close(cdm.stopChan)
|
||
cdm.wg.Wait()
|
||
cdm.active.Store(false)
|
||
|
||
cdm.saveTasks()
|
||
cdm.logger.Info("Cross-datacenter migrator stopped")
|
||
}
|
||
|
||
// =============================================================================
|
||
// ОСНОВНЫЕ МЕТОДЫ МИГРАЦИИ
|
||
// =============================================================================
|
||
|
||
// StartMigration начинает миграцию данных
|
||
func (cdm *CrossDCMigrator) StartMigration(sourceDC, targetDC string, databases []string, collections map[string][]string) (*MigrationTask, error) {
|
||
if !cdm.config.Enabled {
|
||
return nil, fmt.Errorf("migration is disabled in configuration")
|
||
}
|
||
|
||
cdm.mu.Lock()
|
||
defer cdm.mu.Unlock()
|
||
|
||
// Проверяем, есть ли активная миграция
|
||
if cdm.currentTaskID != "" {
|
||
if task, ok := cdm.tasks[cdm.currentTaskID]; ok {
|
||
task.mu.RLock()
|
||
status := task.Status
|
||
task.mu.RUnlock()
|
||
if status == MigrationStatusMigrating || status == MigrationStatusDeltaSync {
|
||
return nil, fmt.Errorf("a migration is already in progress")
|
||
}
|
||
}
|
||
}
|
||
|
||
taskID := fmt.Sprintf("mig_%d_%s", time.Now().UnixNano(), sourceDC)
|
||
|
||
task := &MigrationTask{
|
||
ID: taskID,
|
||
SourceDC: sourceDC,
|
||
TargetDC: targetDC,
|
||
Databases: databases,
|
||
Collections: collections,
|
||
Status: MigrationStatusPreparing,
|
||
ProgressPercent: 0,
|
||
StartTime: time.Now().UnixMilli(),
|
||
CheckpointData: make(map[string]interface{}),
|
||
CreatedAt: time.Now().UnixMilli(),
|
||
UpdatedAt: time.Now().UnixMilli(),
|
||
}
|
||
|
||
cdm.tasks[taskID] = task
|
||
cdm.currentTaskID = taskID
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Migration task %s created: %s -> %s", taskID, sourceDC, targetDC))
|
||
|
||
// Запускаем миграцию асинхронно
|
||
cdm.wg.Add(1)
|
||
go cdm.runMigration(task)
|
||
|
||
return task, nil
|
||
}
|
||
|
||
// runMigration выполняет миграцию
|
||
func (cdm *CrossDCMigrator) runMigration(task *MigrationTask) {
|
||
defer cdm.wg.Done()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
cdm.logger.Error(fmt.Sprintf("Migration %s panicked: %v", task.ID, r))
|
||
task.mu.Lock()
|
||
task.Status = MigrationStatusFailed
|
||
task.Error = fmt.Sprintf("panic: %v", r)
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
}
|
||
}()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Starting migration %s", task.ID))
|
||
|
||
// 1. Подготовка - сбор метаданных
|
||
if err := cdm.prepareMigration(task); err != nil {
|
||
cdm.setTaskError(task, err)
|
||
return
|
||
}
|
||
|
||
// 2. Основная миграция данных
|
||
if err := cdm.migrateData(task); err != nil {
|
||
cdm.setTaskError(task, err)
|
||
return
|
||
}
|
||
|
||
// 3. Дельта-синхронизация (если включена)
|
||
if cdm.config.Delta.Enabled && cdm.mode != MigrationModeManual {
|
||
task.mu.Lock()
|
||
task.Status = MigrationStatusDeltaSync
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
if err := cdm.runDeltaSync(task); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Delta sync failed: %v", err))
|
||
// Не считаем это фатальной ошибкой
|
||
}
|
||
}
|
||
|
||
// 4. Валидация данных
|
||
if cdm.config.Validation.Enabled {
|
||
task.mu.Lock()
|
||
task.Status = MigrationStatusValidating
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
if err := cdm.validateMigration(task); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Validation failed: %v", err))
|
||
// Продолжаем, но отмечаем проблемы
|
||
}
|
||
}
|
||
|
||
// 5. Завершение
|
||
task.mu.Lock()
|
||
task.Status = MigrationStatusCompleted
|
||
task.EndTime = time.Now().UnixMilli()
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Migration %s completed successfully", task.ID))
|
||
}
|
||
|
||
// prepareMigration подготавливает миграцию
|
||
func (cdm *CrossDCMigrator) prepareMigration(task *MigrationTask) error {
|
||
cdm.logger.Debug(fmt.Sprintf("Preparing migration %s", task.ID))
|
||
|
||
// Получаем список баз данных
|
||
databases := task.Databases
|
||
if len(databases) == 0 {
|
||
databases = cdm.store.ListDatabases()
|
||
}
|
||
|
||
totalDocs := int64(0)
|
||
for _, dbName := range databases {
|
||
db, err := cdm.store.GetDatabase(dbName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
collections := task.Collections[dbName]
|
||
if len(collections) == 0 {
|
||
collections = db.ListCollections()
|
||
}
|
||
|
||
for _, collName := range collections {
|
||
coll, err := db.GetCollection(collName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
totalDocs += coll.Count()
|
||
}
|
||
}
|
||
|
||
task.mu.Lock()
|
||
task.TotalDocuments = totalDocs
|
||
task.Status = MigrationStatusMigrating
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Prepared migration %s: %d documents to migrate", task.ID, totalDocs))
|
||
return nil
|
||
}
|
||
|
||
// migrateData выполняет основную миграцию данных
|
||
func (cdm *CrossDCMigrator) migrateData(task *MigrationTask) error {
|
||
cdm.logger.Debug(fmt.Sprintf("Migrating data for task %s", task.ID))
|
||
|
||
databases := task.Databases
|
||
if len(databases) == 0 {
|
||
databases = cdm.store.ListDatabases()
|
||
}
|
||
|
||
// Проверяем наличие чекпоинта для возобновления
|
||
var checkpoint *MigrationCheckpoint
|
||
if cdm.config.Settings.ResumeEnabled {
|
||
checkpoint = cdm.loadCheckpoint(task.ID)
|
||
if checkpoint != nil {
|
||
// Преобразуем map[string]bool в map[string]interface{} для совместимости
|
||
checkpointData := make(map[string]interface{})
|
||
for k, v := range checkpoint.ProcessedDocs {
|
||
checkpointData[k] = v
|
||
}
|
||
task.mu.Lock()
|
||
task.CheckpointData = checkpointData
|
||
task.mu.Unlock()
|
||
cdm.logger.Info(fmt.Sprintf("Resuming migration %s from checkpoint (LSN: %d)", task.ID, checkpoint.LastLSN))
|
||
}
|
||
}
|
||
|
||
for _, dbName := range databases {
|
||
// Проверяем остановку
|
||
if !cdm.active.Load() {
|
||
return fmt.Errorf("migration stopped")
|
||
}
|
||
|
||
db, err := cdm.store.GetDatabase(dbName)
|
||
if err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Database %s not found: %v", dbName, err))
|
||
continue
|
||
}
|
||
|
||
collections := task.Collections[dbName]
|
||
if len(collections) == 0 {
|
||
collections = db.ListCollections()
|
||
}
|
||
|
||
// Проверяем исключения
|
||
excludeMap := make(map[string]bool)
|
||
for _, excl := range cdm.config.Settings.ExcludeCollections {
|
||
excludeMap[excl] = true
|
||
}
|
||
|
||
for _, collName := range collections {
|
||
if excludeMap[collName] {
|
||
continue
|
||
}
|
||
|
||
// Проверяем остановку
|
||
if !cdm.active.Load() {
|
||
return fmt.Errorf("migration stopped")
|
||
}
|
||
|
||
// Проверяем, была ли коллекция уже обработана (по чекпоинту)
|
||
checkpointKey := fmt.Sprintf("%s:%s", dbName, collName)
|
||
if checkpoint != nil {
|
||
if done, ok := checkpoint.ProcessedDocs[checkpointKey]; ok && done {
|
||
cdm.logger.Debug(fmt.Sprintf("Collection %s.%s already migrated, skipping", dbName, collName))
|
||
continue
|
||
}
|
||
}
|
||
|
||
if err := cdm.migrateCollection(task, dbName, collName, checkpoint); err != nil {
|
||
cdm.logger.Error(fmt.Sprintf("Failed to migrate collection %s.%s: %v", dbName, collName, err))
|
||
// Продолжаем с другими коллекциями
|
||
}
|
||
|
||
// Сохраняем чекпоинт для коллекции
|
||
if cdm.config.Settings.ResumeEnabled {
|
||
if checkpoint == nil {
|
||
checkpoint = &MigrationCheckpoint{
|
||
TaskID: task.ID,
|
||
ProcessedDocs: make(map[string]bool),
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Version: 1,
|
||
}
|
||
}
|
||
checkpoint.ProcessedDocs[checkpointKey] = true
|
||
checkpoint.Timestamp = time.Now().UnixMilli()
|
||
checkpoint.Version++
|
||
cdm.saveCheckpoint(task.ID, checkpoint)
|
||
// Преобразуем для task.CheckpointData
|
||
checkpointData := make(map[string]interface{})
|
||
for k, v := range checkpoint.ProcessedDocs {
|
||
checkpointData[k] = v
|
||
}
|
||
task.mu.Lock()
|
||
task.CheckpointData = checkpointData
|
||
task.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// migrateCollection мигрирует одну коллекцию
|
||
func (cdm *CrossDCMigrator) migrateCollection(task *MigrationTask, dbName, collName string, checkpoint *MigrationCheckpoint) error {
|
||
db, err := cdm.store.GetDatabase(dbName)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
coll, err := db.GetCollection(collName)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// Получаем все документы
|
||
docs := coll.GetAllDocuments()
|
||
if len(docs) == 0 {
|
||
return nil
|
||
}
|
||
|
||
batchSize := cdm.config.Settings.BatchSize
|
||
workers := cdm.config.Settings.Workers
|
||
|
||
// Создаём каналы для параллельной обработки
|
||
docChan := make(chan *storage.Document, batchSize)
|
||
resultChan := make(chan error, workers)
|
||
|
||
// Запускаем воркеры
|
||
var wg sync.WaitGroup
|
||
for i := 0; i < workers; i++ {
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
for doc := range docChan {
|
||
err := cdm.sendDocumentToTarget(task, dbName, collName, doc)
|
||
if err != nil {
|
||
resultChan <- err
|
||
task.mu.Lock()
|
||
task.FailedDocs++
|
||
task.mu.Unlock()
|
||
} else {
|
||
task.mu.Lock()
|
||
task.MigratedDocs++
|
||
// Обновляем прогресс
|
||
if task.TotalDocuments > 0 {
|
||
task.ProgressPercent = float64(task.MigratedDocs) / float64(task.TotalDocuments) * 100
|
||
}
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
// Отправляем документы в канал
|
||
sent := 0
|
||
for _, doc := range docs {
|
||
// Проверяем остановку
|
||
if !cdm.active.Load() {
|
||
close(docChan)
|
||
wg.Wait()
|
||
return fmt.Errorf("migration stopped")
|
||
}
|
||
|
||
// Проверяем, был ли документ уже обработан (по чекпоинту)
|
||
if checkpoint != nil {
|
||
if done, ok := checkpoint.ProcessedDocs[doc.ID]; ok && done {
|
||
task.mu.Lock()
|
||
task.SkippedDocs++
|
||
task.mu.Unlock()
|
||
continue
|
||
}
|
||
}
|
||
|
||
docChan <- doc
|
||
sent++
|
||
|
||
// Сохраняем чекпоинт каждые N документов
|
||
if sent%batchSize == 0 && cdm.config.Settings.ResumeEnabled {
|
||
if checkpoint == nil {
|
||
checkpoint = &MigrationCheckpoint{
|
||
TaskID: task.ID,
|
||
ProcessedDocs: make(map[string]bool),
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Version: 1,
|
||
}
|
||
}
|
||
checkpoint.ProcessedDocs[doc.ID] = true
|
||
checkpoint.Timestamp = time.Now().UnixMilli()
|
||
checkpoint.Version++
|
||
cdm.saveCheckpoint(task.ID, checkpoint)
|
||
}
|
||
}
|
||
|
||
close(docChan)
|
||
wg.Wait()
|
||
|
||
// Проверяем ошибки
|
||
var lastErr error
|
||
select {
|
||
case err := <-resultChan:
|
||
lastErr = err
|
||
default:
|
||
}
|
||
|
||
// Обновляем статистику
|
||
task.mu.Lock()
|
||
task.ProgressPercent = 100.0
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
return lastErr
|
||
}
|
||
|
||
// sendDocumentToTarget отправляет документ на целевой датацентр
|
||
func (cdm *CrossDCMigrator) sendDocumentToTarget(task *MigrationTask, dbName, collName string, doc *storage.Document) error {
|
||
// Формируем данные документа
|
||
docData := doc.GetFields()
|
||
|
||
// Добавляем метаданные
|
||
docData["_meta"] = map[string]interface{}{
|
||
"source_dc": task.SourceDC,
|
||
"source_db": dbName,
|
||
"source_collection": collName,
|
||
"migration_id": task.ID,
|
||
"migrated_at": time.Now().UnixMilli(),
|
||
}
|
||
|
||
// Создаём запись изменения
|
||
change := &ChangeRecord{
|
||
ID: fmt.Sprintf("%s_%s_%s", task.ID, doc.ID, time.Now().Format("20060102150405")),
|
||
Database: dbName,
|
||
Collection: collName,
|
||
DocumentID: doc.ID,
|
||
ChangeType: ChangeInsert,
|
||
Document: docData,
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Version: doc.Version,
|
||
Applied: false,
|
||
}
|
||
|
||
// Вычисляем контрольную сумму
|
||
data, _ := json.Marshal(docData)
|
||
hash := sha256.Sum256(data)
|
||
change.Checksum = hex.EncodeToString(hash[:])
|
||
|
||
// Добавляем в очередь изменений
|
||
cdm.changeQueue.Add(change)
|
||
|
||
// Пытаемся отправить напрямую
|
||
targetEndpoint := cdm.config.Target.Endpoint
|
||
if targetEndpoint == "" {
|
||
return fmt.Errorf("target endpoint not configured")
|
||
}
|
||
|
||
// Формируем запрос
|
||
reqData := map[string]interface{}{
|
||
"database": dbName,
|
||
"collection": collName,
|
||
"document": docData,
|
||
"migration": map[string]interface{}{
|
||
"id": task.ID,
|
||
"source": task.SourceDC,
|
||
},
|
||
}
|
||
|
||
jsonData, err := json.Marshal(reqData)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// Сжимаем данные
|
||
compressedData, err := cdm.compressData(jsonData)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// Отправляем HTTP запрос
|
||
url := fmt.Sprintf("%s/api/migration/receive", targetEndpoint)
|
||
req, err := http.NewRequest("POST", url, bytes.NewReader(compressedData))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
req.Header.Set("Content-Type", "application/octet-stream")
|
||
req.Header.Set("Content-Encoding", cdm.config.Settings.Compression)
|
||
req.Header.Set("X-Migration-ID", task.ID)
|
||
|
||
// Отправляем с повторными попытками
|
||
var lastErr error
|
||
for retry := 0; retry < cdm.config.Settings.MaxRetries; retry++ {
|
||
resp, err := cdm.httpClient.Do(req)
|
||
if err == nil {
|
||
resp.Body.Close()
|
||
if resp.StatusCode == http.StatusOK {
|
||
change.Applied = true
|
||
change.AppliedAt = time.Now().UnixMilli()
|
||
return nil
|
||
}
|
||
lastErr = fmt.Errorf("status: %s", resp.Status)
|
||
} else {
|
||
lastErr = err
|
||
}
|
||
|
||
time.Sleep(time.Duration(cdm.config.Settings.RetryBackoffSec) * time.Second)
|
||
}
|
||
|
||
return lastErr
|
||
}
|
||
|
||
// =============================================================================
|
||
// ДЕЛЬТА-СИНХРОНИЗАЦИЯ
|
||
// =============================================================================
|
||
|
||
// runDeltaSync выполняет дельта-синхронизацию
|
||
func (cdm *CrossDCMigrator) runDeltaSync(task *MigrationTask) error {
|
||
cdm.logger.Info(fmt.Sprintf("Starting delta sync for migration %s", task.ID))
|
||
|
||
// Получаем изменения после последнего чекпоинта
|
||
since := task.LastCheckpoint
|
||
if since == 0 {
|
||
since = task.StartTime
|
||
}
|
||
|
||
// Собираем изменения из очереди
|
||
changes := cdm.changeQueue.GetChangesAfter(since)
|
||
|
||
batchSize := cdm.config.Settings.BatchSize
|
||
for i := 0; i < len(changes); i += batchSize {
|
||
end := i + batchSize
|
||
if end > len(changes) {
|
||
end = len(changes)
|
||
}
|
||
|
||
batch := changes[i:end]
|
||
if err := cdm.applyChangeBatch(task, batch); err != nil {
|
||
cdm.logger.Error(fmt.Sprintf("Failed to apply change batch: %v", err))
|
||
return err
|
||
}
|
||
|
||
// Обновляем прогресс
|
||
task.mu.Lock()
|
||
task.MigratedDocs += int64(len(batch))
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
}
|
||
|
||
task.mu.Lock()
|
||
task.LastCheckpoint = time.Now().UnixMilli()
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Delta sync completed for migration %s: %d changes applied", task.ID, len(changes)))
|
||
return nil
|
||
}
|
||
|
||
// applyChangeBatch применяет пакет изменений
|
||
func (cdm *CrossDCMigrator) applyChangeBatch(task *MigrationTask, changes []*ChangeRecord) error {
|
||
for _, change := range changes {
|
||
if !cdm.active.Load() {
|
||
return fmt.Errorf("migration stopped")
|
||
}
|
||
|
||
// Применяем изменение в целевом датацентре
|
||
if err := cdm.applyChange(task, change); err != nil {
|
||
cdm.logger.Error(fmt.Sprintf("Failed to apply change %s: %v", change.ID, err))
|
||
continue
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// applyChange применяет одно изменение
|
||
func (cdm *CrossDCMigrator) applyChange(task *MigrationTask, change *ChangeRecord) error {
|
||
// Формируем запрос к целевому датацентру
|
||
reqData := map[string]interface{}{
|
||
"database": change.Database,
|
||
"collection": change.Collection,
|
||
"document_id": change.DocumentID,
|
||
"change_type": string(change.ChangeType),
|
||
"document": change.Document,
|
||
"previous_doc": change.PreviousDoc,
|
||
"migration_id": task.ID,
|
||
"timestamp": change.Timestamp,
|
||
}
|
||
|
||
jsonData, err := json.Marshal(reqData)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
compressedData, err := cdm.compressData(jsonData)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
url := fmt.Sprintf("%s/api/migration/apply", cdm.config.Target.Endpoint)
|
||
req, err := http.NewRequest("POST", url, bytes.NewReader(compressedData))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
req.Header.Set("Content-Type", "application/octet-stream")
|
||
req.Header.Set("Content-Encoding", cdm.config.Settings.Compression)
|
||
|
||
var lastErr error
|
||
for retry := 0; retry < cdm.config.Settings.MaxRetries; retry++ {
|
||
resp, err := cdm.httpClient.Do(req)
|
||
if err == nil {
|
||
resp.Body.Close()
|
||
if resp.StatusCode == http.StatusOK {
|
||
change.Applied = true
|
||
change.AppliedAt = time.Now().UnixMilli()
|
||
cdm.stats.mu.Lock()
|
||
cdm.stats.AppliedChanges++
|
||
cdm.stats.mu.Unlock()
|
||
return nil
|
||
}
|
||
lastErr = fmt.Errorf("status: %s", resp.Status)
|
||
} else {
|
||
lastErr = err
|
||
}
|
||
|
||
time.Sleep(time.Duration(cdm.config.Settings.RetryBackoffSec) * time.Second)
|
||
}
|
||
|
||
cdm.stats.mu.Lock()
|
||
cdm.stats.FailedChanges++
|
||
cdm.stats.mu.Unlock()
|
||
|
||
return lastErr
|
||
}
|
||
|
||
// =============================================================================
|
||
// ВАЛИДАЦИЯ
|
||
// =============================================================================
|
||
|
||
// validateMigration выполняет валидацию мигрированных данных
|
||
func (cdm *CrossDCMigrator) validateMigration(task *MigrationTask) error {
|
||
cdm.logger.Info(fmt.Sprintf("Validating migration %s", task.ID))
|
||
|
||
if cdm.config.Validation.SamplePercent <= 0 {
|
||
return nil
|
||
}
|
||
|
||
databases := task.Databases
|
||
if len(databases) == 0 {
|
||
databases = cdm.store.ListDatabases()
|
||
}
|
||
|
||
errors := 0
|
||
|
||
for _, dbName := range databases {
|
||
db, err := cdm.store.GetDatabase(dbName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
collections := task.Collections[dbName]
|
||
if len(collections) == 0 {
|
||
collections = db.ListCollections()
|
||
}
|
||
|
||
for _, collName := range collections {
|
||
coll, err := db.GetCollection(collName)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
|
||
docs := coll.GetAllDocuments()
|
||
sampleSize := len(docs) * cdm.config.Validation.SamplePercent / 100
|
||
if sampleSize < 1 {
|
||
sampleSize = 1
|
||
}
|
||
|
||
for i := 0; i < sampleSize && i < len(docs); i++ {
|
||
doc := docs[i]
|
||
if err := cdm.validateDocument(task, dbName, collName, doc); err != nil {
|
||
errors++
|
||
if errors >= cdm.config.Validation.MaxErrors {
|
||
return fmt.Errorf("validation stopped: too many errors (%d)", errors)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Validation completed for %s: %d errors", task.ID, errors))
|
||
return nil
|
||
}
|
||
|
||
// validateDocument валидирует один документ
|
||
func (cdm *CrossDCMigrator) validateDocument(task *MigrationTask, dbName, collName string, doc *storage.Document) error {
|
||
docData := doc.GetFields()
|
||
data, _ := json.Marshal(docData)
|
||
hash := sha256.Sum256(data)
|
||
checksum := hex.EncodeToString(hash[:])
|
||
|
||
// Запрашиваем документ из целевого датацентра
|
||
url := fmt.Sprintf("%s/api/migration/validate?database=%s&collection=%s&document_id=%s",
|
||
cdm.config.Target.Endpoint, dbName, collName, doc.ID)
|
||
|
||
resp, err := cdm.httpClient.Get(url)
|
||
if err != nil {
|
||
return fmt.Errorf("validation request failed: %v", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode != http.StatusOK {
|
||
return fmt.Errorf("target document not found: %s", doc.ID)
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||
return fmt.Errorf("failed to decode validation response: %v", err)
|
||
}
|
||
|
||
targetChecksum, ok := result["checksum"].(string)
|
||
if !ok {
|
||
return fmt.Errorf("invalid checksum format")
|
||
}
|
||
|
||
if targetChecksum != checksum {
|
||
return fmt.Errorf("checksum mismatch for %s: source=%s, target=%s", doc.ID, checksum, targetChecksum)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// =============================================================================
|
||
// ЧЕКПОИНТЫ
|
||
// =============================================================================
|
||
|
||
// saveCheckpoint сохраняет чекпоинт миграции
|
||
func (cdm *CrossDCMigrator) saveCheckpoint(taskID string, checkpoint *MigrationCheckpoint) {
|
||
if !cdm.config.Settings.ResumeEnabled {
|
||
return
|
||
}
|
||
|
||
cdm.mu.Lock()
|
||
defer cdm.mu.Unlock()
|
||
|
||
path := filepath.Join(cdm.checkpointPath, fmt.Sprintf("%s.json", taskID))
|
||
|
||
checkpoint.Timestamp = time.Now().UnixMilli()
|
||
checkpoint.Version++
|
||
|
||
jsonData, err := json.MarshalIndent(checkpoint, "", " ")
|
||
if err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to marshal checkpoint: %v", err))
|
||
return
|
||
}
|
||
|
||
if err := os.WriteFile(path, jsonData, 0644); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to save checkpoint: %v", err))
|
||
return
|
||
}
|
||
|
||
cdm.logger.Debug(fmt.Sprintf("Checkpoint saved for task %s (LSN: %d, docs: %d)",
|
||
taskID, checkpoint.LastLSN, len(checkpoint.ProcessedDocs)))
|
||
}
|
||
|
||
// loadCheckpoint загружает чекпоинт миграции
|
||
func (cdm *CrossDCMigrator) loadCheckpoint(taskID string) *MigrationCheckpoint {
|
||
path := filepath.Join(cdm.checkpointPath, fmt.Sprintf("%s.json", taskID))
|
||
|
||
data, err := os.ReadFile(path)
|
||
if err != nil {
|
||
if !os.IsNotExist(err) {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to read checkpoint: %v", err))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
var checkpoint MigrationCheckpoint
|
||
if err := json.Unmarshal(data, &checkpoint); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to unmarshal checkpoint: %v", err))
|
||
return nil
|
||
}
|
||
|
||
cdm.logger.Debug(fmt.Sprintf("Checkpoint loaded for task %s (LSN: %d, docs: %d)",
|
||
taskID, checkpoint.LastLSN, len(checkpoint.ProcessedDocs)))
|
||
return &checkpoint
|
||
}
|
||
|
||
// =============================================================================
|
||
// ДЕЛЬТА-СИНХРОНИЗАЦИЯ В ЦИКЛЕ
|
||
// =============================================================================
|
||
|
||
// deltaSyncLoop выполняет периодическую дельта-синхронизацию
|
||
func (cdm *CrossDCMigrator) deltaSyncLoop() {
|
||
defer cdm.wg.Done()
|
||
|
||
interval := time.Duration(cdm.config.Delta.IntervalSec) * time.Second
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-cdm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
if !cdm.active.Load() {
|
||
continue
|
||
}
|
||
cdm.runPeriodicDeltaSync()
|
||
}
|
||
}
|
||
}
|
||
|
||
// runPeriodicDeltaSync выполняет периодическую дельта-синхронизацию
|
||
func (cdm *CrossDCMigrator) runPeriodicDeltaSync() {
|
||
cdm.mu.RLock()
|
||
taskID := cdm.currentTaskID
|
||
cdm.mu.RUnlock()
|
||
|
||
if taskID == "" {
|
||
return
|
||
}
|
||
|
||
task, ok := cdm.tasks[taskID]
|
||
if !ok {
|
||
return
|
||
}
|
||
|
||
task.mu.RLock()
|
||
status := task.Status
|
||
task.mu.RUnlock()
|
||
|
||
if status != MigrationStatusCompleted && status != MigrationStatusDeltaSync {
|
||
return
|
||
}
|
||
|
||
cdm.logger.Debug("Running periodic delta sync")
|
||
|
||
if err := cdm.runDeltaSync(task); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Periodic delta sync failed: %v", err))
|
||
}
|
||
}
|
||
|
||
// =============================================================================
|
||
// ОБРАБОТКА ИЗМЕНЕНИЙ В ОЧЕРЕДИ
|
||
// =============================================================================
|
||
|
||
// processChanges обрабатывает изменения из очереди
|
||
func (cdm *CrossDCMigrator) processChanges() {
|
||
defer cdm.wg.Done()
|
||
|
||
ticker := time.NewTicker(5 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-cdm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
if !cdm.active.Load() {
|
||
continue
|
||
}
|
||
cdm.flushChangeQueue()
|
||
}
|
||
}
|
||
}
|
||
|
||
// flushChangeQueue сбрасывает очередь изменений
|
||
func (cdm *CrossDCMigrator) flushChangeQueue() {
|
||
changes := cdm.changeQueue.GetAll()
|
||
if len(changes) == 0 {
|
||
return
|
||
}
|
||
|
||
cdm.mu.RLock()
|
||
taskID := cdm.currentTaskID
|
||
cdm.mu.RUnlock()
|
||
|
||
if taskID == "" {
|
||
cdm.logger.Warn("No active migration, dropping changes")
|
||
cdm.changeQueue.Clear()
|
||
return
|
||
}
|
||
|
||
task, ok := cdm.tasks[taskID]
|
||
if !ok {
|
||
cdm.logger.Warn("Migration task not found, dropping changes")
|
||
cdm.changeQueue.Clear()
|
||
return
|
||
}
|
||
|
||
// Применяем изменения
|
||
for _, change := range changes {
|
||
if !cdm.active.Load() {
|
||
return
|
||
}
|
||
|
||
if err := cdm.applyChange(task, change); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to apply change: %v", err))
|
||
}
|
||
}
|
||
|
||
cdm.logger.Debug(fmt.Sprintf("Flushed %d changes to target", len(changes)))
|
||
}
|
||
|
||
// =============================================================================
|
||
// ВСПОМОГАТЕЛЬНЫЕ МЕТОДЫ
|
||
// =============================================================================
|
||
|
||
// compressData сжимает данные
|
||
func (cdm *CrossDCMigrator) compressData(data []byte) ([]byte, error) {
|
||
var buf bytes.Buffer
|
||
|
||
switch cdm.config.Settings.Compression {
|
||
case "gzip", "":
|
||
w := gzip.NewWriter(&buf)
|
||
defer w.Close()
|
||
if _, err := w.Write(data); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := w.Flush(); err != nil {
|
||
return nil, err
|
||
}
|
||
default:
|
||
return data, nil
|
||
}
|
||
|
||
return buf.Bytes(), nil
|
||
}
|
||
|
||
// setTaskError устанавливает ошибку задачи
|
||
func (cdm *CrossDCMigrator) setTaskError(task *MigrationTask, err error) {
|
||
task.mu.Lock()
|
||
task.Status = MigrationStatusFailed
|
||
task.Error = err.Error()
|
||
task.EndTime = time.Now().UnixMilli()
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Error(fmt.Sprintf("Migration %s failed: %v", task.ID, err))
|
||
}
|
||
|
||
// loadTasks загружает сохранённые задачи
|
||
func (cdm *CrossDCMigrator) loadTasks() {
|
||
path := "futriis/migration/tasks.json"
|
||
data, err := os.ReadFile(path)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
var tasks map[string]*MigrationTask
|
||
if err := json.Unmarshal(data, &tasks); err != nil {
|
||
return
|
||
}
|
||
|
||
cdm.mu.Lock()
|
||
for id, task := range tasks {
|
||
cdm.tasks[id] = task
|
||
}
|
||
cdm.mu.Unlock()
|
||
|
||
cdm.logger.Debug(fmt.Sprintf("Loaded %d migration tasks", len(tasks)))
|
||
}
|
||
|
||
// saveTasks сохраняет задачи
|
||
func (cdm *CrossDCMigrator) saveTasks() {
|
||
cdm.mu.RLock()
|
||
tasks := make(map[string]*MigrationTask)
|
||
for id, task := range cdm.tasks {
|
||
tasks[id] = task
|
||
}
|
||
cdm.mu.RUnlock()
|
||
|
||
data, err := json.MarshalIndent(tasks, "", " ")
|
||
if err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to marshal tasks: %v", err))
|
||
return
|
||
}
|
||
|
||
if err := os.WriteFile("futriis/migration/tasks.json", data, 0644); err != nil {
|
||
cdm.logger.Warn(fmt.Sprintf("Failed to save tasks: %v", err))
|
||
}
|
||
}
|
||
|
||
// resumePendingMigrations восстанавливает незавершённые миграции
|
||
func (cdm *CrossDCMigrator) resumePendingMigrations() {
|
||
cdm.mu.RLock()
|
||
var pending []*MigrationTask
|
||
for _, task := range cdm.tasks {
|
||
task.mu.RLock()
|
||
status := task.Status
|
||
task.mu.RUnlock()
|
||
if status == MigrationStatusMigrating || status == MigrationStatusDeltaSync {
|
||
pending = append(pending, task)
|
||
}
|
||
}
|
||
cdm.mu.RUnlock()
|
||
|
||
for _, task := range pending {
|
||
if cdm.config.Settings.ResumeEnabled {
|
||
cdm.logger.Info(fmt.Sprintf("Resuming migration %s", task.ID))
|
||
cdm.wg.Add(1)
|
||
go cdm.runMigration(task)
|
||
}
|
||
}
|
||
}
|
||
|
||
// checkpointLoop периодически сохраняет чекпоинты
|
||
func (cdm *CrossDCMigrator) checkpointLoop() {
|
||
defer cdm.wg.Done()
|
||
|
||
interval := time.Duration(cdm.config.Settings.CheckpointIntervalSec) * time.Second
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-cdm.stopChan:
|
||
return
|
||
case <-ticker.C:
|
||
if !cdm.active.Load() {
|
||
continue
|
||
}
|
||
cdm.mu.RLock()
|
||
taskID := cdm.currentTaskID
|
||
cdm.mu.RUnlock()
|
||
|
||
if taskID == "" {
|
||
continue
|
||
}
|
||
|
||
task, ok := cdm.tasks[taskID]
|
||
if !ok {
|
||
continue
|
||
}
|
||
|
||
task.mu.RLock()
|
||
status := task.Status
|
||
checkpointData := task.CheckpointData
|
||
task.mu.RUnlock()
|
||
|
||
if status != MigrationStatusMigrating && status != MigrationStatusDeltaSync {
|
||
continue
|
||
}
|
||
|
||
// Сохраняем чекпоинт
|
||
checkpoint := &MigrationCheckpoint{
|
||
TaskID: taskID,
|
||
LastLSN: cdm.changeQueue.GetLastLSN(),
|
||
ProcessedDocs: make(map[string]bool),
|
||
Timestamp: time.Now().UnixMilli(),
|
||
Version: 1,
|
||
}
|
||
|
||
// Конвертируем checkpointData в map[string]bool
|
||
for k, v := range checkpointData {
|
||
if b, ok := v.(bool); ok {
|
||
checkpoint.ProcessedDocs[k] = b
|
||
}
|
||
}
|
||
|
||
cdm.saveCheckpoint(taskID, checkpoint)
|
||
}
|
||
}
|
||
}
|
||
|
||
// =============================================================================
|
||
// ПУБЛИЧНЫЕ МЕТОДЫ ДЛЯ REPL
|
||
// =============================================================================
|
||
|
||
// GetMigrationStatus возвращает статус миграции
|
||
func (cdm *CrossDCMigrator) GetMigrationStatus(taskID string) (*MigrationTask, error) {
|
||
cdm.mu.RLock()
|
||
defer cdm.mu.RUnlock()
|
||
|
||
task, ok := cdm.tasks[taskID]
|
||
if !ok {
|
||
return nil, fmt.Errorf("task %s not found", taskID)
|
||
}
|
||
|
||
return task, nil
|
||
}
|
||
|
||
// GetCurrentTaskID возвращает ID текущей задачи
|
||
func (cdm *CrossDCMigrator) GetCurrentTaskID() string {
|
||
cdm.mu.RLock()
|
||
defer cdm.mu.RUnlock()
|
||
return cdm.currentTaskID
|
||
}
|
||
|
||
// ListTasks возвращает список всех задач
|
||
func (cdm *CrossDCMigrator) ListTasks() []*MigrationTask {
|
||
cdm.mu.RLock()
|
||
defer cdm.mu.RUnlock()
|
||
|
||
tasks := make([]*MigrationTask, 0, len(cdm.tasks))
|
||
for _, task := range cdm.tasks {
|
||
tasks = append(tasks, task)
|
||
}
|
||
|
||
// Сортируем по времени создания (новые сверху)
|
||
sort.Slice(tasks, func(i, j int) bool {
|
||
return tasks[i].CreatedAt > tasks[j].CreatedAt
|
||
})
|
||
|
||
return tasks
|
||
}
|
||
|
||
// PauseMigration приостанавливает миграцию
|
||
func (cdm *CrossDCMigrator) PauseMigration(taskID string) error {
|
||
cdm.mu.RLock()
|
||
task, ok := cdm.tasks[taskID]
|
||
cdm.mu.RUnlock()
|
||
|
||
if !ok {
|
||
return fmt.Errorf("task %s not found", taskID)
|
||
}
|
||
|
||
task.mu.Lock()
|
||
if task.Status != MigrationStatusMigrating && task.Status != MigrationStatusDeltaSync {
|
||
task.mu.Unlock()
|
||
return fmt.Errorf("task %s is not in a running state", taskID)
|
||
}
|
||
task.Status = MigrationStatusPaused
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Migration %s paused", taskID))
|
||
return nil
|
||
}
|
||
|
||
// ResumeMigration возобновляет миграцию
|
||
func (cdm *CrossDCMigrator) ResumeMigration(taskID string) error {
|
||
cdm.mu.RLock()
|
||
task, ok := cdm.tasks[taskID]
|
||
cdm.mu.RUnlock()
|
||
|
||
if !ok {
|
||
return fmt.Errorf("task %s not found", taskID)
|
||
}
|
||
|
||
task.mu.Lock()
|
||
if task.Status != MigrationStatusPaused {
|
||
task.mu.Unlock()
|
||
return fmt.Errorf("task %s is not paused", taskID)
|
||
}
|
||
task.Status = MigrationStatusMigrating
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Migration %s resumed", taskID))
|
||
cdm.wg.Add(1)
|
||
go cdm.runMigration(task)
|
||
|
||
return nil
|
||
}
|
||
|
||
// CancelMigration отменяет миграцию
|
||
func (cdm *CrossDCMigrator) CancelMigration(taskID string) error {
|
||
cdm.mu.RLock()
|
||
task, ok := cdm.tasks[taskID]
|
||
cdm.mu.RUnlock()
|
||
|
||
if !ok {
|
||
return fmt.Errorf("task %s not found", taskID)
|
||
}
|
||
|
||
task.mu.Lock()
|
||
task.Status = MigrationStatusFailed
|
||
task.Error = "cancelled by user"
|
||
task.EndTime = time.Now().UnixMilli()
|
||
task.UpdatedAt = time.Now().UnixMilli()
|
||
task.mu.Unlock()
|
||
|
||
cdm.logger.Info(fmt.Sprintf("Migration %s cancelled", taskID))
|
||
return nil
|
||
}
|
||
|
||
// GetMigrationStats возвращает статистику миграции
|
||
func (cdm *CrossDCMigrator) GetMigrationStats() *MigrationStats {
|
||
cdm.stats.mu.RLock()
|
||
defer cdm.stats.mu.RUnlock()
|
||
return &MigrationStats{
|
||
TotalChanges: cdm.stats.TotalChanges,
|
||
AppliedChanges: cdm.stats.AppliedChanges,
|
||
FailedChanges: cdm.stats.FailedChanges,
|
||
SkippedChanges: cdm.stats.SkippedChanges,
|
||
LatencyAvg: cdm.stats.LatencyAvg,
|
||
LatencyMax: cdm.stats.LatencyMax,
|
||
Throughput: cdm.stats.Throughput,
|
||
}
|
||
}
|
||
|
||
// SubscribeChanges подписывается на изменения
|
||
func (cdm *CrossDCMigrator) SubscribeChanges() <-chan *ChangeRecord {
|
||
ch := make(chan *ChangeRecord, 1000)
|
||
cdm.subscriberMu.Lock()
|
||
cdm.changeSubscribers = append(cdm.changeSubscribers, ch)
|
||
cdm.subscriberMu.Unlock()
|
||
return ch
|
||
}
|
||
|
||
// UnsubscribeChanges отписывается от изменений
|
||
func (cdm *CrossDCMigrator) UnsubscribeChanges(ch <-chan *ChangeRecord) {
|
||
cdm.subscriberMu.Lock()
|
||
defer cdm.subscriberMu.Unlock()
|
||
|
||
for i, sub := range cdm.changeSubscribers {
|
||
if sub == ch {
|
||
cdm.changeSubscribers = append(cdm.changeSubscribers[:i], cdm.changeSubscribers[i+1:]...)
|
||
close(sub)
|
||
break
|
||
}
|
||
}
|
||
}
|
||
|
||
// GetQueueStats возвращает статистику очереди
|
||
func (cdm *CrossDCMigrator) GetQueueStats() map[string]interface{} {
|
||
return map[string]interface{}{
|
||
"size": cdm.changeQueue.Size(),
|
||
"last_lsn": cdm.changeQueue.GetLastLSN(),
|
||
"max_size": 100000,
|
||
"file_path": cdm.changeQueue.filePath,
|
||
}
|
||
}
|
||
|
||
// GetConfig возвращает конфигурацию мигратора (публичный метод для доступа из REPL)
|
||
func (cdm *CrossDCMigrator) GetConfig() *config.MigrationConfig {
|
||
return cdm.config
|
||
}
|