Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Refactor default labels and retention metrics

pull/10616/head
Edd Robinson 2018-11-07 16:44:25 +00:00
parent 6c5dec8f88
commit 8ca637bd80
4 changed files with 56 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
storage/engine.go
View File

@ -39,6 +39,8 @@ type Engine struct {
wal *tsm1.WAL
retentionEnforcer *retentionEnforcer
defaultMetricLabels prometheus.Labels
// Tracks all goroutines started by the Engine.
wg sync.WaitGroup
@ -61,6 +63,7 @@ func WithTSMFilenameFormatter(fn tsm1.FormatFileNameFunc) Option {
func WithEngineID(id int) Option {
return func(e *Engine) {
e.engineID = &id
e.defaultMetricLabels["engine_id"] = fmt.Sprint(*e.engineID)
}
}
@ -69,6 +72,7 @@ func WithEngineID(id int) Option {
func WithNodeID(id int) Option {
return func(e *Engine) {
e.nodeID = &id
e.defaultMetricLabels["node_id"] = fmt.Sprint(*e.nodeID)
}
}
@ -78,17 +82,6 @@ func WithNodeID(id int) Option {
func WithRetentionEnforcer(finder BucketFinder) Option {
return func(e *Engine) {
e.retentionEnforcer = newRetentionEnforcer(e, finder)
if e.engineID != nil {
e.retentionEnforcer.defaultMetricLabels["engine_id"] = fmt.Sprint(*e.engineID)
}
if e.nodeID != nil {
e.retentionEnforcer.defaultMetricLabels["node_id"] = fmt.Sprint(*e.nodeID)
}
// As new labels may have been set, set the new metrics on the enforcer.
e.retentionEnforcer.retentionMetrics = newRetentionMetrics(e.retentionEnforcer.defaultMetricLabels)
}
}
@ -110,9 +103,11 @@ func WithCompactionPlanner(planner tsm1.CompactionPlanner) Option {
// TSM engine.
func NewEngine(path string, c Config, options ...Option) *Engine {
e := &Engine{
config: c,
path: path,
logger: zap.NewNop(),
config: c,
path: path,
sfile: tsdb.NewSeriesFile(c.GetSeriesFilePath(path)),
defaultMetricLabels: prometheus.Labels{},
logger: zap.NewNop(),
}
// Initialize series file.
@ -140,6 +135,9 @@ func NewEngine(path string, c Config, options ...Option) *Engine {
for _, option := range options {
option(e)
}
// Set default metrics labels.
e.engine.WithDefaultMetricLabels(e.defaultMetricLabels)
return e
}
@ -197,6 +195,7 @@ func (e *Engine) Open() error {
e.engine.SetCompactionsEnabled(true) // TODO(edd):is this needed?
e.closing = make(chan struct{})
// TODO(edd) background tasks will be run in priority order via a scheduler.
// For now we will just run on an interval as we only have the retention
// policy enforcer.
@ -221,6 +220,11 @@ func (e *Engine) runRetentionEnforcer() {
return
}
if e.retentionEnforcer != nil {
// Set default metric labels on retention enforcer.
e.retentionEnforcer.metrics = newRetentionMetrics(e.defaultMetricLabels)
}
l := e.logger.With(zap.String("component", "retention_enforcer"), logger.DurationLiteral("check_interval", interval))
l.Info("Starting")

18
storage/metrics.go
View File

@ -1,6 +1,10 @@
package storage
import "github.com/prometheus/client_golang/prometheus"
import (
"sort"
"github.com/prometheus/client_golang/prometheus"
)
// namespace is the leading part of all published metrics for the Storage service.
const namespace = "storage"
@ -9,6 +13,7 @@ const retentionSubsystem = "retention" // sub-system associated with metrics for
// retentionMetrics is a set of metrics concerned with tracking data about retention policies.
type retentionMetrics struct {
labels prometheus.Labels
Checks *prometheus.CounterVec
CheckDuration *prometheus.HistogramVec
Unprocessable *prometheus.CounterVec
@ -20,6 +25,8 @@ func newRetentionMetrics(labels prometheus.Labels) *retentionMetrics {
for k := range labels {
names = append(names, k)
}
names = append(names, "status") // All metrics include status
sort.Strings(names)
return &retentionMetrics{
Checks: prometheus.NewCounterVec(prometheus.CounterOpts{
@ -54,6 +61,15 @@ func newRetentionMetrics(labels prometheus.Labels) *retentionMetrics {
}
}
// Labels returns a copy of labels for use with retention metrics.
func (m *retentionMetrics) Labels() prometheus.Labels {
l := make(map[string]string, len(m.labels))
for k, v := range m.labels {
l[k] = v
}
return l
}
// PrometheusCollectors satisfies the prom.PrometheusCollector interface.
func (rm *retentionMetrics) PrometheusCollectors() []prometheus.Collector {
return []prometheus.Collector{

41
storage/retention.go
View File

@ -48,8 +48,7 @@ type retentionEnforcer struct {
logger *zap.Logger
retentionMetrics *retentionMetrics
defaultMetricLabels prometheus.Labels // N.B this must not be mutated after Open is called.
metrics *retentionMetrics
}
// newRetentionEnforcer returns a new enforcer that ensures expired data is
@ -57,24 +56,14 @@ type retentionEnforcer struct {
// disabling the service.
func newRetentionEnforcer(engine Deleter, bucketService BucketFinder) *retentionEnforcer {
s := &retentionEnforcer{
Engine: engine,
BucketService: bucketService,
logger: zap.NewNop(),
defaultMetricLabels: prometheus.Labels{"status": ""},
Engine: engine,
BucketService: bucketService,
logger: zap.NewNop(),
}
s.retentionMetrics = newRetentionMetrics(s.defaultMetricLabels)
s.metrics = newRetentionMetrics(nil)
return s
}
// metricLabels returns a new copy of the default metric labels.
func (s *retentionEnforcer) metricLabels() prometheus.Labels {
labels := make(map[string]string, len(s.defaultMetricLabels))
for k, v := range s.defaultMetricLabels {
labels[k] = v
}
return labels
}
// WithLogger sets the logger l on the service. It must be called before Open.
func (s *retentionEnforcer) WithLogger(l *zap.Logger) {
if s == nil {
@ -96,15 +85,15 @@ func (s *retentionEnforcer) run() {
}
now := time.Now().UTC()
labels := s.metricLabels()
labels := s.metrics.Labels()
labels["status"] = "ok"
if err := s.expireData(rpByBucketID, now); err != nil {
log.Error("Deletion not successful", zap.Error(err))
labels["status"] = "error"
}
s.retentionMetrics.CheckDuration.With(labels).Observe(time.Since(now).Seconds())
s.retentionMetrics.Checks.With(labels).Inc()
s.metrics.CheckDuration.With(labels).Observe(time.Since(now).Seconds())
s.metrics.Checks.With(labels).Inc()
}
// expireData runs a delete operation on the storage engine.
@ -162,21 +151,21 @@ func (s *retentionEnforcer) expireData(rpByBucketID map[platform.ID]time.Duratio
}
defer func() {
if s.retentionMetrics == nil {
if s.metrics == nil {
return
}
labels := s.metricLabels()
labels := s.metrics.Labels()
labels["status"] = "bad_measurement"
s.retentionMetrics.Unprocessable.With(labels).Add(float64(len(badMSketch)))
s.metrics.Unprocessable.With(labels).Add(float64(len(badMSketch)))
labels["status"] = "missing_bucket"
s.retentionMetrics.Unprocessable.With(labels).Add(float64(len(missingBSketch)))
s.metrics.Unprocessable.With(labels).Add(float64(len(missingBSketch)))
labels["status"] = "ok"
s.retentionMetrics.Series.With(labels).Add(float64(atomic.LoadUint64(&seriesDeleted)))
s.metrics.Series.With(labels).Add(float64(atomic.LoadUint64(&seriesDeleted)))
labels["status"] = "skipped"
s.retentionMetrics.Series.With(labels).Add(float64(atomic.LoadUint64(&seriesSkipped)))
s.metrics.Series.With(labels).Add(float64(atomic.LoadUint64(&seriesSkipped)))
}()
return s.Engine.DeleteSeriesRangeWithPredicate(newSeriesIteratorAdapter(cur), fn)
@ -200,7 +189,7 @@ func (s *retentionEnforcer) getRetentionPeriodPerBucket() (map[platform.ID]time.
// PrometheusCollectors satisfies the prom.PrometheusCollector interface.
func (s *retentionEnforcer) PrometheusCollectors() []prometheus.Collector {
return s.retentionMetrics.PrometheusCollectors()
return s.metrics.PrometheusCollectors()
}
// A BucketService is an platform.BucketService that the retentionEnforcer can open,

6
tsdb/tsm1/engine.go
View File

@ -250,6 +250,12 @@ func (e *Engine) WithCompactionPlanner(planner CompactionPlanner) {
e.CompactionPlan = planner
}
// WithDefaultMetricLabels sets the default labels for metrics on the engine.
// It must be called before the Engine is opened.
func (e *Engine) WithDefaultMetricLabels(labels prometheus.Labels) {
e.defaultMetricLabels = labels
}
// SetEnabled sets whether the engine is enabled.
func (e *Engine) SetEnabled(enabled bool) {
e.enableCompactionsOnOpen = enabled