103 lines
2.9 KiB
Go
103 lines
2.9 KiB
Go
package main
|
|
|
|
import (
|
|
"database/sql"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"syscall"
|
|
|
|
"git.ma-al.com/goc_marek/replica/replica"
|
|
_ "github.com/go-sql-driver/mysql"
|
|
)
|
|
|
|
func main() {
|
|
// Initialize the logger
|
|
replica.InitLogger()
|
|
|
|
// Load configuration from environment
|
|
cfg, err := replica.LoadEnvConfig()
|
|
if err != nil {
|
|
replica.Fatalf("Failed to load configuration: %v", err)
|
|
}
|
|
|
|
// Setup Graylog if enabled
|
|
if cfg.Graylog.Enabled {
|
|
replica.Infof("Graylog enabled: %s", cfg.Graylog.Endpoint)
|
|
if err := replica.SetupGlobalGraylog(replica.GraylogConfig{
|
|
Endpoint: cfg.Graylog.Endpoint,
|
|
Protocol: cfg.Graylog.Protocol,
|
|
Timeout: cfg.Graylog.Timeout,
|
|
Source: cfg.Graylog.Source,
|
|
ExtraFields: cfg.Graylog.ExtraFields,
|
|
}); err != nil {
|
|
replica.Warnf("Failed to setup Graylog: %v", err)
|
|
} else {
|
|
replica.Info("Graylog setup successful")
|
|
}
|
|
}
|
|
|
|
replica.Infof("Loaded configuration: %d secondary(ies)", len(cfg.Secondaries))
|
|
|
|
// Handle shutdown signals
|
|
sigChan := make(chan os.Signal, 1)
|
|
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
|
|
|
|
// Connect to primary MariaDB
|
|
primaryDSN := cfg.Primary.User + ":" + cfg.Primary.Password + "@tcp(" + cfg.Primary.Host + ":" +
|
|
strconv.FormatUint(uint64(cfg.Primary.Port), 10) + ")/?multiStatements=true"
|
|
primaryDB, err := sql.Open("mysql", primaryDSN)
|
|
if err != nil {
|
|
replica.Fatalf("Failed to connect to primary MariaDB: %v", err)
|
|
}
|
|
defer primaryDB.Close()
|
|
|
|
primaryDB.SetMaxOpenConns(25)
|
|
primaryDB.SetMaxIdleConns(5)
|
|
|
|
if err := primaryDB.Ping(); err != nil {
|
|
replica.Fatalf("Failed to ping primary MariaDB: %v", err)
|
|
}
|
|
replica.Info("Connected to primary MariaDB")
|
|
|
|
// Create multi-service manager
|
|
multiService := replica.NewMultiBinlogSyncService()
|
|
|
|
// Connect to each secondary and create services
|
|
for _, secCfg := range cfg.Secondaries {
|
|
replica.Infof("Connecting to secondary: %s (%s:%d)", secCfg.Name, secCfg.Host, secCfg.Port)
|
|
|
|
secondaryDB, err := sql.Open("mysql", secCfg.DSN)
|
|
if err != nil {
|
|
replica.Fatalf("Failed to connect to secondary %s: %v", secCfg.Name, err)
|
|
}
|
|
defer secondaryDB.Close()
|
|
|
|
secondaryDB.SetMaxOpenConns(25)
|
|
secondaryDB.SetMaxIdleConns(5)
|
|
|
|
if err := secondaryDB.Ping(); err != nil {
|
|
replica.Fatalf("Failed to ping secondary %s: %v", secCfg.Name, err)
|
|
}
|
|
replica.Infof("Connected to secondary: %s", secCfg.Name)
|
|
|
|
// Create service for this secondary
|
|
service := replica.NewBinlogSyncService(cfg.Primary, primaryDB, secondaryDB, secCfg.Name)
|
|
multiService.AddService(service)
|
|
}
|
|
|
|
// Start all services
|
|
replica.Info("Starting binlog replication...")
|
|
if err := multiService.StartAll(cfg.BatchSize, cfg.WorkerCount, cfg.ExcludeSchemas); err != nil {
|
|
replica.Fatalf("Failed to start all services: %v", err)
|
|
}
|
|
|
|
// Wait for shutdown signal
|
|
sig := <-sigChan
|
|
replica.Infof("Received %v, shutting down...", sig)
|
|
|
|
// Stop all services
|
|
multiService.StopAll()
|
|
replica.Info("Service stopped.")
|
|
}
|