Fix sockets left in CLOSE_WAIT and cleanups

In https://lab.nexedi.com/nexedi/slapos/blob/7102c878b2bda5a56be31e02b0cd90914b291478/component/fluentbit-plugin-wendelin/buildout.cfg#L15 we are using tag 0.2.1 but master branch was something completely different. This merges 0.2.1 and remove some unused/obsolete code that was still on master.

This fixes the problem of leaked files and CLOSE_WAIT, the fix is actually very simple ( b2052811 ), the responses were never closed, so each ingested line leaked a file descriptor.

This also removes an "UnsecureSkipVerify" TLS option, I don't see any reason to use it.

See merge request !2

Makefile

+
+ifeq ($(PREFIX),)
+	PREFIX=/opt/fluentbit-plugin-wendelin
+endif
+
+.PHONY: all
+all: build
+
+build:
+	go build -v -buildmode=c-shared -o build/libfluentbit_wendelin.so src/fluentbit_wendelin.go
+
+.PHONY: install
+install: all
+	install -d $(DESTDIR)$(PREFIX)/etc/
+	install -d $(DESTDIR)$(PREFIX)/include/
+	install -d $(DESTDIR)$(PREFIX)/lib/
+	cp src/configuration-files/flb.conf.in	$(DESTDIR)$(PREFIX)/etc/
+	cp build/libfluentbit_wendelin.h	$(DESTDIR)$(PREFIX)/include/
+	cp build/libfluentbit_wendelin.so	$(DESTDIR)$(PREFIX)/lib/
+
+.PHONY: uninstall
+uninstall:
+	rm -f $(DESTDIR)$(PREFIX)/etc/flb.conf.in
+	rm -f $(DESTDIR)$(PREFIX)/include/libfluentbit_wendelin.h
+	rm -f $(DESTDIR)$(PREFIX)/lib/libfluentbit_wendelin.so
+
+.PHONY: clean
+clean:
+	rm -rf build/
+
+
+# export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$(pwd)
+.PHONY: test
+test:
+	gcc -L. -lfluentbit_wendelin test/test.c

README.md

-fluentbit-plugin-wendelin
-======================
+output plugin for fluentbit
 
+more :
 
-This is output plugin for [Fluentbit][] to forward data to [Wendelin][] system.
-
-[Fluentbit]:  http://fluentbit.io/
-[Wendelin]: http://wendelin.io/
-
-See *example/to_wendelin.conf* for fully setup example.
\ No newline at end of file
+https://docs.fluentbit.io/manual/development/golang-output-plugins
+https://github.com/fluent/fluent-bit-go

example/to_wendelin.conf

-[INPUT]
-    Name  dummy
-    Tag   dummy
-
-[OUTPUT]
-    Name  wendelin_out
-    Match *
-    User <your_wendelin_user>
-    Password <your_wendelin_password>
-    Uri   <Wendelin_URL>/erp5/portal_ingestion_policies/<YOUR_INGESTION_POLICY_ID>
-    Reference <input_stream_reference>
-

go.mod

+module fluentbit-plugin-wendelin
+
+go 1.16
+
+require github.com/fluent/fluent-bit-go v0.0.0-20201210173045-3fd1e0486df2 // indirect

go.sum

+github.com/fluent/fluent-bit-go v0.0.0-20201210173045-3fd1e0486df2 h1:G57WNyWS0FQf43hjRXLy5JT1V5LWVsSiEpkUcT67Ugk=
+github.com/fluent/fluent-bit-go v0.0.0-20201210173045-3fd1e0486df2/go.mod h1:L92h+dgwElEyUuShEwjbiHjseW410WIcNz+Bjutc8YQ=
+github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=
+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=

out_http.go

