mirror of
https://github.com/openziti/ziti.git
synced 2026-10-11 23:12:35 +00:00
Merge pull request #1521 from openziti/dont-stop-router-on-dns-failure
Don't terminate router on DNS failure. Fixes #1513
This commit is contained in:
@@ -19,12 +19,12 @@ package xgress_edge_tunnel
|
||||
import (
|
||||
"github.com/michaelquigley/pfxlog"
|
||||
"github.com/openziti/ziti/router/fabric"
|
||||
"github.com/openziti/ziti/router/xgress"
|
||||
"github.com/openziti/ziti/tunnel/dns"
|
||||
"github.com/openziti/ziti/tunnel/intercept"
|
||||
"github.com/openziti/ziti/tunnel/intercept/host"
|
||||
"github.com/openziti/ziti/tunnel/intercept/proxy"
|
||||
"github.com/openziti/ziti/tunnel/intercept/tproxy"
|
||||
"github.com/openziti/ziti/router/xgress"
|
||||
cmap "github.com/orcaman/concurrent-map/v2"
|
||||
"github.com/pkg/errors"
|
||||
"math"
|
||||
@@ -96,7 +96,11 @@ func (self *tunneler) Start(notifyClose <-chan struct{}) error {
|
||||
return errors.Errorf("unsupported tunnel mode '%v'", self.listenOptions.mode)
|
||||
}
|
||||
|
||||
resolver := dns.NewResolver(self.listenOptions.resolver)
|
||||
resolver, err := dns.NewResolver(self.listenOptions.resolver)
|
||||
if err != nil {
|
||||
pfxlog.Logger().WithError(err).Error("failed to start DNS resolver")
|
||||
}
|
||||
|
||||
if err = intercept.SetDnsInterceptIpRange(self.listenOptions.dnsSvcIpRange); err != nil {
|
||||
pfxlog.Logger().Errorf("invalid dns service IP range %s: %v", self.listenOptions.dnsSvcIpRange, err)
|
||||
return err
|
||||
|
||||
+15
-12
@@ -60,29 +60,32 @@ func flushDnsCaches() {
|
||||
}
|
||||
}
|
||||
|
||||
func NewResolver(config string) Resolver {
|
||||
func NewResolver(config string) (Resolver, error) {
|
||||
flushDnsCaches()
|
||||
if config == "" {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
resolverURL, err := url.Parse(config)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to parse resolver configuration '%s': %s", config, err)
|
||||
return nil, fmt.Errorf("failed to parse resolver configuration '%s': %w", config, err)
|
||||
}
|
||||
|
||||
switch resolverURL.Scheme {
|
||||
case "", "file":
|
||||
return NewRefCountingResolver(NewHostFile(resolverURL.Path))
|
||||
return NewRefCountingResolver(NewHostFile(resolverURL.Path)), nil
|
||||
case "udp":
|
||||
return NewRefCountingResolver(NewDnsServer(resolverURL.Host))
|
||||
dnsResolver, err := NewDnsServer(resolverURL.Host)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return NewRefCountingResolver(dnsResolver), nil
|
||||
}
|
||||
|
||||
log.Fatalf("invalid resolver configuration '%s'. must be 'file://' or 'udp://' URL", config)
|
||||
return nil
|
||||
return nil, fmt.Errorf("invalid resolver configuration '%s'. must be 'file://' or 'udp://' URL", config)
|
||||
}
|
||||
|
||||
func NewDnsServer(addr string) Resolver {
|
||||
func NewDnsServer(addr string) (Resolver, error) {
|
||||
log.Infof("starting dns server...")
|
||||
s := &dns.Server{
|
||||
Addr: addr,
|
||||
@@ -108,9 +111,9 @@ func NewDnsServer(addr string) Resolver {
|
||||
select {
|
||||
case err := <-errChan:
|
||||
if err != nil {
|
||||
log.Fatalf("dns server failed to start: %s", err)
|
||||
return nil, fmt.Errorf("dns server failed to start: %w", err)
|
||||
} else {
|
||||
log.Fatal("dns server stopped prematurely")
|
||||
return nil, fmt.Errorf("dns server stopped prematurely")
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
log.Infof("dns server running at %s", s.Addr)
|
||||
@@ -125,10 +128,10 @@ func NewDnsServer(addr string) Resolver {
|
||||
err := r.testSystemResolver()
|
||||
if err != nil {
|
||||
_ = r.Cleanup()
|
||||
log.Fatalf("system resolver test failed: %s\n\n"+resolverConfigHelp, err, addr)
|
||||
return nil, fmt.Errorf("system resolver test failed: %s\n\n"+resolverConfigHelp, err, addr)
|
||||
}
|
||||
|
||||
return r
|
||||
return r, nil
|
||||
}
|
||||
|
||||
func (r *resolver) testSystemResolver() error {
|
||||
|
||||
@@ -17,15 +17,14 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/michaelquigley/pfxlog"
|
||||
"github.com/openziti/foundation/v2/info"
|
||||
"github.com/openziti/foundation/v2/mempool"
|
||||
"github.com/openziti/ziti/tunnel"
|
||||
"github.com/openziti/ziti/tunnel/dns"
|
||||
"github.com/openziti/ziti/tunnel/entities"
|
||||
"github.com/openziti/ziti/tunnel/intercept"
|
||||
"github.com/openziti/ziti/tunnel/udp_vconn"
|
||||
"github.com/openziti/foundation/v2/info"
|
||||
"github.com/openziti/foundation/v2/mempool"
|
||||
"github.com/pkg/errors"
|
||||
"io"
|
||||
"net"
|
||||
@@ -61,7 +60,6 @@ func (self *Service) Stop() error {
|
||||
type interceptor struct {
|
||||
interceptIP net.IP
|
||||
services map[string]*Service
|
||||
closeCh chan interface{}
|
||||
}
|
||||
|
||||
func New(ip net.IP, serviceList []string) (intercept.Interceptor, error) {
|
||||
@@ -98,7 +96,6 @@ func New(ip net.IP, serviceList []string) (intercept.Interceptor, error) {
|
||||
p := interceptor{
|
||||
interceptIP: ip,
|
||||
services: services,
|
||||
closeCh: make(chan interface{}),
|
||||
}
|
||||
return &p, nil
|
||||
}
|
||||
@@ -117,27 +114,23 @@ func (p *interceptor) Intercept(service *entities.Service, _ dns.Resolver, _ int
|
||||
// pre-fetch network session todo move this to service poller?
|
||||
service.FabricProvider.PrepForUse(*service.ID)
|
||||
|
||||
go p.runServiceListener(proxiedService)
|
||||
return nil
|
||||
return p.runServiceListener(proxiedService)
|
||||
}
|
||||
|
||||
func (p *interceptor) runServiceListener(service *Service) {
|
||||
func (p *interceptor) runServiceListener(service *Service) error {
|
||||
if service.Protocol == intercept.TCP {
|
||||
p.handleTCP(service)
|
||||
} else {
|
||||
p.handleUDP(service)
|
||||
return p.handleTCP(service)
|
||||
}
|
||||
return p.handleUDP(service)
|
||||
}
|
||||
|
||||
func (p *interceptor) handleTCP(service *Service) {
|
||||
func (p *interceptor) handleTCP(service *Service) error {
|
||||
log := pfxlog.Logger().WithField("service", service.Name)
|
||||
|
||||
listenAddr := net.TCPAddr{IP: p.interceptIP, Port: service.Port}
|
||||
server, err := net.Listen("tcp4", listenAddr.String())
|
||||
if err != nil {
|
||||
log.Fatalln(err)
|
||||
p.closeCh <- err
|
||||
return
|
||||
return err
|
||||
}
|
||||
service.setCloser(server)
|
||||
|
||||
@@ -145,33 +138,31 @@ func (p *interceptor) handleTCP(service *Service) {
|
||||
|
||||
log.Info("service is listening")
|
||||
defer log.Info("service stopped")
|
||||
defer func() {
|
||||
p.closeCh <- fmt.Sprintf("service listener %s exited", service.Name)
|
||||
|
||||
go func() {
|
||||
for {
|
||||
conn, err := server.Accept()
|
||||
if err != nil {
|
||||
log.WithError(err).Error("accept failed")
|
||||
return
|
||||
}
|
||||
sourceAddr := service.TunnelService.GetSourceAddr(conn.RemoteAddr(), conn.LocalAddr())
|
||||
appInfo := tunnel.GetAppInfo("tcp", "", p.interceptIP.String(), strconv.Itoa(service.Port), sourceAddr)
|
||||
identity := service.TunnelService.GetDialIdentity(conn.RemoteAddr(), conn.LocalAddr())
|
||||
go tunnel.DialAndRun(service.TunnelService, identity, conn, appInfo, true)
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
conn, err := server.Accept()
|
||||
if err != nil {
|
||||
log.WithError(err).Error("accept failed")
|
||||
p.closeCh <- err
|
||||
return
|
||||
}
|
||||
sourceAddr := service.TunnelService.GetSourceAddr(conn.RemoteAddr(), conn.LocalAddr())
|
||||
appInfo := tunnel.GetAppInfo("tcp", "", p.interceptIP.String(), strconv.Itoa(service.Port), sourceAddr)
|
||||
identity := service.TunnelService.GetDialIdentity(conn.RemoteAddr(), conn.LocalAddr())
|
||||
go tunnel.DialAndRun(service.TunnelService, identity, conn, appInfo, true)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *interceptor) handleUDP(service *Service) {
|
||||
func (p *interceptor) handleUDP(service *Service) error {
|
||||
log := pfxlog.Logger().WithField("service", service.Name)
|
||||
|
||||
listenAddr := &net.UDPAddr{IP: p.interceptIP, Port: service.Port}
|
||||
udpPacketConn, err := net.ListenUDP("udp", listenAddr)
|
||||
if err != nil {
|
||||
log.Fatalln(err)
|
||||
p.closeCh <- err
|
||||
return
|
||||
return err
|
||||
}
|
||||
|
||||
service.setCloser(udpPacketConn)
|
||||
@@ -185,6 +176,7 @@ func (p *interceptor) handleUDP(service *Service) {
|
||||
}
|
||||
vconnManager := udp_vconn.NewManager(service.TunnelService.FabricProvider, udp_vconn.NewUnlimitedConnectionPolicy(), udp_vconn.NewDefaultExpirationPolicy())
|
||||
go reader.generateReadEvents(vconnManager)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *interceptor) Stop() {
|
||||
|
||||
+7
-4
@@ -28,13 +28,13 @@ import (
|
||||
|
||||
"github.com/michaelquigley/pfxlog"
|
||||
"github.com/openziti/agent"
|
||||
"github.com/openziti/sdk-golang/ziti"
|
||||
"github.com/openziti/ziti/common/enrollment"
|
||||
"github.com/openziti/ziti/common/version"
|
||||
"github.com/openziti/ziti/tunnel"
|
||||
"github.com/openziti/ziti/tunnel/dns"
|
||||
"github.com/openziti/ziti/tunnel/entities"
|
||||
"github.com/openziti/ziti/tunnel/intercept"
|
||||
"github.com/openziti/sdk-golang/ziti"
|
||||
"github.com/openziti/ziti/common/enrollment"
|
||||
"github.com/openziti/ziti/common/version"
|
||||
"github.com/sirupsen/logrus"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
@@ -136,7 +136,10 @@ func rootPostRun(cmd *cobra.Command, _ []string) {
|
||||
sdkinfo.SetApplication("ziti-tunnel", version.GetVersion())
|
||||
|
||||
resolverConfig := cmd.Flag(resolverCfgFlag).Value.String()
|
||||
resolver := dns.NewResolver(resolverConfig)
|
||||
resolver, err := dns.NewResolver(resolverConfig)
|
||||
if err != nil {
|
||||
log.WithError(err).Fatal("failed to start DNS resolver")
|
||||
}
|
||||
|
||||
serviceListenerGroup := intercept.NewServiceListenerGroup(interceptor, resolver)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user