From fd2158e7464231accdc4690db823d2c7b512a016 Mon Sep 17 00:00:00 2001
From: Julius Volz <julius.volz@gmail.com>
Date: Sun, 2 Feb 2014 16:45:53 +0100
Subject: [PATCH] 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
---
 storage/metric/memory.go      |  9 ++++++--
 storage/metric/tiered_test.go | 39 +++++++++++++++++++++++++++++++++++
 2 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/storage/metric/memory.go b/storage/metric/memory.go
index 996283243c..1ea929dcf3 100644
--- a/storage/metric/memory.go
+++ b/storage/metric/memory.go
@@ -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 {
diff --git a/storage/metric/tiered_test.go b/storage/metric/tiered_test.go
index cd989e68ee..2c61dd2d2f 100644
--- a/storage/metric/tiered_test.go
+++ b/storage/metric/tiered_test.go
@@ -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])
+	}
+}