wip

Author: alecholmes

cshared.go

@@ -135,7 +135,7 @@ func FLBPluginInputPreRun(useHotReload C.int) int {
 		return input.FLB_ERROR
 	}
 
-	if err := instance.inputPreRun(); err != nil {
+	if err := instance.resume(); err != nil {
 		fmt.Fprintf(os.Stderr, "plugin pre-run error: %v\n", err)
 		return flbReturnCode(err)
 	}
@@ -211,7 +211,7 @@ func FLBPluginOutputPreRun(useHotReload C.int) int {
 		panic("plugin not initialized")
 	}
 
-	if err := instance.outputPreRun(); err != nil {
+	if err := instance.resume(); err != nil {
 		fmt.Fprintf(os.Stderr, "plugin pre-run error: %v\n", err)
 		return flbReturnCode(err)
 	}

cshared_test.go

@@ -100,27 +100,54 @@ func TestInputCallbackLifecycle(t *testing.T) {
 	require.Equal(t, input.FLB_RETRY, callbackResp, "pre-run must be called before callback")
 
 	// Pre-run
-	require.NoError(t, inst.inputPreRun())
+	require.NoError(t, inst.resume())
 	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"`)
+	require.ErrorContains(t, inst.resume(), `invalid plugin state "runnable"`)
 
 	// Callback
-	time.Sleep(time.Second)
+	//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
+	// Stop (ensuring collect loop exits cleanly)
+	plugin.onCollectDone = func(ch chan<- Message) {
+		// Keep enqueueing after stop to ensure the plugin message channel wasn't closed early
+		time.Sleep(10 * time.Millisecond)
+		ch <- testMessage(map[string]any{"name": "m3"})
+	}
 	require.NoError(t, inst.stop())
 	require.False(t, plugin.collectRunning.Load())
+	require.NoError(t, inst.stop(), "stop should be idempotent")
+
+	callbackBytes, callbackResp = testCallback(inst)
+	require.Equal(t, input.FLB_RETRY, callbackResp)
+	assert.Empty(t, callbackBytes)
+
+	// Resume stopped pipeline
+	require.NoError(t, inst.resume())
+	require.ErrorContains(t, inst.resume(), `invalid plugin state "runnable"`)
+	callbackBytes, callbackResp = testCallback(inst)
+	require.Equal(t, input.FLB_OK, callbackResp)
+	assert.Empty(t, callbackBytes, "m3 message from earlier not dequeued")
+	require.Eventually(t, plugin.collectRunning.Load, time.Second, time.Millisecond,
+		"collect background loop should have started running")
+	m4 := testMessage(map[string]any{"name": "m4"})
+	plugin.enqueue(m4)()
+	callbackBytes, callbackResp = testCallback(inst)
+	require.Equal(t, input.FLB_OK, callbackResp)
+	require.Equal(t, []Message{m4}, decodeMessages(t, callbackBytes))
+
+	// Stop again
 	require.NoError(t, inst.stop())
+	require.False(t, plugin.collectRunning.Load())
 }
 
 func testMessage(record map[string]any) Message {
@@ -147,6 +174,7 @@ type testInputPlugin struct {
 	initCount      atomic.Int64
 	collectRunning atomic.Bool
 	inputs         chan *collectMessage
+	onCollectDone  func(ch chan<- Message)
 }
 
 var _ InputPlugin = (*testInputPlugin)(nil)
@@ -166,6 +194,9 @@ func (t *testInputPlugin) Collect(ctx context.Context, ch chan<- Message) error
 			ch <- m.msg
 			m.collectedWG.Done()
 		case <-ctx.Done():
+			if t.onCollectDone != nil {
+				t.onCollectDone(ch)
+			}
 			return nil
 		}
 	}
@@ -198,7 +229,7 @@ func TestInputCallbackCtrlC(t *testing.T) {
 	inst := newTestInputInstance(t, testPluginInputCallbackCtrlC{})
 
 	require.NoError(t, inst.init(nil))
-	require.NoError(t, inst.inputPreRun())
+	require.NoError(t, inst.resume())
 
 	cdone := make(chan struct{})
 	timeout := time.After(1 * time.Second)
@@ -247,7 +278,7 @@ func TestInputCallbackDangle(t *testing.T) {
 
 	// prepare channel for input explicitly.
 	require.NoError(t, inst.init(ptr))
-	require.NoError(t, inst.inputPreRun())
+	require.NoError(t, inst.resume())
 
 	go func() {
 		ticker := time.NewTicker(collectInterval)
@@ -312,7 +343,7 @@ func TestInputCallbackInfinite(t *testing.T) {
 
 	// prepare channel for input explicitly.
 	require.NoError(t, inst.init(ptr))
-	require.NoError(t, inst.inputPreRun())
+	require.NoError(t, inst.resume())
 
 	go func() {
 		ticker := time.NewTicker(collectInterval)
@@ -387,7 +418,7 @@ func TestInputCallbackLatency(t *testing.T) {
 
 	// prepare channel for input explicitly.
 	require.NoError(t, inst.init(nil))
-	require.NoError(t, inst.inputPreRun())
+	require.NoError(t, inst.resume())
 
 	go func() {
 		ticker := time.NewTicker(collectInterval)
@@ -503,7 +534,7 @@ func TestInputCallbackInfiniteConcurrent(t *testing.T) {
 
 	// prepare channel for input explicitly.
 	require.NoError(t, inst.init(ptr))
-	require.NoError(t, inst.inputPreRun())
+	require.NoError(t, inst.resume())
 
 	go func() {
 		ticker := time.NewTicker(time.Second * 1)
@@ -716,7 +747,7 @@ func TestOutputFlush(t *testing.T) {
 	}
 	inst := newTestOutputInstance(t, out)
 	require.NoError(t, inst.init(nil))
-	require.NoError(t, inst.outputPreRun())
+	require.NoError(t, inst.resume())
 
 	msg := Message{
 		Time: now,

plugin.go

@@ -300,10 +300,7 @@ func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 }
 
 // inputPreRun transitions an initialized input plugin into runnable state.
-func (p *pluginInstance) inputPreRun() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
+func (p *pluginInstance) inputPreRunWithLock() error {
 	// Only input plugins have a pre-run step
 	if p.meta.input == nil {
 		return nil
@@ -339,10 +336,7 @@ func (p *pluginInstance) inputPreRun() error {
 }
 
 // outputPreRun transitions an output plugin into runnable state.
-func (p *pluginInstance) outputPreRun() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
+func (p *pluginInstance) outputPreRunWithLock() error {
 	// Only input plugins have a pre-run step
 	if p.meta.output == nil {
 		return fmt.Errorf("plugin is not an output plugin")
@@ -468,15 +462,15 @@ func (p *pluginInstance) resume() error {
 	defer p.mu.Unlock()
 
 	if p.state != instanceStateInitialized {
-		return fmt.Errorf("cannot resume plugin in state %q", p.state)
+		return fmt.Errorf("invalid plugin state %q", p.state)
 	}
 
 	if p.meta.input != nil {
-		if err := p.inputPreRun(); err != nil {
+		if err := p.inputPreRunWithLock(); err != nil {
 			return err
 		}
 	} else if p.meta.output != nil {
-		if err := p.outputPreRun(); err != nil {
+		if err := p.outputPreRunWithLock(); err != nil {
 			return err
 		}
 	}