-package main
-
-import "github.com/fluent/fluent-bit-go/output"
-import (
-	"C"
-	"bytes"
-	"fmt"
-	"net/http"
-	"regexp"
-	"strconv"
-	"unsafe"
-)
-
-// configuration parameters
-var user string
-var password string
-var uri string
-var reference string
-
-//export FLBPluginRegister
-func FLBPluginRegister(ctx unsafe.Pointer) int {
-	return output.FLBPluginRegister(ctx, "wendelin_out", "Wendelin Out GO!")
-}
-
-//export FLBPluginInit
-// (fluentbit will call this)
-// ctx (context) pointer to fluentbit context (state/ c code)
-func FLBPluginInit(ctx unsafe.Pointer) int {
-	// Example to retrieve an optional configuration parameter
-	//	param := output.FLBPluginConfigKey(ctx, "param")
-	user = output.FLBPluginConfigKey(ctx, "User")
-	password = output.FLBPluginConfigKey(ctx, "Password")
-	uri = output.FLBPluginConfigKey(ctx, "Uri")
-	reference = output.FLBPluginConfigKey(ctx, "Reference")
-
-	//fmt.Printf("[flb-go] plugin parameter = '%s'\n", param)
-	fmt.Printf("[flb-go user] plugin parameter = '%s'\n", user)
-	fmt.Printf("[flb-go password] plugin parameter = '%s'\n", password)
-	fmt.Printf("[flb-go uri] plugin parameter = '%s'\n", uri)
-	fmt.Printf("[flb-go reference] plugin parameter = '%s'\n", reference)
-
-	return output.FLB_OK
-}
-
-//export FLBPluginFlush
-func FLBPluginFlush(data unsafe.Pointer, length C.int, tag *C.char) int {
-
-	request_string := uri + "/ingest?reference=" + reference
-	var b []byte
-	b = C.GoBytes(data, C.int(length))
-
-	hc := http.Client{}
-
-	req, err := http.NewRequest("POST", request_string, bytes.NewBuffer(b))
-
-	if err != nil {
-		return output.FLB_ERROR
-	}
-
-	req.Header.Set("Content-Type", "application/octet-stream")
-	req.SetBasicAuth(user, password)
-
-	resp, err := hc.Do(req)
-
-	if err != nil {
-		return output.FLB_ERROR
-	}
-
-	/*
-	 * Only allow the following HTTP status:
-	 *
-	 *  - 200: OK
-	 *  - 201: Created
-	 *  - 202: Accepted
-	 *  - 203: no authorative resp
-	 *  - 204: No Content
-	 *  - 205: Reset content
-	 */
-	re := regexp.MustCompile("[0-9]+") // get only the status code
-	status_code := re.FindAllString(resp.Status, -1)
-	resp_status, err := strconv.Atoi(status_code[0])
-	if err != nil {
-		fmt.Println(err)
-		return output.FLB_RETRY
-	}
-
-	fmt.Println(resp.Status)
-	fmt.Println(err)
-
-	if resp_status < 200 && resp_status > 205 {
-		return output.FLB_RETRY
-	}
-
-	defer resp.Body.Close()
-
-	/*
-	 * Return options:
-	 *
-	 * - output.FLB_OK    = data have been processed.
-	 * - output.FLB_ERROR = unrecoverable error, do not try this again.
-	 * - output.FLB_RETRY = retry to flush later.
-	 */
-	return output.FLB_OK
-}
-
-//export FLBPluginExit
-func FLBPluginExit() int {
-	return output.FLB_OK
-}
-
-func main() {
-}

src/configuration-files/flb.conf.in

+[SERVICE]
+  flush 5
+
+[INPUT]
+  name tail
+  path %SCAN_FILE%
+  refresh_interval 2
+
+[output]
+  name fluentbit_wendelin
+  match *
+  streamtool_uri %WENDELIN_URL%
+  user %WENDELIN_USER%
+  password %WENDELIN_PWD%
+  buffer_type memory
+  flush_interval 60s
+  disable_retry_limit true
+  reference %WENDELIN_REFERENCE%

src/fluentbit_wendelin.go

