feat(platform): support flux queries for 1x sources

http/influxdb/query.go

@@ -3,10 +3,14 @@ package influxdb
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"fmt"
+	nethttp "net/http"
+	"net/url"
 
 	"github.com/influxdata/platform"
 	"github.com/influxdata/platform/chronograf"
+	"github.com/influxdata/platform/http"
 )
 
 // SourceQuerier connects to Influx via HTTP using tokens to manage buckets
@@ -22,7 +26,7 @@ func (s *SourceQuerier) Query(ctx context.Context, q *platform.SourceQuery) (*pl
 		return s.fluxQuery(ctx, q)
 	}
 
-	return nil, fmt.Errorf("unsupport language %v", q.Type)
+	return nil, fmt.Errorf("unsupported language %v", q.Type)
 }
 
 func (s *SourceQuerier) influxQuery(ctx context.Context, q *platform.SourceQuery) (*platform.SourceQueryResult, error) {
@@ -52,5 +56,73 @@ func (s *SourceQuerier) influxQuery(ctx context.Context, q *platform.SourceQuery
 }
 
 func (s *SourceQuerier) fluxQuery(ctx context.Context, q *platform.SourceQuery) (*platform.SourceQueryResult, error) {
-	panic("not implemented")
+	req, err := s.newFluxHTTPRequest(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+
+	hc := newHTTPClient(req.URL.Scheme, s.Source.InsecureSkipVerify)
+	resp, err := hc.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := http.CheckError(resp); err != nil {
+		return nil, err
+	}
+
+	return &platform.SourceQueryResult{
+		Reader: resp.Body,
+	}, nil
+
+}
+
+func newURL(addr, path string) (*url.URL, error) {
+	u, err := url.Parse(addr)
+	if err != nil {
+		return nil, err
+	}
+	u.Path = path
+	return u, nil
 }
+
+func (s *SourceQuerier) newFluxHTTPRequest(ctx context.Context, q *platform.SourceQuery) (*nethttp.Request, error) {
+	u, err := newURL(s.Source.FluxURL, "/v1/query")
+	if err != nil {
+		return nil, err
+	}
+	values := url.Values{}
+	values.Set("q", q.Query)
+	// TODO(desa): replace this with real value.
+	values.Set("orgName", "an org has no name")
+	u.RawQuery = values.Encode()
+
+	req, err := nethttp.NewRequest("POST", u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	// TODO(desa): replace with real token
+	req.Header.Set("Authorization", "Token 123")
+	req.Header.Set("Accept", "text/csv")
+
+	return req, nil
+}
+
+func newHTTPClient(scheme string, insecure bool) *nethttp.Client {
+	hc := &nethttp.Client{
+		Transport: defaultTransport,
+	}
+	if scheme == "https" && insecure {
+		hc.Transport = skipVerifyTransport
+	}
+
+	return hc
+}
+
+// Shared transports for all clients to prevent leaking connections
+var (
+	skipVerifyTransport = &nethttp.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+	defaultTransport = &nethttp.Transport{}
+)

source.go

@@ -51,6 +51,7 @@ type V1SourceFields struct {
 	SharedSecret string `json:"sharedSecret,omitempty"` // ShareSecret is the optional signing secret for Influx JWT authorization
 	MetaURL      string `json:"metaUrl,omitempty"`      // MetaURL is the url for the meta node
 	DefaultRP    string `json:"defaultRP"`              // DefaultRP is the default retention policy used in database queries to this source
+	FluxURL      string `json:"fluxURL,omitempty"`      // FluxURL is the url for a flux connected to a 1x source
 }
 
 // SourceFields
@@ -85,7 +86,8 @@ type SourceUpdate struct {
 	Username           *string     `json:"username,omitempty"`
 	Password           *string     `json:"password,omitempty"`
 	SharedSecret       *string     `json:"sharedSecret,omitempty"`
-	MetaURL            *string     `json:"metaUrl,omitempty"`
+	MetaURL            *string     `json:"metaURL,omitempty"`
+	FluxURL            *string     `json:"fluxURL,omitempty"`
 	Role               *string     `json:"role,omitempty"`
 	DefaultRP          *string     `json:"defaultRP"`
 }
@@ -122,6 +124,9 @@ func (u SourceUpdate) Apply(s *Source) error {
 	if u.MetaURL != nil {
 		s.MetaURL = *u.MetaURL
 	}
+	if u.FluxURL != nil {
+		s.FluxURL = *u.FluxURL
+	}
 	if u.DefaultRP != nil {
 		s.DefaultRP = *u.DefaultRP
 	}