Method Push

pkg/ingester/ingester.go:1269–1685

Push adds metrics to a block

(ctx context.Context, req *cortexpb.WriteRequest)

// Push adds metrics to a block
func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) {
	if err := i.checkRunning(); err != nil {
		return nil, err
	}

	span, ctx := opentracing.StartSpanFromContext(ctx, "Ingester.Push")
	defer span.Finish()

	userID, err := users.TenantID(ctx)
	if err != nil {
		return nil, err
	}

	// We will report *this* request in the error too.
	inflight := i.inflightPushRequests.Inc()
	i.maxInflightPushRequests.Track(inflight)
	defer i.inflightPushRequests.Dec()

	gl := i.getInstanceLimits()
	if gl != nil && gl.MaxInflightPushRequests > 0 {
		if inflight > gl.MaxInflightPushRequests {
			i.metrics.pushErrorsTotal.WithLabelValues(userID, pushErrTooManyInflightRequests).Inc()
			return nil, errTooManyInflightPushRequests
		}
	}

	var firstPartialErr error

	// NOTE: because we use `unsafe` in deserialisation, we must not
	// retain anything from `req` past the call to ReuseSlice
	defer req.Free()
	defer cortexpb.ReuseSlice(req.Timeseries)

	il := i.getInstanceLimits()
	if il != nil && il.MaxIngestionRate > 0 {
		if rate := i.ingestionRate.Rate(); rate >= il.MaxIngestionRate {
			return nil, errMaxSamplesPushRateLimitReached
		}
	}

	db, err := i.getOrCreateTSDB(userID, false)
	if err != nil {
		return nil, wrapWithUser(err, userID)
	}

	// Ensure the ingester shutdown procedure hasn't started
	i.stoppedMtx.RLock()
	if i.stopped {
		i.stoppedMtx.RUnlock()
		return nil, errIngesterStopping
	}
	i.stoppedMtx.RUnlock()

	if err := db.acquireAppendLock(); err != nil {
		return &cortexpb.WriteResponse{}, httpgrpc.Errorf(http.StatusServiceUnavailable, "%s", wrapWithUser(err, userID).Error())
	}
	defer db.releaseAppendLock()