fluentbit-go: add quic output pluging

output_quic/out_quic.go

+package main
+
+import "github.com/fluent/fluent-bit-go/output"
+import quic "github.com/lucas-clemente/quic-go"
+import "github.com/lucas-clemente/quic-go/h2quic"
+
+import (
+	"C"
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"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 {
+
+	var roundTripper = &h2quic.RoundTripper{
+		QuicConfig: &quic.Config{KeepAlive: true},
+	}
+
+	var hclient = &http.Client{
+		Transport: roundTripper,
+	}
+
+	request_string := uri + "/ingest?reference=" + reference
+
+	var b []byte
+	b = C.GoBytes(data, C.int(length))
+	fmt.Println("POST %s", request_string)
+	req, err := http.NewRequest("POST", request_string, bytes.NewBuffer(b))
+
+	if err != nil {
+		fmt.Println("POST error = ")
+		fmt.Println(err)
+		return output.FLB_ERROR
+	}
+
+	req.Header.Set("Content-Type", "application/octet-stream")
+	req.SetBasicAuth(user, password)
+
+	resp, err := hclient.Do(req)
+
+	if err != nil {
+		fmt.Println("DO POST error = ")
+		fmt.Println(err)
+		return output.FLB_ERROR
+	}
+	log.Printf("Got response for POST %s: %#v", request_string, resp)
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		log.Fatalf("Couldn't parse response body. %+v", err)
+	}
+
+	log.Println("Response Body:", string(body))
+	defer resp.Body.Close()
+
+	/*  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_ERROR
+	  }
+
+	  fmt.Println(resp.Status)
+	  fmt.Println(err)
+
+	  if resp_status < 200 && resp_status > 205 {
+	    return output.FLB_ERROR
+	  }
+
+	  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() {
+}