From a3bf2efdd5ecfc7a81de1fe19385fefa8e4a325d Mon Sep 17 00:00:00 2001
From: "Matt T. Proud" <matt.proud@gmail.com>
Date: Wed, 7 Aug 2013 12:07:35 +0200
Subject: [PATCH 1/2] Replace index writes with wrapped interface.

This commit is the first of several and should not be regarded as the
desired end state for these cleanups.  What this one does it, however,
is wrap the query index writing behind an interface type that can be
injected into the storage stack and have its lifecycle managed
separately as needed.  It also would mean we can swap out underlying
implementations to support remote indexing, buffering, no-op indexing
very easily.

In the future, most of the individual index interface members in the
tiered storage will go away in favor of agents that can query and
resolve what they need from the datastore without the user knowing
how and why they work.
---
 storage/metric/index.go   | 187 +++++++++++++++++++++++++++++---
 storage/metric/leveldb.go | 219 +++++---------------------------------
 storage/metric/tiered.go  |   2 +
 3 files changed, 197 insertions(+), 211 deletions(-)

diff --git a/storage/metric/index.go b/storage/metric/index.go
index 89f5d9734..0ae2ca443 100644
--- a/storage/metric/index.go
+++ b/storage/metric/index.go
@@ -25,6 +25,7 @@ import (
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/storage/raw"
 	"github.com/prometheus/prometheus/storage/raw/leveldb"
+	"github.com/prometheus/prometheus/utility"
 )
 
 type FingerprintMetricMapping map[clientmodel.Fingerprint]clientmodel.Metric
@@ -211,20 +212,20 @@ func NewLevelLabelNameFingerprintIndex(o *LevelDBLabelNameFingerprintIndexOption
 	}, nil
 }
 
-type LabelSetFingerprintMapping map[LabelPair]clientmodel.Fingerprints
+type LabelPairFingerprintMapping map[LabelPair]clientmodel.Fingerprints
 
