wip

Author: alecholmes

cshared.go

@@ -36,10 +36,6 @@ const (
 	collectInterval = 1000 * time.Nanosecond
 )
 
-const (
-	maxBufferedMessages = defaultMaxBufferedMessages
-)
-
 // FLBPluginPreRegister -
 //
 //export FLBPluginPreRegister
@@ -138,17 +134,10 @@ func (p *pluginInstance) inputPreRun() error {
 		return fmt.Errorf("invalid plugin state %q", p.state)
 	}
 
-	if p.runCtx != nil || p.runCancel != nil {
-		return fmt.Errorf("plugin context already set")
-	}
-	if p.msgChannel != nil {
-		return fmt.Errorf("plugin channel already set")
-	}
-
 	runCtx, runCancel := context.WithCancel(context.Background())
 	p.runCtx = runCtx
 	p.runCancel = runCancel
-	p.msgChannel = make(chan Message, maxBufferedMessages)
+	p.msgChannel = make(chan Message, p.maxBufferedMessages)
 	p.state = instanceStateRunnable
 
 	go func() {
@@ -204,21 +193,24 @@ func FLBPluginInputPause() {
 		panic(fmt.Errorf("plugin not initialized"))
 	}
 
-	if err := instance.pause(); err != nil {
+	if err := instance.stop(); err != nil {
 		panic(err)
 	}
 }
 
-func (p *pluginInstance) pause() error {
+func (p *pluginInstance) stop() error {
 	p.mu.Lock()
+	p.runningWG.Wait()
 	defer p.mu.Unlock()
 
-	if p.state != instanceStateRunnable {
-		return fmt.Errorf("cannot pause plugin in state %q", p.state)
+	if p.state != instanceStateRunnable && p.state != instanceStatePreExit {
+		return fmt.Errorf("cannot stop plugin in state %q", p.state)
 	}
 
-	p.runCancel()
-	close(p.msgChannel)
+	if p.state == instanceStateRunnable {
+		p.runCancel()
+		close(p.msgChannel)
+	}
 
 	p.state = instanceStateInitialized
 	p.runCtx = nil
@@ -286,10 +278,10 @@ func (p *pluginInstance) outputPreExit() error {
 
 	// Only output plugins have a pre-exit step
 	if p.meta.output == nil {
-		return nil
+		return fmt.Errorf("plugin is not an output plugin")
 	}
 
-	if p.state != instanceStateInitialized {
+	if p.state != instanceStateRunnable {
 		return fmt.Errorf("invalid plugin state %q", p.state)
 	}
 
@@ -377,16 +369,18 @@ func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int {
 }
 
 func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-
+	p.mu.Lock()
 	if p.state != instanceStateRunnable {
 		return input.FLB_RETRY
 	}
 
+	p.runningWG.Add(1)
+	defer p.runningWG.Done()
+	p.mu.Unlock()
+
 	buf := bytes.NewBuffer([]byte{})
 
-	for loop := min(len(p.msgChannel), maxBufferedMessages); loop > 0; loop-- {
+	for loop := min(len(p.msgChannel), p.maxBufferedMessages); loop > 0; loop-- {
 		select {
 		case msg, ok := <-p.msgChannel:
 			if !ok {
@@ -561,7 +555,7 @@ func FLBPluginExit() int {
 	defer currInstanceMu.Unlock()
 
 	if currInstance != nil {
-		currInstance.pause() //nolint:errcheck
+		currInstance.stop() //nolint:errcheck
 		currInstance = nil
 	}
 

cshared_test.go

@@ -13,7 +13,7 @@ import (
 	"time"
 	"unsafe"
 
-	"github.com/alecthomas/assert/v2"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/vmihailenco/msgpack/v5"
 
@@ -45,7 +45,7 @@ func newTestInputInstance(t testing.TB, input InputPlugin) *pluginInstance {
 		desc:  "test plugin",
 		input: input,
 	}))
-	t.Cleanup(inst.stop)
+	t.Cleanup(func() { assert.NoError(t, inst.stop()) })
 	return inst
 }
 
@@ -55,7 +55,7 @@ func newTestOutputInstance(t testing.TB, output OutputPlugin) *pluginInstance {
 		desc:   "test plugin",
 		output: output,
 	}))
-	t.Cleanup(inst.stop)
+	t.Cleanup(func() { assert.NoError(t, inst.stop()) })
 	return inst
 }
 
@@ -587,18 +587,16 @@ func TestOutputFlush(t *testing.T) {
 		Check: func(t *testing.T, msg Message) error {
 			defer wg.Done()
 
-			assert.Equal(t, now, msg.Time)
-
-			record := assertType[map[string]any](t, msg.Record)
-
-			foo := assertType[string](t, record["foo"])
-			assert.Equal(t, "bar", foo)
-
-			bar := assertType[int8](t, record["bar"])
-			assert.Equal(t, 3, bar)
-
-			foobar := assertType[float64](t, record["foobar"])
-			assert.Equal(t, 1.337, foobar)
+			expectTag := "foobar"
+			assert.Equal(t, Message{
+				Time: now,
+				Record: map[string]any{
+					"foo":    "bar",
+					"bar":    int8(3),
+					"foobar": 1.337,
+				},
+				tag: &expectTag,
+			}, msg)
 
 			return nil
 		},

plugin.go

@@ -78,7 +78,8 @@ func newPluginInstance(meta pluginMetadata) *pluginInstance {
 		configLoaderProvider: func(ptr unsafe.Pointer) ConfigLoader {
 			return &flbInputConfigLoader{ptr: ptr}
 		},
-		state: instanceStateCreated,
+		state:               instanceStateCreated,
+		maxBufferedMessages: defaultMaxBufferedMessages,
 	}
 }
 
@@ -87,12 +88,14 @@ type pluginInstance struct {
 	meta                 pluginMetadata
 	cmetricsCtxProvider  cmetricsContextProvider
 	configLoaderProvider configLoaderProvider
-
-	mu         sync.RWMutex // Protects all members below:
-	state      instanceState
-	runCtx     context.Context
-	runCancel  context.CancelFunc
-	msgChannel chan Message
+	runningWG            sync.WaitGroup
+
+	mu                  sync.Mutex // Protects all members below:
+	state               instanceState
+	runCtx              context.Context
+	runCancel           context.CancelFunc
+	msgChannel          chan Message
+	maxBufferedMessages int
 }
 
 // withCMetricsContextProvider overrides the cmetricsContextProvider.
@@ -151,28 +154,45 @@ func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 				return fmt.Errorf("go.MaxBufferedMessages must be an integer, got %q", maxBufferedStr)
 			}
 
-			maxBufferedMessages = maxBuffered // TODO: put this on plugin?
+			p.maxBufferedMessages = maxBuffered
 		}
 
 		if err := p.meta.input.Init(ctx, fbit); err != nil {
 			return fmt.Errorf("initializing plugin %q: %w", p.meta.name, err)
 		}
-	}
+	} else if p.meta.output != nil {
+		defer cancel()
+		cmt, err := p.cmetricsCtxProvider(ptr)
+		if err != nil {
+			return err
+		}
 
-	// TODO: other branches
+		fbit := &Fluentbit{
+			Conf:    p.configLoaderProvider(ptr),
+			Metrics: makeMetrics(cmt),
+			Logger:  &flbOutputLogger{ptr: ptr},
+		}
 
-	return nil
-}
+		if err := p.meta.output.Init(ctx, fbit); err != nil {
+			return fmt.Errorf("initializing plugin %q: %w", p.meta.name, err)
+		}
+	} else {
+		// Custom plugins don't have preInit functions that set context
+		p.runCtx, p.runCancel = ctx, cancel
+		p.state = instanceStateRunnable
 
-func (p *pluginInstance) stop() {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+		fbit := &Fluentbit{
+			Conf:    p.configLoaderProvider(ptr),
+			Metrics: nil,
+			Logger:  &flbOutputLogger{ptr: ptr},
+		}
 
-	if p.state == instanceStateRunnable {
-		if err := p.pause(); err != nil {
-			panic(err)
+		if err := p.meta.custom.Init(ctx, fbit); err != nil {
+			return fmt.Errorf("initializing plugin %q: %w", p.meta.name, err)
 		}
 	}
+
+	return nil
 }
 
 type Fluentbit struct {