wip

Author: alecholmes

cshared.go

@@ -36,8 +36,7 @@ const (
 	collectInterval = 1000 * time.Nanosecond
 )
 
-var (
-	unregister          func()
+const (
 	maxBufferedMessages = defaultMaxBufferedMessages
 )
 
@@ -71,32 +70,32 @@ func FLBPluginRegister(def unsafe.Pointer) (returnCode int) {
 		return input.FLB_RETRY
 	}
 
-	if meta.input != nil {
-		out := input.FLBPluginRegister(def, meta.name, meta.desc)
-		unregister = func() {
-			input.FLBPluginUnregister(def)
-		}
-		return out
+	if unregisterFunc.Load() != nil {
+		fmt.Fprintf(os.Stderr, "plugin already registered\n")
+		return input.FLB_ERROR
 	}
 
-	if meta.output != nil {
-		out := output.FLBPluginRegister(def, meta.name, meta.desc)
-		unregister = func() {
-			output.FLBPluginUnregister(def)
-		}
-		return out
-	}
+	var newUnregisterFunc func()
 
-	if meta.custom != nil {
-		out := custom.FLBPluginRegister(def, meta.name, meta.desc)
-		unregister = func() {
-			custom.FLBPluginUnregister(def)
-		}
+	if meta.input != nil {
+		returnCode = input.FLBPluginRegister(def, meta.name, meta.desc)
+		newUnregisterFunc = func() { input.FLBPluginUnregister(def) }
+	} else if meta.output != nil {
+		returnCode = output.FLBPluginRegister(def, meta.name, meta.desc)
+		newUnregisterFunc = func() { output.FLBPluginUnregister(def) }
+	} else if meta.custom != nil {
+		returnCode = custom.FLBPluginRegister(def, meta.name, meta.desc)
+		newUnregisterFunc = func() { custom.FLBPluginUnregister(def) }
+	} else {
+		fmt.Fprintf(os.Stderr, "no input, output, or custom plugin registered\n")
+		return input.FLB_RETRY
+	}
 
-		return out
+	if returnCode == input.FLB_OK {
+		unregisterFunc.Store(&newUnregisterFunc)
 	}
 
-	return input.FLB_ERROR
+	return
 }
 
 // FLBPluginInit this method gets invoked once by the fluent-bit runtime at initialisation phase.
@@ -558,18 +557,18 @@ func decodeMsg(dec *msgpack.Decoder, tag string) (Message, error) {
 //
 //export FLBPluginExit
 func FLBPluginExit() int {
-	if unregister != nil {
-		unregister()
-		unregister = nil
-	}
-
 	currInstanceMu.Lock()
 	defer currInstanceMu.Unlock()
 
 	if currInstance != nil {
-		currInstance.stop()
+		currInstance.pause() //nolint:errcheck
+		currInstance = nil
+	}
+
+	if unregister := unregisterFunc.Load(); unregister != nil {
+		(*unregister)()
+		unregisterFunc.Store(nil)
 	}
-	currInstance = nil
 
 	return input.FLB_OK
 }

plugin.go

@@ -27,30 +27,33 @@ var (
 
 	// 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     atomic.Pointer[pluginMetadata]
+	unregisterFunc atomic.Pointer[func()]
+
+	// currInstanceMu guards access to currInstance
+	currInstanceMu sync.Mutex
+	// Current instance of the plugin, used by functions called from fluent-bit like FLBPluginInit.
+	currInstance *pluginInstance
 )
 
 // pluginMeta describes a plugin and exposes hooks into its implementation.
 type pluginMetadata struct {
-	name   string
-	desc   string
+	name string
+	desc string
+
+	// Exactly one of the following will be set:
 	input  InputPlugin
 	output OutputPlugin
 	custom CustomPlugin
 }
 
-var (
-	currInstanceMu sync.Mutex // Guards access to currInstance
-	currInstance   *pluginInstance
-)
-
 type instanceState string
 
 const (
+	instanceStateCreated     instanceState = ""
 	instanceStateInitialized instanceState = "initialized"
 	instanceStateRunnable    instanceState = "runnable"
 	instanceStatePreExit     instanceState = "preExit"
-	instanceStateStopped     instanceState = "stopped"
 )
 
 type retryableError error
@@ -75,31 +78,38 @@ func newPluginInstance(meta pluginMetadata) *pluginInstance {
 		configLoaderProvider: func(ptr unsafe.Pointer) ConfigLoader {
 			return &flbInputConfigLoader{ptr: ptr}
 		},
+		state: instanceStateCreated,
 	}
 }
 
+// pluginInstance is an instance of a plugin.
 type pluginInstance struct {
 	meta                 pluginMetadata
 	cmetricsCtxProvider  cmetricsContextProvider
 	configLoaderProvider configLoaderProvider
 
-	mu         sync.RWMutex
+	mu         sync.RWMutex // Protects all members below:
 	state      instanceState
 	runCtx     context.Context
 	runCancel  context.CancelFunc
 	msgChannel chan Message
 }
 
+// withCMetricsContextProvider overrides the cmetricsContextProvider.
+// This must be called immediately after creating the instance.
 func (p *pluginInstance) withCMetricsContextProvider(provider cmetricsContextProvider) *pluginInstance {
 	p.cmetricsCtxProvider = provider
 	return p
 }
 
+// withConfigLoaderProvider overrides the configLoaderProvider.
+// This must be called immediately after creating the instance.
 func (p *pluginInstance) withConfigLoaderProvider(provider configLoaderProvider) *pluginInstance {
 	p.configLoaderProvider = provider
 	return p
 }
 
+// init initializes a newly created plugin instance.
 func (p *pluginInstance) init(ptr unsafe.Pointer) (initErr error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()