[exporter/syslog] send syslog messages in batches (open-telemetry#27799)

**Description:**

This changes the behavior of the Syslog exporter to send each batch of
Syslog messages in a single request (with messages separated by
newlines), instead of sending each message in a separate request and
closing the connection after each message.

The batching only happens when using TCP. For UDP, each syslog message
is still sent in a separate request, as defined by [the
spec](https://datatracker.ietf.org/doc/html/rfc5426#section-3.1).

This also significantly refactors (and hopefully simplifies) the
exporter's code, extracting the code that formats the syslog messages
from the `sender` type into separate `formatter` types. Hopefully this
will make the development of this component easier.

**Link to tracking Issue:**

-
open-telemetry#21244

**Testing:**

The unit tests have been updated to reflect the refactored codebase. The
integration tests introduced in
open-telemetry#27464
are unchanged, as the format of the output messages hasn't changed.

**Documentation:**

No documentation updates.

.chloggen/syslog-exporter-single-request.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: exporter/syslog
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: send syslog messages in batches
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [21244]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: This changes the behavior of the Syslog exporter to send each batch of Syslog messages in a single request (with messages separated by newlines), instead of sending each message in a separate request and closing the connection after each message.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/syslogexporter/exporter.go

@@ -8,12 +8,10 @@ import (
  "crypto/tls"
  "fmt"
  "strings"
- "time"
 
  "go.opentelemetry.io/collector/consumer/consumererror"
  "go.opentelemetry.io/collector/exporter"
  "go.opentelemetry.io/collector/exporter/exporterhelper"
- "go.opentelemetry.io/collector/pdata/pcommon"
  "go.opentelemetry.io/collector/pdata/plog"
  "go.uber.org/multierr"
  "go.uber.org/zap"
@@ -23,6 +21,7 @@ type syslogexporter struct {
  config *Config
  logger *zap.Logger
  tlsConfig *tls.Config
+ formatter formatter
 }
 
 func initExporter(cfg *Config, createSettings exporter.CreateSettings) (*syslogexporter, error) {
@@ -37,11 +36,12 @@ func initExporter(cfg *Config, createSettings exporter.CreateSettings) (*sysloge
  config: cfg,
  logger: createSettings.Logger,
  tlsConfig: tlsConfig,
+ formatter: createFormatter(cfg.Protocol),
  }
 
  s.logger.Info("Syslog Exporter configured",
  zap.String("endpoint", cfg.Endpoint),
- zap.String("Protocol", cfg.Protocol),
+ zap.String("protocol", cfg.Protocol),
  zap.Int("port", cfg.Port),
  )
 
@@ -69,78 +69,77 @@ func newLogsExporter(
  )
 }
 
-func (se *syslogexporter) logsToMap(record plog.LogRecord) map[string]any {
- attributes := record.Attributes().AsRaw()
- return attributes
-}
-
-func (se *syslogexporter) getTimestamp(record plog.LogRecord) time.Time {
- timestamp := record.Timestamp().AsTime()
- return timestamp
+func (se *syslogexporter) pushLogsData(_ context.Context, logs plog.Logs) error {
+ batchMessages := strings.ToLower(se.config.Network) == "tcp"
+ var err error
+ if batchMessages {
+ err = se.exportBatch(logs)
+ } else {
+ err = se.exportNonBatch(logs)
+ }
+ return err
 }
 
-func (se *syslogexporter) pushLogsData(_ context.Context, ld plog.Logs) error {
- type droppedResourceRecords struct {
- resource pcommon.Resource
- records []plog.LogRecord
- }
- var (
- errs []error
- dropped []droppedResourceRecords
- )
- rls := ld.ResourceLogs()
- for i := 0; i < rls.Len(); i++ {
- rl := rls.At(i)
- if droppedRecords, err := se.sendSyslogs(rl); err != nil {
- dropped = append(dropped, droppedResourceRecords{
- resource: rl.Resource(),
- records: droppedRecords,
- })
- errs = append(errs, err)
+func (se *syslogexporter) exportBatch(logs plog.Logs) error {
+ var payload strings.Builder
+ for i := 0; i < logs.ResourceLogs().Len(); i++ {
+ resourceLogs := logs.ResourceLogs().At(i)
+ for j := 0; j < resourceLogs.ScopeLogs().Len(); j++ {
+ scopeLogs := resourceLogs.ScopeLogs().At(j)
+ for k := 0; k < scopeLogs.LogRecords().Len(); k++ {
+ logRecord := scopeLogs.LogRecords().At(k)
+ formatted := se.formatter.format(logRecord)
+ payload.WriteString(formatted)
+ }
  }
  }
- if len(dropped) > 0 {
- ld = plog.NewLogs()
- for i := range dropped {
- rls := ld.ResourceLogs().AppendEmpty()
- logRecords := rls.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty()
- dropped[i].resource.MoveTo(rls.Resource())
- for j := 0; j < len(dropped[i].records); j++ {
- dropped[i].records[j].MoveTo(logRecords)
- }
+
+ if payload.Len() > 0 {
+ sender, err := connect(se.logger, se.config, se.tlsConfig)
+ if err != nil {
+ return consumererror.NewLogs(err, logs)
+ }
+ defer sender.close()
+ err = sender.Write(payload.String())
+ if err != nil {
+ return consumererror.NewLogs(err, logs)
  }
- errs = deduplicateErrors(errs)
- return consumererror.NewLogs(multierr.Combine(errs...), ld)
  }
- se.logger.Info("Connected successfully, exporting logs....")
  return nil
 }
 
-func (se *syslogexporter) sendSyslogs(rl plog.ResourceLogs) ([]plog.LogRecord, error) {
- var (
- errs []error
- droppedRecords []plog.LogRecord
- )
- slgs := rl.ScopeLogs()
- for i := 0; i < slgs.Len(); i++ {
- slg := slgs.At(i)
- for j := 0; j < slg.LogRecords().Len(); j++ {
- lr := slg.LogRecords().At(j)
- formattedLine := se.logsToMap(lr)
- timestamp := se.getTimestamp(lr)
- s, errConn := connect(se.logger, se.config, se.tlsConfig)
- if errConn != nil {
- droppedRecords = append(droppedRecords, lr)
- errs = append(errs, errConn)
- continue
- }
- defer s.close()
- err := s.Write(formattedLine, timestamp)
- if err != nil {
- droppedRecords = append(droppedRecords, lr)
- errs = append(errs, err)
+func (se *syslogexporter) exportNonBatch(logs plog.Logs) error {
+ sender, err := connect(se.logger, se.config, se.tlsConfig)
+ if err != nil {
+ return consumererror.NewLogs(err, logs)
+ }
+ defer sender.close()
+
+ errs := []error{}
+ droppedLogs := plog.NewLogs()
+ for i := 0; i < logs.ResourceLogs().Len(); i++ {
+ resourceLogs := logs.ResourceLogs().At(i)
+ droppedResourceLogs := droppedLogs.ResourceLogs().AppendEmpty()
+ for j := 0; j < resourceLogs.ScopeLogs().Len(); j++ {
+ scopeLogs := resourceLogs.ScopeLogs().At(j)
+ droppedScopeLogs := droppedResourceLogs.ScopeLogs().AppendEmpty()
+ for k := 0; k < scopeLogs.LogRecords().Len(); k++ {
+ logRecord := scopeLogs.LogRecords().At(k)
+ formatted := se.formatter.format(logRecord)
+ err = sender.Write(formatted)
+ if err != nil {
+ errs = append(errs, err)
+ droppedLogRecord := droppedScopeLogs.LogRecords().AppendEmpty()
+ logRecord.CopyTo(droppedLogRecord)
+ }
  }
  }
  }
- return droppedRecords, multierr.Combine(errs...)
+
+ if len(errs) > 0 {
+ errs = deduplicateErrors(errs)
+ return consumererror.NewLogs(multierr.Combine(errs...), droppedLogs)
+ }
+
+ return nil
 }

exporter/syslogexporter/formatter.go

@@ -0,0 +1,29 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package syslogexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/syslogexporter"
+
+import (
+ "go.opentelemetry.io/collector/pdata/plog"
+)
+
+func createFormatter(protocol string) formatter {
+ if protocol == protocolRFC5424Str {
+ return newRFC5424Formatter()
+ }
+ return newRFC3164Formatter()
+}
+
+type formatter interface {
+ format(plog.LogRecord) string
+}
+
+// getAttributeValueOrDefault returns the value of the requested log record's attribute as a string.
+// If the attribute was not found, it returns the provided default value.
+func getAttributeValueOrDefault(logRecord plog.LogRecord, attributeName string, defaultValue string) string {
+ value := defaultValue
+ if attributeValue, found := logRecord.Attributes().Get(attributeName); found {
+ value = attributeValue.AsString()
+ }
+ return value
+}

exporter/syslogexporter/rfc3164_formatter.go

@@ -0,0 +1,56 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package syslogexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/syslogexporter"
+
+import (
+ "fmt"
+ "strconv"
+
+ "go.opentelemetry.io/collector/pdata/plog"
+)
+
+type rfc3164Formatter struct {
+}
+
+func newRFC3164Formatter() *rfc3164Formatter {
+ return &rfc3164Formatter{}
+}
+
+func (f *rfc3164Formatter) format(logRecord plog.LogRecord) string {
+ priorityString := f.formatPriority(logRecord)
+ timestampString := f.formatTimestamp(logRecord)
+ hostnameString := f.formatHostname(logRecord)
+ appnameString := f.formatAppname(logRecord)
+ messageString := f.formatMessage(logRecord)
+ appnameMessageDelimiter := ""
+ if len(appnameString) > 0 && messageString != emptyMessage {
+ appnameMessageDelimiter = " "
+ }
+ formatted := fmt.Sprintf("<%s>%s %s %s%s%s\n", priorityString, timestampString, hostnameString, appnameString, appnameMessageDelimiter, messageString)
+ return formatted
+}
+
+func (f *rfc3164Formatter) formatPriority(logRecord plog.LogRecord) string {
+ return getAttributeValueOrDefault(logRecord, priority, strconv.Itoa(defaultPriority))
+}
+
+func (f *rfc3164Formatter) formatTimestamp(logRecord plog.LogRecord) string {
+ return logRecord.Timestamp().AsTime().Format("Jan 02 15:04:05")
+}
+
+func (f *rfc3164Formatter) formatHostname(logRecord plog.LogRecord) string {
+ return getAttributeValueOrDefault(logRecord, hostname, emptyValue)
+}
+
+func (f *rfc3164Formatter) formatAppname(logRecord plog.LogRecord) string {
+ value := getAttributeValueOrDefault(logRecord, app, "")
+ if value != "" {
+ value += ":"
+ }
+ return value
+}
+
+func (f *rfc3164Formatter) formatMessage(logRecord plog.LogRecord) string {
+ return getAttributeValueOrDefault(logRecord, message, emptyMessage)
+}

exporter/syslogexporter/rfc3164_formatter_test.go

@@ -0,0 +1,41 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package syslogexporter
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "go.opentelemetry.io/collector/pdata/pcommon"
+ "go.opentelemetry.io/collector/pdata/plog"
+)
+
+func TestRFC3164Formatter(t *testing.T) {
+ expected := "<34>Aug 24 05:14:15 mymachine su: 'su root' failed for lonvick on /dev/pts/8\n"
+ logRecord := plog.NewLogRecord()
+ logRecord.Attributes().PutStr("appname", "su")
+ logRecord.Attributes().PutStr("hostname", "mymachine")
+ logRecord.Attributes().PutStr("message", "'su root' failed for lonvick on /dev/pts/8")
+ logRecord.Attributes().PutInt("priority", 34)
+ timestamp, err := time.Parse(time.RFC3339Nano, "2003-08-24T05:14:15.000003Z")
+ require.NoError(t, err)
+ logRecord.SetTimestamp(pcommon.NewTimestampFromTime(timestamp))
+
+ actual := newRFC3164Formatter().format(logRecord)
+ assert.NoError(t, err)
+ assert.Equal(t, expected, actual)
+
+ expected = "<165>Aug 24 05:14:15 - -\n"
+ logRecord = plog.NewLogRecord()
+ logRecord.Attributes().PutStr("message", "-")
+ timestamp, err = time.Parse(time.RFC3339Nano, "2003-08-24T05:14:15.000003Z")
+ require.NoError(t, err)
+ logRecord.SetTimestamp(pcommon.NewTimestampFromTime(timestamp))
+
+ actual = newRFC3164Formatter().format(logRecord)
+ assert.NoError(t, err)
+ assert.Equal(t, expected, actual)
+}