[query] Allow lookback duration to be set in query parameters

docs/query_engine/api/index.md

@@ -26,6 +26,7 @@ Query using PromQL and returns JSON datapoints compatible with the Prometheus Gr
 #### Optional
 
 - `debug=[bool]`
+- `lookback=[string|time duration]`: This sets the per request lookback duration to something other than the default set in config, can either be a time duration or the string "step" which sets the lookback to the same as the `step` request parameter.
 
 ### Header Params
 

scripts/development/m3_stack/start_m3.sh

@@ -68,11 +68,9 @@ else
     echo "Not running aggregator pipeline"
 fi
 
-
-echo "Sleeping to wait for nodes to initialize"
-sleep 10
-
-echo "Nodes online"
+echo "Wait for coordinator API to be up"
+ATTEMPTS=10 MAX_TIMEOUT=4 TIMEOUT=1 retry_with_backoff  \
+  'curl -vvvsSf localhost:7201/health'
 
 echo "Initializing namespaces"
 curl -vvvsSf -X POST localhost:7201/api/v1/namespace -d '{

scripts/docker-integration-tests/prometheus/test.sh

@@ -124,13 +124,13 @@ function test_query_limits_applied {
 }
 
 function prometheus_query_native {
-  local endpoint="$1"
-  local query="$2"
-  local params="$3"
-  local metrics_type="$4"
-  local metrics_storage_policy="$5"
-  local jq_path="$6"
-  local expected_value="$7"
+  local endpoint=${endpoint:-}
+  local query=${query:-}
+  local params=${params:-}
+  local metrics_type=${metrics_type:-}
+  local metrics_storage_policy=${metrics_storage_policy:-}
+  local jq_path=${jq_path:-}
+  local expected_value=${expected_value:-}
 
   params_prefixed=""
   if [[ "$params" != "" ]]; then
@@ -156,19 +156,27 @@ function test_query_restrict_metrics_type {
 
   # Test restricting to unaggregated metrics
   echo "Test query restrict to unaggregated metrics type (instant)"
-  prometheus_query_native query "$METRIC_NAME_TEST_RESTRICT_WRITE" "$params_instant" \
-    "unaggregated" "" "$jq_path_instant" "42.42"
+  ATTEMPTS=50 TIMEOUT=2 MAX_TIMEOUT=4 \
+    endpoint=query query="$METRIC_NAME_TEST_RESTRICT_WRITE" params="$params_instant" \
+    metrics_type="unaggregated" jq_path="$jq_path_instant" expected_value="42.42" \
+    retry_with_backoff prometheus_query_native
   echo "Test query restrict to unaggregated metrics type (range)"
-  prometheus_query_native query_range "$METRIC_NAME_TEST_RESTRICT_WRITE" "$params_range" \
-    "unaggregated" "" "$jq_path_range" "42.42"
+  ATTEMPTS=50 TIMEOUT=2 MAX_TIMEOUT=4 \
+    endpoint=query_range query="$METRIC_NAME_TEST_RESTRICT_WRITE" params="$params_range" \
+    metrics_type="unaggregated" jq_path="$jq_path_range" expected_value="42.42" \
+    retry_with_backoff prometheus_query_native
 
   # Test restricting to aggregated metrics
   echo "Test query restrict to aggregated metrics type (instant)"
-  prometheus_query_native query "$METRIC_NAME_TEST_RESTRICT_WRITE" "$params_instant" \
-    "aggregated" "15s:10h" "$jq_path_instant" "84.84"
+  ATTEMPTS=50 TIMEOUT=2 MAX_TIMEOUT=4 \
+    endpoint=query query="$METRIC_NAME_TEST_RESTRICT_WRITE" params="$params_instant" \
+    metrics_type="aggregated" metrics_storage_policy="15s:10h" jq_path="$jq_path_instant" expected_value="84.84" \
+    retry_with_backoff prometheus_query_native
   echo "Test query restrict to aggregated metrics type (range)"
-  prometheus_query_native query_range "$METRIC_NAME_TEST_RESTRICT_WRITE" "$params_range" \
-    "aggregated" "15s:10h" "$jq_path_range" "84.84"
+  ATTEMPTS=50 TIMEOUT=2 MAX_TIMEOUT=4 \
+    endpoint=query_range query="$METRIC_NAME_TEST_RESTRICT_WRITE" params="$params_range" \
+    metrics_type="aggregated" metrics_storage_policy="15s:10h" jq_path="$jq_path_range" expected_value="84.84" \
+    retry_with_backoff prometheus_query_native
 }
 
 # Run all tests

src/query/api/v1/handler/fetch_options.go

@@ -22,14 +22,27 @@ package handler
 
 import (
 	"fmt"
+	"math"
 	"net/http"
 	"strconv"
+	"strings"
+	"time"
 
 	"github.com/m3db/m3/src/metrics/policy"
+	"github.com/m3db/m3/src/query/errors"
 	"github.com/m3db/m3/src/query/storage"
 	xhttp "github.com/m3db/m3/src/x/net/http"
 )
 
+const (
+	// StepParam is the step parameter.
+	StepParam = "step"
+	// LookbackParam is the lookback parameter.
+	LookbackParam = "lookback"
+	maxInt64      = float64(math.MaxInt64)
+	minInt64      = float64(math.MinInt64)
+)
+
 // FetchOptionsBuilder builds fetch options based on a request and default
 // config.
 type FetchOptionsBuilder interface {
@@ -94,6 +107,24 @@ func (b fetchOptionsBuilder) NewFetchOptions(
 			return nil, xhttp.NewParseError(err, http.StatusBadRequest)
 		}
 	}
+
+	// NB(r): Eventually all query parameters that are common across the
+	// implementations should be parsed here so they can be fed to the engine.
+	if step, ok, err := ParseStep(req); err != nil {
+		err = fmt.Errorf(
+			"could not parse step: err=%v", err)
+		return nil, xhttp.NewParseError(err, http.StatusBadRequest)
+	} else if ok {
+		fetchOpts.Step = step
+	}
+	if lookback, ok, err := ParseLookbackDuration(req); err != nil {
+		err = fmt.Errorf(
+			"could not parse lookback: err=%v", err)
+		return nil, xhttp.NewParseError(err, http.StatusBadRequest)
+	} else if ok {
+		fetchOpts.LookbackDuration = &lookback
+	}
+
 	return fetchOpts, nil
 }
 
