694 lines
19 KiB
Go
694 lines
19 KiB
Go
//go:generate ../../../tools/readme_config_includer/generator
|
|
package stackdriver
|
|
|
|
import (
|
|
"context"
|
|
_ "embed"
|
|
"errors"
|
|
"fmt"
|
|
"hash/fnv"
|
|
"path"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
|
|
monitoring "cloud.google.com/go/monitoring/apiv3/v2"
|
|
"cloud.google.com/go/monitoring/apiv3/v2/monitoringpb"
|
|
"google.golang.org/api/option"
|
|
"google.golang.org/genproto/googleapis/api/distribution"
|
|
metricpb "google.golang.org/genproto/googleapis/api/metric"
|
|
monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
|
|
"google.golang.org/grpc/status"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
|
|
"github.com/influxdata/telegraf"
|
|
"github.com/influxdata/telegraf/filter"
|
|
"github.com/influxdata/telegraf/internal"
|
|
"github.com/influxdata/telegraf/plugins/outputs"
|
|
)
|
|
|
|
//go:embed sample.conf
|
|
var sampleConfig string
|
|
|
|
// Stackdriver is the Google Stackdriver config info.
|
|
type Stackdriver struct {
|
|
Project string `toml:"project"`
|
|
Namespace string `toml:"namespace"`
|
|
ResourceType string `toml:"resource_type"`
|
|
ResourceLabels map[string]string `toml:"resource_labels"`
|
|
MetricTypePrefix string `toml:"metric_type_prefix"`
|
|
MetricNameFormat string `toml:"metric_name_format"`
|
|
MetricDataType string `toml:"metric_data_type"`
|
|
TagsAsResourceLabels []string `toml:"tags_as_resource_label"`
|
|
MetricCounter []string `toml:"metric_counter"`
|
|
MetricGauge []string `toml:"metric_gauge"`
|
|
MetricHistogram []string `toml:"metric_histogram"`
|
|
Log telegraf.Logger `toml:"-"`
|
|
|
|
client *monitoring.MetricClient
|
|
counterCache *counterCache
|
|
filterCounter filter.Filter
|
|
filterGauge filter.Filter
|
|
filterHistogram filter.Filter
|
|
}
|
|
|
|
const (
|
|
// The user-defined limits are documented below:
|
|
// https://cloud.google.com/monitoring/quotas#custom_metrics_quotas
|
|
|
|
// QuotaLabelsPerMetricDescriptor is the limit
|
|
// to labels (tags) per metric descriptor.
|
|
QuotaLabelsPerMetricDescriptor = 30
|
|
// QuotaStringLengthForLabelKey is the limit
|
|
// to string length for label key.
|
|
QuotaStringLengthForLabelKey = 100
|
|
// QuotaStringLengthForLabelValue is the limit
|
|
// to string length for label value.
|
|
QuotaStringLengthForLabelValue = 1024
|
|
|
|
// MaxInt is the max int64 value.
|
|
MaxInt = int(^uint(0) >> 1)
|
|
)
|
|
|
|
func (s *Stackdriver) Init() error {
|
|
if s.MetricTypePrefix == "" {
|
|
s.MetricTypePrefix = "custom.googleapis.com"
|
|
}
|
|
|
|
switch s.MetricNameFormat {
|
|
case "":
|
|
s.MetricNameFormat = "path"
|
|
case "path", "official":
|
|
default:
|
|
return fmt.Errorf("unrecognized metric name format: %s", s.MetricNameFormat)
|
|
}
|
|
|
|
switch s.MetricDataType {
|
|
case "":
|
|
s.MetricDataType = "source"
|
|
case "source", "double":
|
|
default:
|
|
return fmt.Errorf("unrecognized metric data type: %s", s.MetricDataType)
|
|
}
|
|
|
|
var err error
|
|
s.filterCounter, err = filter.Compile(s.MetricCounter)
|
|
if err != nil {
|
|
return fmt.Errorf("creating counter filter failed: %w", err)
|
|
}
|
|
s.filterGauge, err = filter.Compile(s.MetricGauge)
|
|
if err != nil {
|
|
return fmt.Errorf("creating gauge filter failed: %w", err)
|
|
}
|
|
s.filterHistogram, err = filter.Compile(s.MetricHistogram)
|
|
if err != nil {
|
|
return fmt.Errorf("creating histogram filter failed: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (*Stackdriver) SampleConfig() string {
|
|
return sampleConfig
|
|
}
|
|
|
|
// Connect initiates the primary connection to the GCP project.
|
|
func (s *Stackdriver) Connect() error {
|
|
if s.Project == "" {
|
|
return errors.New("project is a required field for stackdriver output")
|
|
}
|
|
|
|
if s.Namespace == "" {
|
|
s.Log.Warn("plugin-level namespace is empty")
|
|
}
|
|
|
|
if s.ResourceType == "" {
|
|
s.ResourceType = "global"
|
|
}
|
|
|
|
if s.ResourceLabels == nil {
|
|
s.ResourceLabels = make(map[string]string, 1)
|
|
}
|
|
|
|
if s.counterCache == nil {
|
|
s.counterCache = NewCounterCache(s.Log)
|
|
}
|
|
|
|
s.ResourceLabels["project_id"] = s.Project
|
|
|
|
if s.client == nil {
|
|
ctx := context.Background()
|
|
client, err := monitoring.NewMetricClient(ctx, option.WithUserAgent(internal.ProductToken()))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.client = client
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Sorted returns a copy of the metrics in time ascending order. A copy is
|
|
// made to avoid modifying the input metric slice since doing so is not
|
|
// allowed.
|
|
func sorted(metrics []telegraf.Metric) []telegraf.Metric {
|
|
batch := make([]telegraf.Metric, 0, len(metrics))
|
|
for i := len(metrics) - 1; i >= 0; i-- {
|
|
batch = append(batch, metrics[i])
|
|
}
|
|
sort.Slice(batch, func(i, j int) bool {
|
|
return batch[i].Time().Before(batch[j].Time())
|
|
})
|
|
return batch
|
|
}
|
|
|
|
type timeSeriesBuckets map[uint64][]*monitoringpb.TimeSeries
|
|
|
|
func (tsb timeSeriesBuckets) Add(m telegraf.Metric, f []*telegraf.Field, ts *monitoringpb.TimeSeries) {
|
|
h := fnv.New64a()
|
|
h.Write([]byte(m.Name()))
|
|
h.Write([]byte{'\n'})
|
|
for _, field := range f {
|
|
h.Write([]byte(field.Key))
|
|
h.Write([]byte{'\n'})
|
|
}
|
|
for key, value := range m.Tags() {
|
|
h.Write([]byte(key))
|
|
h.Write([]byte{'\n'})
|
|
h.Write([]byte(value))
|
|
h.Write([]byte{'\n'})
|
|
}
|
|
k := h.Sum64()
|
|
|
|
s := tsb[k]
|
|
s = append(s, ts)
|
|
tsb[k] = s
|
|
}
|
|
|
|
// Split metrics up by timestamp and send to Google Cloud Stackdriver
|
|
func (s *Stackdriver) Write(metrics []telegraf.Metric) error {
|
|
metricBatch := make(map[int64][]telegraf.Metric)
|
|
timestamps := make([]int64, 0, len(metrics))
|
|
for _, metric := range sorted(metrics) {
|
|
timestamp := metric.Time().UnixNano()
|
|
if existingSlice, ok := metricBatch[timestamp]; ok {
|
|
metricBatch[timestamp] = append(existingSlice, metric)
|
|
} else {
|
|
metricBatch[timestamp] = []telegraf.Metric{metric}
|
|
timestamps = append(timestamps, timestamp)
|
|
}
|
|
}
|
|
|
|
// sort the timestamps we collected
|
|
sort.Slice(timestamps, func(i, j int) bool { return timestamps[i] < timestamps[j] })
|
|
|
|
s.Log.Debugf("received %d metrics\n", len(metrics))
|
|
s.Log.Debugf("split into %d groups by timestamp\n", len(metricBatch))
|
|
for _, timestamp := range timestamps {
|
|
if err := s.sendBatch(metricBatch[timestamp]); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Write the metrics to Google Cloud Stackdriver.
|
|
func (s *Stackdriver) sendBatch(batch []telegraf.Metric) error {
|
|
ctx := context.Background()
|
|
|
|
buckets := make(timeSeriesBuckets)
|
|
for _, m := range batch {
|
|
// Set metric types based on user-provided filter
|
|
metricType := m.Type()
|
|
if s.filterCounter != nil && s.filterCounter.Match(m.Name()) {
|
|
metricType = telegraf.Counter
|
|
}
|
|
if s.filterGauge != nil && s.filterGauge.Match(m.Name()) {
|
|
metricType = telegraf.Gauge
|
|
}
|
|
if s.filterHistogram != nil && s.filterHistogram.Match(m.Name()) {
|
|
metricType = telegraf.Histogram
|
|
}
|
|
|
|
metricKind, err := getStackdriverMetricKind(metricType)
|
|
if err != nil {
|
|
s.Log.Errorf("Get kind for metric %q (%T) failed: %s", m.Name(), metricType, err)
|
|
continue
|
|
}
|
|
|
|
// Convert any declared tag to a resource label and remove it from
|
|
// the metric
|
|
resourceLabels := make(map[string]string, len(s.ResourceLabels)+len(s.TagsAsResourceLabels))
|
|
for k, v := range s.ResourceLabels {
|
|
resourceLabels[k] = v
|
|
}
|
|
for _, tag := range s.TagsAsResourceLabels {
|
|
if val, ok := m.GetTag(tag); ok {
|
|
resourceLabels[tag] = val
|
|
m.RemoveTag(tag)
|
|
}
|
|
}
|
|
|
|
if m.Type() == telegraf.Histogram {
|
|
value, err := buildHistogram(m)
|
|
if err != nil {
|
|
s.Log.Errorf("Unable to build distribution from metric %s: %s", m, err)
|
|
continue
|
|
}
|
|
|
|
startTime, endTime := getStackdriverIntervalEndpoints(metricKind, value, m, nil, s.counterCache)
|
|
timeInterval, err := getStackdriverTimeInterval(metricKind, startTime, endTime)
|
|
if err != nil {
|
|
s.Log.Errorf("Get time interval failed: %s", err)
|
|
continue
|
|
}
|
|
|
|
// Prepare an individual data point.
|
|
dataPoint := &monitoringpb.Point{
|
|
Interval: timeInterval,
|
|
Value: value,
|
|
}
|
|
|
|
// Prepare time series.
|
|
timeSeries := &monitoringpb.TimeSeries{
|
|
Metric: &metricpb.Metric{
|
|
Type: s.generateHistogramName(m),
|
|
Labels: s.getStackdriverLabels(m.TagList()),
|
|
},
|
|
MetricKind: metricKind,
|
|
Resource: &monitoredrespb.MonitoredResource{
|
|
Type: s.ResourceType,
|
|
Labels: resourceLabels,
|
|
},
|
|
Points: []*monitoringpb.Point{
|
|
dataPoint,
|
|
},
|
|
}
|
|
|
|
buckets.Add(m, m.FieldList(), timeSeries)
|
|
continue
|
|
}
|
|
|
|
for _, f := range m.FieldList() {
|
|
value, err := s.getStackdriverTypedValue(f.Value)
|
|
if err != nil {
|
|
s.Log.Errorf("Get type failed: %q", err)
|
|
continue
|
|
}
|
|
if value == nil {
|
|
continue
|
|
}
|
|
|
|
startTime, endTime := getStackdriverIntervalEndpoints(metricKind, value, m, f, s.counterCache)
|
|
timeInterval, err := getStackdriverTimeInterval(metricKind, startTime, endTime)
|
|
if err != nil {
|
|
s.Log.Errorf("Get time interval failed: %s", err)
|
|
continue
|
|
}
|
|
|
|
// Prepare an individual data point.
|
|
dataPoint := &monitoringpb.Point{
|
|
Interval: timeInterval,
|
|
Value: value,
|
|
}
|
|
|
|
// Prepare time series.
|
|
timeSeries := &monitoringpb.TimeSeries{
|
|
Metric: &metricpb.Metric{
|
|
Type: s.generateMetricName(m, metricType, f.Key),
|
|
Labels: s.getStackdriverLabels(m.TagList()),
|
|
},
|
|
MetricKind: metricKind,
|
|
Resource: &monitoredrespb.MonitoredResource{
|
|
Type: s.ResourceType,
|
|
Labels: resourceLabels,
|
|
},
|
|
Points: []*monitoringpb.Point{
|
|
dataPoint,
|
|
},
|
|
}
|
|
|
|
buckets.Add(m, []*telegraf.Field{f}, timeSeries)
|
|
|
|
// If the metric is untyped, it will end with unknown. We will also
|
|
// send another metric with the unknown:counter suffix. Google will
|
|
// do some heuristics to know which one to use for queries. This
|
|
// only occurs when using the official name format.
|
|
if s.MetricNameFormat == "official" && strings.HasSuffix(timeSeries.Metric.Type, "unknown") {
|
|
metricKind := metricpb.MetricDescriptor_CUMULATIVE
|
|
startTime, endTime := getStackdriverIntervalEndpoints(metricKind, value, m, f, s.counterCache)
|
|
timeInterval, err := getStackdriverTimeInterval(metricKind, startTime, endTime)
|
|
if err != nil {
|
|
s.Log.Errorf("Get time interval failed: %s", err)
|
|
continue
|
|
}
|
|
dataPoint := &monitoringpb.Point{
|
|
Interval: timeInterval,
|
|
Value: value,
|
|
}
|
|
|
|
counterTimeSeries := &monitoringpb.TimeSeries{
|
|
Metric: &metricpb.Metric{
|
|
Type: s.generateMetricName(m, metricType, f.Key) + ":counter",
|
|
Labels: s.getStackdriverLabels(m.TagList()),
|
|
},
|
|
MetricKind: metricpb.MetricDescriptor_CUMULATIVE,
|
|
Resource: &monitoredrespb.MonitoredResource{
|
|
Type: s.ResourceType,
|
|
Labels: resourceLabels,
|
|
},
|
|
Points: []*monitoringpb.Point{
|
|
dataPoint,
|
|
},
|
|
}
|
|
buckets.Add(m, []*telegraf.Field{f}, counterTimeSeries)
|
|
}
|
|
}
|
|
}
|
|
|
|
// process the buckets in order
|
|
keys := make([]uint64, 0, len(buckets))
|
|
for k := range buckets {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Slice(keys, func(i, j int) bool { return keys[i] < keys[j] })
|
|
|
|
for len(buckets) != 0 {
|
|
// can send up to 200 time series to stackdriver
|
|
timeSeries := make([]*monitoringpb.TimeSeries, 0, 200)
|
|
for i := 0; i < len(keys) && len(timeSeries) < cap(timeSeries); i++ {
|
|
k := keys[i]
|
|
s := buckets[k]
|
|
timeSeries = append(timeSeries, s[0])
|
|
if len(s) == 1 {
|
|
delete(buckets, k)
|
|
keys = append(keys[:i], keys[i+1:]...)
|
|
i--
|
|
continue
|
|
}
|
|
|
|
s = s[1:]
|
|
buckets[k] = s
|
|
}
|
|
|
|
// Prepare time series request.
|
|
timeSeriesRequest := &monitoringpb.CreateTimeSeriesRequest{
|
|
Name: "projects/" + s.Project,
|
|
TimeSeries: timeSeries,
|
|
}
|
|
|
|
// Create the time series in Stackdriver.
|
|
err := s.client.CreateTimeSeries(ctx, timeSeriesRequest)
|
|
if err != nil {
|
|
if errStatus, ok := status.FromError(err); ok {
|
|
if errStatus.Code().String() == "InvalidArgument" {
|
|
s.Log.Warnf("Unable to write to Stackdriver - dropping metrics: %s", err)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
s.Log.Errorf("Unable to write to Stackdriver: %s", err)
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Stackdriver) generateMetricName(m telegraf.Metric, metricType telegraf.ValueType, key string) string {
|
|
if s.MetricNameFormat == "path" {
|
|
return path.Join(s.MetricTypePrefix, s.Namespace, m.Name(), key)
|
|
}
|
|
|
|
name := m.Name() + "_" + key
|
|
if s.Namespace != "" {
|
|
name = s.Namespace + "_" + m.Name() + "_" + key
|
|
}
|
|
|
|
var kind string
|
|
switch metricType {
|
|
case telegraf.Gauge:
|
|
kind = "gauge"
|
|
case telegraf.Untyped:
|
|
kind = "unknown"
|
|
case telegraf.Counter:
|
|
kind = "counter"
|
|
case telegraf.Histogram:
|
|
kind = "histogram"
|
|
default:
|
|
kind = ""
|
|
}
|
|
|
|
return path.Join(s.MetricTypePrefix, name, kind)
|
|
}
|
|
|
|
func (s *Stackdriver) generateHistogramName(m telegraf.Metric) string {
|
|
if s.MetricNameFormat == "path" {
|
|
return path.Join(s.MetricTypePrefix, s.Namespace, m.Name())
|
|
}
|
|
|
|
name := m.Name()
|
|
if s.Namespace != "" {
|
|
name = s.Namespace + "_" + m.Name()
|
|
}
|
|
|
|
return path.Join(s.MetricTypePrefix, name, "histogram")
|
|
}
|
|
|
|
func getStackdriverIntervalEndpoints(
|
|
kind metricpb.MetricDescriptor_MetricKind,
|
|
value *monitoringpb.TypedValue,
|
|
m telegraf.Metric,
|
|
f *telegraf.Field,
|
|
cc *counterCache,
|
|
) (start, end *timestamppb.Timestamp) {
|
|
endTime := timestamppb.New(m.Time())
|
|
var startTime *timestamppb.Timestamp
|
|
if kind == metricpb.MetricDescriptor_CUMULATIVE {
|
|
// Interval starts for stackdriver CUMULATIVE metrics must reset any time
|
|
// the counter resets, so we keep a cache of the start times and last
|
|
// observed values for each counter in the batch.
|
|
startTime = cc.GetStartTime(GetCounterCacheKey(m, f), value, endTime)
|
|
}
|
|
return startTime, endTime
|
|
}
|
|
|
|
func getStackdriverTimeInterval(m metricpb.MetricDescriptor_MetricKind, startTime, endTime *timestamppb.Timestamp) (*monitoringpb.TimeInterval, error) {
|
|
switch m {
|
|
case metricpb.MetricDescriptor_GAUGE:
|
|
return &monitoringpb.TimeInterval{
|
|
EndTime: endTime,
|
|
}, nil
|
|
case metricpb.MetricDescriptor_CUMULATIVE:
|
|
return &monitoringpb.TimeInterval{
|
|
StartTime: startTime,
|
|
EndTime: endTime,
|
|
}, nil
|
|
case metricpb.MetricDescriptor_DELTA, metricpb.MetricDescriptor_METRIC_KIND_UNSPECIFIED:
|
|
fallthrough
|
|
default:
|
|
return nil, fmt.Errorf("unsupported metric kind %T", m)
|
|
}
|
|
}
|
|
|
|
func getStackdriverMetricKind(vt telegraf.ValueType) (metricpb.MetricDescriptor_MetricKind, error) {
|
|
switch vt {
|
|
case telegraf.Untyped:
|
|
return metricpb.MetricDescriptor_GAUGE, nil
|
|
case telegraf.Gauge:
|
|
return metricpb.MetricDescriptor_GAUGE, nil
|
|
case telegraf.Counter:
|
|
return metricpb.MetricDescriptor_CUMULATIVE, nil
|
|
case telegraf.Histogram:
|
|
return metricpb.MetricDescriptor_CUMULATIVE, nil
|
|
case telegraf.Summary:
|
|
fallthrough
|
|
default:
|
|
return metricpb.MetricDescriptor_METRIC_KIND_UNSPECIFIED, fmt.Errorf("unsupported telegraf value type: %T", vt)
|
|
}
|
|
}
|
|
|
|
func (s *Stackdriver) getStackdriverTypedValue(value interface{}) (*monitoringpb.TypedValue, error) {
|
|
if s.MetricDataType == "double" {
|
|
v, err := internal.ToFloat64(value)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_DoubleValue{
|
|
DoubleValue: v,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
switch v := value.(type) {
|
|
case uint64:
|
|
if v <= uint64(MaxInt) {
|
|
return &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_Int64Value{
|
|
Int64Value: int64(v),
|
|
},
|
|
}, nil
|
|
}
|
|
return &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_Int64Value{
|
|
Int64Value: int64(MaxInt),
|
|
},
|
|
}, nil
|
|
case int64:
|
|
return &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_Int64Value{
|
|
Int64Value: v,
|
|
},
|
|
}, nil
|
|
case float64:
|
|
return &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_DoubleValue{
|
|
DoubleValue: v,
|
|
},
|
|
}, nil
|
|
case bool:
|
|
return &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_BoolValue{
|
|
BoolValue: v,
|
|
},
|
|
}, nil
|
|
case string:
|
|
// String value types are not available for custom metrics
|
|
return nil, nil
|
|
default:
|
|
return nil, fmt.Errorf("value type \"%T\" not supported for stackdriver custom metrics", v)
|
|
}
|
|
}
|
|
|
|
func buildHistogram(m telegraf.Metric) (*monitoringpb.TypedValue, error) {
|
|
sumInter, ok := m.GetField("sum")
|
|
if !ok {
|
|
return nil, errors.New("no sum field present")
|
|
}
|
|
sum, err := internal.ToFloat64(sumInter)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to convert sum value to float64: %w", err)
|
|
}
|
|
m.RemoveField("sum")
|
|
|
|
countInter, ok := m.GetField("count")
|
|
if !ok {
|
|
return nil, errors.New("no count field present")
|
|
}
|
|
count, err := internal.ToFloat64(countInter)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to convert count value to float64: %w", err)
|
|
}
|
|
m.RemoveField("count")
|
|
|
|
// Build map of the buckets and their values
|
|
buckets := make([]float64, 0)
|
|
bucketCounts := make([]int64, 0)
|
|
for _, field := range m.FieldList() {
|
|
// Add the +inf value to bucket counts, no need to define a bound
|
|
if strings.Contains(strings.ToLower(field.Key), "+inf") {
|
|
count, err := internal.ToInt64(field.Value)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
bucketCounts = append(bucketCounts, count)
|
|
continue
|
|
}
|
|
|
|
bucket, err := strconv.ParseFloat(field.Key, 64)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
count, err := internal.ToInt64(field.Value)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
buckets = append(buckets, bucket)
|
|
bucketCounts = append(bucketCounts, count)
|
|
}
|
|
|
|
sort.Slice(buckets, func(i, j int) bool {
|
|
return buckets[i] < buckets[j]
|
|
})
|
|
sort.Slice(bucketCounts, func(i, j int) bool {
|
|
return bucketCounts[i] < bucketCounts[j]
|
|
})
|
|
|
|
// Bucket counts contain the count for a specific bucket, not the running
|
|
// total like Prometheus histograms use. Loop backwards to determine the
|
|
// count of each bucket rather than the running total count.
|
|
for i := len(bucketCounts) - 1; i > 0; i-- {
|
|
bucketCounts[i] = bucketCounts[i] - bucketCounts[i-1]
|
|
}
|
|
|
|
v := &monitoringpb.TypedValue{
|
|
Value: &monitoringpb.TypedValue_DistributionValue{
|
|
DistributionValue: &distribution.Distribution{
|
|
Count: int64(count),
|
|
Mean: sum / count,
|
|
BucketCounts: bucketCounts,
|
|
BucketOptions: &distribution.Distribution_BucketOptions{
|
|
Options: &distribution.Distribution_BucketOptions_ExplicitBuckets{
|
|
ExplicitBuckets: &distribution.Distribution_BucketOptions_Explicit{
|
|
Bounds: buckets,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
return v, nil
|
|
}
|
|
|
|
func (s *Stackdriver) getStackdriverLabels(tags []*telegraf.Tag) map[string]string {
|
|
labels := make(map[string]string)
|
|
for _, t := range tags {
|
|
labels[t.Key] = t.Value
|
|
}
|
|
for k, v := range labels {
|
|
if len(k) > QuotaStringLengthForLabelKey {
|
|
s.Log.Warnf("Removing tag %q key exceeds string length for label key [%d]", k, QuotaStringLengthForLabelKey)
|
|
delete(labels, k)
|
|
continue
|
|
}
|
|
if len(v) > QuotaStringLengthForLabelValue {
|
|
s.Log.Warnf("Removing tag %q value exceeds string length for label value [%d]", k, QuotaStringLengthForLabelValue)
|
|
delete(labels, k)
|
|
continue
|
|
}
|
|
}
|
|
if len(labels) > QuotaLabelsPerMetricDescriptor {
|
|
excess := len(labels) - QuotaLabelsPerMetricDescriptor
|
|
s.Log.Warnf("Tag count [%d] exceeds quota for stackdriver labels [%d] removing [%d] random tags", len(labels), QuotaLabelsPerMetricDescriptor, excess)
|
|
for k := range labels {
|
|
if excess == 0 {
|
|
break
|
|
}
|
|
excess--
|
|
delete(labels, k)
|
|
}
|
|
}
|
|
|
|
return labels
|
|
}
|
|
|
|
// Close will terminate the session to the backend, returning error if an issue arises.
|
|
func (s *Stackdriver) Close() error {
|
|
return s.client.Close()
|
|
}
|
|
|
|
func newStackdriver() *Stackdriver {
|
|
return &Stackdriver{}
|
|
}
|
|
|
|
func init() {
|
|
outputs.Add("stackdriver", func() telegraf.Output {
|
|
return newStackdriver()
|
|
})
|
|
}
|