Repository: prometheus/statsd_exporter
Pull request #164 (2018-11-13 19:54:18)

Remove stale metrics
This patch implements clearing of metrics with variable labels with per-mapping configurable timeout (ttl). Our monitoring installation is suffered from the behavior desribed in #129. There are metrics from Kubernetes pods with constantly changing values for labels. For example, metric `ingress_nginx_upstream_retries_count` with label `pod` and changing pod name as a value for this label. There is no sense to store metrics with old pod names as they are never repeated. Removing metrics with this stale values will not cause any damage to subsequent aggregations or multiple prometheus servers.

The patch is divided by 3 parts to simplify a review:
- replace Metrics with Collectors to gain ability of deletion metrics. So Elements in CounterContainer stores CounterVecs instead of Counters.
- implement labels values storage in Exporter to detect metrics staleness
- implement configurable timeout (ttl). Add key `ttl` to `default` section and to `mapping` sections.

`ttl` is a timeout in seconds for stale metrics. Exporter saves each labels values set and updates last time it receives metric with these labels values. If no metric with the same labels values received for `ttl` seconds, then statsd_exporter stops reporting metric with this labels values set.

Mark this as WIP. It would be great to hear your thoughts on this.

diafour: Tests are broken for now because of changes in Elements fields. Will fix it at the daylight.
matthiasr: Awesome, thank you for contributing! It will take me a little while to go through the code changes in detail, but I wanted to say that I think the general approach is good.
R0quef0rt: This is a great idea. Really hoping to see it implemented soon. 

I work in an environment where our central infrastructure is in the cloud, but our clients are behind corporate firewalls. We're developing an application-specific plugin that will send metrics to Prometheus - but we keep running into this problem. 

The pushgateway - and now, statsd - currently "cache" metrics indefinitely. This makes it difficult to determine if a client has gone down - because their metrics are being persisted indefinitely. 

Here's hoping we'll see a TTL feature added soon!
matthiasr: One workaround for this specific question and situation is to push a metric
that is the current (unix) time from all clients, then you can see when
that stops updating / gets too far away from the server time().

On Tue, Nov 27, 2018, 09:56 R0quef0rt <notifications@github.com> wrote:

> This is a great idea. Really hoping to see it implemented soon.
>
> I work in an environment where our central infrastructure is in the cloud,
> but our clients are behind corporate firewalls. We're developing an
> application-specific plugin that will send metrics to Prometheus - but we
> keep running into this problem.
>
> The pushgateway - and now, statsd - currently "cache" metrics
> indefinitely. This makes it difficult to determine if a client has gone
> down - because their metrics are being persisted indefinitely.
>
> Here's hoping we'll see a TTL feature added soon!
>
> —
> You are receiving this because you commented.
>
>
> Reply to this email directly, view it on GitHub
> <https://github.com/prometheus/statsd_exporter/pull/164#issuecomment-441977803>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/AAICBgNOO_rwtYaDB8NI7LUUsT3OKz-sks5uzP4ngaJpZM4YcZUb>
> .
>

diafour: Tests are good now. Problem was in TestHistogramUnits with mocking of Histogram metric. Resolve it with Write a dto.Metric as in a client_golang https://github.com/prometheus/client_golang/blob/master/prometheus/examples_test.go#L497-L498

`hashNameAndLabels` is used to create a key for label values in intermediate storage (Exporter.labelValues)

`labelNames` is got proper name
diafour: @matthiasr Any comments? statsd_exporter with this patch now lives on 5 production clusters and works well.
matthiasr: sorry, I got distracted and then forgot. the title still says WIP?
matthiasr: What happens when there are two mappings with different TTLs mapping to the same metric name / label? That is a perfectly valid situation, and we need to at least document it.
diafour: > Could you please add tests for the added mapping fields, and tests that demonstrate/verify the expiry behaviour?

Done

> Please update the documentation (README) as well.

Done

> The exporter allows live-reloading of the configuration, so the label set or TTL of a given metric can change during the lifetime of the exporter. What happens in that case?

Modified ttl for a metric will not be in use until call to handleEvent with a new value for metric. In theory one would want to expire a stale metric with `ttl: 30m` without reload by decreasing ttl to 1s — this is not possible now. Should we handle such situation? I think it requires some kind of signalization about config reloads from main to exporter.

> What happens when there are two mappings with different TTLs mapping to the same metric name / label? That is a perfectly valid situation, and we need to at least document it.

`handleEvent` use mapper.GetMapping which use FSM.GetMapping to get a *MetricMapping by metric name and metric type. This mapping defines ttl to use. As I can see labels are not used to find a 
mapping. Can you tell what is an algorithm to find the mapping in FSM? May be you have an example of config to run some tests?

