Method sendSamplesWithBackoff

storage/remote/queue_manager.go:1750–1852

sendSamplesWithBackoff to the remote storage with backoff for recoverable errors.

(ctx context.Context, samples []prompb.TimeSeries, sampleCount, exemplarCount, histogramCount, metadataCount int, pBuf *proto.Buffer, buf compression.EncodeBuffer, compr compression.Type)

// sendSamplesWithBackoff to the remote storage with backoff for recoverable errors.
func (s *shards) sendSamplesWithBackoff(ctx context.Context, samples []prompb.TimeSeries, sampleCount, exemplarCount, histogramCount, metadataCount int, pBuf *proto.Buffer, buf compression.EncodeBuffer, compr compression.Type) (WriteResponseStats, error) {
	// Build the WriteRequest with no metadata.
	req, highest, lowest, err := buildWriteRequest(s.qm.logger, samples, nil, pBuf, nil, buf, compr)
	s.qm.buildRequestLimitTimestamp.Store(lowest)
	if err != nil {
		// Failing to build the write request is non-recoverable, since it will
		// only error if marshaling the proto to bytes fails.
		return WriteResponseStats{}, err
	}

	reqSize := len(req)
	sc := sendBatchContext{
		ctx:            ctx,
		sampleCount:    sampleCount,
		exemplarCount:  exemplarCount,
		histogramCount: histogramCount,
		metadataCount:  metadataCount,
		reqSize:        reqSize,
	}

	metricsUpdater := batchMetricsUpdater{
		metrics: s.qm.metrics,
	}

	// Since we retry writes via attemptStore and sendWriteRequestWithBackoff we need
	// to track the total amount of accepted data across the various attempts.
	accumulatedStats := WriteResponseStats{}
	var accumulatedStatsMu sync.Mutex
	addStats := func(rs WriteResponseStats) {
		accumulatedStatsMu.Lock()
		accumulatedStats = accumulatedStats.Add(rs)
		accumulatedStatsMu.Unlock()
	}

	// An anonymous function allows us to defer the completion of our per-try spans
	// without causing a memory leak, and it has the nice effect of not propagating any
	// parameters for sendSamplesWithBackoff/3.
	attemptStore := func(try int) error {
		currentTime := time.Now()
		lowest := s.qm.buildRequestLimitTimestamp.Load()
		if isSampleOld(currentTime, time.Duration(s.qm.cfg.SampleAgeLimit), lowest) {
			// This will filter out old samples during retries.
			req2, _, lowest, err := buildWriteRequest(
				s.qm.logger,
				samples,
				nil,
				pBuf,
				isTimeSeriesOldFilter(s.qm.metrics, currentTime, time.Duration(s.qm.cfg.SampleAgeLimit)),
				buf,
				compr,
			)
			s.qm.buildRequestLimitTimestamp.Store(lowest)
			if err != nil {
				return err
			}
			req = req2
		}