97 lines
2.4 KiB
Go
97 lines
2.4 KiB
Go
package cdc
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/go-mysql-org/go-mysql/canal"
|
|
|
|
"dashboard-cdc-worker/internal/config"
|
|
"dashboard-cdc-worker/internal/dashboard"
|
|
)
|
|
|
|
type Worker struct {
|
|
cfg config.Config
|
|
store dashboard.Store
|
|
projector *dashboard.Projector
|
|
logger *slog.Logger
|
|
canal *canal.Canal
|
|
}
|
|
|
|
func NewWorker(cfg config.Config, store dashboard.Store, projector *dashboard.Projector, logger *slog.Logger) (*Worker, error) {
|
|
canalCfg := canal.NewDefaultConfig()
|
|
canalCfg.Addr = cfg.MySQL.CDCAddr
|
|
canalCfg.User = cfg.MySQL.CDCUser
|
|
canalCfg.Password = cfg.MySQL.CDCPassword
|
|
canalCfg.ServerID = cfg.CDC.ServerID
|
|
canalCfg.Flavor = "mysql"
|
|
canalCfg.ParseTime = true
|
|
canalCfg.UseDecimal = true
|
|
if location, err := time.LoadLocation(cfg.Dashboard.StorageTimezone); err == nil {
|
|
canalCfg.TimestampStringLocation = location
|
|
}
|
|
canalCfg.HeartbeatPeriod = cfg.CDC.HeartbeatEvery
|
|
canalCfg.ReadTimeout = 90 * time.Second
|
|
canalCfg.IncludeTableRegex = includeRegex(cfg.CDC.IncludeTables)
|
|
canalCfg.Dump.ExecutionPath = ""
|
|
canalCfg.DiscardNoMetaRowEvent = true
|
|
canalCfg.Logger = logger
|
|
|
|
instance, err := canal.NewCanal(canalCfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &Worker{
|
|
cfg: cfg,
|
|
store: store,
|
|
projector: projector,
|
|
logger: logger,
|
|
canal: instance,
|
|
}, nil
|
|
}
|
|
|
|
func (w *Worker) Run(ctx context.Context) error {
|
|
handler := newHandler(w.cfg, w.store, w.projector, w.logger)
|
|
w.canal.SetEventHandler(handler)
|
|
go func() {
|
|
<-ctx.Done()
|
|
w.canal.Close()
|
|
}()
|
|
|
|
if w.cfg.CDC.StartMode == "checkpoint" {
|
|
pos, ok, err := w.store.LoadCheckpoint(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ok && pos.Name != "" && pos.Pos > 0 {
|
|
handler.setCurrentFile(pos.Name)
|
|
w.logger.Info("starting cdc from checkpoint", "position", pos.String())
|
|
return w.canal.RunFrom(pos)
|
|
}
|
|
}
|
|
|
|
pos, err := w.canal.GetMasterPos()
|
|
if err != nil {
|
|
return fmt.Errorf("get master position: %w", err)
|
|
}
|
|
handler.setCurrentFile(pos.Name)
|
|
w.logger.Info("starting cdc from current master position", "position", pos.String())
|
|
return w.canal.RunFrom(pos)
|
|
}
|
|
|
|
func includeRegex(tables []string) []string {
|
|
out := make([]string, 0, len(tables))
|
|
for _, table := range tables {
|
|
parts := strings.Split(strings.TrimSpace(table), ".")
|
|
if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
|
|
continue
|
|
}
|
|
out = append(out, "^"+regexp.QuoteMeta(parts[0])+"\\."+regexp.QuoteMeta(parts[1])+"$")
|
|
}
|
|
return out
|
|
}
|