Make hot reloadable

cshared.go

@@ -44,6 +44,16 @@ var (
 	maxBufferedMessages = defaultMaxBufferedMessages
 )
 
+//export FLBPluginPreRegister
+func FLBPluginPreRegister(hotReloading C.int) int {
+	if hotReloading == C.int(1) {
+		initWG.Add(1)
+		registerWG.Add(1)
+	}
+
+	return input.FLB_OK
+}
+
 // FLBPluginRegister registers a plugin in the context of the fluent-bit runtime, a name and description
 // can be provided.
 //
@@ -72,6 +82,24 @@ func FLBPluginRegister(def unsafe.Pointer) int {
 	return out
 }
 
+func cleanup() int {
+	if unregister != nil {
+		unregister()
+		unregister = nil
+	}
+
+	if runCancel != nil {
+		runCancel()
+		runCancel = nil
+	}
+
+	if theChannel != nil {
+		defer close(theChannel)
+	}
+
+	return input.FLB_OK
+}
+
 // FLBPluginInit this method gets invoked once by the fluent-bit runtime at initialisation phase.
 // here all the plugin context should be initialized and any data or flag required for
 // plugins to execute the collect or flush callback.
@@ -80,8 +108,6 @@ func FLBPluginRegister(def unsafe.Pointer) int {
 func FLBPluginInit(ptr unsafe.Pointer) int {
 	defer initWG.Done()
 
-	registerWG.Wait()
-
 	if theInput == nil && theOutput == nil {
 		fmt.Fprintf(os.Stderr, "no input or output registered\n")
 		return input.FLB_RETRY
@@ -167,6 +193,118 @@ var theInputLock sync.Mutex
 //
 // This function will invoke Collect only once to preserve backward
 // compatible behavior. There are unit tests to enforce this behavior.
+func prepareInputCollector() (err error) {
+	runCtx, runCancel = context.WithCancel(context.Background())
+	theChannel = make(chan Message, maxBufferedMessages)
+
+	theInputLock.Lock()
+
+	go func(theChannel chan<- Message) {
+		defer theInputLock.Unlock()
+
+		go func(theChannel chan<- Message) {
+			err = theInput.Collect(runCtx, theChannel)
+		}(theChannel)
+
+		for {
+			select {
+			case <-runCtx.Done():
+				log.Printf("goroutine will be stopping: name=%q\n", theName)
+				return
+			}
+		}
+
+		if err != nil {
+			fmt.Fprintf(os.Stderr,
+				"collect error: %s\n", err.Error())
+		}
+	}(theChannel)
+
+	return err
+}
+
+// FLBPluginInputPreRun this method gets invoked by the fluent-bit runtime, once the plugin has been
+// initialised, the plugin invoked only once before executing the input callbacks.
+//
+//export FLBPluginInputPreRun
+func FLBPluginInputPreRun(useHotReload C.int) int {
+	registerWG.Wait()
+
+	var err error
+	err = prepareInputCollector()
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "run: %s\n", err)
+		return input.FLB_ERROR
+	}
+
+	return input.FLB_OK
+}
+
+
+// FLBPluginInputPause this method gets invoked by the fluent-bit runtime, once the plugin has been
+// paused, the plugin invoked this method and entering paused state.
+//
+//export FLBPluginInputPause
+func FLBPluginInputPause() {
+	if runCancel != nil {
+		runCancel()
+		runCancel = nil
+	}
+
+	if theChannel != nil {
+		close(theChannel)
+		theChannel = nil
+	}
+}
+
+// FLBPluginInputResume this method gets invoked by the fluent-bit runtime, once the plugin has been
+// resumeed, the plugin invoked this method and re-running state.
+//
+//export FLBPluginInputResume
+func FLBPluginInputResume() {
+	var err error
+	err = prepareInputCollector()
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "run: %s\n", err)
+	}
+}
+
+//export FLBPluginOutputPreRun
+func FLBPluginOutputPreRun(useHotReload C.int) int {
+	registerWG.Wait()
+
+	var err error
+	runCtx, runCancel = context.WithCancel(context.Background())
+	theChannel = make(chan Message)
+	go func(runCtx context.Context) {
+		go func(runCtx context.Context) {
+			err = theOutput.Flush(runCtx, theChannel)
+		}(runCtx)
+
+		for {
+			select {
+			case <-runCtx.Done():
+				log.Printf("goroutine will be stopping: name=%q\n", theName)
+				return
+			}
+		}
+
+	}(runCtx)
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "run: %s\n", err)
+		return output.FLB_ERROR
+	}
+
+	return output.FLB_OK
+}
+
+// FLBPluginInputCallback this method gets invoked by the fluent-bit runtime, once the plugin has been
+// initialised, the plugin implementation is responsible for handling the incoming data and the context
+// that gets past, for long-living collectors the plugin itself should keep a running thread and fluent-bit
+// will not execute further callbacks.
 //
 //export FLBPluginInputCallback
 func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int {
@@ -177,23 +315,6 @@ func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int {
 		return input.FLB_RETRY
 	}
 
-	once.Do(func() {
-		runCtx, runCancel = context.WithCancel(context.Background())
-		theChannel = make(chan Message, maxBufferedMessages)
-
-		theInputLock.Lock()
-
-		go func(theChannel chan<- Message) {
-			defer theInputLock.Unlock()
-
-			err := theInput.Collect(runCtx, theChannel)
-			if err != nil {
-				fmt.Fprintf(os.Stderr,
-					"collect error: %s\n", err.Error())
-			}
-		}(theChannel)
-	})
-
 	buf := bytes.NewBuffer([]byte{})
 
 	for loop := min(len(theChannel), maxBufferedMessages); loop > 0; loop-- {
@@ -261,18 +382,6 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 	}
 
 	var err error
-	once.Do(func() {
-		runCtx, runCancel = context.WithCancel(context.Background())
-		theChannel = make(chan Message)
-		go func() {
-			err = theOutput.Flush(runCtx, theChannel)
-		}()
-	})
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "run: %s\n", err)
-		return output.FLB_ERROR
-	}
-
 	select {
 	case <-runCtx.Done():
 		err = runCtx.Err()
@@ -319,24 +428,29 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 			return output.FLB_ERROR
 		}
 
