feat(pkg/runner): change runner's interface to return error

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

Author: hainenber

component/loki/source/docker/runner.go

@@ -94,7 +94,7 @@ func newTailer(l log.Logger, task *tailerTask) *tailer {
 	}
 }
 
-func (t *tailer) Run(ctx context.Context) {
+func (t *tailer) Run(ctx context.Context) error {
 	ch, chErr := t.opts.client.ContainerWait(ctx, t.target.Name(), container.WaitConditionNextExit)
 
 	t.target.StartIfNotRunning()
@@ -108,10 +108,10 @@ func (t *tailer) Run(ctx context.Context) {
 		// refresh.
 		level.Error(t.log).Log("msg", "could not set up a wait request to the Docker client", "error", err)
 		t.target.Stop()
-		return
+		return err
 	case <-ch:
 		t.target.Stop()
-		return
+		return nil
 	}
 }
 

component/loki/source/kubernetes/kubetail/tailer.go

@@ -89,7 +89,7 @@ var retailBackoff = backoff.Config{
 	MaxBackoff: time.Minute,
 }
 
-func (t *tailer) Run(ctx context.Context) {
+func (t *tailer) Run(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -111,18 +111,22 @@ func (t *tailer) Run(ctx context.Context) {
 			terminated, err := t.containerTerminated(ctx)
 			if terminated {
 				// The container shut down and won't come back; we can stop tailing it.
-				return
+				return nil
 			} else if err != nil {
 				level.Warn(t.log).Log("msg", "could not determine if container terminated; will retry tailing", "err", err)
+				return err
 			}
 		}
 
 		if err != nil {
 			t.target.Report(time.Now().UTC(), err)
 			level.Warn(t.log).Log("msg", "tailer stopped; will retry", "err", err)
+			return err
 		}
 		bo.Wait()
 	}
+
+	return nil
 }
 
 func (t *tailer) tail(ctx context.Context, handler loki.EntryHandler) error {

component/loki/source/kubernetes_events/event_controller.go

@@ -79,15 +79,18 @@ func newEventController(task eventControllerTask) *eventController {
 	}
 }
 
-func (ctrl *eventController) Run(ctx context.Context) {
+func (ctrl *eventController) Run(ctx context.Context) error {
 	defer ctrl.handler.Stop()
 
 	level.Info(ctrl.log).Log("msg", "watching events for namespace", "namespace", ctrl.task.Namespace)
 	defer level.Info(ctrl.log).Log("msg", "stopping watcher for events", "namespace", ctrl.task.Namespace)
 
 	if err := ctrl.runError(ctx); err != nil {
 		level.Error(ctrl.log).Log("msg", "event watcher exited with error", "err", err)
+		return err
 	}
+
+	return nil
 }
 
 func (ctrl *eventController) runError(ctx context.Context) error {

pkg/runner/runner.go

@@ -32,7 +32,7 @@ type Worker interface {
 	// Run starts a Worker, blocking until the provided ctx is canceled or a
 	// fatal error occurs. Run is guaranteed to be called exactly once for any
 	// given Worker.
-	Run(ctx context.Context)
+	Run(ctx context.Context) error
 }
 
 // The Runner manages a set of running Workers based on an active set of tasks.

pkg/runner/runner_test.go

@@ -101,9 +101,11 @@ type genericWorker struct {
 
 var _ runner.Worker = (*genericWorker)(nil)
 
-func (w *genericWorker) Run(ctx context.Context) {
+func (w *genericWorker) Run(ctx context.Context) error {
 	w.workerCount.Inc()
 	defer w.workerCount.Dec()
 
 	<-ctx.Done()
+
+	return nil
 }