wip

Author: alecholmes

cshared.go

@@ -506,17 +506,18 @@ func makeMetrics(cmp *cmetrics.Context) Metrics {
 	}
 }
 
-// testFLBPluginInputCallback cannot be in a test file since test files can't use CGO.
-func testFLBPluginInputCallback(inst *pluginInstance) ([]byte, error) {
+// testCallback invokes the callback and returns the bytes outputted from it.
+// This cannot be in the test file since test files can't use CGO.
+func testCallback(inst *pluginInstance) ([]byte, int) {
 	data := unsafe.Pointer(nil)
 	var csize C.size_t
 
-	inst.callback(&data, &csize)
+	retCode := inst.callback(&data, &csize)
 
 	if data == nil {
-		return []byte{}, nil
+		return []byte{}, retCode
 	}
 
 	defer C.free(data)
-	return C.GoBytes(data, C.int(csize)), nil
+	return C.GoBytes(data, C.int(csize)), retCode
 }

cshared_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/vmihailenco/msgpack/v5"
 
 	"github.com/calyptia/cmetrics-go"
+	"github.com/calyptia/plugin/input"
 	"github.com/calyptia/plugin/metric"
 )
 
@@ -45,7 +46,21 @@ func newTestInputInstance(t testing.TB, input InputPlugin) *pluginInstance {
 		desc:  "test plugin",
 		input: input,
 	}))
-	t.Cleanup(func() { assert.NoError(t, inst.stop()) })
+	t.Cleanup(func() {
+		stopErr := make(chan error)
+		go func() {
+			stopErr <- inst.stop()
+		}()
+
+		select {
+		case err := <-stopErr:
+			assert.NoError(t, err)
+			return
+		case <-time.After(time.Second):
+			require.Fail(t, "timed out waiting for instance to stop")
+		}
+	})
+
 	return inst
 }
 
@@ -69,6 +84,116 @@ func pluginInstanceWithFakes(inst *pluginInstance) *pluginInstance {
 		})
 }
 
