diff --git a/cshared.go b/cshared.go index 68bf910..951d3e0 100644 --- a/cshared.go +++ b/cshared.go @@ -44,6 +44,16 @@ var ( maxBufferedMessages = defaultMaxBufferedMessages ) +//export FLBPluginPreRegister +func FLBPluginPreRegister(hotReloading C.int) int { + if hotReloading == C.int(1) { + initWG.Add(1) + registerWG.Add(1) + } + + return input.FLB_OK +} + // FLBPluginRegister registers a plugin in the context of the fluent-bit runtime, a name and description // can be provided. // @@ -72,6 +82,24 @@ func FLBPluginRegister(def unsafe.Pointer) int { return out } +func cleanup() int { + if unregister != nil { + unregister() + unregister = nil + } + + if runCancel != nil { + runCancel() + runCancel = nil + } + + if theChannel != nil { + defer close(theChannel) + } + + return input.FLB_OK +} + // FLBPluginInit this method gets invoked once by the fluent-bit runtime at initialisation phase. // here all the plugin context should be initialized and any data or flag required for // plugins to execute the collect or flush callback. @@ -80,8 +108,6 @@ func FLBPluginRegister(def unsafe.Pointer) int { func FLBPluginInit(ptr unsafe.Pointer) int { defer initWG.Done() - registerWG.Wait() - if theInput == nil && theOutput == nil { fmt.Fprintf(os.Stderr, "no input or output registered\n") return input.FLB_RETRY @@ -167,6 +193,118 @@ var theInputLock sync.Mutex // // This function will invoke Collect only once to preserve backward // compatible behavior. There are unit tests to enforce this behavior. +func prepareInputCollector() (err error) { + runCtx, runCancel = context.WithCancel(context.Background()) + theChannel = make(chan Message, maxBufferedMessages) + + theInputLock.Lock() + + go func(theChannel chan<- Message) { + defer theInputLock.Unlock() + + go func(theChannel chan<- Message) { + err = theInput.Collect(runCtx, theChannel) + }(theChannel) + + for { + select { + case <-runCtx.Done(): + log.Printf("goroutine will be stopping: name=%q\n", theName) + return + } + } + + if err != nil { + fmt.Fprintf(os.Stderr, + "collect error: %s\n", err.Error()) + } + }(theChannel) + + return err +} + +// FLBPluginInputPreRun this method gets invoked by the fluent-bit runtime, once the plugin has been +// initialised, the plugin invoked only once before executing the input callbacks. +// +//export FLBPluginInputPreRun +func FLBPluginInputPreRun(useHotReload C.int) int { + registerWG.Wait() + + var err error + err = prepareInputCollector() + + if err != nil { + fmt.Fprintf(os.Stderr, "run: %s\n", err) + return input.FLB_ERROR + } + + return input.FLB_OK +} + + +// FLBPluginInputPause this method gets invoked by the fluent-bit runtime, once the plugin has been +// paused, the plugin invoked this method and entering paused state. +// +//export FLBPluginInputPause +func FLBPluginInputPause() { + if runCancel != nil { + runCancel() + runCancel = nil + } + + if theChannel != nil { + close(theChannel) + theChannel = nil + } +} + +// FLBPluginInputResume this method gets invoked by the fluent-bit runtime, once the plugin has been +// resumeed, the plugin invoked this method and re-running state. +// +//export FLBPluginInputResume +func FLBPluginInputResume() { + var err error + err = prepareInputCollector() + + if err != nil { + fmt.Fprintf(os.Stderr, "run: %s\n", err) + } +} + +//export FLBPluginOutputPreRun +func FLBPluginOutputPreRun(useHotReload C.int) int { + registerWG.Wait() + + var err error + runCtx, runCancel = context.WithCancel(context.Background()) + theChannel = make(chan Message) + go func(runCtx context.Context) { + go func(runCtx context.Context) { + err = theOutput.Flush(runCtx, theChannel) + }(runCtx) + + for { + select { + case <-runCtx.Done(): + log.Printf("goroutine will be stopping: name=%q\n", theName) + return + } + } + + }(runCtx) + + if err != nil { + fmt.Fprintf(os.Stderr, "run: %s\n", err) + return output.FLB_ERROR + } + + return output.FLB_OK +} + +// FLBPluginInputCallback this method gets invoked by the fluent-bit runtime, once the plugin has been +// initialised, the plugin implementation is responsible for handling the incoming data and the context +// that gets past, for long-living collectors the plugin itself should keep a running thread and fluent-bit +// will not execute further callbacks. // //export FLBPluginInputCallback func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int { @@ -177,23 +315,6 @@ func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int { return input.FLB_RETRY } - once.Do(func() { - runCtx, runCancel = context.WithCancel(context.Background()) - theChannel = make(chan Message, maxBufferedMessages) - - theInputLock.Lock() - - go func(theChannel chan<- Message) { - defer theInputLock.Unlock() - - err := theInput.Collect(runCtx, theChannel) - if err != nil { - fmt.Fprintf(os.Stderr, - "collect error: %s\n", err.Error()) - } - }(theChannel) - }) - buf := bytes.NewBuffer([]byte{}) for loop := min(len(theChannel), maxBufferedMessages); loop > 0; loop-- { @@ -261,18 +382,6 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int { } var err error - once.Do(func() { - runCtx, runCancel = context.WithCancel(context.Background()) - theChannel = make(chan Message) - go func() { - err = theOutput.Flush(runCtx, theChannel) - }() - }) - if err != nil { - fmt.Fprintf(os.Stderr, "run: %s\n", err) - return output.FLB_ERROR - } - select { case <-runCtx.Done(): err = runCtx.Err() @@ -319,24 +428,29 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int { return output.FLB_ERROR } - if d := len(entry); d != 2 { - fmt.Fprintf(os.Stderr, "unexpected entry length: %d\n", d) + slice := reflect.ValueOf(entry) + if slice.Kind() != reflect.Slice || slice.Len() < 2 { + fmt.Fprintf(os.Stderr, "unexpected entry length: %d\n", slice.Len()) return output.FLB_ERROR } - ft, ok := entry[0].(bigEndianTime) - if !ok { + var t time.Time + ts := slice.Index(0).Interface() + switch ft := ts.(type) { + case bigEndianTime: + t = time.Time(ft) + case []interface{}: + s := reflect.ValueOf(ft) + st := s.Index(0).Interface() + ty := st.(bigEndianTime) + t = time.Time(ty) + default: fmt.Fprintf(os.Stderr, "unexpected entry time type: %T\n", entry[0]) return output.FLB_ERROR } - t := time.Time(ft) - - recVal, ok := entry[1].(map[any]any) - if !ok { - fmt.Fprintf(os.Stderr, "unexpected entry record type: %T\n", entry[1]) - return output.FLB_ERROR - } + data := slice.Index(1) + recVal := data.Interface().(map[interface{}]interface{}) var rec map[string]string if d := len(recVal); d != 0 { @@ -348,13 +462,22 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int { return output.FLB_ERROR } - val, ok := v.([]uint8) - if !ok { + var val string + switch tv := v.(type) { + case []uint8: + val = string(tv) + case uint64: + val = strconv.FormatUint(tv, 10) + case int64: + val = strconv.FormatInt(tv, 10) + case bool: + val = strconv.FormatBool(tv) + default: fmt.Fprintf(os.Stderr, "unexpected record value type: %T\n", v) return output.FLB_ERROR } - rec[key] = string(val) + rec[key] = val } } @@ -374,21 +497,7 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int { // //export FLBPluginExit func FLBPluginExit() int { - log.Printf("calling FLBPluginExit(): name=%q\n", theName) - - if unregister != nil { - unregister() - } - - if runCancel != nil { - runCancel() - } - - if theChannel != nil { - defer close(theChannel) - } - - return input.FLB_OK + return cleanup() } type flbInputConfigLoader struct { diff --git a/cshared_test.go b/cshared_test.go index 76f2fe7..a4d0156 100644 --- a/cshared_test.go +++ b/cshared_test.go @@ -40,6 +40,13 @@ func TestInputCallbackCtrlC(t *testing.T) { ptr := unsafe.Pointer(nil) + // prepare channel for input explicitly. + err := prepareInputCollector() + if err != nil { + t.Fail() + return + } + go func() { FLBPluginInputCallback(&ptr, nil) cdone <- true @@ -86,6 +93,12 @@ func TestInputCallbackDangle(t *testing.T) { cdone := make(chan bool) ptr := unsafe.Pointer(nil) + // prepare channel for input explicitly. + err := prepareInputCollector() + if err != nil { + t.Fail() + } + go func() { t := time.NewTicker(collectInterval) defer t.Stop() @@ -156,6 +169,13 @@ func TestInputCallbackInfinite(t *testing.T) { cshutdown := make(chan bool) ptr := unsafe.Pointer(nil) + // prepare channel for input explicitly. + err := prepareInputCollector() + if err != nil { + t.Fail() + return + } + go func() { t := time.NewTicker(collectInterval) defer t.Stop() @@ -237,6 +257,13 @@ func TestInputCallbackLatency(t *testing.T) { cstarted := make(chan bool) cmsg := make(chan []byte) + // prepare channel for input explicitly. + err := prepareInputCollector() + if err != nil { + t.Fail() + return + } + go func() { t := time.NewTicker(collectInterval) defer t.Stop() @@ -363,6 +390,12 @@ func TestInputCallbackInfiniteConcurrent(t *testing.T) { concurrentWait.Add(64) + // prepare channel for input explicitly. + err := prepareInputCollector() + if err != nil { + t.Fail() + } + go func(cstarted chan bool) { ticker := time.NewTicker(time.Second * 1) defer ticker.Stop() diff --git a/examples/out_gstdout/go.mod b/examples/out_gstdout/go.mod index 4829621..e5635fe 100644 --- a/examples/out_gstdout/go.mod +++ b/examples/out_gstdout/go.mod @@ -2,6 +2,11 @@ module github.com/fluent/fluent-bit-go/examples/gstdout go 1.21.0 -require github.com/fluent/fluent-bit-go v0.0.0-20200420155746-e125cab17963 +require github.com/calyptia/plugin v0.1.6 -replace github.com/fluent/fluent-bit-go => ../.. +require ( + github.com/calyptia/cmetrics-go v0.1.7 // indirect + github.com/ugorji/go/codec v1.2.11 // indirect +) + +replace github.com/calyptia/plugin => ../.. diff --git a/examples/out_gstdout/go.sum b/examples/out_gstdout/go.sum index d612ace..297fc41 100644 --- a/examples/out_gstdout/go.sum +++ b/examples/out_gstdout/go.sum @@ -119,7 +119,10 @@ github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx2 github.com/bonitoo-io/go-sql-bigquery v0.3.4-1.4.0/go.mod h1:J4Y6YJm0qTWB9aFziB7cPeSyc6dOZFyJdteSeybVpXQ= github.com/c-bata/go-prompt v0.2.2/go.mod h1:VzqtzE2ksDBcdln8G7mk2RX9QyGjH+OVqOCSiVIqS34= github.com/cactus/go-statsd-client/statsd v0.0.0-20191106001114-12b4e2b38748/go.mod h1:l/bIBLeOl9eX+wxJAzxS4TveKRtAqlyDpHjhkfO0MEI= +github.com/calyptia/cmetrics-go v0.1.7 h1:A4kEFuFqVuWzytIbbey9KivHi0GQVjOkE2JJkdRbQ2U= github.com/calyptia/cmetrics-go v0.1.7/go.mod h1:K1IEPgICDtD4mJW7RVhfG4BkCywnjCdYZwbKs0jSw/U= +github.com/calyptia/plugin v1.1.1 h1:vOQvSKTXsAlNQ+/+VA1xDbhOwqKf/IeRBs0zMdOhT2k= +github.com/calyptia/plugin v1.1.1/go.mod h1:V5LqfR3UJ7G+NMf80Hm3VUgSocDJQJ7OuXlgorCw++M= github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ= github.com/cenkalti/backoff v0.0.0-20181003080854-62661b46c409/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM= github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM= @@ -714,6 +717,7 @@ 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= +github.com/ugorji/go/codec v1.2.11 h1:BMaWp1Bb6fHwEtbplGBGJ498wD+LKlNSl25MjdZY4dU= github.com/ugorji/go/codec v1.2.11/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA= github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= diff --git a/examples/out_gstdout/out_gstdout.go b/examples/out_gstdout/out_gstdout.go index 4ddc854..de37386 100644 --- a/examples/out_gstdout/out_gstdout.go +++ b/examples/out_gstdout/out_gstdout.go @@ -1,82 +1,60 @@ package main import ( - "C" + "context" "fmt" - "time" - "unsafe" + "reflect" - "github.com/calyptia/plugin/output" + "github.com/calyptia/plugin" + "github.com/calyptia/plugin/metric" ) -//export FLBPluginRegister -func FLBPluginRegister(def unsafe.Pointer) int { - return output.FLBPluginRegister(def, "gstdout", "Stdout GO!") +func init() { + plugin.RegisterOutput("gstdout", "StdOut GO!", &gstdoutPlugin{}) } -// (fluentbit will call this) -// plugin (context) pointer to fluentbit context (state/ c code) -// -//export FLBPluginInit -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 +type gstdoutPlugin struct { + param string + flushCounter metric.Counter + log plugin.Logger } -//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{} +func (plug *gstdoutPlugin) Init(ctx context.Context, fbit *plugin.Fluentbit) error { + plug.flushCounter = fbit.Metrics.NewCounter("flush_total", "Total number of flushes", "gstdout") + plug.param = fbit.Conf.String("param") + plug.log = fbit.Logger - // Create Fluent Bit decoder - dec := output.NewDecoder(data, int(length)) + return nil +} +func (plug gstdoutPlugin) Flush(ctx context.Context, ch <-chan plugin.Message) error { // Iterate Records - count = 0 - for { - // Extract Record - ret, ts, record = output.GetRecord(dec) - if ret != 0 { - break - } + count := 0 - 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() - } + for msg := range ch { + plug.flushCounter.Add(1) + plug.log.Debug("[gstdout] operation proceeded") // 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("[%d] %s: [%d.%d, {", count, msg.Tag(), + msg.Time.Unix(), msg.Time.Nanosecond()) + rec := reflect.ValueOf(msg.Record) + if rec.Kind() == reflect.Map { + keyCount := 0 + for _, key := range rec.MapKeys() { + if keyCount > 0 { + fmt.Printf(", ") + } + strct := rec.MapIndex(key) + fmt.Printf("\"%s\":\"%v\"", key.Interface(), strct.Interface()) + keyCount++ + } } - fmt.Printf("}\n") + 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 + return nil } -//export FLBPluginExit -func FLBPluginExit() int { - return output.FLB_OK -} - -func main() { -} +func main() {} diff --git a/plugin.go b/plugin.go index 6f6c581..a9e16c8 100644 --- a/plugin.go +++ b/plugin.go @@ -33,6 +33,7 @@ var ( func init() { registerWG.Add(1) initWG.Add(1) + theChannel = nil } type Fluentbit struct {