Add call to optional Init function for all plugins (#5899)

agent/agent.go

@@ -39,8 +39,14 @@ func (a *Agent) Run(ctx context.Context) error {
 		return ctx.Err()
 	}
 
+	log.Printf("D! [agent] Initializing plugins")
+	err := a.initPlugins()
+	if err != nil {
+		return err
+	}
+
 	log.Printf("D! [agent] Connecting outputs")
-	err := a.connectOutputs(ctx)
+	err = a.connectOutputs(ctx)
 	if err != nil {
 		return err
 	}
@@ -185,6 +191,11 @@ func (a *Agent) Test(ctx context.Context, waitDuration time.Duration) error {
 	}
 
 	for _, input := range a.Config.Inputs {
+		err := input.Init()
+		if err != nil {
+			return err
+		}
+
 		select {
 		case <-ctx.Done():
 			return nil
@@ -596,6 +607,39 @@ func (a *Agent) flushOnce(
 
 }
 
+// initPlugins runs the Init function on plugins.
+func (a *Agent) initPlugins() error {
+	for _, input := range a.Config.Inputs {
+		err := input.Init()
+		if err != nil {
+			return fmt.Errorf("could not initialize input %s: %v",
+				input.Config.Name, err)
+		}
+	}
+	for _, processor := range a.Config.Processors {
+		err := processor.Init()
+		if err != nil {
+			return fmt.Errorf("could not initialize processor %s: %v",
+				processor.Config.Name, err)
+		}
+	}
+	for _, aggregator := range a.Config.Aggregators {
+		err := aggregator.Init()
+		if err != nil {
+			return fmt.Errorf("could not initialize aggregator %s: %v",
+				aggregator.Config.Name, err)
+		}
+	}
+	for _, output := range a.Config.Outputs {
+		err := output.Init()
+		if err != nil {
+			return fmt.Errorf("could not initialize output %s: %v",
+				output.Config.Name, err)
+		}
+	}
+	return nil
+}
+
 // connectOutputs connects to all outputs.
 func (a *Agent) connectOutputs(ctx context.Context) error {
 	for _, output := range a.Config.Outputs {

docs/AGGREGATORS.md

@@ -52,6 +52,10 @@ var sampleConfig = `
   drop_original = false
 `
 
+func (m *Min) Init() error {
+	return nil
+}
+
 func (m *Min) SampleConfig() string {
 	return sampleConfig
 }

docs/INPUTS.md

@@ -52,6 +52,10 @@ func (s *Simple) SampleConfig() string {
 `
 }
 
+func (s *Simple) Init() error {
+	return nil
+}
+
 func (s *Simple) Gather(acc telegraf.Accumulator) error {
     if s.Ok {
         acc.AddFields("state", map[string]interface{}{"value": "pretty good"}, nil)

docs/OUTPUTS.md

@@ -43,6 +43,10 @@ func (s *Simple) SampleConfig() string {
 `
 }
 
+func (s *Simple) Init() error {
+	return nil
+}
+
 func (s *Simple) Connect() error {
     // Make a connection to the URL here
     return nil

docs/PROCESSORS.md

@@ -46,6 +46,10 @@ func (p *Printer) Description() string {
 	return "Print all metrics that pass through this filter."
 }
 
+func (p *Printer) Init() error {
+	return nil
+}
+
 func (p *Printer) Apply(in ...telegraf.Metric) []telegraf.Metric {
 	for _, metric := range in {
 		fmt.Println(metric.String())

input.go

@@ -1,5 +1,14 @@
 package telegraf
 
+// Initializer is an interface that all plugin types: Inputs, Outputs,
+// Processors, and Aggregators can optionally implement to initialize the
+// plugin.
+type Initializer interface {
+	// Init performs one time setup of the plugin and returns an error if the
+	// configuration is invalid.
+	Init() error
+}
+
 type Input interface {
 	// SampleConfig returns the default configuration of the Input
 	SampleConfig() string

internal/models/running_aggregator.go

@@ -71,6 +71,16 @@ func (r *RunningAggregator) Name() string {
 	return "aggregators." + r.Config.Name
 }
 
+func (r *RunningAggregator) Init() error {
+	if p, ok := r.Aggregator.(telegraf.Initializer); ok {
+		err := p.Init()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (r *RunningAggregator) Period() time.Duration {
 	return r.Config.Period
 }

internal/models/running_input.go

@@ -56,6 +56,16 @@ func (r *RunningInput) metricFiltered(metric telegraf.Metric) {
 	metric.Drop()
 }
 
+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)

internal/models/running_output.go

@@ -97,6 +97,16 @@ func (ro *RunningOutput) metricFiltered(metric telegraf.Metric) {
 	metric.Drop()
 }
 
+func (ro *RunningOutput) Init() error {
+	if p, ok := ro.Output.(telegraf.Initializer); ok {
+		err := p.Init()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // AddMetric adds a metric to the output.
 //
 // Takes ownership of metric

internal/models/running_processor.go

@@ -40,6 +40,16 @@ func containsMetric(item telegraf.Metric, metrics []telegraf.Metric) bool {
 	return false
 }
 
+func (rp *RunningProcessor) Init() error {
+	if p, ok := rp.Processor.(telegraf.Initializer); ok {
+		err := p.Init()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (rp *RunningProcessor) Apply(in ...telegraf.Metric) []telegraf.Metric {
 	rp.Lock()
 	defer rp.Unlock()

plugins/inputs/http/http.go

@@ -1,7 +1,6 @@
 package http
 
 import (
-	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -89,18 +88,12 @@ func (*HTTP) Description() string {
 	return "Read formatted metrics from one or more HTTP endpoints"
 }
 
-// Gather takes in an accumulator and adds the metrics that the Input
+func (h *HTTP) Init() error {
-// gathers. This is called every "interval"
-func (h *HTTP) Gather(acc telegraf.Accumulator) error {
-	if h.parser == nil {
-		return errors.New("Parser is not set")
-	}
-
-	if h.client == nil {
 	tlsCfg, err := h.ClientConfig.TLSConfig()
 	if err != nil {
 		return err
 	}
+
 	h.client = &http.Client{
 		Transport: &http.Transport{
 			TLSClientConfig: tlsCfg,
@@ -108,8 +101,12 @@ func (h *HTTP) Gather(acc telegraf.Accumulator) error {
 		},
 		Timeout: h.Timeout.Duration,
 	}
+	return nil
 }
 
+// Gather takes in an accumulator and adds the metrics that the Input
+// gathers. This is called every "interval"
+func (h *HTTP) Gather(acc telegraf.Accumulator) error {
 	var wg sync.WaitGroup
 	for _, u := range h.URLs {
 		wg.Add(1)

plugins/inputs/http/http_test.go

@@ -37,6 +37,7 @@ func TestHTTPwithJSONFormat(t *testing.T) {
 	plugin.SetParser(p)
 
 	var acc testutil.Accumulator
+	plugin.Init()
 	require.NoError(t, acc.GatherError(plugin.Gather))
 
 	require.Len(t, acc.Metrics, 1)
@@ -78,6 +79,7 @@ func TestHTTPHeaders(t *testing.T) {
 	plugin.SetParser(p)
 
 	var acc testutil.Accumulator
+	plugin.Init()
 	require.NoError(t, acc.GatherError(plugin.Gather))
 }
 
@@ -100,6 +102,7 @@ func TestInvalidStatusCode(t *testing.T) {
 	plugin.SetParser(p)
 
 	var acc testutil.Accumulator
+	plugin.Init()
 	require.Error(t, acc.GatherError(plugin.Gather))
 }
 
@@ -125,28 +128,10 @@ func TestMethod(t *testing.T) {
 	plugin.SetParser(p)
 
 	var acc testutil.Accumulator
+	plugin.Init()
 	require.NoError(t, acc.GatherError(plugin.Gather))
 }
 
-func TestParserNotSet(t *testing.T) {
-	fakeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path == "/endpoint" {
-			_, _ = w.Write([]byte(simpleJSON))
-		} else {
-			w.WriteHeader(http.StatusNotFound)
-		}
-	}))
-	defer fakeServer.Close()
-
-	url := fakeServer.URL + "/endpoint"
-	plugin := &plugin.HTTP{
-		URLs: []string{url},
-	}
-
-	var acc testutil.Accumulator
-	require.Error(t, acc.GatherError(plugin.Gather))
-}
-
 const simpleJSON = `
 {
     "a": 1.2
@@ -237,6 +222,7 @@ func TestBodyAndContentEncoding(t *testing.T) {
 			tt.plugin.SetParser(parser)
 
 			var acc testutil.Accumulator
+			tt.plugin.Init()
 			err = tt.plugin.Gather(&acc)
 			require.NoError(t, err)
 		})