Merge pull request prometheus#5503 from prometheus/release292

Cut 2.9.2

CHANGELOG.md

@@ -1,3 +1,11 @@
+## 2.9.2 / 2019-04-24
+
+* [BUGFIX] Make sure subquery range is taken into account for selection #5467
+* [BUGFIX] Exhaust every request body before closing it #5166
+* [BUGFIX] Cmd/promtool: return errors from rule evaluations #5483
+* [BUGFIX] Remote Storage: string interner should not panic in release #5487
+* [BUGFIX] Fix memory allocation regression in mergedPostings.Seek tsdb#586
+
 ## 2.9.1 / 2019-04-16
 
 * [BUGFIX] Discovery/kubernetes: fix missing label sanitization #5462

VERSION

@@ -1 +1 @@
-2.9.1
+2.9.2

cmd/promtool/unittest.go

@@ -210,6 +210,12 @@ func (tg *testGroup) test(mint, maxt time.Time, evalInterval time.Duration, grou
 			}
 			for _, g := range groups {
 				g.Eval(suite.Context(), ts)
+				for _, r := range g.Rules() {
+					if r.LastError() != nil {
+						errs = append(errs, errors.Errorf("    rule: %s, time: %s, err: %v",
+							r.Name(), ts.Sub(time.Unix(0, 0)), r.LastError()))
+					}
+				}
 			}
 		})
 		if len(errs) > 0 {

discovery/marathon/marathon.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math/rand"
 	"net"
@@ -306,7 +307,10 @@ func fetchApps(ctx context.Context, client *http.Client, url string) (*appList,
 	if err != nil {
 		return nil, err
 	}
-	defer resp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, resp.Body)
+		resp.Body.Close()
+	}()
 
 	if (resp.StatusCode < 200) || (resp.StatusCode >= 300) {
 		return nil, errors.Errorf("non 2xx status '%v' response during marathon service discovery", resp.StatusCode)

documentation/examples/custom-sd/adapter-usage/main.go

@@ -17,6 +17,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
@@ -98,7 +100,10 @@ func (d *discovery) parseServiceNodes(resp *http.Response, name string) (*target
 	}
 
 	dec := json.NewDecoder(resp.Body)
-	defer resp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, resp.Body)
+		resp.Body.Close()
+	}()
 	err := dec.Decode(&nodes)
 
 	if err != nil {

documentation/examples/remote_storage/remote_storage_adapter/opentsdb/client.go

@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"io"
 	"io/ioutil"
 	"math"
 	"net/http"
@@ -114,7 +115,10 @@ func (c *Client) Write(samples model.Samples) error {
 	if err != nil {
 		return err
 	}
-	defer resp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, resp.Body)
+		resp.Body.Close()
+	}()
 
 	// API returns status code 204 for successful writes.
 	// http://opentsdb.net/docs/build/html/api_http/put.html

go.mod

@@ -79,7 +79,7 @@ require (
 	github.com/prometheus/client_golang v0.9.3-0.20190127221311-3c4408c8b829
 	github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f
 	github.com/prometheus/common v0.3.0
-	github.com/prometheus/tsdb v0.7.0
+	github.com/prometheus/tsdb v0.7.1
 	github.com/rlmcpherson/s3gof3r v0.5.0 // indirect
 	github.com/rubyist/circuitbreaker v2.2.1+incompatible // indirect
 	github.com/samuel/go-zookeeper v0.0.0-20161028232340-1d7be4effb13

go.sum

@@ -277,8 +277,8 @@ github.com/prometheus/common v0.3.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190117184657-bf6a532e95b1 h1:/K3IL0Z1quvmJ7X0A1AwNEK7CRkVK3YwfOU/QAL4WGg=
 github.com/prometheus/procfs v0.0.0-20190117184657-bf6a532e95b1/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
-github.com/prometheus/tsdb v0.7.0 h1:2rQ5LUp0GGnSOMyZTEnEE82YDAAcQ2d4M2nGeUBH5tg=
-github.com/prometheus/tsdb v0.7.0/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/prometheus/tsdb v0.7.1 h1:YZcsG11NqnK4czYLrWd9mpEuAJIHVQLwdrleYfszMAA=
+github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rlmcpherson/s3gof3r v0.5.0 h1:1izOJpTiohSibfOHuNyEA/yQnAirh05enzEdmhez43k=

notifier/notifier.go

@@ -19,6 +19,8 @@ import (
 	"crypto/md5"
 	"encoding/json"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"net/url"
@@ -508,7 +510,10 @@ func (n *Manager) sendOne(ctx context.Context, c *http.Client, url string, b []b
 	if err != nil {
 		return err
 	}
-	defer resp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, resp.Body)
+		resp.Body.Close()
+	}()
 
 	// Any HTTP status 2xx is OK.
 	if resp.StatusCode/100 != 2 {

notifier/notifier_test.go

@@ -14,6 +14,7 @@
 package notifier
 
 import (
+	"bytes"
 	"context"
 	"crypto/md5"
 	"encoding/json"
@@ -224,7 +225,7 @@ func TestCustomDo(t *testing.T) {
 			testutil.Equals(t, testURL, req.URL.String())
 
 			return &http.Response{
-				Body: ioutil.NopCloser(nil),
+				Body: ioutil.NopCloser(bytes.NewBuffer(nil)),
 			}, nil
 		},
 	}, nil)

promql/engine.go

@@ -552,6 +552,14 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev
 			Step:  durationToInt64Millis(s.Interval),
 		}
 
