// Command vrtxsqlweb 给 migrate 库套一个本地 Web 界面。 // // 和 cmd/vrtxsql 是同一套能力, 只是把 flag 换成浏览器里的表单: // 填两个 DSN、看一眼源库结构、勾选要迁的表、点一下, 迁移过程以日志流实时回显, // 结束给出每张表的报告 (行数 / 列数 / 索引数 / 耗时)。 // // go run ./cmd/vrtxsqlweb -addr :8099 // // 实现只用了标准库 + 一个内嵌的 index.html, 没有前端工具链, 也没有外部依赖: // 它把 migrate 这个库直接暴露给 HTTP, 不复制任何迁移逻辑。 package main import ( _ "embed" "encoding/json" "flag" "fmt" "log" "net/http" "os" "path/filepath" "runtime" "cmp" "slices" "strings" "time" "git.ycl.pub/yl32940/TsDB2/migrate" ) // index.html 是这份单页界面, 编译期嵌进二进制 —— 运行时不需要任何静态文件。 // //go:embed index.html var indexHTML []byte func main() { // 默认只听回环: 这是个本地工具, 而且 /api/browse 能列本机任意目录。 addr := flag.String("addr", "127.0.0.1:8099", "监听地址, 如 127.0.0.1:8099") flag.Parse() mux := http.NewServeMux() mux.HandleFunc("/", handleIndex) mux.HandleFunc("/api/browse", handleBrowse) mux.HandleFunc("/api/stat", handleStat) mux.HandleFunc("/api/list", handleList) mux.HandleFunc("/api/databases", handleDatabases) mux.HandleFunc("/api/migrate", handleMigrate) url := "http://" + *addr if strings.HasPrefix(*addr, ":") { url = "http://localhost" + *addr } log.Printf("vrtxsqlweb 已启动: %s", url) log.Printf("提示: 服务运行的当前目录就是相对路径的基准目录") // 迁移大表可能跑很久, 这里刻意不设 WriteTimeout。 if err := http.ListenAndServe(*addr, mux); err != nil { log.Fatal(err) } } func handleIndex(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/" { http.NotFound(w, r) return } w.Header().Set("Content-Type", "text/html; charset=utf-8") _, _ = w.Write(indexHTML) } // ============================ /api/browse ============================ // browseReq.Path 用指针是为了区分「字段没传」和「传了空串」: // // nil → 从服务进程的当前目录起步 (前端首次打开弹窗) // "" → 虚拟根 (Windows 列出所有盘符; Unix 就是 /) // 其他 → 该目录; 若给的是文件, 自动退到它所在目录 type browseReq struct { Path *string `json:"path"` } type browseEntry struct { Name string `json:"name"` Path string `json:"path"` IsDir bool `json:"isDir"` Size int64 `json:"size"` Ext string `json:"ext"` } type browseResp struct { Path string `json:"path"` Parent string `json:"parent"` Root bool `json:"root"` // 已到顶 (此电脑 / 文件系统根) Dirs []browseEntry `json:"dirs"` Files []browseEntry `json:"files"` Error string `json:"error,omitempty"` } // handleBrowse 给前端的「浏览…」弹窗列目录。 // 这是本地工具且默认只监听回环地址, 所以不做沙箱限制。 func handleBrowse(w http.ResponseWriter, r *http.Request) { var req browseReq if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeJSON(w, browseResp{Error: "请求体解析失败: " + err.Error()}) return } var dir string switch { case req.Path == nil: wd, err := os.Getwd() if err != nil { writeJSON(w, browseResp{Error: "取当前目录失败: " + err.Error()}) return } dir = wd case *req.Path == "": if runtime.GOOS == "windows" { writeJSON(w, browseResp{Dirs: windowsDrives(), Root: true}) return } dir = string(filepath.Separator) default: dir = *req.Path } abs, err := filepath.Abs(dir) if err != nil { writeJSON(w, browseResp{Error: "路径非法: " + err.Error()}) return } if st, err := os.Stat(abs); err == nil && !st.IsDir() { abs = filepath.Dir(abs) } resp := browseResp{Path: abs, Dirs: []browseEntry{}, Files: []browseEntry{}} switch parent := filepath.Dir(abs); { case parent == abs && runtime.GOOS == "windows": resp.Parent = "" // 盘根再往上一级 = 此电脑 case parent == abs: resp.Root = true // Unix 的 / default: resp.Parent = parent } ents, err := os.ReadDir(abs) if err != nil { resp.Error = "读取目录失败: " + err.Error() writeJSON(w, resp) return } for _, e := range ents { name := e.Name() if strings.HasPrefix(name, ".") || strings.HasPrefix(name, "$") { continue // 隐藏项和系统目录不展示 } full := filepath.Join(abs, name) if e.IsDir() { resp.Dirs = append(resp.Dirs, browseEntry{Name: name, Path: full, IsDir: true}) continue } var size int64 if info, err := e.Info(); err == nil { size = info.Size() } resp.Files = append(resp.Files, browseEntry{ Name: name, Path: full, Size: size, Ext: strings.ToLower(filepath.Ext(name)), }) } slices.SortFunc(resp.Dirs, func(a, b browseEntry) int { return cmp.Compare(a.Name, b.Name) }) slices.SortFunc(resp.Files, func(a, b browseEntry) int { return cmp.Compare(a.Name, b.Name) }) writeJSON(w, resp) } func windowsDrives() []browseEntry { out := make([]browseEntry, 0, 8) for c := 'A'; c <= 'Z'; c++ { root := string(c) + ":\\" if st, err := os.Stat(root); err == nil && st.IsDir() { out = append(out, browseEntry{Name: root, Path: root, IsDir: true}) } } return out } // ============================ /api/stat ============================ type statReq struct { Path string `json:"path"` } type statResp struct { Path string `json:"path"` Exists bool `json:"exists"` IsDir bool `json:"isDir"` Size int64 `json:"size"` Parent string `json:"parent"` ParentExists bool `json:"parentExists"` Error string `json:"error,omitempty"` } // handleStat 让前端能显示「文件已存在 / 不存在(将新建) / 父目录不存在」。 // 目标库一般是新建的, 所以要能区分「还没这个文件」和「路径写错了」。 func handleStat(w http.ResponseWriter, r *http.Request) { var req statReq if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeJSON(w, statResp{Error: "请求体解析失败: " + err.Error()}) return } p := strings.TrimSpace(req.Path) if p == "" { writeJSON(w, statResp{}) return } abs, err := filepath.Abs(p) if err != nil { writeJSON(w, statResp{Error: "路径非法: " + err.Error()}) return } resp := statResp{Path: abs, Parent: filepath.Dir(abs)} if st, err := os.Stat(abs); err == nil { resp.Exists = true resp.IsDir = st.IsDir() resp.Size = st.Size() } if st, err := os.Stat(resp.Parent); err == nil && st.IsDir() { resp.ParentExists = true } writeJSON(w, resp) } // ============================ /api/list ============================ type listReq struct { DSN string `json:"dsn"` Tables []string `json:"tables"` } type colInfo struct { Name string `json:"name"` Type string `json:"type"` Size int `json:"size"` Dynamic bool `json:"dynamic"` IsPK bool `json:"isPK"` Synthetic bool `json:"synthetic"` } type idxInfo struct { Name string `json:"name"` Type string `json:"type"` Columns []string `json:"columns"` RowStore bool `json:"rowStore"` } type tableInfo struct { Name string `json:"name"` Columns []colInfo `json:"columns"` PKColumn string `json:"pkColumn"` PKIndexType string `json:"pkIndexType"` RowStore []string `json:"rowStore"` Indexes []idxInfo `json:"indexes"` Rows uint64 `json:"rows"` } type listResp struct { Kind string `json:"kind"` Identity string `json:"identity"` Tables []tableInfo `json:"tables"` Error string `json:"error,omitempty"` } type dbListReq struct { Host string `json:"host"` Port string `json:"port"` User string `json:"user"` Pass string `json:"pass"` } type dbListResp struct { Databases []string `json:"databases,omitempty"` Error string `json:"error,omitempty"` } // sysSchemas MySQL 自带的库, 迁它们没有意义, 从候选里去掉。 var sysSchemas = map[string]bool{ "information_schema": true, "performance_schema": true, "mysql": true, "sys": true, } // handleDatabases 只回答"这台 MySQL 上有哪些库", 给连接表单把库名做成候选。 // 库名不必先知道 —— 能不能列出库只取决于账号权限。 func handleDatabases(w http.ResponseWriter, r *http.Request) { var req dbListReq if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeJSON(w, dbListResp{Error: "请求体解析失败: " + err.Error()}) return } user := strings.TrimSpace(req.User) if user == "" { writeJSON(w, dbListResp{Error: "请先填写用户名"}) return } host := strings.TrimSpace(req.Host) if host == "" { host = "127.0.0.1" } port := strings.TrimSpace(req.Port) if port == "" { port = "3306" } // 库名留空: / 后面只挂连接参数, 顺序不能反过来 dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/?charset=utf8mb4", user, req.Pass, host, port) names, err := migrate.MySQLDatabases(dsn) if err != nil { writeJSON(w, dbListResp{Error: friendlyDBErr(err, host, port)}) return } keep := make([]string, 0, len(names)) for _, n := range names { if !sysSchemas[n] { keep = append(keep, n) } } writeJSON(w, dbListResp{Databases: keep}) } // friendlyDBErr 把驱动的报错翻成"接下来该查什么"。 // 密码错、权限不足这类原文已经够清楚, 原样带上; 麻烦的是连不上 —— // 那是一句英文 TCP 报错, 看着不知道从哪儿下手。 func friendlyDBErr(err error, host, port string) string { msg := err.Error() switch { case strings.Contains(msg, "refused"), strings.Contains(msg, "connectex"), strings.Contains(msg, "no such host"), strings.Contains(msg, "i/o timeout"), strings.Contains(msg, "deadline exceeded"): return fmt.Sprintf("连不上 %s:%s —— 检查 MySQL 是否在运行、主机名和端口是否写对 (%v)", host, port, err) case strings.Contains(msg, "Access denied"): return "账号或密码不对:" + msg } return msg } // handleList 只读源库结构, 不碰目标库 —— 对应 CLI 的 -list。 func handleList(w http.ResponseWriter, r *http.Request) { var req listReq if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeJSON(w, listResp{Error: "请求体解析失败: " + err.Error()}) return } if req.DSN == "" { writeJSON(w, listResp{Error: "请先填写源库 DSN"}) return } ep, err := migrate.ParseDSN(req.DSN, migrate.DefaultOptions()) if err != nil { writeJSON(w, listResp{Error: err.Error()}) return } defer ep.Close() resp := listResp{Kind: string(ep.Kind), Identity: ep.Identity()} names, err := ep.Src.Tables() if err != nil { resp.Error = "列出表失败: " + err.Error() writeJSON(w, resp) return } want := map[string]bool{} for _, t := range req.Tables { want[t] = true } for _, name := range names { if name == migrate.MetaTable { continue // 保真表是迁移的随身行李, 不当作业务表展示 } if len(want) > 0 && !want[name] { continue } sc, err := ep.Src.Schema(name) if err != nil { resp.Tables = append(resp.Tables, tableInfo{ Name: name, PKIndexType: "读结构失败: " + err.Error(), }) continue } resp.Tables = append(resp.Tables, toTableInfo(sc)) } writeJSON(w, resp) } func toTableInfo(sc *migrate.TableSchema) tableInfo { ti := tableInfo{ Name: sc.Name, PKColumn: sc.PKColumn, PKIndexType: migrate.IndexTypeName(sc.PKIndexType), RowStore: sc.RowStore(), Rows: sc.Rows, Columns: make([]colInfo, 0, len(sc.Columns)), } for _, c := range sc.Columns { ti.Columns = append(ti.Columns, colInfo{ Name: c.Name, Type: migrate.TypeName(c.Type), Size: c.Size(), Dynamic: c.IsDynamic(), IsPK: c.Name == sc.PKColumn, Synthetic: sc.IsSynthetic(c.Name), }) } for _, idx := range sc.Indexes { ti.Indexes = append(ti.Indexes, idxInfo{ Name: idx.Name, Type: migrate.IndexTypeName(idx.Type), Columns: idx.Columns, RowStore: idx.IsRowStore(), }) } return ti } // ============================ /api/migrate ============================ type migrateReq struct { From string `json:"from"` To string `json:"to"` Tables []string `json:"tables"` Recreate bool `json:"recreate"` Batch int `json:"batch"` Limit uint64 `json:"limit"` DryRun bool `json:"dryRun"` NoMeta bool `json:"noMeta"` AutoPK bool `json:"autoPk"` Loose bool `json:"loose"` // TruncOversize 值超过引擎单值上限 (16360 字节) 时截断并逐行报出 —— // 与 CLI 的 -truncate-oversize 同一个开关, 明确的数据丢失, 默认关。 TruncOversize bool `json:"truncateOversize"` // KeepGoing 单张表失败时跳过它继续迁其余表 (前端默认勾上)。 KeepGoing bool `json:"keepGoing"` // EachRow 每读一行数据就推一条事件, 界面能看着数据一行行滚过去。 // 开销随行数线性上涨 (百万行 = 百万条事件), 默认关。 EachRow bool `json:"eachRow"` } // maxBatchRows 批量接口一次请求允许的最大批大小。batch 直接控制 migrate 包 // make([]Row, 0, Batch) 的预分配 (见 migrate/run.go), 不设上限一次请求就能 // 预分配出超大切片把服务打 OOM。 const maxBatchRows = 50000 // streamEvent 迁移过程以 NDJSON 一行一条推送 (前端按行解析): // // {"type":"log", "at":"15:04:05","msg":"...","level":"stage"} // {"type":"progress", "table":"orders","rows":12000,"elapsed":"1.2s"} // {"type":"row", "table":"orders","rows":12001,"msg":"id=1, name=..."} // {"type":"done", "report":{...}} // {"type":"error", "msg":"..."} type streamEvent struct { Type string `json:"type"` Msg string `json:"msg,omitempty"` At string `json:"at,omitempty"` Level string `json:"level,omitempty"` // 空=普通; "stage"=阶段分隔行 Table string `json:"table,omitempty"` Rows uint64 `json:"rows,omitempty"` Elapsed string `json:"elapsed,omitempty"` Report *reportInfo `json:"report,omitempty"` } type reportInfo struct { From string `json:"from"` To string `json:"to"` Rows uint64 `json:"rows"` Elapsed string `json:"elapsed"` Tables []tableReport `json:"tables"` } type tableReport struct { Table string `json:"table"` Rows uint64 `json:"rows"` Columns int `json:"columns"` Indexes int `json:"indexes"` Elapsed string `json:"elapsed"` Skipped bool `json:"skipped"` Note string `json:"note"` Error string `json:"error"` } func handleMigrate(w http.ResponseWriter, r *http.Request) { var req migrateReq if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "请求体解析失败: "+err.Error(), http.StatusBadRequest) return } if req.Batch > maxBatchRows { http.Error(w, fmt.Sprintf("batch 超过上限 %d", maxBatchRows), http.StatusBadRequest) return } flusher, _ := w.(http.Flusher) w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("X-Accel-Buffering", "no") // 走反代时也别缓冲 w.WriteHeader(http.StatusOK) enc := json.NewEncoder(w) emit := func(ev streamEvent) { _ = enc.Encode(ev) if flusher != nil { flusher.Flush() } } now := func() string { return time.Now().Format("15:04:05") } if req.From == "" || req.To == "" { emit(streamEvent{Type: "error", Msg: "请同时填写源库与目标库 DSN"}) return } if req.From == req.To { emit(streamEvent{Type: "error", Msg: "源库与目标库不能是同一个"}) return } opts := migrate.DefaultOptions() opts.Create = true opts.Recreate = req.Recreate opts.EmbedMeta = !req.NoMeta opts.StrictValue = !req.Loose opts.TruncateOversize = req.TruncOversize opts.AutoPK = req.AutoPK opts.ContinueOnError = req.KeepGoing opts.DryRun = req.DryRun opts.Limit = req.Limit opts.Tables = req.Tables if req.Batch > 0 { opts.Batch = req.Batch } // 把库的逐表进度接到 HTTP 流上: Run 内部每完成一张表就回调这里。 opts.Verbose = true opts.Log = func(format string, args ...any) { emit(streamEvent{Type: "log", At: now(), Msg: fmt.Sprintf(format, args...)}) } // 大表"已写 N 行"走结构化事件: 前端原地刷一行, 不往日志里灌流水 // (灌进去会把日志顶到看不见, 滚动条也就一直在跳)。 opts.Progress = func(table string, rows uint64, elapsed time.Duration) { emit(streamEvent{Type: "progress", Table: table, Rows: rows, Elapsed: elapsed.Round(time.Millisecond).String()}) } // 逐行事件 (可选): 让人眼看着数据一行行滚过去。日志区有行数上限, 不会无限长。 if req.EachRow { opts.EachRow = func(table string, n uint64, row migrate.Row) { emit(streamEvent{Type: "row", Table: table, Rows: n, Msg: formatRowValues(row)}) } } src, err := migrate.ParseDSN(req.From, opts) if err != nil { emit(streamEvent{Type: "error", Msg: "打开源库: " + err.Error()}) return } defer src.Close() dst, err := migrate.ParseDSN(req.To, opts) if err != nil { emit(streamEvent{Type: "error", Msg: "打开目标库: " + err.Error()}) return } defer dst.Close() emit(streamEvent{Type: "log", At: now(), Msg: fmt.Sprintf("开始: %s → %s", src.Identity(), dst.Identity())}) // 两阶段: 先把所有表的结构立起来, 再灌数据。结构阶段是秒级的 —— 点完按钮 // 很快就能看到 52 张表全部就位, 不用盯着一张表猜是卡住还是死锁。 var rep1 *migrate.Report if !req.DryRun { schemaOpts := *opts schemaOpts.SchemaOnly = true // 结构阶段尽量往前推: 一张表建不出来不该挡住后面几十张, 明细留给数据阶段报。 schemaOpts.ContinueOnError = true emit(streamEvent{Type: "log", At: now(), Level: "stage", Msg: "阶段 1/2: 同步表结构 (建表 + 建索引, 不搬数据)"}) rep1, _ = migrate.Run(src.Src, dst.Dst, &schemaOpts) if rep1 != nil { ok, bad := 0, 0 for _, tr := range rep1.Tables { if tr.Error != "" { bad++ } else { ok++ } } msg := fmt.Sprintf("阶段 1/2 完成: %d 张表结构就绪", ok) if bad > 0 { msg += fmt.Sprintf(", %d 张建表失败 (具体错误在数据阶段报)", bad) } emit(streamEvent{Type: "log", At: now(), Msg: msg}) } } dataOpts := *opts // 表已经在阶段 1 建好了, 这里再 Recreate 一次纯属白干 (还会把刚建的表删掉)。 dataOpts.Recreate = false if !req.DryRun { emit(streamEvent{Type: "log", At: now(), Level: "stage", Msg: "阶段 2/2: 迁移数据"}) } rep, runErr := migrate.Run(src.Src, dst.Dst, &dataOpts) if rep != nil { mergeSchemaNotes(rep, rep1) emit(streamEvent{Type: "done", Report: toReportInfo(rep)}) } if runErr != nil { emit(streamEvent{Type: "error", Msg: runErr.Error()}) } } // mergeSchemaNotes 把阶段 1 才知道的事补进数据阶段的报告。 // // 为什么需要: 数据阶段面对的已经是建好的表, CreateTable 直接返回, 于是"哪个索引 // 被跳过"这类只发生在建表时刻的信息在阶段 2 的报告里是空的 —— 不补上, 用户看 // 报告会以为索引都建全了。 func mergeSchemaNotes(rep, schemaRep *migrate.Report) { if rep == nil || schemaRep == nil { return } notes := make(map[string]string, len(schemaRep.Tables)) for _, tr := range schemaRep.Tables { n := strings.TrimSuffix(strings.TrimSuffix(tr.Note, "仅结构"), "; ") if n != "" { notes[tr.Table] = n } } for i := range rep.Tables { n, ok := notes[rep.Tables[i].Table] if !ok { continue } if rep.Tables[i].Note == "" { rep.Tables[i].Note = n } else { rep.Tables[i].Note += "; " + n } } } // formatRowValues 把一行数据拼成 "值, 值, ..." 供逐行滚动显示。 // 截断长值: 这些行是给人看"数据在动"的, 不是给人读全文的。 func formatRowValues(row migrate.Row) string { var b strings.Builder for i, v := range row.Values { if i > 0 { b.WriteString(", ") } s := fmt.Sprintf("%v", v) if r := []rune(s); len(r) > 32 { s = string(r[:32]) + "…" } b.WriteString(s) } s := b.String() if r := []rune(s); len(r) > 220 { s = string(r[:220]) + "…" } return s } func toReportInfo(rep *migrate.Report) *reportInfo { ri := &reportInfo{ From: string(rep.From), To: string(rep.To), Rows: rep.Rows, Elapsed: rep.Elapsed.Round(time.Millisecond).String(), Tables: make([]tableReport, 0, len(rep.Tables)), } for _, tr := range rep.Tables { ri.Tables = append(ri.Tables, tableReport{ Table: tr.Table, Rows: tr.Rows, Columns: tr.Columns, Indexes: tr.Indexes, Elapsed: tr.Elapsed.Round(time.Millisecond).String(), Skipped: tr.Skipped, Note: tr.Note, Error: tr.Error, }) } return ri } func writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json; charset=utf-8") _ = json.NewEncoder(w).Encode(v) }