support per-pattern configurable index pruning

fix #868

Author: Dieterbe

conf/init.go

@@ -3,6 +3,7 @@
 // see https://graphite.readthedocs.io/en/0.9.9/config-carbon.html#storage-schemas-conf
 // * storage-aggregation.conf
 // see http://graphite.readthedocs.io/en/latest/config-carbon.html#storage-aggregation-conf
+// as well as our own file index-rules.conf
 //
 // it also adds defaults (the same ones as graphite),
 // so that even if nothing is matched in the user provided schemas or aggregations,

docker/docker-chaos/metrictank.ini

@@ -332,8 +332,6 @@ timeout = 1s
 num-conns = 10
 # Max number of metricDefs allowed to be unwritten to cassandra
 write-queue-size = 100000
-#automatically clear series from the index if they have not been seen for this much time.
-max-stale = 0
 #Interval at which the index should be checked for stale series.
 prune-interval = 3h
 # synchronize index changes to cassandra. not all your nodes need to do this.
@@ -366,3 +364,5 @@ tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# path to index-rules.conf file
+index-rules-conf = /etc/metrictank/index-rules.conf

docker/docker-cluster/metrictank.ini

@@ -332,8 +332,6 @@ timeout = 1s
 num-conns = 10
 # Max number of metricDefs allowed to be unwritten to cassandra
 write-queue-size = 100000
-#automatically clear series from the index if they have not been seen for this much time.
-max-stale = 0
 #Interval at which the index should be checked for stale series.
 prune-interval = 3h
 # synchronize index changes to cassandra. not all your nodes need to do this.
@@ -366,3 +364,5 @@ tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# path to index-rules.conf file
+index-rules-conf = /etc/metrictank/index-rules.conf

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -332,8 +332,6 @@ timeout = 1s
 num-conns = 10
 # Max number of metricDefs allowed to be unwritten to cassandra
 write-queue-size = 100000
-#automatically clear series from the index if they have not been seen for this much time.
-max-stale = 0
 #Interval at which the index should be checked for stale series.
 prune-interval = 3h
 # synchronize index changes to cassandra. not all your nodes need to do this.
@@ -366,3 +364,5 @@ tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# path to index-rules.conf file
+index-rules-conf = /etc/metrictank/index-rules.conf

docs/config.md

