(batch []timeSeries, pendingData []prompb.TimeSeries, sendExemplars, sendNativeHistograms bool)
| 1654 | } |
| 1655 | |
| 1656 | func populateTimeSeries(batch []timeSeries, pendingData []prompb.TimeSeries, sendExemplars, sendNativeHistograms bool) (int, int, int) { |
| 1657 | var nPendingSamples, nPendingExemplars, nPendingHistograms int |
| 1658 | for nPending, d := range batch { |
| 1659 | pendingData[nPending].Samples = pendingData[nPending].Samples[:0] |
| 1660 | if sendExemplars { |
| 1661 | pendingData[nPending].Exemplars = pendingData[nPending].Exemplars[:0] |
| 1662 | } |
| 1663 | if sendNativeHistograms { |
| 1664 | pendingData[nPending].Histograms = pendingData[nPending].Histograms[:0] |
| 1665 | } |
| 1666 | |
| 1667 | // Number of pending samples is limited by the fact that sendSamples (via sendSamplesWithBackoff) |
| 1668 | // retries endlessly, so once we reach max samples, if we can never send to the endpoint we'll |
| 1669 | // stop reading from the queue. This makes it safe to reference pendingSamples by index. |
| 1670 | pendingData[nPending].Labels = prompb.FromLabels(d.seriesLabels, pendingData[nPending].Labels) |
| 1671 | |
| 1672 | switch d.sType { |
| 1673 | case tSample: |
| 1674 | pendingData[nPending].Samples = append(pendingData[nPending].Samples, prompb.Sample{ |
| 1675 | Value: d.value, |
| 1676 | Timestamp: d.timestamp, |
| 1677 | }) |
| 1678 | nPendingSamples++ |
| 1679 | case tExemplar: |
| 1680 | pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.Exemplar{ |
| 1681 | Labels: prompb.FromLabels(d.exemplarLabels, nil), |
| 1682 | Value: d.value, |
| 1683 | Timestamp: d.timestamp, |
| 1684 | }) |
| 1685 | nPendingExemplars++ |
| 1686 | case tHistogram: |
| 1687 | pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, prompb.FromIntHistogram(d.timestamp, d.histogram)) |
| 1688 | nPendingHistograms++ |
| 1689 | case tFloatHistogram: |
| 1690 | pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, prompb.FromFloatHistogram(d.timestamp, d.floatHistogram)) |
| 1691 | nPendingHistograms++ |
| 1692 | } |
| 1693 | } |
| 1694 | return nPendingSamples, nPendingExemplars, nPendingHistograms |
| 1695 | } |
| 1696 | |
| 1697 | func (s *shards) sendSamples(ctx context.Context, samples []prompb.TimeSeries, sampleCount, exemplarCount, histogramCount int, pBuf *proto.Buffer, buf compression.EncodeBuffer, compr compression.Type) error { |
| 1698 | begin := time.Now() |
searching dependent graphs…