zkcoi/Meshray-Manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rebuild: split core into separate repo; move wg-go integration to internal/ctr/corebind

Browse Source 
- go.mod: require git.zkcoi.com/zkcoi/meshray/core, replace => ../Meshray
- internal/ctr/corebind: EnhancedBind + registry + adapter (wg-go integration isolated)
- cmd/mr-wg migrated from old core/cmd/meshray-core
- core/ removed; CHANGELOG updated; fix checkdb vet warning
This commit is contained in:
zkcoi
2026-07-15 16:08:13 +08:00
parent 15dab96872
commit 59e3059246
39 changed files with 2350 additions and 4361 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ctr/ctr.go
+135 -131
View File
@@ -5,8 +5,7 @@ import (
"strconv"
"sync"
"git.zkcoi.com/zkcoi/meshray/core"
"git.zkcoi.com/zkcoi/meshray/core/connect"
contract "git.zkcoi.com/zkcoi/meshray-contract"
"go.uber.org/zap"
)
@@ -20,8 +19,8 @@ type Ctr struct {
// WireGuard 管理器
wgManager *WGManager
// Core 实例（直接集成，无需 gRPC）
coreInst *core.Core
// 增强引擎实例（由 meshray-core 提供，按 networkID 索引）；原生模式为空
engines map[string]contract.EngineController
mu sync.RWMutex
}
@@ -38,14 +37,12 @@ func NewCtr(name string, networkID uint64, config *CtrConfig, logger *zap.Logger
networkID: networkID,
config: config,
logger: logger,
engines: make(map[string]contract.EngineController),
}
// 初始化 WireGuard 管理器
// 初始化 WireGuard 管理器（两种模式都需要）
ctr.wgManager = NewWGManager(logger)
// 初始化 Core 实例（直接集成，无需 gRPC）
ctr.coreInst = core.NewCore(logger)
return ctr, nil
}
@@ -58,9 +55,6 @@ func (c *Ctr) Start() error {
zap.String("name", c.name),
zap.Uint64("network_id", c.networkID))
// ✅ Watchdog 监控暂不实现（P3 阶段）
// go c.watchdog.Start()
return nil
}
@@ -72,10 +66,13 @@ func (c *Ctr) Stop() error {
c.logger.Info("停止 meshray-ctr",
zap.Uint64("network_id", c.networkID))
// 停止 Core 实例（直接调用方法）
if c.coreInst != nil {
// ✅ Close 方法已实现（会停止所有 Engine）
c.coreInst.Close()
// 停止所有增强引擎
for networkID, engineCtl := range c.engines {
if err := engineCtl.Stop(); err != nil {
c.logger.Warn("停止增强引擎失败",
zap.String("network_id", networkID), zap.Error(err))
}
delete(c.engines, networkID)
}
// 停止 WireGuard 管理
@@ -87,8 +84,8 @@ func (c *Ctr) Stop() error {
}
// CreateNetwork 创建网络（根据组网模式决定行为）
// - 原生模式：仅创建 WG 设备
// - 增强模式：创建 WG 设备 + 启动 Core 实例
// - 原生模式：仅创建 WG 设备（wireguard-go 默认 Bind 直连）
// - 增强模式：WG 设备使用 meshray-core 的自定义 conn.Bind 接管收发；并启动增强引擎控制面
func (c *Ctr) CreateNetwork(networkID uint64, subnet string, listenPort int, meshMode string) error {
c.mu.Lock()
defer c.mu.Unlock()
@@ -99,41 +96,30 @@ func (c *Ctr) CreateNetwork(networkID uint64, subnet string, listenPort int, mes
zap.Int("listen_port", listenPort),
zap.String("mesh_mode", meshMode))
// 1. 创建 WireGuard 设备（两种模式都需要）
// 1. 创建 WireGuard 设备（两种模式都需要；增强模式使用 core 注册的自定义 Bind）
networkIDStr := strconv.FormatUint(networkID, 10)
if err := c.wgManager.CreateDevice(networkIDStr, subnet, listenPort); err != nil {
if err := c.wgManager.CreateDevice(networkIDStr, subnet, listenPort, meshMode); err != nil {
return fmt.Errorf("创建 WireGuard 设备失败：%w", err)
}
c.logger.Debug("WireGuard 设备创建成功",
zap.Uint64("network_id", networkID))
// 2. 仅增强模式需要创建并启动 Core 实例
// 2. 仅增强模式需要启动增强引擎（控制面）
if meshMode == "enhanced" {
// 直接调用 Core 实例的方法（无需 gRPC）
metrics := core.NewMetrics()
engine, err := c.coreInst.CreateEngine(networkIDStr, metrics)
engineCtl, err := contract.NewEngineController(networkIDStr, c.logger)
if err != nil {
c.wgManager.DeleteDevice(networkIDStr) // 回滚 WG设备
c.logger.Error("创建 Engine 失败",
zap.Uint64("network_id", networkID),
zap.Error(err))
return fmt.Errorf("创建 Engine 失败：%w", err)
}
if err := engine.Start(); err != nil {
// ✅ 调用 Stop 方法停止（会清理所有资源）
engine.Stop()
c.wgManager.DeleteDevice(networkIDStr) // 回滚 WG 设备
c.logger.Error("启动 Engine 失败",
zap.Uint64("network_id", networkID),
zap.Error(err))
return fmt.Errorf("启动 Engine 失败：%w", err)
return fmt.Errorf("增强模式需 meshray-core：%w", err)
}
c.logger.Info("增强模式：Engine 实例已启动",
if err := engineCtl.Start(); err != nil {
c.wgManager.DeleteDevice(networkIDStr) // 回滚 WG 设备
return fmt.Errorf("启动增强引擎失败：%w", err)
}
c.engines[networkIDStr] = engineCtl
c.logger.Info("增强模式：引擎已启动（收发由 meshray-core conn.Bind 接管）",
zap.Uint64("network_id", networkID))
} else {
c.logger.Info("原生模式：仅创建 WG 设备，不启动 Core",
c.logger.Info("原生模式：仅创建 WG 设备，标准直连",
zap.Uint64("network_id", networkID))
}
@@ -152,11 +138,17 @@ func (c *Ctr) DeleteNetwork(networkID uint64) error {
c.logger.Info("开始删除网络",
zap.Uint64("network_id", networkID))
// 1. Core 实例不需要显式删除（它是内存中的对象）
// 如果需要清理，可以在未来实现 Engine.Stop() 方法
networkIDStr := strconv.FormatUint(networkID, 10)
// 2. 删除 WireGuard 设备
// 停止并清理增强引擎（若存在）
if engineCtl, ok := c.engines[networkIDStr]; ok {
if err := engineCtl.Stop(); err != nil {
c.logger.Warn("停止增强引擎失败", zap.String("network_id", networkIDStr), zap.Error(err))
}
delete(c.engines, networkIDStr)
}
// 删除 WireGuard 设备
if err := c.wgManager.DeleteDevice(networkIDStr); err != nil {
return fmt.Errorf("删除 WireGuard 设备失败：%w", err)
}
@@ -182,29 +174,13 @@ func (c *Ctr) AddPeer(networkID uint64, publicKey, allowedIP string) error {
return fmt.Errorf("添加 Peer 失败：%w", err)
}
// 2. 通知 Core
// 如果已经在 Enhanced 模式，需要接管刚刚加入的这个 Peer
engine, err := c.coreInst.GetEngine(networkIDStr)
if err == nil && engine != nil {
// 引擎存在，说明是 enhanced 模式
// 由 Core 分配一个本地端口
localPort, err := engine.Bind(publicKey, 0)
if err != nil {
c.logger.Error("增强模式动态 AddPeer Bind 失败", zap.Error(err))
} else {
// 通知 WG 更新为本地代理 Endpoint
newEndpoint := fmt.Sprintf("127.0.0.1:%d", localPort)
if err := c.wgManager.UpdatePeerEndpoint(networkIDStr, publicKey, newEndpoint); err != nil {
c.logger.Warn("增强模式自动重写 Endpoint 失败", zap.Error(err))
engine.Unbind(publicKey)
} else {
c.logger.Info("增强模式：动态 Peer 添加并成功代理给 Core",
zap.String("public_key", publicKey[:8]+"..."),
zap.String("new_endpoint", newEndpoint))
}
}
// 2. 增强模式：meshray-core 的 conn.Bind 已接管该组网收发，peer Endpoint 保持真实地址，
// 无需改写回环端口；候选地址后续由信令层经 NotifyPeerCandidates 注入触发建连。
if _, ok := c.engines[networkIDStr]; ok {
c.logger.Info("增强模式：动态 Peer 已交由 meshray-core 接管",
zap.String("public_key", publicKey[:8]+"..."))
} else {
c.logger.Debug("Core 未启动，普通模式无需代理 Peer", zap.Uint64("network_id", networkID))
c.logger.Debug("原生模式无需代理 Peer", zap.Uint64("network_id", networkID))
}
return nil
@@ -225,36 +201,38 @@ func (c *Ctr) RemovePeer(networkID uint64, publicKey string) error {
return fmt.Errorf("移除 Peer 失败：%w", err)
}
// 通知 Core 卸载这台 Peer 的监听
engine, err := c.coreInst.GetEngine(networkIDStr)
if err == nil && engine != nil {
engine.Unbind(publicKey)
c.logger.Info("增强模式：Core 已解除对该 Peer 的挂载", zap.String("public_key", publicKey[:8]+"..."))
// 增强模式：meshray-core 的 conn.Bind 按 WG peer 状态自动管理收发，
// 移除 WG peer 后不再有该 peer 的包，无需显式解绑
if _, ok := c.engines[networkIDStr]; ok {
c.logger.Info("增强模式：Core 已随 WG Peer 移除停止对该 Peer 的转发",
zap.String("public_key", publicKey[:8]+"..."))
}
return nil
}
// SetSTUNTURNConfig 为指定网络设置 STUN/TURN 配置
func (c *Ctr) SetSTUNTURNConfig(networkID uint64, stunServers []string, turnServers []connect.TURNServerConfig) error {
// SetSTUNTURNConfig 为指定网络设置 STUN/TURN 配置（仅增强模式生效）
func (c *Ctr) SetSTUNTURNConfig(networkID uint64, stunServers []string, turnServers []string) error {
c.mu.RLock()
defer c.mu.RUnlock()
networkIDStr := strconv.FormatUint(networkID, 10)
// 获取 Engine 实例
engine, err := c.coreInst.GetEngine(networkIDStr)
if err != nil {
// 获取增强引擎
engineCtl, ok := c.engines[networkIDStr]
if !ok {
c.logger.Debug("网络未启动增强模式，跳过 STUN/TURN 配置",
zap.Uint64("network_id", networkID))
return nil // 无需错误，因为原生模式不需要
}
// 更新 WebRTC 工厂的 ICE 配置
engine.SetICEConfig(connect.ICEConfig{
if err := engineCtl.SetICEConfig(contract.ICEConfig{
STUNServers: stunServers,
TURNServers: turnServers,
})
}); err != nil {
return fmt.Errorf("设置 ICE 配置失败：%w", err)
}
c.logger.Info("STUN/TURN 配置已设置",
zap.Uint64("network_id", networkID),
@@ -263,6 +241,7 @@ func (c *Ctr) SetSTUNTURNConfig(networkID uint64, stunServers []string, turnServ
return nil
}
func (c *Ctr) GetStatus(networkID uint64) (*NetworkStatus, error) {
c.mu.RLock()
defer c.mu.RUnlock()
@@ -279,11 +258,9 @@ func (c *Ctr) GetStatus(networkID uint64) (*NetworkStatus, error) {
}
status.WGStatus = wgStatus
// 获取 Core 状态（直接调用 Engine 的方法）
engine, err := c.coreInst.GetEngine(networkIDStr)
if err == nil && engine != nil {
coreStatus, err := engine.GetStatus()
if err == nil {
// 获取增强引擎状态
if engineCtl, ok := c.engines[networkIDStr]; ok {
if coreStatus, err := engineCtl.GetStatus(); err == nil {
status.CoreStatus = coreStatus
}
}
@@ -292,7 +269,8 @@ func (c *Ctr) GetStatus(networkID uint64) (*NetworkStatus, error) {
}
// SwitchMode 切换传输模式（原生→增强）
// 仅支持 "native" → "enhanced" 单向切换
// 仅支持 "native" → "enhanced" 单向切换。增强模式由 meshray-core 的 conn.Bind 接管收发，
// 不再改写 peer Endpoint 为回环端口（根治入站接不通的缺陷）。
func (c *Ctr) SwitchMode(networkID uint64, mode string) error {
c.mu.Lock()
defer c.mu.Unlock()
@@ -317,18 +295,17 @@ func (c *Ctr) SwitchMode(networkID uint64, mode string) error {
c.logger.Info("获取到 WG 状态",
zap.Int("peer_count", wgStatus.PeerCount))
// 3. 创建并启动 Core Engine（增强模式核心）
metrics := core.NewMetrics()
engine, err := c.coreInst.CreateEngine(networkIDStr, metrics)
// 3. 创建并启动增强引擎（控制面）
engineCtl, err := contract.NewEngineController(networkIDStr, c.logger)
if err != nil {
return fmt.Errorf("创建 Core Engine 失败：%w", err)
return fmt.Errorf("创建增强引擎失败（需 meshray-core）：%w", err)
}
if err := engine.Start(); err != nil {
return fmt.Errorf("启动 Core Engine 失败：%w", err)
if err := engineCtl.Start(); err != nil {
return fmt.Errorf("启动增强引擎失败：%w", err)
}
c.engines[networkIDStr] = engineCtl
c.logger.Info("Core Engine 已启动",
c.logger.Info("增强引擎已启动（收发由 meshray-core conn.Bind 接管）",
zap.Uint64("network_id", networkID))
// 4. 获取 Peer 列表
@@ -337,60 +314,87 @@ func (c *Ctr) SwitchMode(networkID uint64, mode string) error {
return fmt.Errorf("获取 Peer 列表失败：%w", err)
}
// 5. 修改 Peer Endpoint → 127.0.0.1:{port}，并且让 Core 绑定
// 增强模式下，所有 Peer 的 Endpoint 都指向本地回环地址
// 5. 把已知真实 Endpoint 作为候选注入，触发 P2P 建连
// （不再改写 peer Endpoint 为回环端口；peer Endpoint 保持真实地址，WG 入站匹配成功）
for _, peer := range peers {
// 1. Core 引擎动态接管，传 0 意为让操作系统挑个未占用的随意回环端口
localPort, err := engine.Bind(peer.PublicKey, 0)
if err != nil {
c.logger.Error("Engine Bind 失败",
zap.String("public_key", peer.PublicKey[:8]+"..."),
zap.Error(err))
continue
if peer.Endpoint != "" {
candidates := []contract.Candidate{
{Addr: peer.Endpoint, Type: "host", Priority: 1, Protocol: "udp"},
}
if err := engineCtl.NotifyPeerCandidates(peer.PublicKey, candidates); err != nil {
c.logger.Warn("注入对端候选地址失败",
zap.String("public_key", peer.PublicKey[:8]+"..."),
zap.Error(err))
}
}
newEndpoint := fmt.Sprintf("127.0.0.1:%d", localPort)
// 2. 将此端口告知 WG 更新 Endpoint
if err := c.wgManager.UpdatePeerEndpoint(networkIDStr, peer.PublicKey, newEndpoint); err != nil {
c.logger.Warn("更新 Peer Endpoint 失败",
zap.String("public_key", peer.PublicKey[:8]+"..."),
zap.Error(err))
// 如果更新失败，解绑 Core
engine.Unbind(peer.PublicKey)
continue
}
c.logger.Info("Peer Endpoint 已成功切换并接入 Core",
zap.String("public_key", peer.PublicKey[:8]+"..."),
zap.String("original_endpoint", peer.Endpoint),
zap.String("new_endpoint", newEndpoint))
}
c.logger.Info("传输模式切换成功（基础功能已实现，P3 阶段待完善）",
c.logger.Info("传输模式切换成功（增强模式：meshray-core 接管收发）",
zap.Uint64("network_id", networkID),
zap.String("mode", mode))
return nil
}
// CoreConfigUpdate 核心配置更新请求（由 server 层在策略变更时构造并传入）
type CoreConfigUpdate struct {
STUNServers []string // STUN 服务器列表
TURNServers []string // TURN 服务器列表（URL）
EnabledLayers []contract.Layer // 启用的传输层顺序（为空保持默认 9 层）
}
// NotifyPeerCandidates 由信令层调用：注入对端候选地址并触发 P2P 建连
func (c *Ctr) NotifyPeerCandidates(networkID uint64, publicKey string, candidates []contract.Candidate) error {
c.mu.RLock()
defer c.mu.RUnlock()
networkIDStr := strconv.FormatUint(networkID, 10)
engineCtl, ok := c.engines[networkIDStr]
if !ok {
return fmt.Errorf("网络未启用增强模式")
}
return engineCtl.NotifyPeerCandidates(publicKey, candidates)
}
// UpdateCoreConfig 更新 Core 配置（策略修改后同步）
// ✅ P3-1 阶段实现
// 真实生效：更新 ICE/STUN/TURN 配置到 WebRTC 工厂，并按需调整传输层优先级顺序
func (c *Ctr) UpdateCoreConfig(networkID uint64, config interface{}) error {
c.mu.RLock()
defer c.mu.RUnlock()
c.logger.Info("更新 Core 配置",
zap.Uint64("network_id", networkID))
update, ok := config.(*CoreConfigUpdate)
if !ok {
return fmt.Errorf("不支持的配置类型：%T", config)
}
// ✅ P3-1 阶段 - 暂不实现
// 未来实现：
// 1. 根据策略类型解析配置
// 2. 调用 Core 的 UpdateConfig 方法
// 3. 持久化到数据库
networkIDStr := strconv.FormatUint(networkID, 10)
engineCtl, ok := c.engines[networkIDStr]
if !ok {
return fmt.Errorf("网络未启用增强模式")
}
return fmt.Errorf("UpdateCoreConfig 尚未实现（P3-1 阶段）")
// 1. 更新 ICE/STUN/TURN 配置（下发到 WebRTC 工厂）
if err := engineCtl.SetICEConfig(contract.ICEConfig{
STUNServers: update.STUNServers,
TURNServers: update.TURNServers,
}); err != nil {
return fmt.Errorf("设置 ICE 配置失败：%w", err)
}
// 2. 更新传输层优先级顺序（策略核心配置）
if len(update.EnabledLayers) > 0 {
if err := engineCtl.SetLayerOrder(update.EnabledLayers); err != nil {
return fmt.Errorf("设置传输层顺序失败：%w", err)
}
}
c.logger.Info("Core 配置已更新",
zap.Uint64("network_id", networkID),
zap.Int("stun_count", len(update.STUNServers)),
zap.Int("turn_count", len(update.TURNServers)),
zap.Int("enabled_layers", len(update.EnabledLayers)))
return nil
}
// GetWGMode 获取系统 WG 模式
@@ -400,7 +404,7 @@ func (c *Ctr) GetWGMode() string {
// NetworkStatus 网络状态
type NetworkStatus struct {
NetworkID string `json:"network_id"`
WGStatus *WGStatus `json:"wg_status"`
CoreStatus *core.EngineStatus `json:"core_status,omitempty"`
NetworkID string `json:"network_id"`
WGStatus *WGStatus `json:"wg_status"`
CoreStatus *contract.EngineStatus `json:"core_status,omitempty"`
}