Method starting

modules/backendworker/backendworker.go:129–181

(ctx context.Context)

}

func (w *BackendWorker) starting(ctx context.Context) (err error) {
	defer func() {
		if err == nil || w.subservices == nil {
			return
		}

		if stopErr := services.StopManagerAndAwaitStopped(context.Background(), w.subservices); stopErr != nil {
			level.Error(log.Logger).Log("msg", "failed to gracefully stop backend-worker dependencies", "err", stopErr)
		}
	}()

	if w.isSharded() {
		w.subservices, err = services.NewManager(w.ringLifecycler, w.Ring)
		if err != nil {
			return fmt.Errorf("failed to create subservices: %w", err)
		}
		w.subservicesWatcher = services.NewFailureWatcher()
		w.subservicesWatcher.WatchManager(w.subservices)

		err := services.StartManagerAndAwaitHealthy(ctx, w.subservices)
		if err != nil {
			return fmt.Errorf("failed to start subservices: %w", err)
		}

		// Wait until the ring client detected this instance in the ACTIVE state.
		level.Info(log.Logger).Log("msg", "waiting until backend-worker is ACTIVE in the ring")
		ctxWithTimeout, cancel := context.WithTimeout(ctx, w.cfg.Ring.WaitActiveInstanceTimeout)
		defer cancel()
		if err := ring.WaitInstanceState(ctxWithTimeout, w.Ring, w.ringLifecycler.GetInstanceID(), ring.ACTIVE); err != nil {
			return err
		}
		level.Info(log.Logger).Log("msg", "backend-worker is ACTIVE in the ring")

		// In the event of a cluster cold start we may end up in a situation where each new backend-worker
		// instance starts at a slightly different time and thus each one starts with a different state
		// of the ring. It's better to just wait the ring stability for a short time.
		if w.cfg.Ring.WaitStabilityMinDuration > 0 {
			minWaiting := w.cfg.Ring.WaitStabilityMinDuration
			maxWaiting := w.cfg.Ring.WaitStabilityMaxDuration

			level.Info(log.Logger).Log("msg", "waiting until backend-worker ring topology is stable", "min_waiting", minWaiting.String(), "max_waiting", maxWaiting.String())
			if err := ring.WaitRingStability(ctx, w.Ring, ringOp, minWaiting, maxWaiting); err != nil {
				level.Warn(log.Logger).Log("msg", "backend-worker ring topology is not stable after the max waiting time, proceeding anyway")
			} else {
				level.Info(log.Logger).Log("msg", "backend-worker ring topology is stable")
			}
		}
	}

	w.store.EnablePolling(ctx, w, false)

	return nil
}

func (w *BackendWorker) running(ctx context.Context) error {
	level.Info(log.Logger).Log("msg", "backend worker running")