Loki HTTP/JSON Model Layer

.circleci/config.yml

@@ -78,7 +78,7 @@ jobs:
       - checkout
       - run:
           name: build
-          command: make GOOS=windows promtail
+          command: make GOOS=windows GOGC=10 promtail
 
   # Docker driver
   build/docker-driver:

Makefile

@@ -214,7 +214,7 @@ lint:
 ########
 
 test: all
-	go test -p=6 ./...
+	GOGC=10 go test -p=4 ./...
 
 #########
 # Clean #

pkg/canary/reader/reader.go

@@ -18,8 +18,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
+	loghttp "github.com/grafana/loki/pkg/loghttp/legacy"
 	"github.com/grafana/loki/pkg/logproto"
-	"github.com/grafana/loki/pkg/querier"
 )
 
 var (
@@ -159,7 +159,7 @@ func (r *Reader) run() {
 
 	r.closeAndReconnect()
 
-	tailResponse := &querier.TailResponse{}
+	tailResponse := &loghttp.TailResponse{}
 
 	for {
 		err := r.conn.ReadJSON(tailResponse)

pkg/logcli/client/client.go

@@ -11,13 +11,10 @@ import (
 	"strings"
 	"time"
 
-	"github.com/prometheus/common/model"
-
-	"github.com/grafana/loki/pkg/logql"
+	"github.com/grafana/loki/pkg/loghttp"
 
 	"github.com/gorilla/websocket"
 	"github.com/prometheus/common/config"
-	"github.com/prometheus/prometheus/promql"
 
 	"github.com/grafana/loki/pkg/logproto"
 )
@@ -38,16 +35,10 @@ type Client struct {
 	Address   string
 }
 
-// QueryResult contains fields necessary to return data from Loki endpoints
-type QueryResult struct {
-	ResultType promql.ValueType
-	Result     interface{}
-}
-
 // Query uses the /api/v1/query endpoint to execute an instant query
 // excluding interfacer b/c it suggests taking the interface promql.Node instead of logproto.Direction b/c it happens to have a String() method
 // nolint:interfacer
-func (c *Client) Query(queryStr string, limit int, time time.Time, direction logproto.Direction, quiet bool) (*QueryResult, error) {
+func (c *Client) Query(queryStr string, limit int, time time.Time, direction logproto.Direction, quiet bool) (*loghttp.QueryResponse, error) {
 	path := fmt.Sprintf(queryPath,
 		url.QueryEscape(queryStr), // query
 		limit,                     // limit
@@ -61,7 +52,7 @@ func (c *Client) Query(queryStr string, limit int, time time.Time, direction log
 // QueryRange uses the /api/v1/query_range endpoint to execute a range query
 // excluding interfacer b/c it suggests taking the interface promql.Node instead of logproto.Direction b/c it happens to have a String() method
 // nolint:interfacer
-func (c *Client) QueryRange(queryStr string, limit int, from, through time.Time, direction logproto.Direction, quiet bool) (*QueryResult, error) {
+func (c *Client) QueryRange(queryStr string, limit int, from, through time.Time, direction logproto.Direction, quiet bool) (*loghttp.QueryResponse, error) {
 	path := fmt.Sprintf(queryRangePath,
 		url.QueryEscape(queryStr), // query
 		limit,                     // limit
@@ -74,64 +65,33 @@ func (c *Client) QueryRange(queryStr string, limit int, from, through time.Time,
 }
 
 // ListLabelNames uses the /api/v1/label endpoint to list label names
-func (c *Client) ListLabelNames(quiet bool) (*logproto.LabelResponse, error) {
-	var labelResponse logproto.LabelResponse
+func (c *Client) ListLabelNames(quiet bool) (*loghttp.LabelResponse, error) {
+	var labelResponse loghttp.LabelResponse
 	if err := c.doRequest(labelsPath, quiet, &labelResponse); err != nil {
 		return nil, err
 	}
 	return &labelResponse, nil
 }
 
 // ListLabelValues uses the /api/v1/label endpoint to list label values
-func (c *Client) ListLabelValues(name string, quiet bool) (*logproto.LabelResponse, error) {
+func (c *Client) ListLabelValues(name string, quiet bool) (*loghttp.LabelResponse, error) {
 	path := fmt.Sprintf(labelValuesPath, url.PathEscape(name))
-	var labelResponse logproto.LabelResponse
+	var labelResponse loghttp.LabelResponse
 	if err := c.doRequest(path, quiet, &labelResponse); err != nil {
 		return nil, err
 	}
 	return &labelResponse, nil
 }
 
-func (c *Client) doQuery(path string, quiet bool) (*QueryResult, error) {
+func (c *Client) doQuery(path string, quiet bool) (*loghttp.QueryResponse, error) {
 	var err error
+	var r loghttp.QueryResponse
 
-	unmarshal := struct {
-		Type   promql.ValueType `json:"resultType"`
-		Result json.RawMessage  `json:"result"`
-	}{}
-
-	if err = c.doRequest(path, quiet, &unmarshal); err != nil {
-		return nil, err
-	}
-
-	var value interface{}
-
-	// unmarshal results
-	switch unmarshal.Type {
-	case logql.ValueTypeStreams:
-		var s logql.Streams
-		err = json.Unmarshal(unmarshal.Result, &s)
-		value = s
-	case promql.ValueTypeMatrix:
-		var m model.Matrix
-		err = json.Unmarshal(unmarshal.Result, &m)
-		value = m
-	case promql.ValueTypeVector:
-		var m model.Vector
-		err = json.Unmarshal(unmarshal.Result, &m)
-		value = m
-	default:
-		return nil, fmt.Errorf("Unknown type: %s", unmarshal.Type)
-	}
-
-	if err != nil {
+	if err = c.doRequest(path, quiet, &r); err != nil {
 		return nil, err
 	}
 
-	return &QueryResult{
-		ResultType: unmarshal.Type,
-		Result:     value,
-	}, nil
+	return &r, nil
 }
 
 func (c *Client) doRequest(path string, quiet bool, out interface{}) error {

pkg/logcli/labelquery/labels.go

@@ -5,7 +5,7 @@ import (
 	"log"
 
 	"github.com/grafana/loki/pkg/logcli/client"
-	"github.com/grafana/loki/pkg/logproto"
+	"github.com/grafana/loki/pkg/loghttp"
 )
 
 // LabelQuery contains all necessary fields to execute label queries and print out the resutls
@@ -25,7 +25,7 @@ func (q *LabelQuery) DoLabels(c *client.Client) {
 
 // ListLabels returns an array of label strings
 func (q *LabelQuery) ListLabels(c *client.Client) []string {
-	var labelResponse *logproto.LabelResponse
+	var labelResponse *loghttp.LabelResponse
 	var err error
 	if len(q.LabelName) > 0 {
 		labelResponse, err = c.ListLabelValues(q.LabelName, q.Quiet)

pkg/logcli/output/default.go

@@ -6,7 +6,7 @@ import (
 	"time"
 
 	"github.com/fatih/color"
-	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/grafana/loki/pkg/loghttp"
 )
 
 // DefaultOutput provides logs and metadata in human readable format
@@ -15,19 +15,19 @@ type DefaultOutput struct {
 }
 
 // Format a log entry in a human readable format
-func (o *DefaultOutput) Format(ts time.Time, lbls *labels.Labels, maxLabelsLen int, line string) string {
+func (o *DefaultOutput) Format(ts time.Time, lbls loghttp.LabelSet, maxLabelsLen int, line string) string {
 	timestamp := ts.In(o.options.Timezone).Format(time.RFC3339)
 	line = strings.TrimSpace(line)
 
 	if o.options.NoLabels {
 		return fmt.Sprintf("%s %s", color.BlueString(timestamp), line)
 	}
 
-	return fmt.Sprintf("%s %s %s", color.BlueString(timestamp), color.RedString(padLabel(*lbls, maxLabelsLen)), line)
+	return fmt.Sprintf("%s %s %s", color.BlueString(timestamp), color.RedString(padLabel(lbls, maxLabelsLen)), line)
 }
 
 // add some padding after labels
-func padLabel(ls labels.Labels, maxLabelsLen int) string {
+func padLabel(ls loghttp.LabelSet, maxLabelsLen int) string {
 	labels := ls.String()
 	if len(labels) < maxLabelsLen {
 		labels += strings.Repeat(" ", maxLabelsLen-len(labels))

pkg/logcli/output/default_test.go

@@ -5,69 +5,71 @@ import (
 	"testing"
 	"time"
 
-	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/grafana/loki/pkg/loghttp"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestDefaultOutput_Format(t *testing.T) {
 	t.Parallel()
 
 	timestamp, _ := time.Parse(time.RFC3339, "2006-01-02T15:04:05+07:00")
-	emptyLabels := labels.New()
-	someLabels := labels.New(labels.Label{Name: "type", Value: "test"})
+	emptyLabels := loghttp.LabelSet{}
+	someLabels := loghttp.LabelSet(map[string]string{
+		"type": "test",
+	})
 
 	tests := map[string]struct {
 		options      *LogOutputOptions
 		timestamp    time.Time
-		lbls         *labels.Labels
+		lbls         loghttp.LabelSet
 		maxLabelsLen int
 		line         string
 		expected     string
 	}{
 		"empty line with no labels": {
 			&LogOutputOptions{Timezone: time.UTC, NoLabels: false},
 			timestamp,
-			&emptyLabels,
+			emptyLabels,
 			0,
 			"",
 			"2006-01-02T08:04:05Z {} ",
 		},
 		"empty line with labels": {
 			&LogOutputOptions{Timezone: time.UTC, NoLabels: false},
 			timestamp,
-			&someLabels,
+			someLabels,
 			len(someLabels.String()),
 			"",
 			"2006-01-02T08:04:05Z {type=\"test\"} ",
 		},
 		"max labels length shorter than input labels": {
 			&LogOutputOptions{Timezone: time.UTC, NoLabels: false},
 			timestamp,
-			&someLabels,
+			someLabels,
 			0,
 			"Hello",
 			"2006-01-02T08:04:05Z {type=\"test\"} Hello",
 		},
 		"max labels length longer than input labels": {
 			&LogOutputOptions{Timezone: time.UTC, NoLabels: false},
 			timestamp,
-			&someLabels,
+			someLabels,
 			20,
 			"Hello",
 			"2006-01-02T08:04:05Z {type=\"test\"}        Hello",
 		},
 		"timezone option set to a Local one": {
 			&LogOutputOptions{Timezone: time.FixedZone("test", 2*60*60), NoLabels: false},
 			timestamp,
-			&someLabels,
+			someLabels,
 			0,
 			"Hello",
 			"2006-01-02T10:04:05+02:00 {type=\"test\"} Hello",
 		},
 		"labels output disabled": {
 			&LogOutputOptions{Timezone: time.UTC, NoLabels: true},
 			timestamp,
-			&someLabels,
+			someLabels,
 			0,
 			"Hello",
 			"2006-01-02T08:04:05Z Hello",
@@ -92,10 +94,17 @@ func TestDefaultOutput_FormatLabelsPadding(t *testing.T) {
 	t.Parallel()
 
 	// Define a list of labels that - once formatted - have a different length
-	labelsList := []labels.Labels{
-		labels.New(labels.Label{Name: "type", Value: "test"}),
-		labels.New(labels.Label{Name: "type", Value: "test"}, labels.Label{Name: "foo", Value: "bar"}),
-		labels.New(labels.Label{Name: "type", Value: "a-longer-test"}),
+	labelsList := []loghttp.LabelSet{
+		loghttp.LabelSet(map[string]string{
+			"type": "test",
+		}),
+		loghttp.LabelSet(map[string]string{
+			"type": "test",
+			"foo":  "bar",
+		}),
+		loghttp.LabelSet(map[string]string{
+			"type": "a-longer-test",
+		}),
 	}
 
 	timestamp, _ := time.Parse(time.RFC3339, "2006-01-02T15:04:05+07:00")
@@ -106,7 +115,7 @@ func TestDefaultOutput_FormatLabelsPadding(t *testing.T) {
 	// Format the same log line with different labels
 	formattedEntries := make([]string, 0, len(labelsList))
 	for _, lbls := range labelsList {
-		formattedEntries = append(formattedEntries, out.Format(timestamp, &lbls, maxLabelsLen, "XXX"))
+		formattedEntries = append(formattedEntries, out.Format(timestamp, lbls, maxLabelsLen, "XXX"))
 	}
 
 	// Ensure the log line starts at the same position in each formatted output
@@ -122,7 +131,7 @@ func TestDefaultOutput_FormatLabelsPadding(t *testing.T) {
 	}
 }
 
-func findMaxLabelsLength(labelsList []labels.Labels) int {
+func findMaxLabelsLength(labelsList []loghttp.LabelSet) int {
 	maxLabelsLen := 0
 
 	for _, lbls := range labelsList {

pkg/logcli/output/jsonl.go

@@ -5,7 +5,7 @@ import (
 	"log"
 	"time"
 
-	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/grafana/loki/pkg/loghttp"
 )
 
 // JSONLOutput prints logs and metadata as JSON Lines, suitable for scripts
@@ -14,7 +14,7 @@ type JSONLOutput struct {
 }
 
 // Format a log entry as json line
-func (o *JSONLOutput) Format(ts time.Time, lbls *labels.Labels, maxLabelsLen int, line string) string {
+func (o *JSONLOutput) Format(ts time.Time, lbls loghttp.LabelSet, maxLabelsLen int, line string) string {
 	entry := map[string]interface{}{
 		"timestamp": ts.In(o.options.Timezone),
 		"line":      line,