package node import ( "time" "github.com/InazumaV/V2bX/api/panel" "github.com/InazumaV/V2bX/common/task" vCore "github.com/InazumaV/V2bX/core" "github.com/InazumaV/V2bX/limiter" log "github.com/sirupsen/logrus" ) func (c *Controller) startTasks(node *panel.NodeInfo) { // fetch node info task c.nodeInfoMonitorPeriodic = &task.Task{ Interval: node.PullInterval, Execute: c.nodeInfoMonitor, } // fetch user list task c.userReportPeriodic = &task.Task{ Interval: node.PushInterval, Execute: c.reportUserTrafficTask, } log.WithField("tag", c.tag).Info("Start monitor node status") // delay to start nodeInfoMonitor _ = c.nodeInfoMonitorPeriodic.Start(false) log.WithField("tag", c.tag).Info("Start report node status") _ = c.userReportPeriodic.Start(false) if node.Tls { switch c.CertConfig.CertMode { case "reality", "none", "": default: c.renewCertPeriodic = &task.Task{ Interval: time.Hour * 24, Execute: c.renewCertTask, } log.WithField("tag", c.tag).Info("Start renew cert") // delay to start renewCert _ = c.renewCertPeriodic.Start(true) } } if c.LimitConfig.EnableDynamicSpeedLimit { c.traffic = make(map[string]int64) c.dynamicSpeedLimitPeriodic = &task.Task{ Interval: time.Duration(c.LimitConfig.DynamicSpeedLimitConfig.Periodic) * time.Second, Execute: c.SpeedChecker, } } } func (c *Controller) nodeInfoMonitor() (err error) { // get node info newNodeInfo, err := c.apiClient.GetNodeInfo() if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Get node info failed") return nil } // get user info newUserInfo, err := c.apiClient.GetUserList() if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Get user list failed") return nil } if newNodeInfo != nil { c.info = newNodeInfo // nodeInfo changed if newUserInfo != nil { c.userList = newUserInfo } c.traffic = make(map[string]int64) // Remove old tag log.WithField("tag", c.tag).Info("Node changed, reload") err = c.server.DelNode(c.tag) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Delete node failed") return nil } // Remove Old limiter limiter.DeleteLimiter(c.tag) // Add new Limiter c.tag = c.buildNodeTag(newNodeInfo) l := limiter.AddLimiter(c.tag, &c.LimitConfig, c.userList) // check cert if newNodeInfo.Tls || newNodeInfo.Type == "hysteria" { err = c.requestCert() if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Request cert failed") return nil } } // add new node err = c.server.AddNode(c.tag, newNodeInfo, c.Options) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Add node failed") return nil } _, err = c.server.AddUsers(&vCore.AddUsersParams{ Tag: c.tag, Config: c.Options, UserInfo: c.userList, NodeInfo: newNodeInfo, }) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Add users failed") return nil } err = l.UpdateRule(&newNodeInfo.Rules) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Update Rule failed") return nil } c.limiter = l // Check interval if c.nodeInfoMonitorPeriodic.Interval != newNodeInfo.PullInterval && newNodeInfo.PullInterval != 0 { c.nodeInfoMonitorPeriodic.Interval = newNodeInfo.PullInterval c.nodeInfoMonitorPeriodic.Close() _ = c.nodeInfoMonitorPeriodic.Start(false) } if c.userReportPeriodic.Interval != newNodeInfo.PushInterval && newNodeInfo.PushInterval != 0 { c.userReportPeriodic.Interval = newNodeInfo.PullInterval c.userReportPeriodic.Close() _ = c.userReportPeriodic.Start(false) } log.WithField("tag", c.tag).Infof("Added %d new users", len(c.userList)) // exit return nil } // node no changed, check users if len(newUserInfo) == 0 { return nil } deleted, added := compareUserList(c.userList, newUserInfo) if len(deleted) > 0 { // have deleted users err = c.server.DelUsers(deleted, c.tag) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Delete users failed") return nil } } if len(added) > 0 { // have added users _, err = c.server.AddUsers(&vCore.AddUsersParams{ Tag: c.tag, Config: c.Options, NodeInfo: c.info, UserInfo: added, }) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("Add users failed") return nil } } if len(added) > 0 || len(deleted) > 0 { // update Limiter c.limiter.UpdateUser(c.tag, added, deleted) if err != nil { log.WithFields(log.Fields{ "tag": c.tag, "err": err, }).Error("limiter users failed") return nil } // clear traffic record if c.LimitConfig.EnableDynamicSpeedLimit { for i := range deleted { delete(c.traffic, deleted[i].Uuid) } } } c.userList = newUserInfo if len(added)+len(deleted) != 0 { log.WithField("tag", c.tag). Infof("%d user deleted, %d user added", len(deleted), len(added)) } return nil } func (c *Controller) SpeedChecker() error { for u, t := range c.traffic { if t >= c.LimitConfig.DynamicSpeedLimitConfig.Traffic { err := c.limiter.UpdateDynamicSpeedLimit(c.tag, u, c.LimitConfig.DynamicSpeedLimitConfig.SpeedLimit, time.Now().Add(time.Duration(c.LimitConfig.DynamicSpeedLimitConfig.ExpireTime)*time.Minute)) log.WithField("err", err).Error("Update dynamic speed limit failed") delete(c.traffic, u) } } return nil }