@@ -105,3 +136,81 @@ func newOrExistingRestrictFetchOptions(
 	}
 	return &storage.RestrictFetchOptions{}
 }
+
+// ParseStep parses the step duration for an HTTP request.
+func ParseStep(r *http.Request) (time.Duration, bool, error) {
+	step := r.FormValue(StepParam)
+	if step == "" {
+		return 0, false, nil
+	}
+	value, err := parseStep(r, StepParam)
+	if err != nil {
+		return 0, false, err
+	}
+	return value, true, err
+}
+
+func parseStep(r *http.Request, key string) (time.Duration, error) {
+	step, err := ParseDuration(r, key)
+	if err != nil {
+		return 0, err
+	}
+	if step <= 0 {
+		return 0, fmt.Errorf("expected postive step size, instead got: %d", step)
+	}
+	return step, nil
+}
+
+// ParseLookbackDuration parses a lookback duration for an HTTP request.
+func ParseLookbackDuration(r *http.Request) (time.Duration, bool, error) {
+	lookback := r.FormValue(LookbackParam)
+	if lookback == "" {
+		return 0, false, nil
+	}
+
+	if lookback == StepParam {
+		// Use the step size as lookback.
+		step, err := parseStep(r, StepParam)
+		if err != nil {
+			return 0, false, err
+		}
+		return step, true, nil
+	}
+
+	// Otherwise it is specified as duration value.
+	value, err := parseStep(r, LookbackParam)
+	if err != nil {
+		return 0, false, err
+	}
+
+	return value, true, nil
+}
+
+// ParseDuration parses a duration HTTP parameter.
+// nolint: unparam
+func ParseDuration(r *http.Request, key string) (time.Duration, error) {
+	str := strings.TrimSpace(r.FormValue(key))
+	if str == "" {
+		return 0, errors.ErrNotFound
+	}
+
+	value, durationErr := time.ParseDuration(str)
+	if durationErr == nil {
+		return value, nil
+	}
+
+	// Try parsing as a float value specifying seconds, the Prometheus default.
+	seconds, floatErr := strconv.ParseFloat(str, 64)
+	if floatErr == nil {
+		ts := seconds * float64(time.Second)
+		if ts > maxInt64 || ts < minInt64 {
+			return 0, fmt.Errorf("cannot parse duration='%s': as_float_err="+
+				"int64 overflow after float conversion", str)
+		}
+
+		return time.Duration(ts), nil
+	}
+
+	return 0, fmt.Errorf("cannot parse duration='%s': as_duration_err=%s, as_float_err=%s",
+		str, durationErr, floatErr)
+}

src/query/api/v1/handler/fetch_options_test.go

