Update go and dependencies versions

Dockerfile

@@ -1,17 +1,19 @@
-FROM golang:1.13  as building-stage
+FROM golang:1.20-bullseye as building-stage
 
-RUN go get github.com/fluent/fluent-bit-go/output && \ 
-    go get github.com/streadway/amqp
+RUN go install github.com/fluent/fluent-bit-go/output@latest; exit 0 && \ 
+    go install github.com/rabbitmq/amqp091-go@latest; exit 0
+
 
 COPY ./*.go /go/src/
-
+COPY ./go.mod /go/src/
+COPY ./go.sum /go/src/
 COPY ./Makefile /go/src
 
 WORKDIR /go/src
 
 RUN make
 
-FROM fluent/fluent-bit:1.3
+FROM fluent/fluent-bit:2.1
 
 LABEL maintainer="Björn Franke"
 

README.md

@@ -25,6 +25,7 @@ build
 | RoutingKey                   | The routing-key pattern                                                                         | ""            |
 | RoutingKeyDelimiter          | The Delemiter which seperates the routing-key parts                                             |  "."          |
 | RemoveRkValuesFromRecord     | If enabled fluentd deletes the values of the record,  which have been stored in the routing-key | ""            |
+| ContentEncoding              | Sets the Content Encoding if needed                                                             | ""            |
 
 ### Routing-Key pattern
 

conf/fluent-bit-docker.conf

@@ -20,4 +20,5 @@
     RoutingKey  $["loglevel_3"].$["loglevel_1"][2]["sublevel"][0].$["loglevel_2"]["info_loglevel"]
     RemoveRkValuesFromRecord true
     AddTagToRecord false
-    AddTimestampToRecord true
+    AddTimestampToRecord true
+    ContentEncoding utf8

go.mod

@@ -0,0 +1,10 @@
+module github.com/igd-geo/fluent-bit-rabbitmq-output
+
+go 1.20
+
+require (
+        github.com/fluent/fluent-bit-go v0.0.0-20230731091245-a7a013e2473c
+        github.com/rabbitmq/amqp091-go v1.9.0
+)
+
+require github.com/ugorji/go/codec v1.1.7 // indirect

go.sum

@@ -0,0 +1,23 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fluent/fluent-bit-go v0.0.0-20230731091245-a7a013e2473c h1:yKN46XJHYC/gvgH2UsisJ31+n4K3S7QYZSfU2uAWjuI=
+github.com/fluent/fluent-bit-go v0.0.0-20230731091245-a7a013e2473c/go.mod h1:L92h+dgwElEyUuShEwjbiHjseW410WIcNz+Bjutc8YQ=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rabbitmq/amqp091-go v1.9.0 h1:qrQtyzB4H8BQgEuJwhmVQqVHB9O4+MNDJCCAcpc3Aoo=
+github.com/rabbitmq/amqp091-go v1.9.0/go.mod h1:+jPrT9iY2eLjRaMSRHUhc3z14E/l85kv/f+6luSD3pc=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
+github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
+github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

out_rabbitmq.go

@@ -8,7 +8,7 @@ import (
 	"unsafe"
 
 	"github.com/fluent/fluent-bit-go/output"
-	"github.com/streadway/amqp"
+	amqp "github.com/rabbitmq/amqp091-go"
 )
 
 var (
@@ -20,6 +20,7 @@ var (
 	removeRkValuesFromRecord bool
 	addTagToRecord           bool
 	addTimestampToRecord     bool
+	contentEncoding          string
 )
 
 //export FLBPluginRegister
@@ -44,6 +45,7 @@ func FLBPluginInit(plugin unsafe.Pointer) int {
 	removeRkValuesFromRecordStr := output.FLBPluginConfigKey(plugin, "RemoveRkValuesFromRecord")
 	addTagToRecordStr := output.FLBPluginConfigKey(plugin, "AddTagToRecord")
 	addTimestampToRecordStr := output.FLBPluginConfigKey(plugin, "AddTimestampToRecord")
+	contentEncoding := output.FLBPluginConfigKey(plugin, "ContentEncoding")
 
 	if len(routingKeyDelimiter) < 1 {
 		routingKeyDelimiter = "."
@@ -74,6 +76,10 @@ func FLBPluginInit(plugin unsafe.Pointer) int {
 		return output.FLB_ERROR
 	}
 
+	if len(contentEncoding) < 1 {
+		contentEncoding = ""
+	}
+
 	connection, err = amqp.Dial("amqp://" + user + ":" + password + "@" + host + ":" + port + "/")
 	if err != nil {
 		logError("Failed to establish a connection to RabbitMQ: ", err)
@@ -152,8 +158,9 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
 			false,        // mandatory
 			false,        // immediate
 			amqp.Publishing{
-				ContentType: "application/json",
-				Body:        jsonString,
+				ContentType:     "application/json",
+				ContentEncoding: contentEncoding,
+				Body:            jsonString,
 			})
 		if err != nil {
 			logError("Couldn't publish record: ", err)