diafour: @matthiasr some comments about tests. Previous algorithm with getting a Metric instance and call a Write works good for simple cases such as test for 1 value or test values in one routine. So I use a Gatherer which is thread safe and to be more close to what a scraper get. Method `Gather` returns many additional metrics but for test purposes it is not important.

matthiasr: > Should we handle such situation? I think it requires some kind of signalization about config reloads from main to exporter.

I think for now it's okay to document this and leave it be. If someone really needs instant expiration, they can restart or implement that signalling.
matthiasr: > Can you tell what is an algorithm to find the mapping in FSM

The FSM finds exactly one mapping for a given statsd metric name. The Prometheus metric name and labels are an output of that. I think what will happen in this case is that whichever mapping last matched will win. Again, I think that's okay as long as we document it.
matthiasr: Nevermind my last comment, I was confusing myself about what level TTLs are applied at. I don't expect anyone to have multiple mappings to the same Prometheus metric name, labels _and_ label values, so we don't need to cover that case. If you're really mapping multiple event streams to the same metric, all bets are off, but last-event-wins is what you get either way.

Patch:
diff --git a/README.md b/README.md
index 529fbb1..16f2646 100644
--- a/README.md
+++ b/README.md
@@ -268,6 +268,7 @@ defaults:
   buckets: [.005, .01, .025, .05, .1, .25, .5, 1, 2.5 ]
   match_type: glob
   glob_disable_ordering: false
+  ttl: 0 # metrics do not expire
 mappings:
 # This will be a histogram using the buckets set in `defaults`.
 - match: test.timing.*.*.*
@@ -350,6 +351,18 @@ mappings:
 
 Possible values for `match_metric_type` are `gauge`, `counter` and `timer`.
 
+### Time series expiration
+
+The `ttl` parameter can be used to define the expiration time for stale metrics.
+The value is a time duration with valid time units: "ns", "us" (or "µs"),
+"ms", "s", "m", "h". For example, `ttl: 1m20s`. `0` value is used to indicate
+metrics that do not expire.
+
+ TTLs are applied to each mapped metric name/labels combination whenever
+ new samples are received. This means that you cannot immediately expire a
+ metric only by changing the mapping configuration. At least one sample must
+ be received for updated mappings to take effect.
+
 ## Using Docker
 
 You can deploy this exporter using the [prom/statsd-exporter](https://registry.hub.docker.com/u/prom/statsd-exporter/) Docker image.
diff --git a/exporter.go b/exporter.go
index b227516..13c98f1 100644
--- a/exporter.go
+++ b/exporter.go
@@ -22,14 +22,17 @@ import (
 	"io"
 	"net"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
+	"time"
 	"unicode/utf8"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/log"
 	"github.com/prometheus/common/model"
 
+	"github.com/prometheus/statsd_exporter/pkg/clock"
 	"github.com/prometheus/statsd_exporter/pkg/mapper"
 )
 
@@ -48,6 +51,15 @@ var (
 	intBuf = make([]byte, 8)
 )
 
+func labelNames(labels prometheus.Labels) []string {
+	names := make([]string, 0, len(labels))
+	for labelName := range labels {
+		names = append(names, labelName)
+	}
+	sort.Strings(names)
+	return names
+}
+
 // hashNameAndLabels returns a hash value of the provided name string and all
 // the label names and values in the provided labels map.
 //
@@ -64,74 +76,82 @@ func hashNameAndLabels(name string, labels prometheus.Labels) uint64 {
 }
 
 type CounterContainer struct {
-	Elements map[uint64]prometheus.Counter
+	//           metric name
+	Elements map[string]*prometheus.CounterVec
 }
 
 func NewCounterContainer() *CounterContainer {
 	return &CounterContainer{
-		Elements: make(map[uint64]prometheus.Counter),
+		Elements: make(map[string]*prometheus.CounterVec),
 	}
 }
 
 func (c *CounterContainer) Get(metricName string, labels prometheus.Labels, help string) (prometheus.Counter, error) {
-	hash := hashNameAndLabels(metricName, labels)
-	counter, ok := c.Elements[hash]
+	counterVec, ok := c.Elements[metricName]
 	if !ok {
-		counter = prometheus.NewCounter(prometheus.CounterOpts{
-			Name:        metricName,
-			Help:        help,
-			ConstLabels: labels,
-		})
-		if err := prometheus.Register(counter); err != nil {
+		counterVec = prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: metricName,
+			Help: help,
+		}, labelNames(labels))
+		if err := prometheus.Register(counterVec); err != nil {
 			return nil, err
 		}
-		c.Elements[hash] = counter
+		c.Elements[metricName] = counterVec
+	}
+	return counterVec.GetMetricWith(labels)
+}
+
+func (c *CounterContainer) Delete(metricName string, labels prometheus.Labels) {
+	if _, ok := c.Elements[metricName]; ok {
+		c.Elements[metricName].Delete(labels)
 	}
-	return counter, nil
 }
 
 type GaugeContainer struct {
-	Elements map[uint64]prometheus.Gauge
+	Elements map[string]*prometheus.GaugeVec
 }
 
 func NewGaugeContainer() *GaugeContainer {
 	return &GaugeContainer{
-		Elements: make(map[uint64]prometheus.Gauge),
+		Elements: make(map[string]*prometheus.GaugeVec),
 	}
 }
 
 func (c *GaugeContainer) Get(metricName string, labels prometheus.Labels, help string) (prometheus.Gauge, error) {
-	hash := hashNameAndLabels(metricName, labels)
-	gauge, ok := c.Elements[hash]
+	gaugeVec, ok := c.Elements[metricName]
 	if !ok {
-		gauge = prometheus.NewGauge(prometheus.GaugeOpts{
-			Name:        metricName,
-			Help:        help,
-			ConstLabels: labels,
-		})
-		if err := prometheus.Register(gauge); err != nil {
+		gaugeVec = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: metricName,
+			Help: help,
+		}, labelNames(labels))
+		if err := prometheus.Register(gaugeVec); err != nil {
 			return nil, err
 		}
-		c.Elements[hash] = gauge
+		c.Elements[metricName] = gaugeVec
+	}
+	return gaugeVec.GetMetricWith(labels)
+}
+
+func (c *GaugeContainer) Delete(metricName string, labels prometheus.Labels) {
+	if _, ok := c.Elements[metricName]; ok {
+		c.Elements[metricName].Delete(labels)
 	}
-	return gauge, nil
 }
 
 type SummaryContainer struct {
-	Elements map[uint64]prometheus.Summary
+	Elements map[string]*prometheus.SummaryVec
 	mapper   *mapper.MetricMapper
 }
 
 func NewSummaryContainer(mapper *mapper.MetricMapper) *SummaryContainer {
 	return &SummaryContainer{
-		Elements: make(map[uint64]prometheus.Summary),
+		Elements: make(map[string]*prometheus.SummaryVec),
 		mapper:   mapper,
 	}
 }
 
