wip

Author: alecholmes

cshared.go

@@ -11,9 +11,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log"
 	"os"
-	"runtime"
 	"strconv"
 	"strings"
 	"time"
@@ -121,41 +119,6 @@ func FLBPluginInit(ptr unsafe.Pointer) (respCode int) {
 	return input.FLB_OK
 }
 
-func (p *pluginInstance) inputPreRun() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	// Only input plugins have a pre-run step
-	if p.meta.input == nil {
-		return nil
-	}
-
-	if p.state != instanceStateInitialized {
-		return fmt.Errorf("invalid plugin state %q", p.state)
-	}
-
-	runCtx, runCancel := context.WithCancel(context.Background())
-	p.runCtx = runCtx
-	p.runCancel = runCancel
-	p.msgChannel = make(chan Message, p.maxBufferedMessages)
-	p.state = instanceStateRunnable
-
-	go func() {
-		err := p.meta.input.Collect(runCtx, p.msgChannel)
-		if err != nil {
-			fmt.Fprintf(os.Stderr, "collect error: %v\n", err)
-		}
-	}()
-
-	go func() {
-		<-runCtx.Done()
-
-		log.Printf("goroutine will be stopping: name=%q\n", p.meta.name)
-	}()
-
-	return nil
-}
-
 // FLBPluginInputPreRun is invoked by the fluent-bit runtime after the plugin has been
 // initialized using FLBPluginRegister but before executing plugin callback functions.
 //
@@ -198,48 +161,6 @@ func FLBPluginInputPause() {
 	}
 }
 
-func (p *pluginInstance) stop() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	if p.state != instanceStateRunnable && p.state != instanceStatePreExit {
-		return fmt.Errorf("cannot stop plugin in state %q", p.state)
-	}
-
-	if p.runCancel != nil {
-		p.runCancel()
-	}
-
-	p.runningWG.Wait()
-
-	if p.msgChannel != nil {
-		close(p.msgChannel)
-	}
-
-	p.state = instanceStateInitialized
-	p.runCtx = nil
-	p.runCancel = nil
-	p.msgChannel = nil
-
-	return nil
-}
-
-func (p *pluginInstance) resume() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	if p.state != instanceStateInitialized {
-		return fmt.Errorf("cannot resume plugin in state %q", p.state)
-	}
-
-	if err := p.inputPreRun(); err != nil {
-		return err
-	}
-
-	p.state = instanceStateInitialized
-	return nil
-}
-
 // FLBPluginInputResume this method gets invoked by the fluent-bit runtime, once the plugin has been
 // resumed, the plugin invoked this method and re-running state.
 //
@@ -276,32 +197,6 @@ func FLBPluginOutputPreExit() {
 	}
 }
 
-func (p *pluginInstance) outputPreExit() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	// Only output plugins have a pre-exit step
-	if p.meta.output == nil {
-		return fmt.Errorf("plugin is not an output plugin")
-	}
-
-	if p.state != instanceStateRunnable {
-		return fmt.Errorf("invalid plugin state %q", p.state)
-	}
-
-	p.runCancel()
-
-	// Wait for any running callback/flush to finish before closing the message channel
-	p.runningWG.Wait()
-	close(p.msgChannel)
-
-	p.state = instanceStatePreExit
-	p.runCtx = nil
-	p.runCancel = nil
-
-	return nil
-}
-
 // FLBPluginOutputPreRun -
 //
 //export FLBPluginOutputPreRun
@@ -324,40 +219,6 @@ func FLBPluginOutputPreRun(useHotReload C.int) int {
 	return input.FLB_OK
 }
 
-func (p *pluginInstance) outputPreRun() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
-	// 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() {
-		p.runningWG.Add(1)
-		defer p.runningWG.Done()
-		if err := p.meta.output.Flush(p.runCtx, p.msgChannel); err != nil {
-			fmt.Fprintf(os.Stderr, "FLBPluginOutputPreRun error: %v\n", err)
-		}
-	}()
-
-	go func() {
-		<-p.runCtx.Done()
-
-		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
 // initialized, 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
@@ -379,61 +240,6 @@ func FLBPluginInputCallback(data *unsafe.Pointer, csize *C.size_t) int {
 	return instance.callback(data, csize)
 }
 
-func (p *pluginInstance) callback(data *unsafe.Pointer, csize *C.size_t) int {
-	p.mu.Lock()
-	if p.state != instanceStateRunnable {
-		return input.FLB_RETRY
-	}
-
-	p.runningWG.Add(1)
-	defer p.runningWG.Done()
-	p.mu.Unlock()
-
-	buf := bytes.NewBuffer([]byte{})
-
-	for loop := min(len(p.msgChannel), p.maxBufferedMessages); loop > 0; loop-- {
-		select {
-		case msg, ok := <-p.msgChannel:
-			if !ok {
-				return input.FLB_ERROR
-			}
-
-			b, err := msgpack.Marshal([]any{&EventTime{msg.Time}, msg.Record})
-			if err != nil {
-				fmt.Fprintf(os.Stderr, "msgpack marshal: %s\n", err)
-				return input.FLB_ERROR
-			}
-
-			buf.Grow(len(b))
-			buf.Write(b)
-		case <-p.runCtx.Done():
-			err := p.runCtx.Err()
-			if err != nil && !errors.Is(err, context.Canceled) {
-				fmt.Fprintf(os.Stderr, "run: %s\n", err)
-				return input.FLB_ERROR
-			}
-			// enforce a runtime gc, to prevent the thread finalizer on
-			// fluent-bit to kick in before any remaining data has not been GC'ed
-			// causing a sigsegv.
-			defer runtime.GC()
-			loop = 0
-		default:
-			loop = 0
-		}
-	}
-
-	if buf.Len() > 0 {
-		b := buf.Bytes()
-		cdata := C.CBytes(b)
-		*data = cdata
-		if csize != nil {
-			*csize = C.size_t(len(b))
-		}
-	}
-
-	return input.FLB_OK
-}
-
 // FLBPluginInputCleanupCallback releases the memory used during the input callback
 //
 //export FLBPluginInputCleanupCallback

cshared_test.go

@@ -624,14 +624,3 @@ func TestOutputFlush(t *testing.T) {
 	assert.NoError(t, pluginFlush(inst, "foobar", b))
 	wg.Wait()
 }
-
-func assertType[T any](tb testing.TB, got any) T {
-	tb.Helper()
-
-	var want T
-
-	v, ok := got.(T)
-	assert.True(tb, ok, "Expected types to be equal:\n-%T\n+%T", want, got)
-
-	return v
-}