+		// We need to make sure we select the timerange selected by the subquery.
+		// TODO(gouthamve): cumulativeSubqueryOffset gives the sum of range and the offset
+		// we can optimise it by separating out the range and offsets, and substracting the offsets
+		// from end also.
+		subqOffset := ng.cumulativeSubqueryOffset(path)
+		offsetMilliseconds := durationMilliseconds(subqOffset)
+		params.Start = params.Start - offsetMilliseconds
+
 		switch n := node.(type) {
 		case *VectorSelector:
 			params.Start = params.Start - durationMilliseconds(LookbackDelta)

promql/engine_test.go

@@ -227,6 +227,166 @@ func TestQueryError(t *testing.T) {
 	}
 }
 
+// paramCheckerQuerier implements storage.Querier which checks the start and end times
+// in params.
+type paramCheckerQuerier struct {
+	start int64
+	end   int64
+
+	t *testing.T
+}
+
+func (q *paramCheckerQuerier) Select(sp *storage.SelectParams, _ ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
+	testutil.Equals(q.t, q.start, sp.Start)
+	testutil.Equals(q.t, q.end, sp.End)
+
+	return errSeriesSet{err: nil}, nil, nil
+}
+func (*paramCheckerQuerier) LabelValues(name string) ([]string, error) { return nil, nil }
+func (*paramCheckerQuerier) LabelNames() ([]string, error)             { return nil, nil }
+func (*paramCheckerQuerier) Close() error                              { return nil }
+
+func TestParamsSetCorrectly(t *testing.T) {
+	opts := EngineOpts{
+		Logger:        nil,
+		Reg:           nil,
+		MaxConcurrent: 10,
+		MaxSamples:    10,
+		Timeout:       10 * time.Second,
+	}
+
+	// Set the lookback to be smaller and reset at the end.
+	currLookback := LookbackDelta
+	LookbackDelta = 5 * time.Second
+	defer func() {
+		LookbackDelta = currLookback
+	}()
+
+	cases := []struct {
+		query string
+
+		// All times are in seconds.
+		start int64
+		end   int64
+
+		paramStart int64
+		paramEnd   int64
+	}{{
+		query: "foo",
+		start: 10,
+
+		paramStart: 5,
+		paramEnd:   10,
+	}, {
+		query: "foo[2m]",
+		start: 200,
+
+		paramStart: 80, // 200 - 120
+		paramEnd:   200,
+	}, {
+		query: "foo[2m] offset 2m",
+		start: 300,
+
+		paramStart: 60,
+		paramEnd:   180,
+	}, {
+		query: "foo[2m:1s]",
+		start: 300,
+
+		paramStart: 175, // 300 - 120 - 5
+		paramEnd:   300,
+	}, {
+		query: "count_over_time(foo[2m:1s])",
+		start: 300,
+
+		paramStart: 175, // 300 - 120 - 5
+		paramEnd:   300,
+	}, {
+		query: "count_over_time(foo[2m:1s] offset 10s)",
+		start: 300,
+
+		paramStart: 165, // 300 - 120 - 5 - 10
+		paramEnd:   300,
+	}, {
+		query: "count_over_time((foo offset 10s)[2m:1s] offset 10s)",
+		start: 300,
+
+		paramStart: 155, // 300 - 120 - 5 - 10 - 10
+		paramEnd:   290,
+	}, {
+		// Range queries now.
+		query: "foo",
+		start: 10,
+		end:   20,
+
+		paramStart: 5,
+		paramEnd:   20,
+	}, {
+		query: "rate(foo[2m])",
+		start: 200,
+		end:   500,
+
+		paramStart: 80, // 200 - 120
+		paramEnd:   500,
+	}, {
+		query: "rate(foo[2m] offset 2m)",
+		start: 300,
+		end:   500,
+
+		paramStart: 60,
+		paramEnd:   380,
+	}, {
+		query: "rate(foo[2m:1s])",
+		start: 300,
+		end:   500,
+
+		paramStart: 175, // 300 - 120 - 5
+		paramEnd:   500,
+	}, {
+		query: "count_over_time(foo[2m:1s])",
+		start: 300,
+		end:   500,
+
+		paramStart: 175, // 300 - 120 - 5
+		paramEnd:   500,
+	}, {
+		query: "count_over_time(foo[2m:1s] offset 10s)",
+		start: 300,
+		end:   500,
+
+		paramStart: 165, // 300 - 120 - 5 - 10
+		paramEnd:   500,
+	}, {
+		query: "count_over_time((foo offset 10s)[2m:1s] offset 10s)",
+		start: 300,
+		end:   500,
+
+		paramStart: 155, // 300 - 120 - 5 - 10 - 10
+		paramEnd:   490,
+	}}
+
+	for _, tc := range cases {
+		engine := NewEngine(opts)
+		queryable := storage.QueryableFunc(func(ctx context.Context, mint, maxt int64) (storage.Querier, error) {
+			return &paramCheckerQuerier{start: tc.paramStart * 1000, end: tc.paramEnd * 1000, t: t}, nil
+		})
+
+		var (
+			query Query
+			err   error
+		)
+		if tc.end == 0 {
+			query, err = engine.NewInstantQuery(queryable, tc.query, time.Unix(tc.start, 0))
+		} else {
+			query, err = engine.NewRangeQuery(queryable, tc.query, time.Unix(tc.start, 0), time.Unix(tc.end, 0), time.Second)
+		}
+		testutil.Ok(t, err)
+
+		res := query.Exec(context.Background())
+		testutil.Ok(t, res.Err)
+	}
+}
+
 func TestEngineShutdown(t *testing.T) {
 	opts := EngineOpts{
 		Logger:        nil,

scrape/scrape.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"net/http"
 	"sync"
@@ -543,7 +544,10 @@ func (s *targetScraper) scrape(ctx context.Context, w io.Writer) (string, error)
 	if err != nil {
 		return "", err
 	}
-	defer resp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, resp.Body)
+		resp.Body.Close()
+	}()
 
 	if resp.StatusCode != http.StatusOK {
 		return "", errors.Errorf("server returned HTTP status %s", resp.Status)

storage/remote/client.go

@@ -95,7 +95,10 @@ func (c *Client) Store(ctx context.Context, req []byte) error {
 		// recoverable.
 		return recoverableError{err}
 	}
-	defer httpResp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, httpResp.Body)
+		httpResp.Body.Close()
+	}()
 
 	if httpResp.StatusCode/100 != 2 {
 		scanner := bufio.NewScanner(io.LimitReader(httpResp.Body, maxErrMsgLen))
@@ -148,7 +151,10 @@ func (c *Client) Read(ctx context.Context, query *prompb.Query) (*prompb.QueryRe
 	if err != nil {
 		return nil, errors.Wrap(err, "error sending request")
 	}
-	defer httpResp.Body.Close()
+	defer func() {
+		io.Copy(ioutil.Discard, httpResp.Body)
+		httpResp.Body.Close()
+	}()
 	if httpResp.StatusCode/100 != 2 {
 		return nil, errors.Errorf("server returned HTTP status %s", httpResp.Status)
 	}