middleware/proxy: implement Exchanger (#480)

By defining and using an proxy.Exchanger interface we make the proxy
more generic and we can then fold back httproxy into proxy.

This overrides #463 and #473 and should make futures extensions rather
trivial

* Add docs that talk about `protocol` and how to set it.
* middleware/proxy: rename New to NewLookup
  It's used as a Lookup mechanism not as a completely new proxy,
  reflect that in the name.
* Set maxfails to 3 by default when looking up names.

Most of the changes have been copied
from https://github.com/johnbelamaric/coredns/pull/1/files
This commit is contained in:
Miek Gieben 2017-01-15 08:12:58 +00:00 committed by GitHub
parent a6d232a622
commit 52e01264e8
25 changed files with 140 additions and 61 deletions

View file

@ -153,7 +153,7 @@ func autoParse(c *caddy.Controller) (Auto, error) {
if err != nil {
return a, err
}
a.loader.proxy = proxy.New(ups)
a.loader.proxy = proxy.NewLookup(ups)
default:
t, _, e := file.TransferParse(c, false)

View file

@ -16,7 +16,7 @@ import (
func TestProxyLookupFailDebug(t *testing.T) {
etc := newEtcdMiddleware()
etc.Proxy = proxy.New([]string{"127.0.0.1:154"})
etc.Proxy = proxy.NewLookup([]string{"127.0.0.1:154"})
etc.Debugging = true
for _, serv := range servicesProxy {

View file

@ -49,7 +49,7 @@ func setup(c *caddy.Controller) error {
func etcdParse(c *caddy.Controller) (*Etcd, bool, error) {
stub := make(map[string]proxy.Proxy)
etc := Etcd{
Proxy: proxy.New([]string{"8.8.8.8:53", "8.8.4.4:53"}),
Proxy: proxy.NewLookup([]string{"8.8.8.8:53", "8.8.4.4:53"}),
PathPrefix: "skydns",
Ctx: context.Background(),
Inflight: &singleflight.Group{},
@ -96,7 +96,7 @@ func etcdParse(c *caddy.Controller) (*Etcd, bool, error) {
if err != nil {
return &Etcd{}, false, err
}
etc.Proxy = proxy.New(ups)
etc.Proxy = proxy.NewLookup(ups)
case "tls": // cert key cacertfile
args := c.RemainingArgs()
tlsConfig, err = mwtls.NewTLSConfigFromArgs(args...)
@ -134,7 +134,7 @@ func etcdParse(c *caddy.Controller) (*Etcd, bool, error) {
if err != nil {
return &Etcd{}, false, c.ArgErr()
}
etc.Proxy = proxy.New(ups)
etc.Proxy = proxy.NewLookup(ups)
case "tls": // cert key cacertfile
args := c.RemainingArgs()
tlsConfig, err = mwtls.NewTLSConfigFromArgs(args...)

View file

@ -12,9 +12,9 @@ import (
"github.com/miekg/coredns/middleware/etcd/msg"
"github.com/miekg/coredns/middleware/pkg/dnsrecorder"
"github.com/miekg/coredns/middleware/pkg/singleflight"
"github.com/miekg/coredns/middleware/pkg/tls"
"github.com/miekg/coredns/middleware/proxy"
"github.com/miekg/coredns/middleware/test"
"github.com/miekg/coredns/middleware/pkg/tls"
etcdc "github.com/coreos/etcd/client"
"github.com/mholt/caddy"
@ -33,7 +33,7 @@ func newEtcdMiddleware() *Etcd {
client, _ := newEtcdClient(endpoints, tlsc)
return &Etcd{
Proxy: proxy.New([]string{"8.8.8.8:53"}),
Proxy: proxy.NewLookup([]string{"8.8.8.8:53"}),
PathPrefix: "skydns",
Ctx: context.Background(),
Inflight: &singleflight.Group{},

View file

@ -67,7 +67,7 @@ Services:
}
for domain, nss := range nameservers {
stubmap[domain] = proxy.New(nss)
stubmap[domain] = proxy.NewLookup(nss)
}
// atomic swap (at least that's what we hope it is)
if len(stubmap) > 0 {

View file

@ -93,7 +93,7 @@ func TestLookupCNAMEExternal(t *testing.T) {
if err != nil {
t.Fatalf("Expected no error when reading zone, got %q", err)
}
zone.Proxy = proxy.New([]string{"8.8.8.8:53"}) // TODO(miek): point to local instance
zone.Proxy = proxy.NewLookup([]string{"8.8.8.8:53"}) // TODO(miek): point to local instance
fm := File{Next: test.ErrorHandler(), Zones: Zones{Z: map[string]*Zone{name: zone}, Names: []string{name}}}
ctx := context.TODO()

View file

@ -110,7 +110,7 @@ func fileParse(c *caddy.Controller) (Zones, error) {
if err != nil {
return Zones{}, err
}
prxy = proxy.New(ups)
prxy = proxy.NewLookup(ups)
}
for _, origin := range origins {

View file

@ -26,6 +26,7 @@ proxy FROM TO... {
health_check PATH:PORT [DURATION]
except IGNORED_NAMES...
spray
protocol [dns|https_google]
}
~~~
@ -37,6 +38,8 @@ proxy FROM TO... {
* `health_check` will check path (on port) on each backend. If a backend returns a status code of 200-399, then that backend is healthy. If it doesn't, the backend is marked as unhealthy for duration and no requests are routed to it. If this option is not provided then health checks are disabled. The default duration is 10 seconds ("10s").
* `ignored_names...` is a space-separated list of paths to exclude from proxying. Requests that match any of these paths will be passed through.
* `spray` when all backends are unhealthy, randomly pick one to send the traffic to. (This is a failsafe.)
* `protocol` specifies what protocol to use to speak to an upstream, `dns` (the default) is plain old DNS, and
`https_google` uses `https://dns.google.com` and speaks a JSON DNS dialect.
## Policies
@ -48,14 +51,20 @@ There are three load-balancing policies available:
All polices implement randomly spraying packets to backend hosts when *no healthy* hosts are
available. This is to preeempt the case where the healthchecking (as a mechanism) fails.
## Upstream Protocols
Currently supported are `dns` (i.e., standard DNS over UDP) and `https_google`. Note that with
`https_google` the entire transport is encrypted. Only *you* and *Google* can see your DNS activity.
## Metrics
If monitoring is enabled (via the *prometheus* directive) then the following metric is exported:
* coredns_proxy_request_count_total{zone, proto, family}
* coredns_proxy_request_count_total{protocol, zone, family}
This has some overlap with `coredns_dns_request_count_total{zone, proto, family}`, but allows for
specifics on upstream query resolving. See the *prometheus* documentation for more details.
`protocol` is the protocol used to query the upstream.
## Examples

View file

@ -10,25 +10,30 @@ import (
"github.com/miekg/dns"
)
type client struct {
type dnsEx struct {
Timeout time.Duration
Address string // address/name of this upstream
group *singleflight.Group
}
func newClient() *client {
return &client{Timeout: defaultTimeout, group: new(singleflight.Group)}
func newDNSEx(address string) *dnsEx {
return &dnsEx{Address: address, group: new(singleflight.Group), Timeout: defaultTimeout * time.Second}
}
// ServeDNS does not satisfy middleware.Handler, instead it interacts with the upstream
// and returns the respons or an error.
func (c *client) ServeDNS(w dns.ResponseWriter, r *dns.Msg, u *UpstreamHost) (*dns.Msg, error) {
co, err := net.DialTimeout(request.Proto(w), u.Name, c.Timeout)
func (d *dnsEx) OnStartup() error { return nil }
func (d *dnsEx) OnShutdown() error { return nil }
func (d *dnsEx) SetUpstream(u Upstream) error { return nil }
func (d *dnsEx) Protocol() protocol { return dnsProto }
// Exchange implements the Exchanger interface.
func (d *dnsEx) Exchange(state request.Request) (*dns.Msg, error) {
co, err := net.DialTimeout(state.Proto(), d.Address, d.Timeout)
if err != nil {
return nil, err
}
reply, _, err := c.Exchange(r, co)
reply, _, err := d.ExchangeConn(state.Req, co)
co.Close()
@ -42,12 +47,12 @@ func (c *client) ServeDNS(w dns.ResponseWriter, r *dns.Msg, u *UpstreamHost) (*d
}
reply.Compress = true
reply.Id = r.Id
reply.Id = state.Req.Id
return reply, nil
}
func (c *client) Exchange(m *dns.Msg, co net.Conn) (*dns.Msg, time.Duration, error) {
func (d *dnsEx) ExchangeConn(m *dns.Msg, co net.Conn) (*dns.Msg, time.Duration, error) {
t := "nop"
if t1, ok := dns.TypeToString[m.Question[0].Qtype]; ok {
t = t1
@ -60,9 +65,8 @@ func (c *client) Exchange(m *dns.Msg, co net.Conn) (*dns.Msg, time.Duration, err
start := time.Now()
// Name needs to be normalized! Bug in go dns.
r, err := c.group.Do(m.Question[0].Name+t+cl, func() (interface{}, error) {
ret, e := c.exchange(m, co)
return ret, e
r, err := d.group.Do(m.Question[0].Name+t+cl, func() (interface{}, error) {
return exchange(m, co)
})
r1 := r.(dns.Msg)
@ -72,7 +76,7 @@ func (c *client) Exchange(m *dns.Msg, co net.Conn) (*dns.Msg, time.Duration, err
// exchange does *not* return a pointer to dns.Msg because that leads to buffer reuse when
// group.Do is used in Exchange.
func (c *client) exchange(m *dns.Msg, co net.Conn) (dns.Msg, error) {
func exchange(m *dns.Msg, co net.Conn) (dns.Msg, error) {
opt := m.IsEdns0()
udpsize := uint16(dns.MinMsgSize)
@ -97,3 +101,5 @@ func (c *client) exchange(m *dns.Msg, co net.Conn) (dns.Msg, error) {
}
return *r, err
}
const dnsProto protocol = "dns"

View file

@ -0,0 +1,18 @@
package proxy
import (
"github.com/miekg/coredns/request"
"github.com/miekg/dns"
)
// Exchanger is an interface that specifies a type implementing a DNS resolver that
// can use whatever transport it likes.
type Exchanger interface {
Exchange(request.Request) (*dns.Msg, error)
SetUpstream(Upstream) error // (Re)set the upstream
OnStartup() error
OnShutdown() error
Protocol() protocol
}
type protocol string

View file

@ -11,9 +11,10 @@ import (
"github.com/miekg/dns"
)
// New create a new proxy with the hosts in host and a Random policy.
func New(hosts []string) Proxy {
p := Proxy{Next: nil, Client: newClient()}
// NewLookup create a new proxy with the hosts in host and a Random policy.
func NewLookup(hosts []string) Proxy {
// TODO(miek): maybe add optional protocol parameter?
p := Proxy{Next: nil}
upstream := &staticUpstream{
from: "",
@ -21,7 +22,7 @@ func New(hosts []string) Proxy {
Policy: &Random{},
Spray: nil,
FailTimeout: 10 * time.Second,
MaxFails: 1,
MaxFails: 3,
}
for i, host := range hosts {
@ -30,6 +31,7 @@ func New(hosts []string) Proxy {
Conns: 0,
Fails: 0,
FailTimeout: upstream.FailTimeout,
Exchanger: newDNSEx(host),
Unhealthy: false,
CheckDown: func(upstream *staticUpstream) UpstreamHostDownFunc {
@ -59,15 +61,17 @@ func (p Proxy) Lookup(state request.Request, name string, typ uint16) (*dns.Msg,
req.SetQuestion(name, typ)
state.SizeAndDo(req)
return p.lookup(state, req)
state2 := request.Request{W: state.W, Req: req}
return p.lookup(state2)
}
// Forward forward the request in state as-is. Unlike Lookup that adds EDNS0 suffix to the message.
func (p Proxy) Forward(state request.Request) (*dns.Msg, error) {
return p.lookup(state, state.Req)
return p.lookup(state)
}
func (p Proxy) lookup(state request.Request, r *dns.Msg) (*dns.Msg, error) {
func (p Proxy) lookup(state request.Request) (*dns.Msg, error) {
for _, upstream := range p.Upstreams {
start := time.Now()
@ -80,11 +84,11 @@ func (p Proxy) lookup(state request.Request, r *dns.Msg) (*dns.Msg, error) {
}
// duplicated from proxy.go, but with a twist, we don't write the
// reply back to the client, we return it.
// reply back to the client, we return it and there is no monitoring.
atomic.AddInt64(&host.Conns, 1)
reply, backendErr := p.Client.ServeDNS(state.W, r, host)
reply, backendErr := host.Exchange(state)
atomic.AddInt64(&host.Conns, -1)

View file

@ -12,14 +12,14 @@ import (
var (
RequestDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
Namespace: middleware.Namespace,
Subsystem: subsystem,
Subsystem: "proxy",
Name: "request_duration_milliseconds",
Buckets: append(prometheus.DefBuckets, []float64{50, 100, 200, 500, 1000, 2000, 3000, 4000, 5000, 10000}...),
Help: "Histogram of the time (in milliseconds) each request took.",
}, []string{"zone"})
}, []string{"protocol", "zone"})
)
// OnStartup sets up the metrics on startup.
// OnStartup sets up the metrics on startup. This is done for all proxy protocols.
func OnStartup() error {
metricsOnce.Do(func() {
prometheus.MustRegister(RequestDuration)
@ -28,5 +28,3 @@ func OnStartup() error {
}
var metricsOnce sync.Once
const subsystem = "proxy"

View file

@ -7,17 +7,20 @@ import (
"time"
"github.com/miekg/coredns/middleware"
"github.com/miekg/coredns/request"
"github.com/miekg/dns"
"golang.org/x/net/context"
)
var errUnreachable = errors.New("unreachable backend")
var (
errUnreachable = errors.New("unreachable backend")
errInvalidProtocol = errors.New("invalid protocol")
)
// Proxy represents a middleware instance that can proxy requests to another DNS server.
// Proxy represents a middleware instance that can proxy requests to another (DNS) server.
type Proxy struct {
Next middleware.Handler
Client *client
Upstreams []Upstream
}
@ -46,6 +49,7 @@ type UpstreamHost struct {
Unhealthy bool
CheckDown UpstreamHostDownFunc
WithoutPathPrefix string
Exchanger
}
// Down checks whether the upstream host is down or not.
@ -66,6 +70,7 @@ var tryDuration = 60 * time.Second
// ServeDNS satisfies the middleware.Handler interface.
func (p Proxy) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
state := request.Request{W: w, Req: r}
for _, upstream := range p.Upstreams {
start := time.Now()
@ -82,7 +87,7 @@ func (p Proxy) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (
atomic.AddInt64(&host.Conns, 1)
reply, backendErr := p.Client.ServeDNS(w, r, host)
reply, backendErr := host.Exchange(state)
atomic.AddInt64(&host.Conns, -1)

View file

@ -20,7 +20,7 @@ func setup(c *caddy.Controller) error {
return middleware.Error("proxy", err)
}
dnsserver.GetConfig(c).AddMiddleware(func(next middleware.Handler) middleware.Handler {
return Proxy{Next: next, Client: newClient(), Upstreams: upstreams}
return Proxy{Next: next, Upstreams: upstreams}
})
c.OnStartup(OnStartup)

View file

@ -1,6 +1,7 @@
package proxy
import (
"fmt"
"io"
"io/ioutil"
"net"
@ -37,6 +38,7 @@ type staticUpstream struct {
WithoutPathPrefix string
IgnoredSubDomains []string
options Options
Protocol protocol
}
// Options ...
@ -56,6 +58,7 @@ func NewStaticUpstreams(c *caddyfile.Dispenser) ([]Upstream, error) {
Spray: nil,
FailTimeout: 10 * time.Second,
MaxFails: 1,
Protocol: dnsProto,
}
if !c.Args(&upstream.from) {
@ -86,6 +89,7 @@ func NewStaticUpstreams(c *caddyfile.Dispenser) ([]Upstream, error) {
Fails: 0,
FailTimeout: upstream.FailTimeout,
Unhealthy: false,
Exchanger: newDNSEx(host),
CheckDown: func(upstream *staticUpstream) UpstreamHostDownFunc {
return func(uh *UpstreamHost) bool {
@ -102,6 +106,15 @@ func NewStaticUpstreams(c *caddyfile.Dispenser) ([]Upstream, error) {
}(upstream),
WithoutPathPrefix: upstream.WithoutPathPrefix,
}
switch upstream.Protocol {
// case https_google:
case dnsProto:
fallthrough
default:
// Already done in the initialization above.
}
upstream.Hosts[i] = uh
}
@ -188,6 +201,19 @@ func parseBlock(c *caddyfile.Dispenser, u *staticUpstream) error {
u.IgnoredSubDomains = ignoredDomains
case "spray":
u.Spray = &Spray{}
case "protocol":
encArgs := c.RemainingArgs()
if len(encArgs) == 0 {
return c.ArgErr()
}
switch encArgs[0] {
case "dns":
u.Protocol = dnsProto
case "https_google":
// Nothing yet.
default:
return fmt.Errorf("%s: %s", errInvalidProtocol, encArgs[0])
}
default:
return c.Errf("unknown property '%s'", c.Val())

View file

@ -165,6 +165,20 @@ proxy . 8.8.8.8:53 {
proxy . some_bogus_filename`,
true,
},
{
`
proxy . 8.8.8.8:53 {
protocol dns
}`,
false,
},
{
`
proxy . 8.8.8.8:53 {
protocol foobar
}`,
true,
},
}
for i, test := range tests {
c := caddy.NewTestController("dns", test.inputUpstreams)

View file

@ -42,7 +42,7 @@ func TestAuto(t *testing.T) {
log.SetOutput(ioutil.Discard)
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "www.example.org.", dns.TypeA)
@ -108,7 +108,7 @@ func TestAutoNonExistentZone(t *testing.T) {
}
defer i.Stop()
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "example.org.", dns.TypeA)
@ -155,7 +155,7 @@ func TestAutoAXFR(t *testing.T) {
time.Sleep(1100 * time.Millisecond) // wait for it to be picked up
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
m := new(dns.Msg)
m.SetAxfr("example.org.")
state := request.Request{W: &test.ResponseWriter{}, Req: m}

View file

@ -56,7 +56,7 @@ func TestLookupCache(t *testing.T) {
log.SetOutput(ioutil.Discard)
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "example.org.", dns.TypeA)
@ -65,7 +65,7 @@ func TestLookupCache(t *testing.T) {
}
// expect answer section with A record in it
if len(resp.Answer) == 0 {
t.Error("Expected to at least one RR in the answer section, got none")
t.Fatal("Expected to at least one RR in the answer section, got none")
}
ttl := resp.Answer[0].Header().Ttl

View file

@ -57,7 +57,7 @@ func TestLookupDS(t *testing.T) {
log.SetOutput(ioutil.Discard)
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &mtest.ResponseWriter{}, Req: new(dns.Msg)}
for _, tc := range dsTestCases {

View file

@ -47,7 +47,7 @@ func TestEtcdCacheAndDebug(t *testing.T) {
defer delete(ctx, t, etc, serv.Key)
}
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "b.example.skydns.test.", dns.TypeA)

View file

@ -66,15 +66,14 @@ func TestEtcdStubAndProxyLookup(t *testing.T) {
defer delete(ctx, t, etc, serv.Key)
}
p := proxy.New([]string{udp}) // use udp port from the server
p := proxy.NewLookup([]string{udp}) // use udp port from the server
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "example.com.", dns.TypeA)
if err != nil {
t.Error("Expected to receive reply, but didn't")
return
t.Fatalf("Expected to receive reply, but didn't", err)
}
if len(resp.Answer) == 0 {
t.Error("Expected to at least one RR in the answer section, got none")
t.Fatalf("Expected to at least one RR in the answer section, got none")
}
if resp.Answer[0].Header().Rrtype != dns.TypeA {
t.Errorf("Expected RR to A, got: %d", resp.Answer[0].Header().Rrtype)

View file

@ -42,7 +42,7 @@ example.net:0 {
}
defer i.Stop()
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "example.org.", dns.TypeA)

View file

@ -33,7 +33,7 @@ func TestProxyErratic(t *testing.T) {
}
defer backend.Stop()
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
// We do one lookup that should not time out.

View file

@ -38,7 +38,7 @@ func TestLookupProxy(t *testing.T) {
log.SetOutput(ioutil.Discard)
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
resp, err := p.Lookup(state, "example.org.", dns.TypeA)
if err != nil {
@ -82,7 +82,7 @@ func BenchmarkLookupProxy(b *testing.B) {
log.SetOutput(ioutil.Discard)
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
b.ResetTimer()

View file

@ -38,7 +38,7 @@ func TestLookupWildcard(t *testing.T) {
log.SetOutput(ioutil.Discard)
p := proxy.New([]string{udp})
p := proxy.NewLookup([]string{udp})
state := request.Request{W: &test.ResponseWriter{}, Req: new(dns.Msg)}
for _, lookup := range []string{"a.w.example.org.", "a.a.w.example.org."} {