-func (c *SummaryContainer) Get(metricName string, labels prometheus.Labels, help string, mapping *mapper.MetricMapping) (prometheus.Summary, error) {
-	hash := hashNameAndLabels(metricName, labels)
-	summary, ok := c.Elements[hash]
+func (c *SummaryContainer) Get(metricName string, labels prometheus.Labels, help string, mapping *mapper.MetricMapping) (prometheus.Observer, error) {
+	summaryVec, ok := c.Elements[metricName]
 	if !ok {
 		quantiles := c.mapper.Defaults.Quantiles
 		if mapping != nil && mapping.Quantiles != nil && len(mapping.Quantiles) > 0 {
@@ -141,54 +161,63 @@ func (c *SummaryContainer) Get(metricName string, labels prometheus.Labels, help
 		for _, q := range quantiles {
 			objectives[q.Quantile] = q.Error
 		}
-		summary = prometheus.NewSummary(
+		summaryVec = prometheus.NewSummaryVec(
 			prometheus.SummaryOpts{
-				Name:        metricName,
-				Help:        help,
-				ConstLabels: labels,
-				Objectives:  objectives,
-			})
-		if err := prometheus.Register(summary); err != nil {
+				Name:       metricName,
+				Help:       help,
+				Objectives: objectives,
+			}, labelNames(labels))
+		if err := prometheus.Register(summaryVec); err != nil {
 			return nil, err
 		}
-		c.Elements[hash] = summary
+		c.Elements[metricName] = summaryVec
+	}
+	return summaryVec.GetMetricWith(labels)
+}
+
+func (c *SummaryContainer) Delete(metricName string, labels prometheus.Labels) {
+	if _, ok := c.Elements[metricName]; ok {
+		c.Elements[metricName].Delete(labels)
 	}
-	return summary, nil
 }
 
 type HistogramContainer struct {
-	Elements map[uint64]prometheus.Histogram
+	Elements map[string]*prometheus.HistogramVec
 	mapper   *mapper.MetricMapper
 }
 
 func NewHistogramContainer(mapper *mapper.MetricMapper) *HistogramContainer {
 	return &HistogramContainer{
-		Elements: make(map[uint64]prometheus.Histogram),
+		Elements: make(map[string]*prometheus.HistogramVec),
 		mapper:   mapper,
 	}
 }
 
-func (c *HistogramContainer) Get(metricName string, labels prometheus.Labels, help string, mapping *mapper.MetricMapping) (prometheus.Histogram, error) {
-	hash := hashNameAndLabels(metricName, labels)
-	histogram, ok := c.Elements[hash]
+func (c *HistogramContainer) Get(metricName string, labels prometheus.Labels, help string, mapping *mapper.MetricMapping) (prometheus.Observer, error) {
+	histogramVec, ok := c.Elements[metricName]
 	if !ok {
 		buckets := c.mapper.Defaults.Buckets
 		if mapping != nil && mapping.Buckets != nil && len(mapping.Buckets) > 0 {
 			buckets = mapping.Buckets
 		}
-		histogram = prometheus.NewHistogram(
+		histogramVec = prometheus.NewHistogramVec(
 			prometheus.HistogramOpts{
-				Name:        metricName,
-				Help:        help,
-				ConstLabels: labels,
-				Buckets:     buckets,
-			})
-		c.Elements[hash] = histogram
-		if err := prometheus.Register(histogram); err != nil {
+				Name:    metricName,
+				Help:    help,
+				Buckets: buckets,
+			}, labelNames(labels))
+		if err := prometheus.Register(histogramVec); err != nil {
 			return nil, err
 		}
+		c.Elements[metricName] = histogramVec
+	}
+	return histogramVec.GetMetricWith(labels)
+}
+
+func (c *HistogramContainer) Delete(metricName string, labels prometheus.Labels) {
+	if _, ok := c.Elements[metricName]; ok {
+		c.Elements[metricName].Delete(labels)
 	}
-	return histogram, nil
 }
 
 type Event interface {
@@ -234,12 +263,19 @@ func (c *TimerEvent) MetricType() mapper.MetricType { return mapper.MetricTypeTi
 
 type Events []Event
 
+type LabelValues struct {
+	lastRegisteredAt time.Time
+	labels           prometheus.Labels
+	ttl              time.Duration
+}
+
 type Exporter struct {
-	Counters   *CounterContainer
-	Gauges     *GaugeContainer
-	Summaries  *SummaryContainer
-	Histograms *HistogramContainer
-	mapper     *mapper.MetricMapper
+	Counters    *CounterContainer
+	Gauges      *GaugeContainer
+	Summaries   *SummaryContainer
+	Histograms  *HistogramContainer
+	mapper      *mapper.MetricMapper
+	labelValues map[string]map[uint64]*LabelValues
 }
 
 func escapeMetricName(metricName string) string {
@@ -256,14 +292,21 @@ func escapeMetricName(metricName string) string {
 // Listen handles all events sent to the given channel sequentially. It
 // terminates when the channel is closed.
 func (b *Exporter) Listen(e <-chan Events) {
+	removeStaleMetricsTicker := clock.NewTicker(time.Second)
+
 	for {
-		events, ok := <-e
-		if !ok {
-			log.Debug("Channel is closed. Break out of Exporter.Listener.")
-			return
-		}
-		for _, event := range events {
-			b.handleEvent(event)
+		select {
+		case <-removeStaleMetricsTicker.C:
+			b.removeStaleMetrics()
+		case events, ok := <-e:
+			if !ok {
+				log.Debug("Channel is closed. Break out of Exporter.Listener.")
+				removeStaleMetricsTicker.Stop()
+				return
+			}
+			for _, event := range events {
+				b.handleEvent(event)
+			}
 		}
 	}
 }
@@ -273,6 +316,9 @@ func (b *Exporter) handleEvent(event Event) {
 	mapping, labels, present := b.mapper.GetMapping(event.MetricName(), event.MetricType())
 	if mapping == nil {
 		mapping = &mapper.MetricMapping{}
+		if b.mapper.Defaults.Ttl != 0 {
+			mapping.Ttl = b.mapper.Defaults.Ttl
+		}
 	}
 
 	if mapping.Action == mapper.ActionTypeDrop {
@@ -313,7 +359,7 @@ func (b *Exporter) handleEvent(event Event) {
 		)
 		if err == nil {
 			counter.Add(event.Value())
-
+			b.saveLabelValues(metricName, prometheusLabels, mapping.Ttl)
 			eventStats.WithLabelValues("counter").Inc()
 		} else {
 			log.Debugf(regErrF, metricName, err)
@@ -333,7 +379,7 @@ func (b *Exporter) handleEvent(event Event) {
 			} else {
 				gauge.Set(event.Value())
 			}
-
+			b.saveLabelValues(metricName, prometheusLabels, mapping.Ttl)
 			eventStats.WithLabelValues("gauge").Inc()
 		} else {
 			log.Debugf(regErrF, metricName, err)
@@ -359,6 +405,7 @@ func (b *Exporter) handleEvent(event Event) {
 			)
 			if err == nil {
 				histogram.Observe(event.Value() / 1000) // prometheus presumes seconds, statsd millisecond
+				b.saveLabelValues(metricName, prometheusLabels, mapping.Ttl)
 				eventStats.WithLabelValues("timer").Inc()
 			} else {
 				log.Debugf(regErrF, metricName, err)
@@ -374,6 +421,7 @@ func (b *Exporter) handleEvent(event Event) {
 			)
 			if err == nil {
 				summary.Observe(event.Value())
+				b.saveLabelValues(metricName, prometheusLabels, mapping.Ttl)
 				eventStats.WithLabelValues("timer").Inc()
 			} else {
 				log.Debugf(regErrF, metricName, err)
@@ -390,13 +438,56 @@ func (b *Exporter) handleEvent(event Event) {
 	}
 }
 
+// removeStaleMetrics removes label values set from metric with stale values
+func (b *Exporter) removeStaleMetrics() {
+	now := clock.Now()
+	// delete timeseries with expired ttl
+	for metricName := range b.labelValues {
+		for hash, lvs := range b.labelValues[metricName] {
+			if lvs.ttl == 0 {
+				continue
+			}
+			if lvs.lastRegisteredAt.Add(lvs.ttl).Before(now) {
+				b.Counters.Delete(metricName, lvs.labels)
+				b.Gauges.Delete(metricName, lvs.labels)
+				b.Summaries.Delete(metricName, lvs.labels)
+				b.Histograms.Delete(metricName, lvs.labels)
+				delete(b.labelValues[metricName], hash)
+			}
+		}
+	}
+}
+
+// saveLabelValues stores label values set to labelValues and update lastRegisteredAt time and ttl value
+func (b *Exporter) saveLabelValues(metricName string, labels prometheus.Labels, ttl time.Duration) {
+	metric, hasMetric := b.labelValues[metricName]
+	if !hasMetric {
+		metric = make(map[uint64]*LabelValues)
+		b.labelValues[metricName] = metric
+	}
+	hash := hashNameAndLabels(metricName, labels)
+	metricLabelValues, ok := metric[hash]
+	if !ok {
+		metricLabelValues = &LabelValues{
+			labels: labels,
+			ttl:    ttl,
+		}
+		b.labelValues[metricName][hash] = metricLabelValues
+	}
+	now := clock.Now()
+	metricLabelValues.lastRegisteredAt = now
+	// Update ttl from mapping
+	metricLabelValues.ttl = ttl
+}
+
 func NewExporter(mapper *mapper.MetricMapper) *Exporter {
 	return &Exporter{
-		Counters:   NewCounterContainer(),
-		Gauges:     NewGaugeContainer(),
-		Summaries:  NewSummaryContainer(mapper),
-		Histograms: NewHistogramContainer(mapper),
-		mapper:     mapper,
+		Counters:    NewCounterContainer(),
+		Gauges:      NewGaugeContainer(),
+		Summaries:   NewSummaryContainer(mapper),
+		Histograms:  NewHistogramContainer(mapper),
+		mapper:      mapper,
+		labelValues: make(map[string]map[uint64]*LabelValues, 0),
 	}
 }
 
diff --git a/go.mod b/go.mod
index b69800f..488a957 100644
--- a/go.mod
+++ b/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/onsi/gomega v1.4.3 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v0.9.2
+	github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910
 	github.com/prometheus/common v0.0.0-20181126121408-4724e9255275
 	github.com/sergi/go-diff v1.0.0 // indirect
 	github.com/sirupsen/logrus v1.0.3 // indirect
diff --git a/pkg/clock/clock.go b/pkg/clock/clock.go
new file mode 100644
index 0000000..2dc8394
--- /dev/null
+++ b/pkg/clock/clock.go
@@ -0,0 +1,28 @@
+package clock
+
+import (
+	"time"
+)
+
+var ClockInstance *Clock
+
+type Clock struct {
+	Instant  time.Time
+	TickerCh chan time.Time
+}
+
+func Now() time.Time {
+	if ClockInstance == nil {
+		return time.Now()
+	}
+	return ClockInstance.Instant
+}
+
+func NewTicker(d time.Duration) *time.Ticker {
+	if ClockInstance == nil || ClockInstance.TickerCh == nil {
+		return time.NewTicker(d)
+	}
+	return &time.Ticker{
+		C: ClockInstance.TickerCh,
+	}
+}
diff --git a/pkg/mapper/mapper.go b/pkg/mapper/mapper.go
index 46e90e6..b9fe3d7 100644
--- a/pkg/mapper/mapper.go
+++ b/pkg/mapper/mapper.go
@@ -22,6 +22,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/statsd_exporter/pkg/mapper/fsm"
 	yaml "gopkg.in/yaml.v2"
+	"time"
 )
 
 var (
@@ -39,6 +40,7 @@ type mapperConfigDefaults struct {
 	Quantiles           []metricObjective `yaml:"quantiles"`
 	MatchType           MatchType         `yaml:"match_type"`
 	GlobDisableOrdering bool              `yaml:"glob_disable_ordering"`
+	Ttl                 time.Duration     `yaml:"ttl"`
 }
 
 type MetricMapper struct {
@@ -69,6 +71,7 @@ type MetricMapping struct {
 	HelpText        string            `yaml:"help"`
 	Action          ActionType        `yaml:"action"`
 	MatchMetricType MetricType        `yaml:"match_metric_type"`
+	Ttl             time.Duration     `yaml:"ttl"`
 }
 
 type metricObjective struct {
@@ -177,6 +180,10 @@ func (m *MetricMapper) InitFromYAMLString(fileContents string) error {
 			currentMapping.Quantiles = n.Defaults.Quantiles
 		}
 
+		if currentMapping.Ttl == 0 && n.Defaults.Ttl > 0 {
+			currentMapping.Ttl = n.Defaults.Ttl
+		}
+
 	}
 
 	m.mutex.Lock()


Test patch:
diff --git a/exporter_test.go b/exporter_test.go
index 0722240..b4615b8 100644
--- a/exporter_test.go
+++ b/exporter_test.go
@@ -20,7 +20,9 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 
+	"github.com/prometheus/statsd_exporter/pkg/clock"
 	"github.com/prometheus/statsd_exporter/pkg/mapper"
 )
 
@@ -38,22 +40,19 @@ func TestNegativeCounter(t *testing.T) {
 		}
 	}()
 
-	events := make(chan Events, 1)
-	c := Events{
-		&CounterEvent{
-			metricName: "foo",
-			value:      -1,
-		},
-	}
-	events <- c
-	ex := NewExporter(&mapper.MetricMapper{})
-
-	// Close channel to signify we are done with the listener after a short period.
+	events := make(chan Events, 0)
 	go func() {
-		time.Sleep(time.Millisecond * 100)
+		c := Events{
+			&CounterEvent{
+				metricName: "foo",
+				value:      -1,
+			},
+		}
+		events <- c
 		close(events)
 	}()
 
+	ex := NewExporter(&mapper.MetricMapper{})
 	ex.Listen(events)
 }
 
@@ -62,34 +61,37 @@ func TestNegativeCounter(t *testing.T) {
 // It sends the same tags first with a valid value, then with an invalid one.
 // The exporter should not panic, but drop the invalid event
 func TestInvalidUtf8InDatadogTagValue(t *testing.T) {
-	ex := NewExporter(&mapper.MetricMapper{})
-	for _, l := range []statsDPacketHandler{&StatsDUDPListener{}, &mockStatsDTCPListener{}} {
-		events := make(chan Events, 2)
-
-		l.handlePacket([]byte("bar:200|c|#tag:value\nbar:200|c|#tag:\xc3\x28invalid"), events)
-
-		// Close channel to signify we are done with the listener after a short period.
-		go func() {
-			time.Sleep(time.Millisecond * 100)
-			close(events)
-		}()
+	defer func() {
+		if e := recover(); e != nil {
+			err := e.(error)
+			t.Fatalf("Exporter listener should not panic on bad utf8: %q", err.Error())
+		}
+	}()
 
-		ex.Listen(events)
-	}
-}
+	events := make(chan Events, 0)
 
-type MockHistogram struct {
-	prometheus.Metric
-	prometheus.Collector
-	value float64
-}
+	go func() {
+		for _, l := range []statsDPacketHandler{&StatsDUDPListener{}, &mockStatsDTCPListener{}} {
+			l.handlePacket([]byte("bar:200|c|#tag:value\nbar:200|c|#tag:\xc3\x28invalid"), events)
+		}
+		close(events)
+	}()
 
-func (h *MockHistogram) Observe(n float64) {
-	h.value = n
+	ex := NewExporter(&mapper.MetricMapper{})
+	ex.Listen(events)
 }
 
 func TestHistogramUnits(t *testing.T) {
-	events := make(chan Events, 1)
+	// Start exporter with a synchronous channel
+	events := make(chan Events, 0)
+	go func() {
+		ex := NewExporter(&mapper.MetricMapper{})
+		ex.mapper.Defaults.TimerType = mapper.TimerTypeHistogram
+		ex.Listen(events)
+	}()
+
+	// Synchronously send a statsd event to wait for handleEvent execution.
+	// Then close events channel to stop a listener.
 	name := "foo"
 	c := Events{
 		&TimerEvent{
@@ -98,22 +100,22 @@ func TestHistogramUnits(t *testing.T) {
 		},
 	}
 	events <- c
-	ex := NewExporter(&mapper.MetricMapper{})
-	ex.mapper.Defaults.TimerType = mapper.TimerTypeHistogram
+	events <- Events{}
+	close(events)
 
-	// Close channel to signify we are done with the listener after a short period.
-	go func() {
-		time.Sleep(time.Millisecond * 100)
-		close(events)
-	}()
-	mock := &MockHistogram{}
-	key := hashNameAndLabels(name, nil)
-	ex.Histograms.Elements[key] = mock
-	ex.Listen(events)
-	if mock.value == 300 {
+	// Check histogram value
+	metrics, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatalf("Cannot gather from DefaultGatherer: %v", err)
+	}
+	value := getFloat64(metrics, name, prometheus.Labels{})
+	if value == nil {
+		t.Fatal("Histogram value should not be nil")
+	}
+	if *value == 300 {
 		t.Fatalf("Histogram observations not scaled into Seconds")
-	} else if mock.value != .300 {
-		t.Fatalf("Received unexpected value for histogram observation %f != .300", mock.value)
+	} else if *value != .300 {
+		t.Fatalf("Received unexpected value for histogram observation %f != .300", *value)
 	}
 }
 
@@ -173,3 +175,183 @@ func TestEscapeMetricName(t *testing.T) {
 		}
 	}
 }
+
+// TestTtlExpiration validates expiration of time series.
+// foobar metric without mapping should expire with default ttl of 1s
+// bazqux metric should expire with ttl of 2s
+func TestTtlExpiration(t *testing.T) {
+	// Mock a time.NewTicker
+	tickerCh := make(chan time.Time, 0)
+	clock.ClockInstance = &clock.Clock{
+		TickerCh: tickerCh,
+	}
+
+	config := `
+defaults:
+  ttl: 1s
+mappings:
+- match: bazqux.*
+  name: bazqux
+  ttl: 2s
+`
+	// Create mapper from config and start an Exporter with a synchronous channel
+	testMapper := &mapper.MetricMapper{}
+	err := testMapper.InitFromYAMLString(config)
+	if err != nil {
+		t.Fatalf("Config load error: %s %s", config, err)
+	}
+	events := make(chan Events, 0)
+	defer close(events)
+	go func() {
+		ex := NewExporter(testMapper)
+		ex.Listen(events)
+	}()
+
+	ev := Events{
+		// event with default ttl = 1s
+		&GaugeEvent{
+			metricName: "foobar",
+			value:      200,
+		},
+		// event with ttl = 2s from a mapping
+		&TimerEvent{
+			metricName: "bazqux.main",
+			value:      42,
+		},
+	}
+
+	var metrics []*dto.MetricFamily
+	var foobarValue *float64
+	var bazquxValue *float64
+
+	// Step 1. Send events with statsd metrics.
+	// Send empty Events to wait for events are handled.
+	// saveLabelValues will use fake instant as a lastRegisteredAt time.
+	clock.ClockInstance.Instant = time.Unix(0, 0)
+	events <- ev
+	events <- Events{}
+
+	// Check values
+	metrics, err = prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatal("Gather should not fail")
+	}
+	foobarValue = getFloat64(metrics, "foobar", prometheus.Labels{})
+	bazquxValue = getFloat64(metrics, "bazqux", prometheus.Labels{})
+	if foobarValue == nil || bazquxValue == nil {
+		t.Fatalf("Gauge `foobar` and Summary `bazqux` should be gathered")
+	}
+	if *foobarValue != 200 {
+		t.Fatalf("Gauge `foobar` observation %f is not expected. Should be 200", *foobarValue)
+	}
+	if *bazquxValue != 42 {
+		t.Fatalf("Summary `bazqux` observation %f is not expected. Should be 42", *bazquxValue)
+	}
+
+	// Step 2. Increase Instant to emulate metrics expiration after 1s
+	clock.ClockInstance.Instant = time.Unix(1, 10)
+	clock.ClockInstance.TickerCh <- time.Unix(0, 0)
+	events <- Events{}
+
+	// Check values
+	metrics, err = prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatal("Gather should not fail")
+	}
+	foobarValue = getFloat64(metrics, "foobar", prometheus.Labels{})
+	bazquxValue = getFloat64(metrics, "bazqux", prometheus.Labels{})
+	if foobarValue != nil {
+		t.Fatalf("Gauge `foobar` should be expired")
+	}
+	if bazquxValue == nil {
+		t.Fatalf("Summary `bazqux` should be gathered")
+	}
+	if *bazquxValue != 42 {
+		t.Fatalf("Summary `bazqux` observation %f is not expected. Should be 42", *bazquxValue)
+	}
+
+	// Step 3. Increase Instant to emulate metrics expiration after 2s
+	clock.ClockInstance.Instant = time.Unix(2, 200)
+	clock.ClockInstance.TickerCh <- time.Unix(0, 0)
+	events <- Events{}
+
+	// Check values
+	metrics, err = prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatal("Gather should not fail")
+	}
+	foobarValue = getFloat64(metrics, "foobar", prometheus.Labels{})
+	bazquxValue = getFloat64(metrics, "bazqux", prometheus.Labels{})
+	if bazquxValue != nil {
+		t.Fatalf("Summary `bazqux` should be expired")
+	}
+	if foobarValue != nil {
+		t.Fatalf("Gauge `foobar` should not be gathered after expiration")
+	}
+}
+
+// getFloat64 search for metric by name in array of MetricFamily and then search a value by labels.
+// Method returns a value or nil if metric is not found.
+func getFloat64(metrics []*dto.MetricFamily, name string, labels prometheus.Labels) *float64 {
+	var metricFamily *dto.MetricFamily
+	for _, m := range metrics {
+		if *m.Name == name {
+			metricFamily = m
+			break
+		}
+	}
+	if metricFamily == nil {
+		return nil
+	}
+
+	var metric *dto.Metric
+	labelsHash := hashNameAndLabels(name, labels)
+	for _, m := range metricFamily.Metric {
+		h := hashNameAndLabels(name, labelPairsAsLabels(m.GetLabel()))
+		if h == labelsHash {
+			metric = m
+			break
+		}
+	}
+	if metric == nil {
+		return nil
+	}
+
+	var value float64
+	if metric.Gauge != nil {
+		value = metric.Gauge.GetValue()
+		return &value
+	}
+	if metric.Counter != nil {
+		value = metric.Counter.GetValue()
+		return &value
+	}
+	if metric.Histogram != nil {
+		value = metric.Histogram.GetSampleSum()
+		return &value
+	}
+	if metric.Summary != nil {
+		value = metric.Summary.GetSampleSum()
+		return &value
+	}
+	if metric.Untyped != nil {
+		value = metric.Untyped.GetValue()
+		return &value
+	}
+	panic(fmt.Errorf("collected a non-gauge/counter/histogram/summary/untyped metric: %s", metric))
+}
+
+func labelPairsAsLabels(pairs []*dto.LabelPair) (labels prometheus.Labels) {
+	labels = prometheus.Labels{}
+	for _, pair := range pairs {
+		if pair.Name == nil {
+			continue
+		}
+		value := ""
+		if pair.Value != nil {
+			value = *pair.Value
+		}
+		labels[*pair.Name] = value
+	}
+	return
+}
diff --git a/pkg/mapper/mapper_test.go b/pkg/mapper/mapper_test.go
index 08e4306..8d1ea49 100644
--- a/pkg/mapper/mapper_test.go
+++ b/pkg/mapper/mapper_test.go
@@ -15,6 +15,7 @@ package mapper
 
 import (
 	"testing"
+	"time"
 )
 
 type mappings map[string]struct {
@@ -22,6 +23,7 @@ type mappings map[string]struct {
 	labels     map[string]string
 	quantiles  []metricObjective
 	notPresent bool
+	ttl        time.Duration
 }
 
 func TestMetricMapperYAML(t *testing.T) {
@@ -603,6 +605,66 @@ mappings:
 				},
 			},
 		},
+		// Config that has a ttl.
+		{
+			config: `mappings:
+- match: web.*
+  name: "web"
+  ttl: 10s
+  labels:
+    site: "$1"`,
+			mappings: mappings{
+				"test.a": {},
+				"web.localhost": {
+					name: "web",
+					labels: map[string]string{
+						"site": "localhost",
+					},
+					ttl: time.Second * 10,
+				},
+			},
+		},
+		// Config that has a default ttl.
+		{
+			config: `defaults:
+  ttl: 1m2s
+mappings:
+- match: web.*
+  name: "web"
+  labels:
+    site: "$1"`,
+			mappings: mappings{
+				"test.a": {},
+				"web.localhost": {
+					name: "web",
+					labels: map[string]string{
+						"site": "localhost",
+					},
+					ttl: time.Minute + time.Second*2,
+				},
+			},
+		},
+		// Config that override a default ttl.
+		{
+			config: `defaults:
+  ttl: 1m2s
+mappings:
+- match: web.*
+  name: "web"
+  ttl: 5s
+  labels:
+    site: "$1"`,
+			mappings: mappings{
+				"test.a": {},
+				"web.localhost": {
+					name: "web",
+					labels: map[string]string{
+						"site": "localhost",
+					},
+					ttl: time.Second * 5,
+				},
+			},
+		},
 	}
 
 	mapper := MetricMapper{}
@@ -633,6 +695,9 @@ mappings:
 					t.Fatalf("%d.%q: Expected labels %v, got %v", i, metric, mapping, labels)
 				}
 			}
+			if mapping.ttl > 0 && mapping.ttl != m.Ttl {
+				t.Fatalf("%d.%q: Expected ttl of %s, got %s", i, metric, mapping.ttl.String(), m.Ttl.String())
+			}
 
 			if len(mapping.quantiles) != 0 {
 				if len(mapping.quantiles) != len(m.Quantiles) {