mirrors/dae
1
0
Fork 0
mirror of https://github.com/daeuniverse/dae.git synced 2025-07-04 07:17:55 +07:00
Code Issues Packages Projects Releases Wiki Activity

fix: rollback reloading and refresh fixed outbound alive state

Browse Source
This commit is contained in:
mzz2017
2023-03-23 15:34:56 +08:00
parent 7adbb2fbb9
commit b69cb63a11
5 changed files with 101 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
cmd/run.go
View File

@ -186,6 +186,7 @@ loop:
"err": err,
}).Fatalln("[Reload] Failed to roll back configuration")
}
newConf = conf
log.Warnln("[Reload] Last reload failed; rolled back configuration")
} else {
log.Warnln("[Reload] Stopped old control plane")

102
component/outbound/dialer_group.go
View File

@ -7,10 +7,10 @@ package outbound
import (
"fmt"
"github.com/mzz2017/softwind/netproxy"
"github.com/sirupsen/logrus"
"github.com/daeuniverse/dae/common/consts"
"github.com/daeuniverse/dae/component/outbound/dialer"
"github.com/mzz2017/softwind/netproxy"
"github.com/sirupsen/logrus"
"time"
)
@ -38,84 +38,96 @@ func NewDialerGroup(option *dialer.GlobalOption, name string, dialers []*dialer.
var aliveDnsUdp4DialerSet *dialer.AliveDialerSet
var aliveDnsUdp6DialerSet *dialer.AliveDialerSet
var needAliveState bool
switch p.Policy {
case consts.DialerSelectionPolicy_Random,
consts.DialerSelectionPolicy_MinLastLatency,
consts.DialerSelectionPolicy_MinAverage10Latencies,
consts.DialerSelectionPolicy_MinMovingAverageLatencies:
// Need to know the alive state or latency.
networkType := &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_4,
IsDns: false,
}
needAliveState = true
case consts.DialerSelectionPolicy_Fixed:
// No need to know if the dialer is alive.
needAliveState = false
default:
log.Panicf("Unexpected dialer selection policy: %v", p.Policy)
}
networkType := &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_4,
IsDns: false,
}
if needAliveState {
aliveTcp4DialerSet = dialer.NewAliveDialerSet(
log, name, networkType, option.CheckTolerance, p.Policy, dialers,
func(networkType *dialer.NetworkType) func(alive bool) {
// Use the trick to copy a pointer of *dialer.NetworkType.
return func(alive bool) { aliveChangeCallback(alive, networkType, false) }
}(networkType), true)
aliveChangeCallback(true, networkType, true)
}
aliveChangeCallback(true, networkType, true)
networkType = &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_6,
IsDns: false,
}
networkType = &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_6,
IsDns: false,
}
if needAliveState {
aliveTcp6DialerSet = dialer.NewAliveDialerSet(
log, name, networkType, option.CheckTolerance, p.Policy, dialers,
func(networkType *dialer.NetworkType) func(alive bool) {
// Use the trick to copy a pointer of *dialer.NetworkType.
return func(alive bool) { aliveChangeCallback(alive, networkType, false) }
}(networkType), true)
aliveChangeCallback(true, networkType, true)
}
aliveChangeCallback(true, networkType, true)
networkType = &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_UDP,
IpVersion: consts.IpVersionStr_4,
IsDns: true,
}
networkType = &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_UDP,
IpVersion: consts.IpVersionStr_4,
IsDns: true,
}
if needAliveState {
aliveDnsUdp4DialerSet = dialer.NewAliveDialerSet(
log, name, networkType, option.CheckTolerance, p.Policy, dialers,
func(networkType *dialer.NetworkType) func(alive bool) {
// Use the trick to copy a pointer of *dialer.NetworkType.
return func(alive bool) { aliveChangeCallback(alive, networkType, false) }
}(networkType), true)
aliveChangeCallback(true, networkType, true)
}
aliveChangeCallback(true, networkType, true)
networkType = &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_UDP,
IpVersion: consts.IpVersionStr_6,
IsDns: true,
}
networkType = &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_UDP,
IpVersion: consts.IpVersionStr_6,
IsDns: true,
}
if needAliveState {
aliveDnsUdp6DialerSet = dialer.NewAliveDialerSet(
log, name, networkType, option.CheckTolerance, p.Policy, dialers,
func(networkType *dialer.NetworkType) func(alive bool) {
// Use the trick to copy a pointer of *dialer.NetworkType.
return func(alive bool) { aliveChangeCallback(alive, networkType, false) }
}(networkType), true)
aliveChangeCallback(true, networkType, true)
}
aliveChangeCallback(true, networkType, true)
if option.CheckDnsTcp {
aliveDnsTcp4DialerSet = dialer.NewAliveDialerSet(log, name, &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_4,
IsDns: true,
}, option.CheckTolerance, p.Policy, dialers, func(alive bool) {}, true)
if option.CheckDnsTcp && needAliveState {
aliveDnsTcp4DialerSet = dialer.NewAliveDialerSet(log, name, &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_4,
IsDns: true,
}, option.CheckTolerance, p.Policy, dialers, func(alive bool) {}, true)
aliveDnsTcp6DialerSet = dialer.NewAliveDialerSet(log, name, &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_6,
IsDns: true,
}, option.CheckTolerance, p.Policy, dialers, func(alive bool) {}, true)
}
case consts.DialerSelectionPolicy_Fixed:
// No need to know if the dialer is alive.
default:
log.Panicf("Unexpected dialer selection policy: %v", p.Policy)
aliveDnsTcp6DialerSet = dialer.NewAliveDialerSet(log, name, &dialer.NetworkType{
L4Proto: consts.L4ProtoStr_TCP,
IpVersion: consts.IpVersionStr_6,
IsDns: true,
}, option.CheckTolerance, p.Policy, dialers, func(alive bool) {}, true)
}
for _, d := range dialers {

26
control/connectivity.go
View File

@ -7,8 +7,8 @@ package control
import (
"github.com/cilium/ebpf"
"github.com/sirupsen/logrus"
"github.com/daeuniverse/dae/component/outbound/dialer"
"github.com/sirupsen/logrus"
"golang.org/x/sys/unix"
"strconv"
)
@ -25,11 +25,17 @@ func FormatL4Proto(l4proto uint8) string {
func (c *controlPlaneCore) OutboundAliveChangeCallback(outbound uint8) func(alive bool, networkType *dialer.NetworkType, isInit bool) {
return func(alive bool, networkType *dialer.NetworkType, isInit bool) {
if !isInit {
select {
case <-c.closed.Done():
return
default:
}
if !isInit || c.log.IsLevelEnabled(logrus.TraceLevel) {
c.log.WithFields(logrus.Fields{
"alive": alive,
"network": networkType.StringWithoutDns(),
"outbound": c.outboundId2Name[outbound],
"alive": alive,
"network": networkType.StringWithoutDns(),
"outboundId": outbound,
"outbound": c.outboundId2Name[outbound],
}).Warnf("Outbound alive state changed, notify the kernel program.")
}
@ -37,10 +43,16 @@ func (c *controlPlaneCore) OutboundAliveChangeCallback(outbound uint8) func(aliv
if alive {
value = 1
}
_ = c.bpf.OutboundConnectivityMap.Update(bpfOutboundConnectivityQuery{
if err := c.bpf.OutboundConnectivityMap.Update(bpfOutboundConnectivityQuery{
Outbound: outbound,
L4proto: networkType.L4Proto.ToL4Proto(),
Ipversion: networkType.IpVersion.ToIpVersion(),
}, value, ebpf.UpdateAny)
}, value, ebpf.UpdateAny); err != nil {
c.log.WithFields(logrus.Fields{
"alive": alive,
"network": networkType.StringWithoutDns(),
"outbound": c.outboundId2Name[outbound],
}).Warnf("Failed to notify the kernel program: %v", err)
}
}
}

20
control/control_plane_core.go
View File

@ -6,15 +6,16 @@
package control
import (
"context"
"fmt"
"github.com/cilium/ebpf"
ciliumLink "github.com/cilium/ebpf/link"
"github.com/mohae/deepcopy"
"github.com/safchain/ethtool"
"github.com/sirupsen/logrus"
"github.com/daeuniverse/dae/common"
"github.com/daeuniverse/dae/common/consts"
internal "github.com/daeuniverse/dae/pkg/ebpf_internal"
"github.com/mohae/deepcopy"
"github.com/safchain/ethtool"
"github.com/sirupsen/logrus"
"github.com/vishvananda/netlink"
"golang.org/x/net/dns/dnsmessage"
"golang.org/x/sys/unix"
@ -38,6 +39,9 @@ type controlPlaneCore struct {
flip int
isReload bool
bpfEjected bool
closed context.Context
close context.CancelFunc
}
func newControlPlaneCore(log *logrus.Logger,
@ -53,6 +57,7 @@ func newControlPlaneCore(log *logrus.Logger,
if !isReload {
deferFuncs = append(deferFuncs, bpf.Close)
}
closed, toClose := context.WithCancel(context.Background())
return &controlPlaneCore{
log: log,
deferFuncs: deferFuncs,
@ -61,6 +66,9 @@ func newControlPlaneCore(log *logrus.Logger,
kernelVersion: kernelVersion,
flip: coreFlip,
isReload: isReload,
bpfEjected: false,
closed: closed,
close: toClose,
}
}
@ -68,6 +76,11 @@ func (c *controlPlaneCore) Flip() {
coreFlip = coreFlip&1 ^ 1
}
func (c *controlPlaneCore) Close() (err error) {
select {
case <-c.closed.Done():
return nil
default:
}
// Invoke defer funcs in reverse order.
for i := len(c.deferFuncs) - 1; i >= 0; i-- {
if e := c.deferFuncs[i](); e != nil {
@ -79,6 +92,7 @@ func (c *controlPlaneCore) Close() (err error) {
}
}
}
c.close()
return err
}

8
control/dns_control.go
View File

@ -289,6 +289,12 @@ type dialArgument struct {
}
func (c *DnsController) Handle_(dnsMessage *dnsmessage.Message, req *udpRequest) (err error) {
if c.log.IsLevelEnabled(logrus.TraceLevel) && len(dnsMessage.Questions) > 0 {
q := dnsMessage.Questions[0]
c.log.Tracef("Received UDP(DNS) %v <-> %v: %v %v",
RefineSourceToShow(req.realSrc, req.realDst.Addr(), req.lanWanFlag), req.realDst.String(), strings.ToLower(q.Name.String()), q.Type,
)
}
if resp := c.LookupDnsRespCache_(dnsMessage); resp != nil {
// Send cache to client directly.
if err = sendPkt(resp, req.realDst, req.realSrc, req.src, req.lConn, req.lanWanFlag); err != nil {
@ -296,7 +302,7 @@ func (c *DnsController) Handle_(dnsMessage *dnsmessage.Message, req *udpRequest)
}
if c.log.IsLevelEnabled(logrus.DebugLevel) && len(dnsMessage.Questions) > 0 {
q := dnsMessage.Questions[0]
c.log.Tracef("UDP(DNS) %v <-> Cache: %v %v",
c.log.Debugf("UDP(DNS) %v <-> Cache: %v %v",
RefineSourceToShow(req.realSrc, req.realDst.Addr(), req.lanWanFlag), strings.ToLower(q.Name.String()), q.Type,
)
}