Merge pull request #15005 from influxdata/fix/deadman-check

fix(notification/check): remove the aggregateWindow from the deadman query

notification/check/deadman.go

@@ -45,6 +45,7 @@ func (c Deadman) GenerateFlux() (string, error) {
 // an error for each error found when the script is parsed.
 func (c Deadman) GenerateFluxAST() (*ast.Package, error) {
 	p := parser.ParseSource(c.Query.Text)
+	removeAggregateWindow(p)
 	replaceDurationsWithEvery(p, c.StaleTime)
 	removeStopFromRange(p)
 

notification/check/deadman_test.go

@@ -21,6 +21,65 @@ func TestDeadman_GenerateFlux(t *testing.T) {
 		args  args
 		wants wants
 	}{
+		{
+			name: "with aggregateWindow",
+			args: args{
+				deadman: check.Deadman{
+					Base: check.Base{
+						ID:   10,
+						Name: "moo",
+						Tags: []influxdb.Tag{
+							{Key: "aaa", Value: "vaaa"},
+							{Key: "bbb", Value: "vbbb"},
+						},
+						Every:                 mustDuration("1h"),
+						StatusMessageTemplate: "whoa! {r.dead}",
+						Query: influxdb.DashboardQuery{
+							Text: `from(bucket: "foo") |> range(start: -1d, stop: now()) |> aggregateWindow(fn: mean, every: 1m) |> yield()`,
+							BuilderConfig: influxdb.BuilderConfig{
+								Tags: []struct {
+									Key    string   `json:"key"`
+									Values []string `json:"values"`
+								}{
+									{
+										Key:    "_field",
+										Values: []string{"usage_user"},
+									},
+								},
+							},
+						},
+					},
+					TimeSince: mustDuration("60s"),
+					StaleTime: mustDuration("10m"),
+					Level:     notification.Info,
+				},
+			},
+			wants: wants{
+				script: `package main
+import "influxdata/influxdb/monitor"
+import "experimental"
+
+data = from(bucket: "foo")
+	|> range(start: -10m)
+
+option task = {name: "moo", every: 1h}
+
+check = {
+	_check_id: "000000000000000a",
+	_check_name: "moo",
+	_type: "deadman",
+	tags: {aaa: "vaaa", bbb: "vbbb"},
+}
+info = (r) =>
+	(r.dead)
+messageFn = (r) =>
+	("whoa! {r.dead}")
+
+data
+	|> monitor.deadman(t: experimental.subDuration(from: now(), d: 60s))
+	|> monitor.check(data: check, messageFn: messageFn, info: info)`,
+			},
+		},
 		{
 			name: "basic",
 			args: args{

notification/check/threshold.go

@@ -208,6 +208,20 @@ func removeStopFromRange(pkg *ast.Package) {
 	})
 }
 
+// TODO(desa): we'll likely want to remove all other arguments to range that are provided, but for now this should work.
+// When we decide to implement the full feature we'll have to do something more sophisticated.
+func removeAggregateWindow(pkg *ast.Package) {
+	ast.Visit(pkg, func(n ast.Node) {
+		if pipe, ok := n.(*ast.PipeExpression); ok {
+			if id, ok := pipe.Call.Callee.(*ast.Identifier); ok && id.Name == "aggregateWindow" {
+				if subPipe, ok := pipe.Argument.(*ast.PipeExpression); ok {
+					*pipe = *subPipe
+				}
+			}
+		}
+	})
+}
+
 func assignPipelineToData(f *ast.File) error {
 	if len(f.Body) != 1 {
 		return fmt.Errorf("expected there to be a single statement in the flux script body, recieved %d", len(f.Body))