Store copy of metric during fingerprint caching

Problem description:
====================
If a rule evaluation referencing a metric/timeseries M happens at a time
when M doesn't have a memory timeseries yet, looking up the fingerprint
for M (via TieredStorage.GetMetricForFingerprint()) will create a new
Metric object for M which gets both: a) attached to a new empty memory
timeseries (so we don't have to ask disk for the Metric's fingerprint
next time), and b) returned to the rule evaluation layer. However, the
rule evaluation layer replaces the name label (and possibly other
labels) of the metric with the name of the recorded rule.  Since both
the rule evaluator and the memory storage share a reference to the same
Metric object, the original memory timeseries will now also be
incorrectly renamed.

Fix:
====
Instead of storing a reference to a shared metric object, take a copy of
the object when creating an empty memory timeseries for caching
purposes.

Change-Id: I9f2172696c16c10b377e6708553a46ef29390f1e
This commit is contained in:
Julius Volz 2014-02-02 16:45:53 +01:00
parent 7e9ecaac3a
commit fd2158e746
2 changed files with 46 additions and 2 deletions

View File

@ -231,9 +231,14 @@ func (s *memorySeriesStorage) CreateEmptySeries(metric clientmodel.Metric) {
s.Lock()
defer s.Unlock()
m := clientmodel.Metric{}
for label, value := range metric {
m[label] = value
}
fingerprint := &clientmodel.Fingerprint{}
fingerprint.LoadFromMetric(metric)
s.getOrCreateSeries(metric, fingerprint)
fingerprint.LoadFromMetric(m)
s.getOrCreateSeries(m, fingerprint)
}
func (s *memorySeriesStorage) getOrCreateSeries(metric clientmodel.Metric, fingerprint *clientmodel.Fingerprint) stream {

View File

@ -729,3 +729,42 @@ func testTruncateBefore(t test.Tester) {
func TestTruncateBefore(t *testing.T) {
testTruncateBefore(t)
}
func TestGetMetricForFingerprintCachesCopyOfMetric(t *testing.T) {
ts, closer := NewTestTieredStorage(t)
defer closer.Close()
m := clientmodel.Metric{
clientmodel.MetricNameLabel: "testmetric",
}
samples := clientmodel.Samples{
&clientmodel.Sample{
Metric: m,
Value: 0,
Timestamp: clientmodel.Now(),
},
}
if err := ts.AppendSamples(samples); err != nil {
t.Fatalf(err.Error())
}
ts.Flush()
fp := &clientmodel.Fingerprint{}
fp.LoadFromMetric(m)
m, err := ts.GetMetricForFingerprint(fp)
if err != nil {
t.Fatalf(err.Error())
}
m[clientmodel.MetricNameLabel] = "changedmetric"
m, err = ts.GetMetricForFingerprint(fp)
if err != nil {
t.Fatalf(err.Error())
}
if m[clientmodel.MetricNameLabel] != "testmetric" {
t.Fatal("Metric name label value has changed: ", m[clientmodel.MetricNameLabel])
}
}