IP endpoint for dnstap (#1002)

* adds the option to log to a remote endpoint

* examples

* tests

* tcp:// or default to unix://

* cosmetic update

* bad naked returns

middleware/dnstap/out/tcp.go

@@ -0,0 +1,59 @@
+package out
+
+import (
+	"net"
+	"time"
+
+	fs "github.com/farsightsec/golang-framestream"
+)
+
+// TCP is a Frame Streams encoder over TCP.
+type TCP struct {
+	address string
+	frames  [][]byte
+}
+
+// NewTCP returns a TCP writer.
+func NewTCP(address string) *TCP {
+	s := &TCP{address: address}
+	s.frames = make([][]byte, 0, 13) // 13 messages buffer
+	return s
+}
+
+// Write a single Frame Streams frame.
+func (s *TCP) Write(frame []byte) (n int, err error) {
+	s.frames = append(s.frames, frame)
+	if len(s.frames) == cap(s.frames) {
+		return len(frame), s.Flush()
+	}
+	return len(frame), nil
+}
+
+// Flush the remaining frames.
+func (s *TCP) Flush() error {
+	defer func() {
+		s.frames = s.frames[0:]
+	}()
+	c, err := net.DialTimeout("tcp", s.address, time.Second)
+	if err != nil {
+		return err
+	}
+	enc, err := fs.NewEncoder(c, &fs.EncoderOptions{
+		ContentType:   []byte("protobuf:dnstap.Dnstap"),
+		Bidirectional: true,
+	})
+	if err != nil {
+		return err
+	}
+	for _, frame := range s.frames {
+		if _, err = enc.Write(frame); err != nil {
+			return err
+		}
+	}
+	return enc.Flush()
+}
+
+// Close is an alias to Flush to satisfy io.WriteCloser similarly to type Socket.
+func (s *TCP) Close() error {
+	return s.Flush()
+}

middleware/dnstap/out/tcp_test.go

@@ -0,0 +1,66 @@
+package out
+
+import (
+	"net"
+	"testing"
+)
+
+func sendOneTcp(tcp *TCP) error {
+	if _, err := tcp.Write([]byte("frame")); err != nil {
+		return err
+	}
+	if err := tcp.Flush(); err != nil {
+		return err
+	}
+	return nil
+}
+func TestTcp(t *testing.T) {
+	tcp := NewTCP("localhost:14000")
+
+	if err := sendOneTcp(tcp); err == nil {
+		t.Fatal("Not listening but no error.")
+		return
+	}
+
+	l, err := net.Listen("tcp", "localhost:14000")
+	if err != nil {
+		t.Fatal(err)
+		return
+	}
+
+	wait := make(chan bool)
+	go func() {
+		acceptOne(t, l)
+		wait <- true
+	}()
+
+	if err := sendOneTcp(tcp); err != nil {
+		t.Fatalf("send one: %s", err)
+		return
+	}
+
+	<-wait
+
+	// TODO: When the server isn't responding according to the framestream protocol
+	// the thread is blocked.
+	/*
+		if err := sendOneTcp(tcp); err == nil {
+			panic("must fail")
+		}
+	*/
+
+	go func() {
+		acceptOne(t, l)
+		wait <- true
+	}()
+
+	if err := sendOneTcp(tcp); err != nil {
+		t.Fatalf("send one: %s", err)
+		return
+	}
+
+	<-wait
+	if err := l.Close(); err != nil {
+		t.Error(err)
+	}
+}