@@ -21,22 +21,34 @@
 package handler
 
 import (
+	"fmt"
+	"math"
+	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"testing"
+	"time"
 
 	"github.com/m3db/m3/src/metrics/policy"
 	"github.com/m3db/m3/src/query/storage"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func TestFetchOptionsBuilder(t *testing.T) {
+	type expectedLookback struct {
+		value time.Duration
+	}
+
 	tests := []struct {
 		name             string
 		defaultLimit     int
 		headers          map[string]string
+		query            string
 		expectedLimit    int
 		expectedRestrict *storage.RestrictFetchOptions
+		expectedLookback *expectedLookback
 		expectedErr      bool
 	}{
 		{
@@ -103,6 +115,35 @@ func TestFetchOptionsBuilder(t *testing.T) {
 			},
 			expectedErr: true,
 		},
+		{
+			name:  "can set lookback duration",
+			query: "lookback=10s",
+			expectedLookback: &expectedLookback{
+				value: 10 * time.Second,
+			},
+		},
+		{
+			name:  "can set lookback duration based on step",
+			query: "lookback=step&step=10s",
+			expectedLookback: &expectedLookback{
+				value: 10 * time.Second,
+			},
+		},
+		{
+			name:        "bad lookback returns error",
+			query:       "lookback=foo",
+			expectedErr: true,
+		},
+		{
+			name:        "loookback step but step is bad",
+			query:       "lookback=step&step=invalid",
+			expectedErr: true,
+		},
+		{
+			name:        "lookback step but step is negative",
+			query:       "lookback=step&step=-1",
+			expectedErr: true,
+		},
 	}
 
 	for _, test := range tests {
@@ -111,7 +152,11 @@ func TestFetchOptionsBuilder(t *testing.T) {
 				Limit: test.defaultLimit,
 			})
 
-			req := httptest.NewRequest("GET", "/foo", nil)
+			url := "/foo"
+			if test.query != "" {
+				url += "?" + test.query
+			}
+			req := httptest.NewRequest("GET", url, nil)
 			for k, v := range test.headers {
 				req.Header.Add(k, v)
 			}
@@ -127,9 +172,100 @@ func TestFetchOptionsBuilder(t *testing.T) {
 					require.NotNil(t, opts.RestrictFetchOptions)
 					require.Equal(t, *test.expectedRestrict, *opts.RestrictFetchOptions)
 				}
+				if test.expectedLookback == nil {
+					require.Nil(t, opts.LookbackDuration)
+				} else {
+					require.NotNil(t, opts.LookbackDuration)
+					require.Equal(t, test.expectedLookback.value, *opts.LookbackDuration)
+				}
 			} else {
 				require.Error(t, err)
 			}
 		})
 	}
 }
+
+func TestInvalidStep(t *testing.T) {
+	req := httptest.NewRequest("GET", "/foo", nil)
+	vals := make(url.Values)
+	vals.Del(StepParam)
+	vals.Add(StepParam, "-10.50s")
+	req.URL.RawQuery = vals.Encode()
+	_, _, err := ParseStep(req)
+	require.NotNil(t, err, "unable to parse request")
+}
+
+func TestParseLookbackDuration(t *testing.T) {
+	r := httptest.NewRequest(http.MethodGet, "/foo", nil)
+	_, ok, err := ParseLookbackDuration(r)
+	require.NoError(t, err)
+	require.False(t, ok)
+
+	r = httptest.NewRequest(http.MethodGet, "/foo?step=60s&lookback=step", nil)
+	v, ok, err := ParseLookbackDuration(r)
+	require.NoError(t, err)
+	require.True(t, ok)
+	assert.Equal(t, time.Minute, v)
+
+	r = httptest.NewRequest(http.MethodGet, "/foo?step=60s&lookback=120s", nil)
+	v, ok, err = ParseLookbackDuration(r)
+	require.NoError(t, err)
+	require.True(t, ok)
+	assert.Equal(t, 2*time.Minute, v)
+
+	r = httptest.NewRequest(http.MethodGet, "/foo?step=60s&lookback=foobar", nil)
+	_, _, err = ParseLookbackDuration(r)
+	require.Error(t, err)
+}
+
+func TestParseDuration(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, "/foo?step=10s", nil)
+	require.NoError(t, err)
+	v, err := ParseDuration(r, StepParam)
+	require.NoError(t, err)
+	assert.Equal(t, 10*time.Second, v)
+}
+
+func TestParseFloatDuration(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, "/foo?step=10.50m", nil)
+	require.NoError(t, err)
+	v, err := ParseDuration(r, StepParam)
+	require.NoError(t, err)
+	assert.Equal(t, 10*time.Minute+30*time.Second, v)
+
+	r = httptest.NewRequest(http.MethodGet, "/foo?step=10.00m", nil)
+	require.NoError(t, err)
+	v, err = ParseDuration(r, StepParam)
+	require.NoError(t, err)
+	assert.Equal(t, 10*time.Minute, v)
+}
+
+func TestParseDurationParsesIntAsSeconds(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, "/foo?step=30", nil)
+	require.NoError(t, err)
+	v, err := ParseDuration(r, StepParam)
+	require.NoError(t, err)
+	assert.Equal(t, 30*time.Second, v)
+}
+
+func TestParseDurationParsesFloatAsSeconds(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, "/foo?step=30.00", nil)
+	require.NoError(t, err)
+	v, err := ParseDuration(r, StepParam)
+	require.NoError(t, err)
+	assert.Equal(t, 30*time.Second, v)
+}
+
+func TestParseDurationError(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, "/foo?step=bar10", nil)
+	require.NoError(t, err)
+	_, err = ParseDuration(r, StepParam)
+	assert.Error(t, err)
+}
+
+func TestParseDurationOverflowError(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, fmt.Sprintf("/foo?step=%f", float64(math.MaxInt64)), nil)
+	require.NoError(t, err)
+	_, err = ParseDuration(r, StepParam)
+	assert.Error(t, err)
+}