telegraf/internal/models/running_input.go

119 lines
2.4 KiB
Go
Raw Normal View History

package models
import (
"time"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/selfstat"
)
var GlobalMetricsGathered = selfstat.Register("agent", "metrics_gathered", map[string]string{})
type RunningInput struct {
Input telegraf.Input
Config *InputConfig
log telegraf.Logger
defaultTags map[string]string
MetricsGathered selfstat.Stat
GatherTime selfstat.Stat
}
func NewRunningInput(input telegraf.Input, config *InputConfig) *RunningInput {
tags := map[string]string{"input": config.Name}
if config.Alias != "" {
tags["alias"] = config.Alias
}
logger := &Logger{
Name: logName("inputs", config.Name, config.Alias),
Errs: selfstat.Register("gather", "errors", tags),
}
setLogIfExist(input, logger)
return &RunningInput{
Input: input,
Config: config,
MetricsGathered: selfstat.Register(
"gather",
"metrics_gathered",
tags,
),
GatherTime: selfstat.RegisterTiming(
"gather",
"gather_time_ns",
tags,
),
log: logger,
}
}
// InputConfig is the common config for all inputs.
type InputConfig struct {
Name string
Alias string
Interval time.Duration
NameOverride string
MeasurementPrefix string
MeasurementSuffix string
Tags map[string]string
Filter Filter
}
func (r *RunningInput) metricFiltered(metric telegraf.Metric) {
metric.Drop()
}
func (r *RunningInput) LogName() string {
return logName("inputs", r.Config.Name, r.Config.Alias)
}
func (r *RunningInput) Init() error {
if p, ok := r.Input.(telegraf.Initializer); ok {
err := p.Init()
if err != nil {
return err
}
}
return nil
}
func (r *RunningInput) MakeMetric(metric telegraf.Metric) telegraf.Metric {
if ok := r.Config.Filter.Select(metric); !ok {
r.metricFiltered(metric)
return nil
}
m := makemetric(
metric,
r.Config.NameOverride,
r.Config.MeasurementPrefix,
r.Config.MeasurementSuffix,
r.Config.Tags,
r.defaultTags)
r.Config.Filter.Modify(metric)
if len(metric.FieldList()) == 0 {
r.metricFiltered(metric)
return nil
}
r.MetricsGathered.Incr(1)
GlobalMetricsGathered.Incr(1)
return m
}
func (r *RunningInput) Gather(acc telegraf.Accumulator) error {
start := time.Now()
err := r.Input.Gather(acc)
elapsed := time.Since(start)
r.GatherTime.Incr(elapsed.Nanoseconds())
return err
}
func (r *RunningInput) SetDefaultTags(tags map[string]string) {
r.defaultTags = tags
}