Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1180 (2021-04-04 08:30:05)

fix access to nil *url.URL
Signed-off-by: Megrez Lu <lujiajing1126@gmail.com>

Fix access to nil `*url.URL` pointer, since there is no nil check in [`URL.String()`](https://github.com/golang/go/blob/776d8d387c262d3dd159f7868b3c04fb0f617865/src/net/url/url.go#L811-L816)


Patch:
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 4a0dd2950..f1294214e 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -9,13 +9,17 @@
 * FEATURE: vmagent: add support for `follow_redirects` option to `scrape_configs` section in the same way as [Prometheus 2.26 does](https://github.com/prometheus/prometheus/pull/8546).
 * FEATURE: vmagent: add support for `authorization` section in `-promscrape.config` in the same way as [Prometheus 2.26 does](https://github.com/prometheus/prometheus/pull/8512).
 * FEATURE: vmagent: add support for socks5 proxy in `proxy_url` config option. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1177).
+* FEATURE: vmagent: add support for `socks5 over tls` proxy in `proxy_url` config option. It can be set up with the following config: `proxy_url: tls+socks5://proxy-addr:port`.
 * FEATURE: vmagent: reduce memory usage when `-remoteWrite.queues` is set to a big value. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1167).
 * FEATURE: vmagent: add AWS IAM roles for tasks support for EC2 service discovery according to [these docs](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-iam-roles.html).
-* FEATURE: vmagent: add support for `proxy_tls_config`, `proxy_authorization`, `proxy_basic_auth`, `proxy_bearer_token` and `proxy_bearer_token_file` options to `consul_sd_config`, `dockerswarm_sd_config` and `eureka_sd_config` sections.
+* FEATURE: vmagent: add support for `proxy_tls_config`, `proxy_authorization`, `proxy_basic_auth`, `proxy_bearer_token` and `proxy_bearer_token_file` options in `consul_sd_config`, `dockerswarm_sd_config` and `eureka_sd_config` sections.
+* FEATURE: vmagent: pass `X-Prometheus-Scrape-Timeout-Seconds` header to scrape targets as Prometheus does. In this case scrape targets can limit the time needed for performing the scrape. See [this comment](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1179#issuecomment-813118733) for details.
+* FEATURE: vmagent: drop corrupted persistent queue files at `-remoteWrite.tmpDataPath` instead of throwing a fatal error. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1030).
 * FEATURE: vmauth: add support for authorization via [bearer token](https://swagger.io/docs/specification/authentication/bearer-authentication/). See [the docs](https://victoriametrics.github.io/vmauth.html#auth-config) for details.
 
 * BUGFIX: vmagent: properly work with simple HTTP proxies which don't support `CONNECT` method. For example, [PushProx](https://github.com/prometheus-community/PushProx). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1179).
 * BUGFIX: vmagent: properly discover targets if multiple namespace selectors are put inside `kubernetes_sd_config`. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1170).
+* BUGFIX: vmagent: properly discover `role: endpoints` and `role: endpointslices` targets in `kubernetes_sd_config`. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1182).
 * BUGFIX: properly generate filename for `*.tar.gz` archive inside `_checksums.txt` file posted at [releases page](https://github.com/VictoriaMetrics/VictoriaMetrics/releases). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1171).
 
 
diff --git a/lib/persistentqueue/fastqueue.go b/lib/persistentqueue/fastqueue.go
index 5c25933f6..7dda73a86 100644
--- a/lib/persistentqueue/fastqueue.go
+++ b/lib/persistentqueue/fastqueue.go
@@ -8,7 +8,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 )
 
