Merge branch 'main' into sparsehistogram

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 2.32.1 / 2021-12-17
+
+* [BUGFIX] Scrape: Fix reporting metrics when sample limit is reached during the report. #9996
+* [BUGFIX] Scrape: Ensure that scrape interval and scrape timeout are always set. #10023
+* [BUGFIX] TSDB: Expose and fix bug in iterators' `Seek()` method. #10030
+
 ## 2.32.0 / 2021-12-09
 
 This release introduces the Prometheus Agent, a new mode of operation for

VERSION

@@ -1 +1 @@
-2.32.0
+2.32.1

cmd/promtool/sd_test.go

@@ -15,6 +15,7 @@ package main
 
 import (
 	"testing"
+	"time"
 
 	"github.com/prometheus/common/model"
 
@@ -37,6 +38,8 @@ func TestSDCheckResult(t *testing.T) {
 	require.Nil(t, err)
 
 	scrapeConfig := &config.ScrapeConfig{
+		ScrapeInterval: model.Duration(1 * time.Minute),
+		ScrapeTimeout:  model.Duration(10 * time.Second),
 		RelabelConfigs: []*relabel.Config{{
 			SourceLabels: model.LabelNames{"foo"},
 			Action:       relabel.Replace,
@@ -50,14 +53,14 @@ func TestSDCheckResult(t *testing.T) {
 		{
 			DiscoveredLabels: labels.Labels{
 				labels.Label{Name: "__address__", Value: "localhost:8080"},
-				labels.Label{Name: "__scrape_interval__", Value: "0s"},
+				labels.Label{Name: "__scrape_interval__", Value: "1m"},
-				labels.Label{Name: "__scrape_timeout__", Value: "0s"},
+				labels.Label{Name: "__scrape_timeout__", Value: "10s"},
 				labels.Label{Name: "foo", Value: "bar"},
 			},
 			Labels: labels.Labels{
 				labels.Label{Name: "__address__", Value: "localhost:8080"},
-				labels.Label{Name: "__scrape_interval__", Value: "0s"},
+				labels.Label{Name: "__scrape_interval__", Value: "1m"},
-				labels.Label{Name: "__scrape_timeout__", Value: "0s"},
+				labels.Label{Name: "__scrape_timeout__", Value: "10s"},
 				labels.Label{Name: "foo", Value: "bar"},
 				labels.Label{Name: "instance", Value: "localhost:8080"},
 				labels.Label{Name: "newfoo", Value: "bar"},

docs/configuration/template_reference.md

@@ -70,6 +70,7 @@ versions.
 | title         | string        | string  | [strings.Title](https://golang.org/pkg/strings/#Title), capitalises first character of each word.|
 | toUpper       | string        | string  | [strings.ToUpper](https://golang.org/pkg/strings/#ToUpper), converts all characters to upper case.|
 | toLower       | string        | string  | [strings.ToLower](https://golang.org/pkg/strings/#ToLower), converts all characters to lower case.|
+| stripPort     | string        | string  | [net.SplitHostPort](https://pkg.go.dev/net#SplitHostPort), splits string into host and port, then returns only host.|
 | match         | pattern, text | boolean | [regexp.MatchString](https://golang.org/pkg/regexp/#MatchString) Tests for a unanchored regexp match. |
 | reReplaceAll  | pattern, replacement, text | string | [Regexp.ReplaceAllString](https://golang.org/pkg/regexp/#Regexp.ReplaceAllString) Regexp substitution, unanchored. |
 | graphLink  | expr | string | Returns path to graph view in the [expression browser](https://prometheus.io/docs/visualization/browser/) for the expression. |

scrape/scrape.go

@@ -306,7 +306,7 @@ func newScrapePool(cfg *config.ScrapeConfig, app storage.Appendable, jitterSeed
 				return mutateSampleLabels(l, opts.target, opts.honorLabels, opts.mrc)
 			},
 			func(l labels.Labels) labels.Labels { return mutateReportSampleLabels(l, opts.target) },
-			func(ctx context.Context) storage.Appender { return appender(app.Appender(ctx), opts.sampleLimit) },
+			func(ctx context.Context) storage.Appender { return app.Appender(ctx) },
 			cache,
 			jitterSeed,
 			opts.honorTimestamps,
@@ -1427,6 +1427,9 @@ func (sl *scrapeLoop) append(app storage.Appender, b []byte, contentType string,
 		e              exemplar.Exemplar // escapes to heap so hoisted out of loop
 	)
 
+	// Take an appender with limits.
+	app = appender(app, sl.sampleLimit)
+
 	defer func() {
 		if err != nil {
 			return

scrape/scrape_test.go

@@ -464,7 +464,7 @@ func TestScrapePoolAppender(t *testing.T) {
 	appl, ok := loop.(*scrapeLoop)
 	require.True(t, ok, "Expected scrapeLoop but got %T", loop)
 
-	wrapped := appl.appender(context.Background())
+	wrapped := appender(appl.appender(context.Background()), 0)
 
 	tl, ok := wrapped.(*timeLimitAppender)
 	require.True(t, ok, "Expected timeLimitAppender but got %T", wrapped)
@@ -472,20 +472,21 @@ func TestScrapePoolAppender(t *testing.T) {
 	_, ok = tl.Appender.(nopAppender)
 	require.True(t, ok, "Expected base appender but got %T", tl.Appender)
 
+	sampleLimit := 100
 	loop = sp.newLoop(scrapeLoopOptions{
 		target:      &Target{},
-		sampleLimit: 100,
+		sampleLimit: sampleLimit,
 	})
 	appl, ok = loop.(*scrapeLoop)
 	require.True(t, ok, "Expected scrapeLoop but got %T", loop)
 
-	wrapped = appl.appender(context.Background())
+	wrapped = appender(appl.appender(context.Background()), sampleLimit)
 
 	sl, ok := wrapped.(*limitAppender)
 	require.True(t, ok, "Expected limitAppender but got %T", wrapped)
 
 	tl, ok = sl.Appender.(*timeLimitAppender)
-	require.True(t, ok, "Expected limitAppender but got %T", sl.Appender)
+	require.True(t, ok, "Expected timeLimitAppender but got %T", sl.Appender)
 
 	_, ok = tl.Appender.(nopAppender)
 	require.True(t, ok, "Expected base appender but got %T", tl.Appender)
@@ -2773,6 +2774,72 @@ func TestScrapeReportSingleAppender(t *testing.T) {
 	}
 }
 
+func TestScrapeReportLimit(t *testing.T) {
+	s := teststorage.New(t)
+	defer s.Close()
+
+	cfg := &config.ScrapeConfig{
+		JobName:        "test",
+		SampleLimit:    5,
+		Scheme:         "http",
+		ScrapeInterval: model.Duration(100 * time.Millisecond),
+		ScrapeTimeout:  model.Duration(100 * time.Millisecond),
+	}
+
+	var (
+		scrapes      int
+		scrapedTwice = make(chan bool)
+	)
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, "metric_a 44\nmetric_b 44\nmetric_c 44\nmetric_d 44\n")
+		scrapes++
+		if scrapes == 2 {
+			close(scrapedTwice)
+		}
+	}))
+	defer ts.Close()
+
+	sp, err := newScrapePool(cfg, s, 0, nil, false)
+	require.NoError(t, err)
+	defer sp.stop()
+
+	testURL, err := url.Parse(ts.URL)
+	require.NoError(t, err)
+	sp.Sync([]*targetgroup.Group{
+		{
+			Targets: []model.LabelSet{{model.AddressLabel: model.LabelValue(testURL.Host)}},
+		},
+	})
+
+	select {
+	case <-time.After(5 * time.Second):
+		t.Fatalf("target was not scraped twice")
+	case <-scrapedTwice:
+		// If the target has been scraped twice, report samples from the first
+		// scrape have been inserted in the database.
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	q, err := s.Querier(ctx, time.Time{}.UnixNano(), time.Now().UnixNano())
+	require.NoError(t, err)
+	defer q.Close()
+	series := q.Select(false, nil, labels.MustNewMatcher(labels.MatchRegexp, "__name__", "up"))
+
+	var found bool
+	for series.Next() {
+		i := series.At().Iterator()
+		for i.Next() {
+			_, v := i.At()
+			require.Equal(t, 1.0, v)
+			found = true
+		}
+	}
+
+	require.True(t, found)
+}
+
 func TestScrapeLoopLabelLimit(t *testing.T) {
 	tests := []struct {
 		title           string

scrape/target.go

@@ -418,28 +418,22 @@ func PopulateLabels(lset labels.Labels, cfg *config.ScrapeConfig) (res, orig lab
 		return nil, nil, err
 	}
 
-	var interval string
+	interval := lset.Get(model.ScrapeIntervalLabel)
-	var intervalDuration model.Duration
+	intervalDuration, err := model.ParseDuration(interval)
-	if interval = lset.Get(model.ScrapeIntervalLabel); interval != cfg.ScrapeInterval.String() {
+	if err != nil {
-		intervalDuration, err = model.ParseDuration(interval)
+		return nil, nil, errors.Errorf("error parsing scrape interval: %v", err)
-		if err != nil {
+	}
-			return nil, nil, errors.Errorf("error parsing scrape interval: %v", err)
+	if time.Duration(intervalDuration) == 0 {
-		}
+		return nil, nil, errors.New("scrape interval cannot be 0")
-		if time.Duration(intervalDuration) == 0 {
-			return nil, nil, errors.New("scrape interval cannot be 0")
-		}
 	}
 
-	var timeout string
+	timeout := lset.Get(model.ScrapeTimeoutLabel)
-	var timeoutDuration model.Duration
+	timeoutDuration, err := model.ParseDuration(timeout)
-	if timeout = lset.Get(model.ScrapeTimeoutLabel); timeout != cfg.ScrapeTimeout.String() {
+	if err != nil {
-		timeoutDuration, err = model.ParseDuration(timeout)
+		return nil, nil, errors.Errorf("error parsing scrape timeout: %v", err)
-		if err != nil {
+	}
-			return nil, nil, errors.Errorf("error parsing scrape timeout: %v", err)
+	if time.Duration(timeoutDuration) == 0 {
-		}
+		return nil, nil, errors.New("scrape timeout cannot be 0")
-		if time.Duration(timeoutDuration) == 0 {
-			return nil, nil, errors.New("scrape timeout cannot be 0")
-		}
 	}
 
 	if timeoutDuration > intervalDuration {

scrape/target_test.go

@@ -371,7 +371,11 @@ func TestNewClientWithBadTLSConfig(t *testing.T) {
 func TestTargetsFromGroup(t *testing.T) {
 	expectedError := "instance 0 in group : no address"
 
-	targets, failures := TargetsFromGroup(&targetgroup.Group{Targets: []model.LabelSet{{}, {model.AddressLabel: "localhost:9090"}}}, &config.ScrapeConfig{})
+	cfg := config.ScrapeConfig{
+		ScrapeTimeout:  model.Duration(10 * time.Second),
+		ScrapeInterval: model.Duration(1 * time.Minute),
+	}
+	targets, failures := TargetsFromGroup(&targetgroup.Group{Targets: []model.LabelSet{{}, {model.AddressLabel: "localhost:9090"}}}, &cfg)
 	if len(targets) != 1 {
 		t.Fatalf("Expected 1 target, got %v", len(targets))
 	}

storage/remote/codec.go

@@ -362,6 +362,9 @@ func (c *concreteSeriesIterator) Seek(t int64) chunkenc.ValueType {
 	if c.cur == -1 {
 		c.cur = 0
 	}
+	if c.cur >= len(c.series.samples) {
+		return false
+	}
 	// No-op check.
 	if s := c.series.samples[c.cur]; s.Timestamp >= t {
 		return chunkenc.ValFloat

storage/remote/codec_test.go

@@ -234,6 +234,11 @@ func TestConcreteSeriesIterator(t *testing.T) {
 	ts, v = it.At()
 	require.Equal(t, int64(3), ts)
 	require.Equal(t, 3., v)
+
+	// Seek beyond the end.
+	require.False(t, it.Seek(5))
+	// And we don't go back. (This exposes issue #10027.)
+	require.False(t, it.Seek(2))
 }
 
 func TestFromQueryResultWithDuplicates(t *testing.T) {

storage/series.go

@@ -118,6 +118,9 @@ func (it *listSeriesIterator) Seek(t int64) chunkenc.ValueType {
 	if it.idx == -1 {
 		it.idx = 0
 	}
+	if it.idx >= it.samples.Len() {
+		return false
+	}
 	// No-op check.
 	if s := it.samples.Get(it.idx); s.T() >= t {
 		return s.Type()

storage/series_test.go

@@ -59,4 +59,9 @@ func TestListSeriesIterator(t *testing.T) {
 	ts, v = it.At()
 	require.Equal(t, int64(3), ts)
 	require.Equal(t, 3., v)
+
+	// Seek beyond the end.
+	require.False(t, it.Seek(5))
+	// And we don't go back. (This exposes issue #10027.)
+	require.False(t, it.Seek(2))
 }

template/template.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	html_template "html/template"
 	"math"
+	"net"
 	"net/url"
 	"regexp"
 	"sort"
@@ -184,6 +185,13 @@ func NewTemplateExpander(
 				sort.Stable(sorter)
 				return v
 			},
+			"stripPort": func(hostPort string) string {
+				host, _, err := net.SplitHostPort(hostPort)
+				if err != nil {
+					return hostPort
+				}
+				return host
+			},
 			"humanize": func(i interface{}) (string, error) {
 				v, err := convertToFloat(i)
 				if err != nil {

template/template_test.go

@@ -161,6 +161,41 @@ func TestTemplateExpansion(t *testing.T) {
 			},
 			output: "a:11: b:21: ",
 		},
+		{
+			// Simple hostname.
+			text:   "{{ \"foo.example.com\" | stripPort }}",
+			output: "foo.example.com",
+		},
+		{
+			// Hostname with port.
+			text:   "{{ \"foo.example.com:12345\" | stripPort }}",
+			output: "foo.example.com",
+		},
+		{
+			// Simple IPv4 address.
+			text:   "{{ \"192.0.2.1\" | stripPort }}",
+			output: "192.0.2.1",
+		},
+		{
+			// IPv4 address with port.
+			text:   "{{ \"192.0.2.1:12345\" | stripPort }}",
+			output: "192.0.2.1",
+		},
+		{
+			// Simple IPv6 address.
+			text:   "{{ \"2001:0DB8::1\" | stripPort }}",
+			output: "2001:0DB8::1",
+		},
+		{
+			// IPv6 address with port.
+			text:   "{{ \"[2001:0DB8::1]:12345\" | stripPort }}",
+			output: "2001:0DB8::1",
+		},
+		{
+			// Value can't be split into host and port.
+			text:   "{{ \"[2001:0DB8::1]::12345\" | stripPort }}",
+			output: "[2001:0DB8::1]::12345",
+		},
 		{
 			// Missing value is no value for nil options.
 			text:   "{{ .Foo }}",

tsdb/tsdbutil/buffer_test.go

@@ -180,6 +180,9 @@ func (it *listSeriesIterator) Seek(t int64) chunkenc.ValueType {
 	if it.idx == -1 {
 		it.idx = 0
 	}
+	if it.idx >= len(it.list) {
+		return false
+	}
 	// No-op check.
 	if s := it.list[it.idx]; s.t >= t {
 		return s.Type()

web/web_test.go

@@ -567,39 +567,68 @@ func TestAgentAPIEndPoints(t *testing.T) {
 
 	webHandler := New(nil, opts)
 	webHandler.Ready()
-
+	webHandler.config = &config.Config{}
-	baseURL := "http://localhost" + port
+	webHandler.notifier = &notifier.Manager{}
-
+	l, err := webHandler.Listener()
-	// Test for non-available endpoints in the Agent mode.
+	if err != nil {
-	for _, u := range []string{
+		panic(fmt.Sprintf("Unable to start web listener: %s", err))
-		"/-/labels",
-		"/label",
-		"/series",
-		"/alertmanagers",
-		"/query",
-		"/query_range",
-		"/query_exemplars",
-		"/graph",
-		"/rules",
-	} {
-		w := httptest.NewRecorder()
-		req, err := http.NewRequest("GET", baseURL+u, nil)
-		require.NoError(t, err)
-		webHandler.router.ServeHTTP(w, req)
-		require.Equal(t, http.StatusNotFound, w.Code)
 	}
 
-	// Test for available endpoints in the Agent mode.
+	ctx, cancel := context.WithCancel(context.Background())
-	for _, u := range []string{
+	defer cancel()
-		"/agent",
+	go func() {
-		"/targets",
+		err := webHandler.Run(ctx, l, "")
-		"/status",
+		if err != nil {
+			panic(fmt.Sprintf("Can't start web handler:%s", err))
+		}
+	}()
+
+	// Give some time for the web goroutine to run since we need the server
+	// to be up before starting tests.
+	time.Sleep(5 * time.Second)
+	baseURL := "http://localhost" + port + "/api/v1"
+
+	// Test for non-available endpoints in the Agent mode.
+	for path, methods := range map[string][]string{
+		"/labels":                      {http.MethodGet, http.MethodPost},
+		"/label/:name/values":          {http.MethodGet},
+		"/series":                      {http.MethodGet, http.MethodPost, http.MethodDelete},
+		"/alertmanagers":               {http.MethodGet},
+		"/query":                       {http.MethodGet, http.MethodPost},
+		"/query_range":                 {http.MethodGet, http.MethodPost},
+		"/query_exemplars":             {http.MethodGet, http.MethodPost},
+		"/status/tsdb":                 {http.MethodGet},
+		"/alerts":                      {http.MethodGet},
+		"/rules":                       {http.MethodGet},
+		"/admin/tsdb/delete_series":    {http.MethodPost, http.MethodPut},
+		"/admin/tsdb/clean_tombstones": {http.MethodPost, http.MethodPut},
+		"/admin/tsdb/snapshot":         {http.MethodPost, http.MethodPut},
 	} {
-		w := httptest.NewRecorder()
+		for _, m := range methods {
-		req, err := http.NewRequest("GET", baseURL+u, nil)
+			req, err := http.NewRequest(m, baseURL+path, nil)
-		require.NoError(t, err)
+			require.NoError(t, err)
-		webHandler.router.ServeHTTP(w, req)
+			resp, err := http.DefaultClient.Do(req)
-		require.Equal(t, http.StatusOK, w.Code)
+			require.NoError(t, err)
+			require.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode)
+		}
+	}
+
+	// Test for some of available endpoints in the Agent mode.
+	for path, methods := range map[string][]string{
+		"/targets":            {http.MethodGet},
+		"/targets/metadata":   {http.MethodGet},
+		"/metadata":           {http.MethodGet},
+		"/status/config":      {http.MethodGet},
+		"/status/runtimeinfo": {http.MethodGet},
+		"/status/flags":       {http.MethodGet},
+	} {
+		for _, m := range methods {
+			req, err := http.NewRequest(m, baseURL+path, nil)
+			require.NoError(t, err)
+			resp, err := http.DefaultClient.Do(req)
+			require.NoError(t, err)
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+		}
 	}
 }