Merge pull request #13569 from influxdata/flux-staging

 chore(dep): update Flux to v0.27.0

cmd/influxd/launcher/launcher.go

@@ -478,15 +478,17 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 		pointsWriter = m.engine
 
 		const (
-			concurrencyQuota = 10
-			memoryBytesQuota = 1e6
+			concurrencyQuota         = 10
+			memoryBytesQuotaPerQuery = 1e6
+			QueueSize                = 10
 		)
 
 		cc := control.Config{
-			ExecutorDependencies: make(execute.Dependencies),
-			ConcurrencyQuota:     concurrencyQuota,
-			MemoryBytesQuota:     int64(memoryBytesQuota),
-			Logger:               m.logger.With(zap.String("service", "storage-reads")),
+			ExecutorDependencies:     make(execute.Dependencies),
+			ConcurrencyQuota:         concurrencyQuota,
+			MemoryBytesQuotaPerQuery: int64(memoryBytesQuotaPerQuery),
+			QueueSize:                QueueSize,
+			Logger:                   m.logger.With(zap.String("service", "storage-reads")),
 		}
 
 		if err := readservice.AddControllerConfigDependencies(
@@ -496,7 +498,12 @@ func (m *Launcher) run(ctx context.Context) (err error) {
 			return err
 		}
 
-		m.queryController = pcontrol.New(cc)
+		c, err := pcontrol.New(cc)
+		if err != nil {
+			m.logger.Error("Failed to create query controller", zap.Error(err))
+			return err
+		}
+		m.queryController = c
 		m.reg.MustRegister(m.queryController.PrometheusCollectors()...)
 	}
 

go.mod

@@ -63,7 +63,7 @@ require (
 	github.com/hashicorp/vault v0.11.5
 	github.com/hashicorp/vault-plugin-secrets-kv v0.0.0-20181106190520-2236f141171e // indirect
 	github.com/hashicorp/yamux v0.0.0-20181012175058-2f1d1f20f75d // indirect
-	github.com/influxdata/flux v0.26.0
+	github.com/influxdata/flux v0.27.0
 	github.com/influxdata/influxql v0.0.0-20180925231337-1cbfca8e56b6
 	github.com/influxdata/usage-client v0.0.0-20160829180054-6d3895376368
 	github.com/jefferai/jsonx v0.0.0-20160721235117-9cc31c3135ee // indirect

go.sum

@@ -225,8 +225,8 @@ github.com/imdario/mergo v0.3.6 h1:xTNEAn+kxVO7dTZGu0CegyqKZmoWFI0rF8UxjlB2d28=
 github.com/imdario/mergo v0.3.6/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
-github.com/influxdata/flux v0.26.0 h1:NumEU26pyNAWDmhVj5yaV5/l5WZHSE1GFsByE5pwdIk=
-github.com/influxdata/flux v0.26.0/go.mod h1:YwRX+zCiwIRI+QZB+ywDfz8yqNnwu9g2g5aL4/n/bXY=
+github.com/influxdata/flux v0.27.0 h1:wSJ9WHJnoc08XMSWFs+7jlHL4o3yokFe92mN5FSeD84=
+github.com/influxdata/flux v0.27.0/go.mod h1:YwRX+zCiwIRI+QZB+ywDfz8yqNnwu9g2g5aL4/n/bXY=
 github.com/influxdata/goreleaser v0.97.0-influx h1:jT5OrcW7WfS0e2QxfwmTBjhLvpIC9CDLRhNgZJyhj8s=
 github.com/influxdata/goreleaser v0.97.0-influx/go.mod h1:MnjA0e0Uq6ISqjG1WxxMAl+3VS1QYjILSWVnMYDxasE=
 github.com/influxdata/influxql v0.0.0-20180925231337-1cbfca8e56b6 h1:CFx+pP90q/qg3spoiZjf8donE4WpAdjeJfPOcoNqkWo=

query/control/controller.go

@@ -22,10 +22,13 @@ type Controller struct {
 }
 
 // NewController creates a new Controller specific to platform.
-func New(config control.Config) *Controller {
+func New(config control.Config) (*Controller, error) {
 	config.MetricLabelKeys = append(config.MetricLabelKeys, orgLabel)
-	c := control.New(config)
-	return &Controller{c: c}
+	c, err := control.New(config)
+	if err != nil {
+		return nil, err
+	}
+	return &Controller{c: c}, nil
 }
 
 // Query satisfies the AsyncQueryService while ensuring the request is propagated on the context.

task/backend/analytical_storage_test.go

@@ -105,15 +105,17 @@ func newAnalyticalBackend(t *testing.T, orgSvc influxdb.OrganizationService, buc
 	}()
 
 	const (
-		concurrencyQuota = 10
-		memoryBytesQuota = 1e6
+		concurrencyQuota         = 10
+		memoryBytesQuotaPerQuery = 1e6
+		queueSize                = 10
 	)
 
 	cc := control.Config{
-		ExecutorDependencies: make(execute.Dependencies),
-		ConcurrencyQuota:     concurrencyQuota,
-		MemoryBytesQuota:     int64(memoryBytesQuota),
-		Logger:               logger.With(zap.String("service", "storage-reads")),
+		ExecutorDependencies:     make(execute.Dependencies),
+		ConcurrencyQuota:         concurrencyQuota,
+		MemoryBytesQuotaPerQuery: int64(memoryBytesQuotaPerQuery),
+		QueueSize:                queueSize,
+		Logger:                   logger.With(zap.String("service", "storage-reads")),
 	}
 
 	if err := readservice.AddControllerConfigDependencies(
@@ -122,7 +124,10 @@ func newAnalyticalBackend(t *testing.T, orgSvc influxdb.OrganizationService, buc
 		t.Fatal(err)
 	}
 
-	queryController := pcontrol.New(cc)
+	queryController, err := pcontrol.New(cc)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	return &analyticalBackend{
 		queryController: queryController,

task/backend/logreaderwriter_test.go

@@ -120,15 +120,17 @@ func newFullStackAwareLogReaderWriter(t *testing.T) *fullStackAwareLogReaderWrit
 	svc := inmem.NewService()
 
 	const (
-		concurrencyQuota = 10
-		memoryBytesQuota = 1e6
+		concurrencyQuota         = 10
+		memoryBytesQuotaPerQuery = 1e6
+		queueSize                = 10
 	)
 
 	cc := control.Config{
-		ExecutorDependencies: make(execute.Dependencies),
-		ConcurrencyQuota:     concurrencyQuota,
-		MemoryBytesQuota:     int64(memoryBytesQuota),
-		Logger:               logger.With(zap.String("service", "storage-reads")),
+		ExecutorDependencies:     make(execute.Dependencies),
+		ConcurrencyQuota:         concurrencyQuota,
+		MemoryBytesQuotaPerQuery: int64(memoryBytesQuotaPerQuery),
+		QueueSize:                queueSize,
+		Logger:                   logger.With(zap.String("service", "storage-reads")),
 	}
 
 	if err := readservice.AddControllerConfigDependencies(
@@ -137,7 +139,10 @@ func newFullStackAwareLogReaderWriter(t *testing.T) *fullStackAwareLogReaderWrit
 		t.Fatal(err)
 	}
 
-	queryController := pcontrol.New(cc)
+	queryController, err := pcontrol.New(cc)
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	return &fullStackAwareLogReaderWriter{
 		PointLogWriter: backend.NewPointLogWriter(engine),