test: add fluxtest harness

.gitignore

@@ -12,6 +12,10 @@ config.json
 # ignore generated files.
 cmd/influxd/version.go
 
+# Flux test harness artifacts
+fluxtest
+flux.zip
+
 # executables
 
 *.test

cmd/influxd/run/command.go

@@ -15,6 +15,7 @@ import (
 	"strconv"
 	"time"
 
+	fluxinit "github.com/influxdata/influxdb/flux/init"
 	"github.com/influxdata/influxdb/logger"
 	"go.uber.org/zap"
 )
@@ -101,6 +102,11 @@ func (cmd *Command) Run(args ...string) error {
 		go func() { http.ListenAndServe("localhost:6060", nil) }()
 	}
 
+	// Initialize the Flux built-ins if enabled.
+	if config.HTTPD.FluxEnabled {
+		fluxinit.Initialize()
+	}
+
 	// Print sweet InfluxDB logo.
 	if !config.Logging.SuppressLogo && logger.IsTerminal(cmd.Stdout) {
 		fmt.Fprint(cmd.Stdout, logo)

etc/test-flux.sh

@@ -0,0 +1,51 @@
+#!/bin/bash
+set -eu -o pipefail
+readonly GO=${GO:-go}
+
+log() {
+  local now
+  now=$(date '+%Y/%m/%d %H:%M:%S')
+  echo "[${now}]" "$@"
+}
+
+determine_flux_revision() {
+  local version revision
+  version=$("$GO" list -m -f '{{.Version}}' github.com/influxdata/flux)
+  revision=$(printf "%s" "${version}" | cut -d- -f 3)
+  if [[ ${revision} != "" ]]; then
+    printf "%s\n" "${revision}"
+  else
+    printf "%s\n" "${version}"
+  fi
+}
+
+download_flux_archive() {
+  local revision
+  revision=$(determine_flux_revision)
+  log "Downloading flux archive (${revision})..."
+  curl -sLo flux.zip "https://github.com/influxdata/flux/archive/${revision}.zip"
+}
+
+build_test_harness() {
+  log "Building test harness..."
+  "$GO" build -o fluxtest ./internal/cmd/fluxtest-harness-influxdb
+}
+
+run_integration_tests() {
+  log "Running flux integration tests..."
+  ./fluxtest -v -p flux.zip
+  log "Running influxdb integration tests..."
+  ./fluxtest -v -p flux/stdlib
+}
+
+cleanup() {
+  rm -f flux.zip fluxtest
+}
+
+main() {
+  build_test_harness
+  download_flux_archive
+  run_integration_tests
+  cleanup
+}
+main

flux/control/controller.go

@@ -5,11 +5,11 @@ import (
 	"time"
 
 	"github.com/influxdata/flux"
+	"github.com/influxdata/flux/dependencies/testing"
 	"github.com/influxdata/flux/lang"
 	"github.com/influxdata/flux/memory"
 	"github.com/influxdata/flux/runtime"
 	"github.com/influxdata/influxdb/coordinator"
-	"github.com/influxdata/influxdb/flux/builtin"
 	"github.com/influxdata/influxdb/flux/stdlib/influxdata/influxdb"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
@@ -72,15 +72,13 @@ func (p *program) Start(ctx context.Context, allocator *memory.Allocator) (flux.
 }
 
 func NewController(mc MetaClient, reader influxdb.Reader, auth Authorizer, authEnabled bool, writer influxdb.PointsWriter, logger *zap.Logger) *Controller {
-	builtin.Initialize()
-
 	storageDeps, err := influxdb.NewDependencies(mc, reader, auth, authEnabled, writer)
 	if err != nil {
 		panic(err)
 	}
 
 	return &Controller{
-		deps:   []flux.Dependency{storageDeps},
+		deps:   append([]flux.Dependency{storageDeps}, testing.FrameworkConfig{}),
 		logger: logger,
 	}
 }

flux/builtin/builtin.go → flux/init/init.go

@@ -1,23 +1,20 @@
-// Package builtin ensures all packages related to Flux built-ins are imported and initialized.
+// Package init ensures all packages related to Flux built-ins are imported and initialized.
 // This should only be imported from main or test packages.
 // It is a mistake to import it from any other package.
-package builtin
+//
+// NOTE: This is a superset-wrapper of Flux's built-in initialization logic.
+// It also ensures V1-specific flux builtins are initialized.
+package init
 
 import (
-	"sync"
-
 	"github.com/influxdata/flux/runtime"
 	_ "github.com/influxdata/flux/stdlib"
 	_ "github.com/influxdata/influxdb/flux/stdlib"
 )
 
-var once sync.Once
-
 // Initialize ensures all Flux builtins are configured and should be called
 // prior to using the Flux runtime. Initialize is safe to call concurrently
 // and is idempotent.
 func Initialize() {
-	once.Do(func() {
-		runtime.FinalizeBuiltIns()
-	})
+	runtime.FinalizeBuiltIns()
 }

flux/init/static/static.go

@@ -0,0 +1,9 @@
+// The init/static package can be imported in test cases and other uses
+// cases where it is okay to always initialize flux.
+package static
+
+import fluxinit "github.com/influxdata/influxdb/flux/init"
+
+func init() {
+	fluxinit.Initialize()
+}

flux/stdlib/influxdata/influxdb/filter_test.flux

@@ -0,0 +1,41 @@
+package influxdb_test
+
+import "csv"
+import "testing"
+import "testing/expect"
+
+option now = () => (2030-01-01T00:00:00Z)
+
+input = "#datatype,string,long,dateTime:RFC3339,string,string,string,double
+#group,false,false,false,true,true,true,false
+#default,_result,,,,,,
+,result,table,_time,_measurement,host,_field,_value
+,,0,2018-05-22T19:53:26Z,system,host.local,load1,1.83
+,,0,2018-05-22T19:53:36Z,system,host.local,load1,1.63
+,,1,2018-05-22T19:53:26Z,system,host.local,load3,1.72
+,,2,2018-05-22T19:53:26Z,system,host.local,load4,1.77
+,,2,2018-05-22T19:53:36Z,system,host.local,load4,1.78
+,,2,2018-05-22T19:53:46Z,system,host.local,load4,1.77
+"
+
+testcase filter {
+    expect.planner(rules: [
+        "influxdata/influxdb.FromStorageRule": 1,
+        "PushDownRangeRule": 1,
+        "PushDownFilterRule": 1,
+    ])
+
+    want = csv.from(csv: "#datatype,string,long,dateTime:RFC3339,string,string,string,double
+#group,false,false,false,true,true,true,false
+#default,_result,,,,,,
+,result,table,_time,_measurement,host,_field,_value
+,,0,2018-05-22T19:53:26Z,system,host.local,load1,1.83
+,,0,2018-05-22T19:53:36Z,system,host.local,load1,1.63
+")
+
+    got = testing.loadStorage(csv: input)
+        |> range(start: -100y)
+        |> filter(fn: (r) => r._measurement == "system" and r._field == "load1")
+        |> drop(columns: ["_start", "_stop"])
+    testing.diff(want, got)
+}

flux/stdlib/influxdata/influxdb/from.go

@@ -6,97 +6,32 @@ import (
 	"github.com/influxdata/flux"
 	"github.com/influxdata/flux/codes"
 	"github.com/influxdata/flux/plan"
-	"github.com/influxdata/flux/runtime"
 	"github.com/influxdata/flux/stdlib/influxdata/influxdb"
 )
 
 const FromKind = "influxDBFrom"
 
-type FromOpSpec struct {
-	Bucket   string `json:"bucket,omitempty"`
-	BucketID string `json:"bucketID,omitempty"`
-}
-
-func init() {
-	fromSignature := runtime.MustLookupBuiltinType("influxdata/influxdb", "from")
-	runtime.ReplacePackageValue("influxdata/influxdb", influxdb.FromKind, flux.MustValue(flux.FunctionValue(FromKind, createFromOpSpec, fromSignature)))
-	flux.RegisterOpSpec(FromKind, newFromOp)
-	plan.RegisterProcedureSpec(FromKind, newFromProcedure, FromKind)
-}
-
-func createFromOpSpec(args flux.Arguments, a *flux.Administration) (flux.OperationSpec, error) {
-	spec := new(FromOpSpec)
-
-	if bucket, ok, err := args.GetString("bucket"); err != nil {
-		return nil, err
-	} else if ok {
-		spec.Bucket = bucket
-	}
-
-	if bucketID, ok, err := args.GetString("bucketID"); err != nil {
-		return nil, err
-	} else if ok {
-		spec.BucketID = bucketID
-	}
-
-	if spec.Bucket == "" && spec.BucketID == "" {
-		return nil, &flux.Error{
-			Code: codes.Invalid,
-			Msg:  "must specify one of bucket or bucketID",
-		}
-	}
-	if spec.Bucket != "" && spec.BucketID != "" {
-		return nil, &flux.Error{
-			Code: codes.Invalid,
-			Msg:  "must specify only one of bucket or bucketID",
-		}
-	}
-	return spec, nil
-}
-
-func newFromOp() flux.OperationSpec {
-	return new(FromOpSpec)
-}
-
-func (s *FromOpSpec) Kind() flux.OperationKind {
-	return FromKind
-}
-
-type FromProcedureSpec struct {
-	Bucket   string
-	BucketID string
-}
-
-func newFromProcedure(qs flux.OperationSpec, pa plan.Administration) (plan.ProcedureSpec, error) {
-	spec, ok := qs.(*FromOpSpec)
-	if !ok {
-		return nil, &flux.Error{
-			Code: codes.Internal,
-			Msg:  fmt.Sprintf("invalid spec type %T", qs),
-		}
-	}
+type (
+	NameOrID   = influxdb.NameOrID
+	FromOpSpec = influxdb.FromOpSpec
+)
 
-	return &FromProcedureSpec{
-		Bucket:   spec.Bucket,
-		BucketID: spec.BucketID,
-	}, nil
+type FromStorageProcedureSpec struct {
+	Bucket influxdb.NameOrID
 }
 
-func (s *FromProcedureSpec) Kind() plan.ProcedureKind {
+func (s *FromStorageProcedureSpec) Kind() plan.ProcedureKind {
 	return FromKind
 }
 
-func (s *FromProcedureSpec) Copy() plan.ProcedureSpec {
-	ns := new(FromProcedureSpec)
-
+func (s *FromStorageProcedureSpec) Copy() plan.ProcedureSpec {
+	ns := new(FromStorageProcedureSpec)
 	ns.Bucket = s.Bucket
-	ns.BucketID = s.BucketID
-
 	return ns
 }
 
-func (s *FromProcedureSpec) PostPhysicalValidate(id plan.NodeID) error {
-	// FromProcedureSpec is a logical operation representing any read
+func (s *FromStorageProcedureSpec) PostPhysicalValidate(id plan.NodeID) error {
+	// FromStorageProcedureSpec is a logical operation representing any read
 	// from storage. However as a logical operation, it doesn't specify
 	// how data is to be read from storage. It is the query planner's
 	// job to determine the optimal read strategy and to convert this
@@ -108,10 +43,10 @@ func (s *FromProcedureSpec) PostPhysicalValidate(id plan.NodeID) error {
 	// not support unbounded reads, and so this query must not be
 	// validated.
 	var bucket string
-	if len(s.Bucket) > 0 {
-		bucket = s.Bucket
+	if len(s.Bucket.Name) > 0 {
+		bucket = s.Bucket.Name
 	} else {
-		bucket = s.BucketID
+		bucket = s.Bucket.ID
 	}
 	return &flux.Error{
 		Code: codes.Invalid,

flux/stdlib/influxdata/influxdb/from_test.go

@@ -1 +1,44 @@
 package influxdb_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/influxdata/flux"
+	"github.com/influxdata/flux/plan"
+	"github.com/influxdata/flux/plan/plantest"
+	"github.com/influxdata/flux/stdlib/influxdata/influxdb"
+	"github.com/influxdata/flux/stdlib/universe"
+	qinfluxdb "github.com/influxdata/influxdb/flux/stdlib/influxdata/influxdb"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFromValidation(t *testing.T) {
+	spec := plantest.PlanSpec{
+		// from |> group (cannot query an infinite time range)
+		Nodes: []plan.Node{
+			plan.CreateLogicalNode("from", &influxdb.FromProcedureSpec{
+				Bucket: influxdb.NameOrID{Name: "my-bucket"},
+			}),
+			plan.CreatePhysicalNode("group", &universe.GroupProcedureSpec{
+				GroupMode: flux.GroupModeBy,
+				GroupKeys: []string{"_measurement", "_field"},
+			}),
+		},
+		Edges: [][2]int{
+			{0, 1},
+		},
+	}
+
+	ps := plantest.CreatePlanSpec(&spec)
+	pp := plan.NewPhysicalPlanner(plan.OnlyPhysicalRules(
+		qinfluxdb.FromStorageRule{},
+		qinfluxdb.PushDownRangeRule{},
+		qinfluxdb.PushDownFilterRule{},
+		qinfluxdb.PushDownGroupRule{},
+	))
+	_, err := pp.Plan(context.Background(), ps)
+	require.Error(t, err, "Expected query with no call to range to fail physical planning")
+	want := `cannot submit unbounded read to "my-bucket"; try bounding 'from' with a call to 'range'`
+	require.Equal(t, want, err.Error())
+}