wip

Author: alecholmes

.gitignore

@@ -176,6 +176,7 @@ cmake-build-*/
 
 # IntelliJ
 out/
+/.idea/
 
 # mpeltonen/sbt-idea plugin
 .idea_modules/

cshared.go

@@ -57,8 +57,13 @@ func FLBPluginPreRegister(hotReloading C.int) int {
 // can be provided.
 //
 //export FLBPluginRegister
-func FLBPluginRegister(def unsafe.Pointer) int {
-	defer registerWG.Done()
+func FLBPluginRegister(def unsafe.Pointer) (returnCode int) {
+	defer func() {
+		// Only unblock waiters on registerWG if registration succeeds
+		if returnCode == input.FLB_OK {
+			registerWG.Done()
+		}
+	}()
 
 	meta := pluginMeta.Load()
 	if meta == nil {
@@ -130,7 +135,7 @@ func (p *pluginInstance) inputPreRun() error {
 		return nil
 	}
 
-	if p.state != instanceStateInitialized && p.state != instanceStatePaused {
+	if p.state != instanceStateInitialized {
 		return fmt.Errorf("invalid plugin state %q", p.state)
 	}
 
@@ -145,6 +150,7 @@ func (p *pluginInstance) inputPreRun() error {
 	p.runCtx = runCtx
 	p.runCancel = runCancel
 	p.msgChannel = make(chan Message, maxBufferedMessages)
+	p.state = instanceStateRunnable
 
 	go func() {
 		err := p.meta.input.Collect(runCtx, p.msgChannel)
@@ -162,8 +168,8 @@ func (p *pluginInstance) inputPreRun() error {
 	return nil
 }
 
-// FLBPluginInputPreRun this method gets invoked by the fluent-bit runtime, once the plugin has been
-// initialized, the plugin invoked only once before executing the input callbacks.
+// FLBPluginInputPreRun is invoked by the fluent-bit runtime after the plugin has been
+// initialized using FLBPluginRegister but before executing plugin callback functions.
 //
 //export FLBPluginInputPreRun
 func FLBPluginInputPreRun(useHotReload C.int) int {
@@ -208,14 +214,14 @@ func (p *pluginInstance) pause() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if !(p.state == instanceStateInitialized) {
+	if p.state != instanceStateRunnable {
 		return fmt.Errorf("cannot pause plugin in state %q", p.state)
 	}
 
 	p.runCancel()
 	close(p.msgChannel)
 
-	p.state = instanceStatePaused
+	p.state = instanceStateInitialized
 	p.runCtx = nil
 	p.runCancel = nil
 	p.msgChannel = nil
@@ -227,7 +233,7 @@ func (p *pluginInstance) resume() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if p.state != instanceStatePaused {
+	if p.state != instanceStateInitialized {
 		return fmt.Errorf("cannot resume plugin in state %q", p.state)
 	}
 
@@ -310,19 +316,30 @@ func FLBPluginOutputPreRun(useHotReload C.int) int {
 		panic("plugin not initialized")
 	}
 
-	instance.outputPreRun()
+	if err := instance.outputPreRun(); err != nil {
+		fmt.Fprintf(os.Stderr, "plugin pre-run error: %v\n", err)
+		return flbReturnCode(err)
+	}
 
 	return input.FLB_OK
 }
 
-func (p *pluginInstance) outputPreRun() {
+func (p *pluginInstance) outputPreRun() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	// TODO: state check
+	// Only input plugins have a pre-run step
+	if p.meta.output == nil {
+		return fmt.Errorf("plugin is not an output plugin")
+	}
+
+	if p.state != instanceStateInitialized {
+		return fmt.Errorf("invalid plugin state %q", p.state)
+	}
 
 	p.runCtx, p.runCancel = context.WithCancel(context.Background())
 	p.msgChannel = make(chan Message)
+	p.state = instanceStateRunnable
 
 	go func() {
 		if err := p.meta.output.Flush(p.runCtx, p.msgChannel); err != nil {
@@ -335,6 +352,8 @@ func (p *pluginInstance) outputPreRun() {
 
 		log.Printf("goroutine will be stopping: name=%q\n", p.meta.name)
 	}()
+
+	return nil
 }
 
 // FLBPluginInputCallback this method gets invoked by the fluent-bit runtime, once the plugin has been
@@ -362,7 +381,7 @@ func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	if p.state != instanceStateInitialized {
+	if p.state != instanceStateRunnable {
 		return input.FLB_RETRY
 	}
 
@@ -428,7 +447,7 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 	instance := currInstance
 	currInstanceMu.Unlock()
 
-	if instance == nil || instance.state != instanceStateInitialized { // TODO: ok?
+	if instance == nil || instance.state != instanceStateRunnable {
 		return output.FLB_RETRY
 	}
 
@@ -437,14 +456,9 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 		return output.FLB_RETRY
 	}
 
-	if instance == nil || instance.runCtx == nil {
-		return output.FLB_RETRY
-	}
-
-	var err error
 	select {
 	case <-instance.runCtx.Done():
-		err = instance.runCtx.Err()
+		err := instance.runCtx.Err()
 		if err != nil && !errors.Is(err, context.Canceled) {
 			fmt.Fprintf(os.Stderr, "run: %s\n", err)
 			return output.FLB_ERROR
@@ -465,10 +479,6 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 }
 
 func pluginFlush(instance *pluginInstance, tag string, b []byte) error {
-	if instance.msgChannel == nil {
-		return fmt.Errorf("no instance or channel available")
-	}
-
 	dec := msgpack.NewDecoder(bytes.NewReader(b))
 	for {
 		select {

cshared_test.go

@@ -605,7 +605,7 @@ func TestOutputFlush(t *testing.T) {
 	}
 	inst := newTestOutputInstance(t, out)
 	require.NoError(t, inst.init(nil))
-	inst.outputPreRun()
+	require.NoError(t, inst.outputPreRun())
 
 	msg := Message{
 		Time: now,

plugin.go

@@ -17,11 +17,20 @@ import (
 	"github.com/calyptia/plugin/metric"
 )
 
+func init() {
+	registerWG.Add(1)
+}
+
 var (
+	// registerWG blocks until FLBPluginRegister has successfully run.
+	registerWG sync.WaitGroup
+
 	// Current plugin metadata. This is nil if the plugin has not been registered yet.
+	// This is set by the RegisterInput, RegisterOutput, or RegisterCustom functions.
 	pluginMeta atomic.Pointer[pluginMetadata]
 )
 
+// pluginMeta describes a plugin and exposes hooks into its implementation.
 type pluginMetadata struct {
 	name   string
 	desc   string
@@ -31,9 +40,6 @@ type pluginMetadata struct {
 }
 
 var (
-	// registerWG gates access to pluginMeta before the plugin has been registered.
-	registerWG sync.WaitGroup
-
 	currInstanceMu sync.Mutex // Guards access to currInstance
 	currInstance   *pluginInstance
 )
@@ -42,10 +48,9 @@ type instanceState string
 
 const (
 	instanceStateInitialized instanceState = "initialized"
-	//instanceStatePreRun      instanceState = "preRun"
-	instanceStatePaused  instanceState = "paused"
-	instanceStatePreExit instanceState = "preExit"
-	instanceStateStopped instanceState = "stopped"
+	instanceStateRunnable    instanceState = "runnable"
+	instanceStatePreExit     instanceState = "preExit"
+	instanceStateStopped     instanceState = "stopped"
 )
 
 type retryableError error
@@ -103,13 +108,17 @@ func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 		return fmt.Errorf("unexpected plugin state %q", p.state)
 	}
 
+	newState := instanceStateInitialized
 	ctx, cancel := context.WithCancel(context.Background())
 	defer func() {
 		if initErr != nil {
 			cancel()
 			p.runCtx = nil
 			p.runCancel = nil
+
+			return
 		}
+		p.state = newState
 	}()
 
 	if p.meta.input != nil {
@@ -141,7 +150,6 @@ func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 	}
 
 	// TODO: other branches
-	p.state = instanceStateInitialized
 
 	return nil
 }
@@ -150,49 +158,11 @@ func (p *pluginInstance) stop() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if p.state == instanceStateStopped {
-		return
-	}
-
-	if p.runCancel != nil {
-		p.runCancel()
-	}
-	if p.msgChannel != nil {
-		close(p.msgChannel)
+	if p.state == instanceStateRunnable {
+		if err := p.pause(); err != nil {
+			panic(err)
+		}
 	}
-
-	p.state = instanceStatePaused
-}
-
-//// getCurrInstance safely gets the current plugin instance
-//func getCurrInstance() *pluginInstance {
-//	currInstanceMu.Lock()
-//	defer currInstanceMu.Unlock()
-//
-//	return currInstance
-//}
-//
-//// setCurrInstance safely sets the current plugin instance
-//func setCurrInstance(instance *pluginInstance) {
-//	currInstanceMu.Lock()
-//	defer currInstanceMu.Unlock()
-//
-//	currInstance = instance
-//}
-
-//// getOrCreateCurrInstance safely gets or creates the current plugin instance
-//func getOrCreateCurrInstance() *pluginInstance {
-//	currInstanceMu.Lock()
-//	defer currInstanceMu.Unlock()
-//
-//	if currInstance == nil {
-//		currInstance = &pluginInstance{}
-//	}
-//	return currInstance
-//}
-
-func init() {
-	registerWG.Add(1)
 }
 
 type Fluentbit struct {