-type LabelSetFingerprintIndex interface {
+type LabelPairFingerprintIndex interface {
 	raw.ForEacher
 	raw.Pruner
 
-	IndexBatch(LabelSetFingerprintMapping) error
+	IndexBatch(LabelPairFingerprintMapping) error
 	Lookup(*LabelPair) (m clientmodel.Fingerprints, ok bool, err error)
 	Has(*LabelPair) (ok bool, err error)
 	State() *raw.DatabaseState
 	Size() (s uint64, present bool, err error)
 }
 
-type LevelDBLabelSetFingerprintIndex struct {
+type LevelDBLabelPairFingerprintIndex struct {
 	p *leveldb.LevelDBPersistence
 }
 
@@ -232,7 +233,7 @@ type LevelDBLabelSetFingerprintIndexOptions struct {
 	leveldb.LevelDBOptions
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) IndexBatch(m LabelSetFingerprintMapping) error {
+func (i *LevelDBLabelPairFingerprintIndex) IndexBatch(m LabelPairFingerprintMapping) error {
 	batch := leveldb.NewBatch()
 	defer batch.Close()
 
@@ -256,7 +257,7 @@ func (i *LevelDBLabelSetFingerprintIndex) IndexBatch(m LabelSetFingerprintMappin
 	return i.p.Commit(batch)
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) Lookup(p *LabelPair) (m clientmodel.Fingerprints, ok bool, err error) {
+func (i *LevelDBLabelPairFingerprintIndex) Lookup(p *LabelPair) (m clientmodel.Fingerprints, ok bool, err error) {
 	k := &dto.LabelPair{
 		Name:  proto.String(string(p.Name)),
 		Value: proto.String(string(p.Value)),
@@ -281,7 +282,7 @@ func (i *LevelDBLabelSetFingerprintIndex) Lookup(p *LabelPair) (m clientmodel.Fi
 	return m, true, nil
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) Has(p *LabelPair) (ok bool, err error) {
+func (i *LevelDBLabelPairFingerprintIndex) Has(p *LabelPair) (ok bool, err error) {
 	k := &dto.LabelPair{
 		Name:  proto.String(string(p.Name)),
 		Value: proto.String(string(p.Value)),
@@ -290,35 +291,35 @@ func (i *LevelDBLabelSetFingerprintIndex) Has(p *LabelPair) (ok bool, err error)
 	return i.p.Has(k)
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) ForEach(d storage.RecordDecoder, f storage.RecordFilter, o storage.RecordOperator) (bool, error) {
+func (i *LevelDBLabelPairFingerprintIndex) ForEach(d storage.RecordDecoder, f storage.RecordFilter, o storage.RecordOperator) (bool, error) {
 	return i.p.ForEach(d, f, o)
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) Prune() (bool, error) {
+func (i *LevelDBLabelPairFingerprintIndex) Prune() (bool, error) {
 	i.p.Prune()
 	return false, nil
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) Close() {
+func (i *LevelDBLabelPairFingerprintIndex) Close() {
 	i.p.Close()
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) Size() (uint64, bool, error) {
+func (i *LevelDBLabelPairFingerprintIndex) Size() (uint64, bool, error) {
 	s, err := i.p.Size()
 	return s, true, err
 }
 
-func (i *LevelDBLabelSetFingerprintIndex) State() *raw.DatabaseState {
+func (i *LevelDBLabelPairFingerprintIndex) State() *raw.DatabaseState {
 	return i.p.State()
 }
 
-func NewLevelDBLabelSetFingerprintIndex(o *LevelDBLabelSetFingerprintIndexOptions) (*LevelDBLabelSetFingerprintIndex, error) {
+func NewLevelDBLabelSetFingerprintIndex(o *LevelDBLabelSetFingerprintIndexOptions) (*LevelDBLabelPairFingerprintIndex, error) {
 	s, err := leveldb.NewLevelDBPersistence(&o.LevelDBOptions)
 	if err != nil {
 		return nil, err
 	}
 
-	return &LevelDBLabelSetFingerprintIndex{
+	return &LevelDBLabelPairFingerprintIndex{
 		p: s,
 	}, nil
 }
@@ -326,7 +327,7 @@ func NewLevelDBLabelSetFingerprintIndex(o *LevelDBLabelSetFingerprintIndexOption
 type MetricMembershipIndex interface {
 	raw.Pruner
 
-	IndexBatch([]clientmodel.Metric) error
+	IndexBatch(FingerprintMetricMapping) error
 	Has(clientmodel.Metric) (ok bool, err error)
 	State() *raw.DatabaseState
 	Size() (s uint64, present bool, err error)
@@ -338,11 +339,11 @@ type LevelDBMetricMembershipIndex struct {
 
 var existenceIdentity = new(dto.MembershipIndexValue)
 
-func (i *LevelDBMetricMembershipIndex) IndexBatch(ms []clientmodel.Metric) error {
+func (i *LevelDBMetricMembershipIndex) IndexBatch(b FingerprintMetricMapping) error {
 	batch := leveldb.NewBatch()
 	defer batch.Close()
 
-	for _, m := range ms {
+	for _, m := range b {
 		k := new(dto.Metric)
 		dumpMetric(k, m)
 		batch.Put(k, existenceIdentity)
@@ -391,3 +392,155 @@ func NewLevelDBMetricMembershipIndex(o *LevelDBMetricMembershipIndexOptions) (*L
 		p: s,
 	}, nil
 }
+
+// MetricIndexer indexes facets of a clientmodel.Metric.
+type MetricIndexer interface {
+	// IndexMetric makes no assumptions about the concurrency safety of the
+	// underlying implementer.
+	IndexMetrics(FingerprintMetricMapping) error
+}
+
+// TotalIndexer is a MetricIndexer that indexes all standard facets of a metric
+// that a user or the Prometheus subsystem would want to query against:
+//
+//    "<Label Name>" -> {Fingerprint, ...}
+//    "<Label Name> <Label Pair>" -> {Fingerprint, ...}
+//
+//    "<Fingerprint>" -> Metric
+//
+//    "<Metric>" -> Existence Value
+//
+//  This type supports concrete queries but only single writes, and it has no
+// locking semantics to enforce this.
+type TotalIndexer struct {
+	FingerprintToMetric    FingerprintMetricIndex
+	LabelNameToFingerprint LabelNameFingerprintIndex
+	LabelPairToFingerprint LabelPairFingerprintIndex
+	MetricMembership       MetricMembershipIndex
+}
+
+func findUnindexed(i MetricMembershipIndex, b FingerprintMetricMapping) (FingerprintMetricMapping, error) {
+	out := FingerprintMetricMapping{}
+
+	for fp, m := range b {
+		has, err := i.Has(m)
+		if err != nil {
+			return nil, err
+		}
+		if !has {
+			out[fp] = m
+		}
+	}
+
+	return out, nil
+}
+
+func extendLabelNameIndex(i LabelNameFingerprintIndex, b FingerprintMetricMapping) (LabelNameFingerprintMapping, error) {
+	collection := map[clientmodel.LabelName]utility.Set{}
+
+	for fp, m := range b {
+		for l := range m {
+			set, ok := collection[l]
+			if !ok {
+				baseFps, _, err := i.Lookup(l)
+				if err != nil {
+					return nil, err
+				}
+
+				set = utility.Set{}
+
+				for _, baseFp := range baseFps {
+					set.Add(*baseFp)
+				}
+
+				collection[l] = set
+			}
+
+			set.Add(fp)
+		}
+	}
+
+	batch := LabelNameFingerprintMapping{}
+	for l, set := range collection {
+		fps := clientmodel.Fingerprints{}
+		for e := range set {
+			fp := e.(clientmodel.Fingerprint)
+			fps = append(fps, &fp)
+		}
+
+		batch[l] = fps
+	}
+
+	return batch, nil
+}
+
+func extendLabelPairIndex(i LabelPairFingerprintIndex, b FingerprintMetricMapping) (LabelPairFingerprintMapping, error) {
+	collection := map[LabelPair]utility.Set{}
+
+	for fp, m := range b {
+		for n, v := range m {
+			pair := LabelPair{
+				Name:  n,
+				Value: v,
+			}
+			set, ok := collection[pair]
+			if !ok {
+				baseFps, _, err := i.Lookup(&pair)
+				if err != nil {
+					return nil, err
+				}
+
+				set = utility.Set{}
+				for _, baseFp := range baseFps {
+					set.Add(*baseFp)
+				}
+
+				collection[pair] = set
+			}
+
+			set.Add(fp)
+		}
+	}
+
+	batch := LabelPairFingerprintMapping{}
+
+	for pair, set := range collection {
+		fps := batch[pair]
+		for element := range set {
+			fp := element.(clientmodel.Fingerprint)
+			fps = append(fps, &fp)
+		}
+		batch[pair] = fps
+	}
+
+	return batch, nil
+}
+
+func (i *TotalIndexer) IndexMetrics(b FingerprintMetricMapping) error {
+	unindexed, err := findUnindexed(i.MetricMembership, b)
+	if err != nil {
+		return err
+	}
+
+	labelNames, err := extendLabelNameIndex(i.LabelNameToFingerprint, unindexed)
+	if err != nil {
+		return err
+	}
+	if err := i.LabelNameToFingerprint.IndexBatch(labelNames); err != nil {
+		return err
+	}
+
+	labelPairs, err := extendLabelPairIndex(i.LabelPairToFingerprint, unindexed)
+	if err != nil {
+		return err
+	}
+	if err := i.LabelPairToFingerprint.IndexBatch(labelPairs); err != nil {
+		return err
+	}
+
+	if err := i.FingerprintToMetric.IndexBatch(unindexed); err != nil {
+		return err
+	}
+
+	return i.MetricMembership.IndexBatch(unindexed)
+}
diff --git a/storage/metric/leveldb.go b/storage/metric/leveldb.go
index 6571f8aaf..f901385f3 100644
--- a/storage/metric/leveldb.go
+++ b/storage/metric/leveldb.go
@@ -39,10 +39,25 @@ type LevelDBMetricPersistence struct {
 	CurationRemarks         CurationRemarker
 	fingerprintToMetrics    FingerprintMetricIndex
 	labelNameToFingerprints LabelNameFingerprintIndex
-	labelSetToFingerprints  LabelSetFingerprintIndex
+	labelSetToFingerprints  LabelPairFingerprintIndex
 	MetricHighWatermarks    HighWatermarker
 	metricMembershipIndex   MetricMembershipIndex
-	MetricSamples           *leveldb.LevelDBPersistence
+
+	Indexer MetricIndexer
+
+	MetricSamples *leveldb.LevelDBPersistence
+
+	// The remaining indices will be replaced with generalized interface resolvers:
+	//
+	// type FingerprintResolver interface {
+	// 	GetFingerprintForMetric(clientmodel.Metric) (*clientmodel.Fingerprint, bool, error)
+	// 	GetFingerprintsForLabelName(clientmodel.LabelName) (clientmodel.Fingerprints, bool, error)
+	// 	GetFingerprintsForLabelSet(LabelPair) (clientmodel.Fingerprints, bool, error)
+	// }
+
+	// type MetricResolver interface {
+	// 	GetMetricsForFingerprint(clientmodel.Fingerprints) (FingerprintMetricMapping, bool, error)
+	// }
 }
 
 var (
@@ -228,6 +243,13 @@ func NewLevelDBMetricPersistence(baseDirectory string) (*LevelDBMetricPersistenc
 		return nil, fmt.Errorf("Unable to open metric persistence.")
 	}
 
+	emission.Indexer = &TotalIndexer{
+		FingerprintToMetric:    emission.fingerprintToMetrics,
+		LabelNameToFingerprint: emission.labelNameToFingerprints,
+		LabelPairToFingerprint: emission.labelSetToFingerprints,
+		MetricMembership:       emission.metricMembershipIndex,
+	}
+
 	return emission, nil
 }
 
@@ -277,197 +299,6 @@ func groupByFingerprint(samples clientmodel.Samples) map[clientmodel.Fingerprint
 	return fingerprintToSamples
 }
 
-// findUnindexedMetrics scours the metric membership index for each given Metric
-// in the keyspace and returns a map of Fingerprint-Metric pairs that are
-// absent.
-func (l *LevelDBMetricPersistence) findUnindexedMetrics(candidates map[clientmodel.Fingerprint]clientmodel.Metric) (unindexed FingerprintMetricMapping, err error) {
-	defer func(begin time.Time) {
-		duration := time.Since(begin)
-
-		recordOutcome(duration, err, map[string]string{operation: findUnindexedMetrics, result: success}, map[string]string{operation: findUnindexedMetrics, result: failure})
-	}(time.Now())
-
-	unindexed = FingerprintMetricMapping{}
-	for fingerprint, metric := range candidates {
-		indexHas, err := l.hasIndexMetric(metric)
-		if err != nil {
-			return unindexed, err
-		}
-		if !indexHas {
-			unindexed[fingerprint] = metric
-		}
-	}
-
-	return unindexed, nil
-}
-
-// indexLabelNames accumulates all label name to fingerprint index entries for
-// the dirty metrics, appends the new dirtied metrics, sorts, and bulk updates
-// the index to reflect the new state.
-//
-// This operation is idempotent.
-func (l *LevelDBMetricPersistence) indexLabelNames(metrics FingerprintMetricMapping) (err error) {
-	defer func(begin time.Time) {
-		duration := time.Since(begin)
-
-		recordOutcome(duration, err, map[string]string{operation: indexLabelNames, result: success}, map[string]string{operation: indexLabelNames, result: failure})
-	}(time.Now())
-
-	retrieved := map[clientmodel.LabelName]utility.Set{}
-
-	for fingerprint, metric := range metrics {
-		for labelName := range metric {
-			fingerprintSet, ok := retrieved[labelName]
-			if !ok {
-				fingerprints, err := l.GetFingerprintsForLabelName(labelName)
-				if err != nil {
-					return err
-				}
-
-				fingerprintSet = utility.Set{}
-				retrieved[labelName] = fingerprintSet
-
-				for _, fingerprint := range fingerprints {
-					fingerprintSet.Add(*fingerprint)
-				}
-			}
-
-			fingerprintSet.Add(fingerprint)
-		}
-	}
-
-	pending := LabelNameFingerprintMapping{}
-	for name, set := range retrieved {
-		fps := pending[name]
-		for fp := range set {
-			f := fp.(clientmodel.Fingerprint)
-			fps = append(fps, &f)
-		}
-		pending[name] = fps
-	}
-
-	return l.labelNameToFingerprints.IndexBatch(pending)
-}
-
-// indexLabelPairs accumulates all label pair to fingerprint index entries for
-// the dirty metrics, appends the new dirtied metrics, sorts, and bulk updates
-// the index to reflect the new state.
-//
-// This operation is idempotent.
-func (l *LevelDBMetricPersistence) indexLabelPairs(metrics map[clientmodel.Fingerprint]clientmodel.Metric) (err error) {
-	defer func(begin time.Time) {
-		duration := time.Since(begin)
-
-		recordOutcome(duration, err, map[string]string{operation: indexLabelPairs, result: success}, map[string]string{operation: indexLabelPairs, result: failure})
-	}(time.Now())
-
-	collection := map[LabelPair]utility.Set{}
-
-	for fingerprint, metric := range metrics {
-		for labelName, labelValue := range metric {
-			labelPair := LabelPair{
-				Name:  labelName,
-				Value: labelValue,
-			}
-			fingerprintSet, ok := collection[labelPair]
-			if !ok {
-				fingerprints, _, err := l.labelSetToFingerprints.Lookup(&labelPair)
-				if err != nil {
-					return err
-				}
-
-				fingerprintSet = utility.Set{}
-				for _, fingerprint := range fingerprints {
-					fingerprintSet.Add(*fingerprint)
-				}
-
-				collection[labelPair] = fingerprintSet
-			}
-
-			fingerprintSet.Add(fingerprint)
-		}
-	}
-
-	batch := LabelSetFingerprintMapping{}
-
-	for pair, elements := range collection {
-		fps := batch[pair]
-		for element := range elements {
-			fp := element.(clientmodel.Fingerprint)
-			fps = append(fps, &fp)
-		}
-		batch[pair] = fps
-	}
-
-	return l.labelSetToFingerprints.IndexBatch(batch)
-}
-
-// indexFingerprints updates all of the Fingerprint to Metric reverse lookups
-// in the index and then bulk updates.
-//
-// This operation is idempotent.
-func (l *LevelDBMetricPersistence) indexFingerprints(b FingerprintMetricMapping) (err error) {
-	defer func(begin time.Time) {
-		duration := time.Since(begin)
-
-		recordOutcome(duration, err, map[string]string{operation: indexFingerprints, result: success}, map[string]string{operation: indexFingerprints, result: failure})
-	}(time.Now())
-
-	return l.fingerprintToMetrics.IndexBatch(b)
-}
-
-// indexMetrics takes groups of samples, determines which ones contain metrics
-// that are unknown to the storage stack, and then proceeds to update all
-// affected indices.
-func (l *LevelDBMetricPersistence) indexMetrics(fingerprints FingerprintMetricMapping) (err error) {
-	defer func(begin time.Time) {
-		duration := time.Since(begin)
-
-		recordOutcome(duration, err, map[string]string{operation: indexMetrics, result: success}, map[string]string{operation: indexMetrics, result: failure})
-	}(time.Now())
-
-	absentees, err := l.findUnindexedMetrics(fingerprints)
-	if err != nil {
-		return
-	}
-
-	if len(absentees) == 0 {
-		return
-	}
-
-	// TODO: For the missing fingerprints, determine what label names and pairs
-	// are absent and act accordingly and append fingerprints.
-	workers := utility.NewUncertaintyGroup(3)
-
-	go func() {
-		workers.MayFail(l.indexLabelNames(absentees))
-	}()
-
-	go func() {
-		workers.MayFail(l.indexLabelPairs(absentees))
-	}()
-
-	go func() {
-		workers.MayFail(l.indexFingerprints(absentees))
-	}()
-
-	// If any of the preceding operations failed, we will have inconsistent
-	// indices.  Thusly, the Metric membership index should NOT be updated, as
-	// its state is used to determine whether to bulk update the other indices.
-	// Given that those operations are idempotent, it is OK to repeat them;
-	// however, it will consume considerable amounts of time.
-	if !workers.Wait() {
-		return fmt.Errorf("Could not index due to %s", workers.Errors())
-	}
-
-	ms := []clientmodel.Metric{}
-	for _, m := range absentees {
-		ms = append(ms, m)
-	}
-
-	return l.metricMembershipIndex.IndexBatch(ms)
-}
-
 func (l *LevelDBMetricPersistence) refreshHighWatermarks(groups map[clientmodel.Fingerprint]clientmodel.Samples) (err error) {
 	defer func(begin time.Time) {
 		duration := time.Since(begin)
@@ -505,7 +336,7 @@ func (l *LevelDBMetricPersistence) AppendSamples(samples clientmodel.Samples) (e
 			metrics[fingerprint] = samples[0].Metric
 		}
 
-		indexErrChan <- l.indexMetrics(metrics)
+		indexErrChan <- l.Indexer.IndexMetrics(metrics)
 	}(fingerprintToSamples)
 
 	go func(groups map[clientmodel.Fingerprint]clientmodel.Samples) {
diff --git a/storage/metric/tiered.go b/storage/metric/tiered.go
index 1206a268b..34eded717 100644
--- a/storage/metric/tiered.go
+++ b/storage/metric/tiered.go
@@ -95,6 +95,8 @@ type TieredStorage struct {
 	diskSemaphore   chan bool
 
 	wmCache *WatermarkCache
+
+	Indexer MetricIndexer
 }
 
 // viewJob encapsulates a request to extract sample values from the datastore.

From 820e5519889378e3c01d8b817d044794fe6e9177 Mon Sep 17 00:00:00 2001
From: "Matt T. Proud" <matt.proud@gmail.com>
Date: Wed, 7 Aug 2013 13:29:10 +0200
Subject: [PATCH 2/2] Code Review: Nits.

---
 storage/metric/index.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/storage/metric/index.go b/storage/metric/index.go
index 0ae2ca443..1ba652c97 100644
--- a/storage/metric/index.go
+++ b/storage/metric/index.go
@@ -404,13 +404,13 @@ type MetricIndexer interface {
 // that a user or the Prometheus subsystem would want to query against:
 //
 //    "<Label Name>" -> {Fingerprint, ...}
-//    "<Label Name> <Label Pair>" -> {Fingerprint, ...}
+//    "<Label Name> <Label Value>" -> {Fingerprint, ...}
 //
 //    "<Fingerprint>" -> Metric
 //
 //    "<Metric>" -> Existence Value
 //
-//  This type supports concrete queries but only single writes, and it has no
+// This type supports concrete queries but only single writes, and it has no
 // locking semantics to enforce this.
 type TotalIndexer struct {
 	FingerprintToMetric    FingerprintMetricIndex