Improves tests.

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

pkg/querier/queryrange/codec.go

@@ -55,8 +55,8 @@ func (codec) DecodeRequest(_ context.Context, r *http.Request) (queryrange.Reque
 		Query:     req.Query,
 		Limit:     req.Limit,
 		Direction: req.Direction,
-		StartTs:   req.Start,
-		EndTs:     req.End,
+		StartTs:   req.Start.UTC(),
+		EndTs:     req.End.UTC(),
 		// GetStep must return milliseconds
 		Step: int64(req.Step) / 1e6,
 		Path: r.URL.Path,
@@ -76,7 +76,7 @@ func (codec) EncodeRequest(ctx context.Context, r queryrange.Request) (*http.Req
 		"limit":     []string{fmt.Sprintf("%d", lokiReq.Limit)},
 	}
 	if lokiReq.Step != 0 {
-		params["step"] = []string{fmt.Sprintf("%d", lokiReq.Step/int64(1e3))}
+		params["step"] = []string{fmt.Sprintf("%f", float64(lokiReq.Step)/float64(1e3))}
 	}
 	u := &url.URL{
 		// the request could come /api/prom/query but we want to only use the new api.

pkg/querier/queryrange/codec_test.go

@@ -0,0 +1,330 @@
+package queryrange
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"reflect"
+	strings "strings"
+	"testing"
+	"time"
+
+	"github.com/cortexproject/cortex/pkg/ingester/client"
+	"github.com/cortexproject/cortex/pkg/querier/queryrange"
+	"github.com/grafana/loki/pkg/loghttp"
+	"github.com/grafana/loki/pkg/logproto"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	start = testTime //  Marshalling the time drops the monotonic clock so we can't use time.Now
+	end   = start.Add(1 * time.Hour)
+)
+
+func Test_codec_DecodeRequest(t *testing.T) {
+	tests := []struct {
+		name       string
+		reqBuilder func() (*http.Request, error)
+		want       queryrange.Request
+		wantErr    bool
+	}{
+		{"wrong", func() (*http.Request, error) { return http.NewRequest(http.MethodGet, "/bad?step=bad", nil) }, nil, true},
+		{"ok", func() (*http.Request, error) {
+			return http.NewRequest(http.MethodGet,
+				fmt.Sprintf(`/query_range?start=%d&end=%d&query={foo="bar"}&step=1&limit=200&direction=FORWARD`, start.UnixNano(), end.UnixNano()), nil)
+		}, &LokiRequest{
+			Query:     `{foo="bar"}`,
+			Limit:     200,
+			Step:      1000, // step is expected in ms.
+			Direction: logproto.FORWARD,
+			Path:      "/query_range",
+			StartTs:   start,
+			EndTs:     end,
+		}, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req, err := tt.reqBuilder()
+			if err != nil {
+				t.Fatal(err)
+			}
+			got, err := lokiCodec.DecodeRequest(context.TODO(), req)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("codec.DecodeRequest() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			require.Equal(t, got, tt.want)
+		})
+	}
+}
+
+func Test_codec_DecodeResponse(t *testing.T) {
+	tests := []struct {
+		name    string
+		res     *http.Response
+		req     queryrange.Request
+		want    queryrange.Response
+		wantErr bool
+	}{
+		{"500", &http.Response{StatusCode: 500, Body: ioutil.NopCloser(strings.NewReader("some error"))}, nil, nil, true},
+		{"no body", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(badReader{})}, nil, nil, true},
+		{"bad json", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader(""))}, nil, nil, true},
+		{"not success", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader(`{"status":"fail"}`))}, nil, nil, true},
+		{"unknown", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader(`{"status":"success"}`))}, nil, nil, true},
+		{"matrix", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader(matrixString))}, nil,
+			&queryrange.PrometheusResponse{
+				Status: loghttp.QueryStatusSuccess,
+				Data: queryrange.PrometheusData{
+					ResultType: loghttp.ResultTypeMatrix,
+					Result:     sampleStreams,
+				},
+			}, false},
+		{"streams v1", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader(streamsString))},
+			&LokiRequest{Direction: logproto.FORWARD, Limit: 100, Path: "/loki/api/v1/query_range"},
+			&LokiResponse{
+				Status:    loghttp.QueryStatusSuccess,
+				Direction: logproto.FORWARD,
+				Limit:     100,
+				Version:   uint32(loghttp.VersionV1),
+				Data: LokiData{
+					ResultType: loghttp.ResultTypeStream,
+					Result:     logStreams,
+				},
+			}, false},
+		{"streams legacy", &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader(streamsString))},
+			&LokiRequest{Direction: logproto.FORWARD, Limit: 100, Path: "/api/prom/query_range"},
+			&LokiResponse{
+				Status:    loghttp.QueryStatusSuccess,
+				Direction: logproto.FORWARD,
+				Limit:     100,
+				Version:   uint32(loghttp.VersionLegacy),
+				Data: LokiData{
+					ResultType: loghttp.ResultTypeStream,
+					Result:     logStreams,
+				},
+			}, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := lokiCodec.DecodeResponse(context.TODO(), tt.res, tt.req)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("codec.DecodeResponse() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("codec.DecodeResponse() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_codec_EncodeRequest(t *testing.T) {
+
+	// we only accept LokiRequest.
+	got, err := lokiCodec.EncodeRequest(context.TODO(), &queryrange.PrometheusRequest{})
+	require.Error(t, err)
+	require.Nil(t, got)
+
+	ctx := context.Background()
+	toEncode := &LokiRequest{
+		Query:     `{foo="bar"}`,
+		Limit:     200,
+		Step:      1010,
+		Direction: logproto.FORWARD,
+		Path:      "/query_range",
+		StartTs:   start,
+		EndTs:     end,
+	}
+	got, err = lokiCodec.EncodeRequest(ctx, toEncode)
+	require.NoError(t, err)
+	require.Equal(t, ctx, got.Context())
+	require.Equal(t, "/loki/api/v1/query_range", got.URL.Path)
+	require.Equal(t, fmt.Sprintf("%d", start.UnixNano()), got.URL.Query().Get("start"))
+	require.Equal(t, fmt.Sprintf("%d", end.UnixNano()), got.URL.Query().Get("end"))
+	require.Equal(t, `{foo="bar"}`, got.URL.Query().Get("query"))
+	require.Equal(t, fmt.Sprintf("%d", 200), got.URL.Query().Get("limit"))
+	require.Equal(t, `FORWARD`, got.URL.Query().Get("direction"))
+	require.Equal(t, "1.010000", got.URL.Query().Get("step"))
+
+	// testing a full roundtrip
+	req, err := lokiCodec.DecodeRequest(context.TODO(), got)
+	require.NoError(t, err)
+	require.Equal(t, toEncode.Query, req.(*LokiRequest).Query)
+	require.Equal(t, toEncode.Step, req.(*LokiRequest).Step)
+	require.Equal(t, toEncode.StartTs, req.(*LokiRequest).StartTs)
+	require.Equal(t, toEncode.EndTs, req.(*LokiRequest).EndTs)
+	require.Equal(t, toEncode.Direction, req.(*LokiRequest).Direction)
+	require.Equal(t, toEncode.Limit, req.(*LokiRequest).Limit)
+	require.Equal(t, "/loki/api/v1/query_range", req.(*LokiRequest).Path)
+}
+
+func Test_codec_EncodeResponse(t *testing.T) {
+
+	tests := []struct {
+		name    string
+		res     queryrange.Response
+		body    string
+		wantErr bool
+	}{
+		{"error", &badResponse{}, "", true},
+		{"prom", &queryrange.PrometheusResponse{
+			Status: loghttp.QueryStatusSuccess,
+			Data: queryrange.PrometheusData{
+				ResultType: loghttp.ResultTypeMatrix,
+				Result:     sampleStreams,
+			},
+		}, matrixString, false},
+		{"loki v1",
+			&LokiResponse{
+				Status:    loghttp.QueryStatusSuccess,
+				Direction: logproto.FORWARD,
+				Limit:     100,
+				Version:   uint32(loghttp.VersionV1),
+				Data: LokiData{
+					ResultType: loghttp.ResultTypeStream,
+					Result:     logStreams,
+				},
+			}, streamsString, false},
+		{"loki legacy",
+			&LokiResponse{
+				Status:    loghttp.QueryStatusSuccess,
+				Direction: logproto.FORWARD,
+				Limit:     100,
+				Version:   uint32(loghttp.VersionLegacy),
+				Data: LokiData{
+					ResultType: loghttp.ResultTypeStream,
+					Result:     logStreams,
+				},
+			}, streamsStringLegacy, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := lokiCodec.EncodeResponse(context.TODO(), tt.res)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("codec.EncodeResponse() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if err == nil {
+				require.Equal(t, 200, got.StatusCode)
+				body, err := ioutil.ReadAll(got.Body)
+				require.Nil(t, err)
+				bodyString := string(body)
+				require.JSONEq(t, tt.body, bodyString)
+			}
+		})
+	}
+}
+
+func Test_codec_MergeResponse(t *testing.T) {
+	tests := []struct {
+		name      string
+		responses []queryrange.Response
+		want      queryrange.Response
+		wantErr   bool
+	}{
+		{"one", []queryrange.Response{}, nil, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := lokiCodec.MergeResponse(tt.responses...)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("codec.MergeResponse() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("codec.MergeResponse() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+type badResponse struct{}
+
+func (badResponse) Reset()         {}
+func (badResponse) String() string { return "noop" }
+func (badResponse) ProtoMessage()  {}
+
+type badReader struct{}
+
+func (badReader) Read(p []byte) (n int, err error) {
+	return 0, errors.New("")
+}
+
+var (
+	matrixString = `{
+	"data": {
+	  "resultType": "matrix",
+	  "result": [
+		{
+		  "metric": {
+			"filename": "\/var\/hostlog\/apport.log",
+			"job": "varlogs"
+		  },
+		  "values": [
+			  [
+				1568404331.324,
+				"0.013333333333333334"
+			  ]
+			]
+		},
+		{
+		  "metric": {
+			"filename": "\/var\/hostlog\/syslog",
+			"job": "varlogs"
+		  },
+		  "values": [
+				[
+					1568404331.324,
+					"3.45"
+				],
+				[
+					1568404331.339,
+					"4.45"
+				]
+			]
+		}
+	  ]
+	},
+	"status": "success"
+  }`
+	sampleStreams = []queryrange.SampleStream{
+		queryrange.SampleStream{
+			Labels:  []client.LabelAdapter{{Name: "filename", Value: "/var/hostlog/apport.log"}, {Name: "job", Value: "varlogs"}},
+			Samples: []client.Sample{{Value: 0.013333333333333334, TimestampMs: 1568404331324}},
+		},
+		queryrange.SampleStream{
+			Labels:  []client.LabelAdapter{{Name: "filename", Value: "/var/hostlog/syslog"}, {Name: "job", Value: "varlogs"}},
+			Samples: []client.Sample{{Value: 3.45, TimestampMs: 1568404331324}, {Value: 4.45, TimestampMs: 1568404331339}},
+		},
+	}
+	streamsString = `{
+		"status": "success",
+		"data": {
+			"resultType": "streams",
+			"result": [
+				{
+					"stream": {
+						"test": "test"
+					},
+					"values":[
+						[ "123456789012345", "super line" ]
+					]
+				}
+			]
+		}
+	}`
+	streamsStringLegacy = `{"streams":[{"labels":"{test=\"test\"}","entries":[{"ts":"1970-01-02T05:17:36.789012345-05:00","line":"super line"}]}]}`
+	logStreams          = []logproto.Stream{
+		logproto.Stream{
+			Labels: `{test="test"}`,
+			Entries: []logproto.Entry{
+				logproto.Entry{
+					Line:      "super line",
+					Timestamp: time.Unix(0, 123456789012345),
+				},
+			},
+		},
+	}
+)

pkg/querier/queryrange/roundtrip.go

@@ -11,6 +11,7 @@ import (
 	"github.com/grafana/loki/pkg/logql"
 )
 
+// Config is the configuration for the queryrange tripperware
 type Config struct {
 	queryrange.Config `yaml:",inline"`
 	IntervalBatchSize int `yaml:"interval_batch_size"`
@@ -22,6 +23,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.IntervalBatchSize, "interval-batch-size", 16, "The number of batched requests per interval. (only applied to regex)")
 }
 
+// Stopper gracefully shutdown resources created
 type Stopper interface {
 	Stop() error
 }