chore: Remove platform dependency

Author: nathanielc

compiler.go

@@ -2,48 +2,25 @@ package flux
 
 import (
 	"context"
-	"time"
+	"fmt"
 )
 
-const (
-	FluxCompilerType = "flux"
-	SpecCompilerType = "spec"
-)
-
-// AddCompilerMappings adds the Flux specific compiler mappings.
-func AddCompilerMappings(mappings CompilerMappings) error {
-	if err := mappings.Add(FluxCompilerType, func() Compiler {
-		return new(FluxCompiler)
-
-	}); err != nil {
-		return err
-	}
-	return mappings.Add(SpecCompilerType, func() Compiler {
-		return new(SpecCompiler)
-	})
+// Compiler produces a specification for the query.
+type Compiler interface {
+	// Compile produces a specification for the query.
+	Compile(ctx context.Context) (*Spec, error)
+	CompilerType() CompilerType
 }
 
-// FluxCompiler compiles a Flux script into a spec.
-type FluxCompiler struct {
-	Query string `json:"query"`
-}
-
-func (c FluxCompiler) Compile(ctx context.Context) (*Spec, error) {
-	return Compile(ctx, c.Query, time.Now())
-}
+// CompilerType is the name of a query compiler.
+type CompilerType string
+type CreateCompiler func() Compiler
+type CompilerMappings map[CompilerType]CreateCompiler
 
-func (c FluxCompiler) CompilerType() CompilerType {
-	return FluxCompilerType
-}
-
-// SpecCompiler implements Compiler by returning a known spec.
-type SpecCompiler struct {
-	Spec *Spec `json:"spec"`
-}
-
-func (c SpecCompiler) Compile(ctx context.Context) (*Spec, error) {
-	return c.Spec, nil
-}
-func (c SpecCompiler) CompilerType() CompilerType {
-	return SpecCompilerType
+func (m CompilerMappings) Add(t CompilerType, c CreateCompiler) error {
+	if _, ok := m[t]; ok {
+		return fmt.Errorf("duplicate compiler mapping for %q", t)
+	}
+	m[t] = c
+	return nil
 }

control/controller.go

@@ -31,7 +31,6 @@ import (
 	"github.com/influxdata/flux"
 	"github.com/influxdata/flux/execute"
 	"github.com/influxdata/flux/plan"
-	"github.com/influxdata/platform"
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -48,7 +47,8 @@ type Controller struct {
 	queryDone     chan *Query
 	cancelRequest chan QueryID
 
-	metrics *controllerMetrics
+	metrics   *controllerMetrics
+	labelKeys []string
 
 	verbose bool
 
@@ -69,6 +69,10 @@ type Config struct {
 	PlannerOptions       []plan.Option
 	Logger               *zap.Logger
 	Verbose              bool
+	// MetricLabelKeys is a list of labels to add to the metrics produced by the controller.
+	// The value for a given key will be read off the context.
+	// The context value must be a string or an implementation of the Stringer interface.
+	MetricLabelKeys []string
 }
 
 type QueryID uint64
@@ -90,18 +94,19 @@ func New(c Config) *Controller {
 		pplanner:             plan.NewPlanner(c.PlannerOptions...),
 		executor:             execute.NewExecutor(c.ExecutorDependencies, logger),
 		logger:               logger,
-		metrics:              newControllerMetrics(),
 		verbose:              c.Verbose,
+		metrics:              newControllerMetrics(c.MetricLabelKeys),
+		labelKeys:            c.MetricLabelKeys,
 	}
 	go ctrl.run()
 	return ctrl
 }
 
 // Query submits a query for execution returning immediately.
 // Done must be called on any returned Query objects.
-func (c *Controller) Query(ctx context.Context, req *flux.Request) (flux.Query, error) {
-	q := c.createQuery(ctx, req.OrganizationID)
-	if err := c.compileQuery(q, req.Compiler); err != nil {
+func (c *Controller) Query(ctx context.Context, compiler flux.Compiler) (flux.Query, error) {
+	q := c.createQuery(ctx, compiler.CompilerType())
+	if err := c.compileQuery(q, compiler); err != nil {
 		q.parentSpan.Finish()
 		return nil, err
 	}
@@ -112,11 +117,28 @@ func (c *Controller) Query(ctx context.Context, req *flux.Request) (flux.Query,
 	return q, nil
 }
 
-func (c *Controller) createQuery(ctx context.Context, orgID platform.ID) *Query {
+type Stringer interface {
+	String() string
+}
+
+func (c *Controller) createQuery(ctx context.Context, ct flux.CompilerType) *Query {
 	id := c.nextID()
-	labelValues := []string{
-		orgID.String(),
+	labelValues := make([]string, len(c.labelKeys))
+	compileLabelValues := make([]string, len(c.labelKeys)+1)
+	for i, k := range c.labelKeys {
+		value := ctx.Value(k)
+		var str string
+		switch v := value.(type) {
+		case string:
+			str = v
+		case Stringer:
+			str = v.String()
+		}
+		labelValues[i] = str
+		compileLabelValues[i] = str
 	}
+	compileLabelValues[len(compileLabelValues)-1] = string(ct)
+
 	cctx, cancel := context.WithCancel(ctx)
 	parentSpan, parentCtx := StartSpanFromContext(
 		cctx,
@@ -126,16 +148,16 @@ func (c *Controller) createQuery(ctx context.Context, orgID platform.ID) *Query
 	)
 	ready := make(chan map[string]flux.Result, 1)
 	return &Query{
-		id:          id,
-		orgID:       orgID,
-		labelValues: labelValues,
-		state:       Created,
-		c:           c,
-		now:         time.Now().UTC(),
-		ready:       ready,
-		parentCtx:   parentCtx,
-		parentSpan:  parentSpan,
-		cancel:      cancel,
+		id:                 id,
+		labelValues:        labelValues,
+		compileLabelValues: compileLabelValues,
+		state:              Created,
+		c:                  c,
+		now:                time.Now().UTC(),
+		ready:              ready,
+		parentCtx:          parentCtx,
+		parentSpan:         parentSpan,
+		cancel:             cancel,
 	}
 }
 
@@ -296,7 +318,7 @@ func (c *Controller) processQuery(q *Query) (pop bool, err error) {
 			return true, errors.New("failed to transition query into executing state")
 		}
 		q.alloc = new(execute.Allocator)
-		r, err := c.executor.Execute(q.executeCtx, q.orgID, q.plan, q.alloc)
+		r, err := c.executor.Execute(q.executeCtx, q.plan, q.alloc)
 		if err != nil {
 			return true, errors.Wrap(err, "failed to execute query")
 		}
@@ -338,9 +360,8 @@ func (c *Controller) PrometheusCollectors() []prometheus.Collector {
 type Query struct {
 	id QueryID
 
-	orgID platform.ID
-
-	labelValues []string
+	labelValues        []string
+	compileLabelValues []string
 
 	c *Controller
 
@@ -383,10 +404,6 @@ func (q *Query) ID() QueryID {
 	return q.id
 }
 
-func (q *Query) OrganizationID() platform.ID {
-	return q.orgID
-}
-
 func (q *Query) Spec() *flux.Spec {
 	return &q.spec
 }
@@ -549,8 +566,8 @@ func (q *Query) tryCompile() bool {
 		q.compileSpan, q.compilingCtx = StartSpanFromContext(
 			q.parentCtx,
 			"compiling",
-			q.c.metrics.compilingDur.WithLabelValues(q.labelValues...),
-			q.c.metrics.compiling.WithLabelValues(q.labelValues...),
+			q.c.metrics.compilingDur.WithLabelValues(q.compileLabelValues...),
+			q.c.metrics.compiling.WithLabelValues(q.compileLabelValues...),
 		)
 
 		q.state = Compiling

control/controller_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/influxdata/flux/execute"
 	"github.com/influxdata/flux/mock"
 	"github.com/influxdata/flux/plan"
-	"github.com/influxdata/platform"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
@@ -33,18 +32,14 @@ func TestController_CompileQuery_Failure(t *testing.T) {
 	}
 
 	ctrl := New(Config{})
-	req := &flux.Request{
-		OrganizationID: platform.ID("a"),
-		Compiler:       compiler,
-	}
 
 	// Run the query. It should return an error.
-	if _, err := ctrl.Query(context.Background(), req); err == nil {
+	if _, err := ctrl.Query(context.Background(), compiler); err == nil {
 		t.Fatal("expected error")
 	}
 
 	// Verify the metrics say there are no queries.
-	gauge, err := ctrl.metrics.all.GetMetricWithLabelValues(req.OrganizationID.String())
+	gauge, err := ctrl.metrics.all.GetMetricWithLabelValues()
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -71,13 +66,9 @@ func TestController_EnqueueQuery_Failure(t *testing.T) {
 	}
 
 	ctrl := New(Config{})
-	req := &flux.Request{
-		OrganizationID: platform.ID("a"),
-		Compiler:       compiler,
-	}
 
 	// Run the query. It should return an error.
-	if _, err := ctrl.Query(context.Background(), req); err == nil {
+	if _, err := ctrl.Query(context.Background(), compiler); err == nil {
 		t.Fatal("expected error")
 	}
 
@@ -86,7 +77,7 @@ func TestController_EnqueueQuery_Failure(t *testing.T) {
 		"all":      ctrl.metrics.all,
 		"queueing": ctrl.metrics.queueing,
 	} {
-		gauge, err := gaugeVec.GetMetricWithLabelValues(req.OrganizationID.String())
+		gauge, err := gaugeVec.GetMetricWithLabelValues()
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -104,19 +95,15 @@ func TestController_EnqueueQuery_Failure(t *testing.T) {
 
 func TestController_ExecuteQuery_Failure(t *testing.T) {
 	executor := mock.NewExecutor()
-	executor.ExecuteFn = func(context.Context, platform.ID, *plan.PlanSpec, *execute.Allocator) (map[string]flux.Result, error) {
+	executor.ExecuteFn = func(context.Context, *plan.PlanSpec, *execute.Allocator) (map[string]flux.Result, error) {
 		return nil, errors.New("expected")
 	}
 
 	ctrl := New(Config{})
 	ctrl.executor = executor
-	req := &flux.Request{
-		OrganizationID: platform.ID("a"),
-		Compiler:       mockCompiler,
-	}
 
 	// Run a query and then wait for it to be ready.
-	q, err := ctrl.Query(context.Background(), req)
+	q, err := ctrl.Query(context.Background(), mockCompiler)
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -134,7 +121,7 @@ func TestController_ExecuteQuery_Failure(t *testing.T) {
 	q.Done()
 
 	// Verify the metrics say there are no queries.
-	gauge, err := ctrl.metrics.all.GetMetricWithLabelValues(req.OrganizationID.String())
+	gauge, err := ctrl.metrics.all.GetMetricWithLabelValues()
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -151,20 +138,16 @@ func TestController_ExecuteQuery_Failure(t *testing.T) {
 
 func TestController_CancelQuery(t *testing.T) {
 	executor := mock.NewExecutor()
-	executor.ExecuteFn = func(context.Context, platform.ID, *plan.PlanSpec, *execute.Allocator) (map[string]flux.Result, error) {
+	executor.ExecuteFn = func(context.Context, *plan.PlanSpec, *execute.Allocator) (map[string]flux.Result, error) {
 		// Return an empty result.
 		return map[string]flux.Result{}, nil
 	}
 
 	ctrl := New(Config{})
 	ctrl.executor = executor
-	req := &flux.Request{
-		OrganizationID: platform.ID("a"),
-		Compiler:       mockCompiler,
-	}
 
 	// Run a query and then wait for it to be ready.
-	q, err := ctrl.Query(context.Background(), req)
+	q, err := ctrl.Query(context.Background(), mockCompiler)
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -179,7 +162,7 @@ func TestController_CancelQuery(t *testing.T) {
 	q.Done()
 
 	// Verify the metrics say there are no queries.
-	gauge, err := ctrl.metrics.all.GetMetricWithLabelValues(req.OrganizationID.String())
+	gauge, err := ctrl.metrics.all.GetMetricWithLabelValues()
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -198,20 +181,16 @@ func TestController_BlockedExecutor(t *testing.T) {
 	done := make(chan struct{})
 
 	executor := mock.NewExecutor()
-	executor.ExecuteFn = func(context.Context, platform.ID, *plan.PlanSpec, *execute.Allocator) (map[string]flux.Result, error) {
+	executor.ExecuteFn = func(context.Context, *plan.PlanSpec, *execute.Allocator) (map[string]flux.Result, error) {
 		<-done
 		return nil, nil
 	}
 
 	ctrl := New(Config{})
 	ctrl.executor = executor
-	req := &flux.Request{
-		OrganizationID: platform.ID("a"),
-		Compiler:       mockCompiler,
-	}
 
 	// Run a query that will cause the controller to stall.
-	q, err := ctrl.Query(context.Background(), req)
+	q, err := ctrl.Query(context.Background(), mockCompiler)
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -234,7 +213,7 @@ func TestController_BlockedExecutor(t *testing.T) {
 		}
 	}()
 
-	if _, err := ctrl.Query(ctx, req); err == nil {
+	if _, err := ctrl.Query(ctx, mockCompiler); err == nil {
 		t.Fatal("expected error")
 	} else if got, want := err, context.Canceled; got != want {
 		t.Fatalf("unexpected error: got=%q want=%q", got, want)