Merge pull request #472 from gliderlabs/jszwedko/add-octet-framing

Add optional TCP framing to syslog adapter

Author: michaelshobbs

adapters/syslog/syslog.go

@@ -2,6 +2,7 @@ package syslog
 
 import (
 	"bytes"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -19,14 +20,25 @@ import (
 	"github.com/gliderlabs/logspout/router"
 )
 
-const defaultRetryCount = 10
+const (
+	// TraditionalTCPFraming is the traditional LF framing of syslog messages on the wire
+	TraditionalTCPFraming TCPFraming = "traditional"
+	// OctetCountedTCPFraming prepends the size of each message before the message. https://tools.ietf.org/html/rfc6587#section-3.4.1
+	OctetCountedTCPFraming TCPFraming = "octet-counted"
+
+	defaultRetryCount = 10
+)
 
 var (
 	hostname         string
 	retryCount       uint
+	tcpFraming       TCPFraming
 	econnResetErrStr string
 )
 
+// TCPFraming represents the type of framing to use for syslog messages
+type TCPFraming string
+
 func init() {
 	hostname, _ = os.Hostname()
 	econnResetErrStr = fmt.Sprintf("write: %s", syscall.ECONNRESET.Error())
@@ -89,6 +101,12 @@ func NewSyslogAdapter(route *router.Route) (router.LogAdapter, error) {
 		structuredData = fmt.Sprintf("[%s]", structuredData)
 	}
 
+	if isTCPConnecion(conn) {
+		if err = setTCPFraming(); err != nil {
+			return nil, err
+		}
+	}
+
 	var tmplStr string
 	switch format {
 	case "rfc5424":
@@ -120,6 +138,19 @@ func NewSyslogAdapter(route *router.Route) (router.LogAdapter, error) {
 	}, nil
 }
 
+func setTCPFraming() error {
+	switch s := cfg.GetEnvDefault("SYSLOG_TCP_FRAMING", "traditional"); s {
+	case "traditional":
+		tcpFraming = TraditionalTCPFraming
+		return nil
+	case "octet-counted":
+		tcpFraming = OctetCountedTCPFraming
+		return nil
+	default:
+		return fmt.Errorf("unknown SYSLOG_TCP_FRAMING value: %s", s)
+	}
+}
+
 // Adapter streams log output to a connection in the Syslog format
 type Adapter struct {
 	conn      net.Conn
@@ -137,6 +168,19 @@ func (a *Adapter) Stream(logstream chan *router.Message) {
 			log.Println("syslog:", err)
 			return
 		}
+
+		if isTCPConnecion(a.conn) {
+			switch tcpFraming {
+			case OctetCountedTCPFraming:
+				buf = append([]byte(fmt.Sprintf("%d ", len(buf))), buf...)
+			case TraditionalTCPFraming:
+				// leave as-is
+			default:
+				// should never get here, validated above
+				panic("unknown framing format: " + tcpFraming)
+			}
+		}
+
 		if _, err = a.conn.Write(buf); err != nil {
 			log.Println("syslog:", err)
 			switch a.conn.(type) {
@@ -226,6 +270,17 @@ func retryExp(fun func() error, tries uint) error {
 	}
 }
 
+func isTCPConnecion(conn net.Conn) bool {
+	switch conn.(type) {
+	case *net.TCPConn:
+		return true
+	case *tls.Conn:
+		return true
+	default:
+		return false
+	}
+}
+
 // Message extends router.Message for the syslog standard
 type Message struct {
 	*router.Message

adapters/syslog/syslog_test.go

@@ -2,6 +2,7 @@ package syslog
 
 import (
 	"bufio"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"log"
@@ -40,6 +41,56 @@ var (
 	badHostnameContent   = "hostname\r\n"
 )
 
+func TestSyslogOctetFraming(t *testing.T) {
+	os.Setenv("SYSLOG_TCP_FRAMING", "octet-counted")
+	defer os.Unsetenv("SYSLOG_TCP_FRAMING")
+
+	done := make(chan string)
+	addr, sock, srvWG := startServer("tcp", "", done)
+	defer srvWG.Wait()
+	defer os.Remove(addr)
+	defer sock.Close()
+
+	route := &router.Route{Adapter: "syslog+tcp", Address: addr}
+	adapter, err := NewSyslogAdapter(route)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer adapter.(*Adapter).conn.Close()
+
+	stream := make(chan *router.Message)
+	go adapter.Stream(stream)
+
+	count := 1
+	messages := make(chan string, count)
+	go sendLogstream(stream, messages, adapter, count)
+
+	timeout := time.After(6 * time.Second)
+	msgnum := 1
+	select {
+	case msg := <-done:
+		sizeStr := ""
+		_, err := fmt.Sscan(msg, &sizeStr)
+		if err != nil {
+			t.Fatal("unable to scan size from message: ", err)
+		}
+
+		size, err := strconv.ParseInt(sizeStr, 10, 32)
+		if err != nil {
+			t.Fatal("unable to scan size from message: ", err)
+		}
+
+		expectedOctetFrame := len(sizeStr) + 1 + int(size)
+		if len(msg) != expectedOctetFrame {
+			t.Errorf("expected octet frame to be %d. got %d instead for message %s", expectedOctetFrame, size, msg)
+		}
+		return
+	case <-timeout:
+		t.Fatal("timeout after", msgnum, "messages")
+		return
+	}
+}
+
 func TestSyslogRetryCount(t *testing.T) {
 	newRetryCount := uint(20)
 	os.Setenv("RETRY_COUNT", strconv.Itoa(int(newRetryCount)))