coredns/plugin/grpc/grpc.go

package grpc

import (
	"context"
	"crypto/tls"
	"errors"
	"time"

	"github.com/coredns/coredns/plugin"
	"github.com/coredns/coredns/plugin/debug"
	"github.com/coredns/coredns/plugin/pkg/fall"
	"github.com/coredns/coredns/request"

	"github.com/miekg/dns"
	ot "github.com/opentracing/opentracing-go"
)

// GRPC represents a plugin instance that can proxy requests to another (DNS) server via gRPC protocol.
// It has a list of proxies each representing one upstream proxy.
type GRPC struct {
	proxies []*Proxy
	p       Policy

	from    string
	ignored []string

	tlsConfig     *tls.Config
	tlsServerName string

	Fall fall.F
	Next plugin.Handler
}

// ServeDNS implements the plugin.Handler interface.
func (g *GRPC) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
	state := request.Request{W: w, Req: r}
	if !g.match(state) {
		if g.Next != nil {
			return plugin.NextOrFailure(g.Name(), g.Next, ctx, w, r)
		}
		// No next plugin, return SERVFAIL
		return dns.RcodeServerFailure, nil
	}

	var (
		span ot.Span
		ret  *dns.Msg
		err  error
		i    int
	)
	span = ot.SpanFromContext(ctx)
	list := g.list()
	deadline := time.Now().Add(defaultTimeout)

	for time.Now().Before(deadline) {
		if i >= len(list) {
			// reached the end of list without any answer
			if ret != nil {
				// write empty response and finish
				w.WriteMsg(ret)
			}
			break
		}

		proxy := list[i]
		i++

		callCtx := ctx
		var child ot.Span
		if span != nil {
			child, callCtx = ot.StartSpanFromContext(callCtx, "query")
		}

		var cancel context.CancelFunc
		callCtx, cancel = context.WithDeadline(callCtx, deadline)

		ret, err = proxy.query(callCtx, r)
		cancel()

		if child != nil {
			child.Finish()
		}
		if err != nil {
			// Continue with the next proxy
			continue
		}

		// Check if the reply is correct; if not return FormErr.
		if !state.Match(ret) {
			debug.Hexdumpf(ret, "Wrong reply for id: %d, %s %d", ret.Id, state.QName(), state.QType())

			formerr := new(dns.Msg)
			formerr.SetRcode(state.Req, dns.RcodeFormatError)
			w.WriteMsg(formerr)
			return 0, nil
		}

		// Check if we should fallthrough on NXDOMAIN responses
		if ret.Rcode == dns.RcodeNameError && g.Fall.Through(state.Name()) {
			if g.Next != nil {
				return plugin.NextOrFailure(g.Name(), g.Next, ctx, w, r)
			}
			// No next plugin to fallthrough to, return the NXDOMAIN response
		}

		w.WriteMsg(ret)
		return 0, nil
	}

	// SERVFAIL if all healthy proxys returned errors.
	if err != nil {
		// If fallthrough is enabled, try the next plugin instead of returning SERVFAIL
		if g.Fall.Through(state.Name()) && g.Next != nil {
			return plugin.NextOrFailure(g.Name(), g.Next, ctx, w, r)
		}
		// just return the last error received
		return dns.RcodeServerFailure, err
	}

	// If fallthrough is enabled, try the next plugin instead of returning SERVFAIL
	if g.Fall.Through(state.Name()) && g.Next != nil {
		return plugin.NextOrFailure(g.Name(), g.Next, ctx, w, r)
	}

	return dns.RcodeServerFailure, ErrNoHealthy
}

// NewGRPC returns a new GRPC.
func newGRPC() *GRPC {
	g := &GRPC{
		p: new(random),
	}
	return g
}

// Name implements the Handler interface.
func (g *GRPC) Name() string { return "grpc" }

// Len returns the number of configured proxies.
func (g *GRPC) len() int { return len(g.proxies) }

func (g *GRPC) match(state request.Request) bool {
	if !plugin.Name(g.from).Matches(state.Name()) || !g.isAllowedDomain(state.Name()) {
		return false
	}

	return true
}

func (g *GRPC) isAllowedDomain(name string) bool {
	if dns.Name(name) == dns.Name(g.from) {
		return true
	}

	for _, ignore := range g.ignored {
		if plugin.Name(ignore).Matches(name) {
			return false
		}
	}
	return true
}

// List returns a set of proxies to be used for this client depending on the policy in p.
func (g *GRPC) list() []*Proxy { return g.p.List(g.proxies) }

const defaultTimeout = 5 * time.Second

var (
	// ErrNoHealthy means no healthy proxies left.
	ErrNoHealthy = errors.New("no healthy gRPC proxies")
)