Repository: prometheus/prometheus
File: storage/remote/queue_manager.go (Go)

Code before:
	if errors.Is(err, context.Canceled) {
		// When there is resharding, we cancel the context for this queue, which means the data is not sent.
		// So we exit early to not update the metrics.
		return accumulatedStats, err
	}

	s.qm.metrics.sentBytesTotal.Add(float64(reqSize))
	s.qm.metrics.highestSentTimestamp.Set(float64(highest / 1000))
	return accumulatedStats, err
}

func populateV2TimeSeries(symbolTable *writev2.SymbolsTable, batch []timeSeries, pendingData []writev2.TimeSeries, sendExemplars, sendNativeHistograms, enableTypeAndUnitLabels bool) (int, int, int, int) {
	var nPendingSamples, nPendingExemplars, nPendingHistograms, nPendingMetadata int
	for nPending, d := range batch {
		pendingData[nPending].Samples = pendingData[nPending].Samples[:0]
		if d.metadata != nil {
			pendingData[nPending].Metadata.Type = writev2.FromMetadataType(d.metadata.Type)
			pendingData[nPending].Metadata.HelpRef = symbolTable.Symbolize(d.metadata.Help)
			pendingData[nPending].Metadata.UnitRef = symbolTable.Symbolize(d.metadata.Unit)
			nPendingMetadata++
		} else {
			metricType := writev2.Metadata_METRIC_TYPE_UNSPECIFIED
			var unit string
			if enableTypeAndUnitLabels {
				typeValue := d.seriesLabels.Get("__type__")
				unit = d.seriesLabels.Get("__unit__")
				// convert string type to MetricMetadata_MetricType
				switch typeValue {
				case "counter":
					metricType = writev2.Metadata_METRIC_TYPE_COUNTER
				case "gauge":
					metricType = writev2.Metadata_METRIC_TYPE_GAUGE
				case "histogram":
					metricType = writev2.Metadata_METRIC_TYPE_HISTOGRAM
				case "summary":
					metricType = writev2.Metadata_METRIC_TYPE_SUMMARY
				case "info":
					metricType = writev2.Metadata_METRIC_TYPE_INFO
				case "stateset":
					metricType = writev2.Metadata_METRIC_TYPE_STATESET
				default:
					metricType = writev2.Metadata_METRIC_TYPE_UNSPECIFIED
				}

			}
			// Safeguard against sending garbage in case of not having metadata
			// for whatever reason.
			pendingData[nPending].Metadata.Type = metricType
			pendingData[nPending].Metadata.HelpRef = 0
			pendingData[nPending].Metadata.UnitRef = symbolTable.Symbolize(unit)
		}

Review comment (bug):
I could be wrong here, but I'm assuming that I can access this labels, because @ArthurSens recently added this line ⬇️, where when the feature flag is enabled, the labels __type__ and __unit__ will be populated.

https://github.com/prometheus/prometheus/blob/294f36e80261bb2e2364aa3d0700ebaf20a17bca/storage/remote/otlptranslator/prometheusremotewrite/helper.go#L206-L214
 



Code after:
	err = s.qm.sendWriteRequestWithBackoff(ctx, attemptStore, onRetry)
	if errors.Is(err, context.Canceled) {
		// When there is resharding, we cancel the context for this queue, which means the data is not sent.
		// So we exit early to not update the metrics.
		return accumulatedStats, err
	}

	s.qm.metrics.sentBytesTotal.Add(float64(reqSize))
	s.qm.metrics.highestSentTimestamp.Set(float64(highest / 1000))
	return accumulatedStats, err
}

func populateV2TimeSeries(symbolTable *writev2.SymbolsTable, batch []timeSeries, pendingData []writev2.TimeSeries, sendExemplars, sendNativeHistograms, enableTypeAndUnitLabels bool) (int, int, int, int, int) {
	var nPendingSamples, nPendingExemplars, nPendingHistograms, nPendingMetadata, nUnexpectedMetadata int
	for nPending, d := range batch {
		pendingData[nPending].Samples = pendingData[nPending].Samples[:0]
		if d.metadata != nil {
			pendingData[nPending].Metadata.Type = writev2.FromMetadataType(d.metadata.Type)
			pendingData[nPending].Metadata.HelpRef = symbolTable.Symbolize(d.metadata.Help)
			pendingData[nPending].Metadata.UnitRef = symbolTable.Symbolize(d.metadata.Unit)
			nPendingMetadata++
		} else {
			var m schema.Metadata
			if enableTypeAndUnitLabels {
				m = schema.NewMetadataFromLabels(d.seriesLabels)
			}
			// Safeguard against sending garbage in case of not having metadata
			// for whatever reason.
			pendingData[nPending].Metadata.Type = writev2.FromMetadataType(m.Type)
			pendingData[nPending].Metadata.UnitRef = symbolTable.Symbolize(m.Unit)
			pendingData[nPending].Metadata.HelpRef = 0
		}

		if sendExemplars {
			pendingData[nPending].Exemplars = pendingData[nPending].Exemplars[:0]
		}
		if sendNativeHistograms {
			pendingData[nPending].Histograms = pendingData[nPending].Histograms[:0]
		}

		// Number of pending samples is limited by the fact that sendSamples (via sendSamplesWithBackoff)
		// retries endlessly, so once we reach max samples, if we can never send to the endpoint we'll
		// stop reading from the queue. This makes it safe to reference pendingSamples by index.
		pendingData[nPending].LabelsRefs = symbolTable.SymbolizeLabels(d.seriesLabels, pendingData[nPending].LabelsRefs)
		switch d.sType {
		case tSample:
			pendingData[nPending].Samples = append(pendingData[nPending].Samples, writev2.Sample{
				Value:     d.value,
				Timestamp: d.timestamp,
			})
			nPendingSamples++