fix(pkger): fix issue with imports causing option task to be injected…

… at wrong point

closes: #17069

CHANGELOG.md

@@ -4,7 +4,8 @@
 
 ### Bug Fixes
 1. [17039](https://github.com/influxdata/influxdb/pull/17039): Fixed issue where tasks are exported for notification rules
-1. [17042](https://github.com/influxdata/influxdb/pull/17039): Fixed issue where tasks are not exported when exporting by org id
+1. [17042](https://github.com/influxdata/influxdb/pull/17042): Fixed issue where tasks are not exported when exporting by org id
+1. [17070](https://github.com/influxdata/influxdb/pull/17070): Fixed issue where tasks with imports in query break in pkger
 
 ## v2.0.0-beta.5 [2020-02-27]
 

cmd/influxd/launcher/pkger_test.go

@@ -635,6 +635,60 @@ spec:
 		})
 	})
 
+	t.Run("apply a task pkg with a complex query", func(t *testing.T) {
+		// validates bug: https://github.com/influxdata/influxdb/issues/17069
+
+		pkgStr := fmt.Sprintf(`
+apiVersion: %[1]s
+kind: Task
+metadata:
+    name: Http.POST Synthetic (POST)
+spec:
+    every: 5m
+    query: |-
+        import "strings"
+        import "csv"
+        import "http"
+        import "system"
+
+        timeDiff = (t1, t2) => {
+        	return duration(v: uint(v: t2) - uint(v: t1))
+        }
+        timeDiffNum = (t1, t2) => {
+        	return uint(v: t2) - uint(v: t1)
+        }
+        urlToPost = "http://www.duckduckgo.com"
+        timeBeforeCall = system.time()
+        responseCode = http.post(url: urlToPost, data: bytes(v: "influxdata"))
+        timeAfterCall = system.time()
+        responseTime = timeDiff(t1: timeBeforeCall, t2: timeAfterCall)
+        responseTimeNum = timeDiffNum(t1: timeBeforeCall, t2: timeAfterCall)
+        data = "#group,false,false,true,true,true,true,true,true
+        #datatype,string,long,string,string,string,string,string,string
+        #default,mean,,,,,,,
+        ,result,table,service,response_code,time_before,time_after,response_time_duration,response_time_ns
+        ,,0,http_post_ping,${string(v: responseCode)},${string(v: timeBeforeCall)},${string(v: timeAfterCall)},${string(v: responseTime)},${string(v: responseTimeNum)}"
+        theTable = csv.from(csv: data)
+
+        theTable
+        	|> map(fn: (r) =>
+        		({r with _time: now()}))
+        	|> map(fn: (r) =>
+        		({r with _measurement: "PingService", url: urlToPost, method: "POST"}))
+        	|> drop(columns: ["time_before", "time_after", "response_time_duration"])
+        	|> to(bucket: "Pingpire", orgID: "039346c3777a1000", fieldFn: (r) =>
+        		({"responseCode": r.response_code, "responseTime": int(v: r.response_time_ns)}))
+`, pkger.APIVersion)
+
+		pkg, err := pkger.Parse(pkger.EncodingYAML, pkger.FromString(pkgStr))
+		require.NoError(t, err)
+
+		sum, err := svc.Apply(timedCtx(time.Second), l.Org.ID, l.User.ID, pkg)
+		require.NoError(t, err)
+
+		require.Len(t, sum.Tasks, 1)
+	})
+
 	t.Run("apply a package with env refs", func(t *testing.T) {
 		pkgStr := fmt.Sprintf(`
 apiVersion: %[1]s

pkger/models.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/url"
 	"reflect"
+	"regexp"
 	"sort"
 	"strconv"
 	"strings"
@@ -1828,6 +1829,8 @@ func (t *task) Status() influxdb.Status {
 	return influxdb.Status(t.status)
 }
 
+var fluxRegex = regexp.MustCompile(`import\s+\".*\"`)
+
 func (t *task) flux() string {
 	taskOpts := []string{fmt.Sprintf("name: %q", t.name)}
 	if t.cron != "" {
@@ -1839,11 +1842,24 @@ func (t *task) flux() string {
 	if t.offset > 0 {
 		taskOpts = append(taskOpts, fmt.Sprintf("offset: %s", t.offset))
 	}
+
 	// this is required by the API, super nasty. Will be super challenging for
 	// anyone outside org to figure out how to do this within an hour of looking
 	// at the API :sadpanda:. Would be ideal to let the API translate the arguments
 	// into this required form instead of forcing that complexity on the caller.
-	return fmt.Sprintf("option task = { %s }\n%s", strings.Join(taskOpts, ", "), t.query)
+	taskOptStr := fmt.Sprintf("\noption task = { %s }", strings.Join(taskOpts, ", "))
+
+	if indices := fluxRegex.FindAllIndex([]byte(t.query), -1); len(indices) > 0 {
+		lastImportIdx := indices[len(indices)-1][1]
+		pieces := append([]string{},
+			t.query[:lastImportIdx],
+			taskOptStr,
+			t.query[lastImportIdx:],
+		)
+		return fmt.Sprint(strings.Join(pieces, "\n"))
+	}
+
+	return fmt.Sprintf("%s\n%s", taskOptStr, t.query)
 }
 
 func (t *task) summarize() SummaryTask {