Method AppendWatcherMetadata

storage/remote/queue_manager.go:562–589

AppendWatcherMetadata sends metadata to the remote storage. Metadata is sent in batches, but is not parallelized. This is only used for the metadata_config.send setting and 1.x Remote Write.

(ctx context.Context, metadata []scrape.MetricMetadata)

// AppendWatcherMetadata sends metadata to the remote storage. Metadata is sent in batches, but is not parallelized.
// This is only used for the metadata_config.send setting and 1.x Remote Write.
func (t *QueueManager) AppendWatcherMetadata(ctx context.Context, metadata []scrape.MetricMetadata) {
	// no op for any newer proto format, which will cache metadata sent to it from the WAL watcher.
	if t.protoMsg != remoteapi.WriteV1MessageType {
		return
	}

	// 1.X will still get metadata in batches.
	mm := make([]prompb.MetricMetadata, 0, len(metadata))
	for _, entry := range metadata {
		mm = append(mm, prompb.MetricMetadata{
			MetricFamilyName: entry.MetricFamily,
			Help:             entry.Help,
			Type:             prompb.FromMetadataType(entry.Type),
			Unit:             entry.Unit,
		})
	}

	pBuf := proto.NewBuffer(nil)
	numSends := int(math.Ceil(float64(len(metadata)) / float64(t.mcfg.MaxSamplesPerSend)))
	for i := range numSends {
		last := min((i+1)*t.mcfg.MaxSamplesPerSend, len(metadata))
		err := t.sendMetadataWithBackoff(ctx, mm[i*t.mcfg.MaxSamplesPerSend:last], pBuf)
		if err != nil {
			t.metrics.failedMetadataTotal.Add(float64(last - (i * t.mcfg.MaxSamplesPerSend)))
			t.logger.Error("non-recoverable error while sending metadata", "count", last-(i*t.mcfg.MaxSamplesPerSend), "err", err)
		}
	}
}

func (t *QueueManager) sendMetadataWithBackoff(ctx context.Context, metadata []prompb.MetricMetadata, pBuf *proto.Buffer) error {
	// Build the WriteRequest with no samples (v1 flow).