remove error channel, handle upload error within immediate upload loop.

Signed-off-by: sspaink <[email protected]>

Author: sspaink

Diff for: v1/plugins/logs/plugin.go

@@ -836,26 +836,23 @@ func (p *Plugin) compilerUpdated(storage.Transaction) {
 func (p *Plugin) loop() {
 	ctx, cancel := context.WithCancel(context.Background())
 
-	var errCh chan error
-	var stop chan chan struct{}
+	var stopImmediateLoop chan chan struct{}
 	if *p.config.Reporting.Trigger == plugins.TriggerImmediate && p.config.Service != "" {
-		errCh = make(chan error)
-		stop = make(chan chan struct{})
-
+		stopImmediateLoop = make(chan chan struct{})
 		switch p.config.Reporting.BufferType {
 		case sizeBufferType:
 			go immediateUpload[[][]byte](ctx,
-				errCh,
-				stop,
+				stopImmediateLoop,
 				p.config.Reporting.MaxDelaySeconds,
 				p,
+				p,
 			)
 		case eventBufferType:
 			go immediateUpload[*bufferItem](ctx,
-				errCh,
-				stop,
+				stopImmediateLoop,
 				p.config.Reporting.MaxDelaySeconds,
 				p.eventBuffer,
+				p,
 			)
 		}
 	}
@@ -898,16 +895,14 @@ func (p *Plugin) loop() {
 		}
 
 		select {
-		case err := <-errCh:
-			p.handleUploadError(err)
 		case <-waitC:
 		case update := <-p.reconfig:
 			p.reconfigure(ctx, update.config)
 			update.done <- struct{}{}
 		case done := <-p.stop:
 			if *p.config.Reporting.Trigger == plugins.TriggerImmediate {
 				d := make(chan struct{})
-				stop <- d
+				stopImmediateLoop <- d
 				<-d
 			}
 			cancel()
@@ -1303,7 +1298,7 @@ type buffer[T any] interface {
 }
 
 // immediateUpload continuously waits to receive a chunk or event and then attempts an upload
-func immediateUpload[T any](ctx context.Context, errCh chan error, stop chan chan struct{}, maxDelay *int64, b buffer[T]) {
+func immediateUpload[T any](ctx context.Context, stopImmediateLoop chan chan struct{}, maxDelay *int64, b buffer[T], p *Plugin) {
 	timer := time.NewTimer(time.Duration(*maxDelay))
 
 	for {
@@ -1315,18 +1310,14 @@ func immediateUpload[T any](ctx context.Context, errCh chan error, stop chan cha
 			chunks = b.Handle(e)
 		case <-timer.C:
 			chunks = b.Flush()
-		case d := <-stop:
-			d <- struct{}{}
+		case done := <-stopImmediateLoop:
+			done <- struct{}{}
 			return
 		}
 
 		err := b.UploadChunks(ctx, chunks)
 		if err != nil {
-			// this is consumed by p.loop to set the plugin status
-			errCh <- err
-			// delay briefly to prevent overloading the service
-			time.Sleep(minRetryDelay)
-
+			p.handleUploadError(err)
 			continue
 		}
 

Diff for: v1/plugins/logs/plugin_test.go

@@ -3782,6 +3782,8 @@ func TestPluginTriggerImmediateErrorHandling(t *testing.T) {
 				t.Fatal(err)
 			}
 
+			fixture.plugin.Stop(ctx)
+
 			// wait until there is an error entry
 			var errReceived string
 			for {