feat: RunOnce / Serverless Telegraf

agent/agent.go

@@ -142,6 +142,95 @@ func (a *Agent) Run(ctx context.Context) error {
 	return nil
 }
 
+// Run starts and runs only once
+func (a *Agent) RunOnce(ctx context.Context) error {
+	log.Printf("I! [agent] Config: Interval:%s, Quiet:%#v, Hostname:%#v, "+
+		"Flush Interval:%s",
+		a.Config.Agent.Interval.Duration, a.Config.Agent.Quiet,
+		a.Config.Agent.Hostname, a.Config.Agent.FlushInterval.Duration)
+
+	if ctx.Err() != nil {
+		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)
+	if err != nil {
+		return err
+	}
+
+	inputC := make(chan telegraf.Metric, 100)
+	procC := make(chan telegraf.Metric, 100)
+	outputC := make(chan telegraf.Metric, 100)
+
+	startTime := time.Now()
+
+	log.Printf("D! [agent] Starting service inputs")
+	err = a.startServiceInputs(ctx, inputC)
+	if err != nil {
+		return err
+	}
+
+	src := inputC
+	dst := inputC
+
+	err = a.runInputs(ctx, startTime, dst)
+	if err != nil {
+		log.Printf("E! [agent] Error running inputs: %v", err)
+	}
+
+	log.Printf("D! [agent] Stopping service inputs")
+	a.stopServiceInputs()
+
+	close(dst)
+	log.Printf("D! [agent] Input channel closed")
+
+	src = dst
+
+	if len(a.Config.Processors) > 0 {
+		dst = procC
+
+		err = a.runProcessors(src, dst)
+		if err != nil {
+			log.Printf("E! [agent] Error running processors: %v", err)
+		}
+		close(dst)
+		log.Printf("D! [agent] Processor channel closed")
+
+		src = dst
+	}
+
+	if len(a.Config.Aggregators) > 0 {
+		dst = outputC
+
+		err = a.runAggregators(startTime, src, dst)
+		if err != nil {
+			log.Printf("E! [agent] Error running aggregators: %v", err)
+		}
+		close(dst)
+		log.Printf("D! [agent] Output channel closed")
+
+		src = dst
+	}
+
+	err = a.runOutputs(startTime, src)
+	if err != nil {
+		log.Printf("E! [agent] Error running outputs: %v", err)
+	}
+
+	log.Printf("D! [agent] Closing outputs")
+	a.closeOutputs()
+
+	log.Printf("D! [agent] Stopped Successfully")
+	return nil
+}
+
 // Test runs the inputs once and prints the output to stdout in line protocol.
 func (a *Agent) Test(ctx context.Context, waitDuration time.Duration) error {
 	var wg sync.WaitGroup
@@ -272,6 +361,14 @@ func (a *Agent) runInputs(
 		go func(input *models.RunningInput) {
 			defer wg.Done()
 
+			if a.Config.Agent.RunOnce {
+				err := a.gatherOnce(acc, input, interval)
+				if err != nil {
+					acc.AddError(err)
+				}
+				return
+			}
+
 			if a.Config.Agent.RoundInterval {
 				err := internal.SleepContext(
 					ctx, internal.AlignDuration(startTime, interval))
@@ -517,6 +614,23 @@ func (a *Agent) runOutputs(
 		wg.Add(1)
 		go func(output *models.RunningOutput) {
 			defer wg.Done()
+
+			if a.Config.Agent.RunOnce {
+				err := a.flushOnce(output, interval, output.Write)
+				if err != nil {
+					fmt.Println("Dead")
+				}
+				return
+			}
+
+			if !a.Config.Agent.RoundInterval {
+				err := internal.SleepContext(
+					ctx, internal.AlignDuration(startTime, interval))
+				if err != nil {
+					return
+				}
+			}
+
 			a.flushLoop(ctx, startTime, output, interval, jitter)
 		}(output)
 	}

cmd/telegraf/telegraf.go

@@ -67,6 +67,7 @@ var fServiceDisplayName = flag.String("service-display-name", "Telegraf Data Col
 var fRunAsConsole = flag.Bool("console", false, "run as console application (windows only)")
 var fPlugins = flag.String("plugin-directory", "",
 	"path to directory containing external plugins")
+var fRunOnce = flag.Bool("run-once", false, "collect metrics from inputs once and exit")
 
 var (
 	version string
@@ -151,6 +152,12 @@ func runAgent(ctx context.Context,
 			c.Agent.Interval.Duration)
 	}
 
+	// Command-line argument takes precedence over configuration file
+	if *fRunOnce {
+		c.Agent.RunOnce = true
+
+	}
+
 	ag, err := agent.NewAgent(c)
 	if err != nil {
 		return err
@@ -198,6 +205,10 @@ func runAgent(ctx context.Context,
 		}
 	}
 
+	if ag.Config.Agent.RunOnce {
+		return ag.RunOnce(ctx)
+	}
+
 	return ag.Run(ctx)
 }
 

internal/config/config.go

@@ -77,6 +77,7 @@ func NewConfig() *Config {
 			FlushInterval:              internal.Duration{Duration: 10 * time.Second},
 			LogTarget:                  "file",
 			LogfileRotationMaxArchives: 5,
+			RunOnce:                    false,
 		},
 
 		Tags:          make(map[string]string),
@@ -168,6 +169,9 @@ type AgentConfig struct {
 	// If set to -1, no archives are removed.
 	LogfileRotationMaxArchives int `toml:"logfile_rotation_max_archives"`
 
+	// Run the collection once, then exit
+	RunOnce bool `toml:"run_once"`
+
 	Hostname     string
 	OmitHostname bool
 }