Add option to use nontransparent framing

Co-authored-by: Pavel Busko <[email protected]>

Author: pbusko

src/cmd/syslog-server/config.go

@@ -24,6 +24,8 @@ type Config struct {
 
 	SyslogClientTrustedCAFile string `env:"SYSLOG_CLIENT_TRUSTED_CA_FILE,  report"`
 
+	SyslogNonTransparentFraming bool `env:"SYSLOG_NON_TRANSPARENT_FRAMING, report"`
+
 	MetricsServer config.MetricsServer
 	UseRFC339     bool `env:"USE_RFC339"`
 }

src/cmd/syslog-server/main.go

@@ -84,6 +84,10 @@ func main() {
 		serverOptions = append(serverOptions, syslog.WithSyslogClientCA(cfg.SyslogClientTrustedCAFile))
 	}
 
+	if cfg.SyslogNonTransparentFraming {
+		serverOptions = append(serverOptions, syslog.WithNonTransparentFraming())
+	}
+
 	server := syslog.NewServer(
 		loggr,
 		m,

src/internal/syslog/server.go

@@ -12,6 +12,7 @@ import (
 	metrics "code.cloudfoundry.org/go-metric-registry"
 	"code.cloudfoundry.org/tlsconfig"
 	"github.com/leodido/go-syslog/v4"
+	"github.com/leodido/go-syslog/v4/nontransparent"
 	"github.com/leodido/go-syslog/v4/octetcounting"
 	"github.com/leodido/go-syslog/v4/rfc5424"
 
@@ -34,9 +35,11 @@ type Server struct {
 	idleTimeout           time.Duration
 	maxMessageLength      int
 	trimMessageWhitespace bool
+	nonTransparentFraming bool
 
 	ingress        metrics.Counter
 	invalidIngress metrics.Counter
+	newParserFunc  func(...syslog.ParserOption) syslog.Parser
 
 	loggr *log.Logger
 }
@@ -58,12 +61,17 @@ func NewServer(
 		idleTimeout:           2 * time.Minute,
 		maxMessageLength:      65 * 1024, // Diego should never send logs bigger than 64Kib
 		trimMessageWhitespace: true,
+		newParserFunc:         octetcounting.NewParser,
 	}
 
 	for _, o := range opts {
 		o(s)
 	}
 
+	if s.nonTransparentFraming {
+		s.newParserFunc = nontransparent.NewParser
+	}
+
 	s.ingress = m.NewCounter(
 		"ingress",
 		"Total syslog messages ingressed successfully.",
@@ -107,6 +115,12 @@ func WithSyslogClientCA(syslogClientCA string) ServerOption {
 	}
 }
 
+func WithNonTransparentFraming() ServerOption {
+	return func(s *Server) {
+		s.nonTransparentFraming = true
+	}
+}
+
 func WithIdleTimeout(d time.Duration) ServerOption {
 	return func(s *Server) {
 		s.idleTimeout = d
@@ -159,7 +173,7 @@ func (s *Server) handleConnection(conn net.Conn) {
 	defer conn.Close()
 	s.setReadDeadline(conn)
 
-	p := octetcounting.NewParser(
+	p := s.newParserFunc(
 		syslog.WithMaxMessageLength(s.maxMessageLength),
 		syslog.WithListener(s.parseListenerForConnection(conn)),
 	)

src/internal/syslog/server_test.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"log"
 	"net"
+	"strings"
 	"time"
 
 	"code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2"
@@ -27,6 +28,15 @@ const (
 	TIMER_MSG   = `128 <14>1 1970-01-01T00:00:00.012345+00:00 test-hostname test-app-id [APP/2] - [timer@47450 name="some-name" start="10" stop="20"] ` + "\n"
 )
 
+var (
+	// Same messages but without the MSGLEN
+	_, LOG_MSG_NT, _     = strings.Cut(LOG_MSG, " ")
+	_, COUNTER_MSG_NT, _ = strings.Cut(COUNTER_MSG, " ")
+	_, GAUGE_MSG_NT, _   = strings.Cut(GAUGE_MSG, " ")
+	_, EVENT_MSG_NT, _   = strings.Cut(EVENT_MSG, " ")
+	_, TIMER_MSG_NT, _   = strings.Cut(TIMER_MSG, " ")
+)
+
 var _ = Describe("Server", func() {
 	var (
 		spyRegistry *testhelpers.SpyMetricsRegistry
@@ -221,6 +231,89 @@ var _ = Describe("Server", func() {
 			Entry("explicitly enabled", syslog.WithServerTrimMessageWhitespace(true), "just a test with with whitespace"),
 		)
 
+		DescribeTableSubtree("nontransparent framing",
+			func(msg string, expected *loggregator_v2.Envelope) {
+				BeforeEach(func() {
+					serverOpts = append(serverOpts, syslog.WithNonTransparentFraming())
+				})
+
+				It("correctly parses RFC5424 formatted messages without MSGLEN", func() {
+					_, err := fmt.Fprint(clientConn, msg)
+					Expect(err).ToNot(HaveOccurred())
+
+					Expect(server.Stream(context.Background(), &loggregator_v2.EgressBatchRequest{})()).Should(ContainElement(expected))
+				})
+			},
+			Entry("log", LOG_MSG_NT, &loggregator_v2.Envelope{
+				Tags: map[string]string{
+					"source_type": "actual-source-type",
+					"key":         "value",
+				},
+				InstanceId: "2",
+				Timestamp:  12345000,
+				SourceId:   "test-app-id",
+				Message: &loggregator_v2.Envelope_Log{
+					Log: &loggregator_v2.Log{
+						Payload: []byte("just a test"),
+						Type:    loggregator_v2.Log_OUT,
+					},
+				},
+			}),
+			Entry("counter", COUNTER_MSG_NT, &loggregator_v2.Envelope{
+				InstanceId: "1",
+				Timestamp:  12345000,
+				SourceId:   "test-app-id",
+				Tags: map[string]string{
+					"key": "value",
+				},
+				Message: &loggregator_v2.Envelope_Counter{
+					Counter: &loggregator_v2.Counter{
+						Name:  "test",
+						Delta: 5,
+						Total: 10,
+					},
+				},
+			}),
+			Entry("gauge", GAUGE_MSG_NT, &loggregator_v2.Envelope{
+				InstanceId: "3",
+				Timestamp:  12345000,
+				SourceId:   "test-app-id",
+				Tags:       map[string]string{},
+				Message: &loggregator_v2.Envelope_Gauge{
+					Gauge: &loggregator_v2.Gauge{
+						Metrics: map[string]*loggregator_v2.GaugeValue{
+							"cpu": {Unit: "percentage", Value: 0.23},
+						},
+					},
+				},
+			}),
+			Entry("event", EVENT_MSG_NT, &loggregator_v2.Envelope{
+				InstanceId: "2",
+				Timestamp:  12345000,
+				SourceId:   "test-app-id",
+				Tags:       map[string]string{},
+				Message: &loggregator_v2.Envelope_Event{
+					Event: &loggregator_v2.Event{
+						Title: "event-title",
+						Body:  "event-body",
+					},
+				},
+			}),
+			Entry("timer", TIMER_MSG_NT, &loggregator_v2.Envelope{
+				InstanceId: "2",
+				Timestamp:  12345000,
+				SourceId:   "test-app-id",
+				Tags:       map[string]string{},
+				Message: &loggregator_v2.Envelope_Timer{
+					Timer: &loggregator_v2.Timer{
+						Name:  "some-name",
+						Start: 10,
+						Stop:  20,
+					},
+				},
+			}),
+		)
+
 		Context("when max message length is exceeded", func() {
 			BeforeEach(func() {
 				serverOpts = append(serverOpts, syslog.WithServerMaxMessageLength(128))