-// FastQueue is a wrapper around Queue, which prefers sending data via memory.
+// FastQueue is fast persistent queue, which prefers sending data via memory.
 //
 // It falls back to sending data via file when readers don't catch up with writers.
 type FastQueue struct {
@@ -20,7 +20,7 @@ type FastQueue struct {
 	cond sync.Cond
 
 	// pq is file-based queue
-	pq *Queue
+	pq *queue
 
 	// ch is in-memory queue
 	ch chan *bytesutil.ByteBuffer
@@ -40,7 +40,7 @@ type FastQueue struct {
 // Otherwise its size is limited by maxPendingBytes. The oldest data is dropped when the queue
 // reaches maxPendingSize.
 func MustOpenFastQueue(path, name string, maxInmemoryBlocks, maxPendingBytes int) *FastQueue {
-	pq := MustOpen(path, name, maxPendingBytes)
+	pq := mustOpen(path, name, maxPendingBytes)
 	fq := &FastQueue{
 		pq: pq,
 		ch: make(chan *bytesutil.ByteBuffer, maxInmemoryBlocks),
@@ -174,7 +174,12 @@ func (fq *FastQueue) MustReadBlock(dst []byte) ([]byte, bool) {
 			return dst, true
 		}
 		if n := fq.pq.GetPendingBytes(); n > 0 {
-			return fq.pq.MustReadBlock(dst)
+			data, ok := fq.pq.MustReadBlockNonblocking(dst)
+			if ok {
+				return data, true
+			}
+			dst = data
+			continue
 		}
 
 		// There are no blocks. Wait for new block.
diff --git a/lib/persistentqueue/persistentqueue.go b/lib/persistentqueue/persistentqueue.go
index 916276c3f..4c082a9f8 100644
--- a/lib/persistentqueue/persistentqueue.go
+++ b/lib/persistentqueue/persistentqueue.go
@@ -8,7 +8,6 @@ import (
 	"os"
 	"regexp"
 	"strconv"
-	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
@@ -26,8 +25,10 @@ const defaultChunkFileSize = (MaxBlockSize + 8) * 16
 
 var chunkFileNameRegex = regexp.MustCompile("^[0-9A-F]{16}$")
 
-// Queue represents persistent queue.
-type Queue struct {
+// queue represents persistent queue.
+//
+// It is unsafe to call queue methods from concurrent goroutines.
+type queue struct {
 	chunkFileSize   uint64
 	maxBlockSize    uint64
 	maxPendingBytes uint64
@@ -37,13 +38,6 @@ type Queue struct {
 
 	flockF *os.File
 
-	// mu protects all the fields below.
-	mu sync.Mutex
-
-	// cond is used for notifying blocked readers when new data has been added
-	// or when MustClose is called.
-	cond sync.Cond
-
 	reader            *filestream.Reader
 	readerPath        string
 	readerOffset      uint64
@@ -74,10 +68,7 @@ type Queue struct {
 // ResetIfEmpty resets q if it is empty.
 //
 // This is needed in order to remove chunk file associated with empty q.
-func (q *Queue) ResetIfEmpty() {
-	q.mu.Lock()
-	defer q.mu.Unlock()
-
+func (q *queue) ResetIfEmpty() {
 	if q.readerOffset != q.writerOffset {
 		// The queue isn't empty.
 		return
@@ -86,10 +77,13 @@ func (q *Queue) ResetIfEmpty() {
 		// The file is too small to drop. Leave it as is in order to reduce filesystem load.
 		return
 	}
+	q.mustResetFiles()
+}
+
+func (q *queue) mustResetFiles() {
 	if q.readerPath != q.writerPath {
 		logger.Panicf("BUG: readerPath=%q doesn't match writerPath=%q", q.readerPath, q.writerPath)
 	}
-
 	q.reader.MustClose()
 	q.writer.MustClose()
 	fs.MustRemoveAll(q.readerPath)
@@ -115,31 +109,29 @@ func (q *Queue) ResetIfEmpty() {
 	}
 	q.reader = r
 
-	if err := q.flushMetainfoLocked(); err != nil {
+	if err := q.flushMetainfo(); err != nil {
 		logger.Panicf("FATAL: cannot flush metainfo: %s", err)
 	}
 }
 
 // GetPendingBytes returns the number of pending bytes in the queue.
-func (q *Queue) GetPendingBytes() uint64 {
-	q.mu.Lock()
+func (q *queue) GetPendingBytes() uint64 {
 	n := q.writerOffset - q.readerOffset
-	q.mu.Unlock()
 	return n
 }
 
-// MustOpen opens persistent queue from the given path.
+// mustOpen opens persistent queue from the given path.
 //
 // If maxPendingBytes is greater than 0, then the max queue size is limited by this value.
 // The oldest data is deleted when queue size exceeds maxPendingBytes.
-func MustOpen(path, name string, maxPendingBytes int) *Queue {
+func mustOpen(path, name string, maxPendingBytes int) *queue {
 	if maxPendingBytes < 0 {
 		maxPendingBytes = 0
 	}
-	return mustOpen(path, name, defaultChunkFileSize, MaxBlockSize, uint64(maxPendingBytes))
+	return mustOpenInternal(path, name, defaultChunkFileSize, MaxBlockSize, uint64(maxPendingBytes))
 }
 
-func mustOpen(path, name string, chunkFileSize, maxBlockSize, maxPendingBytes uint64) *Queue {
+func mustOpenInternal(path, name string, chunkFileSize, maxBlockSize, maxPendingBytes uint64) *queue {
 	if chunkFileSize < 8 || chunkFileSize-8 < maxBlockSize {
 		logger.Panicf("BUG: too small chunkFileSize=%d for maxBlockSize=%d; chunkFileSize must fit at least one block", chunkFileSize, maxBlockSize)
 	}
@@ -166,15 +158,14 @@ func mustCreateFlockFile(path string) *os.File {
 	return f
 }
 
-func tryOpeningQueue(path, name string, chunkFileSize, maxBlockSize, maxPendingBytes uint64) (*Queue, error) {
+func tryOpeningQueue(path, name string, chunkFileSize, maxBlockSize, maxPendingBytes uint64) (*queue, error) {
 	// Protect from concurrent opens.
-	var q Queue
+	var q queue
 	q.chunkFileSize = chunkFileSize
 	q.maxBlockSize = maxBlockSize
 	q.maxPendingBytes = maxPendingBytes
 	q.dir = path
 	q.name = name
-	q.cond.L = &q.mu
 
 	q.blocksDropped = metrics.GetOrCreateCounter(fmt.Sprintf(`vm_persistentqueue_blocks_dropped_total{path=%q}`, path))
 	q.bytesDropped = metrics.GetOrCreateCounter(fmt.Sprintf(`vm_persistentqueue_bytes_dropped_total{path=%q}`, path))
@@ -346,17 +337,8 @@ func tryOpeningQueue(path, name string, chunkFileSize, maxBlockSize, maxPendingB
 
 // MustClose closes q.
 //
-// It unblocks all the MustReadBlock calls.
-//
 // MustWriteBlock mustn't be called during and after the call to MustClose.
-func (q *Queue) MustClose() {
-	q.mu.Lock()
-	defer q.mu.Unlock()
-
-	// Unblock goroutines blocked on cond in MustReadBlock.
-	q.mustStop = true
-	q.cond.Broadcast()
-
+func (q *queue) MustClose() {
 	// Close writer.
 	q.writer.MustClose()
 	q.writer = nil
@@ -366,7 +348,7 @@ func (q *Queue) MustClose() {
 	q.reader = nil
 
 	// Store metainfo
-	if err := q.flushMetainfoLocked(); err != nil {
+	if err := q.flushMetainfo(); err != nil {
 		logger.Panicf("FATAL: cannot flush chunked queue metainfo: %s", err)
 	}
 
@@ -377,11 +359,11 @@ func (q *Queue) MustClose() {
 	q.flockF = nil
 }
 
-func (q *Queue) chunkFilePath(offset uint64) string {
+func (q *queue) chunkFilePath(offset uint64) string {
 	return fmt.Sprintf("%s/%016X", q.dir, offset)
 }
 
-func (q *Queue) metainfoPath() string {
+func (q *queue) metainfoPath() string {
 	return q.dir + "/metainfo.json"
 }
 
@@ -390,14 +372,10 @@ func (q *Queue) metainfoPath() string {
 // The block size cannot exceed MaxBlockSize.
 //
 // It is safe calling this function from concurrent goroutines.
-func (q *Queue) MustWriteBlock(block []byte) {
+func (q *queue) MustWriteBlock(block []byte) {
 	if uint64(len(block)) > q.maxBlockSize {
 		logger.Panicf("BUG: too big block to send: %d bytes; it mustn't exceed %d bytes", len(block), q.maxBlockSize)
 	}
-
-	q.mu.Lock()
-	defer q.mu.Unlock()
-
 	if q.mustStop {
 		logger.Panicf("BUG: MustWriteBlock cannot be called after MustClose")
 	}
@@ -416,7 +394,10 @@ func (q *Queue) MustWriteBlock(block []byte) {
 		bb := blockBufPool.Get()
 		for q.writerOffset-q.readerOffset > maxPendingBytes {
 			var err error
-			bb.B, err = q.readBlockLocked(bb.B[:0])
+			bb.B, err = q.readBlock(bb.B[:0])
+			if err == errEmptyQueue {
+				break
+			}
 			if err != nil {
 				logger.Panicf("FATAL: cannot read the oldest block %s", err)
 			}
@@ -429,38 +410,18 @@ func (q *Queue) MustWriteBlock(block []byte) {
 			return
 		}
 	}
-	if err := q.writeBlockLocked(block); err != nil {
+	if err := q.writeBlock(block); err != nil {
 		logger.Panicf("FATAL: %s", err)
 	}
-
-	// Notify blocked reader if any.
-	// See https://github.com/VictoriaMetrics/VictoriaMetrics/pull/484 for details.
-	q.cond.Signal()
 }
 
 var blockBufPool bytesutil.ByteBufferPool
 
-func (q *Queue) writeBlockLocked(block []byte) error {
+func (q *queue) writeBlock(block []byte) error {
 	if q.writerLocalOffset+q.maxBlockSize+8 > q.chunkFileSize {
-		// Finalize the current chunk and start new one.
-		q.writer.MustClose()
-		// There is no need to do fs.MustSyncPath(q.writerPath) here,
-		// since MustClose already does this.
-		if n := q.writerOffset % q.chunkFileSize; n > 0 {
-			q.writerOffset += (q.chunkFileSize - n)
+		if err := q.nextChunkFileForWrite(); err != nil {
+			return fmt.Errorf("cannot create next chunk file: %w", err)
 		}
-		q.writerFlushedOffset = q.writerOffset
-		q.writerLocalOffset = 0
-		q.writerPath = q.chunkFilePath(q.writerOffset)
-		w, err := filestream.Create(q.writerPath, false)
-		if err != nil {
-			return fmt.Errorf("cannot create chunk file %q: %w", q.writerPath, err)
-		}
-		q.writer = w
-		if err := q.flushMetainfoLocked(); err != nil {
-			return fmt.Errorf("cannot flush metainfo: %w", err)
-		}
-		fs.MustSyncPath(q.dir)
 	}
 
 	// Write block len.
@@ -479,62 +440,61 @@ func (q *Queue) writeBlockLocked(block []byte) error {
 	}
 	q.blocksWritten.Inc()
 	q.bytesWritten.Add(len(block))
-	return q.flushMetainfoIfNeededLocked(true)
+	return q.flushWriterMetainfoIfNeeded()
 }
 
-// MustReadBlock appends the next block from q to dst and returns the result.
-//
-// false is returned after MustClose call.
-//
-// It is safe calling this function from concurrent goroutines.
-func (q *Queue) MustReadBlock(dst []byte) ([]byte, bool) {
-	q.mu.Lock()
-	defer q.mu.Unlock()
-
-	for {
-		if q.mustStop {
-			return dst, false
-		}
-		if q.readerOffset > q.writerOffset {
-			logger.Panicf("BUG: readerOffset=%d cannot exceed writerOffset=%d", q.readerOffset, q.writerOffset)
-		}
-		if q.readerOffset < q.writerOffset {
-			break
-		}
-		q.cond.Wait()
+func (q *queue) nextChunkFileForWrite() error {
+	// Finalize the current chunk and start new one.
+	q.writer.MustClose()
+	// There is no need to do fs.MustSyncPath(q.writerPath) here,
+	// since MustClose already does this.
+	if n := q.writerOffset % q.chunkFileSize; n > 0 {
+		q.writerOffset += q.chunkFileSize - n
 	}
+	q.writerFlushedOffset = q.writerOffset
+	q.writerLocalOffset = 0
+	q.writerPath = q.chunkFilePath(q.writerOffset)
+	w, err := filestream.Create(q.writerPath, false)
+	if err != nil {
+		return fmt.Errorf("cannot create chunk file %q: %w", q.writerPath, err)
+	}
+	q.writer = w
+	if err := q.flushMetainfo(); err != nil {
+		return fmt.Errorf("cannot flush metainfo: %w", err)
+	}
+	fs.MustSyncPath(q.dir)
+	return nil
+}
 
-	data, err := q.readBlockLocked(dst)
+// MustReadBlockNonblocking appends the next block from q to dst and returns the result.
+//
+// false is returned if q is empty.
+func (q *queue) MustReadBlockNonblocking(dst []byte) ([]byte, bool) {
+	if q.readerOffset > q.writerOffset {
+		logger.Panicf("BUG: readerOffset=%d cannot exceed writerOffset=%d", q.readerOffset, q.writerOffset)
+	}
+	if q.readerOffset == q.writerOffset {
+		return dst, false
+	}
+	var err error
+	dst, err = q.readBlock(dst)
 	if err != nil {
-		// Skip the current chunk, since it may be broken.
-		q.readerOffset += q.chunkFileSize - q.readerOffset%q.chunkFileSize
-		_ = q.flushMetainfoLocked()
+		if err == errEmptyQueue {
+			return dst, false
+		}
 		logger.Panicf("FATAL: %s", err)
 	}
-	return data, true
+	return dst, true
 }
 
-func (q *Queue) readBlockLocked(dst []byte) ([]byte, error) {
+func (q *queue) readBlock(dst []byte) ([]byte, error) {
 	if q.readerLocalOffset+q.maxBlockSize+8 > q.chunkFileSize {
-		// Remove the current chunk and go to the next chunk.
-		q.reader.MustClose()
-		fs.MustRemoveAll(q.readerPath)
-		if n := q.readerOffset % q.chunkFileSize; n > 0 {
-			q.readerOffset += (q.chunkFileSize - n)
+		if err := q.nextChunkFileForRead(); err != nil {
+			return dst, fmt.Errorf("cannot open next chunk file: %w", err)
 		}
-		q.readerLocalOffset = 0
-		q.readerPath = q.chunkFilePath(q.readerOffset)
-		r, err := filestream.Open(q.readerPath, true)
-		if err != nil {
-			return dst, fmt.Errorf("cannot open chunk file %q: %w", q.readerPath, err)
-		}
-		q.reader = r
-		if err := q.flushMetainfoLocked(); err != nil {
-			return dst, fmt.Errorf("cannot flush metainfo: %w", err)
-		}
-		fs.MustSyncPath(q.dir)
 	}
 
+again:
 	// Read block len.
 	header := headerBufPool.Get()
 	header.B = bytesutil.Resize(header.B, 8)
@@ -542,27 +502,73 @@ func (q *Queue) readBlockLocked(dst []byte) ([]byte, error) {
 	blockLen := encoding.UnmarshalUint64(header.B)
 	headerBufPool.Put(header)
 	if err != nil {
-		return dst, fmt.Errorf("cannot read header with size 8 bytes from %q: %w", q.readerPath, err)
+		logger.Errorf("skipping corrupted %q, since header with size 8 bytes cannot be read from it: %s", q.readerPath, err)
+		if err := q.skipBrokenChunkFile(); err != nil {
+			return dst, err
+		}
+		goto again
 	}
 	if blockLen > q.maxBlockSize {
-		return dst, fmt.Errorf("too big block size read from %q: %d bytes; cannot exceed %d bytes", q.readerPath, blockLen, q.maxBlockSize)
+		logger.Errorf("skipping corrupted %q, since too big block size is read from it: %d bytes; cannot exceed %d bytes", q.readerPath, blockLen, q.maxBlockSize)
+		if err := q.skipBrokenChunkFile(); err != nil {
+			return dst, err
+		}
+		goto again
 	}
 
 	// Read block contents.
 	dstLen := len(dst)
 	dst = bytesutil.Resize(dst, dstLen+int(blockLen))
 	if err := q.readFull(dst[dstLen:]); err != nil {
-		return dst, fmt.Errorf("cannot read block contents with size %d bytes from %q: %w", blockLen, q.readerPath, err)
+		logger.Errorf("skipping corrupted %q, since contents with size %d bytes cannot be read from it: %s", q.readerPath, blockLen, err)
+		if err := q.skipBrokenChunkFile(); err != nil {
+			return dst[:dstLen], err
+		}
+		goto again
 	}
 	q.blocksRead.Inc()
 	q.bytesRead.Add(int(blockLen))
-	if err := q.flushMetainfoIfNeededLocked(false); err != nil {
+	if err := q.flushReaderMetainfoIfNeeded(); err != nil {
 		return dst, err
 	}
 	return dst, nil
 }
 
-func (q *Queue) write(buf []byte) error {
+func (q *queue) skipBrokenChunkFile() error {
+	// Try to recover from broken chunk file by skipping it.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1030
+	q.readerOffset += q.chunkFileSize - q.readerOffset%q.chunkFileSize
+	if q.readerOffset >= q.writerOffset {
+		q.mustResetFiles()
+		return errEmptyQueue
+	}
+	return q.nextChunkFileForRead()
+}
+
+var errEmptyQueue = fmt.Errorf("the queue is empty")
+
+func (q *queue) nextChunkFileForRead() error {
+	// Remove the current chunk and go to the next chunk.
+	q.reader.MustClose()
+	fs.MustRemoveAll(q.readerPath)
+	if n := q.readerOffset % q.chunkFileSize; n > 0 {
+		q.readerOffset += q.chunkFileSize - n
+	}
+	q.readerLocalOffset = 0
+	q.readerPath = q.chunkFilePath(q.readerOffset)
+	r, err := filestream.Open(q.readerPath, true)
+	if err != nil {
+		return fmt.Errorf("cannot open chunk file %q: %w", q.readerPath, err)
+	}
+	q.reader = r
+	if err := q.flushMetainfo(); err != nil {
+		return fmt.Errorf("cannot flush metainfo: %w", err)
+	}
+	fs.MustSyncPath(q.dir)
+	return nil
+}
+
+func (q *queue) write(buf []byte) error {
 	bufLen := uint64(len(buf))
 	n, err := q.writer.Write(buf)
 	if err != nil {
@@ -576,7 +582,7 @@ func (q *Queue) write(buf []byte) error {
 	return nil
 }
 
-func (q *Queue) readFull(buf []byte) error {
+func (q *queue) readFull(buf []byte) error {
 	bufLen := uint64(len(buf))
 	if q.readerOffset+bufLen > q.writerFlushedOffset {
 		q.writer.MustFlush(false)
@@ -594,22 +600,32 @@ func (q *Queue) readFull(buf []byte) error {
 	return nil
 }
 
-func (q *Queue) flushMetainfoIfNeededLocked(flushData bool) error {
+func (q *queue) flushReaderMetainfoIfNeeded() error {
 	t := fasttime.UnixTimestamp()
 	if t == q.lastMetainfoFlushTime {
 		return nil
 	}
-	if flushData {
-		q.writer.MustFlush(true)
+	if err := q.flushMetainfo(); err != nil {
+		return fmt.Errorf("cannot flush metainfo: %w", err)
+	}
+	q.lastMetainfoFlushTime = t
+	return nil
+}
+
+func (q *queue) flushWriterMetainfoIfNeeded() error {
+	t := fasttime.UnixTimestamp()
+	if t == q.lastMetainfoFlushTime {
+		return nil
 	}
-	if err := q.flushMetainfoLocked(); err != nil {
+	q.writer.MustFlush(true)
+	if err := q.flushMetainfo(); err != nil {
 		return fmt.Errorf("cannot flush metainfo: %w", err)
 	}
 	q.lastMetainfoFlushTime = t
 	return nil
 }
 
-func (q *Queue) flushMetainfoLocked() error {
+func (q *queue) flushMetainfo() error {
 	mi := &metainfo{
 		Name:         q.name,
 		ReaderOffset: q.readerOffset,
diff --git a/lib/promscrape/client.go b/lib/promscrape/client.go
index 565e6915c..e023a95d3 100644
--- a/lib/promscrape/client.go
+++ b/lib/promscrape/client.go
@@ -43,14 +43,15 @@ type client struct {
 	// It may be useful for scraping targets with millions of metrics per target.
 	sc *http.Client
 
-	scrapeURL          string
-	host               string
-	requestURI         string
-	authHeader         string
-	proxyAuthHeader    string
-	denyRedirects      bool
-	disableCompression bool
-	disableKeepAlive   bool
+	scrapeURL               string
+	scrapeTimeoutSecondsStr string
+	host                    string
+	requestURI              string
+	authHeader              string
+	proxyAuthHeader         string
+	denyRedirects           bool
+	disableCompression      bool
+	disableKeepAlive        bool
 }
 
 func newClient(sw *ScrapeWork) *client {
@@ -127,16 +128,17 @@ func newClient(sw *ScrapeWork) *client {
 		}
 	}
 	return &client{
-		hc:                 hc,
-		sc:                 sc,
-		scrapeURL:          sw.ScrapeURL,
-		host:               host,
-		requestURI:         requestURI,
-		authHeader:         sw.AuthConfig.Authorization,
-		proxyAuthHeader:    proxyAuthHeader,
-		denyRedirects:      sw.DenyRedirects,
-		disableCompression: sw.DisableCompression,
-		disableKeepAlive:   sw.DisableKeepAlive,
+		hc:                      hc,
+		sc:                      sc,
+		scrapeURL:               sw.ScrapeURL,
+		scrapeTimeoutSecondsStr: fmt.Sprintf("%.3f", sw.ScrapeTimeout.Seconds()),
+		host:                    host,
+		requestURI:              requestURI,
+		authHeader:              sw.AuthConfig.Authorization,
+		proxyAuthHeader:         proxyAuthHeader,
+		denyRedirects:           sw.DenyRedirects,
+		disableCompression:      sw.DisableCompression,
+		disableKeepAlive:        sw.DisableKeepAlive,
 	}
 }
 
@@ -154,6 +156,9 @@ func (c *client) GetStreamReader() (*streamReader, error) {
 	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/608 for details.
 	// Do not bloat the `Accept` header with OpenMetrics shit, since it looks like dead standard now.
 	req.Header.Set("Accept", "text/plain;version=0.0.4;q=1,*/*;q=0.1")
+	// Set X-Prometheus-Scrape-Timeout-Seconds like Prometheus does, since it is used by some exporters such as PushProx.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1179#issuecomment-813117162
+	req.Header.Set("X-Prometheus-Scrape-Timeout-Seconds", c.scrapeTimeoutSecondsStr)
 	if c.authHeader != "" {
 		req.Header.Set("Authorization", c.authHeader)
 	}
@@ -191,6 +196,9 @@ func (c *client) ReadData(dst []byte) ([]byte, error) {
 	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/608 for details.
 	// Do not bloat the `Accept` header with OpenMetrics shit, since it looks like dead standard now.
 	req.Header.Set("Accept", "text/plain;version=0.0.4;q=1,*/*;q=0.1")
+	// Set X-Prometheus-Scrape-Timeout-Seconds like Prometheus does, since it is used by some exporters such as PushProx.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1179#issuecomment-813117162
+	req.Header.Set("X-Prometheus-Scrape-Timeout-Seconds", c.scrapeTimeoutSecondsStr)
 	if c.authHeader != "" {
 		req.Header.Set("Authorization", c.authHeader)
 	}
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index b4a8d5d2c..1f93c4be5 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -60,6 +60,15 @@ type Config struct {
 	baseDir string
 }
 
+func (cfg *Config) mustStart() {
+	startTime := time.Now()
+	logger.Infof("starting service discovery routines...")
+	for i := range cfg.ScrapeConfigs {
+		cfg.ScrapeConfigs[i].mustStart(cfg.baseDir)
+	}
+	logger.Infof("started service discovery routines in %.3f seconds", time.Since(startTime).Seconds())
+}
+
 func (cfg *Config) mustStop() {
 	startTime := time.Now()
 	logger.Infof("stopping service discovery routines...")
@@ -120,6 +129,21 @@ type ScrapeConfig struct {
 	swc *scrapeWorkConfig
 }
 
+func (sc *ScrapeConfig) mustStart(baseDir string) {
+	for i := range sc.KubernetesSDConfigs {
+		swosFunc := func(metaLabels map[string]string) interface{} {
+			target := metaLabels["__address__"]
+			sw, err := sc.swc.getScrapeWork(target, nil, metaLabels)
+			if err != nil {
+				logger.Errorf("cannot create kubernetes_sd_config target %q for job_name %q: %s", target, sc.swc.jobName, err)
+				return nil
+			}
+			return sw
+		}
+		sc.KubernetesSDConfigs[i].MustStart(baseDir, swosFunc)
+	}
+}
+
 func (sc *ScrapeConfig) mustStop() {
 	for i := range sc.KubernetesSDConfigs {
 		sc.KubernetesSDConfigs[i].MustStop()
@@ -243,15 +267,7 @@ func (cfg *Config) getKubernetesSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
 		ok := true
 		for j := range sc.KubernetesSDConfigs {
 			sdc := &sc.KubernetesSDConfigs[j]
-			swos, err := sdc.GetScrapeWorkObjects(cfg.baseDir, func(metaLabels map[string]string) interface{} {
-				target := metaLabels["__address__"]
-				sw, err := sc.swc.getScrapeWork(target, nil, metaLabels)
-				if err != nil {
-					logger.Errorf("cannot create kubernetes_sd_config target %q for job_name %q: %s", target, sc.swc.jobName, err)
-					return nil
-				}
-				return sw
-			})
+			swos, err := sdc.GetScrapeWorkObjects()
 			if err != nil {
 				logger.Errorf("skipping kubernetes_sd_config targets for job_name %q because of error: %s", sc.swc.jobName, err)
 				ok = false
diff --git a/lib/promscrape/discovery/kubernetes/api.go b/lib/promscrape/discovery/kubernetes/api.go
index a09221087..8afa6b613 100644
--- a/lib/promscrape/discovery/kubernetes/api.go
+++ b/lib/promscrape/discovery/kubernetes/api.go
@@ -7,7 +7,6 @@ import (
 	"strings"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
 // apiConfig contains config for API server
@@ -15,18 +14,12 @@ type apiConfig struct {
 	aw *apiWatcher
 }
 
-func (ac *apiConfig) mustStop() {
-	ac.aw.mustStop()
+func (ac *apiConfig) mustStart() {
+	ac.aw.mustStart()
 }
 
-var configMap = discoveryutils.NewConfigMap()
-
-func getAPIConfig(sdc *SDConfig, baseDir string, swcFunc ScrapeWorkConstructorFunc) (*apiConfig, error) {
-	v, err := configMap.Get(sdc, func() (interface{}, error) { return newAPIConfig(sdc, baseDir, swcFunc) })
-	if err != nil {
-		return nil, err
-	}
-	return v.(*apiConfig), nil
+func (ac *apiConfig) mustStop() {
+	ac.aw.mustStop()
 }
 
 func newAPIConfig(sdc *SDConfig, baseDir string, swcFunc ScrapeWorkConstructorFunc) (*apiConfig, error) {
diff --git a/lib/promscrape/discovery/kubernetes/api_watcher.go b/lib/promscrape/discovery/kubernetes/api_watcher.go
index e9f096e72..d5def2f29 100644
--- a/lib/promscrape/discovery/kubernetes/api_watcher.go
+++ b/lib/promscrape/discovery/kubernetes/api_watcher.go
@@ -72,6 +72,10 @@ func newAPIWatcher(apiServer string, ac *promauth.Config, sdc *SDConfig, swcFunc
 	}
 }
 
+func (aw *apiWatcher) mustStart() {
+	aw.gw.startWatchersForRole(aw.role, aw)
+}
+
 func (aw *apiWatcher) mustStop() {
 	aw.gw.unsubscribeAPIWatcher(aw)
 	aw.swosByNamespaceLock.Lock()
@@ -87,7 +91,7 @@ func (aw *apiWatcher) reloadScrapeWorks(namespace string, swosByKey map[string][
 }
 
 func (aw *apiWatcher) setScrapeWorks(namespace, key string, labels []map[string]string) {
-	swos := getScrapeWorkObjectsForLabels(aw.swcFunc, labels)
+	swos := aw.getScrapeWorkObjectsForLabels(labels)
 	aw.swosByNamespaceLock.Lock()
 	swosByKey := aw.swosByNamespace[namespace]
 	if swosByKey == nil {
@@ -114,10 +118,10 @@ func (aw *apiWatcher) removeScrapeWorks(namespace, key string) {
 	aw.swosByNamespaceLock.Unlock()
 }
 
-func getScrapeWorkObjectsForLabels(swcFunc ScrapeWorkConstructorFunc, labelss []map[string]string) []interface{} {
+func (aw *apiWatcher) getScrapeWorkObjectsForLabels(labelss []map[string]string) []interface{} {
 	swos := make([]interface{}, 0, len(labelss))
 	for _, labels := range labelss {
-		swo := swcFunc(labels)
+		swo := aw.swcFunc(labels)
 		// The reflect check is needed because of https://mangatmodi.medium.com/go-check-nil-interface-the-right-way-d142776edef1
 		if swo != nil && !reflect.ValueOf(swo).IsNil() {
 			swos = append(swos, swo)
@@ -128,7 +132,8 @@ func getScrapeWorkObjectsForLabels(swcFunc ScrapeWorkConstructorFunc, labelss []
 
 // getScrapeWorkObjects returns all the ScrapeWork objects for the given aw.
 func (aw *apiWatcher) getScrapeWorkObjects() []interface{} {
-	aw.gw.startWatchersForRole(aw.role, aw)
+	aw.gw.registerPendingAPIWatchers()
+
 	aw.swosByNamespaceLock.Lock()
 	defer aw.swosByNamespaceLock.Unlock()
 
@@ -185,8 +190,12 @@ func newGroupWatcher(apiServer string, ac *promauth.Config, namespaces []string,
 }
 
 func getGroupWatcher(apiServer string, ac *promauth.Config, namespaces []string, selectors []Selector, proxyURL *url.URL) *groupWatcher {
-	key := fmt.Sprintf("apiServer=%s, namespaces=%s, selectors=%s, proxyURL=%v, authConfig=%s",
-		apiServer, namespaces, selectorsKey(selectors), proxyURL, ac.String())
+	proxyURLStr := "<nil>"
+	if proxyURL != nil {
+		proxyURLStr = proxyURL.String()
+	}
+	key := fmt.Sprintf("apiServer=%s, namespaces=%s, selectors=%s, proxyURL=%s, authConfig=%s",
+		apiServer, namespaces, selectorsKey(selectors), proxyURLStr, ac.String())
 	groupWatchersLock.Lock()
 	gw := groupWatchers[key]
 	if gw == nil {
@@ -223,12 +232,57 @@ func (gw *groupWatcher) getObjectByRole(role, namespace, name string) object {
 		// this is needed for testing
 		return nil
 	}
+	o := gw.getCachedObjectByRole(role, namespace, name)
+	if o != nil {
+		// Fast path: the object has been found in the cache.
+		return o
+	}
+
+	// The object wasn't found in the cache. Try querying it directly from API server.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1182#issuecomment-813353359 for details.
+	metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_direct_object_loads_total{role=%q}`, role)).Inc()
+	objectType := getObjectTypeByRole(role)
+	path := getAPIPath(objectType, namespace, "")
+	path += "/" + name
+	requestURL := gw.apiServer + path
+	resp, err := gw.doRequest(requestURL)
+	if err != nil {
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_direct_object_load_errors_total{role=%q}`, role)).Inc()
+		logger.Errorf("cannot obtain data for object %s (namespace=%q, name=%q): %s", role, namespace, name, err)
+		return nil
+	}
+	data, err := ioutil.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if err != nil {
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_direct_object_load_errors_total{role=%q}`, role)).Inc()
+		logger.Errorf("cannot read response from %q: %s", requestURL, err)
+		return nil
+	}
+	if resp.StatusCode != http.StatusOK {
+		if resp.StatusCode == http.StatusNotFound {
+			metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_direct_object_load_misses_total{role=%q}`, role)).Inc()
+			return nil
+		}
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_direct_object_load_errors_total{role=%q}`, role)).Inc()
+		logger.Errorf("unexpected status code when reading response from %q; got %d; want %d; response body: %q", requestURL, resp.StatusCode, http.StatusOK, data)
+		return nil
+	}
+	parseObject, _ := getObjectParsersForRole(role)
+	o, err = parseObject(data)
+	if err != nil {
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_direct_object_load_errors_total{role=%q}`, role)).Inc()
+		logger.Errorf("cannot parse object obtained from %q: %s; response body: %q", requestURL, err, data)
+		return nil
+	}
+	// There is no need in storing the object in urlWatcher cache, since it should be eventually populated there by urlWatcher itself.
+	return o
+}
+
+func (gw *groupWatcher) getCachedObjectByRole(role, namespace, name string) object {
 	key := namespace + "/" + name
 	gw.startWatchersForRole(role, nil)
-	gw.mu.Lock()
-	defer gw.mu.Unlock()
-
-	for _, uw := range gw.m {
+	uws := gw.getURLWatchers()
+	for _, uw := range uws {
 		if uw.role != role {
 			// Role mismatch
 			continue
@@ -253,34 +307,19 @@ func (gw *groupWatcher) startWatchersForRole(role string, aw *apiWatcher) {
 		apiURL := gw.apiServer + path
 		gw.mu.Lock()
 		uw := gw.m[apiURL]
-		if uw == nil {
+		needStart := uw == nil
+		if needStart {
 			uw = newURLWatcher(role, namespaces[i], apiURL, gw)
 			gw.m[apiURL] = uw
 		}
 		gw.mu.Unlock()
-		uw.subscribeAPIWatcher(aw)
-	}
-}
-
-func (gw *groupWatcher) reloadScrapeWorksForAPIWatchers(namespace string, aws []*apiWatcher, objectsByKey map[string]object) {
-	if len(aws) == 0 {
-		return
-	}
-	swosByKey := make([]map[string][]interface{}, len(aws))
-	for i := range aws {
-		swosByKey[i] = make(map[string][]interface{})
-	}
-	for key, o := range objectsByKey {
-		labels := o.getTargetLabels(gw)
-		for i, aw := range aws {
-			swos := getScrapeWorkObjectsForLabels(aw.swcFunc, labels)
-			if len(swos) > 0 {
-				swosByKey[i][key] = swos
-			}
+		if needStart {
+			uw.reloadObjects()
+			go uw.watchForUpdates()
+		}
+		if aw != nil {
+			uw.subscribeAPIWatcher(aw)
 		}
-	}
-	for i, aw := range aws {
-		aw.reloadScrapeWorks(namespace, swosByKey[i])
 	}
 }
 
@@ -296,12 +335,28 @@ func (gw *groupWatcher) doRequest(requestURL string) (*http.Response, error) {
 	return gw.client.Do(req)
 }
 
-func (gw *groupWatcher) unsubscribeAPIWatcher(aw *apiWatcher) {
+func (gw *groupWatcher) registerPendingAPIWatchers() {
+	uws := gw.getURLWatchers()
+	for _, uw := range uws {
+		uw.registerPendingAPIWatchers()
+	}
+}
+
+func (gw *groupWatcher) getURLWatchers() []*urlWatcher {
 	gw.mu.Lock()
+	uws := make([]*urlWatcher, 0, len(gw.m))
 	for _, uw := range gw.m {
-		uw.unsubscribeAPIWatcher(aw)
+		uws = append(uws, uw)
 	}
 	gw.mu.Unlock()
+	return uws
+}
+
+func (gw *groupWatcher) unsubscribeAPIWatcher(aw *apiWatcher) {
+	uws := gw.getURLWatchers()
+	for _, uw := range uws {
+		uw.unsubscribeAPIWatcher(aw)
+	}
 }
 
 // urlWatcher watches for an apiURL and updates object states in objectsByKey.
@@ -317,12 +372,16 @@ type urlWatcher struct {
 	// mu protects aws, awsPending, objectsByKey and resourceVersion
 	mu sync.Mutex
 
+	// awsPending contains pending apiWatcher objects, which are registered in a batch.
+	// Batch registering saves CPU time needed for registering big number of Kubernetes objects
+	// shared among big number of scrape jobs, since per-object labels are generated only once
+	// for all the scrape jobs (each scrape job is associated with a single apiWatcher).
+	// See reloadScrapeWorksForAPIWatchers for details.
+	awsPending map[*apiWatcher]struct{}
+
 	// aws contains registered apiWatcher objects
 	aws map[*apiWatcher]struct{}
 
-	// awsPending contains pending apiWatcher objects, which must be moved to aws in a batch
-	awsPending map[*apiWatcher]struct{}
-
 	// objectsByKey contains the latest state for objects obtained from apiURL
 	objectsByKey map[string]object
 
@@ -347,8 +406,8 @@ func newURLWatcher(role, namespace, apiURL string, gw *groupWatcher) *urlWatcher
 		parseObject:     parseObject,
 		parseObjectList: parseObjectList,
 
-		aws:          make(map[*apiWatcher]struct{}),
 		awsPending:   make(map[*apiWatcher]struct{}),
+		aws:          make(map[*apiWatcher]struct{}),
 		objectsByKey: make(map[string]object),
 
 		objectsCount:          metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_objects{role=%q}`, role)),
@@ -358,64 +417,44 @@ func newURLWatcher(role, namespace, apiURL string, gw *groupWatcher) *urlWatcher
 		staleResourceVersions: metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_stale_resource_versions_total{role=%q}`, role)),
 	}
 	logger.Infof("started %s watcher for %q", uw.role, uw.apiURL)
-	go uw.watchForUpdates()
-	go uw.processPendingSubscribers()
 	return uw
 }
 
 func (uw *urlWatcher) subscribeAPIWatcher(aw *apiWatcher) {
-	if aw == nil {
-		return
-	}
 	uw.mu.Lock()
 	if _, ok := uw.aws[aw]; !ok {
 		if _, ok := uw.awsPending[aw]; !ok {
 			uw.awsPending[aw] = struct{}{}
-			metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscibers{role=%q,type="pending"}`, uw.role)).Inc()
+			metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscribers{role=%q,status="pending"}`, uw.role)).Inc()
 		}
 	}
 	uw.mu.Unlock()
 }
 
-func (uw *urlWatcher) unsubscribeAPIWatcher(aw *apiWatcher) {
+func (uw *urlWatcher) registerPendingAPIWatchers() {
 	uw.mu.Lock()
-	if _, ok := uw.aws[aw]; ok {
-		delete(uw.aws, aw)
-		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscibers{role=%q,type="permanent"}`, uw.role)).Dec()
-	} else if _, ok := uw.awsPending[aw]; ok {
+	awsPending := make([]*apiWatcher, 0, len(uw.awsPending))
+	for aw := range uw.awsPending {
+		awsPending = append(awsPending, aw)
 		delete(uw.awsPending, aw)
-		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscibers{role=%q,type="pending"}`, uw.role)).Dec()
 	}
+	uw.reloadScrapeWorksForAPIWatchers(awsPending, uw.objectsByKey)
 	uw.mu.Unlock()
+	metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscribers{role=%q,status="working"}`, uw.role)).Add(len(awsPending))
+	metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscribers{role=%q,status="pending"}`, uw.role)).Add(-len(awsPending))
 }
 
-func (uw *urlWatcher) processPendingSubscribers() {
-	t := time.NewTicker(time.Second)
-	for range t.C {
-		var awsPending []*apiWatcher
-		var objectsByKey map[string]object
-
-		uw.mu.Lock()
-		if len(uw.awsPending) > 0 {
-			awsPending = getAPIWatchers(uw.awsPending)
-			for _, aw := range awsPending {
-				if _, ok := uw.aws[aw]; ok {
-					logger.Panicf("BUG: aw=%p already exists in uw.aws", aw)
-				}
-				uw.aws[aw] = struct{}{}
-				delete(uw.awsPending, aw)
-			}
-			objectsByKey = make(map[string]object, len(uw.objectsByKey))
-			for key, o := range uw.objectsByKey {
-				objectsByKey[key] = o
-			}
-		}
-		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscibers{role=%q,type="pending"}`, uw.role)).Add(-len(awsPending))
-		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscibers{role=%q,type="permanent"}`, uw.role)).Add(len(awsPending))
-		uw.mu.Unlock()
-
-		uw.gw.reloadScrapeWorksForAPIWatchers(uw.namespace, awsPending, objectsByKey)
+func (uw *urlWatcher) unsubscribeAPIWatcher(aw *apiWatcher) {
+	uw.mu.Lock()
+	if _, ok := uw.awsPending[aw]; ok {
+		delete(uw.awsPending, aw)
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscribers{role=%q,status="pending"}`, uw.role)).Dec()
+	}
+	if _, ok := uw.aws[aw]; ok {
+		delete(uw.aws, aw)
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_discovery_kubernetes_subscribers{role=%q,status="working"}`, uw.role)).Dec()
 	}
+	uw.mu.Unlock()
 }
 
 func (uw *urlWatcher) setResourceVersion(resourceVersion string) {
@@ -478,9 +517,31 @@ func (uw *urlWatcher) reloadObjects() string {
 	aws := getAPIWatchers(uw.aws)
 	uw.mu.Unlock()
 
-	uw.gw.reloadScrapeWorksForAPIWatchers(uw.namespace, aws, objectsByKey)
+	uw.reloadScrapeWorksForAPIWatchers(aws, objectsByKey)
 	logger.Infof("reloaded %d objects from %q", len(objectsByKey), requestURL)
-	return metadata.ResourceVersion
+	return uw.resourceVersion
+}
+
+func (uw *urlWatcher) reloadScrapeWorksForAPIWatchers(aws []*apiWatcher, objectsByKey map[string]object) {
+	if len(aws) == 0 {
+		return
+	}
+	swosByKey := make([]map[string][]interface{}, len(aws))
+	for i := range aws {
+		swosByKey[i] = make(map[string][]interface{})
+	}
+	for key, o := range objectsByKey {
+		labels := o.getTargetLabels(uw.gw)
+		for i, aw := range aws {
+			swos := aw.getScrapeWorkObjectsForLabels(labels)
+			if len(swos) > 0 {
+				swosByKey[i][key] = swos
+			}
+		}
+	}
+	for i, aw := range aws {
+		aw.reloadScrapeWorks(uw.namespace, swosByKey[i])
+	}
 }
 
 func getAPIWatchers(awsMap map[*apiWatcher]struct{}) []*apiWatcher {
@@ -652,32 +713,32 @@ func parseError(data []byte) (*Error, error) {
 }
 
 func getAPIPathsWithNamespaces(role string, namespaces []string, selectors []Selector) ([]string, []string) {
-	objectName := getObjectNameByRole(role)
-	if objectName == "nodes" || len(namespaces) == 0 {
+	objectType := getObjectTypeByRole(role)
+	if objectType == "nodes" || len(namespaces) == 0 {
 		query := joinSelectors(role, selectors)
-		path := getAPIPath(objectName, "", query)
+		path := getAPIPath(objectType, "", query)
 		return []string{path}, []string{""}
 	}
 	query := joinSelectors(role, selectors)
 	paths := make([]string, len(namespaces))
 	for i, namespace := range namespaces {
-		paths[i] = getAPIPath(objectName, namespace, query)
+		paths[i] = getAPIPath(objectType, namespace, query)
 	}
 	return paths, namespaces
 }
 
-func getAPIPath(objectName, namespace, query string) string {
-	suffix := objectName
+func getAPIPath(objectType, namespace, query string) string {
+	suffix := objectType
 	if namespace != "" {
-		suffix = "namespaces/" + namespace + "/" + objectName
+		suffix = "namespaces/" + namespace + "/" + objectType
 	}
 	if len(query) > 0 {
 		suffix += "?" + query
 	}
-	if objectName == "ingresses" {
+	if objectType == "ingresses" {
 		return "/apis/networking.k8s.io/v1beta1/" + suffix
 	}
-	if objectName == "endpointslices" {
+	if objectType == "endpointslices" {
 		return "/apis/discovery.k8s.io/v1beta1/" + suffix
 	}
 	return "/api/v1/" + suffix
@@ -706,7 +767,7 @@ func joinSelectors(role string, selectors []Selector) string {
 	return strings.Join(args, "&")
 }
 
-func getObjectNameByRole(role string) string {
+func getObjectTypeByRole(role string) string {
 	switch role {
 	case "node":
 		return "nodes"
diff --git a/lib/promscrape/discovery/kubernetes/endpoints.go b/lib/promscrape/discovery/kubernetes/endpoints.go
index 805a88b01..af4b16b5e 100644
--- a/lib/promscrape/discovery/kubernetes/endpoints.go
+++ b/lib/promscrape/discovery/kubernetes/endpoints.go
@@ -139,8 +139,10 @@ func appendEndpointLabelsForAddresses(ms []map[string]string, gw *groupWatcher,
 	eas []EndpointAddress, epp EndpointPort, svc *Service, ready string) []map[string]string {
 	for _, ea := range eas {
 		var p *Pod
-		if o := gw.getObjectByRole("pod", ea.TargetRef.Namespace, ea.TargetRef.Name); o != nil {
-			p = o.(*Pod)
+		if ea.TargetRef.Name != "" {
+			if o := gw.getObjectByRole("pod", ea.TargetRef.Namespace, ea.TargetRef.Name); o != nil {
+				p = o.(*Pod)
+			}
 		}
 		m := getEndpointLabelsForAddressAndPort(podPortsSeen, eps, ea, epp, p, svc, ready)
 		ms = append(ms, m)
diff --git a/lib/promscrape/discovery/kubernetes/kubernetes.go b/lib/promscrape/discovery/kubernetes/kubernetes.go
index 9315730f2..28160743b 100644
--- a/lib/promscrape/discovery/kubernetes/kubernetes.go
+++ b/lib/promscrape/discovery/kubernetes/kubernetes.go
@@ -17,6 +17,9 @@ type SDConfig struct {
 	ProxyURL         proxy.URL                 `yaml:"proxy_url,omitempty"`
 	Namespaces       Namespaces                `yaml:"namespaces,omitempty"`
 	Selectors        []Selector                `yaml:"selectors,omitempty"`
+
+	cfg      *apiConfig
+	startErr error
 }
 
 // Namespaces represents namespaces for SDConfig
@@ -37,23 +40,33 @@ type Selector struct {
 // ScrapeWorkConstructorFunc must construct ScrapeWork object for the given metaLabels.
 type ScrapeWorkConstructorFunc func(metaLabels map[string]string) interface{}
 
-// GetScrapeWorkObjects returns ScrapeWork objects for the given sdc and baseDir.
+// GetScrapeWorkObjects returns ScrapeWork objects for the given sdc.
+//
+// This function must be called after MustStart call.
+func (sdc *SDConfig) GetScrapeWorkObjects() ([]interface{}, error) {
+	if sdc.cfg == nil {
+		return nil, sdc.startErr
+	}
+	return sdc.cfg.aw.getScrapeWorkObjects(), nil
+}
+
+// MustStart initializes sdc before its usage.
 //
 // swcFunc is used for constructing such objects.
-func (sdc *SDConfig) GetScrapeWorkObjects(baseDir string, swcFunc ScrapeWorkConstructorFunc) ([]interface{}, error) {
-	cfg, err := getAPIConfig(sdc, baseDir, swcFunc)
+func (sdc *SDConfig) MustStart(baseDir string, swcFunc ScrapeWorkConstructorFunc) {
+	cfg, err := newAPIConfig(sdc, baseDir, swcFunc)
 	if err != nil {
-		return nil, fmt.Errorf("cannot create API config: %w", err)
+		sdc.startErr = fmt.Errorf("cannot create API config for kubernetes: %w", err)
+		return
 	}
-	return cfg.aw.getScrapeWorkObjects(), nil
+	cfg.aw.mustStart()
+	sdc.cfg = cfg
 }
 
 // MustStop stops further usage for sdc.
 func (sdc *SDConfig) MustStop() {
-	v := configMap.Delete(sdc)
-	if v != nil {
-		// v can be nil if GetLabels wasn't called yet.
-		cfg := v.(*apiConfig)
-		cfg.mustStop()
+	if sdc.cfg != nil {
+		// sdc.cfg can be nil on MustStart error.
+		sdc.cfg.mustStop()
 	}
 }
diff --git a/lib/promscrape/scraper.go b/lib/promscrape/scraper.go
index d02f1732f..7dd454891 100644
--- a/lib/promscrape/scraper.go
+++ b/lib/promscrape/scraper.go
@@ -93,6 +93,7 @@ func runScraper(configFile string, pushData func(wr *prompbmarshal.WriteRequest)
 	if err != nil {
 		logger.Fatalf("cannot read %q: %s", configFile, err)
 	}
+	cfg.mustStart()
 
 	scs := newScrapeConfigs(pushData)
 	scs.add("static_configs", 0, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getStaticScrapeWork() })
@@ -130,6 +131,7 @@ func runScraper(configFile string, pushData func(wr *prompbmarshal.WriteRequest)
 				goto waitForChans
 			}
 			cfg.mustStop()
+			cfgNew.mustStart()
 			cfg = cfgNew
 			data = dataNew
 		case <-tickerCh:
@@ -143,6 +145,7 @@ func runScraper(configFile string, pushData func(wr *prompbmarshal.WriteRequest)
 				goto waitForChans
 			}
 			cfg.mustStop()
+			cfgNew.mustStart()
 			cfg = cfgNew
 			data = dataNew
 		case <-globalStopCh:
@@ -231,17 +234,11 @@ func (scfg *scrapeConfig) run() {
 	cfg := <-scfg.cfgCh
 	var swsPrev []*ScrapeWork
 	updateScrapeWork := func(cfg *Config) {
-		for {
-			startTime := time.Now()
-			sws := scfg.getScrapeWork(cfg, swsPrev)
-			retry := sg.update(sws)
-			swsPrev = sws
-			scfg.discoveryDuration.UpdateDuration(startTime)
-			if !retry {
-				return
-			}
-			time.Sleep(2 * time.Second)
-		}
+		startTime := time.Now()
+		sws := scfg.getScrapeWork(cfg, swsPrev)
+		sg.update(sws)
+		swsPrev = sws
+		scfg.discoveryDuration.UpdateDuration(startTime)
 	}
 	updateScrapeWork(cfg)
 	atomic.AddInt32(&PendingScrapeConfigs, -1)
@@ -301,7 +298,7 @@ func (sg *scraperGroup) stop() {
 	sg.wg.Wait()
 }
 
-func (sg *scraperGroup) update(sws []*ScrapeWork) (retry bool) {
+func (sg *scraperGroup) update(sws []*ScrapeWork) {
 	sg.mLock.Lock()
 	defer sg.mLock.Unlock()
 
@@ -358,7 +355,6 @@ func (sg *scraperGroup) update(sws []*ScrapeWork) (retry bool) {
 		sg.changesCount.Add(additionsCount + deletionsCount)
 		logger.Infof("%s: added targets: %d, removed targets: %d; total targets: %d", sg.name, additionsCount, deletionsCount, len(sg.m))
 	}
-	return deletionsCount > 0 && len(sg.m) == 0
 }
 
 type scraper struct {
diff --git a/lib/proxy/proxy.go b/lib/proxy/proxy.go
index 087d35fb0..c792b6ae2 100644
--- a/lib/proxy/proxy.go
+++ b/lib/proxy/proxy.go
@@ -105,18 +105,13 @@ func (u *URL) NewDialFunc(ac *promauth.Config) (fasthttp.DialFunc, error) {
 		return defaultDialFunc, nil
 	}
 	pu := u.url
-	if pu.Scheme != "http" && pu.Scheme != "https" && pu.Scheme != "socks5" {
-		return nil, fmt.Errorf("unknown scheme=%q for proxy_url=%q, must be http, https or socks5", pu.Scheme, pu.Redacted())
+	switch pu.Scheme {
+	case "http", "https", "socks5", "tls+socks5":
+	default:
+		return nil, fmt.Errorf("unknown scheme=%q for proxy_url=%q, must be http, https, socks5 or tls+socks5", pu.Scheme, pu.Redacted())
 	}
-	isTLS := pu.Scheme == "https"
+	isTLS := (pu.Scheme == "https" || pu.Scheme == "tls+socks5")
 	proxyAddr := addMissingPort(pu.Host, isTLS)
-	if pu.Scheme == "socks5" {
-		return socks5DialFunc(proxyAddr, pu)
-	}
-	authHeader := u.GetAuthHeader(ac)
-	if authHeader != "" {
-		authHeader = "Proxy-Authorization: " + authHeader + "\r\n"
-	}
 	var tlsCfg *tls.Config
 	if isTLS {
 		tlsCfg = ac.NewTLSConfig()
@@ -124,6 +119,13 @@ func (u *URL) NewDialFunc(ac *promauth.Config) (fasthttp.DialFunc, error) {
 			tlsCfg.ServerName = tlsServerName(proxyAddr)
 		}
 	}
+	if pu.Scheme == "socks5" || pu.Scheme == "tls+socks5" {
+		return socks5DialFunc(proxyAddr, pu, tlsCfg)
+	}
+	authHeader := u.GetAuthHeader(ac)
+	if authHeader != "" {
+		authHeader = "Proxy-Authorization: " + authHeader + "\r\n"
+	}
 	dialFunc := func(addr string) (net.Conn, error) {
 		proxyConn, err := defaultDialFunc(proxyAddr)
 		if err != nil {
@@ -142,7 +144,7 @@ func (u *URL) NewDialFunc(ac *promauth.Config) (fasthttp.DialFunc, error) {
 	return dialFunc, nil
 }
 
-func socks5DialFunc(proxyAddr string, pu *url.URL) (fasthttp.DialFunc, error) {
+func socks5DialFunc(proxyAddr string, pu *url.URL, tlsCfg *tls.Config) (fasthttp.DialFunc, error) {
 	var sac *proxy.Auth
 	if pu.User != nil {
 		username := pu.User.Username()
@@ -153,7 +155,13 @@ func socks5DialFunc(proxyAddr string, pu *url.URL) (fasthttp.DialFunc, error) {
 		}
 	}
 	network := netutil.GetTCPNetwork()
-	d, err := proxy.SOCKS5(network, proxyAddr, sac, proxy.Direct)
+	var dialer proxy.Dialer = proxy.Direct
+	if tlsCfg != nil {
+		dialer = &tls.Dialer{
+			Config: tlsCfg,
+		}
+	}
+	d, err := proxy.SOCKS5(network, proxyAddr, sac, dialer)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create socks5 proxy for url: %s, err: %w", pu.Redacted(), err)
 	}


Test patch:
diff --git a/lib/persistentqueue/persistentqueue_test.go b/lib/persistentqueue/persistentqueue_test.go
index e4b83d540..e135eb415 100644
--- a/lib/persistentqueue/persistentqueue_test.go
+++ b/lib/persistentqueue/persistentqueue_test.go
@@ -5,16 +5,14 @@ import (
 	"io/ioutil"
 	"os"
 	"strconv"
-	"sync"
 	"testing"
-	"time"
 )
 
 func TestQueueOpenClose(t *testing.T) {
 	path := "queue-open-close"
 	mustDeleteDir(path)
 	for i := 0; i < 3; i++ {
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		if n := q.GetPendingBytes(); n > 0 {
 			t.Fatalf("pending bytes must be 0; got %d", n)
 		}
@@ -28,7 +26,7 @@ func TestQueueOpen(t *testing.T) {
 		path := "queue-open-invalid-metainfo"
 		mustCreateDir(path)
 		mustCreateFile(path+"/metainfo.json", "foobarbaz")
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -38,7 +36,7 @@ func TestQueueOpen(t *testing.T) {
 		mustCreateEmptyMetainfo(path, "foobar")
 		mustCreateFile(path+"/junk-file", "foobar")
 		mustCreateDir(path + "/junk-dir")
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -47,7 +45,7 @@ func TestQueueOpen(t *testing.T) {
 		mustCreateDir(path)
 		mustCreateEmptyMetainfo(path, "foobar")
 		mustCreateFile(fmt.Sprintf("%s/%016X", path, 1234), "qwere")
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -56,7 +54,7 @@ func TestQueueOpen(t *testing.T) {
 		mustCreateDir(path)
 		mustCreateEmptyMetainfo(path, "foobar")
 		mustCreateFile(fmt.Sprintf("%s/%016X", path, 100*uint64(defaultChunkFileSize)), "asdf")
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -72,7 +70,7 @@ func TestQueueOpen(t *testing.T) {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		mustCreateFile(fmt.Sprintf("%s/%016X", path, 0), "adfsfd")
-		q := MustOpen(path, mi.Name, 0)
+		q := mustOpen(path, mi.Name, 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -86,7 +84,7 @@ func TestQueueOpen(t *testing.T) {
 		if err := mi.WriteToFile(path + "/metainfo.json"); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		q := MustOpen(path, mi.Name, 0)
+		q := mustOpen(path, mi.Name, 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -94,7 +92,7 @@ func TestQueueOpen(t *testing.T) {
 		path := "queue-open-metainfo-dir"
 		mustCreateDir(path)
 		mustCreateDir(path + "/metainfo.json")
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -110,7 +108,7 @@ func TestQueueOpen(t *testing.T) {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		mustCreateFile(fmt.Sprintf("%s/%016X", path, 0), "sdf")
-		q := MustOpen(path, mi.Name, 0)
+		q := mustOpen(path, mi.Name, 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -119,7 +117,7 @@ func TestQueueOpen(t *testing.T) {
 		mustCreateDir(path)
 		mustCreateEmptyMetainfo(path, "foobar")
 		mustCreateFile(fmt.Sprintf("%s/%016X", path, 0), "sdfdsf")
-		q := MustOpen(path, "foobar", 0)
+		q := mustOpen(path, "foobar", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -133,7 +131,7 @@ func TestQueueOpen(t *testing.T) {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		mustCreateFile(fmt.Sprintf("%s/%016X", path, 0), "sdf")
-		q := MustOpen(path, "baz", 0)
+		q := mustOpen(path, "baz", 0)
 		q.MustClose()
 		mustDeleteDir(path)
 	})
@@ -142,7 +140,7 @@ func TestQueueOpen(t *testing.T) {
 func TestQueueResetIfEmpty(t *testing.T) {
 	path := "queue-reset-if-empty"
 	mustDeleteDir(path)
-	q := MustOpen(path, "foobar", 0)
+	q := mustOpen(path, "foobar", 0)
 	defer func() {
 		q.MustClose()
 		mustDeleteDir(path)
@@ -154,14 +152,18 @@ func TestQueueResetIfEmpty(t *testing.T) {
 		for i := 0; i < 10; i++ {
 			q.MustWriteBlock(block)
 			var ok bool
-			buf, ok = q.MustReadBlock(buf[:0])
+			buf, ok = q.MustReadBlockNonblocking(buf[:0])
 			if !ok {
-				t.Fatalf("unexpected ok=false returned from MustReadBlock")
+				t.Fatalf("unexpected ok=false returned from MustReadBlockNonblocking")
 			}
 		}
 		q.ResetIfEmpty()
 		if n := q.GetPendingBytes(); n > 0 {
-			t.Fatalf("unexpected non-zer pending bytes after queue reset: %d", n)
+			t.Fatalf("unexpected non-zero pending bytes after queue reset: %d", n)
+		}
+		q.ResetIfEmpty()
+		if n := q.GetPendingBytes(); n > 0 {
+			t.Fatalf("unexpected non-zero pending bytes after queue reset: %d", n)
 		}
 	}
 }
@@ -169,7 +171,7 @@ func TestQueueResetIfEmpty(t *testing.T) {
 func TestQueueWriteRead(t *testing.T) {
 	path := "queue-write-read"
 	mustDeleteDir(path)
-	q := MustOpen(path, "foobar", 0)
+	q := mustOpen(path, "foobar", 0)
 	defer func() {
 		q.MustClose()
 		mustDeleteDir(path)
@@ -188,9 +190,9 @@ func TestQueueWriteRead(t *testing.T) {
 		var buf []byte
 		var ok bool
 		for _, block := range blocks {
-			buf, ok = q.MustReadBlock(buf[:0])
+			buf, ok = q.MustReadBlockNonblocking(buf[:0])
 			if !ok {
-				t.Fatalf("unexpected ok=%v returned from MustReadBlock; want true", ok)
+				t.Fatalf("unexpected ok=%v returned from MustReadBlockNonblocking; want true", ok)
 			}
 			if string(buf) != string(block) {
 				t.Fatalf("unexpected block read; got %q; want %q", buf, block)
@@ -205,7 +207,7 @@ func TestQueueWriteRead(t *testing.T) {
 func TestQueueWriteCloseRead(t *testing.T) {
 	path := "queue-write-close-read"
 	mustDeleteDir(path)
-	q := MustOpen(path, "foobar", 0)
+	q := mustOpen(path, "foobar", 0)
 	defer func() {
 		q.MustClose()
 		mustDeleteDir(path)
@@ -222,16 +224,16 @@ func TestQueueWriteCloseRead(t *testing.T) {
 			t.Fatalf("pending bytes must be greater than 0; got %d", n)
 		}
 		q.MustClose()
-		q = MustOpen(path, "foobar", 0)
+		q = mustOpen(path, "foobar", 0)
 		if n := q.GetPendingBytes(); n <= 0 {
 			t.Fatalf("pending bytes must be greater than 0; got %d", n)
 		}
 		var buf []byte
 		var ok bool
 		for _, block := range blocks {
-			buf, ok = q.MustReadBlock(buf[:0])
+			buf, ok = q.MustReadBlockNonblocking(buf[:0])
 			if !ok {
-				t.Fatalf("unexpected ok=%v returned from MustReadBlock; want true", ok)
+				t.Fatalf("unexpected ok=%v returned from MustReadBlockNonblocking; want true", ok)
 			}
 			if string(buf) != string(block) {
 				t.Fatalf("unexpected block read; got %q; want %q", buf, block)
@@ -243,137 +245,12 @@ func TestQueueWriteCloseRead(t *testing.T) {
 	}
 }
 
-func TestQueueReadEmpty(t *testing.T) {
-	path := "queue-read-empty"
-	mustDeleteDir(path)
-	q := MustOpen(path, "foobar", 0)
-	defer mustDeleteDir(path)
-
-	resultCh := make(chan error)
-	go func() {
-		data, ok := q.MustReadBlock(nil)
-		var err error
-		if ok {
-			err = fmt.Errorf("unexpected ok=%v returned from MustReadBlock; want false", ok)
-		} else if len(data) > 0 {
-			err = fmt.Errorf("unexpected non-empty data returned from MustReadBlock: %q", data)
-		}
-		resultCh <- err
-	}()
-	if n := q.GetPendingBytes(); n > 0 {
-		t.Fatalf("pending bytes must be 0; got %d", n)
-	}
-	q.MustClose()
-	select {
-	case err := <-resultCh:
-		if err != nil {
-			t.Fatalf("unexpected error: %s", err)
-		}
-	case <-time.After(time.Second):
-		t.Fatalf("timeout")
-	}
-}
-
-func TestQueueReadWriteConcurrent(t *testing.T) {
-	path := "queue-read-write-concurrent"
-	mustDeleteDir(path)
-	q := MustOpen(path, "foobar", 0)
-	defer mustDeleteDir(path)
-
-	blocksMap := make(map[string]bool, 1000)
-	var blocksMapLock sync.Mutex
-	blocks := make([]string, 1000)
-	for i := 0; i < 1000; i++ {
-		block := fmt.Sprintf("block #%d", i)
-		blocksMap[block] = true
-		blocks[i] = block
-	}
-
-	// Start block readers
-	var readersWG sync.WaitGroup
-	for workerID := 0; workerID < 10; workerID++ {
-		readersWG.Add(1)
-		go func() {
-			defer readersWG.Done()
-			for {
-				block, ok := q.MustReadBlock(nil)
-				if !ok {
-					return
-				}
-				blocksMapLock.Lock()
-				if !blocksMap[string(block)] {
-					panic(fmt.Errorf("unexpected block read: %q", block))
-				}
-				delete(blocksMap, string(block))
-				blocksMapLock.Unlock()
-			}
-		}()
-	}
-
-	// Start block writers
-	blocksCh := make(chan string)
-	var writersWG sync.WaitGroup
-	for workerID := 0; workerID < 10; workerID++ {
-		writersWG.Add(1)
-		go func(workerID int) {
-			defer writersWG.Done()
-			for block := range blocksCh {
-				q.MustWriteBlock([]byte(block))
-			}
-		}(workerID)
-	}
-	for _, block := range blocks {
-		blocksCh <- block
-	}
-	close(blocksCh)
-
-	// Wait for block writers to finish
-	writersWG.Wait()
-
-	// Notify readers that the queue is closed
-	q.MustClose()
-
-	// Wait for block readers to finish
-	readersWG.Wait()
-
-	// Read the remaining blocks in q.
-	q = MustOpen(path, "foobar", 0)
-	defer q.MustClose()
-	resultCh := make(chan error)
-	go func() {
-		for len(blocksMap) > 0 {
-			block, ok := q.MustReadBlock(nil)
-			if !ok {
-				resultCh <- fmt.Errorf("unexpected ok=false returned from MustReadBlock")
-				return
-			}
-			if !blocksMap[string(block)] {
-				resultCh <- fmt.Errorf("unexpected block read from the queue: %q", block)
-				return
-			}
-			delete(blocksMap, string(block))
-		}
-		resultCh <- nil
-	}()
-	select {
-	case err := <-resultCh:
-		if err != nil {
-			t.Fatalf("unexpected error: %s", err)
-		}
-	case <-time.After(5 * time.Second):
-		t.Fatalf("timeout")
-	}
-	if n := q.GetPendingBytes(); n > 0 {
-		t.Fatalf("pending bytes must be 0; got %d", n)
-	}
-}
-
 func TestQueueChunkManagementSimple(t *testing.T) {
 	path := "queue-chunk-management-simple"
 	mustDeleteDir(path)
 	const chunkFileSize = 100
 	const maxBlockSize = 20
-	q := mustOpen(path, "foobar", chunkFileSize, maxBlockSize, 0)
+	q := mustOpenInternal(path, "foobar", chunkFileSize, maxBlockSize, 0)
 	defer mustDeleteDir(path)
 	defer q.MustClose()
 	var blocks []string
@@ -386,7 +263,7 @@ func TestQueueChunkManagementSimple(t *testing.T) {
 		t.Fatalf("unexpected zero number of bytes pending")
 	}
 	for _, block := range blocks {
-		data, ok := q.MustReadBlock(nil)
+		data, ok := q.MustReadBlockNonblocking(nil)
 		if !ok {
 			t.Fatalf("unexpected ok=false")
 		}
@@ -404,7 +281,7 @@ func TestQueueChunkManagementPeriodicClose(t *testing.T) {
 	mustDeleteDir(path)
 	const chunkFileSize = 100
 	const maxBlockSize = 20
-	q := mustOpen(path, "foobar", chunkFileSize, maxBlockSize, 0)
+	q := mustOpenInternal(path, "foobar", chunkFileSize, maxBlockSize, 0)
 	defer func() {
 		q.MustClose()
 		mustDeleteDir(path)
@@ -415,13 +292,13 @@ func TestQueueChunkManagementPeriodicClose(t *testing.T) {
 		q.MustWriteBlock([]byte(block))
 		blocks = append(blocks, block)
 		q.MustClose()
-		q = mustOpen(path, "foobar", chunkFileSize, maxBlockSize, 0)
+		q = mustOpenInternal(path, "foobar", chunkFileSize, maxBlockSize, 0)
 	}
 	if n := q.GetPendingBytes(); n == 0 {
 		t.Fatalf("unexpected zero number of bytes pending")
 	}
 	for _, block := range blocks {
-		data, ok := q.MustReadBlock(nil)
+		data, ok := q.MustReadBlockNonblocking(nil)
 		if !ok {
 			t.Fatalf("unexpected ok=false")
 		}
@@ -429,7 +306,7 @@ func TestQueueChunkManagementPeriodicClose(t *testing.T) {
 			t.Fatalf("unexpected block read; got %q; want %q", data, block)
 		}
 		q.MustClose()
-		q = mustOpen(path, "foobar", chunkFileSize, maxBlockSize, 0)
+		q = mustOpenInternal(path, "foobar", chunkFileSize, maxBlockSize, 0)
 	}
 	if n := q.GetPendingBytes(); n != 0 {
 		t.Fatalf("unexpected non-zero number of pending bytes: %d", n)
@@ -440,7 +317,7 @@ func TestQueueLimitedSize(t *testing.T) {
 	const maxPendingBytes = 1000
 	path := "queue-limited-size"
 	mustDeleteDir(path)
-	q := MustOpen(path, "foobar", maxPendingBytes)
+	q := mustOpen(path, "foobar", maxPendingBytes)
 	defer func() {
 		q.MustClose()
 		mustDeleteDir(path)
@@ -456,7 +333,7 @@ func TestQueueLimitedSize(t *testing.T) {
 	var buf []byte
 	var ok bool
 	for _, block := range blocks {
-		buf, ok = q.MustReadBlock(buf[:0])
+		buf, ok = q.MustReadBlockNonblocking(buf[:0])
 		if !ok {
 			t.Fatalf("unexpected ok=false")
 		}
@@ -473,7 +350,7 @@ func TestQueueLimitedSize(t *testing.T) {
 	if n := q.GetPendingBytes(); n > maxPendingBytes {
 		t.Fatalf("too many pending bytes; got %d; mustn't exceed %d", n, maxPendingBytes)
 	}
-	buf, ok = q.MustReadBlock(buf[:0])
+	buf, ok = q.MustReadBlockNonblocking(buf[:0])
 	if !ok {
 		t.Fatalf("unexpected ok=false")
 	}
diff --git a/lib/persistentqueue/persistentqueue_timing_test.go b/lib/persistentqueue/persistentqueue_timing_test.go
index 02e87513f..e7a3b7874 100644
--- a/lib/persistentqueue/persistentqueue_timing_test.go
+++ b/lib/persistentqueue/persistentqueue_timing_test.go
@@ -2,13 +2,14 @@ package persistentqueue
 
 import (
 	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 )
 
 func BenchmarkQueueThroughputSerial(b *testing.B) {
-	const iterationsCount = 10
+	const iterationsCount = 100
 	for _, blockSize := range []int{1e0, 1e1, 1e2, 1e3, 1e4, 1e5, 1e6} {
 		block := make([]byte, blockSize)
 		b.Run(fmt.Sprintf("block-size-%d", blockSize), func(b *testing.B) {
@@ -16,7 +17,7 @@ func BenchmarkQueueThroughputSerial(b *testing.B) {
 			b.SetBytes(int64(blockSize) * iterationsCount)
 			path := fmt.Sprintf("bench-queue-throughput-serial-%d", blockSize)
 			mustDeleteDir(path)
-			q := MustOpen(path, "foobar", 0)
+			q := mustOpen(path, "foobar", 0)
 			defer func() {
 				q.MustClose()
 				mustDeleteDir(path)
@@ -29,7 +30,7 @@ func BenchmarkQueueThroughputSerial(b *testing.B) {
 }
 
 func BenchmarkQueueThroughputConcurrent(b *testing.B) {
-	const iterationsCount = 10
+	const iterationsCount = 100
 	for _, blockSize := range []int{1e0, 1e1, 1e2, 1e3, 1e4, 1e5, 1e6} {
 		block := make([]byte, blockSize)
 		b.Run(fmt.Sprintf("block-size-%d", blockSize), func(b *testing.B) {
@@ -37,28 +38,31 @@ func BenchmarkQueueThroughputConcurrent(b *testing.B) {
 			b.SetBytes(int64(blockSize) * iterationsCount)
 			path := fmt.Sprintf("bench-queue-throughput-concurrent-%d", blockSize)
 			mustDeleteDir(path)
-			q := MustOpen(path, "foobar", 0)
+			q := mustOpen(path, "foobar", 0)
+			var qLock sync.Mutex
 			defer func() {
 				q.MustClose()
 				mustDeleteDir(path)
 			}()
 			b.RunParallel(func(pb *testing.PB) {
 				for pb.Next() {
+					qLock.Lock()
 					writeReadIteration(q, block, iterationsCount)
+					qLock.Unlock()
 				}
 			})
 		})
 	}
 }
 
-func writeReadIteration(q *Queue, block []byte, iterationsCount int) {
+func writeReadIteration(q *queue, block []byte, iterationsCount int) {
 	for i := 0; i < iterationsCount; i++ {
 		q.MustWriteBlock(block)
 	}
 	var ok bool
 	bb := bbPool.Get()
 	for i := 0; i < iterationsCount; i++ {
-		bb.B, ok = q.MustReadBlock(bb.B[:0])
+		bb.B, ok = q.MustReadBlockNonblocking(bb.B[:0])
 		if !ok {
 			panic(fmt.Errorf("unexpected ok=false"))
 		}