Merge branch 'api-0.12'

Author: edsiper

README.md

@@ -2,6 +2,12 @@
 
 This repository contains Go packages that allows to create [Fluent Bit](http://fluentbit.io) plugins. At the moment it only supports the creation of _Output_ plugins.
 
+## Requirements
+
+The code of this package is intended to be used with [Fluent Bit 0.12](https://github.com/fluent/fluent-bit/tree/0.12) branch.
+
+> Fluent Bit on GIT master (0.12) uses a different format to set records timestamps, this package is not backward compatible with Fluent Bit 0.11
+
 ## Usage
 
 Fluent Bit Go packages are exposed on this repository:
@@ -16,7 +22,7 @@ import "github.com/fluent/fluent-bit-go/output"
 
 for a more practical example please refer to the _out\_gstdout_ plugin implementation located at:
 
-https://github.com/fluent/fluent-bit-go/blob/master/examples/out_gstdout/out_gstdout.go
+https://github.com/fluent/fluent-bit-go/blob/api-0.12/examples/out_gstdout/out_gstdout.go
 
 ## Contact
 

examples/out_gstdout/Makefile

@@ -1,5 +1,8 @@
 all:
 	go build -buildmode=c-shared -o out_gstdout.so .
 
+fast:
+	go build out_gstdout.go
+
 clean:
 	rm -rf *.so *.h *~

examples/out_gstdout/README.md

@@ -0,0 +1,72 @@
+# Example: out_gstdout
+
+The following example code implements a simple output plugin that prints the records to the standard output interface (STDOUT).
+
+Every output plugin go through four callbacks associated to different phases:
+
+| Plugin Phase        | Callback                   |
+|---------------------|----------------------------|
+| Registration        | FLBPluginRegister()        |
+| Initialization      | FLBPluginInit()            |
+| Runtime Flush       | FLBPluginFlush()           |
+| Exit                | FLBPluginExit()            |
+
+## Plugin Registration
+
+When Fluent Bit loads a Golang plugin, it lookup and load the registration callback that aims to populate the internal structure with plugin name and description among others:
+
+```go
+//export FLBPluginRegister
+func FLBPluginRegister(ctx unsafe.Pointer) int {
+	return output.FLBPluginRegister(ctx, "gstdout", "Stdout GO!")
+}
+```
+
+This function is invoked at start time _before_ any configuration is done inside the engine.
+
+## Plugin Initialization
+
+Before the engine starts, it initialize all plugins that were requested to start. Upon initialization a configuration context already exists, so the plugin can ask for configuration parameters or do any other internal checks. E.g:
+
+```go
+//export FLBPluginInit
+func FLBPluginInit(ctx unsafe.Pointer) int {
+	return output.FLB_OK
+}
+```
+
+The function must return FLB\_OK when it initialized properly or FLB\_ERROR if something went wrong. If the plugin reports an error, the engine will _not_ load the instance.
+
+## Runtime Flush
+
+Upon flush time, when Fluent Bit want's to flush it buffers, the runtime flush callback will be triggered.
+
+The callback will receive a raw buffer of msgpack data with it proper bytes length and the tag associated.
+
+```go
+//export FLBPluginFlush
+func FLBPluginFlush(data unsafe.Pointer, length C.int, tag *C.char) int {
+    return output.FLB_OK
+}
+```
+
+> for more details about how to process the incoming msgpack data, refer to the [out_gstdout.go](out_gstdout.go) file.
+
+When done, there are three returning values available:
+
+| Return value  | Description                                    |
+|---------------|------------------------------------------------|
+| FLB\_OK       | The data have been processed normally.         |
+| FLB\_ERROR    | An internal error have ocurred, the plugin will not handle the set of records/data again. |
+| FLB\_RETRY    | A recoverable error have ocurred, the engine can try to flush the records/data later.|
+
+## Plugin Exit
+
+When Fluent Bit will stop using the instance of the plugin, it will trigger the exit callback. e.g:
+
+```go
+//export FLBPluginExit
+func FLBPluginExit() int {
+	return output.FLB_OK
+}
+```

examples/out_gstdout/out_gstdout.go

@@ -2,56 +2,50 @@ package main
 
 import "github.com/fluent/fluent-bit-go/output"
 import (
-	"github.com/ugorji/go/codec"
 	"fmt"
 	"unsafe"
 	"C"
-	"reflect"
 )
 
 //export FLBPluginRegister
-// ctx (context) pointer to fluentbit context (state/ c code)
 func FLBPluginRegister(ctx unsafe.Pointer) int {
-	// roll call for the specifics of the plugin
 	return output.FLBPluginRegister(ctx, "gstdout", "Stdout 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")
+	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 h codec.Handle = new(codec.MsgpackHandle)
-	var b []byte
-	var m interface{}
-	var err error
+	var ret int
+	var ts interface{}
+	var record map[interface{}]interface{}
 
-	b = C.GoBytes(data, length)
-	dec := codec.NewDecoderBytes(b, h)
+	// Create Fluent Bit decoder
+	dec := output.NewDecoder(data, int(length))
 
-	// Iterate the original MessagePack array
+	// Iterate Records
 	count = 0
 	for {
-		// Decode the entry
-		err = dec.Decode(&m)
-		if err != nil {
+		// Extract Record
+		ret, ts, record = output.GetRecord(dec)
+		if ret != 0 {
 			break
 		}
 
-		// Get a slice and their two entries: timestamp and map
-		slice := reflect.ValueOf(m)
-		timestamp := slice.Index(0)
-		data := slice.Index(1)
-
-		// Convert slice data to a real map and iterate
-		map_data := data.Interface().(map[interface{}] interface{})
-		fmt.Printf("[%d] %s: [%d, {", count, C.GoString(tag), timestamp)
-		for k, v := range map_data {
+		// Print record keys and values
+		timestamp := ts.(output.FLBTime)
+		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")
@@ -68,7 +62,7 @@ func FLBPluginFlush(data unsafe.Pointer, length C.int, tag *C.char) int {
 
 //export FLBPluginExit
 func FLBPluginExit() int {
-	return 0
+	return output.FLB_OK
 }
 
 func main() {

output/decoder.go

@@ -0,0 +1,86 @@
+//  Fluent Bit Go!
+//  ==============
+//  Copyright (C) 2015-2017 Treasure Data Inc.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package output
+
+import (
+	"unsafe"
+	"reflect"
+	"encoding/binary"
+	"github.com/ugorji/go/codec"
+	"time"
+	"C"
+)
+
+type FLBDecoder struct {
+	handle *codec.MsgpackHandle
+	mpdec  *codec.Decoder
+}
+
+type FLBTime struct {
+	time.Time
+}
+
+func (f FLBTime) WriteExt(interface{}) []byte {
+	panic("unsupported")
+}
+
+func (f FLBTime) ReadExt(i interface{}, b []byte) {
+	out := i.(*FLBTime)
+	sec := binary.BigEndian.Uint32(b)
+	usec := binary.BigEndian.Uint32(b[4:])
+	out.Time = time.Unix(int64(sec), int64(usec))
+}
+
+func (f FLBTime) ConvertExt(v interface{}) interface{} {
+	return nil
+}
+
+func (f FLBTime) UpdateExt(dest interface{}, v interface{}) {
+	panic("unsupported")
+}
+
+func NewDecoder(data unsafe.Pointer, length int) (*FLBDecoder) {
+	var b []byte
+
+	dec := new(FLBDecoder)
+	dec.handle = new(codec.MsgpackHandle)
+	dec.handle.SetExt(reflect.TypeOf(FLBTime{}), 0, &FLBTime{})
+
+	b = C.GoBytes(data, C.int(length))
+	dec.mpdec = codec.NewDecoderBytes(b, dec.handle)
+
+	return dec
+}
+
+func GetRecord(dec *FLBDecoder) (ret int, ts interface{}, rec map[interface{}]interface{}) {
+	var check error
+	var m interface{}
+
+	check = dec.mpdec.Decode(&m)
+	if check != nil {
+		return -1, 0, nil
+	}
+
+	slice := reflect.ValueOf(m)
+	t := slice.Index(0).Interface()
+	data := slice.Index(1)
+
+	map_data := data.Interface().(map[interface{}] interface{})
+
+	return 0, t, map_data
+}

output/flb_output.h

@@ -0,0 +1,44 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit Go!
+ *  ==============
+ *  Copyright (C) 2015-2017 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLBGO_OUTPUT_H
+#define FLBGO_OUTPUT_H
+
+#include <stdio.h>
+
+struct flb_api {
+    char *(*output_get_property) (char *, void *);
+};
+
+struct flbgo_output_plugin {
+    char *name;
+    struct flb_api *api;
+    void *o_ins;
+    int (*cb_init)();
+    int (*cb_flush)(void *, size_t, char *);
+    int (*cb_exit)(void *);
+};
+
+char *output_get_property(char *key, void *ctx)
+{
+    struct flbgo_output_plugin *plugin = ctx;
+    return plugin->api->output_get_property(key, plugin->o_ins);
+}
+
+#endif

output/flb_plugin.h

@@ -1,8 +1,8 @@
 /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
 
-/*  Fluent Bit
- *  ==========
- *  Copyright (C) 2015-2016 Treasure Data Inc.
+/*  Fluent Bit Go!
+ *  ==============
+ *  Copyright (C) 2015-2017 Treasure Data Inc.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@
 #define FLB_PROXY_OUTPUT_PLUGIN    2
 #define FLB_PROXY_GOLANG          11
 
+/* Structure used for registration, it match the one on flb_plugin_proxy.h */
 struct flb_plugin_proxy {
     int type;
     int proxy;

output/output.go

@@ -1,6 +1,6 @@
 //  Fluent Bit Go!
 //  ==============
-//  Copyright (C) 2015-2016 Treasure Data Inc.
+//  Copyright (C) 2015-2017 Treasure Data Inc.
 //
 //  Licensed under the Apache License, Version 2.0 (the "License");
 //  you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ package output
 /*
 #include <stdlib.h>
 #include "flb_plugin.h"
+#include "flb_output.h"
 */
 import "C"
 import "fmt"
@@ -35,6 +36,7 @@ const FLB_PROXY_GOLANG        =  C.FLB_PROXY_GOLANG
 
 // Local type to define a plugin definition
 type FLBPlugin C.struct_flb_plugin_proxy
+type FLBOutPlugin C.struct_flbgo_output_plugin
 
 // When the FLBPluginInit is triggered by Fluent Bit, a plugin context
 // is passed and the next step is to invoke this FLBPluginRegister() function
@@ -53,8 +55,12 @@ func FLBPluginRegister(ctx unsafe.Pointer, name string, desc string) int {
 // Release resources allocated by the plugin initialization
 func FLBPluginUnregister(ctx unsafe.Pointer) {
 	p := (*FLBPlugin) (unsafe.Pointer(ctx))
-
 	fmt.Printf("[flbgo] unregistering %v\n", p)
 	C.free(unsafe.Pointer(p.name))
 	C.free(unsafe.Pointer(p.description))
 }
+
+func FLBPluginConfigKey(ctx unsafe.Pointer, key string) string {
+	_key := C.CString(key)
+	return C.GoString(C.output_get_property(_key, unsafe.Pointer(ctx)))
+}