Method processWriteRequest

storage/diskmetricstore.go:268–309

(wr WriteRequest)

}

func (dms *DiskMetricStore) processWriteRequest(wr WriteRequest) {
	dms.lock.Lock()
	defer dms.lock.Unlock()

	key := groupingKeyFor(wr.Labels)

	if wr.MetricFamilies == nil {
		// No MetricFamilies means delete request. Delete the whole
		// metric group, and we are done here.
		delete(dms.metricGroups, key)
		return
	}
	// Otherwise, it's an update.
	group, ok := dms.metricGroups[key]
	if !ok {
		group = MetricGroup{
			Labels:  wr.Labels,
			Metrics: NameToTimestampedMetricFamilyMap{},
		}
		dms.metricGroups[key] = group
	} else if wr.Replace {
		// For replace, we have to delete all metric families in the
		// group except pre-existing push timestamps.
		for name := range group.Metrics {
			if name != pushMetricName && name != pushFailedMetricName {
				delete(group.Metrics, name)
			}
		}
	}
	wr.MetricFamilies[pushMetricName] = newPushTimestampGauge(wr.Labels, wr.Timestamp)
	// Only add a zero push-failed metric if none is there yet, so that a
	// previously added fail timestamp is retained.
	if _, ok := group.Metrics[pushFailedMetricName]; !ok {
		wr.MetricFamilies[pushFailedMetricName] = newPushFailedTimestampGauge(wr.Labels, time.Time{})
	}
	for name, mf := range wr.MetricFamilies {
		group.Metrics[name] = TimestampedMetricFamily{
			Timestamp:            wr.Timestamp,
			GobbableMetricFamily: (*GobbableMetricFamily)(mf),
		}
	}
}

func (dms *DiskMetricStore) setPushFailedTimestamp(wr WriteRequest) {
	dms.lock.Lock()