+package main
+
+import (
+	"C"
+	"bytes"
+	"crypto/tls"
+	"fmt"
+	"github.com/fluent/fluent-bit-go/output"
+	"net/http"
+	"os"
+	"strings"
+	"unsafe"
+)
+
+//export FLBPluginRegister
+func FLBPluginRegister(def unsafe.Pointer) int {
+	return output.FLBPluginRegister(def, "fluentbit_wendelin", "Fluentbit output plugin for wendelin")
+}
+
+//export FLBPluginInit
+// (fluentbit will call this)
+// plugin (context) pointer to fluentbit context (state/ c code)
+func FLBPluginInit(plugin unsafe.Pointer) int {
+	streamtool_uri := output.FLBPluginConfigKey(plugin, "streamtool_uri")
+	user := output.FLBPluginConfigKey(plugin, "user")
+	password := output.FLBPluginConfigKey(plugin, "password")
+	buffer_type := output.FLBPluginConfigKey(plugin, "buffer_type")
+	flush_interval := output.FLBPluginConfigKey(plugin, "flush_interval")
+	disable_retry_limit := output.FLBPluginConfigKey(plugin, "disable_retry_limit")
+	reference := output.FLBPluginConfigKey(plugin, "reference")
+	dict := map[string]string{
+		"streamtool_uri":      streamtool_uri,
+		"user":                user,
+		"password":            password,
+		"buffer_type":         buffer_type,
+		"flush_interval":      flush_interval,
+		"disable_retry_limit": disable_retry_limit,
+		"reference":           reference,
+	}
+	output.FLBPluginSetContext(plugin, dict)
+	return output.FLB_OK
+}
+
+//export FLBPluginFlushCtx
+func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int {
+	var ret int
+	var record map[interface{}]interface{}
+	// Create Fluent Bit decoder
+	dec := output.NewDecoder(data, int(length))
+	dict := output.FLBPluginGetContext(ctx).(map[string]string)
+	// Iterate Records
+	var result string
+	result = ""
+	var is_end bool = false
+	for {
+		// Extract Record
+		ret, _, record = output.GetRecord(dec)
+		if ret != 0 {
+			break
+		}
+
+		// Print record keys and values
+		for _, v := range record {
+			var output_string string = ""
+			for _, s := range v.([]uint8) {
+				output_string = output_string + string(s)
+			}
+			if strings.Contains(output_string, "fluentbit_end") {
+				is_end = true
+			}
+			result = result + output_string
+		}
+		result += "\n"
+	}
+
+	// Return options:
+	//
+	// output.FLB_OK    = data have been processed.
+	// output.FLB_ERROR = unrecoverable error, do not try this again.
+	// output.FLB_RETRY = retry to flush later.
+	//body result
+	// content type  "application/octet-stream"
+	var b = []byte(result)
+	uri := fmt.Sprintf("%s/ingest?reference=%s", dict["streamtool_uri"], dict["reference"])
+	client := &http.Client{
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{},
+		},
+	}
+	req, err := http.NewRequest("POST", uri, bytes.NewReader(b))
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Got error %s", err.Error())
+		return output.FLB_RETRY
+	}
+	req.SetBasicAuth(dict["user"], dict["password"])
+	req.Header.Set("Content-Type", "application/octet-stream")
+	rsp, err := client.Do(req)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "got error %s", err.Error())
+		return output.FLB_RETRY
+	}
+	defer rsp.Body.Close()
+	if rsp.StatusCode != 204 {
+		fmt.Fprintf(os.Stderr, "status code %d", rsp.StatusCode)
+		return output.FLB_RETRY
+	}
+	if is_end {
+		os.Exit(0)
+	}
+	return output.FLB_OK
+}
+
+//export FLBPluginExit
+func FLBPluginExit() int {
+	return output.FLB_OK
+}
+
+func main() {
+}

test/test.c

+#include <stdio.h>
+#include "../libfluentbit_wendelin.h"
+
+void main() {
+  printf("test");
+
+}