Compare commits
11 Commits
89adb4f8a4
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| e17af95f09 | |||
| 539477de4b | |||
| 54c2903c75 | |||
| 2c818576ef | |||
| 923ca32b98 | |||
| d70bdaae16 | |||
| b71c41424b | |||
| 140235d069 | |||
| 1900f5e82d | |||
| 7e7d706f84 | |||
| e2c87cfa8b |
5
.gitignore
vendored
5
.gitignore
vendored
@@ -27,4 +27,7 @@ go.work.sum
|
||||
.env
|
||||
|
||||
# editor file
|
||||
.idea/
|
||||
.idea/
|
||||
|
||||
cmd/
|
||||
.vscode/
|
||||
13
.zed/debug.json
Normal file
13
.zed/debug.json
Normal file
@@ -0,0 +1,13 @@
|
||||
// Project-local debug tasks
|
||||
//
|
||||
// For more documentation on how to configure debug tasks,
|
||||
// see: https://zed.dev/docs/debugger
|
||||
[
|
||||
{
|
||||
"label": "dev",
|
||||
"adapter": "Delve",
|
||||
"request": "launch",
|
||||
"program": "./cmd/web/main.go",
|
||||
"mode": "debug",
|
||||
},
|
||||
]
|
||||
60
README.md
60
README.md
@@ -1,4 +1,62 @@
|
||||
## TODO
|
||||
|
||||
此实现目前并不是完全并发安全的:
|
||||
- 目前事务等级没有对 cityhash 表的 offset 字段做防丢失,并发操作可能会出问题
|
||||
|
||||
- 目前事务等级没有对 cityhash 表的 offset 字段做防丢失,并发操作可能会出问题
|
||||
|
||||
考虑把命令行接口改为 web 接口
|
||||
|
||||
### 统一节点调度
|
||||
|
||||
节点上下线:
|
||||
|
||||
提供一个接口用来为节点加解锁
|
||||
|
||||
## 目录结构
|
||||
|
||||
整体是一个 go 项目,编译后在服务器执行
|
||||
|
||||
```
|
||||
docs/ 相关文档
|
||||
|
||||
actions/ 程序功能
|
||||
clients/ 外部 api 调用
|
||||
model/ 数据库模型
|
||||
util/ 工具类
|
||||
|
||||
scripts/ shell 脚本
|
||||
```
|
||||
|
||||
## 使用方式
|
||||
|
||||
此项目在本地运行,其编译产物放在服务器运行
|
||||
|
||||
1. 在项目根目录下运行 `./deploy.ps1`,或直接复制脚本中的命令,编译出执行程序并上传到服务器
|
||||
2. (可选)可以把 `update.sh` 和 `rollback.sh` 也上传到服务器,方便更新和回滚
|
||||
3. (可选)可以在服务器将编译出的 cli 可执行文件链接到 /bin 下以实现全局执行
|
||||
|
||||
cli 有三个命令:
|
||||
- `cli sync` 是同步城市节点
|
||||
- `cli update` 是更新网关配置
|
||||
- `cli clear` 是清空网关配置
|
||||
|
||||
> 三个命令对应于代码中 actions/ 目录下三个同名文件
|
||||
|
||||
## 逻辑梳理
|
||||
|
||||
```
|
||||
|
||||
每城市更新数量
|
||||
|
||||
遍历配置[网关][配置]
|
||||
如果无需更新直接放入新配置
|
||||
如果需要更新,放入城市更新统计
|
||||
|
||||
统计[城市][配置]
|
||||
遍历城市
|
||||
取到新节点
|
||||
遍历需要更新的配置:
|
||||
根据配置放入指定槽位[网关][配置]
|
||||
|
||||
新配置[网关][配置]
|
||||
```
|
||||
|
||||
@@ -1,17 +1,17 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"zzman/model"
|
||||
"jhman/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func RecordChange(tx *gorm.DB, changes []model.Change) error {
|
||||
func RecordChanges(tx *gorm.DB, changes []model.Change) error {
|
||||
if len(changes) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
batchSize := 1000
|
||||
batchSize := 500
|
||||
for i := 0; i < len(changes); i += batchSize {
|
||||
end := min(i+batchSize, len(changes))
|
||||
batch := changes[i:end]
|
||||
|
||||
@@ -2,7 +2,7 @@ package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"zzman/model"
|
||||
"jhman/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
@@ -37,16 +37,18 @@ func FindCitiesWithEdgesCount(tx *gorm.DB) ([]model.City, error) {
|
||||
return cities, nil
|
||||
}
|
||||
|
||||
func UpdateCityOffset(tx *gorm.DB, city int, offset int) error {
|
||||
if offset < 0 {
|
||||
return fmt.Errorf("offset must be non-negative")
|
||||
func UpdateCitiesOffset(tx *gorm.DB, updates []model.City) error {
|
||||
if len(updates) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
err := tx.Model(&model.City{}).
|
||||
Where("id = ?", city).
|
||||
Update("offset", offset).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to update cities offset: %w", err)
|
||||
for _, city := range updates {
|
||||
err := tx.Model(new(model.City)).
|
||||
Where("id = ?", city.Id).
|
||||
Update("offset", city.Offset).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to update city id %d offset: %w", city.Id, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -2,8 +2,8 @@ package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"zzman/clients/jd"
|
||||
"zzman/model"
|
||||
"jhman/clients/jd"
|
||||
"jhman/model"
|
||||
)
|
||||
|
||||
func Clear() error {
|
||||
|
||||
@@ -1,29 +1,20 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"zzman/model"
|
||||
"jhman/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func FindConfigsByGateway(tx *gorm.DB, macaddr string) ([]model.Config, error) {
|
||||
func FindConfigs(tx *gorm.DB) ([]model.Config, error) {
|
||||
var configs []model.Config
|
||||
err := tx.Find(&configs, "macaddr = ?", macaddr).Error
|
||||
err := tx.Find(&configs).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return configs, nil
|
||||
}
|
||||
|
||||
func CreateConfigs(tx *gorm.DB, configs []model.Config) error {
|
||||
if len(configs) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// 使用事务批量插入配置
|
||||
return tx.Omit("createtime", "updatetime").Create(&configs).Error
|
||||
}
|
||||
|
||||
func UpdateConfigs(tx *gorm.DB, configs []model.ConfigUpdate) error {
|
||||
if len(configs) == 0 {
|
||||
return nil
|
||||
@@ -31,7 +22,10 @@ func UpdateConfigs(tx *gorm.DB, configs []model.ConfigUpdate) error {
|
||||
|
||||
// 批量更新配置
|
||||
for _, config := range configs {
|
||||
err := tx.Updates(&config).Error
|
||||
err := tx.
|
||||
Where("id = ?", config.Id).
|
||||
Updates(&config).
|
||||
Error
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -39,3 +33,16 @@ func UpdateConfigs(tx *gorm.DB, configs []model.ConfigUpdate) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func CreateConfigs(tx *gorm.DB, configs []model.ConfigCreate) error {
|
||||
if len(configs) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
err := tx.CreateInBatches(&configs, 500).Error
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -2,10 +2,9 @@ package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"zzman/model"
|
||||
"jhman/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
)
|
||||
|
||||
const batchSize = 1000
|
||||
@@ -33,20 +32,76 @@ func SaveEdges(tx *gorm.DB, edges []model.Edge) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// 分批处理边缘设备数据
|
||||
for i := 0; i < len(edges); i += batchSize {
|
||||
end := min(i+batchSize, len(edges))
|
||||
// 手动区分需要创建和更新的节点,以避免 on conflict 更新的 id 占用问题
|
||||
macAddrs := make([]string, len(edges))
|
||||
for i, e := range edges {
|
||||
macAddrs[i] = e.Macaddr
|
||||
}
|
||||
|
||||
batch := edges[i:end]
|
||||
err := tx.Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "macaddr"}},
|
||||
UpdateAll: true,
|
||||
}).Create(&batch).Error
|
||||
var existingEdges []model.Edge
|
||||
if err := tx.Select("id", "macaddr").Where("macaddr IN ?", macAddrs).Find(&existingEdges).Error; err != nil {
|
||||
return fmt.Errorf("failed to find existing edges: %w", err)
|
||||
}
|
||||
existingEdgesMap := make(map[string]model.Edge)
|
||||
for _, e := range existingEdges {
|
||||
existingEdgesMap[e.Macaddr] = e
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to save edges batch %d-%d: %w", i, end-1, err)
|
||||
var toCreate, toUpdate []model.Edge
|
||||
for _, edge := range edges {
|
||||
if existing, ok := existingEdgesMap[edge.Macaddr]; ok {
|
||||
edge.Id = existing.Id
|
||||
toUpdate = append(toUpdate, edge)
|
||||
} else {
|
||||
toCreate = append(toCreate, edge)
|
||||
}
|
||||
}
|
||||
|
||||
// 批量创建
|
||||
if len(toCreate) > 0 {
|
||||
if err := tx.CreateInBatches(toCreate, batchSize).Error; err != nil {
|
||||
return fmt.Errorf("failed to create edge: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 批量更新(使用临时表避免性能问题)
|
||||
if len(toUpdate) > 0 {
|
||||
// 使用临时表进行批量更新
|
||||
tempTableName := "temp_edge_update"
|
||||
if err := tx.Exec(fmt.Sprintf("CREATE TEMPORARY TABLE %s LIKE edge", tempTableName)).Error; err != nil {
|
||||
return fmt.Errorf("failed to create temporary table: %w", err)
|
||||
}
|
||||
|
||||
// 将待更新数据插入临时表
|
||||
if err := tx.Table(tempTableName).CreateInBatches(toUpdate, batchSize).Error; err != nil {
|
||||
return fmt.Errorf("failed to insert into temporary table: %w", err)
|
||||
}
|
||||
|
||||
// 使用 JOIN 更新主表
|
||||
updateQuery := fmt.Sprintf(`
|
||||
UPDATE edge e
|
||||
JOIN %s te ON e.id = te.id
|
||||
SET
|
||||
e.macaddr = te.macaddr,
|
||||
e.public = te.public,
|
||||
e.isp = te.isp,
|
||||
e.single = te.single,
|
||||
e.sole = te.sole,
|
||||
e.arch = te.arch,
|
||||
e.online = te.online,
|
||||
e.city_id = te.city_id,
|
||||
e.active = te.active
|
||||
`, tempTableName)
|
||||
if err := tx.Exec(updateQuery).Error; err != nil {
|
||||
return fmt.Errorf("failed to batch update edges from temporary table: %w", err)
|
||||
}
|
||||
|
||||
// 删除临时表
|
||||
if err := tx.Exec(fmt.Sprintf("DROP TEMPORARY TABLE %s", tempTableName)).Error; err != nil {
|
||||
fmt.Printf("warning: failed to drop temporary table %s: %v\n", tempTableName, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -2,8 +2,8 @@ package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"jhman/model"
|
||||
"math"
|
||||
"zzman/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
55
actions/gen.go
Normal file
55
actions/gen.go
Normal file
@@ -0,0 +1,55 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"jhman/model"
|
||||
"strconv"
|
||||
|
||||
u "jhman/util"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func GenConfig(tx *gorm.DB) error {
|
||||
|
||||
gateways, err := FindGateways(tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cities, err := FindCitiesWithEdgesCount(tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
configs := make([]model.ConfigCreate, len(gateways)*30)
|
||||
for iGateway := range gateways {
|
||||
for j := range 30 {
|
||||
n := iGateway*30 + j
|
||||
iCity := (iGateway*30 + j) % len(cities)
|
||||
gateway := gateways[iGateway]
|
||||
city := cities[iCity]
|
||||
|
||||
configs[n] = model.ConfigCreate{
|
||||
Table: u.P(strconv.Itoa(j + 221)),
|
||||
Cityhash: &city.Hash,
|
||||
CityLabel: &city.Label,
|
||||
Network: u.P(fmt.Sprintf("172.30.168.%d", j+222)),
|
||||
GatewayMac: &gateway.Macaddr,
|
||||
User: u.P(fmt.Sprintf("jdzz%ddt%d", iGateway+1, j+221)),
|
||||
InnerIp: u.P(fmt.Sprintf("%s%d", gateway.ProxyIp, j+222)),
|
||||
EdgeMac: u.P(""),
|
||||
IsChange: u.P(1),
|
||||
IsOnline: u.P(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("生成配置:%d 条", len(configs)))
|
||||
err = CreateConfigs(tx, configs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
27
actions/submit.go
Normal file
27
actions/submit.go
Normal file
@@ -0,0 +1,27 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"jhman/clients/jd"
|
||||
"jhman/model"
|
||||
"time"
|
||||
)
|
||||
|
||||
func RecordSubmit(time time.Time, gatewat model.Gateway, edges []jd.EdgeInfo) error {
|
||||
config, err := json.Marshal(edges)
|
||||
if err != nil {
|
||||
return fmt.Errorf("序列化提交数据失败:%w", err)
|
||||
}
|
||||
|
||||
err = model.DB.Create(&model.Submit{
|
||||
Time: time,
|
||||
Gateway: gatewat.Macaddr,
|
||||
Config: string(config),
|
||||
}).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存提交记录失败:%w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -2,9 +2,9 @@ package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"jhman/clients/jd"
|
||||
"jhman/model"
|
||||
"log/slog"
|
||||
"zzman/clients/jd"
|
||||
"zzman/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
@@ -2,12 +2,12 @@ package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"jhman/clients/jd"
|
||||
"jhman/model"
|
||||
u "jhman/util"
|
||||
"log/slog"
|
||||
"strconv"
|
||||
"os"
|
||||
"time"
|
||||
"zzman/clients/jd"
|
||||
"zzman/model"
|
||||
u "zzman/util"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
@@ -31,72 +31,82 @@ func Update(args ...UpdateArgs) error {
|
||||
|
||||
func update(tx *gorm.DB, arg UpdateArgs) error {
|
||||
var now = time.Now()
|
||||
var step = now
|
||||
|
||||
// 获取所有网关
|
||||
gateways, err := FindGateways(tx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取所有网关失败:%w", err)
|
||||
}
|
||||
var findGateway = make(map[string]model.Gateway)
|
||||
for _, gateway := range gateways {
|
||||
findGateway[gateway.Macaddr] = gateway
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("获取网关:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 获取所有城市
|
||||
cities, err := FindCitiesWithEdgesCount(tx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取所有城市失败:%w", err)
|
||||
}
|
||||
|
||||
// 准备网关查找表,初始化网关配置表
|
||||
type ConfigInfo struct {
|
||||
Item jd.EdgeInfo
|
||||
Type int // 0: 不变, 1: 更新, 2: 新增
|
||||
var findCity = make(map[string]model.City)
|
||||
for _, city := range cities {
|
||||
findCity[city.Hash] = city
|
||||
}
|
||||
var oldConfigsMap = make(map[model.Gateway]map[string]model.Config)
|
||||
var newConfigs = make(map[model.Gateway][]ConfigInfo)
|
||||
|
||||
println(fmt.Sprintf("获取城市:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 获取所有配置
|
||||
configs, err := FindConfigs(tx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取所有配置失败:%w", err)
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("获取配置:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 查找需要更新的配置
|
||||
oldConfigs := make(map[model.City][]model.Config)
|
||||
newConfigs := make(map[model.Gateway][]ConfigInfo)
|
||||
for _, gateway := range gateways {
|
||||
newConfigs[gateway] = make([]ConfigInfo, len(cities))
|
||||
|
||||
oldConfigs, err := FindConfigsByGateway(tx, gateway.Macaddr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取网关 %s 城市节点失败:%w", gateway.Macaddr, err)
|
||||
}
|
||||
|
||||
oldConfigsMap[gateway] = make(map[string]model.Config)
|
||||
for _, config := range oldConfigs {
|
||||
oldConfigsMap[gateway][config.Cityhash] = config
|
||||
}
|
||||
newConfigs[gateway] = make([]ConfigInfo, 250) // 预分配空间,减少扩容
|
||||
}
|
||||
|
||||
// 按城市循环,对比并更新网关配置
|
||||
for iCity, city := range cities {
|
||||
|
||||
// 如果每个网关在此城市都有节点且无需改变,就不需要重新分配节点
|
||||
// 相当于直接重新提交配置,此流程下配置更新是幂等的
|
||||
var gateways2Update []model.Gateway
|
||||
for _, gateway := range gateways {
|
||||
oldConfig, exists := oldConfigsMap[gateway][city.Hash]
|
||||
if exists && oldConfig.IsChange != 1 {
|
||||
newConfigs[gateway][iCity] = ConfigInfo{
|
||||
Type: 0, // 不变
|
||||
Item: jd.EdgeInfo{
|
||||
Mac: oldConfig.Macaddr,
|
||||
City: oldConfig.Cityhash,
|
||||
},
|
||||
}
|
||||
} else {
|
||||
gateways2Update = append(gateways2Update, gateway)
|
||||
for _, config := range configs {
|
||||
city := findCity[config.Cityhash]
|
||||
gateway := findGateway[config.GatewayMac]
|
||||
if config.IsChange == 1 {
|
||||
oldConfigs[city] = append(oldConfigs[city], config)
|
||||
} else {
|
||||
newConfigs[gateway][config.Table-1] = ConfigInfo{
|
||||
Change: false,
|
||||
Remote: jd.EdgeInfo{
|
||||
Mac: config.EdgeMac,
|
||||
City: config.Cityhash,
|
||||
Network: config.Network,
|
||||
},
|
||||
}
|
||||
}
|
||||
if len(gateways2Update) == 0 {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// 否则获取足量新节点
|
||||
offset := city.Offset
|
||||
count := len(gateways2Update)
|
||||
println(fmt.Sprintf("查找更新:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 更新网关配置
|
||||
var citiesUpdate []model.City
|
||||
var changesCreate []model.Change
|
||||
for city, configs := range oldConfigs {
|
||||
|
||||
// 如果有需要变更的节点,获取足量新节点
|
||||
count := len(configs)
|
||||
if count > city.EdgesCount {
|
||||
slog.Warn(fmt.Sprintf("城市节点数量不足,跳过本次更新,城市:%s,节点数:%d,网关数:%d", city.Name, city.EdgesCount, count))
|
||||
continue
|
||||
}
|
||||
|
||||
edges, err := SliceActiveEdges(tx, city.Id, offset, count)
|
||||
edges, err := SliceActiveEdges(tx, city.Id, city.Offset, count)
|
||||
if err != nil {
|
||||
return fmt.Errorf("查询城市 %s 可用节点失败:%w", city.Name, err)
|
||||
}
|
||||
@@ -108,105 +118,102 @@ func update(tx *gorm.DB, arg UpdateArgs) error {
|
||||
}
|
||||
}
|
||||
|
||||
offset = edges[len(edges)-1].Id
|
||||
citiesUpdate = append(citiesUpdate, model.City{
|
||||
Id: city.Id,
|
||||
Offset: edges[len(edges)-1].Id,
|
||||
})
|
||||
|
||||
// 更新网关配置
|
||||
var configs2Create []model.Config
|
||||
var configs2Update []model.ConfigUpdate
|
||||
var changes []model.Change
|
||||
// 分配新节点
|
||||
for i, config := range configs {
|
||||
gateway := findGateway[config.GatewayMac]
|
||||
edge := edges[i]
|
||||
|
||||
for iGateway, gateway := range gateways2Update {
|
||||
slog.Debug(fmt.Sprintf("网关配置变更,网关:%s,旧节点:%s,新节点:%s", gateway.Macaddr, config.EdgeMac, edge.Macaddr))
|
||||
|
||||
oldConfig, exists := oldConfigsMap[gateway][city.Hash]
|
||||
newConfig := edges[iGateway]
|
||||
|
||||
if exists {
|
||||
slog.Debug(fmt.Sprintf("网关配置变更,网关:%s,旧节点:%s,新节点:%s", gateway.Macaddr, oldConfig.Macaddr, newConfig.Macaddr))
|
||||
|
||||
configs2Update = append(configs2Update, model.ConfigUpdate{
|
||||
Id: oldConfig.Id,
|
||||
Macaddr: u.P(oldConfig.Macaddr),
|
||||
newConfigs[gateway][config.Table-1] = ConfigInfo{
|
||||
Change: true,
|
||||
Remote: jd.EdgeInfo{
|
||||
Mac: edge.Macaddr,
|
||||
City: city.Hash,
|
||||
Network: config.Network,
|
||||
},
|
||||
Config: model.ConfigUpdate{
|
||||
Id: config.Id,
|
||||
EdgeMac: u.P(edge.Macaddr),
|
||||
IsChange: u.P(0),
|
||||
})
|
||||
|
||||
changes = append(changes, model.Change{
|
||||
Time: now,
|
||||
Gateway: gateway.Macaddr,
|
||||
OldEdge: oldConfig.Macaddr,
|
||||
NewEdge: newConfig.Macaddr,
|
||||
})
|
||||
|
||||
newConfigs[gateway][iCity] = ConfigInfo{
|
||||
Type: 1, // 更新
|
||||
Item: jd.EdgeInfo{
|
||||
Mac: oldConfig.Macaddr,
|
||||
City: city.Hash,
|
||||
},
|
||||
}
|
||||
} else {
|
||||
slog.Debug(fmt.Sprintf("网关配置新增,网关:%s,新节点:%s", gateway.Macaddr, newConfig.Macaddr))
|
||||
|
||||
configs2Create = append(configs2Create, model.Config{
|
||||
Cityhash: city.Hash,
|
||||
CityLabel: city.Label,
|
||||
GatewayMac: gateway.Macaddr,
|
||||
Macaddr: newConfig.Macaddr,
|
||||
Table: strconv.Itoa(iCity + 1),
|
||||
User: fmt.Sprintf("jdzz%ddt%d", gateway.Id, iCity+1),
|
||||
Network: fmt.Sprintf("172.30.168.%d", iCity+2),
|
||||
InnerIp: fmt.Sprintf("172.16.%d.%d", gateway.Id, iCity+2),
|
||||
})
|
||||
|
||||
changes = append(changes, model.Change{
|
||||
Time: now,
|
||||
Gateway: gateway.Macaddr,
|
||||
OldEdge: "",
|
||||
NewEdge: newConfig.Macaddr,
|
||||
})
|
||||
|
||||
newConfigs[gateway][iCity] = ConfigInfo{
|
||||
Type: 2, // 新增
|
||||
Item: jd.EdgeInfo{
|
||||
Mac: oldConfig.Macaddr,
|
||||
City: city.Hash,
|
||||
},
|
||||
}
|
||||
IsOnline: u.P(0),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
err = CreateConfigs(tx, configs2Create)
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建新配置失败:%w", err)
|
||||
}
|
||||
err = UpdateConfigs(tx, configs2Update)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新配置失败:%w", err)
|
||||
}
|
||||
err = UpdateCityOffset(tx, city.Id, offset)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新城市 %s 的偏移量失败:%w", city.Name, err)
|
||||
}
|
||||
err = RecordChange(tx, changes)
|
||||
if err != nil {
|
||||
return fmt.Errorf("记录变更失败:%w", err)
|
||||
changesCreate = append(changesCreate, model.Change{
|
||||
Time: now,
|
||||
CityId: city.Id,
|
||||
Gateway: gateway.Macaddr,
|
||||
OldEdge: config.EdgeMac,
|
||||
NewEdge: edge.Macaddr,
|
||||
Network: config.Network,
|
||||
Info: edge.Public,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("更新配置:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 更新城市偏移量
|
||||
err = UpdateCitiesOffset(tx, citiesUpdate)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新城市偏移量失败:%w", err)
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("更新城市偏移量:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 记录节点变更
|
||||
if os.Getenv("DEBUG") == "true" {
|
||||
err = RecordChanges(tx, changesCreate)
|
||||
if err != nil {
|
||||
return fmt.Errorf("记录节点变更失败:%w", err)
|
||||
}
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("记录节点变更:%v", time.Since(step)))
|
||||
step = time.Now()
|
||||
|
||||
// 提交所有网关配置到云端
|
||||
for gateway, infos := range newConfigs {
|
||||
change := 0
|
||||
setup := 0
|
||||
edges := make([]jd.EdgeInfo, len(infos))
|
||||
configsUpdate := make([]model.ConfigUpdate, 0)
|
||||
|
||||
// 统计变更数
|
||||
change := 0
|
||||
for i, info := range infos {
|
||||
edges[i] = info.Item
|
||||
switch info.Type {
|
||||
case 1:
|
||||
edges[i] = info.Remote
|
||||
if info.Change {
|
||||
configsUpdate = append(configsUpdate, info.Config)
|
||||
change++
|
||||
case 2:
|
||||
setup++
|
||||
}
|
||||
}
|
||||
slog.Info(fmt.Sprintf("提交网关配置,网关:%s,变更数:%d,新增数:%d", gateway.Macaddr, change, setup))
|
||||
slog.Info(fmt.Sprintf("提交网关配置,网关:%s,变更数:%d", gateway.Macaddr, change))
|
||||
|
||||
// 更新配置版本
|
||||
err = AppendGatewayConfigVersion(tx, gateway.Id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新网关 %s 配置版本失败:%w", gateway.Macaddr, err)
|
||||
}
|
||||
|
||||
// 更新本地配置
|
||||
err = UpdateConfigs(tx, configsUpdate)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新网关 %s 本地配置失败:%w", gateway.Macaddr, err)
|
||||
}
|
||||
|
||||
// 记录提交配置
|
||||
if os.Getenv("DEBUG") == "true" {
|
||||
err = RecordSubmit(now, gateway, edges)
|
||||
if err != nil {
|
||||
return fmt.Errorf("记录网关 %s 提交配置失败:%w", gateway.Macaddr, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 提交配置到云端:配置版本 gateway.ConfigVersion
|
||||
if !arg.Mock {
|
||||
@@ -215,13 +222,18 @@ func update(tx *gorm.DB, arg UpdateArgs) error {
|
||||
return fmt.Errorf("配置网关 %s 失败:%w", gateway.Macaddr, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 更新配置版本
|
||||
err = AppendGatewayConfigVersion(tx, gateway.Id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("更新网关 %s 配置版本失败:%w", gateway.Macaddr, err)
|
||||
}
|
||||
}
|
||||
|
||||
println(fmt.Sprintf("提交配置:%v", time.Since(step)))
|
||||
|
||||
println(fmt.Sprintf("总耗时:%v", time.Since(now)))
|
||||
println(fmt.Sprint("更新总数:", len(changesCreate)))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type ConfigInfo struct {
|
||||
Change bool
|
||||
Remote jd.EdgeInfo
|
||||
Config model.ConfigUpdate
|
||||
}
|
||||
|
||||
@@ -6,11 +6,12 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"io"
|
||||
"jhman/clients"
|
||||
"net/http"
|
||||
"time"
|
||||
"zzman/clients"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
const base = "https://smart.jdbox.xyz:58001"
|
||||
|
||||
@@ -3,8 +3,9 @@ package jd
|
||||
import "fmt"
|
||||
|
||||
type EdgeInfo struct {
|
||||
Mac string
|
||||
City string
|
||||
Mac string
|
||||
City string
|
||||
Network string
|
||||
}
|
||||
|
||||
func GatewayConfigSet(version int, macaddr string, edges []EdgeInfo) error {
|
||||
@@ -23,7 +24,7 @@ func GatewayConfigSet(version int, macaddr string, edges []EdgeInfo) error {
|
||||
rules[i] = GateConfigSetReqRule{
|
||||
Enable: true,
|
||||
Edge: []string{edge.Mac},
|
||||
Network: []string{fmt.Sprintf("172.30.168.%d", i+2)},
|
||||
Network: []string{edge.Network},
|
||||
Cityhash: edge.City, // 每个 edge 的城市应当相同
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,7 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
|
||||
redis "github.com/redis/go-redis/v9"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var Redis *redis.Client
|
||||
|
||||
32
cmd/test.go
32
cmd/test.go
@@ -1,32 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"zzman/actions"
|
||||
"zzman/clients"
|
||||
"zzman/model"
|
||||
|
||||
"github.com/joho/godotenv"
|
||||
)
|
||||
|
||||
func main() {
|
||||
// 初始化环境
|
||||
slog.Debug("初始化环境变量")
|
||||
err := godotenv.Load()
|
||||
if err != nil {
|
||||
slog.Error(fmt.Errorf("初始化变量失败:%w", err).Error())
|
||||
}
|
||||
|
||||
// 初始化数据库和 Redis
|
||||
model.Init()
|
||||
defer model.Close()
|
||||
|
||||
clients.InitRedis()
|
||||
defer clients.CloseRedis()
|
||||
|
||||
// 测试功能
|
||||
actions.Update(actions.UpdateArgs{
|
||||
Mock: true,
|
||||
})
|
||||
}
|
||||
@@ -1,3 +1,3 @@
|
||||
$env:GOOS="linux"; $env:GOARCH="amd64"; $env:CGO_ENABLED=0; go build -o dist/zz main.go
|
||||
$env:GOOS="linux"; $env:GOARCH="amd64"; $env:CGO_ENABLED=0; go build -o dist/zz ./cmd/clt/main.go
|
||||
|
||||
scp ./dist/zz wyk@43.226.58.254:~/zzcli
|
||||
scp ./dist/zz wyk@43.226.58.254:~/zz/cli.new
|
||||
|
||||
@@ -5,19 +5,19 @@ services:
|
||||
redis:
|
||||
image: redis:7
|
||||
environment:
|
||||
REDIS_PASSWORD: 123456
|
||||
REDIS_PASSWORD: ${REDIS_PASSWORD}
|
||||
ports:
|
||||
- "6379:6379"
|
||||
- "${REDIS_PORT}:6379"
|
||||
volumes:
|
||||
- redis_data:/data
|
||||
|
||||
mariadb:
|
||||
image: mariadb:10
|
||||
environment:
|
||||
MYSQL_ROOT_PASSWORD: "hljuip916..."
|
||||
MYSQL_DATABASE: jdbox
|
||||
MYSQL_ROOT_PASSWORD: ${MYSQL_PASSWORD}
|
||||
MYSQL_DATABASE: ${MYSQL_DATABASE}
|
||||
ports:
|
||||
- "3306:3306"
|
||||
- "${MYSQL_PORT}:3306"
|
||||
volumes:
|
||||
- mariadb_data:/var/lib/mysql
|
||||
|
||||
|
||||
4
go.mod
4
go.mod
@@ -1,10 +1,11 @@
|
||||
module zzman
|
||||
module jhman
|
||||
|
||||
go 1.24.0
|
||||
|
||||
require (
|
||||
github.com/joho/godotenv v1.5.1
|
||||
github.com/redis/go-redis/v9 v9.11.0
|
||||
github.com/systemd/slog-journal v0.1.0
|
||||
gorm.io/driver/mysql v1.6.0
|
||||
gorm.io/gorm v1.30.1
|
||||
)
|
||||
@@ -16,7 +17,6 @@ require (
|
||||
github.com/go-sql-driver/mysql v1.9.3 // indirect
|
||||
github.com/jinzhu/inflection v1.0.0 // indirect
|
||||
github.com/jinzhu/now v1.1.5 // indirect
|
||||
github.com/systemd/slog-journal v0.1.0 // indirect
|
||||
golang.org/x/sys v0.29.0 // indirect
|
||||
golang.org/x/text v0.27.0 // indirect
|
||||
)
|
||||
|
||||
97
main.go
97
main.go
@@ -1,97 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"zzman/actions"
|
||||
"zzman/clients"
|
||||
"zzman/model"
|
||||
|
||||
"github.com/joho/godotenv"
|
||||
slogjournal "github.com/systemd/slog-journal"
|
||||
)
|
||||
|
||||
func main() {
|
||||
|
||||
// 初始化环境
|
||||
slog.Debug("初始化环境变量")
|
||||
ex, err := os.Executable()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
exPath := filepath.Dir(ex)
|
||||
file := filepath.Join(exPath, ".env")
|
||||
println("加载环境变量文件:", file)
|
||||
err = godotenv.Load(file)
|
||||
if err != nil {
|
||||
slog.Error(fmt.Errorf("初始化变量失败:%w", err).Error())
|
||||
}
|
||||
|
||||
// 初始化日志
|
||||
handler, err := slogjournal.NewHandler(&slogjournal.Options{
|
||||
ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr {
|
||||
a.Key = strings.ToUpper(a.Key)
|
||||
a.Key = strings.ReplaceAll(a.Key, "-", "_")
|
||||
return a
|
||||
},
|
||||
})
|
||||
slog.SetDefault(slog.New(handler))
|
||||
|
||||
// 初始化数据库和 Redis
|
||||
model.Init()
|
||||
defer model.Close()
|
||||
|
||||
clients.InitRedis()
|
||||
defer clients.CloseRedis()
|
||||
|
||||
// 执行命令
|
||||
if len(os.Args) < 2 {
|
||||
println("缺少命令参数")
|
||||
return
|
||||
}
|
||||
switch os.Args[1] {
|
||||
|
||||
// 同步城市节点
|
||||
case "sync":
|
||||
err := actions.Sync()
|
||||
if err != nil {
|
||||
slog.Error(fmt.Sprintf("同步城市节点数据失败:%s", err.Error()))
|
||||
} else {
|
||||
slog.Info("同步城市节点数据成功")
|
||||
}
|
||||
return
|
||||
|
||||
// 更新网关
|
||||
case "update":
|
||||
var args actions.UpdateArgs
|
||||
if len(os.Args) >= 3 {
|
||||
if slices.Contains(os.Args, "--mock") {
|
||||
args.Mock = true
|
||||
}
|
||||
}
|
||||
|
||||
err := actions.Update(args)
|
||||
if err != nil {
|
||||
slog.Error(fmt.Sprintf("更新节点失败:%s", err.Error()))
|
||||
} else {
|
||||
slog.Info("更新节点成功")
|
||||
}
|
||||
return
|
||||
|
||||
// 清空网关
|
||||
case "clear":
|
||||
err := actions.Clear()
|
||||
if err != nil {
|
||||
slog.Error(fmt.Sprintf("清空节点失败:%s", err.Error()))
|
||||
} else {
|
||||
slog.Info("清空节点成功")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
println("请输入正确的命令参数")
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import "time"
|
||||
type Change struct {
|
||||
Id int `gorm:"column:id;primaryKey"`
|
||||
Time time.Time `gorm:"column:time"`
|
||||
CityId int `gorm:"column:city"`
|
||||
Gateway string `gorm:"column:macaddr"`
|
||||
OldEdge string `gorm:"column:edge_old"`
|
||||
NewEdge string `gorm:"column:edge_new"`
|
||||
|
||||
@@ -3,8 +3,8 @@ package model
|
||||
type Config struct {
|
||||
Id int `gorm:"column:id;primaryKey"`
|
||||
GatewayMac string `gorm:"column:macaddr"`
|
||||
Table string `gorm:"column:table"`
|
||||
Macaddr string `gorm:"column:edge"`
|
||||
Table int `gorm:"column:table"`
|
||||
EdgeMac string `gorm:"column:edge"`
|
||||
Network string `gorm:"column:network"`
|
||||
Cityhash string `gorm:"column:cityhash"`
|
||||
CityLabel string `gorm:"column:label"`
|
||||
@@ -25,7 +25,7 @@ type ConfigUpdate struct {
|
||||
Id int `gorm:"column:id;primaryKey"`
|
||||
GatewayMac *string `gorm:"column:macaddr"`
|
||||
Table *string `gorm:"column:table"`
|
||||
Macaddr *string `gorm:"column:edge"`
|
||||
EdgeMac *string `gorm:"column:edge"`
|
||||
Network *string `gorm:"column:network"`
|
||||
Cityhash *string `gorm:"column:cityhash"`
|
||||
CityLabel *string `gorm:"column:label"`
|
||||
@@ -39,3 +39,20 @@ type ConfigUpdate struct {
|
||||
func (ConfigUpdate) TableName() string {
|
||||
return "gateway"
|
||||
}
|
||||
|
||||
type ConfigCreate struct {
|
||||
GatewayMac *string `gorm:"column:macaddr"`
|
||||
Table *string `gorm:"column:table"`
|
||||
EdgeMac *string `gorm:"column:edge"`
|
||||
Network *string `gorm:"column:network"`
|
||||
Cityhash *string `gorm:"column:cityhash"`
|
||||
CityLabel *string `gorm:"column:label"`
|
||||
User *string `gorm:"column:user"`
|
||||
InnerIp *string `gorm:"column:inner_ip"`
|
||||
IsChange *int `gorm:"column:ischange"`
|
||||
IsOnline *int `gorm:"column:isonline"`
|
||||
}
|
||||
|
||||
func (ConfigCreate) TableName() string {
|
||||
return "gateway"
|
||||
}
|
||||
|
||||
14
model/submit.go
Normal file
14
model/submit.go
Normal file
@@ -0,0 +1,14 @@
|
||||
package model
|
||||
|
||||
import "time"
|
||||
|
||||
type Submit struct {
|
||||
Id int `gorm:"column:id;primaryKey"`
|
||||
Time time.Time `gorm:"column:time"`
|
||||
Gateway string `gorm:"column:gateway"`
|
||||
Config string `gorm:"column:config"`
|
||||
}
|
||||
|
||||
func (Submit) TableName() string {
|
||||
return "submit"
|
||||
}
|
||||
12
rollback.sh
Normal file
12
rollback.sh
Normal file
@@ -0,0 +1,12 @@
|
||||
#!/bin/bash
|
||||
set -e
|
||||
|
||||
# 此脚本应当运行在服务端,与 deploy.ps1 脚本上传到的位置配合,自动切换回退
|
||||
|
||||
cp cli cli.new
|
||||
chmod -x cli.new
|
||||
|
||||
chmod +x cli.bak
|
||||
mv cli.bak cli
|
||||
|
||||
echo 'done~'
|
||||
12
update.sh
Normal file
12
update.sh
Normal file
@@ -0,0 +1,12 @@
|
||||
#!/bin/bash
|
||||
set -e
|
||||
|
||||
# 此脚本应当运行在服务端,与 deploy.ps1 脚本上传到的位置配合,自动切换更新
|
||||
|
||||
cp cli cli.bak
|
||||
chmod -x cli.bak
|
||||
|
||||
chmod +x cli.new
|
||||
mv cli.new cli
|
||||
|
||||
echo 'done~'
|
||||
57
web/web.go
Normal file
57
web/web.go
Normal file
@@ -0,0 +1,57 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
)
|
||||
|
||||
func Start() {
|
||||
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
|
||||
defer cancel()
|
||||
|
||||
// 初始化数据库和 Redis
|
||||
// model.Init()
|
||||
// defer model.Close()
|
||||
|
||||
// clients.InitRedis()
|
||||
// defer clients.CloseRedis()
|
||||
|
||||
// 测试
|
||||
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Write([]byte("Hello, World!"))
|
||||
})
|
||||
|
||||
// 同步节点
|
||||
http.HandleFunc("/sync", func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
})
|
||||
|
||||
// 更新节点
|
||||
http.HandleFunc("/update", func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
})
|
||||
|
||||
// 标记上线
|
||||
http.HandleFunc("/up", func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
})
|
||||
|
||||
// 标记下线
|
||||
http.HandleFunc("/down", func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
})
|
||||
|
||||
// 启动服务
|
||||
server := http.Server{Addr: ":8080"}
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
server.Shutdown(ctx)
|
||||
}()
|
||||
|
||||
if err := server.ListenAndServe(); err != nil {
|
||||
slog.Error("启动服务失败", "error", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user