Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for aggregations. #34

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add support for aggregations. #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 66 additions & 1 deletion collectors/monitoring_collector.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"math"
"strings"
"sync"
"time"

Expand All @@ -20,6 +21,10 @@ type MonitoringCollector struct {
metricsTypePrefixes []string
metricsInterval time.Duration
metricsOffset time.Duration
metricsAggFields []string
metricsAggReducer string
metricsAggAligner string
metricsAggAlignDuration time.Duration
monitoringService *monitoring.Service
apiCallsTotalMetric prometheus.Counter
scrapesTotalMetric prometheus.Counter
Expand All @@ -35,6 +40,10 @@ func NewMonitoringCollector(
metricsInterval time.Duration,
metricsOffset time.Duration,
monitoringService *monitoring.Service,
metricsAggFields []string,
metricsAggReducer string,
metricsAggAligner string,
metricsAggAlignDuration time.Duration,
) (*MonitoringCollector, error) {
apiCallsTotalMetric := prometheus.NewCounter(
prometheus.CounterOpts{
Expand Down Expand Up @@ -101,6 +110,10 @@ func NewMonitoringCollector(
metricsTypePrefixes: metricsTypePrefixes,
metricsInterval: metricsInterval,
metricsOffset: metricsOffset,
metricsAggFields: metricsAggFields,
metricsAggReducer: metricsAggReducer,
metricsAggAligner: metricsAggAligner,
metricsAggAlignDuration: metricsAggAlignDuration,
monitoringService: monitoringService,
apiCallsTotalMetric: apiCallsTotalMetric,
scrapesTotalMetric: scrapesTotalMetric,
Expand Down Expand Up @@ -168,6 +181,14 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri
Filter(fmt.Sprintf("metric.type=\"%s\"", metricDescriptor.Type)).
IntervalStartTime(startTime.Format(time.RFC3339Nano)).
IntervalEndTime(endTime.Format(time.RFC3339Nano))
if len(c.metricsAggFields) != 0 {
timeSeriesListCall = timeSeriesListCall.
AggregationGroupByFields(c.metricsAggFields...).
AggregationCrossSeriesReducer(c.metricsAggReducer).
AggregationPerSeriesAligner(c.metricsAggAligner).
AggregationAlignmentPeriod(fmt.Sprintf("%ds", c.metricsAggAlignDuration/time.Second))
fmt.Printf("duration: %ds", c.metricsAggAlignDuration/time.Second)
}

for {
c.apiCallsTotalMetric.Inc()
Expand Down Expand Up @@ -263,12 +284,56 @@ func (c *MonitoringCollector) reportTimeSeriesMetrics(
labelValues = append(labelValues, value)
}

name := prometheus.BuildFQName("stackdriver", utils.NormalizeMetricName(timeSeries.Resource.Type), utils.NormalizeMetricName(timeSeries.Metric.Type))
if len(c.metricsAggFields) > 0 {
var parts []string
for _, f := range c.metricsAggFields {
subfields := strings.Split(f, ".")
parts = append(parts, subfields[len(subfields)-1])
}
prefix := strings.Join(parts, "_")

var suffix string
switch c.metricsAggReducer {
case "REDUCE_SUM":
suffix = "sum"
case "REDUCE_MEAN":
suffix = "avg"
case "REDUCE_MIN":
suffix = "min"
case "REDUCE_MAX":
suffix = "max"
case "REDUCE_STDDEV":
suffix = "stddex"
case "REDUCE_COUNT":
suffix = "count"
case "REDUCE_COUNT_TRUE":
suffix = "countrue"
case "REDUCE_COUNT_FALSE":
suffix = "countfalse"
case "REDUCE_FRACTION_TRUE":
suffix = "fractrue"
case "REDUCE_PERCENTILE_99":
suffix = "perc99"
case "REDUCE_PERCENTILE_95":
suffix = "perc95"
case "REDUCE_PERCENTILE_50":
suffix = "perc50"
case "REDUCE_PERCENTILE_05":
suffix = "perc05"
default:
suffix = c.metricsAggReducer
}

name = fmt.Sprintf("%s:%s:%s", prefix, name, suffix)
}

// The metric name to report is composed by the 3 parts:
// 1. namespace is a constant prefix (stackdriver)
// 2. subsystem is the monitored resource type (ie gce_instance)
// 3. name is the metric type (ie compute.googleapis.com/instance/cpu/usage_time)
metricDesc = prometheus.NewDesc(
prometheus.BuildFQName("stackdriver", utils.NormalizeMetricName(timeSeries.Resource.Type), utils.NormalizeMetricName(timeSeries.Metric.Type)),
name,
metricDescriptor.Description,
labelKeys,
prometheus.Labels{},
Expand Down
33 changes: 32 additions & 1 deletion stackdriver_exporter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,22 @@ var (
"monitoring.metrics-type-prefixes", "Comma separated Google Stackdriver Monitoring Metric Type prefixes ($STACKDRIVER_EXPORTER_MONITORING_METRICS_TYPE_PREFIXES).",
).Envar("STACKDRIVER_EXPORTER_MONITORING_METRICS_TYPE_PREFIXES").Required().String()

monitoringMetricsAggFields = kingpin.Flag(
"monitoring.metrics-aggregation-fields", "Comma separated list of Stackdriver fields to aggregate over ($STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_FIELDS).",
).Envar("STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_FIELDS").Default("").String()

monitoringMetricsAggReducer = kingpin.Flag(
"monitoring.metrics-aggregation-reducer", "Stackdriver aggregation cross-series reducer ($STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_REDUCER).",
).Envar("STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_REDUCER").Default("REDUCE_SUM").String()

monitoringMetricsAggAligner = kingpin.Flag(
"monitoring.metrics-aggregation-aligner", "Stackdriver aggregation per-series aligner ($STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_ALIGNER).",
).Envar("STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_ALIGNER").Default("ALIGN_DELTA").String()

monitoringMetricsAggAlignDuration = kingpin.Flag(
"monitoring.metrics-aggregation-align-duration", "Stackdriver aggregation alignment duration ($STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_ALIGN_DURATION).",
).Envar("STACKDRIVER_EXPORTER_MONITORING_METRICS_AGGREGATION_ALIGN_DURATION").Default("1m").Duration()

monitoringMetricsInterval = kingpin.Flag(
"monitoring.metrics-interval", "Interval to request the Google Stackdriver Monitoring Metrics for. Only the most recent data point is used ($STACKDRIVER_EXPORTER_MONITORING_METRICS_INTERVAL).",
).Envar("STACKDRIVER_EXPORTER_MONITORING_METRICS_INTERVAL").Default("5m").Duration()
Expand Down Expand Up @@ -89,7 +105,22 @@ func main() {
os.Exit(1)
}

monitoringCollector, err := collectors.NewMonitoringCollector(*projectID, metricsTypePrefixes, *monitoringMetricsInterval, *monitoringMetricsOffset, monitoringService)
var metricsAggFields []string
if *monitoringMetricsAggFields != "" {
metricsAggFields = strings.Split(*monitoringMetricsAggFields, ",")
}

monitoringCollector, err := collectors.NewMonitoringCollector(
*projectID,
metricsTypePrefixes,
*monitoringMetricsInterval,
*monitoringMetricsOffset,
monitoringService,
metricsAggFields,
*monitoringMetricsAggReducer,
*monitoringMetricsAggAligner,
*monitoringMetricsAggAlignDuration)

if err != nil {
log.Error(err)
os.Exit(1)
Expand Down