Compare commits

...

11 Commits

27 changed files with 547 additions and 329 deletions

5
.gitignore vendored
View File

@@ -27,4 +27,7 @@ go.work.sum
.env
# editor file
.idea/
.idea/
cmd/
.vscode/

13
.zed/debug.json Normal file
View 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",
},
]

View File

@@ -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/ 目录下三个同名文件
## 逻辑梳理
```
每城市更新数量
遍历配置[网关][配置]
如果无需更新直接放入新配置
如果需要更新,放入城市更新统计
统计[城市][配置]
遍历城市
取到新节点
遍历需要更新的配置:
根据配置放入指定槽位[网关][配置]
新配置[网关][配置]
```

View File

@@ -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]

View File

@@ -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

View File

@@ -2,8 +2,8 @@ package actions
import (
"fmt"
"zzman/clients/jd"
"zzman/model"
"jhman/clients/jd"
"jhman/model"
)
func Clear() error {

View File

@@ -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
}

View File

@@ -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
}

View File

@@ -2,8 +2,8 @@ package actions
import (
"fmt"
"jhman/model"
"math"
"zzman/model"
"gorm.io/gorm"
)

55
actions/gen.go Normal file
View 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
View 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
}

View File

@@ -2,9 +2,9 @@ package actions
import (
"fmt"
"jhman/clients/jd"
"jhman/model"
"log/slog"
"zzman/clients/jd"
"zzman/model"
"gorm.io/gorm"
)

View File

@@ -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
}

View File

@@ -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"

View File

@@ -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 的城市应当相同
}
}

View File

@@ -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

View File

@@ -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,
})
}

View File

@@ -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

View File

@@ -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
View File

@@ -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
View File

@@ -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("请输入正确的命令参数")
}

View File

@@ -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"`

View File

@@ -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
View 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
View 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
View 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
View 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)
}
}