wip

Author: alecholmes

cshared.go

@@ -6,7 +6,6 @@ package plugin
 import "C"
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -237,7 +236,7 @@ func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int {
 		return input.FLB_RETRY
 	}
 
-	return instance.callback(data, csize)
+	return instance.inputCallback(data, csize)
 }
 
 // FLBPluginInputCleanupCallback releases the memory used during the input callback
@@ -280,42 +279,14 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 
 	in := C.GoBytes(data, clength)
 	tag := C.GoString(ctag)
-	if err := pluginFlush(instance, tag, in); err != nil {
+	if err := instance.outputFlush(tag, in); err != nil {
 		fmt.Fprintf(os.Stderr, "flush: %s\n", err)
 		return output.FLB_ERROR
 	}
 
 	return output.FLB_OK
 }
 
-func pluginFlush(instance *pluginInstance, tag string, b []byte) error {
-	dec := msgpack.NewDecoder(bytes.NewReader(b))
-	for {
-		select {
-		case <-instance.runCtx.Done():
-			err := instance.runCtx.Err()
-			if err != nil && !errors.Is(err, context.Canceled) {
-				fmt.Fprintf(os.Stderr, "run: %s\n", err)
-				return fmt.Errorf("run: %w", err)
-			}
-
-			return nil
-		default:
-		}
-
-		msg, err := decodeMsg(dec, tag)
-		if errors.Is(err, io.EOF) {
-			return nil
-		}
-
-		if err != nil {
-			return err
-		}
-
-		instance.msgChannel <- msg
-	}
-}
-
 // decodeMsg should be called with an already initialized decoder.
 func decodeMsg(dec *msgpack.Decoder, tag string) (Message, error) {
 	var out Message
@@ -512,7 +483,7 @@ func testCallback(inst *pluginInstance) ([]byte, int) {
 	data := unsafe.Pointer(nil)
 	var csize C.size_t
 
-	retCode := inst.callback(&data, &csize)
+	retCode := inst.inputCallback(&data, &csize)
 
 	if data == nil {
 		return []byte{}, retCode

cshared_test.go

@@ -762,7 +762,7 @@ func TestOutputFlush(t *testing.T) {
 	assert.NoError(t, err)
 
 	wg.Add(1)
-	assert.NoError(t, pluginFlush(inst, "foobar", b))
+	assert.NoError(t, inst.outputFlush("foobar", b))
 	wg.Wait()
 }
 

examples/in_gdummy/go.mod

@@ -1,6 +1,8 @@
 module github.com/fluent/fluent-bit-go/examples/gdummy
 
-go 1.22.4
+go 1.23.0
+
+toolchain go1.24.2
 
 require github.com/calyptia/plugin v0.1.6
 

examples/in_gdummy/go.sum

@@ -28,6 +28,9 @@ github.com/ugorji/go v1.2.7/go.mod h1:nF9osbDWLy6bDVv/Rtoh6QgnvNDpmCalQV5urGCCS6
 github.com/ugorji/go/codec v1.2.7 h1:YPXUKf7fYbp/y8xloBqZOw2qaVggbfwMlI8WM3wZUJ0=
 github.com/ugorji/go/codec v1.2.7/go.mod h1:WGN1fab3R1fzQlVQTkfxVtIBhWDRqOviHU95kRgeqEY=
 github.com/ugorji/go/codec v1.2.11/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
+github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
 github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 h1:EzJWgHovont7NscjpAxXsDA8S8BMYve8Y5+7cuRE7R0=
 github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17UxZ74=

examples/out_gstdout/go.mod

@@ -1,12 +1,14 @@
 module github.com/fluent/fluent-bit-go/examples/gstdout
 
-go 1.21.0
+go 1.23.0
+
+toolchain go1.24.2
 
 require github.com/calyptia/plugin v0.1.6
 
 require (
 	github.com/calyptia/cmetrics-go v0.1.7 // indirect
-	github.com/ugorji/go/codec v1.2.11 // indirect
+	github.com/ugorji/go/codec v1.2.12 // indirect
 )
 
 replace github.com/calyptia/plugin => ../..

examples/out_gstdout/go.sum

@@ -719,6 +719,7 @@ 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/ugorji/go/codec v1.2.12/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=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=

examples/out_multiinstance/go.mod

@@ -1,6 +1,8 @@
 module github.com/fluent/fluent-bit-go/examples/multiinstance
 
-go 1.21.0
+go 1.23.0
+
+toolchain go1.24.2
 
 require github.com/fluent/fluent-bit-go v0.0.0-20200420155746-e125cab17963
 

plugin.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"log"
 	"os"
 	"runtime"
@@ -367,11 +368,11 @@ func (p *pluginInstance) outputPreRunWithLock() error {
 	return nil
 }
 
-// callback consumes up to maxBufferedMessages message from the plugin's message channel,
+// inputCallback consumes up to maxBufferedMessages message from the plugin's message channel,
 // returning early if the plugin is shutdown.
 // Consumed messages are marshaled into msgpack bytes and the contents and length of contents
 // set in the respective data and csize input variables.
-func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
+func (p *pluginInstance) inputCallback(data *unsafe.Pointer, csize *C.size_t) int {
 	p.mu.Lock()
 	if p.state != instanceStateRunnable {
 		p.mu.Unlock()
@@ -427,6 +428,46 @@ func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
 	return input.FLB_OK
 }
 
+// outputFlush writes the messages in msgpackBytes to the plugin's message channel,
+// returning early if the plugin is shutdown.
+func (p *pluginInstance) outputFlush(tag string, msgpackBytes []byte) error {
+	p.mu.Lock()
+	if p.state != instanceStateRunnable {
+		p.mu.Unlock()
+		return fmt.Errorf("invalid plugin state %q", p.state)
+	}
+
+	p.runningWG.Add(1)
+	defer p.runningWG.Done()
+	p.mu.Unlock()
+
+	dec := msgpack.NewDecoder(bytes.NewReader(msgpackBytes))
+	for {
+		select {
+		case <-p.runCtx.Done():
+			err := p.runCtx.Err()
+			if err != nil && !errors.Is(err, context.Canceled) {
+				fmt.Fprintf(os.Stderr, "run: %s\n", err)
+				return fmt.Errorf("run: %w", err)
+			}
+
+			return nil
+		default:
+		}
+
+		msg, err := decodeMsg(dec, tag)
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+
+		if err != nil {
+			return err
+		}
+
+		p.msgChannel <- msg
+	}
+}
+
 // stop stops the plugin, freeing resources and returning it to initialized state.
 // Calling stop will signal callbacks to return via the plugin's context, then wait for
 // callbacks to finish and return before freeing resources and returning.