middleware/proxy: dnstap (#786)

* experimental dnstap support into proxy

* proxy reports dnstap errors

* refactoring

* add a message builder for less dnstap code

* msg lint

* context

* proxy by DNS: dnstap comments

* TapBuilder

* resolves conflict

* dnstap into ServeDNS

* testing

* more tests

* `go lint`

* doc update

middleware/proxy/dns.go

@@ -28,6 +28,14 @@ func newDNSExWithOption(opt Options) *dnsEx {
 	return &dnsEx{Timeout: defaultTimeout * time.Second, Options: opt}
 }
 
+func (d *dnsEx) Transport() string {
+	if d.Options.ForceTCP {
+		return "tcp"
+	}
+
+	// The protocol will be determined by `state.Proto()` during Exchange.
+	return ""
+}
 func (d *dnsEx) Protocol() string          { return "dns" }
 func (d *dnsEx) OnShutdown(p *Proxy) error { return nil }
 func (d *dnsEx) OnStartup(p *Proxy) error  { return nil }

middleware/proxy/dnstap_test.go

@@ -0,0 +1,57 @@
+package proxy
+
+import (
+	"testing"
+
+	"github.com/coredns/coredns/middleware/dnstap/msg"
+	"github.com/coredns/coredns/middleware/dnstap/test"
+	mwtest "github.com/coredns/coredns/middleware/test"
+	"github.com/coredns/coredns/request"
+
+	tap "github.com/dnstap/golang-dnstap"
+	"github.com/miekg/dns"
+	"golang.org/x/net/context"
+)
+
+func testCase(t *testing.T, ex Exchanger, q, r *dns.Msg, datq, datr *msg.Data) {
+	tapq := datq.ToOutsideQuery(tap.Message_FORWARDER_QUERY)
+	tapr := datr.ToOutsideResponse(tap.Message_FORWARDER_RESPONSE)
+	ctx := test.Context{}
+	err := toDnstap(&ctx, "10.240.0.1:40212", ex,
+		request.Request{W: &mwtest.ResponseWriter{}, Req: q}, r, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(ctx.Trap) != 2 {
+		t.Fatalf("messages: %d", len(ctx.Trap))
+	}
+	if !test.MsgEqual(ctx.Trap[0], tapq) {
+		t.Errorf("want: %v\nhave: %v", tapq, ctx.Trap[0])
+	}
+	if !test.MsgEqual(ctx.Trap[1], tapr) {
+		t.Errorf("want: %v\nhave: %v", tapr, ctx.Trap[1])
+	}
+}
+
+func TestDnstap(t *testing.T) {
+	q := mwtest.Case{Qname: "example.org", Qtype: dns.TypeA}.Msg()
+	r := mwtest.Case{
+		Qname: "example.org.", Qtype: dns.TypeA,
+		Answer: []dns.RR{
+			mwtest.A("example.org. 3600	IN	A 10.0.0.1"),
+		},
+	}.Msg()
+	tapq, tapr := test.TestingData(), test.TestingData()
+	testCase(t, newDNSEx(), q, r, tapq, tapr)
+	tapq.SocketProto = tap.SocketProtocol_TCP
+	tapr.SocketProto = tap.SocketProtocol_TCP
+	testCase(t, newDNSExWithOption(Options{ForceTCP: true}), q, r, tapq, tapr)
+	testCase(t, newGoogle("", []string{"8.8.8.8:53", "8.8.4.4:53"}), q, r, tapq, tapr)
+}
+
+func TestNoDnstap(t *testing.T) {
+	err := toDnstap(context.TODO(), "", nil, request.Request{}, nil, 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+}

middleware/proxy/exchanger.go

@@ -13,6 +13,10 @@ type Exchanger interface {
 	Exchange(ctx context.Context, addr string, state request.Request) (*dns.Msg, error)
 	Protocol() string
 
+	// Transport returns the only transport protocol used by this Exchanger or "".
+	// If the return value is "", Exchange must use `state.Proto()`.
+	Transport() string
+
 	OnStartup(*Proxy) error
 	OnShutdown(*Proxy) error
 }

middleware/proxy/google.go

@@ -112,6 +112,10 @@ func (g *google) exchangeJSON(addr, json string) ([]byte, error) {
 	return buf, nil
 }
 
+func (g *google) Transport() string {
+	return "tcp"
+}
+
 func (g *google) Protocol() string { return "https_google" }
 
 func (g *google) OnShutdown(p *Proxy) error {

middleware/proxy/grpc.go

@@ -54,6 +54,8 @@ func (g *grpcClient) Exchange(ctx context.Context, addr string, state request.Re
 	return d, nil
 }
 
+func (g *grpcClient) Transport() string { return "tcp" }
+
 func (g *grpcClient) Protocol() string { return "grpc" }
 
 func (g *grpcClient) OnShutdown(p *Proxy) error {

middleware/proxy/proxy.go

@@ -7,9 +7,12 @@ import (
 	"time"
 
 	"github.com/coredns/coredns/middleware"
+	"github.com/coredns/coredns/middleware/dnstap"
+	"github.com/coredns/coredns/middleware/dnstap/msg"
 	"github.com/coredns/coredns/middleware/pkg/healthcheck"
 	"github.com/coredns/coredns/request"
 
+	tap "github.com/dnstap/golang-dnstap"
 	"github.com/miekg/dns"
 	ot "github.com/opentracing/opentracing-go"
 	"golang.org/x/net/context"
@@ -85,22 +88,28 @@ func (p Proxy) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (
 			}
 
 			atomic.AddInt64(&host.Conns, 1)
+			queryEpoch := msg.Epoch()
 
 			reply, backendErr := upstream.Exchanger().Exchange(ctx, host.Name, state)
 
+			respEpoch := msg.Epoch()
 			atomic.AddInt64(&host.Conns, -1)
 
 			if child != nil {
 				child.Finish()
 			}
 
+			taperr := toDnstap(ctx, host.Name, upstream.Exchanger(), state, reply,
+				queryEpoch, respEpoch)
+
 			if backendErr == nil {
 				w.WriteMsg(reply)
 
 				RequestDuration.WithLabelValues(state.Proto(), upstream.Exchanger().Protocol(), upstream.From()).Observe(float64(time.Since(start) / time.Millisecond))
 
-				return 0, nil
+				return 0, taperr
 			}
+
 			timeout := host.FailTimeout
 			if timeout == 0 {
 				timeout = 10 * time.Second
@@ -145,3 +154,40 @@ func (p Proxy) Name() string { return "proxy" }
 
 // defaultTimeout is the default networking timeout for DNS requests.
 const defaultTimeout = 5 * time.Second
+
+func toDnstap(ctx context.Context, host string, ex Exchanger, state request.Request, reply *dns.Msg, queryEpoch, respEpoch uint64) (err error) {
+	if tapper := dnstap.TapperFromContext(ctx); tapper != nil {
+		// Query
+		b := tapper.TapBuilder()
+		b.TimeSec = queryEpoch
+		if err = b.HostPort(host); err != nil {
+			return
+		}
+		t := ex.Transport()
+		if t == "" {
+			t = state.Proto()
+		}
+		if t == "tcp" {
+			b.SocketProto = tap.SocketProtocol_TCP
+		} else {
+			b.SocketProto = tap.SocketProtocol_UDP
+		}
+		if err = b.Msg(state.Req); err != nil {
+			return
+		}
+		err = tapper.TapMessage(b.ToOutsideQuery(tap.Message_FORWARDER_QUERY))
+		if err != nil {
+			return
+		}
+
+		// Response
+		if reply != nil {
+			b.TimeSec = respEpoch
+			if err = b.Msg(reply); err != nil {
+				return
+			}
+			err = tapper.TapMessage(b.ToOutsideResponse(tap.Message_FORWARDER_RESPONSE))
+		}
+	}
+	return
+}