@@ -3,6 +3,7 @@
 Metrictank comes with an [example main config file](https://github.com/grafana/metrictank/blob/master/metrictank-sample.ini),
 a [storage-schemas.conf file](https://github.com/grafana/metrictank/blob/master/scripts/config/storage-schemas.conf) and
 a [storage-aggregation.conf file](https://github.com/grafana/metrictank/blob/master/scripts/config/storage-aggregation.conf)
+an [index-rules.conf file](https://github.com/grafana/metrictank/blob/master/scripts/config/index-rules.conf)
 
 The files themselves are well documented, but for your convenience, they are replicated below.  
 
@@ -394,8 +395,6 @@ timeout = 1s
 num-conns = 10
 # Max number of metricDefs allowed to be unwritten to cassandra
 write-queue-size = 100000
-#automatically clear series from the index if they have not been seen for this much time.
-max-stale = 0
 #Interval at which the index should be checked for stale series.
 prune-interval = 3h
 # synchronize index changes to cassandra. not all your nodes need to do this.
@@ -431,6 +430,44 @@ tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# path to index-rules.conf file
+index-rules-conf = /etc/metrictank/index-rules.conf
+```
+
+# index-rules.conf
+
+```
+# This config file controls when to prune metrics from the index
+# Note:
+# * This file is optional. If it is not present, we won't prune data
+# * Anything not matched will not be pruned
+# * Patterns are string prefix matchers
+# * max-stale is a duration like 7d
+
+[default]
+pattern = 
+max-stale = 0
+
+# storage-aggregation.conf
+
+```
+# This config file controls which summaries are created (using which consolidation functions) for your lower-precision archives, as defined in storage-schemas.conf
+# It is an extension of http://graphite.readthedocs.io/en/latest/config-carbon.html#storage-aggregation-conf
+# Note:
+# * This file is optional. If it is not present, we will use avg for everything
+# * Anything not matched also uses avg for everything
+# * xFilesFactor is not honored yet.  What it is in graphite is a floating point number between 0 and 1 specifying what fraction of the previous retention level's slots must have non-null values in order to aggregate to a non-null value. The default is 0.5.
+# * aggregationMethod specifies the functions used to aggregate values for the next retention level. Legal methods are avg/average, sum, min, max, and last. The default is average.
+# Unlike Graphite, you can specify multiple, as it is often handy to have different summaries available depending on what analysis you need to do.
+# When using multiple, the first one is used for reading.  In the future, we will add capabilities to select the different archives for reading.
+# * the settings configured when metrictank starts are what is applied. So you can enable or disable archives by restarting metrictank.
+#
+# see https://github.com/grafana/metrictank/blob/master/docs/consolidation.md for related info.
+
+[default]
+pattern = .*
+xFilesFactor = 0.1
+aggregationMethod = avg,min,max
 ```
 
 # storage-schemas.conf
@@ -494,27 +531,5 @@ retentions = 1s:35d:10min:7
 # reorderBuffer = 20
 ```
 
-# storage-aggregation.conf
-
-```
-# This config file controls which summaries are created (using which consolidation functions) for your lower-precision archives, as defined in storage-schemas.conf
-# It is an extension of http://graphite.readthedocs.io/en/latest/config-carbon.html#storage-aggregation-conf
-# Note:
-# * This file is optional. If it is not present, we will use avg for everything
-# * Anything not matched also uses avg for everything
-# * xFilesFactor is not honored yet.  What it is in graphite is a floating point number between 0 and 1 specifying what fraction of the previous retention level's slots must have non-null values in order to aggregate to a non-null value. The default is 0.5.
-# * aggregationMethod specifies the functions used to aggregate values for the next retention level. Legal methods are avg/average, sum, min, max, and last. The default is average.
-# Unlike Graphite, you can specify multiple, as it is often handy to have different summaries available depending on what analysis you need to do.
-# When using multiple, the first one is used for reading.  In the future, we will add capabilities to select the different archives for reading.
-# * the settings configured when metrictank starts are what is applied. So you can enable or disable archives by restarting metrictank.
-#
-# see https://github.com/grafana/metrictank/blob/master/docs/consolidation.md for related info.
-
-[default]
-pattern = .*
-xFilesFactor = 0.1
-aggregationMethod = avg,min,max
-```
-
 This file is generated by [config-to-doc](https://github.com/grafana/metrictank/blob/master/scripts/dev/config-to-doc.sh)
 

idx/cassandra/cassandra.go

@@ -64,7 +64,6 @@ var (
 	numConns         int
 	writeQueueSize   int
 	protoVer         int
-	maxStale         time.Duration
 	pruneInterval    time.Duration
 	updateCassIdx    bool
 	updateInterval   time.Duration
@@ -83,7 +82,6 @@ func ConfigSetup() *flag.FlagSet {
 	casIdx.IntVar(&writeQueueSize, "write-queue-size", 100000, "Max number of metricDefs allowed to be unwritten to cassandra")
 	casIdx.BoolVar(&updateCassIdx, "update-cassandra-index", true, "synchronize index changes to cassandra. not all your nodes need to do this.")
 	casIdx.DurationVar(&updateInterval, "update-interval", time.Hour*3, "frequency at which we should update the metricDef lastUpdate field, use 0s for instant updates")
-	casIdx.DurationVar(&maxStale, "max-stale", 0, "clear series from the index if they have not been seen for this much time.")
 	casIdx.DurationVar(&pruneInterval, "prune-interval", time.Hour*3, "Interval at which the index should be checked for stale series.")
 	casIdx.IntVar(&protoVer, "protocol-version", 4, "cql protocol version to use")
 	casIdx.BoolVar(&createKeyspace, "create-keyspace", true, "enable the creation of the index keyspace and tables, only one node needs this")
@@ -234,7 +232,7 @@ func (c *CasIdx) Init() error {
 	//Rebuild the in-memory index.
 	c.rebuildIndex()
 
-	if maxStale > 0 {
+	if memory.IndexRules.Prunable() {
 		if pruneInterval == 0 {
 			return fmt.Errorf("pruneInterval must be greater then 0")
 		}
@@ -355,35 +353,30 @@ func (c *CasIdx) rebuildIndex() {
 	log.Info("cassandra-idx Rebuilding Memory Index from metricDefinitions in Cassandra")
 	pre := time.Now()
 	var defs []schema.MetricDefinition
-	var staleTs uint32
-	if maxStale != 0 {
-		staleTs = uint32(time.Now().Add(maxStale * -1).Unix())
-	}
 	for _, partition := range cluster.Manager.GetPartitions() {
-		defs = c.LoadPartition(partition, defs, staleTs)
+		defs = c.LoadPartition(partition, defs)
 	}
 	num := c.MemoryIdx.Load(defs)
 	log.Info("cassandra-idx Rebuilding Memory Index Complete. Imported %d. Took %s", num, time.Since(pre))
 }
 
 func (c *CasIdx) Load(defs []schema.MetricDefinition, cutoff uint32) []schema.MetricDefinition {
 	iter := c.session.Query("SELECT id, orgid, partition, name, interval, unit, mtype, tags, lastupdate from metric_idx").Iter()
-	return c.load(defs, iter, cutoff)
+	return c.load(defs, iter)
 }
 
-func (c *CasIdx) LoadPartition(partition int32, defs []schema.MetricDefinition, cutoff uint32) []schema.MetricDefinition {
+func (c *CasIdx) LoadPartition(partition int32, defs []schema.MetricDefinition) []schema.MetricDefinition {
 	iter := c.session.Query("SELECT id, orgid, partition, name, interval, unit, mtype, tags, lastupdate from metric_idx where partition=?", partition).Iter()
-	return c.load(defs, iter, cutoff)
+	return c.load(defs, iter)
 }
 
-func (c *CasIdx) load(defs []schema.MetricDefinition, iter cqlIterator, cutoff uint32) []schema.MetricDefinition {
+func (c *CasIdx) load(defs []schema.MetricDefinition, iter cqlIterator) []schema.MetricDefinition {
 	defsByNames := make(map[string][]*schema.MetricDefinition)
 	var id, name, unit, mtype string
 	var orgId, interval int
 	var partition int32
 	var lastupdate int64
 	var tags []string
-	cutoff64 := int64(cutoff)
 	for iter.Scan(&id, &orgId, &partition, &name, &interval, &unit, &mtype, &tags, &lastupdate) {
 		mkey, err := schema.MKeyFromString(id)
 		if err != nil {
@@ -412,10 +405,14 @@ func (c *CasIdx) load(defs []schema.MetricDefinition, iter cqlIterator, cutoff u
 		log.Fatal(4, "Could not close iterator: %s", err.Error())
 	}
 
+	indexChecks := memory.IndexRules.Checks(time.Now())
+
 NAMES:
 	for name, defsByName := range defsByNames {
+		irId, _ := memory.IndexRules.Match(name)
+		check := indexChecks[irId]
 		for _, def := range defsByName {
-			if def.LastUpdate >= cutoff64 {
+			if check.Keep || def.LastUpdate >= check.Cutoff {
 				// if one of the defs in a name is not stale, then we'll need to add
 				// all the associated MDs to the defs slice
 				for _, defToAdd := range defsByNames[name] {
@@ -528,19 +525,17 @@ func (c *CasIdx) deleteDefAsync(key schema.MKey, part int32) {
 	}()
 }
 
-func (c *CasIdx) Prune(oldest time.Time) ([]idx.Archive, error) {
-	pre := time.Now()
-	pruned, err := c.MemoryIdx.Prune(oldest)
-	statPruneDuration.Value(time.Since(pre))
+func (c *CasIdx) Prune(now time.Time) ([]idx.Archive, error) {
+	pruned, err := c.MemoryIdx.Prune(now)
+	statPruneDuration.Value(time.Since(now))
 	return pruned, err
 }
 
 func (c *CasIdx) prune() {
 	ticker := time.NewTicker(pruneInterval)
-	for range ticker.C {
-		log.Debug("cassandra-idx: pruning items from index that have not been seen for %s", maxStale.String())
-		staleTs := time.Now().Add(maxStale * -1)
-		_, err := c.Prune(staleTs)
+	for now := range ticker.C {
+		log.Debug("cassandra-idx: pruning items")
+		_, err := c.Prune(now)
 		if err != nil {
 			log.Error(3, "cassandra-idx: prune error. %s", err)
 		}

idx/idx.go

@@ -28,6 +28,7 @@ type Archive struct {
 	schema.MetricDefinition
 	SchemaId uint16 // index in mdata.schemas (not persisted)
 	AggId    uint16 // index in mdata.aggregations (not persisted)
+	IrId     uint16 // index in mdata.indexrules (not persisted)
 	LastSave uint32 // last time the metricDefinition was saved to a backend store (cassandra)
 }
 

idx/memory/memory.go

@@ -3,12 +3,14 @@ package memory
 import (
 	"flag"
 	"fmt"
+	"io/ioutil"
 	"regexp"
 	"sort"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/grafana/metrictank/conf"
 	"github.com/grafana/metrictank/errors"
 	"github.com/grafana/metrictank/idx"
 	"github.com/grafana/metrictank/mdata"
@@ -50,6 +52,8 @@ var (
 	matchCacheSize  int
 	TagSupport      bool
 	TagQueryWorkers int // number of workers to spin up when evaluation tag expressions
+	indexRulesFile  string
+	IndexRules      conf.IndexRules
 )
 
 func ConfigSetup() {
@@ -58,9 +62,26 @@ func ConfigSetup() {
 	memoryIdx.BoolVar(&TagSupport, "tag-support", false, "enables/disables querying based on tags")
 	memoryIdx.IntVar(&TagQueryWorkers, "tag-query-workers", 50, "number of workers to spin up to evaluate tag queries")
 	memoryIdx.IntVar(&matchCacheSize, "match-cache-size", 1000, "size of regular expression cache in tag query evaluation")
+	memoryIdx.StringVar(&indexRulesFile, "rules-file", "/etc/metrictank/index-rules.conf", "path to index-rules.conf file")
 	globalconf.Register("memory-idx", memoryIdx)
 }
 
+func ConfigProcess() {
+	// read index-rules.conf
+	// file is optional, quit on errors
+	// since we can't distinguish errors reading vs parsing, we'll just try a read separately first
+	_, err := ioutil.ReadFile(indexRulesFile)
+	if err == nil {
+		IndexRules, err = conf.ReadIndexRules(indexRulesFile)
+		if err != nil {
+			log.Fatal(3, "can't read index-rules file %q: %s", indexRulesFile, err.Error())
+		}
+	} else {
+		log.Info("Could not read %s: %s: using defaults", indexRulesFile, err)
+		IndexRules = conf.NewIndexRules()
+	}
+}
+
 type Tree struct {
 	Items map[string]*Node // key is the full path of the node.
 }
@@ -373,11 +394,13 @@ func (m *MemoryIdx) add(def *schema.MetricDefinition) idx.Archive {
 
 	schemaId, _ := mdata.MatchSchema(path, def.Interval)
 	aggId, _ := mdata.MatchAgg(path)
+	irId, _ := IndexRules.Match(path)
 	sort.Strings(def.Tags)
 	archive := &idx.Archive{
 		MetricDefinition: *def,
 		SchemaId:         schemaId,
 		AggId:            aggId,
+		IrId:             irId,
 	}
 
 	if TagSupport && len(def.Tags) > 0 {
@@ -915,7 +938,7 @@ func (m *MemoryIdx) Find(orgId uint32, pattern string, from int64) ([]idx.Node,
 						log.Debug("memory-idx: from is %d, so skipping %s which has LastUpdate %d", from, def.Id, def.LastUpdate)
 						continue
 					}
-					log.Debug("memory-idx Find: adding to path %s archive id=%s name=%s int=%d schemaId=%d aggId=%d lastSave=%d", n.Path, def.Id, def.Name, def.Interval, def.SchemaId, def.AggId, def.LastSave)
+					log.Debug("memory-idx Find: adding to path %s archive id=%s name=%s int=%d schemaId=%d aggId=%d irId=%d lastSave=%d", n.Path, def.Id, def.Name, def.Interval, def.SchemaId, def.AggId, def.IrId, def.LastSave)
 					idxNode.Defs = append(idxNode.Defs, *def)
 				}
 				if len(idxNode.Defs) == 0 {
@@ -1225,9 +1248,8 @@ func (m *MemoryIdx) delete(orgId uint32, n *Node, deleteEmptyParents, deleteChil
 	return deletedDefs
 }
 
-// delete series from the index if they have not been seen since "oldest"
-func (m *MemoryIdx) Prune(oldest time.Time) ([]idx.Archive, error) {
-	oldestUnix := oldest.Unix()
+// Prune prunes series from the index if they have become stale per their index-rule
+func (m *MemoryIdx) Prune(now time.Time) ([]idx.Archive, error) {
 	orgs := make(map[uint32]struct{})
 	log.Info("memory-idx: pruning stale metricDefs across all orgs")
 	m.RLock()
@@ -1251,9 +1273,12 @@ func (m *MemoryIdx) Prune(oldest time.Time) ([]idx.Archive, error) {
 	pre := time.Now()
 
 	m.RLock()
+
+	indexChecks := IndexRules.Checks(now)
 DEFS:
 	for _, def := range m.defById {
-		if def.LastUpdate >= oldestUnix {
+		check := indexChecks[def.IrId]
+		if check.Keep || def.LastUpdate >= check.Cutoff {
 			continue DEFS
 		}
 
@@ -1269,7 +1294,7 @@ DEFS:
 			}
 
 			for _, id := range n.Defs {
-				if m.defById[id].LastUpdate >= oldestUnix {
+				if m.defById[id].LastUpdate >= check.Cutoff {
 					continue DEFS
 				}
 			}
@@ -1280,7 +1305,7 @@ DEFS:
 			// if any other MetricDef with the same tag set is not expired yet,
 			// then we do not want to prune any of them
 			for def := range defs {
-				if def.LastUpdate >= oldestUnix {
+				if def.LastUpdate >= check.Cutoff {
 					continue DEFS
 				}
 			}

metrictank-sample.ini

@@ -335,8 +335,6 @@ timeout = 1s
 num-conns = 10
 # Max number of metricDefs allowed to be unwritten to cassandra
 write-queue-size = 100000
-#automatically clear series from the index if they have not been seen for this much time.
-max-stale = 0
 #Interval at which the index should be checked for stale series.
 prune-interval = 3h
 # synchronize index changes to cassandra. not all your nodes need to do this.
@@ -369,3 +367,5 @@ tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# path to index-rules.conf file
+index-rules-conf = /etc/metrictank/index-rules.conf

metrictank.go

@@ -221,6 +221,7 @@ func main() {
 	***********************************/
 	inCarbon.ConfigProcess()
 	inKafkaMdm.ConfigProcess(*instance)
+	memory.ConfigProcess()
 	inPrometheus.ConfigProcess()
 	notifierNsq.ConfigProcess()
 	notifierKafka.ConfigProcess(*instance)