-		if d := len(entry); d != 2 {
-			fmt.Fprintf(os.Stderr, "unexpected entry length: %d\n", d)
+		slice := reflect.ValueOf(entry)
+		if slice.Kind() != reflect.Slice || slice.Len() < 2 {
+			fmt.Fprintf(os.Stderr, "unexpected entry length: %d\n", slice.Len())
 			return output.FLB_ERROR
 		}
 
-		ft, ok := entry[0].(bigEndianTime)
-		if !ok {
+		var t time.Time
+		ts := slice.Index(0).Interface()
+		switch ft := ts.(type) {
+		case bigEndianTime:
+			t = time.Time(ft)
+		case []interface{}:
+			s := reflect.ValueOf(ft)
+			st := s.Index(0).Interface()
+			ty := st.(bigEndianTime)
+			t = time.Time(ty)
+		default:
 			fmt.Fprintf(os.Stderr, "unexpected entry time type: %T\n", entry[0])
 			return output.FLB_ERROR
 		}
 
-		t := time.Time(ft)
-
-		recVal, ok := entry[1].(map[any]any)
-		if !ok {
-			fmt.Fprintf(os.Stderr, "unexpected entry record type: %T\n", entry[1])
-			return output.FLB_ERROR
-		}
+		data := slice.Index(1)
+		recVal := data.Interface().(map[interface{}]interface{})
 
 		var rec map[string]string
 		if d := len(recVal); d != 0 {
@@ -348,13 +462,22 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 					return output.FLB_ERROR
 				}
 
-				val, ok := v.([]uint8)
-				if !ok {
+				var val string
+				switch tv := v.(type) {
+				case []uint8:
+					val = string(tv)
+				case uint64:
+					val = strconv.FormatUint(tv, 10)
+				case int64:
+					val = strconv.FormatInt(tv, 10)
+				case bool:
+					val = strconv.FormatBool(tv)
+				default:
 					fmt.Fprintf(os.Stderr, "unexpected record value type: %T\n", v)
 					return output.FLB_ERROR
 				}
 
-				rec[key] = string(val)
+				rec[key] = val
 			}
 		}
 
@@ -374,21 +497,7 @@ func FLBPluginFlush(data unsafe.Pointer, clength C.int, ctag *C.char) int {
 //
 //export FLBPluginExit
 func FLBPluginExit() int {
-	log.Printf("calling FLBPluginExit(): name=%q\n", theName)
-
-	if unregister != nil {
-		unregister()
-	}
-
-	if runCancel != nil {
-		runCancel()
-	}
-
-	if theChannel != nil {
-		defer close(theChannel)
-	}
-
-	return input.FLB_OK
+	return cleanup()
 }
 
 type flbInputConfigLoader struct {

cshared_test.go

@@ -40,6 +40,13 @@ func TestInputCallbackCtrlC(t *testing.T) {
 
 	ptr := unsafe.Pointer(nil)
 
+	// prepare channel for input explicitly.
+	err := prepareInputCollector()
+	if err != nil {
+		t.Fail()
+		return
+	}
+
 	go func() {
 		FLBPluginInputCallback(&ptr, nil)
 		cdone <- true
@@ -86,6 +93,12 @@ func TestInputCallbackDangle(t *testing.T) {
 	cdone := make(chan bool)
 	ptr := unsafe.Pointer(nil)
 
+	// prepare channel for input explicitly.
+	err := prepareInputCollector()
+	if err != nil {
+		t.Fail()
+	}
+
 	go func() {
 		t := time.NewTicker(collectInterval)
 		defer t.Stop()
@@ -156,6 +169,13 @@ func TestInputCallbackInfinite(t *testing.T) {
 	cshutdown := make(chan bool)
 	ptr := unsafe.Pointer(nil)
 
+	// prepare channel for input explicitly.
+	err := prepareInputCollector()
+	if err != nil {
+		t.Fail()
+		return
+	}
+
 	go func() {
 		t := time.NewTicker(collectInterval)
 		defer t.Stop()
@@ -237,6 +257,13 @@ func TestInputCallbackLatency(t *testing.T) {
 	cstarted := make(chan bool)
 	cmsg := make(chan []byte)
 
+	// prepare channel for input explicitly.
+	err := prepareInputCollector()
+	if err != nil {
+		t.Fail()
+		return
+	}
+
 	go func() {
 		t := time.NewTicker(collectInterval)
 		defer t.Stop()
@@ -363,6 +390,12 @@ func TestInputCallbackInfiniteConcurrent(t *testing.T) {
 
 	concurrentWait.Add(64)
 
+	// prepare channel for input explicitly.
+	err := prepareInputCollector()
+	if err != nil {
+		t.Fail()
+	}
+
 	go func(cstarted chan bool) {
 		ticker := time.NewTicker(time.Second * 1)
 		defer ticker.Stop()

examples/out_gstdout/go.mod

@@ -2,6 +2,11 @@ module github.com/fluent/fluent-bit-go/examples/gstdout
 
 go 1.21.0
 
-require github.com/fluent/fluent-bit-go v0.0.0-20200420155746-e125cab17963
+require github.com/calyptia/plugin v0.1.6
 
-replace github.com/fluent/fluent-bit-go => ../..
+require (
+	github.com/calyptia/cmetrics-go v0.1.7 // indirect
+	github.com/ugorji/go/codec v1.2.11 // indirect
+)
+
+replace github.com/calyptia/plugin => ../..