Files
EdgeNode/internal/nodes/node.go

854 lines
20 KiB
Go
Raw Normal View History

2020-07-21 11:18:47 +08:00
package nodes
import (
2021-11-11 14:16:57 +08:00
"bytes"
2020-09-09 18:53:53 +08:00
"encoding/json"
"errors"
2020-09-26 08:07:07 +08:00
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
2020-09-13 20:37:40 +08:00
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
2020-10-05 16:55:14 +08:00
"github.com/TeaOSLab/EdgeNode/internal/caches"
"github.com/TeaOSLab/EdgeNode/internal/configs"
2021-01-12 11:48:38 +08:00
teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
2020-10-28 11:19:06 +08:00
"github.com/TeaOSLab/EdgeNode/internal/events"
"github.com/TeaOSLab/EdgeNode/internal/firewalls"
"github.com/TeaOSLab/EdgeNode/internal/goman"
"github.com/TeaOSLab/EdgeNode/internal/iplibrary"
2021-06-30 20:01:00 +08:00
"github.com/TeaOSLab/EdgeNode/internal/metrics"
2021-12-09 17:34:05 +08:00
"github.com/TeaOSLab/EdgeNode/internal/ratelimit"
"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
2020-09-09 18:53:53 +08:00
"github.com/TeaOSLab/EdgeNode/internal/rpc"
2021-01-25 16:40:31 +08:00
"github.com/TeaOSLab/EdgeNode/internal/stats"
"github.com/TeaOSLab/EdgeNode/internal/trackers"
2020-07-22 22:18:47 +08:00
"github.com/TeaOSLab/EdgeNode/internal/utils"
2021-11-11 14:16:57 +08:00
"github.com/andybalholm/brotli"
"github.com/iwind/TeaGo/Tea"
2021-01-12 11:48:38 +08:00
"github.com/iwind/TeaGo/lists"
2021-01-11 18:16:15 +08:00
"github.com/iwind/TeaGo/logs"
2021-07-25 17:14:44 +08:00
"github.com/iwind/TeaGo/maps"
2021-12-02 11:30:47 +08:00
"github.com/iwind/TeaGo/types"
2021-07-25 17:14:44 +08:00
"github.com/iwind/gosock/pkg/gosock"
2022-03-04 12:30:06 +08:00
"gopkg.in/yaml.v3"
"io/ioutil"
2021-01-12 11:48:38 +08:00
"log"
"os"
2021-01-12 11:48:38 +08:00
"os/exec"
"runtime"
2021-12-09 12:07:59 +08:00
"runtime/debug"
"sort"
"sync"
2020-09-09 18:53:53 +08:00
"time"
2020-07-21 11:18:47 +08:00
)
2020-09-26 08:07:07 +08:00
var sharedNodeConfig *nodeconfigs.NodeConfig
var nodeTaskNotify = make(chan bool, 8)
var nodeConfigChangedNotify = make(chan bool, 8)
var nodeConfigUpdatedAt int64
var DaemonIsOn = false
var DaemonPid = 0
2020-07-21 11:18:47 +08:00
// Node 节点
2020-07-21 11:18:47 +08:00
type Node struct {
isLoaded bool
2021-07-25 17:14:44 +08:00
sock *gosock.Sock
locker sync.Mutex
2021-12-09 12:07:59 +08:00
2021-12-09 17:34:05 +08:00
maxCPU int32
maxThreads int
timezone string
2022-01-19 22:16:46 +08:00
updatingServerMap map[int64]*serverconfigs.ServerConfig
2020-07-21 11:18:47 +08:00
}
func NewNode() *Node {
2021-07-25 17:39:09 +08:00
return &Node{
2022-01-19 22:16:46 +08:00
sock: gosock.NewTmpSock(teaconst.ProcessName),
maxThreads: -1,
maxCPU: -1,
updatingServerMap: map[int64]*serverconfigs.ServerConfig{},
2021-07-25 17:39:09 +08:00
}
2020-07-21 11:18:47 +08:00
}
// Test 检查配置
func (this *Node) Test() error {
// 检查是否能连接API
rpcClient, err := rpc.SharedRPC()
if err != nil {
return errors.New("test rpc failed: " + err.Error())
}
_, err = rpcClient.APINodeRPC().FindCurrentAPINodeVersion(rpcClient.Context(), &pb.FindCurrentAPINodeVersionRequest{})
if err != nil {
return errors.New("test rpc failed: " + err.Error())
}
return nil
}
// Start 启动
2020-07-21 11:18:47 +08:00
func (this *Node) Start() {
_, ok := os.LookupEnv("EdgeDaemon")
if ok {
remotelogs.Println("NODE", "start from daemon")
DaemonIsOn = true
DaemonPid = os.Getppid()
}
// 处理异常
this.handlePanic()
// 启动事件
events.Notify(events.EventStart)
// 本地Sock
err := this.listenSock()
if err != nil {
remotelogs.Error("NODE", err.Error())
return
}
2020-09-09 18:53:53 +08:00
// 读取API配置
2021-11-11 14:16:57 +08:00
err = this.syncConfig(0)
if err != nil {
_, err := nodeconfigs.SharedNodeConfig()
2021-01-11 18:16:15 +08:00
if err != nil {
// 无本地数据时,会尝试多次读取
tryTimes := 0
for {
2021-11-11 14:16:57 +08:00
err := this.syncConfig(0)
if err != nil {
tryTimes++
if tryTimes%10 == 0 {
remotelogs.Error("NODE", err.Error())
}
time.Sleep(1 * time.Second)
// 不做长时间的无意义的重试
if tryTimes > 1000 {
return
}
} else {
break
}
2021-01-11 18:16:15 +08:00
}
}
2020-09-09 18:53:53 +08:00
}
// 启动同步计时器
this.startSyncTimer()
// 状态变更计时器
goman.New(func() {
NewNodeStatusExecutor().Listen()
})
2020-09-09 18:53:53 +08:00
2020-07-22 22:18:47 +08:00
// 读取配置
2020-09-26 08:07:07 +08:00
nodeConfig, err := nodeconfigs.SharedNodeConfig()
2020-07-21 11:18:47 +08:00
if err != nil {
remotelogs.Error("NODE", "start failed: read node config failed: "+err.Error())
2020-07-21 11:18:47 +08:00
return
}
teaconst.NodeId = nodeConfig.Id
2021-12-02 11:30:47 +08:00
teaconst.NodeIdString = types.String(teaconst.NodeId)
2021-12-01 15:52:38 +08:00
err, serverErrors := nodeConfig.Init()
2020-09-26 08:07:07 +08:00
if err != nil {
remotelogs.Error("NODE", "init node config failed: "+err.Error())
2020-09-26 08:07:07 +08:00
return
}
2021-12-01 15:52:38 +08:00
if len(serverErrors) > 0 {
for _, serverErr := range serverErrors {
remotelogs.ServerError(serverErr.Id, "NODE", serverErr.Message, nodeconfigs.NodeLogTypeServerConfigInitFailed, maps.Map{})
}
}
2020-09-26 08:07:07 +08:00
sharedNodeConfig = nodeConfig
2021-12-09 12:07:59 +08:00
this.onReload(nodeConfig)
2020-07-21 11:18:47 +08:00
2020-12-03 10:17:28 +08:00
// 发送事件
events.Notify(events.EventLoaded)
2020-07-22 22:18:47 +08:00
// 设置rlimit
_ = utils.SetRLimit(1024 * 1024)
2020-10-04 14:30:42 +08:00
// 连接API
goman.New(func() {
NewAPIStream().Start()
})
2020-10-04 14:30:42 +08:00
2021-01-25 16:40:31 +08:00
// 统计
2021-12-14 10:01:21 +08:00
goman.New(func() {
stats.SharedTrafficStatManager.Start(func() *nodeconfigs.NodeConfig {
return sharedNodeConfig
})
2021-01-25 16:40:31 +08:00
})
goman.New(func() {
stats.SharedHTTPRequestStatManager.Start()
})
2021-01-25 16:40:31 +08:00
2020-07-22 22:18:47 +08:00
// 启动端口
err = sharedListenerManager.Start(nodeConfig)
if err != nil {
remotelogs.Error("NODE", "start failed: "+err.Error())
2020-10-28 11:19:06 +08:00
return
}
2020-07-22 22:18:47 +08:00
// hold住进程
2020-10-04 14:30:42 +08:00
select {}
2020-07-21 11:18:47 +08:00
}
2020-09-09 18:53:53 +08:00
// Daemon 实现守护进程
2021-01-12 11:48:38 +08:00
func (this *Node) Daemon() {
isDebug := lists.ContainsString(os.Args, "debug")
isDebug = true
for {
2021-07-25 17:39:09 +08:00
conn, err := this.sock.Dial()
2021-01-12 11:48:38 +08:00
if err != nil {
if isDebug {
log.Println("[DAEMON]starting ...")
}
// 尝试启动
err = func() error {
exe, err := os.Executable()
if err != nil {
return err
}
// 可以标记当前是从守护进程启动的
_ = os.Setenv("EdgeDaemon", "on")
_ = os.Setenv("EdgeBackground", "on")
2021-01-12 11:48:38 +08:00
cmd := exec.Command(exe)
err = cmd.Start()
if err != nil {
return err
}
err = cmd.Wait()
if err != nil {
return err
}
return nil
}()
if err != nil {
if isDebug {
log.Println("[DAEMON]", err)
}
time.Sleep(1 * time.Second)
} else {
time.Sleep(5 * time.Second)
}
} else {
_ = conn.Close()
time.Sleep(5 * time.Second)
}
}
}
// InstallSystemService 安装系统服务
2021-01-12 11:48:38 +08:00
func (this *Node) InstallSystemService() error {
shortName := teaconst.SystemdServiceName
exe, err := os.Executable()
if err != nil {
return err
}
manager := utils.NewServiceManager(shortName, teaconst.ProductName)
err = manager.Install(exe, []string{})
if err != nil {
return err
}
return nil
}
// 循环
func (this *Node) loop() error {
var tr = trackers.Begin("CHECK_NODE_CONFIG_CHANGES")
defer tr.End()
// 检查api.yaml是否存在
2022-01-19 22:16:46 +08:00
var apiConfigFile = Tea.ConfigFile("api.yaml")
_, err := os.Stat(apiConfigFile)
if err != nil {
return nil
}
rpcClient, err := rpc.SharedRPC()
if err != nil {
return errors.New("create rpc client failed: " + err.Error())
}
2022-01-19 22:16:46 +08:00
var nodeCtx = rpcClient.Context()
tasksResp, err := rpcClient.NodeTaskRPC().FindNodeTasks(nodeCtx, &pb.FindNodeTasksRequest{})
if err != nil {
return errors.New("read node tasks failed: " + err.Error())
}
for _, task := range tasksResp.NodeTasks {
switch task.Type {
case "ipItemChanged":
iplibrary.IPListUpdateNotify <- true
// 修改为已同步
_, err = rpcClient.NodeTaskRPC().ReportNodeTaskDone(nodeCtx, &pb.ReportNodeTaskDoneRequest{
NodeTaskId: task.Id,
IsOk: true,
Error: "",
})
if err != nil {
return err
}
case "configChanged":
2022-01-19 22:16:46 +08:00
if task.ServerId > 0 {
err = this.syncServerConfig(task.ServerId)
} else {
if !task.IsPrimary {
// 我们等等主节点配置准备完毕
time.Sleep(2 * time.Second)
}
err = this.syncConfig(task.Version)
2021-11-11 14:16:57 +08:00
}
if err != nil {
_, err = rpcClient.NodeTaskRPC().ReportNodeTaskDone(nodeCtx, &pb.ReportNodeTaskDoneRequest{
NodeTaskId: task.Id,
IsOk: false,
Error: err.Error(),
})
} else {
_, err = rpcClient.NodeTaskRPC().ReportNodeTaskDone(nodeCtx, &pb.ReportNodeTaskDoneRequest{
NodeTaskId: task.Id,
IsOk: true,
Error: "",
})
}
if err != nil {
return err
}
2022-01-19 22:16:46 +08:00
case "nodeVersionChanged":
goman.New(func() {
sharedUpgradeManager.Start()
})
}
}
return nil
}
2020-09-09 18:53:53 +08:00
// 读取API配置
2021-11-11 14:16:57 +08:00
func (this *Node) syncConfig(taskVersion int64) error {
this.locker.Lock()
defer this.locker.Unlock()
// 检查api.yaml是否存在
apiConfigFile := Tea.ConfigFile("api.yaml")
_, err := os.Stat(apiConfigFile)
if err != nil {
if os.IsNotExist(err) {
clusterErr := this.checkClusterConfig()
if clusterErr != nil {
if os.IsNotExist(clusterErr) {
return err
}
return errors.New("check cluster config failed: " + clusterErr.Error())
}
} else {
return err
}
}
2020-09-09 18:53:53 +08:00
rpcClient, err := rpc.SharedRPC()
if err != nil {
return errors.New("create rpc client failed: " + err.Error())
2020-09-09 18:53:53 +08:00
}
// 获取同步任务
nodeCtx := rpcClient.Context()
2020-09-26 08:07:07 +08:00
// TODO 这里考虑只同步版本号有变更的
configResp, err := rpcClient.NodeRPC().FindCurrentNodeConfig(nodeCtx, &pb.FindCurrentNodeConfigRequest{
2021-11-11 14:16:57 +08:00
Version: -1, // 更新所有版本
Compress: true,
NodeTaskVersion: taskVersion,
2020-12-03 10:17:28 +08:00
})
2020-09-09 18:53:53 +08:00
if err != nil {
return errors.New("read config from rpc failed: " + err.Error())
2020-09-09 18:53:53 +08:00
}
2020-12-03 10:17:28 +08:00
if !configResp.IsChanged {
return nil
}
2020-09-26 08:07:07 +08:00
configJSON := configResp.NodeJSON
2021-11-11 14:16:57 +08:00
if configResp.IsCompressed {
var reader = brotli.NewReader(bytes.NewReader(configJSON))
var configBuf = &bytes.Buffer{}
var buf = make([]byte, 32*1024)
for {
n, err := reader.Read(buf)
if n > 0 {
configBuf.Write(buf[:n])
}
if err != nil {
break
}
}
configJSON = configBuf.Bytes()
}
nodeConfigUpdatedAt = time.Now().Unix()
2020-09-26 08:07:07 +08:00
nodeConfig := &nodeconfigs.NodeConfig{}
err = json.Unmarshal(configJSON, nodeConfig)
2020-09-09 18:53:53 +08:00
if err != nil {
return errors.New("decode config failed: " + err.Error())
2020-09-09 18:53:53 +08:00
}
teaconst.NodeId = nodeConfig.Id
2021-12-02 11:30:47 +08:00
teaconst.NodeIdString = types.String(teaconst.NodeId)
2020-09-09 18:53:53 +08:00
// 写入到文件中
err = nodeConfig.Save()
if err != nil {
return err
}
2021-12-01 15:52:38 +08:00
err, serverErrors := nodeConfig.Init()
2020-09-09 18:53:53 +08:00
if err != nil {
return err
}
2021-12-01 15:52:38 +08:00
if len(serverErrors) > 0 {
for _, serverErr := range serverErrors {
remotelogs.ServerError(serverErr.Id, "NODE", serverErr.Message, nodeconfigs.NodeLogTypeServerConfigInitFailed, maps.Map{})
}
}
2020-09-09 18:53:53 +08:00
2020-09-26 08:07:07 +08:00
// 刷新配置
if this.isLoaded {
remotelogs.Println("NODE", "reloading config ...")
} else {
remotelogs.Println("NODE", "loading config ...")
}
2021-12-09 12:07:59 +08:00
this.onReload(nodeConfig)
2020-09-26 08:07:07 +08:00
2020-12-03 10:17:28 +08:00
// 发送事件
events.Notify(events.EventReload)
if this.isLoaded {
2020-09-09 18:53:53 +08:00
return sharedListenerManager.Start(nodeConfig)
}
this.isLoaded = true
2020-09-09 18:53:53 +08:00
return nil
}
2022-01-19 22:16:46 +08:00
// 读取单个服务配置
func (this *Node) syncServerConfig(serverId int64) error {
rpcClient, err := rpc.SharedRPC()
if err != nil {
return err
}
resp, err := rpcClient.ServerRPC().ComposeServerConfig(rpcClient.Context(), &pb.ComposeServerConfigRequest{ServerId: serverId})
if err != nil {
return err
}
this.locker.Lock()
defer this.locker.Unlock()
if len(resp.ServerConfigJSON) == 0 {
this.updatingServerMap[serverId] = nil
} else {
var config = &serverconfigs.ServerConfig{}
err = json.Unmarshal(resp.ServerConfigJSON, config)
if err != nil {
return err
}
this.updatingServerMap[serverId] = config
}
return nil
}
2020-09-09 18:53:53 +08:00
// 启动同步计时器
func (this *Node) startSyncTimer() {
2020-09-26 08:07:07 +08:00
// TODO 这个时间间隔可以自行设置
2022-01-19 22:16:46 +08:00
var taskTicker = time.NewTicker(60 * time.Second)
var serverChangeTicker = time.NewTicker(5 * time.Second)
2022-01-12 20:31:04 +08:00
events.OnKey(events.EventQuit, this, func() {
remotelogs.Println("NODE", "quit sync timer")
2022-01-19 22:16:46 +08:00
taskTicker.Stop()
serverChangeTicker.Stop()
2020-10-28 11:19:06 +08:00
})
goman.New(func() {
for {
select {
2022-01-19 22:16:46 +08:00
case <-taskTicker.C: // 定期执行
err := this.loop()
if err != nil {
remotelogs.Error("NODE", "sync config error: "+err.Error())
continue
}
2022-01-19 22:16:46 +08:00
case <-serverChangeTicker.C: // 服务变化
this.locker.Lock()
if len(this.updatingServerMap) > 0 {
var updatingServerMap = this.updatingServerMap
this.updatingServerMap = map[int64]*serverconfigs.ServerConfig{}
newNodeConfig, err := nodeconfigs.CloneNodeConfig(sharedNodeConfig)
if err != nil {
remotelogs.Error("NODE", "apply server config error: "+err.Error())
continue
}
for serverId, serverConfig := range updatingServerMap {
if serverConfig != nil {
newNodeConfig.AddServer(serverConfig)
} else {
newNodeConfig.RemoveServer(serverId)
}
}
err, serverErrors := newNodeConfig.Init()
if err != nil {
remotelogs.Error("NODE", "apply server config error: "+err.Error())
continue
}
if len(serverErrors) > 0 {
for _, serverErr := range serverErrors {
remotelogs.ServerError(serverErr.Id, "NODE", serverErr.Message, nodeconfigs.NodeLogTypeServerConfigInitFailed, maps.Map{})
}
}
this.onReload(newNodeConfig)
err = sharedListenerManager.Start(newNodeConfig)
if err != nil {
remotelogs.Error("NODE", "apply server config error: "+err.Error())
}
}
this.locker.Unlock()
case <-nodeTaskNotify: // 有新的更新任务
err := this.loop()
if err != nil {
remotelogs.Error("NODE", "sync config error: "+err.Error())
continue
}
2022-01-19 22:16:46 +08:00
case <-nodeConfigChangedNotify: // 节点变化通知
2021-11-11 14:16:57 +08:00
err := this.syncConfig(0)
if err != nil {
remotelogs.Error("NODE", "sync config error: "+err.Error())
continue
}
2020-09-09 18:53:53 +08:00
}
}
})
2020-09-09 18:53:53 +08:00
}
// 检查集群设置
func (this *Node) checkClusterConfig() error {
configFile := Tea.ConfigFile("cluster.yaml")
data, err := ioutil.ReadFile(configFile)
if err != nil {
return err
}
config := &configs.ClusterConfig{}
err = yaml.Unmarshal(data, config)
if err != nil {
return err
}
rpcClient, err := rpc.NewRPCClient(&configs.APIConfig{
RPC: config.RPC,
NodeId: config.ClusterId,
Secret: config.Secret,
})
if err != nil {
return err
}
logs.Println("[NODE]registering node to cluster ...")
resp, err := rpcClient.NodeRPC().RegisterClusterNode(rpcClient.ClusterContext(config.ClusterId, config.Secret), &pb.RegisterClusterNodeRequest{Name: HOSTNAME})
if err != nil {
return err
}
2021-01-11 18:16:15 +08:00
logs.Println("[NODE]registered successfully")
// 写入到配置文件中
if len(resp.Endpoints) == 0 {
resp.Endpoints = []string{}
}
apiConfig := &configs.APIConfig{
RPC: struct {
Endpoints []string `yaml:"endpoints"`
}{
Endpoints: resp.Endpoints,
},
NodeId: resp.UniqueId,
Secret: resp.Secret,
}
2021-01-11 18:16:15 +08:00
logs.Println("[NODE]writing 'configs/api.yaml' ...")
err = apiConfig.WriteFile(Tea.ConfigFile("api.yaml"))
if err != nil {
return err
}
2021-01-11 18:16:15 +08:00
logs.Println("[NODE]wrote 'configs/api.yaml' successfully")
return nil
}
// 监听本地sock
func (this *Node) listenSock() error {
2021-07-25 17:14:44 +08:00
// 检查是否在运行
if this.sock.IsListening() {
reply, err := this.sock.Send(&gosock.Command{Code: "pid"})
if err == nil {
return errors.New("error: the process is already running, pid: " + maps.NewMap(reply.Params).GetString("pid"))
} else {
2021-07-25 17:14:44 +08:00
return errors.New("error: the process is already running")
}
}
2021-07-25 17:14:44 +08:00
// 启动监听
goman.New(func() {
2021-07-25 17:14:44 +08:00
this.sock.OnCommand(func(cmd *gosock.Command) {
switch cmd.Code {
case "pid":
_ = cmd.Reply(&gosock.Command{
Code: "pid",
Params: map[string]interface{}{
"pid": os.Getpid(),
},
})
case "info":
exePath, _ := os.Executable()
_ = cmd.Reply(&gosock.Command{
Code: "info",
Params: map[string]interface{}{
"pid": os.Getpid(),
"version": teaconst.Version,
"path": exePath,
},
})
2021-07-25 17:14:44 +08:00
case "stop":
_ = cmd.ReplyOk()
// 退出主进程
events.Notify(events.EventQuit)
os.Exit(0)
case "quit":
_ = cmd.ReplyOk()
_ = this.sock.Close()
events.Notify(events.EventQuit)
// 监控连接数如果连接数为0则退出进程
goman.New(func() {
2021-07-25 17:14:44 +08:00
for {
countActiveConnections := sharedListenerManager.TotalActiveConnections()
if countActiveConnections <= 0 {
os.Exit(0)
return
}
time.Sleep(1 * time.Second)
}
})
case "trackers":
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
"labels": trackers.SharedManager.Labels(),
},
})
case "goman":
var posMap = map[string]maps.Map{} // file#line => Map
for _, instance := range goman.List() {
var pos = instance.File + "#" + types.String(instance.Line)
m, ok := posMap[pos]
if ok {
m["count"] = m["count"].(int) + 1
} else {
m = maps.Map{
"pos": pos,
"count": 1,
}
posMap[pos] = m
}
}
var result = []maps.Map{}
for _, m := range posMap {
result = append(result, m)
}
sort.Slice(result, func(i, j int) bool {
return result[i]["count"].(int) > result[j]["count"].(int)
})
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
2021-12-14 10:01:21 +08:00
"total": runtime.NumGoroutine(),
"result": result,
},
2021-12-12 11:48:01 +08:00
})
case "conns":
ipConns, serverConns := sharedClientConnLimiter.Conns()
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
"ipConns": ipConns,
"serverConns": serverConns,
"total": sharedListenerManager.TotalActiveConnections(),
2021-12-19 16:54:56 +08:00
"limiter": sharedConnectionsLimiter.Len(),
2021-12-12 11:48:01 +08:00
},
})
case "dropIP":
var m = maps.NewMap(cmd.Params)
var ip = m.GetString("ip")
var timeSeconds = m.GetInt("timeoutSeconds")
err := firewalls.Firewall().DropSourceIP(ip, timeSeconds)
if err != nil {
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
"error": err.Error(),
},
})
} else {
_ = cmd.ReplyOk()
}
case "rejectIP":
var m = maps.NewMap(cmd.Params)
var ip = m.GetString("ip")
var timeSeconds = m.GetInt("timeoutSeconds")
err := firewalls.Firewall().RejectSourceIP(ip, timeSeconds)
if err != nil {
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
"error": err.Error(),
},
})
} else {
_ = cmd.ReplyOk()
}
case "removeIP":
var m = maps.NewMap(cmd.Params)
var ip = m.GetString("ip")
err := firewalls.Firewall().RemoveSourceIP(ip)
if err != nil {
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
"error": err.Error(),
},
})
} else {
_ = cmd.ReplyOk()
}
case "gc":
runtime.GC()
debug.FreeOSMemory()
2022-01-01 17:18:34 +08:00
_ = cmd.ReplyOk()
case "reload":
err := this.syncConfig(0)
if err != nil {
_ = cmd.Reply(&gosock.Command{
Params: map[string]interface{}{
"error": err.Error(),
},
})
} else {
_ = cmd.ReplyOk()
}
}
2021-07-25 17:14:44 +08:00
})
err := this.sock.Listen()
if err != nil {
logs.Println("NODE", err.Error())
}
})
2022-01-12 20:31:04 +08:00
events.OnKey(events.EventQuit, this, func() {
2021-07-25 17:14:44 +08:00
logs.Println("NODE", "quit unix sock")
_ = this.sock.Close()
})
return nil
}
2021-12-09 12:07:59 +08:00
// 重载配置调用
func (this *Node) onReload(config *nodeconfigs.NodeConfig) {
2022-01-19 22:16:46 +08:00
nodeconfigs.ResetNodeConfig(config)
sharedNodeConfig = config
// 缓存策略
caches.SharedManager.MaxDiskCapacity = config.MaxCacheDiskCapacity
caches.SharedManager.MaxMemoryCapacity = config.MaxCacheMemoryCapacity
if len(config.HTTPCachePolicies) > 0 {
caches.SharedManager.UpdatePolicies(config.HTTPCachePolicies)
} else {
caches.SharedManager.UpdatePolicies([]*serverconfigs.HTTPCachePolicy{})
}
// WAF策略
sharedWAFManager.UpdatePolicies(config.FindAllFirewallPolicies())
iplibrary.SharedActionManager.UpdateActions(config.FirewallActions)
// 统计指标
metrics.SharedManager.Update(config.MetricItems)
2021-12-09 12:07:59 +08:00
// max cpu
2021-12-09 17:34:05 +08:00
if config.MaxCPU != this.maxCPU {
if config.MaxCPU > 0 && config.MaxCPU < int32(runtime.NumCPU()) {
runtime.GOMAXPROCS(int(config.MaxCPU))
remotelogs.Println("NODE", "[CPU]set max cpu to '"+types.String(config.MaxCPU)+"'")
} else {
2021-12-31 19:46:33 +08:00
var threads = runtime.NumCPU() * 4
runtime.GOMAXPROCS(threads)
remotelogs.Println("NODE", "[CPU]set max cpu to '"+types.String(threads)+"'")
2021-12-09 17:34:05 +08:00
}
this.maxCPU = config.MaxCPU
2021-12-09 12:07:59 +08:00
}
// max threads
2021-12-09 17:34:05 +08:00
if config.MaxThreads != this.maxThreads {
if config.MaxThreads > 0 {
debug.SetMaxThreads(config.MaxThreads)
remotelogs.Println("NODE", "[THREADS]set max threads to '"+types.String(config.MaxThreads)+"'")
} else {
debug.SetMaxThreads(nodeconfigs.DefaultMaxThreads)
remotelogs.Println("NODE", "[THREADS]set max threads to '"+types.String(nodeconfigs.DefaultMaxThreads)+"'")
}
this.maxThreads = config.MaxThreads
}
// max tcp connections
if config.TCPMaxConnections <= 0 {
config.TCPMaxConnections = nodeconfigs.DefaultTCPMaxConnections
}
if config.TCPMaxConnections != sharedConnectionsLimiter.Count() {
remotelogs.Println("NODE", "[TCP]changed tcp max connections to '"+types.String(config.TCPMaxConnections)+"'")
sharedConnectionsLimiter.Close()
sharedConnectionsLimiter = ratelimit.NewCounter(config.TCPMaxConnections)
2021-12-09 12:07:59 +08:00
}
// timezone
var timeZone = config.TimeZone
if len(timeZone) == 0 {
timeZone = "Asia/Shanghai"
}
if this.timezone != timeZone {
location, err := time.LoadLocation(timeZone)
if err != nil {
remotelogs.Error("NODE", "[TIMEZONE]change time zone failed: "+err.Error())
return
}
remotelogs.Println("NODE", "[TIMEZONE]change time zone to '"+timeZone+"'")
time.Local = location
this.timezone = timeZone
}
2022-01-09 10:53:21 +08:00
// product information
if config.ProductConfig != nil {
teaconst.GlobalProductName = config.ProductConfig.Name
}
2021-12-09 12:07:59 +08:00
}