Add optional TCP framing to syslog adapter

Enable users to specify "octet-counted" frames as described in RFC6587
(Syslog over TCP) 3.4.1 and RFC5424 (Syslog over TLS). This prefixes
each message with the length of the message to allow consumers to easily
determine where the message ends (rather than traditional LF framing).
This also enables multiline Syslog messages without escaping.

This keeps the default as LF framing for backwards compatibility though
octet-counted framing is preferred when it is known the downstream
consumer can handle it.

Author: jszwedko

adapters/syslog/syslog.go

@@ -2,6 +2,7 @@ package syslog
 
 import (
 	"bytes"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -24,9 +25,17 @@ const defaultRetryCount = 10
 var (
 	hostname         string
 	retryCount       uint
+	tcpFraming       TCPFraming
 	econnResetErrStr string
 )
 
+type TCPFraming string
+
+const (
+	TraditionalTCPFraming  TCPFraming = "traditional"   // LF framing
+	OctetCountedTCPFraming            = "octet-counted" // https://tools.ietf.org/html/rfc6587#section-3.4.1
+)
+
 func init() {
 	hostname, _ = os.Hostname()
 	econnResetErrStr = fmt.Sprintf("write: %s", syscall.ECONNRESET.Error())
@@ -89,6 +98,17 @@ func NewSyslogAdapter(route *router.Route) (router.LogAdapter, error) {
 		structuredData = fmt.Sprintf("[%s]", structuredData)
 	}
 
+	if isTCPConnecion(conn) {
+		switch s := cfg.GetEnvDefault("SYSLOG_TCP_FRAMING", "traditional"); s {
+		case "traditional":
+			tcpFraming = TraditionalTCPFraming
+		case "octet-counted":
+			tcpFraming = OctetCountedTCPFraming
+		default:
+			return nil, fmt.Errorf("unknown SYSLOG_TCP_FRAMING value: %s", s)
+		}
+	}
+
 	var tmplStr string
 	switch format {
 	case "rfc5424":
@@ -137,6 +157,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 +259,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)))