go plugin

src/go_plugin.go

-package main
-
-import "github.com/fluent/fluent-bit-go/output"
-
-//export FLBPluginRegister
-func FLBPluginRegister(def unsafe.Pointer) int {
-    // Gets called only once when the plugin.so is loaded
-    return output.FLBPluginRegister(ctx, "gstdout", "Stdout GO!")
-}
-
-//export FLBPluginInit
-func FLBPluginInit(plugin unsafe.Pointer) int {
-    // Gets called only once for each instance you have configured.
-    return output.FLB_OK
-}
-
-//export FLBPluginFlushCtx
-func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int {
-    // Gets called with a batch of records to be written to an instance.
-    return output.FLB_OK
-}
-
-//export FLBPluginExit
-func FLBPluginExit() int {
-    return output.FLB_OK
-}
-
-func main() {
-}
\ No newline at end of file

src/go_plugin/go.mod

+module github.com/fluent/fluent-bit-go/examples/gstdout
+
+go 1.14
+
+require github.com/fluent/fluent-bit-go v0.0.0-20200420155746-e125cab17963
+
+replace github.com/fluent/fluent-bit-go => ./fluent-bit-go

src/go_plugin/go.sum

+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=
+jithub.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=

src/go_plugin/install.sh

+rm -rf *.so *.h *~ fluent-bit-go
+git clone https://github.com/fluent/fluent-bit-go.git
+go build -buildmode=c-shared -o out_gstdout.so

src/go_plugin/out_gstdout.go

+package main
+
+import (
+	"C"
+	"fmt"
+	"time"
+        "os"
+	"unsafe"
+
+	"github.com/fluent/fluent-bit-go/output"
+)
+
+//export FLBPluginRegister
+func FLBPluginRegister(def unsafe.Pointer) int {
+	return output.FLBPluginRegister(def, "gstdout", "Stdout GO!")
+}
+
+//export FLBPluginInit
+// (fluentbit will call this)
+// plugin (context) pointer to fluentbit context (state/ c code)
+func FLBPluginInit(plugin unsafe.Pointer) int {
+	// Example to retrieve an optional configuration parameter
+	param := output.FLBPluginConfigKey(plugin, "param")
+	fmt.Printf("[flb-go] plugin parameter = '%s'\n", param)
+	return output.FLB_OK
+}
+
+//export FLBPluginFlush
+func FLBPluginFlush(data unsafe.Pointer, length C.int, tag *C.char) int {
+	var count int
+	var ret int
+	var ts interface{}
+	var record map[interface{}]interface{}
+        os.Exit(0)
+	// Create Fluent Bit decoder
+	dec := output.NewDecoder(data, int(length))
+
+	// Iterate Records
+	count = 0
+	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
+		fmt.Printf("[%d] %s: [%s, {", count, C.GoString(tag),
+			timestamp.String())
+		for k, v := range record {
+			fmt.Printf("\"%s\": %v, ", k, v)
+		}
+		fmt.Printf("}\n")
+		count++
+	}
+
+	// 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/go_plugin/out_gstdout.h

+/* Code generated by cmd/cgo; DO NOT EDIT. */
+
+/* package github.com/fluent/fluent-bit-go/examples/gstdout */
+
+
+#line 1 "cgo-builtin-export-prolog"
+
+#include <stddef.h> /* for ptrdiff_t below */
+
+#ifndef GO_CGO_EXPORT_PROLOGUE_H
+#define GO_CGO_EXPORT_PROLOGUE_H
+
+#ifndef GO_CGO_GOSTRING_TYPEDEF
+typedef struct { const char *p; ptrdiff_t n; } _GoString_;
+#endif
+
+#endif
+
+/* Start of preamble from import "C" comments.  */
+
+
+
+
+/* End of preamble from import "C" comments.  */
+
+
+/* Start of boilerplate cgo prologue.  */
+#line 1 "cgo-gcc-export-header-prolog"
+
+#ifndef GO_CGO_PROLOGUE_H
+#define GO_CGO_PROLOGUE_H
+
+typedef signed char GoInt8;
+typedef unsigned char GoUint8;
+typedef short GoInt16;
+typedef unsigned short GoUint16;
+typedef int GoInt32;
+typedef unsigned int GoUint32;
+typedef long long GoInt64;
+typedef unsigned long long GoUint64;
+typedef GoInt64 GoInt;
+typedef GoUint64 GoUint;
+typedef __SIZE_TYPE__ GoUintptr;
+typedef float GoFloat32;
+typedef double GoFloat64;
+typedef float _Complex GoComplex64;
+typedef double _Complex GoComplex128;
+
+/*
+  static assertion to make sure the file is being used on architecture
+  at least with matching size of GoInt.
+*/
+typedef char _check_for_64_bit_pointer_matching_GoInt[sizeof(void*)==64/8 ? 1:-1];
+
+#ifndef GO_CGO_GOSTRING_TYPEDEF
+typedef _GoString_ GoString;
+#endif
+typedef void *GoMap;
+typedef void *GoChan;
+typedef struct { void *t; void *v; } GoInterface;
+typedef struct { void *data; GoInt len; GoInt cap; } GoSlice;
+
+#endif
+
+/* End of boilerplate cgo prologue.  */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+extern GoInt FLBPluginRegister(void* def);
+
+// (fluentbit will call this)
+// plugin (context) pointer to fluentbit context (state/ c code)
+extern GoInt FLBPluginInit(void* plugin);
+extern GoInt FLBPluginFlush(void* data, int length, char* tag);
+extern GoInt FLBPluginExit();
+
+#ifdef __cplusplus
+}
+#endif

src/manually_ingest.c

-#include <fluent-bit.h>
-
-#define JSON_1   "[1449505010, {\"key1\": \"some value\"}]"
-#define JSON_2   "[1449505620, {\"key1\": \"some new value\"}]"
-
-int main()
-{
-    int ret;
-    int in_ffd;
-    int out_ffd;
-    flb_ctx_t *ctx;
-
-    /* Create library context */
-    ctx = flb_create();
-    if (!ctx) {
-        return -1;
-    }
-
-    /* Enable the input plugin for manual data ingestion */
-    in_ffd = flb_input(ctx, "lib", NULL);
-    if (in_ffd == -1) {
-        flb_destroy(ctx);
-        return -1;
-    }
-
-    /* Enable output plugin 'stdout' (print records to the standard output) */
-    out_ffd = flb_output(ctx, "stdout", NULL);
-    if (out_ffd == -1) {
-        flb_destroy(ctx);
-        return -1;
-    }
-
-    /* Start the engine */
-    ret = flb_start(ctx);
-    if (ret == -1) {
-        flb_destroy(ctx);
-        return -1;
-    }
-
-    /* Ingest data manually */
-    flb_lib_push(ctx, in_ffd, JSON_1, sizeof(JSON_1) - 1);
-    flb_lib_push(ctx, in_ffd, JSON_2, sizeof(JSON_2) - 1);
-
-    /* Stop the engine (5 seconds to flush remaining data) */
-    flb_stop(ctx);
-
-    /* Destroy library context, release all resources */
-    flb_destroy(ctx);
-
-    return 0;
-}
\ No newline at end of file