Bug fixes

collector/client.go

@@ -7,6 +7,7 @@ import (
 	"crypto/x509"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"os"
@@ -183,13 +184,15 @@ func (c *MetricsClient) FetchMetricsIterator(target string) (*prompb.Iterator, e
 	}
 
 	if resp.StatusCode != http.StatusOK {
+		defer resp.Body.Close()
 		return nil, fmt.Errorf("collect node metrics for %s: %s", target, resp.Status)
 	}
 
 	br := resp.Body
 	if resp.Header.Get("Content-Encoding") == "gzip" {
-		br, err = gzip.NewReader(resp.Body)
+		br, err = NewGzipReaderWithClose(resp.Body)
 		if err != nil {
+			defer br.Close()
 			return nil, fmt.Errorf("collect node metrics for %s: %w", target, err)
 		}
 	}
@@ -267,3 +270,30 @@ func (c *MetricsClient) refreshToken() {
 		}
 	}
 }
+
+// gzipReaderWithClose wraps gzip.Reader and the underlying reader
+type gzipReaderWithClose struct {
+	*gzip.Reader
+	underlying io.ReadCloser
+}
+
+// Close closes both the gzip.Reader and the underlying reader
+func (gr *gzipReaderWithClose) Close() error {
+	err := gr.Reader.Close()
+	if err != nil {
+		return err
+	}
+	return gr.underlying.Close()
+}
+
+// NewGzipReaderWithClose creates a new gzipReaderWithClose
+func NewGzipReaderWithClose(r io.ReadCloser) (*gzipReaderWithClose, error) {
+	gr, err := gzip.NewReader(r)
+	if err != nil {
+		return nil, err
+	}
+	return &gzipReaderWithClose{
+		Reader:     gr,
+		underlying: r,
+	}, nil
+}

collector/scraper.go

@@ -207,6 +207,7 @@ func (s *Scraper) scrapeTargets(ctx context.Context) {
 	scrapeTime := time.Now().UnixNano() / 1e6
 	wr := &prompb.WriteRequest{}
 	for _, target := range targets {
+		logger.Infof("Scraping %s", target.String())
 		iter, err := s.scrapeClient.FetchMetricsIterator(target.Addr)
 		if err != nil {
 			logger.Errorf("Failed to scrape %s: %s", target.Addr, err.Error())
@@ -257,6 +258,10 @@ func (s *Scraper) scrapeTargets(ctx context.Context) {
 			wr.Timeseries = append(wr.Timeseries, ts)
 			wr = s.flushBatchIfNecessary(ctx, wr)
 		}
+		if err := iter.Err(); err != nil {
+			logger.Errorf("Failed to scrape %s: %s", target.Addr, err.Error())
+		}
+
 		if err := iter.Close(); err != nil {
 			logger.Errorf("Failed to close iterator: %s", err.Error())
 		}

ingestor/cluster/batcher.go

@@ -230,8 +230,6 @@ func (b *batcher) BatchSegments() error {
 // thresholds.  In addition, the batches are ordered as oldest first to allow for prioritizing
 // lagging segments over new ones.
 func (b *batcher) processSegments() ([]*Batch, []*Batch, error) {
-	metrics.IngestorSegmentsTotal.Reset()
-
 	// Groups is b map of metrics name to b list of segments for that metric.
 	groups := make(map[string][]wal.SegmentInfo)
 

pkg/prompb/iterator.go

@@ -24,6 +24,10 @@ func NewIterator(r io.ReadCloser) *Iterator {
 
 func (i *Iterator) Next() bool {
 	if i.scanner.Scan() {
+		if i.scanner.Err() != nil {
+			return false
+		}
+
 		i.current = i.scanner.Text()
 		if i.isComment(i.current) || i.isSpace(i.current) {
 			return i.Next()
@@ -168,6 +172,10 @@ func parseLabels(labels Labels, line string) (Labels, string, error) {
 			}
 
 			key := line[:idx]
+			if len(key) == 0 {
+				return nil, "", fmt.Errorf("invalid label: no key: %s", orig)
+			}
+
 			line = line[idx+1:]
 
 			if len(line) == 0 {

pkg/prompb/iterator_test.go

@@ -120,6 +120,7 @@ func TestIterator_TimeSeries_Malformed(t *testing.T) {
 		`http_requests_total    {key="a",b="c"}     1 a`,
 		`http_requests_total    {}     1 a`,
 		`http_requests_total    {}    `,
+		`http_requests_total    {a="b",=} 1027`,
 	} {
 		t.Run(c, func(t *testing.T) {
 			iter := NewIterator(io.NopCloser(strings.NewReader(c)))