auto stop fluent bit

Makefile

+.PHONY: install
+install: fluentbit_wendelin_output
+
+.PHONY: clean
+clean:
+	rm -f libfluentbit_wendelin.so libfluentbit_wendelin.h
+
+.PHONY: fluentbit_wendelin_output
+fluentbit_wendelin_output:
+	go build -buildmode=c-shared -o libfluentbit_wendelin.so plugin/fluentbit_wendelin.go
+
+# 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

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=

plugin/fluentbit_wendelin.go

+package main
+
+import (
+        "bytes"
+	"C"
+	"fmt"
+	"time"
+        "os"
+	"unsafe"
+        "net/http"
+        "crypto/tls"
+        "strings"
+	"github.com/fluent/fluent-bit-go/output"
+)
+
+//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 ts interface{}
+	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, ts, record = output.GetRecord(dec)
+		if ret != 0 {
+			break
+		}
+
+		var timestamp time.Time
+		switch t := ts.(type) {
+		case output.FLBTime:
+			timestamp = ts.(output.FLBTime).Time
+		case uint64:
+			timestamp = time.Unix(int64(t), 0)
+		default:
+			fmt.Println("time provided invalid, defaulting to now.")
+			timestamp = time.Now()
+		}
+
+		// Print record keys and values
+                result = result + C.GoString(tag) + ":" +timestamp.String()
+		for _, v := range record {
+                   result += "["
+                   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{InsecureSkipVerify: true},
+          },
+        }
+        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
+        }
+        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");
+
+}