Make o365audit input cancellable (#21647)

adriansr · web-flow · commit 1abe97b20d70 · 2020-10-09T09:33:35.000+02:00
PR #21258 introduced a restart mechanism for o365input so that it didn't stop working once a fatal error was found. This updates the restart delay to use a cancellation-context-aware method so that the input doesn't block Filebeat termination.
diff --git a/x-pack/filebeat/input/o365audit/input.go b/x-pack/filebeat/input/o365audit/input.go
@@ -21,14 +21,12 @@ import (
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/o365audit/poll"
 	"github.com/elastic/go-concert/ctxtool"
+	"github.com/elastic/go-concert/timed"
 )
 
 const (
 	pluginName   = "o365audit"
 	fieldsPrefix = pluginName
-
-	// How long to retry when a fatal error is encountered in the input.
-	failureRetryInterval = time.Minute * 5
 )
 
 type o365input struct {
@@ -126,8 +124,8 @@ func (inp *o365input) Run(
 			}
 			publisher.Publish(event, nil)
 			ctx.Logger.Errorf("Input failed: %v", err)
-			ctx.Logger.Infof("Restarting in %v", failureRetryInterval)
-			time.Sleep(failureRetryInterval)
+			ctx.Logger.Infof("Restarting in %v", inp.config.API.ErrorRetryInterval)
+			timed.Wait(ctx.Cancelation, inp.config.API.ErrorRetryInterval)
 		}
 	}
 	return nil

Original file line number	Diff line number	Diff line change
`@@ -21,14 +21,12 @@ import (`
`21`	`21`	`"github.com/elastic/beats/v7/libbeat/logp"`
`22`	`22`	`"github.com/elastic/beats/v7/x-pack/filebeat/input/o365audit/poll"`
`23`	`23`	`"github.com/elastic/go-concert/ctxtool"`
	`24`	`+ "github.com/elastic/go-concert/timed"`
`24`	`25`	`)`
`25`	`26`
`26`	`27`	`const (`
`27`	`28`	`pluginName = "o365audit"`
`28`	`29`	`fieldsPrefix = pluginName`
`29`		`-`
`30`		`- // How long to retry when a fatal error is encountered in the input.`
`31`		`- failureRetryInterval = time.Minute * 5`
`32`	`30`	`)`
`33`	`31`
`34`	`32`	`type o365input struct {`
`@@ -126,8 +124,8 @@ func (inp *o365input) Run(`
`126`	`124`	`}`
`127`	`125`	`publisher.Publish(event, nil)`
`128`	`126`	`ctx.Logger.Errorf("Input failed: %v", err)`
`129`		`- ctx.Logger.Infof("Restarting in %v", failureRetryInterval)`
`130`		`- time.Sleep(failureRetryInterval)`
	`127`	`+ ctx.Logger.Infof("Restarting in %v", inp.config.API.ErrorRetryInterval)`
	`128`	`+ timed.Wait(ctx.Cancelation, inp.config.API.ErrorRetryInterval)`
`131`	`129`	`}`
`132`	`130`	`}`
`133`	`131`	`return nil`