|
| 1 | +// Copyright 2019 Google LLC |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +package retrieval |
| 16 | + |
| 17 | +import ( |
| 18 | + "context" |
| 19 | + "math" |
| 20 | + |
| 21 | + "github.com/go-kit/kit/log" |
| 22 | + "github.com/go-kit/kit/log/level" |
| 23 | + promlabels "github.com/prometheus/prometheus/pkg/labels" |
| 24 | + "github.com/prometheus/tsdb/labels" |
| 25 | + "go.opencensus.io/stats" |
| 26 | + "go.opencensus.io/stats/view" |
| 27 | +) |
| 28 | + |
| 29 | +// CounterAggregator provides the 'aggregated counters' feature of the sidecar. |
| 30 | +// It can be used to export a sum of multiple counters from Prometheus to |
| 31 | +// Stackdriver as a single cumulative metric. |
| 32 | +// Each aggregated counter is associated with a single OpenCensus counter that |
| 33 | +// can then be exported to Stackdriver (as a CUMULATIVE metric) or exposed to |
| 34 | +// Prometheus via the standard `/metrics` endpoint. Regular flushing of counter |
| 35 | +// values is implemented by OpenCensus. |
| 36 | +type CounterAggregator struct { |
| 37 | + logger log.Logger |
| 38 | + counters []*aggregatedCounter |
| 39 | + statsRecord func(context.Context, ...stats.Measurement) // used in testing. |
| 40 | +} |
| 41 | + |
| 42 | +// aggregatedCounter is where CounterAggregator keeps internal state about each |
| 43 | +// exported metric: OpenCensus measure and view as well as a list of Matchers that |
| 44 | +// define which Prometheus metrics will get aggregated. |
| 45 | +type aggregatedCounter struct { |
| 46 | + measure *stats.Float64Measure |
| 47 | + view *view.View |
| 48 | + matchers [][]*promlabels.Matcher |
| 49 | +} |
| 50 | + |
| 51 | +// CounterAggregatorConfig contains configuration for CounterAggregator. Keys of the map |
| 52 | +// are metric names that will be exported by counter aggregator. |
| 53 | +type CounterAggregatorConfig map[string]*CounterAggregatorMetricConfig |
| 54 | + |
| 55 | +// CounterAggregatorMetricConfig provides configuration of a single aggregated counter. |
| 56 | +// Matchers specify what Prometheus metrics (which are expected to be counter metrics) will |
| 57 | +// be re-aggregated. Help provides a description for the exported metric. |
| 58 | +type CounterAggregatorMetricConfig struct { |
| 59 | + Matchers [][]*promlabels.Matcher |
| 60 | + Help string |
| 61 | +} |
| 62 | + |
| 63 | +// counterTracker keeps track of a single time series that has at least one aggregated |
| 64 | +// counter associated with it (i.e. there is at least one aggregated counter that has |
| 65 | +// Matchers covering this time series). Last timestamp and value are tracked |
| 66 | +// to detect counter resets. |
| 67 | +type counterTracker struct { |
| 68 | + lastTimestamp int64 |
| 69 | + lastValue float64 |
| 70 | + measures []*stats.Float64Measure |
| 71 | + ca *CounterAggregator |
| 72 | +} |
| 73 | + |
| 74 | +// NewCounterAggregator creates a counter aggregator. |
| 75 | +func NewCounterAggregator(logger log.Logger, config *CounterAggregatorConfig) (*CounterAggregator, error) { |
| 76 | + aggregator := &CounterAggregator{logger: logger, statsRecord: stats.Record} |
| 77 | + for metric, cfg := range *config { |
| 78 | + measure := stats.Float64(metric, cfg.Help, stats.UnitDimensionless) |
| 79 | + v := &view.View{ |
| 80 | + Name: metric, |
| 81 | + Description: cfg.Help, |
| 82 | + Measure: measure, |
| 83 | + Aggregation: view.Sum(), |
| 84 | + } |
| 85 | + if err := view.Register(v); err != nil { |
| 86 | + return nil, err |
| 87 | + } |
| 88 | + aggregator.counters = append(aggregator.counters, &aggregatedCounter{ |
| 89 | + measure: measure, |
| 90 | + view: v, |
| 91 | + matchers: cfg.Matchers, |
| 92 | + }) |
| 93 | + } |
| 94 | + return aggregator, nil |
| 95 | +} |
| 96 | + |
| 97 | +// Close must be called when CounterAggregator is no longer needed. |
| 98 | +func (c *CounterAggregator) Close() { |
| 99 | + for _, counter := range c.counters { |
| 100 | + view.Unregister(counter.view) |
| 101 | + } |
| 102 | +} |
| 103 | + |
| 104 | +// getTracker returns a counterTracker for a specific time series defined by labelset. |
| 105 | +// If `nil` is returned, it means that there are no aggregated counters that need to |
| 106 | +// be incremented for this time series. |
| 107 | +func (c *CounterAggregator) getTracker(lset labels.Labels) *counterTracker { |
| 108 | + var measures []*stats.Float64Measure |
| 109 | + for _, counter := range c.counters { |
| 110 | + if matchFiltersets(lset, counter.matchers) { |
| 111 | + measures = append(measures, counter.measure) |
| 112 | + } |
| 113 | + } |
| 114 | + if len(measures) == 0 { |
| 115 | + return nil |
| 116 | + } |
| 117 | + return &counterTracker{measures: measures, ca: c} |
| 118 | +} |
| 119 | + |
| 120 | +// newPoint gets called on each new sample (timestamp, value) for time series that need to feed |
| 121 | +// values into aggregated counters. |
| 122 | +func (t *counterTracker) newPoint(ctx context.Context, lset labels.Labels, ts int64, v float64) { |
| 123 | + if math.IsNaN(v) { |
| 124 | + level.Debug(t.ca.logger).Log("msg", "got NaN value", "labels", lset, "last ts", t.lastTimestamp, "ts", t, "lastValue", t.lastValue) |
| 125 | + return |
| 126 | + } |
| 127 | + // Ignore measurements that are earlier than last seen timestamp, since they are already covered by |
| 128 | + // later values. Samples are coming from TSDB in order, so this is unlikely to happen. |
| 129 | + if ts < t.lastTimestamp { |
| 130 | + level.Debug(t.ca.logger).Log("msg", "out of order timestamp", "labels", lset, "last ts", t.lastTimestamp, "ts", ts) |
| 131 | + return |
| 132 | + } |
| 133 | + // Use the first value we see as the starting point for the counter. |
| 134 | + if t.lastTimestamp == 0 { |
| 135 | + level.Debug(t.ca.logger).Log("msg", "first point", "labels", lset) |
| 136 | + t.lastTimestamp = ts |
| 137 | + t.lastValue = v |
| 138 | + return |
| 139 | + } |
| 140 | + var delta float64 |
| 141 | + if v < t.lastValue { |
| 142 | + // Counter was reset. |
| 143 | + delta = v |
| 144 | + level.Debug(t.ca.logger).Log("msg", "counter reset", "labels", lset, "value", v, "lastValue", t.lastValue, "delta", delta) |
| 145 | + } else { |
| 146 | + delta = v - t.lastValue |
| 147 | + level.Debug(t.ca.logger).Log("msg", "got delta", "labels", lset, "value", v, "lastValue", t.lastValue, "delta", delta) |
| 148 | + } |
| 149 | + t.lastTimestamp = ts |
| 150 | + t.lastValue = v |
| 151 | + if delta == 0 { |
| 152 | + return |
| 153 | + } |
| 154 | + ms := make([]stats.Measurement, len(t.measures)) |
| 155 | + for i, measure := range t.measures { |
| 156 | + ms[i] = measure.M(delta) |
| 157 | + } |
| 158 | + t.ca.statsRecord(ctx, ms...) |
| 159 | +} |
0 commit comments