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

@@ -22,24 +22,6 @@ import (
 	"github.com/calyptia/plugin/metric"
 )
 
-type testPluginInputCallbackCtrlC struct{}
-
-func (t testPluginInputCallbackCtrlC) Init(ctx context.Context, fbit *Fluentbit) error {
-	return nil
-}
-
-func (t testPluginInputCallbackCtrlC) Collect(ctx context.Context, ch chan<- Message) error {
-	return nil
-}
-
-type fakeConfigLoader map[string]string
-
-var _ ConfigLoader = (fakeConfigLoader)(nil)
-
-func (f fakeConfigLoader) String(key string) string {
-	return f[key]
-}
-
 func newTestInputInstance(t testing.TB, input InputPlugin) *pluginInstance {
 	inst := pluginInstanceWithFakes(newPluginInstance(pluginMetadata{
 		name:  "test-plugin",
@@ -100,29 +82,56 @@ 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)
 	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())
 }
 
+// testMessage returns a Message with the given record map and current timestamp.
 func testMessage(record map[string]any) Message {
 	tag := ""
 	return Message{
@@ -132,21 +141,19 @@ func testMessage(record map[string]any) Message {
 	}
 }
 
-type collectMessage struct {
-	msg         Message
-	collectedWG sync.WaitGroup
-}
-
 func newTestInputPlugin() *testInputPlugin {
 	return &testInputPlugin{
 		inputs: make(chan *collectMessage),
 	}
 }
 
+// testInputPlugin is an InputPlugin used to help test plugin callback and concurrency behavior.
 type testInputPlugin struct {
-	initCount      atomic.Int64
-	collectRunning atomic.Bool
-	inputs         chan *collectMessage
+	initCount      atomic.Int64            // Count of calls to Init method.
+	collectRunning atomic.Bool             // Indicates whether the Collect method is running.
+	onCollectDone  func(ch chan<- Message) // Settable callback invoked when Collect is about to return.
+
+	inputs chan *collectMessage
 }
 
 var _ InputPlugin = (*testInputPlugin)(nil)
@@ -166,11 +173,16 @@ 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
 		}
 	}
 }
 
+// enqueue the message m to be processed by Collect. When called, the returned function
+// blocks until a running Collect puts m on the plugin's input channel.
 func (t *testInputPlugin) enqueue(m Message) (waitForCollected func()) {
 	cm := &collectMessage{msg: m}
 	cm.collectedWG.Add(1)
@@ -179,6 +191,12 @@ func (t *testInputPlugin) enqueue(m Message) (waitForCollected func()) {
 	return cm.collectedWG.Wait
 }
 
+// collectMessage is a helper wrapper used by testInputPlugin that wraps a Message.
+type collectMessage struct {
+	msg         Message
+	collectedWG sync.WaitGroup // Decremented to 0 when testInputPlugin Collect processes the message.
+}
+
 func decodeMessages(t testing.TB, msgpackBytes []byte) []Message {
 	var messages []Message
 
@@ -194,11 +212,21 @@ func decodeMessages(t testing.TB, msgpackBytes []byte) []Message {
 	}
 }
 
+type testPluginInputCallbackCtrlC struct{}
+
+func (t testPluginInputCallbackCtrlC) Init(ctx context.Context, fbit *Fluentbit) error {
+	return nil
+}
+
+func (t testPluginInputCallbackCtrlC) Collect(ctx context.Context, ch chan<- Message) error {
+	return nil
+}
+
 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 +275,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 +340,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 +415,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 +531,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 +744,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,
@@ -737,3 +765,11 @@ func TestOutputFlush(t *testing.T) {
 	assert.NoError(t, pluginFlush(inst, "foobar", b))
 	wg.Wait()
 }
+
+type fakeConfigLoader map[string]string
+
+var _ ConfigLoader = (fakeConfigLoader)(nil)
+
+func (f fakeConfigLoader) String(key string) string {
+	return f[key]
+}

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
 		}
 	}