✨ 优化忽略规则配置和 Agent 获取 IP

2026-03-23 03:21:50 +00:00 · 2021-06-21 21:30:42 +08:00
parent c4f36d17d5
commit 4b0c0ad288
20 changed files with 370 additions and 291 deletions
--- a/cmd/agent/monitor/myip.go
+++ b/cmd/agent/monitor/myip.go
@@ -1,18 +1,13 @@
 package monitor

 import (
-	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io/ioutil"
-	"net"
 	"net/http"
-	"strings"
-	"sync"
 	"time"

-	"github.com/miekg/dns"
+	"github.com/naiba/nezha/pkg/utils"
 )

 type geoIP struct {
@@ -24,14 +19,16 @@ var (
 	ipv4Servers = []string{
 		"https://api-ipv4.ip.sb/geoip",
 		"https://ip4.seeip.org/geoip",
+		"https://ipapi.co/json",
 	}
 	ipv6Servers = []string{
 		"https://ip6.seeip.org/geoip",
 		"https://api-ipv6.ip.sb/geoip",
+		"https://ipapi.co/json",
 	}
 	cachedIP, cachedCountry string
-	httpClientV4            = newHTTPClient(time.Second*20, time.Second*5, time.Second*10, false)
-	httpClientV6            = newHTTPClient(time.Second*20, time.Second*5, time.Second*10, true)
+	httpClientV4            = utils.NewSingleStackHTTPClient(time.Second*20, time.Second*5, time.Second*10, false)
+	httpClientV6            = utils.NewSingleStackHTTPClient(time.Second*20, time.Second*5, time.Second*10, true)
 )

 func UpdateIP() {
@@ -73,93 +70,3 @@ func fetchGeoIP(servers []string, isV6 bool) geoIP {
 	}
 	return ip
 }
-
-func newHTTPClient(httpTimeout, dialTimeout, keepAliveTimeout time.Duration, ipv6 bool) *http.Client {
-	dialer := &net.Dialer{
-		Timeout:   dialTimeout,
-		KeepAlive: keepAliveTimeout,
-	}
-
-	transport := &http.Transport{
-		Proxy:             http.ProxyFromEnvironment,
-		ForceAttemptHTTP2: false,
-		DialContext: func(ctx context.Context, network string, addr string) (net.Conn, error) {
-			ip, err := resolveIP(addr, ipv6)
-			if err != nil {
-				return nil, err
-			}
-			return dialer.DialContext(ctx, network, ip)
-		},
-	}
-
-	return &http.Client{
-		Transport: transport,
-		Timeout:   httpTimeout,
-	}
-}
-
-func resolveIP(addr string, ipv6 bool) (string, error) {
-	url := strings.Split(addr, ":")
-
-	m := new(dns.Msg)
-	if ipv6 {
-		m.SetQuestion(dns.Fqdn(url[0]), dns.TypeAAAA)
-	} else {
-		m.SetQuestion(dns.Fqdn(url[0]), dns.TypeA)
-	}
-	m.RecursionDesired = true
-
-	dnsServers := []string{"2606:4700:4700::1001", "2001:4860:4860::8844"}
-	if !ipv6 {
-		dnsServers = []string{"1.0.0.1", "8.8.4.4"}
-	}
-
-	var wg sync.WaitGroup
-	var resolveLock sync.RWMutex
-	var ipv4Resolved, ipv6Resolved bool
-
-	wg.Add(len(dnsServers))
-	for i := 0; i < len(dnsServers); i++ {
-		go func(i int) {
-			defer wg.Done()
-			c := new(dns.Client)
-			c.Timeout = time.Second * 3
-			r, _, err := c.Exchange(m, net.JoinHostPort(dnsServers[i], "53"))
-			if err != nil {
-				return
-			}
-			resolveLock.Lock()
-			defer resolveLock.Unlock()
-			if ipv6 && ipv6Resolved {
-				return
-			}
-			if !ipv6 && ipv4Resolved {
-				return
-			}
-			for _, ans := range r.Answer {
-				if ipv6 {
-					if aaaa, ok := ans.(*dns.AAAA); ok {
-						url[0] = "[" + aaaa.AAAA.String() + "]"
-						ipv6Resolved = true
-					}
-				} else {
-					if a, ok := ans.(*dns.A); ok {
-						url[0] = a.A.String()
-						ipv4Resolved = true
-					}
-				}
-			}
-		}(i)
-	}
-	wg.Wait()
-
-	if ipv6 && !ipv6Resolved {
-		return "", errors.New("the AAAA record not resolved")
-	}
-
-	if !ipv6 && !ipv4Resolved {
-		return "", errors.New("the A record not resolved")
-	}
-
-	return strings.Join(url, ":"), nil
-}
--- a/cmd/dashboard/controller/member_api.go
+++ b/cmd/dashboard/controller/member_api.go
@@ -15,7 +15,6 @@ import (
 	"github.com/naiba/nezha/model"
 	"github.com/naiba/nezha/pkg/mygin"
 	"github.com/naiba/nezha/pkg/utils"
-	pb "github.com/naiba/nezha/proto"
 	"github.com/naiba/nezha/service/dao"
 )

@@ -196,6 +195,7 @@ type monitorForm struct {
 	Name           string
 	Target         string
 	Type           uint8
+	Cover          uint8
 	Notify         string
 	SkipServersRaw string
 }
@@ -210,6 +210,7 @@ func (ma *memberAPI) addOrEditMonitor(c *gin.Context) {
 		m.Type = mf.Type
 		m.ID = mf.ID
 		m.SkipServersRaw = mf.SkipServersRaw
+		m.Cover = mf.Cover
 		m.Notify = mf.Notify == "on"
 	}
 	if err == nil {
@@ -239,6 +240,7 @@ type cronForm struct {
 	Scheduler      string
 	Command        string
 	ServersRaw     string
+	Cover          uint8
 	PushSuccessful string
 }

@@ -253,6 +255,7 @@ func (ma *memberAPI) addOrEditCron(c *gin.Context) {
 		cr.ServersRaw = cf.ServersRaw
 		cr.PushSuccessful = cf.PushSuccessful == "on"
 		cr.ID = cf.ID
+		cr.Cover = cf.Cover
 		err = json.Unmarshal([]byte(cf.ServersRaw), &cr.Servers)
 	}
 	if err == nil {
@@ -281,21 +284,7 @@ func (ma *memberAPI) addOrEditCron(c *gin.Context) {
 		dao.Cron.Remove(crOld.CronID)
 	}

-	cr.CronID, err = dao.Cron.AddFunc(cr.Scheduler, func() {
-		dao.ServerLock.RLock()
-		defer dao.ServerLock.RUnlock()
-		for j := 0; j < len(cr.Servers); j++ {
-			if dao.ServerList[cr.Servers[j]].TaskStream != nil {
-				dao.ServerList[cr.Servers[j]].TaskStream.Send(&pb.Task{
-					Id:   cr.ID,
-					Data: cr.Command,
-					Type: model.TaskTypeCommand,
-				})
-			} else {
-				dao.SendNotification(fmt.Sprintf("计划任务：%s，服务器：%s 离线，无法执行。", cr.Name, dao.ServerList[cr.Servers[j]].Name), false)
-			}
-		}
-	})
+	cr.CronID, err = dao.Cron.AddFunc(cr.Scheduler, dao.CronTrigger(cr))
 	if err != nil {
 		panic(err)
 	}
@@ -318,7 +307,7 @@ func (ma *memberAPI) manualTrigger(c *gin.Context) {
 		return
 	}

-	dao.CronTrigger(&cr)
+	dao.ManualTrigger(&cr)

 	c.JSON(http.StatusOK, model.Response{
 		Code: http.StatusOK,
--- a/cmd/dashboard/main.go
+++ b/cmd/dashboard/main.go
@@ -1,7 +1,6 @@
 package main

 import (
-	"fmt"
 	"time"

 	"github.com/patrickmn/go-cache"
@@ -12,7 +11,6 @@ import (
 	"github.com/naiba/nezha/cmd/dashboard/controller"
 	"github.com/naiba/nezha/cmd/dashboard/rpc"
 	"github.com/naiba/nezha/model"
-	pb "github.com/naiba/nezha/proto"
 	"github.com/naiba/nezha/service/dao"
 )

@@ -84,21 +82,13 @@ func loadCrons() {
 	var err error
 	for i := 0; i < len(crons); i++ {
 		cr := crons[i]
-		cr.CronID, err = dao.Cron.AddFunc(cr.Scheduler, func() {
-			dao.ServerLock.RLock()
-			defer dao.ServerLock.RUnlock()
-			for j := 0; j < len(cr.Servers); j++ {
-				if dao.ServerList[cr.Servers[j]].TaskStream != nil {
-					dao.ServerList[cr.Servers[j]].TaskStream.Send(&pb.Task{
-						Id:   cr.ID,
-						Data: cr.Command,
-						Type: model.TaskTypeCommand,
-					})
-				} else {
-					dao.SendNotification(fmt.Sprintf("计划任务：%s，服务器：%s 离线，无法执行。", cr.Name, dao.ServerList[cr.Servers[j]].Name), false)
-				}
-			}
-		})
+
+		crIgnoreMap := make(map[uint64]bool)
+		for j := 0; j < len(cr.Servers); j++ {
+			crIgnoreMap[cr.Servers[j]] = true
+		}
+
+		cr.CronID, err = dao.Cron.AddFunc(cr.Scheduler, dao.CronTrigger(cr))
 		if err != nil {
 			panic(err)
 		}
--- a/cmd/dashboard/rpc/rpc.go
+++ b/cmd/dashboard/rpc/rpc.go
@@ -7,6 +7,7 @@ import (

 	"google.golang.org/grpc"

+	"github.com/naiba/nezha/model"
 	pb "github.com/naiba/nezha/proto"
 	"github.com/naiba/nezha/service/dao"
 	rpcService "github.com/naiba/nezha/service/rpc"
@@ -39,13 +40,21 @@ func DispatchTask(duration time.Duration) {
 				}
 				hasAliveAgent = false
 			}
-			// 1. 如果此任务不可使用此服务器请求，跳过这个服务器（有些 IPv6 only 开了 NAT64 的机器请求 IPv4 总会出问题）
-			// 2. 如果服务器不在线，跳过这个服务器
-			if tasks[i].SkipServers[dao.SortedServerList[index].ID] || dao.SortedServerList[index].TaskStream == nil {
+
+			// 1. 如果服务器不在线，跳过这个服务器
+			if dao.SortedServerList[index].TaskStream == nil {
 				i--
 				index++
 				continue
 			}
+			// 2. 如果此任务不可使用此服务器请求，跳过这个服务器（有些 IPv6 only 开了 NAT64 的机器请求 IPv4 总会出问题）
+			if (tasks[i].Cover == model.MonitorCoverAll && tasks[i].SkipServers[dao.SortedServerList[index].ID]) ||
+				(tasks[i].Cover == model.MonitorCoverIgnoreAll && !tasks[i].SkipServers[dao.SortedServerList[index].ID]) {
+				i--
+				index++
+				continue
+			}
+
 			hasAliveAgent = true
 			dao.SortedServerList[index].TaskStream.Send(tasks[i].PB())
 			index++