telegraf/internal/models/makemetric.go

package models

import (
	"log"
	"math"
	"time"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/metric"
)

// makemetric is used by both RunningAggregator & RunningInput
// to make metrics.
//   nameOverride: override the name of the measurement being made.
//   namePrefix:   add this prefix to each measurement name.
//   nameSuffix:   add this suffix to each measurement name.
//   pluginTags:   these are tags that are specific to this plugin.
//   daemonTags:   these are daemon-wide global tags, and get applied after pluginTags.
//   filter:       this is a filter to apply to each metric being made.
//   applyFilter:  if false, the above filter is not applied to each metric.
//                 This is used by Aggregators, because aggregators use filters
//                 on incoming metrics instead of on created metrics.
// TODO refactor this to not have such a huge func signature.
func makemetric(
	measurement string,
	fields map[string]interface{},
	tags map[string]string,
	nameOverride string,
	namePrefix string,
	nameSuffix string,
	pluginTags map[string]string,
	daemonTags map[string]string,
	filter Filter,
	applyFilter bool,
	mType telegraf.ValueType,
	t time.Time,
) telegraf.Metric {
	if len(fields) == 0 || len(measurement) == 0 {
		return nil
	}
	if tags == nil {
		tags = make(map[string]string)
	}

	// Override measurement name if set
	if len(nameOverride) != 0 {
		measurement = nameOverride
	}
	// Apply measurement prefix and suffix if set
	if len(namePrefix) != 0 {
		measurement = namePrefix + measurement
	}
	if len(nameSuffix) != 0 {
		measurement = measurement + nameSuffix
	}

	// Apply plugin-wide tags if set
	for k, v := range pluginTags {
		if _, ok := tags[k]; !ok {
			tags[k] = v
		}
	}
	// Apply daemon-wide tags if set
	for k, v := range daemonTags {
		if _, ok := tags[k]; !ok {
			tags[k] = v
		}
	}

	// Apply the metric filter(s)
	// for aggregators, the filter does not get applied when the metric is made.
	// instead, the filter is applied to metric incoming into the plugin.
	//   ie, it gets applied in the RunningAggregator.Apply function.
	if applyFilter {
		if ok := filter.Apply(measurement, fields, tags); !ok {
			return nil
		}
	}

	for k, v := range fields {
		// Validate uint64 and float64 fields
		// convert all int & uint types to int64
		switch val := v.(type) {
		case nil:
			// delete nil fields
			delete(fields, k)
		case uint:
			fields[k] = int64(val)
			continue
		case uint8:
			fields[k] = int64(val)
			continue
		case uint16:
			fields[k] = int64(val)
			continue
		case uint32:
			fields[k] = int64(val)
			continue
		case int:
			fields[k] = int64(val)
			continue
		case int8:
			fields[k] = int64(val)
			continue
		case int16:
			fields[k] = int64(val)
			continue
		case int32:
			fields[k] = int64(val)
			continue
		case uint64:
			// InfluxDB does not support writing uint64
			if val < uint64(9223372036854775808) {
				fields[k] = int64(val)
			} else {
				fields[k] = int64(9223372036854775807)
			}
			continue
		case float32:
			fields[k] = float64(val)
			continue
		case float64:
			// NaNs are invalid values in influxdb, skip measurement
			if math.IsNaN(val) || math.IsInf(val, 0) {
				log.Printf("D! Measurement [%s] field [%s] has a NaN or Inf "+
					"field, skipping",
					measurement, k)
				delete(fields, k)
				continue
			}
		default:
			fields[k] = v
		}
	}

	m, err := metric.New(measurement, tags, fields, t, mType)
	if err != nil {
		log.Printf("Error adding point [%s]: %s\n", measurement, err.Error())
		return nil
	}

	return m
}