Method flushHandler

pkg/ingester/ingester.go:3589–3661

Blocks version of Flush handler. It force-compacts blocks, and triggers shipping.

(w http.ResponseWriter, r *http.Request)

// Blocks version of Flush handler. It force-compacts blocks, and triggers shipping.
func (i *Ingester) flushHandler(w http.ResponseWriter, r *http.Request) {
	err := r.ParseForm()
	if err != nil {
		level.Warn(logutil.WithContext(r.Context(), i.logger)).Log("msg", "failed to parse HTTP request in flush handler", "err", err)
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	tenants := r.Form[tenantParam]

	allowedUsers := users.NewAllowedTenants(tenants, nil)
	run := func() {
		ingCtx := i.ServiceContext()
		if ingCtx == nil || ingCtx.Err() != nil {
			level.Info(logutil.WithContext(r.Context(), i.logger)).Log("msg", "flushing TSDB blocks: ingester not running, ignoring flush request")
			return
		}

		compactionCallbackCh := make(chan struct{})

		level.Info(logutil.WithContext(r.Context(), i.logger)).Log("msg", "flushing TSDB blocks: triggering compaction")
		select {
		case i.TSDBState.forceCompactTrigger <- requestWithUsersAndCallback{users: allowedUsers, callback: compactionCallbackCh}:
			// Compacting now.
		case <-ingCtx.Done():
			level.Warn(logutil.WithContext(r.Context(), i.logger)).Log("msg", "failed to compact TSDB blocks, ingester not running anymore")
			return
		}

		// Wait until notified about compaction being finished.
		select {
		case <-compactionCallbackCh:
			level.Info(logutil.WithContext(r.Context(), i.logger)).Log("msg", "finished compacting TSDB blocks")
		case <-ingCtx.Done():
			level.Warn(logutil.WithContext(r.Context(), i.logger)).Log("msg", "failed to compact TSDB blocks, ingester not running anymore")
			return
		}

		if i.cfg.BlocksStorageConfig.TSDB.IsBlocksShippingEnabled() {
			shippingCallbackCh := make(chan struct{}) // must be new channel, as compactionCallbackCh is closed now.

			level.Info(logutil.WithContext(r.Context(), i.logger)).Log("msg", "flushing TSDB blocks: triggering shipping")

			select {
			case i.TSDBState.shipTrigger <- requestWithUsersAndCallback{users: allowedUsers, callback: shippingCallbackCh}:
				// shipping now
			case <-ingCtx.Done():
				level.Warn(logutil.WithContext(r.Context(), i.logger)).Log("msg", "failed to ship TSDB blocks, ingester not running anymore")
				return
			}

			// Wait until shipping finished.
			select {
			case <-shippingCallbackCh:
				level.Info(logutil.WithContext(r.Context(), i.logger)).Log("msg", "shipping of TSDB blocks finished")
			case <-ingCtx.Done():
				level.Warn(logutil.WithContext(r.Context(), i.logger)).Log("msg", "failed to ship TSDB blocks, ingester not running anymore")
				return