+func TestInputCallbackLifecycle(t *testing.T) {
+	plugin := newTestInputPlugin()
+	inst := newTestInputInstance(t, plugin)
+
+	// Initialization
+	require.NoError(t, inst.init(nil))
+	require.Equal(t, int64(1), plugin.initCount.Load())
+
+	require.ErrorContains(t, inst.init(nil), `unexpected plugin state "initialized"`)
+	require.Equal(t, int64(1), plugin.initCount.Load(), "initialization should only run once")
+
+	// Early attempt to callback
+	_, callbackResp := testCallback(inst)
+	require.Equal(t, input.FLB_RETRY, callbackResp, "pre-run must be called before callback")
+
+	// Pre-run
+	require.NoError(t, inst.inputPreRun())
+	require.Eventually(t, plugin.collectRunning.Load, time.Second, time.Millisecond,
+		"collect background loop should have started running")
+	m1 := testMessage(map[string]any{"name": "m1"})
+	m2 := testMessage(map[string]any{"name": "m2"})
+	plugin.enqueue(m1)()
+	plugin.enqueue(m2)()
+
+	require.ErrorContains(t, inst.inputPreRun(), `invalid plugin state "runnable"`)
+
+	// Callback
+	time.Sleep(time.Second)
+	callbackBytes, callbackResp := testCallback(inst)
+	require.Equal(t, input.FLB_OK, callbackResp)
+	require.Equal(t, []Message{m1, m2}, decodeMessages(t, callbackBytes))
+	require.True(t, plugin.collectRunning.Load())
+
+	// Stop
+	require.NoError(t, inst.stop())
+	require.False(t, plugin.collectRunning.Load())
+	require.NoError(t, inst.stop())
+}
+
+func testMessage(record map[string]any) Message {
+	tag := ""
+	return Message{
+		Time:   time.Now().UTC(),
+		Record: record,
+		tag:    &tag,
+	}
+}
+
+type collectMessage struct {
+	msg         Message
+	collectedWG sync.WaitGroup
+}
+
+func newTestInputPlugin() *testInputPlugin {
+	return &testInputPlugin{
+		inputs: make(chan *collectMessage),
+	}
+}
+
+type testInputPlugin struct {
+	initCount      atomic.Int64
+	collectRunning atomic.Bool
+	inputs         chan *collectMessage
+}
+
+var _ InputPlugin = (*testInputPlugin)(nil)
+
+func (t *testInputPlugin) Init(ctx context.Context, fbit *Fluentbit) error {
+	t.initCount.Add(1)
+	return nil
+}
+
+func (t *testInputPlugin) Collect(ctx context.Context, ch chan<- Message) error {
+	t.collectRunning.Store(true)
+	defer t.collectRunning.Store(false)
+
+	for {
+		select {
+		case m := <-t.inputs:
+			ch <- m.msg
+			m.collectedWG.Done()
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+func (t *testInputPlugin) enqueue(m Message) (waitForCollected func()) {
+	cm := &collectMessage{msg: m}
+	cm.collectedWG.Add(1)
+	t.inputs <- cm
+
+	return cm.collectedWG.Wait
+}
+
+func decodeMessages(t testing.TB, msgpackBytes []byte) []Message {
+	var messages []Message
+
+	dec := msgpack.NewDecoder(bytes.NewReader(msgpackBytes))
+	for {
+		msg, err := decodeMsg(dec, "")
+		if errors.Is(err, io.EOF) {
+			return messages
+		}
+		require.NoError(t, err)
+
+		messages = append(messages, msg)
+	}
+}
+
 func TestInputCallbackCtrlC(t *testing.T) {
 	inst := newTestInputInstance(t, testPluginInputCallbackCtrlC{})
 
@@ -78,10 +203,8 @@ func TestInputCallbackCtrlC(t *testing.T) {
 	cdone := make(chan struct{})
 	timeout := time.After(1 * time.Second)
 
-	ptr := unsafe.Pointer(nil)
-
 	go func() {
-		inst.callback(&ptr, nil)
+		testCallback(inst)
 		close(cdone)
 	}()
 
@@ -130,12 +253,11 @@ func TestInputCallbackDangle(t *testing.T) {
 		ticker := time.NewTicker(collectInterval)
 		defer ticker.Stop()
 
-		inst.callback(&ptr, nil)
+		testCallback(inst)
 		for {
 			select {
 			case <-ticker.C:
-				//FLBPluginInputCallback(&ptr, nil)
-				inst.callback(&ptr, nil)
+				testCallback(inst)
 			case <-cdone:
 				return
 			}
@@ -199,7 +321,7 @@ func TestInputCallbackInfinite(t *testing.T) {
 		for {
 			select {
 			case <-ticker.C:
-				inst.callback(&ptr, nil)
+				testCallback(inst)
 				if ptr != nil {
 					close(cdone)
 					return
@@ -271,7 +393,7 @@ func TestInputCallbackLatency(t *testing.T) {
 		ticker := time.NewTicker(collectInterval)
 		defer ticker.Stop()
 
-		buf, _ := testFLBPluginInputCallback(inst)
+		buf, _ := testCallback(inst)
 		if len(buf) > 0 {
 			cmsg <- buf
 		}
@@ -283,7 +405,7 @@ func TestInputCallbackLatency(t *testing.T) {
 				t.Log("---- collect done")
 				return
 			case <-ticker.C:
-				buf, _ := testFLBPluginInputCallback(inst)
+				buf, _ := testCallback(inst)
 				if len(buf) > 0 {
 					cmsg <- buf
 				}
@@ -387,15 +509,13 @@ func TestInputCallbackInfiniteConcurrent(t *testing.T) {
 		ticker := time.NewTicker(time.Second * 1)
 		defer ticker.Stop()
 
-		//FLBPluginInputCallback(&ptr, nil)
-		inst.callback(&ptr, nil)
+		testCallback(inst)
 		close(cstarted)
 
 		for {
 			select {
 			case <-ticker.C:
-				//FLBPluginInputCallback(&ptr, nil)
-				inst.callback(&ptr, nil)
+				testCallback(inst)
 			case <-inst.runCtx.Done():
 				return
 			}

plugin.go

@@ -187,7 +187,7 @@ type pluginInstance struct {
 	meta                 pluginMetadata
 	cmetricsCtxProvider  cmetricsContextProvider
 	configLoaderProvider configLoaderProvider
-	runningWG            sync.WaitGroup
+	runningWG            sync.WaitGroup // Number of running preRun and callback methods.
 
 	// mu protects all members below.
 	// It is generally held during state checks and transitions but not during long-running callbacks.
@@ -319,7 +319,10 @@ func (p *pluginInstance) inputPreRun() error {
 	p.msgChannel = make(chan Message, p.maxBufferedMessages)
 	p.state = instanceStateRunnable
 
+	p.runningWG.Add(1)
 	go func() {
+		defer p.runningWG.Done()
+
 		err := p.meta.input.Collect(runCtx, p.msgChannel)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "collect error: %v\n", err)
@@ -353,8 +356,8 @@ func (p *pluginInstance) outputPreRun() error {
 	p.msgChannel = make(chan Message)
 	p.state = instanceStateRunnable
 
+	p.runningWG.Add(1)
 	go func() {
-		p.runningWG.Add(1)
 		defer p.runningWG.Done()
 		if err := p.meta.output.Flush(p.runCtx, p.msgChannel); err != nil {
 			fmt.Fprintf(os.Stderr, "FLBPluginOutputPreRun error: %v\n", err)
@@ -377,6 +380,7 @@ func (p *pluginInstance) outputPreRun() error {
 func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
 	p.mu.Lock()
 	if p.state != instanceStateRunnable {
+		p.mu.Unlock()
 		return input.FLB_RETRY
 	}
 
@@ -437,7 +441,7 @@ func (p *pluginInstance) stop() error {
 	defer p.mu.Unlock()
 
 	if p.state != instanceStateRunnable && p.state != instanceStatePreExit {
-		return fmt.Errorf("cannot stop plugin in state %q", p.state)
+		return nil
 	}
 
 	if p.runCancel != nil {