Add Tag parameter to fluentbit syslog input

Signed-off-by: Leon Low <[email protected]>

Author: LeonLow97

apis/fluentbit/v1alpha2/plugins/input/syslog.go

@@ -39,6 +39,8 @@ type Syslog struct {
 	ReceiveBufferSize string `json:"receiveBufferSize,omitempty"`
 	// Specify the key where the source address will be injected.
 	SourceAddressKey string `json:"sourceAddressKey,omitempty"`
+	// Specify a tag to route incoming logs through different parsers to different outputs.
+	Tag string `json:"tag,omitempty"`
 	// Specify TLS connector options.
 	*plugins.TLS `json:"tls,omitempty"`
 }
@@ -58,6 +60,7 @@ func (s *Syslog) Params(sl plugins.SecretLoader) (*params.KVs, error) {
 	plugins.InsertKVString(kvs, "Buffer_Max_Size", s.BufferMaxSize)
 	plugins.InsertKVString(kvs, "Receive_Buffer_Size", s.ReceiveBufferSize)
 	plugins.InsertKVString(kvs, "Source_Address_Key", s.SourceAddressKey)
+	plugins.InsertKVString(kvs, "Tag", s.Tag)
 
 	plugins.InsertKVField(kvs, "Port", s.Port)
 	plugins.InsertKVField(kvs, "Unix_Perm", s.UnixPerm)

apis/fluentbit/v1alpha2/plugins/input/syslog_test.go

@@ -0,0 +1,52 @@
+package input
+
+import (
+	"testing"
+
+	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/v3/apis/fluentbit/v1alpha2/plugins/params"
+	"github.com/fluent/fluent-operator/v3/pkg/utils"
+	. "github.com/onsi/gomega"
+)
+
+func TestSyslog_Name(t *testing.T) {
+	g := NewGomegaWithT(t)
+	syslog := Syslog{}
+	g.Expect(syslog.Name()).To(Equal("syslog"))
+}
+
+func TestSyslog_Params(t *testing.T) {
+	g := NewGomegaWithT(t)
+	sl := plugins.NewSecretLoader(nil, "test namespace")
+
+	syslog := Syslog{
+		Mode:              "tcp",
+		Listen:            "0.0.0.0",
+		Port:              utils.ToPtr[int32](514),
+		Path:              "/tmp/syslog.sock",
+		UnixPerm:          utils.ToPtr[int32](644),
+		Parser:            "syslog-rfc5424",
+		BufferChunkSize:   "32KB",
+		BufferMaxSize:     "256KB",
+		ReceiveBufferSize: "1MB",
+		SourceAddressKey:  "source_address",
+		Tag:               "syslog.tag",
+	}
+
+	expected := params.NewKVs()
+	expected.Insert("Mode", "tcp")
+	expected.Insert("Listen", "0.0.0.0")
+	expected.Insert("Path", "/tmp/syslog.sock")
+	expected.Insert("Parser", "syslog-rfc5424")
+	expected.Insert("Buffer_Chunk_Size", "32KB")
+	expected.Insert("Buffer_Max_Size", "256KB")
+	expected.Insert("Receive_Buffer_Size", "1MB")
+	expected.Insert("Source_Address_Key", "source_address")
+	expected.Insert("Tag", "syslog.tag")
+	expected.Insert("Port", "514")
+	expected.Insert("Unix_Perm", "644")
+
+	kvs, err := syslog.Params(sl)
+	g.Expect(err).NotTo(HaveOccurred())
+	g.Expect(kvs).To(Equal(expected))
+}

charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml

@@ -578,6 +578,10 @@ spec:
                     description: Specify the key where the source address will be
                       injected.
                     type: string
+                  tag:
+                    description: Specify a tag to route incoming logs through different
+                      parsers to different outputs.
+                    type: string
                   tls:
                     description: Specify TLS connector options.
                     properties:

config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml

@@ -578,6 +578,10 @@ spec:
                     description: Specify the key where the source address will be
                       injected.
                     type: string
+                  tag:
+                    description: Specify a tag to route incoming logs through different
+                      parsers to different outputs.
+                    type: string
                   tls:
                     description: Specify TLS connector options.
                     properties:

docs/plugins/fluentbit/input/syslog.md

@@ -15,4 +15,5 @@ Syslog input plugins allows to collect Syslog messages through a Unix socket ser
 | bufferMaxSize | Specify the maximum buffer size to receive a Syslog message. If not set, the default size will be the value of Buffer_Chunk_Size. | string |
 | receiveBufferSize | Specify the maximum socket receive buffer size. If not set, the default value is OS-dependant, but generally too low to accept thousands of syslog messages per second without loss on udp or unix_udp sockets. Note that on Linux the value is capped by sysctl net.core.rmem_max. | string |
 | sourceAddressKey | Specify the key where the source address will be injected. | string |
+| tag | Specify a tag to route incoming logs through different parsers to different outputs. | string |
 | tls | Specify TLS connector options. | *[plugins.TLS](../tls.md) |

manifests/setup/fluent-operator-crd.yaml

@@ -2705,6 +2705,10 @@ spec:
                     description: Specify the key where the source address will be
                       injected.
                     type: string
+                  tag:
+                    description: Specify a tag to route incoming logs through different
+                      parsers to different outputs.
+                    type: string
                   tls:
                     description: Specify TLS connector options.
                     properties:

manifests/setup/setup.yaml

@@ -2705,6 +2705,10 @@ spec:
                     description: Specify the key where the source address will be
                       injected.
                     type: string
+                  tag:
+                    description: Specify a tag to route incoming logs through different
+                      parsers to different outputs.
+                    type: string
                   tls:
                     description: Specify TLS connector options.
                     properties:

pkg/utils/utils.go

@@ -6,6 +6,10 @@ import (
 	"strings"
 )
 
+func ToPtr[T any](v T) *T {
+	return &v
+}
+
 func HashCode(msg string) string {
 	var h = md5.New()
 	h.Write([]byte(msg))