Method stopping

modules/livestore/live_store.go:446–511

(error)

}

func (s *LiveStore) stopping(error) error {
	level.Info(s.logger).Log("msg", "live store stopping", "remove_partition_owner", s.ingestPartitionLifecycler.RemoveOwnerOnShutdown())

	// Reject new queries early in shutdown, before tearing down the reader.
	s.readyErr.Store(&ErrStopping)
	metricReady.Set(0)

	var stopErr error

	if s.cfg.ConsumeFromKafka {
		// Stop the kafka lag background worker.
		if s.lagCancel != nil {
			s.lagCancel()
		}
		// Stop consuming.
		if err := services.StopAndAwaitTerminated(context.Background(), s.reader); err != nil {
			level.Warn(s.logger).Log("msg", "failed to stop reader", "err", err)
			stopErr = errors.Join(stopErr, err)
		}

		// Reset lag metrics for our partition when stopping.
		ingest.ResetLagMetricsForRevokedPartitions(s.cfg.IngestConfig.Kafka.ConsumerGroup, []int32{s.ingestPartitionID})
	}

	// Stop both the membership ring and partition ring even if an earlier shutdown step failed.
	if err := services.StopAndAwaitTerminated(context.Background(), s.livestoreLifecycler); err != nil {
		level.Warn(s.logger).Log("msg", "failed to stop livestore lifecycler", "err", err)
		stopErr = errors.Join(stopErr, err)
	}

	if err := services.StopAndAwaitTerminated(context.Background(), s.ingestPartitionLifecycler); err != nil {
		level.Warn(s.logger).Log("msg", "failed to stop partition lifecycler", "err", err)
		stopErr = errors.Join(stopErr, err)
	}

	level.Info(s.logger).Log("msg", "stopping periodic WAL flush goroutines")
	s.stopAllCutToWalLoops()
	level.Info(s.logger).Log("msg", "periodic WAL flush goroutines stopped")

	// Flush all data to disk.
	level.Info(s.logger).Log("msg", "cutting all instances to WAL")
	s.cutAllInstancesToWal()
	level.Info(s.logger).Log("msg", "done cutting all instances to WAL")

	// Remove the shutdown marker if it exists since we are shutting down.
	shutdownMarkerPath := shutdownmarker.GetPath(s.cfg.ShutdownMarkerDir)
	if err := shutdownmarker.Remove(shutdownMarkerPath); err != nil {
		level.Warn(s.logger).Log("msg", "failed to remove shutdown marker", "path", shutdownMarkerPath, "err", err)
		stopErr = errors.Join(stopErr, err)
	}

	if s.cfg.holdAllBackgroundProcesses { // nothing to do
		return stopErr
	}

	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()