Commit 284e0d83 authored by Jérome Perrin's avatar Jérome Perrin

go fmt

parent f790463a
package main
import (
"bytes"
"C"
"bytes"
"crypto/tls"
"fmt"
"os"
"unsafe"
"net/http"
"crypto/tls"
"strings"
"github.com/fluent/fluent-bit-go/output"
"net/http"
"os"
"strings"
"unsafe"
)
//export FLBPluginRegister
......@@ -22,22 +22,22 @@ func FLBPluginRegister(def unsafe.Pointer) int {
// 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")
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)
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
}
......@@ -47,11 +47,11 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
var record map[interface{}]interface{}
// Create Fluent Bit decoder
dec := output.NewDecoder(data, int(length))
dict := output.FLBPluginGetContext(ctx).(map[string]string)
dict := output.FLBPluginGetContext(ctx).(map[string]string)
// Iterate Records
var result string
result = ""
var is_end bool = false
var result string
result = ""
var is_end bool = false
for {
// Extract Record
ret, _, record = output.GetRecord(dec)
......@@ -61,16 +61,16 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
// 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
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"
result += "\n"
}
// Return options:
......@@ -78,35 +78,35 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int
// 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
//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
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment