Repository: VictoriaMetrics/VictoriaMetrics
Pull request #579 (2020-06-22 12:22:55)

vmselect: fix label_replace when mismatch
As per documentation on `label_replace` function: "If the regular
expression doesn't match then the timeseries is returned unchanged".

Currently this behavior is not enforced, if a regexp on an existing
tag doesn't match then the tag value is copied as-is in the destination
tag. This fix first checks that the regular expression matches the
source tag before applying anything.

Given the current implementation, this fix also changes the behavior
of the **MetricsQL** `label_transform` function which does not
document this behavior at the moment.

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=h1) Report
> Merging [#579](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=desc) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/7532dbcdf54439fb33a66d9e9957f93072abf745&el=desc) will **increase** coverage by `0.09%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #579      +/-   ##
==========================================
+ Coverage   68.18%   68.27%   +0.09%     
==========================================
  Files         151      151              
  Lines       21081    21083       +2     
==========================================
+ Hits        14374    14395      +21     
+ Misses       5554     5529      -25     
- Partials     1153     1159       +6     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [app/vmselect/promql/transform.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC90cmFuc2Zvcm0uZ28=) | `87.12% <100.00%> (-0.17%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `90.72% <0.00%> (-1.38%)` | :arrow_down: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `71.54% <0.00%> (ø)` | |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `76.03% <0.00%> (+0.05%)` | :arrow_up: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `68.29% <0.00%> (+0.55%)` | :arrow_up: |
| [lib/storage/part.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydC5nbw==) | `74.48% <0.00%> (+1.37%)` | :arrow_up: |
| [lib/mergeset/block\_stream\_reader.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L2Jsb2NrX3N0cmVhbV9yZWFkZXIuZ28=) | `62.58% <0.00%> (+11.61%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=footer). Last update [7532dbc...9c9038e](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/579?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/app/vmagent/README.md b/app/vmagent/README.md
index ddf7ca715..cf2f1d444 100644
--- a/app/vmagent/README.md
+++ b/app/vmagent/README.md
@@ -170,6 +170,8 @@ Additionally it provides the following extra actions:
 
 * `replace_all`: replaces all the occurences of `regex` in the values of `source_labels` with the `replacement` and stores the result in the `target_label`.
 * `labelmap_all`: replaces all the occurences of `regex` in all the label names with the `replacement`.
+* `keep_if_equal`: keeps the entry if all label values from `source_labels` are equal.
+* `drop_if_equal`: drops the entry if all the label values from `source_labels` are equal.
 
 The relabeling can be defined in the following places:
 
@@ -210,6 +212,14 @@ either via `vmagent` itself or via Prometheus, so the exported metrics could be
   The directory can grow large when remote storage is unavailable for extended periods of time and if `-remoteWrite.maxDiskUsagePerURL` isn't set.
   If you don't want to send all the data from the directory to remote storage, simply stop `vmagent` and delete the directory.
 
+* If you see `skipping duplicate scrape target with identical labels` errors when scraping Kubernetes pods, then it is likely these pods listen multiple ports.
+  Just add the following relabeling rule to `relabel_configs` section in order to filter out targets with unneeded ports:
+
+```yml
+- action: keep_if_equal
+  source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_port, __meta_kubernetes_pod_container_port_number]
+```
+
 
 ### How to build from sources
 
diff --git a/app/vmalert/README.md b/app/vmalert/README.md
index be6830791..e6ba23cad 100644
--- a/app/vmalert/README.md
+++ b/app/vmalert/README.md
@@ -158,6 +158,16 @@ Usage of vmalert:
         Optional basic auth password for -datasource.url
   -datasource.basicAuth.username string
         Optional basic auth username for -datasource.url
+  -datasource.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -datasource.url. By default system CA is used.
+  -datasource.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -datasource.url.
+  -datasource.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -datasource.url
+  -datasource.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -datasource.url.
+  -datasource.tlsServerName value
+        Optional TLS server name to use for connections to -datasource.url. By default the server name from -datasource.url is used.
   -datasource.url string
         Victoria Metrics or VMSelect url. Required parameter. E.g. http://127.0.0.1:8428
   -evaluationInterval duration
@@ -168,6 +178,16 @@ Usage of vmalert:
         Address to listen for http connections (default ":8880")
   -metricsAuthKey string
         Auth key for /metrics. It overrides httpAuth settings
+  -notifier.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -notifier.url. By default system CA is used.
+  -notifier.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -notifier.url.
+  -notifier.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -notifier.url
+  -notifier.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -notifier.url.
+  -notifier.tlsServerName value
+        Optional TLS server name to use for connections to -notifier.url. By default the server name from -notifier.url is used.
   -notifier.url string
         Prometheus alertmanager URL. Required parameter. e.g. http://127.0.0.1:9093
   -remoteRead.basicAuth.password string
@@ -176,6 +196,16 @@ Usage of vmalert:
         Optional basic auth username for -remoteRead.url
   -remoteRead.lookback duration
         Lookback defines how far to look into past for alerts timeseries. For example, if lookback=1h then range from now() to now()-1h will be scanned. (default 1h0m0s)
+  -remoteRead.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -remoteRead.url. By default system CA is used.
+  -remoteRead.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -remoteRead.url.
+  -remoteRead.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -remoteRead.url
+  -remoteRead.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -remoteRead.url.
+  -remoteRead.tlsServerName value
+        Optional TLS server name to use for connections to -remoteRead.url. By default the server name from -remoteRead.url is used.
   -remoteRead.url vmalert
         Optional URL to Victoria Metrics or VMSelect that will be used to restore alerts state. This configuration makes sense only if vmalert was configured with `remoteWrite.url` before and has been successfully persisted its state. E.g. http://127.0.0.1:8428
   -remoteWrite.basicAuth.password string
@@ -188,6 +218,16 @@ Usage of vmalert:
         Defines defines max number of timeseries to be flushed at once (default 1000)
   -remoteWrite.maxQueueSize int
         Defines the max number of pending datapoints to remote write endpoint (default 100000)
+  -remoteWrite.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -remoteWrite.url. By default system CA is used.
+  -remoteWrite.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -remoteWrite.url.
+  -remoteWrite.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -remoteWrite.url
+  -remoteWrite.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -remoteWrite.url.
+  -remoteWrite.tlsServerName value
+        Optional TLS server name to use for connections to -remoteWrite.url. By default the server name from -remoteWrite.url is used.
   -remoteWrite.url string
         Optional URL to Victoria Metrics or VMInsert where to persist alerts state in form of timeseries. E.g. http://127.0.0.1:8428
   -rule value
diff --git a/app/vmalert/main.go b/app/vmalert/main.go
index 550f18899..07d5da715 100644
--- a/app/vmalert/main.go
+++ b/app/vmalert/main.go
@@ -2,8 +2,11 @@ package main
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"flag"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"net/url"
 	"os"
@@ -21,6 +24,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
+	"github.com/VictoriaMetrics/fasthttp"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -39,16 +43,30 @@ absolute path to all .yaml files in root.`)
 
 	datasourceURL = flag.String("datasource.url", "", "Victoria Metrics or VMSelect url. Required parameter."+
 		" E.g. http://127.0.0.1:8428")
-	basicAuthUsername = flag.String("datasource.basicAuth.username", "", "Optional basic auth username for -datasource.url")
-	basicAuthPassword = flag.String("datasource.basicAuth.password", "", "Optional basic auth password for -datasource.url")
+	basicAuthUsername               = flag.String("datasource.basicAuth.username", "", "Optional basic auth username for -datasource.url")
+	basicAuthPassword               = flag.String("datasource.basicAuth.password", "", "Optional basic auth password for -datasource.url")
+	datasourceTLSInsecureSkipVerify = flag.Bool("datasource.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -datasource.url")
+	datasourceTLSCertFile           = flag.String("datasource.tlsCertFile", "", "Optional path to client-side TLS certificate file to use when connecting to -datasource.url")
+	datasourceTLSKeyFile            = flag.String("datasource.tlsKeyFile", "", "Optional path to client-side TLS certificate key to use when connecting to -datasource.url")
+	datasourceTLSCAFile             = flag.String("datasource.tlsCAFile", "", "Optional path to TLS CA file to use for verifying connections to -datasource.url. "+
+		"By default system CA is used")
+	datasourceTLSServerName = flag.String("datasource.tlsServerName", "", "Optional TLS server name to use for connections to -datasource.url. "+
+		"By default the server name from -datasource.url is used")
 
 	remoteWriteURL = flag.String("remoteWrite.url", "", "Optional URL to Victoria Metrics or VMInsert where to persist alerts state"+
 		" and recording rules results in form of timeseries. E.g. http://127.0.0.1:8428")
-	remoteWriteUsername     = flag.String("remoteWrite.basicAuth.username", "", "Optional basic auth username for -remoteWrite.url")
-	remoteWritePassword     = flag.String("remoteWrite.basicAuth.password", "", "Optional basic auth password for -remoteWrite.url")
-	remoteWriteMaxQueueSize = flag.Int("remoteWrite.maxQueueSize", 1e5, "Defines the max number of pending datapoints to remote write endpoint")
-	remoteWriteMaxBatchSize = flag.Int("remoteWrite.maxBatchSize", 1e3, "Defines defines max number of timeseries to be flushed at once")
-	remoteWriteConcurrency  = flag.Int("remoteWrite.concurrency", 1, "Defines number of writers for concurrent writing into remote storage")
+	remoteWriteUsername              = flag.String("remoteWrite.basicAuth.username", "", "Optional basic auth username for -remoteWrite.url")
+	remoteWritePassword              = flag.String("remoteWrite.basicAuth.password", "", "Optional basic auth password for -remoteWrite.url")
+	remoteWriteMaxQueueSize          = flag.Int("remoteWrite.maxQueueSize", 1e5, "Defines the max number of pending datapoints to remote write endpoint")
+	remoteWriteMaxBatchSize          = flag.Int("remoteWrite.maxBatchSize", 1e3, "Defines defines max number of timeseries to be flushed at once")
+	remoteWriteConcurrency           = flag.Int("remoteWrite.concurrency", 1, "Defines number of writers for concurrent writing into remote storage")
+	remoteWriteTLSInsecureSkipVerify = flag.Bool("remoteWrite.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -remoteWrite.url")
+	remoteWriteTLSCertFile           = flag.String("remoteWrite.tlsCertFile", "", "Optional path to client-side TLS certificate file to use when connecting to -remoteWrite.url")
+	remoteWriteTLSKeyFile            = flag.String("remoteWrite.tlsKeyFile", "", "Optional path to client-side TLS certificate key to use when connecting to -remoteWrite.url")
+	remoteWriteTLSCAFile             = flag.String("remoteWrite.tlsCAFile", "", "Optional path to TLS CA file to use for verifying connections to -remoteWrite.url. "+
+		"By default system CA is used")
+	remoteWriteTLSServerName = flag.String("remoteWrite.tlsServerName", "", "Optional TLS server name to use for connections to -remoteWrite.url. "+
+		"By default the server name from -remoteWrite.url is used")
 
 	remoteReadURL = flag.String("remoteRead.url", "", "Optional URL to Victoria Metrics or VMSelect that will be used to restore alerts"+
 		" state. This configuration makes sense only if `vmalert` was configured with `remoteWrite.url` before and has been successfully persisted its state."+
@@ -57,9 +75,23 @@ absolute path to all .yaml files in root.`)
 	remoteReadPassword = flag.String("remoteRead.basicAuth.password", "", "Optional basic auth password for -remoteRead.url")
 	remoteReadLookBack = flag.Duration("remoteRead.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries."+
 		" For example, if lookback=1h then range from now() to now()-1h will be scanned.")
+	remoteReadTLSInsecureSkipVerify = flag.Bool("remoteRead.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -remoteRead.url")
+	remoteReadTLSCertFile           = flag.String("remoteRead.tlsCertFile", "", "Optional path to client-side TLS certificate file to use when connecting to -remoteRead.url")
+	remoteReadTLSKeyFile            = flag.String("remoteRead.tlsKeyFile", "", "Optional path to client-side TLS certificate key to use when connecting to -remoteRead.url")
+	remoteReadTLSCAFile             = flag.String("remoteRead.tlsCAFile", "", "Optional path to TLS CA file to use for verifying connections to -remoteRead.url. "+
+		"By default system CA is used")
+	remoteReadTLSServerName = flag.String("remoteRead.tlsServerName", "", "Optional TLS server name to use for connections to -remoteRead.url. "+
+		"By default the server name from -remoteRead.url is used")
 
-	evaluationInterval  = flag.Duration("evaluationInterval", time.Minute, "How often to evaluate the rules")
-	notifierURL         = flag.String("notifier.url", "", "Prometheus alertmanager URL. Required parameter. e.g. http://127.0.0.1:9093")
+	evaluationInterval            = flag.Duration("evaluationInterval", time.Minute, "How often to evaluate the rules")
+	notifierURL                   = flag.String("notifier.url", "", "Prometheus alertmanager URL. Required parameter. e.g. http://127.0.0.1:9093")
+	notifierTLSInsecureSkipVerify = flag.Bool("notifier.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -notifier.url")
+	notifierTLSCertFile           = flag.String("notifier.tlsCertFile", "", "Optional path to client-side TLS certificate file to use when connecting to -notifier.url")
+	notifierTLSKeyFile            = flag.String("notifier.tlsKeyFile", "", "Optional path to client-side TLS certificate key to use when connecting to -notifier.url")
+	notifierTLSCAFile             = flag.String("notifier.tlsCAFile", "", "Optional path to TLS CA file to use for verifying connections to -notifier.url. "+
+		"By default system CA is used")
+	notifierTLSServerName = flag.String("notifier.tlsServerName", "", "Optional TLS server name to use for connections to -notifier.url. "+
+		"By default the server name from -notifier.url is used")
 	externalURL         = flag.String("external.url", "", "External URL is used as alert's source for sent alerts to the notifier")
 	externalAlertSource = flag.String("external.alert.source", "", `External Alert Source allows to override the Source link for alerts sent to AlertManager for cases where you want to build a custom link to Grafana, Prometheus or any other service.
 eg. 'explore?orgId=1&left=[\"now-1h\",\"now\",\"VictoriaMetrics\",{\"expr\": \"{{$expr|quotesEscape|pathEscape}}\"},{\"mode\":\"Metrics\"},{\"ui\":[true,true,true,\"none\"]}]'.If empty '/api/v1/:groupID/alertID/status' is used`)
@@ -84,12 +116,27 @@ func main() {
 		logger.Fatalf("URL generator error: %s", err)
 	}
 
+	dst, err := getTransport(datasourceURL, datasourceTLSCertFile, datasourceTLSKeyFile, datasourceTLSCAFile, datasourceTLSServerName, datasourceTLSInsecureSkipVerify)
+	if err != nil {
+		logger.Fatalf("cannot create datasource transport: %s", err)
+	}
+
+	nt, err := getTransport(notifierURL, notifierTLSCertFile, notifierTLSKeyFile, notifierTLSCAFile, notifierTLSServerName, notifierTLSInsecureSkipVerify)
+	if err != nil {
+		logger.Fatalf("cannot create notifier transport: %s", err)
+	}
+
 	manager := &manager{
 		groups:   make(map[uint64]*Group),
-		storage:  datasource.NewVMStorage(*datasourceURL, *basicAuthUsername, *basicAuthPassword, &http.Client{}),
-		notifier: notifier.NewAlertManager(*notifierURL, aug, &http.Client{}),
+		storage:  datasource.NewVMStorage(*datasourceURL, *basicAuthUsername, *basicAuthPassword, &http.Client{Transport: dst}),
+		notifier: notifier.NewAlertManager(*notifierURL, aug, &http.Client{Transport: nt}),
 	}
 	if *remoteWriteURL != "" {
+		t, err := getTransport(remoteWriteURL, remoteWriteTLSCertFile, remoteWriteTLSKeyFile, remoteWriteTLSCAFile, remoteWriteTLSServerName, remoteWriteTLSInsecureSkipVerify)
+		if err != nil {
+			logger.Fatalf("cannot create remoteWrite transport: %s", err)
+		}
+
 		c, err := remotewrite.NewClient(ctx, remotewrite.Config{
 			Addr:          *remoteWriteURL,
 			Concurrency:   *remoteWriteConcurrency,
@@ -98,14 +145,21 @@ func main() {
 			FlushInterval: *evaluationInterval,
 			BasicAuthUser: *remoteWriteUsername,
 			BasicAuthPass: *remoteWritePassword,
+			Transport:     t,
 		})
 		if err != nil {
 			logger.Fatalf("failed to init remotewrite client: %s", err)
 		}
 		manager.rw = c
 	}
+
 	if *remoteReadURL != "" {
-		manager.rr = datasource.NewVMStorage(*remoteReadURL, *remoteReadUsername, *remoteReadPassword, &http.Client{})
+		t, err := getTransport(remoteReadURL, remoteReadTLSCertFile, remoteReadTLSKeyFile, remoteReadTLSCAFile, remoteReadTLSServerName, remoteReadTLSInsecureSkipVerify)
+		if err != nil {
+			logger.Fatalf("cannot create remoteRead transport: %s", err)
+		}
+
+		manager.rr = datasource.NewVMStorage(*remoteReadURL, *remoteReadUsername, *remoteReadPassword, &http.Client{Transport: t})
 	}
 
 	if err := manager.start(ctx, *rulePath, *validateTemplates, *validateExpressions); err != nil {
@@ -196,6 +250,57 @@ func getAlertURLGenerator(externalURL *url.URL, externalAlertSource string, vali
 	}, nil
 }
 
+func getTLSConfig(certFile, keyFile, CAFile, serverName *string, insecureSkipVerify *bool) (*tls.Config, error) {
+	var certs []tls.Certificate
+	if *certFile != "" {
+		cert, err := tls.LoadX509KeyPair(*certFile, *keyFile)
+		if err != nil {
+			return nil, fmt.Errorf("cannot load TLS certificate from `cert_file`=%q, `key_file`=%q: %s", *certFile, *keyFile, err)
+		}
+
+		certs = []tls.Certificate{cert}
+	}
+
+	var rootCAs *x509.CertPool
+	if *CAFile != "" {
+		pem, err := ioutil.ReadFile(*CAFile)
+		if err != nil {
+			return nil, fmt.Errorf("cannot read `ca_file` %q: %s", *CAFile, err)
+		}
+
+		rootCAs = x509.NewCertPool()
+		if !rootCAs.AppendCertsFromPEM(pem) {
+			return nil, fmt.Errorf("cannot parse data from `ca_file` %q", *CAFile)
+		}
+	}
+
+	return &tls.Config{
+		Certificates:       certs,
+		InsecureSkipVerify: *insecureSkipVerify,
+		RootCAs:            rootCAs,
+		ServerName:         *serverName,
+	}, nil
+}
+
+func getTransport(URL, certFile, keyFile, CAFile, serverName *string, insecureSkipVerify *bool) (*http.Transport, error) {
+	var u fasthttp.URI
+	u.Update(*URL)
+
+	var t *http.Transport
+	if string(u.Scheme()) == "https" {
+		t = http.DefaultTransport.(*http.Transport).Clone()
+
+		tlsCfg, err := getTLSConfig(certFile, keyFile, CAFile, serverName, insecureSkipVerify)
+		if err != nil {
+			return nil, err
+		}
+
+		t.TLSClientConfig = tlsCfg
+	}
+
+	return t, nil
+}
+
 func checkFlags() {
 	if *notifierURL == "" {
 		flag.PrintDefaults()
diff --git a/app/vmalert/remotewrite/remotewrite.go b/app/vmalert/remotewrite/remotewrite.go
index 8ac8733ae..acb919ceb 100644
--- a/app/vmalert/remotewrite/remotewrite.go
+++ b/app/vmalert/remotewrite/remotewrite.go
@@ -53,6 +53,8 @@ type Config struct {
 	// WriteTimeout defines timeout for HTTP write request
 	// to remote storage
 	WriteTimeout time.Duration
+	// Transport will be used by the underlying http.Client
+	Transport *http.Transport
 }
 
 const (
@@ -85,7 +87,8 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 	}
 	c := &Client{
 		c: &http.Client{
-			Timeout: cfg.WriteTimeout,
+			Timeout:   cfg.WriteTimeout,
+			Transport: cfg.Transport,
 		},
 		addr:          strings.TrimSuffix(cfg.Addr, "/") + writePath,
 		baUser:        cfg.BasicAuthUser,
diff --git a/app/vmselect/netstorage/netstorage.go b/app/vmselect/netstorage/netstorage.go
index 570be1cbb..6bec68f8b 100644
--- a/app/vmselect/netstorage/netstorage.go
+++ b/app/vmselect/netstorage/netstorage.go
@@ -7,13 +7,11 @@ import (
 	"runtime"
 	"sort"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -72,6 +70,47 @@ func (rss *Results) mustClose() {
 	rss.sr = nil
 }
 
+var timeseriesWorkCh = make(chan *timeseriesWork, gomaxprocs)
+
+type timeseriesWork struct {
+	rss    *Results
+	pts    *packedTimeseries
+	f      func(rs *Result, workerID uint)
+	doneCh chan error
+
+	rowsProcessed int
+}
+
+func init() {
+	for i := 0; i < gomaxprocs; i++ {
+		go timeseriesWorker(uint(i))
+	}
+}
+
+func timeseriesWorker(workerID uint) {
+	var rs Result
+	for tsw := range timeseriesWorkCh {
+		rss := tsw.rss
+		if time.Until(rss.deadline.Deadline) < 0 {
+			tsw.doneCh <- fmt.Errorf("timeout exceeded during query execution: %s", rss.deadline.String())
+			continue
+		}
+		if err := tsw.pts.Unpack(&rs, rss.tr, rss.fetchData); err != nil {
+			tsw.doneCh <- fmt.Errorf("error during time series unpacking: %s", err)
+			continue
+		}
+		if len(rs.Timestamps) > 0 || !rss.fetchData {
+			tsw.f(&rs, workerID)
+		}
+		tsw.rowsProcessed = len(rs.Values)
+		tsw.doneCh <- nil
+		if cap(rs.Values) > 1024*1024 && 4*len(rs.Values) < cap(rs.Values) {
+			// Reset rs in order to preseve memory usage after processing big time series with millions of rows.
+			rs = Result{}
+		}
+	}
+}
+
 // RunParallel runs in parallel f for all the results from rss.
 //
 // f shouldn't hold references to rs after returning.
@@ -81,72 +120,36 @@ func (rss *Results) mustClose() {
 func (rss *Results) RunParallel(f func(rs *Result, workerID uint)) error {
 	defer rss.mustClose()
 
-	workersCount := 1 + len(rss.packedTimeseries)/32
-	if workersCount > gomaxprocs {
-		workersCount = gomaxprocs
-	}
-	if workersCount == 0 {
-		logger.Panicf("BUG: workersCount cannot be zero")
-	}
-	workCh := make(chan *packedTimeseries, workersCount)
-	doneCh := make(chan error)
-
-	// Start workers.
-	rowsProcessedTotal := uint64(0)
-	for i := 0; i < workersCount; i++ {
-		go func(workerID uint) {
-			rs := getResult()
-			defer putResult(rs)
-			maxWorkersCount := gomaxprocs / workersCount
-
-			var err error
-			rowsProcessed := 0
-			for pts := range workCh {
-				if time.Until(rss.deadline.Deadline) < 0 {
-					err = fmt.Errorf("timeout exceeded during query execution: %s", rss.deadline.String())
-					break
-				}
-				if err = pts.Unpack(rs, rss.tr, rss.fetchData, maxWorkersCount); err != nil {
-					break
-				}
-				if len(rs.Timestamps) == 0 && rss.fetchData {
-					// Skip empty blocks.
-					continue
-				}
-				rowsProcessed += len(rs.Values)
-				f(rs, workerID)
-			}
-			atomic.AddUint64(&rowsProcessedTotal, uint64(rowsProcessed))
-			// Drain the remaining work
-			for range workCh {
-			}
-			doneCh <- err
-		}(uint(i))
-	}
-
 	// Feed workers with work.
+	tsws := make([]*timeseriesWork, len(rss.packedTimeseries))
 	for i := range rss.packedTimeseries {
-		workCh <- &rss.packedTimeseries[i]
+		tsw := &timeseriesWork{
+			rss:    rss,
+			pts:    &rss.packedTimeseries[i],
+			f:      f,
+			doneCh: make(chan error, 1),
+		}
+		timeseriesWorkCh <- tsw
+		tsws[i] = tsw
 	}
 	seriesProcessedTotal := len(rss.packedTimeseries)
 	rss.packedTimeseries = rss.packedTimeseries[:0]
-	close(workCh)
 
-	// Wait until workers finish.
-	var errors []error
-	for i := 0; i < workersCount; i++ {
-		if err := <-doneCh; err != nil {
-			errors = append(errors, err)
+	// Wait until work is complete.
+	var firstErr error
+	rowsProcessedTotal := 0
+	for _, tsw := range tsws {
+		if err := <-tsw.doneCh; err != nil && firstErr == nil {
+			// Return just the first error, since other errors
+			// are likely duplicate the first error.
+			firstErr = err
 		}
+		rowsProcessedTotal += tsw.rowsProcessed
 	}
+
 	perQueryRowsProcessed.Update(float64(rowsProcessedTotal))
 	perQuerySeriesProcessed.Update(float64(seriesProcessedTotal))
-	if len(errors) > 0 {
-		// Return just the first error, since other errors
-		// is likely duplicate the first error.
-		return errors[0]
-	}
-	return nil
+	return firstErr
 }
 
 var perQueryRowsProcessed = metrics.NewHistogram(`vm_per_query_rows_processed_count`)
@@ -159,70 +162,74 @@ type packedTimeseries struct {
 	brs        []storage.BlockRef
 }
 
-// Unpack unpacks pts to dst.
-func (pts *packedTimeseries) Unpack(dst *Result, tr storage.TimeRange, fetchData bool, maxWorkersCount int) error {
-	dst.reset()
+var unpackWorkCh = make(chan *unpackWork, gomaxprocs)
 
-	if err := dst.MetricName.Unmarshal(bytesutil.ToUnsafeBytes(pts.metricName)); err != nil {
-		return fmt.Errorf("cannot unmarshal metricName %q: %s", pts.metricName, err)
-	}
+type unpackWork struct {
+	br        storage.BlockRef
+	tr        storage.TimeRange
+	fetchData bool
+	doneCh    chan error
+	sb        *sortBlock
+}
 
-	workersCount := 1 + len(pts.brs)/32
-	if workersCount > maxWorkersCount {
-		workersCount = maxWorkersCount
+func init() {
+	for i := 0; i < gomaxprocs; i++ {
+		go unpackWorker()
 	}
-	if workersCount == 0 {
-		logger.Panicf("BUG: workersCount cannot be zero")
+}
+
+func unpackWorker() {
+	for upw := range unpackWorkCh {
+		sb := getSortBlock()
+		if err := sb.unpackFrom(upw.br, upw.tr, upw.fetchData); err != nil {
+			putSortBlock(sb)
+			upw.doneCh <- fmt.Errorf("cannot unpack block: %s", err)
+			continue
+		}
+		upw.sb = sb
+		upw.doneCh <- nil
 	}
+}
 
-	sbs := make([]*sortBlock, 0, len(pts.brs))
-	var sbsLock sync.Mutex
-
-	workCh := make(chan storage.BlockRef, workersCount)
-	doneCh := make(chan error)
-
-	// Start workers
-	for i := 0; i < workersCount; i++ {
-		go func() {
-			var err error
-			for br := range workCh {
-				sb := getSortBlock()
-				if err = sb.unpackFrom(br, tr, fetchData); err != nil {
-					break
-				}
-
-				sbsLock.Lock()
-				sbs = append(sbs, sb)
-				sbsLock.Unlock()
-			}
+// Unpack unpacks pts to dst.
+func (pts *packedTimeseries) Unpack(dst *Result, tr storage.TimeRange, fetchData bool) error {
+	dst.reset()
 
-			// Drain the remaining work
-			for range workCh {
-			}
-			doneCh <- err
-		}()
+	if err := dst.MetricName.Unmarshal(bytesutil.ToUnsafeBytes(pts.metricName)); err != nil {
+		return fmt.Errorf("cannot unmarshal metricName %q: %s", pts.metricName, err)
 	}
 
 	// Feed workers with work
-	for _, br := range pts.brs {
-		workCh <- br
+	upws := make([]*unpackWork, len(pts.brs))
+	for i, br := range pts.brs {
+		upw := &unpackWork{
+			br:        br,
+			tr:        tr,
+			fetchData: fetchData,
+			doneCh:    make(chan error, 1),
+		}
+		unpackWorkCh <- upw
+		upws[i] = upw
 	}
 	pts.brs = pts.brs[:0]
-	close(workCh)
 
-	// Wait until workers finish
-	var errors []error
-	for i := 0; i < workersCount; i++ {
-		if err := <-doneCh; err != nil {
-			errors = append(errors, err)
+	// Wait until work is complete
+	sbs := make([]*sortBlock, 0, len(pts.brs))
+	var firstErr error
+	for _, upw := range upws {
+		if err := <-upw.doneCh; err != nil && firstErr == nil {
+			// Return the first error only, since other errors are likely the same.
+			firstErr = err
+		}
+		if firstErr == nil {
+			sbs = append(sbs, upw.sb)
+		} else {
+			putSortBlock(upw.sb)
 		}
 	}
-	if len(errors) > 0 {
-		// Return the first error only, since other errors are likely the same.
-		return errors[0]
+	if firstErr != nil {
+		return firstErr
 	}
-
-	// Merge blocks
 	mergeSortBlocks(dst, sbs)
 	return nil
 }
@@ -537,25 +544,6 @@ func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadli
 	return &rss, nil
 }
 
-func getResult() *Result {
-	v := rsPool.Get()
-	if v == nil {
-		return &Result{}
-	}
-	return v.(*Result)
-}
-
-func putResult(rs *Result) {
-	if len(rs.Values) > 8192 {
-		// Do not pool big results, since they may occupy too much memory.
-		return
-	}
-	rs.reset()
-	rsPool.Put(rs)
-}
-
-var rsPool sync.Pool
-
 func setupTfss(tagFilterss [][]storage.TagFilter) ([]*storage.TagFilters, error) {
 	tfss := make([]*storage.TagFilters, 0, len(tagFilterss))
 	for _, tagFilters := range tagFilterss {
diff --git a/app/vmselect/promql/transform.go b/app/vmselect/promql/transform.go
index b08f9e8a4..1f0f85812 100644
--- a/app/vmselect/promql/transform.go
+++ b/app/vmselect/promql/transform.go
@@ -1219,6 +1219,9 @@ func labelReplace(tss []*timeseries, srcLabel string, r *regexp.Regexp, dstLabel
 		mn := &ts.MetricName
 		dstValue := getDstValue(mn, dstLabel)
 		srcValue := mn.GetTagValue(srcLabel)
+		if !r.Match(srcValue) {
+			continue
+		}
 		b := r.ReplaceAll(srcValue, replacementBytes)
 		*dstValue = append((*dstValue)[:0], b...)
 		if len(b) == 0 {
diff --git a/docs/vmagent.md b/docs/vmagent.md
index ddf7ca715..cf2f1d444 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -170,6 +170,8 @@ Additionally it provides the following extra actions:
 
 * `replace_all`: replaces all the occurences of `regex` in the values of `source_labels` with the `replacement` and stores the result in the `target_label`.
 * `labelmap_all`: replaces all the occurences of `regex` in all the label names with the `replacement`.
+* `keep_if_equal`: keeps the entry if all label values from `source_labels` are equal.
+* `drop_if_equal`: drops the entry if all the label values from `source_labels` are equal.
 
 The relabeling can be defined in the following places:
 
@@ -210,6 +212,14 @@ either via `vmagent` itself or via Prometheus, so the exported metrics could be
   The directory can grow large when remote storage is unavailable for extended periods of time and if `-remoteWrite.maxDiskUsagePerURL` isn't set.
   If you don't want to send all the data from the directory to remote storage, simply stop `vmagent` and delete the directory.
 
+* If you see `skipping duplicate scrape target with identical labels` errors when scraping Kubernetes pods, then it is likely these pods listen multiple ports.
+  Just add the following relabeling rule to `relabel_configs` section in order to filter out targets with unneeded ports:
+
+```yml
+- action: keep_if_equal
+  source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_port, __meta_kubernetes_pod_container_port_number]
+```
+
 
 ### How to build from sources
 
diff --git a/docs/vmalert.md b/docs/vmalert.md
index be6830791..e6ba23cad 100644
--- a/docs/vmalert.md
+++ b/docs/vmalert.md
@@ -158,6 +158,16 @@ Usage of vmalert:
         Optional basic auth password for -datasource.url
   -datasource.basicAuth.username string
         Optional basic auth username for -datasource.url
+  -datasource.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -datasource.url. By default system CA is used.
+  -datasource.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -datasource.url.
+  -datasource.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -datasource.url
+  -datasource.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -datasource.url.
+  -datasource.tlsServerName value
+        Optional TLS server name to use for connections to -datasource.url. By default the server name from -datasource.url is used.
   -datasource.url string
         Victoria Metrics or VMSelect url. Required parameter. E.g. http://127.0.0.1:8428
   -evaluationInterval duration
@@ -168,6 +178,16 @@ Usage of vmalert:
         Address to listen for http connections (default ":8880")
   -metricsAuthKey string
         Auth key for /metrics. It overrides httpAuth settings
+  -notifier.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -notifier.url. By default system CA is used.
+  -notifier.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -notifier.url.
+  -notifier.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -notifier.url
+  -notifier.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -notifier.url.
+  -notifier.tlsServerName value
+        Optional TLS server name to use for connections to -notifier.url. By default the server name from -notifier.url is used.
   -notifier.url string
         Prometheus alertmanager URL. Required parameter. e.g. http://127.0.0.1:9093
   -remoteRead.basicAuth.password string
@@ -176,6 +196,16 @@ Usage of vmalert:
         Optional basic auth username for -remoteRead.url
   -remoteRead.lookback duration
         Lookback defines how far to look into past for alerts timeseries. For example, if lookback=1h then range from now() to now()-1h will be scanned. (default 1h0m0s)
+  -remoteRead.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -remoteRead.url. By default system CA is used.
+  -remoteRead.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -remoteRead.url.
+  -remoteRead.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -remoteRead.url
+  -remoteRead.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -remoteRead.url.
+  -remoteRead.tlsServerName value
+        Optional TLS server name to use for connections to -remoteRead.url. By default the server name from -remoteRead.url is used.
   -remoteRead.url vmalert
         Optional URL to Victoria Metrics or VMSelect that will be used to restore alerts state. This configuration makes sense only if vmalert was configured with `remoteWrite.url` before and has been successfully persisted its state. E.g. http://127.0.0.1:8428
   -remoteWrite.basicAuth.password string
@@ -188,6 +218,16 @@ Usage of vmalert:
         Defines defines max number of timeseries to be flushed at once (default 1000)
   -remoteWrite.maxQueueSize int
         Defines the max number of pending datapoints to remote write endpoint (default 100000)
+  -remoteWrite.tlsCAFile value
+        Optional path to TLS CA file to use for verifying connections to -remoteWrite.url. By default system CA is used.
+  -remoteWrite.tlsCertFile value
+        Optional path to client-side TLS certificate file to use when connecting to -remoteWrite.url.
+  -remoteWrite.tlsInsecureSkipVerify
+        Whether to skip tls verification when connecting to -remoteWrite.url
+  -remoteWrite.tlsKeyFile value
+        Optional path to client-side TLS certificate key to use when connecting to -remoteWrite.url.
+  -remoteWrite.tlsServerName value
+        Optional TLS server name to use for connections to -remoteWrite.url. By default the server name from -remoteWrite.url is used.
   -remoteWrite.url string
         Optional URL to Victoria Metrics or VMInsert where to persist alerts state in form of timeseries. E.g. http://127.0.0.1:8428
   -rule value
diff --git a/go.mod b/go.mod
index 540be24e2..197a1ad8b 100644
--- a/go.mod
+++ b/go.mod
@@ -13,7 +13,7 @@ require (
 	github.com/aws/aws-sdk-go v1.32.5
 	github.com/cespare/xxhash/v2 v2.1.1
 	github.com/golang/snappy v0.0.1
-	github.com/klauspost/compress v1.10.9
+	github.com/klauspost/compress v1.10.10
 	github.com/valyala/fastjson v1.5.1
 	github.com/valyala/fastrand v1.0.0
 	github.com/valyala/gozstd v1.7.0
diff --git a/go.sum b/go.sum
index 85f27ac54..781e6ceef 100644
--- a/go.sum
+++ b/go.sum
@@ -141,8 +141,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/compress v1.10.4/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/compress v1.10.5 h1:7q6vHIqubShURwQz8cQK6yIe/xC3IF0Vm7TGfqjewrc=
 github.com/klauspost/compress v1.10.5/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
-github.com/klauspost/compress v1.10.9 h1:pPRt1Z78crspaHISkpSSHjDlx+Tt9suHe519dsI0vF4=
-github.com/klauspost/compress v1.10.9/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/klauspost/compress v1.10.10 h1:a/y8CglcM7gLGYmlbP/stPE5sR3hbhFRUjCBfd/0B3I=
+github.com/klauspost/compress v1.10.10/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
diff --git a/lib/fs/reader_at.go b/lib/fs/reader_at.go
index 7dcdbbd04..3cd00afb5 100644
--- a/lib/fs/reader_at.go
+++ b/lib/fs/reader_at.go
@@ -90,6 +90,11 @@ func (r *ReaderAt) MustReadAt(p []byte, off int64) {
 }
 
 func (r *ReaderAt) isInPageCache(start, end int64) bool {
+	if int64(len(r.mmapData))-end < 4096 {
+		// If standard copy(dst, src) from Go may read beyond len(src), then this should help
+		// fixing SIGBUS panic from https://github.com/VictoriaMetrics/VictoriaMetrics/issues/581
+		return false
+	}
 	startBit := uint64(start) / pageSize
 	endBit := uint64(end) / pageSize
 	m := r.pageCacheBitmap.Load().(*pageCacheBitmap).m
@@ -133,7 +138,7 @@ func (r *ReaderAt) MustClose() {
 
 	fname := r.f.Name()
 	if len(r.mmapData) > 0 {
-		if err := unix.Munmap(r.mmapData); err != nil {
+		if err := unix.Munmap(r.mmapData[:cap(r.mmapData)]); err != nil {
 			logger.Panicf("FATAL: cannot unmap data for file %q: %s", fname, err)
 		}
 		r.mmapData = nil
@@ -214,9 +219,16 @@ func mmapFile(f *os.File, size int64) ([]byte, error) {
 	if int64(int(size)) != size {
 		return nil, fmt.Errorf("file is too big to be mmap'ed: %d bytes", size)
 	}
+	// Round size to multiple of 4KB pages as `man 2 mmap` recommends.
+	// This may help preventing SIGBUS panic at https://github.com/VictoriaMetrics/VictoriaMetrics/issues/581
+	// The SIGBUS could occur if standard copy(dst, src) function may read beyond src bounds.
+	sizeOrig := size
+	if size%4096 != 0 {
+		size += 4096 - size%4096
+	}
 	data, err := unix.Mmap(int(f.Fd()), 0, int(size), unix.PROT_READ, unix.MAP_SHARED)
 	if err != nil {
 		return nil, fmt.Errorf("cannot mmap file with size %d: %s", size, err)
 	}
-	return data, nil
+	return data[:sizeOrig], nil
 }
diff --git a/lib/logger/logger.go b/lib/logger/logger.go
index 54393423e..a2eb5d543 100644
--- a/lib/logger/logger.go
+++ b/lib/logger/logger.go
@@ -21,6 +21,9 @@ var (
 	loggerLevel  = flag.String("loggerLevel", "INFO", "Minimum level of errors to log. Possible values: INFO, WARN, ERROR, FATAL, PANIC")
 	loggerFormat = flag.String("loggerFormat", "default", "Format for logs. Possible values: default, json")
 	loggerOutput = flag.String("loggerOutput", "stderr", "Output for the logs. Supported values: stderr, stdout")
+
+	errorsPerSecondLimit = flag.Int("loggerErrorsPerSecondLimit", 10, "Per-second limit on the number of ERROR messages. If more than the given number of errors "+
+		"are emitted per second, then the remaining errors are suppressed. Zero value disables the rate limit")
 )
 
 // Init initializes the logger.
@@ -123,7 +126,7 @@ func logLevelSkipframes(skipframes int, level, format string, args ...interface{
 
 func errorsLoggedCleaner() {
 	for {
-		time.Sleep(5 * time.Second)
+		time.Sleep(time.Second)
 		atomic.StoreUint64(&errorsLogged, 0)
 	}
 }
@@ -141,7 +144,7 @@ func (lw *logWriter) Write(p []byte) (int, error) {
 func logMessage(level, msg string, skipframes int) {
 	// rate limit ERROR log messages
 	if level == "ERROR" {
-		if n := atomic.AddUint64(&errorsLogged, 1); n > 10 {
+		if n := atomic.AddUint64(&errorsLogged, 1); *errorsPerSecondLimit > 0 && n > uint64(*errorsPerSecondLimit) {
 			return
 		}
 	}
diff --git a/lib/promrelabel/config.go b/lib/promrelabel/config.go
index 282c48c08..5a5d60286 100644
--- a/lib/promrelabel/config.go
+++ b/lib/promrelabel/config.go
@@ -92,6 +92,14 @@ func parseRelabelConfig(dst []ParsedRelabelConfig, rc *RelabelConfig) ([]ParsedR
 		if targetLabel == "" {
 			return dst, fmt.Errorf("missing `target_label` for `action=replace`")
 		}
+	case "keep_if_equal":
+		if len(sourceLabels) < 2 {
+			return dst, fmt.Errorf("`source_labels` must contain at least two entries for `action=keep_if_equal`; got %q", sourceLabels)
+		}
+	case "drop_if_equal":
+		if len(sourceLabels) < 2 {
+			return dst, fmt.Errorf("`source_labels` must contain at least two entries for `action=drop_if_equal`; got %q", sourceLabels)
+		}
 	case "keep":
 		if len(sourceLabels) == 0 {
 			return dst, fmt.Errorf("missing `source_labels` for `action=keep`")
diff --git a/lib/promrelabel/relabel.go b/lib/promrelabel/relabel.go
index 211fd77a6..8432f8cff 100644
--- a/lib/promrelabel/relabel.go
+++ b/lib/promrelabel/relabel.go
@@ -148,6 +148,30 @@ func applyRelabelConfig(labels []prompbmarshal.Label, labelsOffset int, prc *Par
 		relabelBufPool.Put(bb)
 		valueStr := prc.Regex.ReplaceAllString(sourceStr, prc.Replacement)
 		return setLabelValue(labels, labelsOffset, prc.TargetLabel, valueStr)
+	case "keep_if_equal":
+		// Keep the entry if all the label values in source_labels are equal.
+		// For example:
+		//
+		//   - source_labels: [foo, bar]
+		//     action: keep_if_equal
+		//
+		// Would leave the entry if `foo` value equals `bar` value
+		if areEqualLabelValues(src, prc.SourceLabels) {
+			return labels
+		}
+		return labels[:labelsOffset]
+	case "drop_if_equal":
+		// Drop the entry if all the label values in source_labels are equal.
+		// For example:
+		//
+		//   - source_labels: [foo, bar]
+		//     action: drop_if_equal
+		//
+		// Would drop the entry if `foo` value equals `bar` value.
+		if areEqualLabelValues(src, prc.SourceLabels) {
+			return labels[:labelsOffset]
+		}
+		return labels
 	case "keep":
 		bb := relabelBufPool.Get()
 		bb.B = concatLabelValues(bb.B[:0], src, prc.SourceLabels, prc.Separator)
@@ -249,6 +273,21 @@ func (prc *ParsedRelabelConfig) expandCaptureGroups(template, source string, mat
 
 var relabelBufPool bytesutil.ByteBufferPool
 
+func areEqualLabelValues(labels []prompbmarshal.Label, labelNames []string) bool {
+	if len(labelNames) < 2 {
+		logger.Panicf("BUG: expecting at least 2 labelNames; got %d", len(labelNames))
+		return false
+	}
+	labelValue := GetLabelValueByName(labels, labelNames[0])
+	for _, labelName := range labelNames[1:] {
+		v := GetLabelValueByName(labels, labelName)
+		if v != labelValue {
+			return false
+		}
+	}
+	return true
+}
+
 func concatLabelValues(dst []byte, labels []prompbmarshal.Label, labelNames []string, separator string) []byte {
 	if len(labelNames) == 0 {
 		return dst
diff --git a/lib/promscrape/client.go b/lib/promscrape/client.go
index b52d227eb..f4e28936b 100644
--- a/lib/promscrape/client.go
+++ b/lib/promscrape/client.go
@@ -133,6 +133,8 @@ var (
 )
 
 func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response) error {
+	attempts := 0
+again:
 	// There is no need in calling DoTimeout, since the timeout must be already set in hc.ReadTimeout.
 	err := hc.Do(req, resp)
 	if err == nil {
@@ -142,5 +144,9 @@ func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request,
 		return err
 	}
 	// Retry request if the server closed the keep-alive connection during the first attempt.
-	return hc.Do(req, resp)
+	attempts++
+	if attempts > 3 {
+		return fmt.Errorf("the server closed 3 subsequent connections: %s", err)
+	}
+	goto again
 }
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index ed06b938c..b71b88f2c 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -158,77 +158,160 @@ func unmarshalMaybeStrict(data []byte, dst interface{}) error {
 	return err
 }
 
+func getSWSByJob(sws []ScrapeWork) map[string][]ScrapeWork {
+	m := make(map[string][]ScrapeWork)
+	for _, sw := range sws {
+		m[sw.jobNameOriginal] = append(m[sw.jobNameOriginal], sw)
+	}
+	return m
+}
+
 // getKubernetesSDScrapeWork returns `kubernetes_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getKubernetesSDScrapeWork() []ScrapeWork {
+func (cfg *Config) getKubernetesSDScrapeWork(prev []ScrapeWork) []ScrapeWork {
+	swsPrevByJob := getSWSByJob(prev)
 	var dst []ScrapeWork
 	for i := range cfg.ScrapeConfigs {
 		sc := &cfg.ScrapeConfigs[i]
+		dstLen := len(dst)
+		ok := true
 		for j := range sc.KubernetesSDConfigs {
 			sdc := &sc.KubernetesSDConfigs[j]
-			dst = appendKubernetesScrapeWork(dst, sdc, cfg.baseDir, sc.swc)
+			var okLocal bool
+			dst, okLocal = appendKubernetesScrapeWork(dst, sdc, cfg.baseDir, sc.swc)
+			if ok {
+				ok = okLocal
+			}
+		}
+		if ok {
+			continue
+		}
+		swsPrev := swsPrevByJob[sc.swc.jobName]
+		if len(swsPrev) > 0 {
+			logger.Errorf("there were errors when discovering kubernetes targets for job %q, so preserving the previous targets", sc.swc.jobName)
+			dst = append(dst[:dstLen], swsPrev...)
 		}
 	}
 	return dst
 }
 
 // getConsulSDScrapeWork returns `consul_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getConsulSDScrapeWork() []ScrapeWork {
+func (cfg *Config) getConsulSDScrapeWork(prev []ScrapeWork) []ScrapeWork {
+	swsPrevByJob := getSWSByJob(prev)
 	var dst []ScrapeWork
 	for i := range cfg.ScrapeConfigs {
 		sc := &cfg.ScrapeConfigs[i]
+		dstLen := len(dst)
+		ok := true
 		for j := range sc.ConsulSDConfigs {
 			sdc := &sc.ConsulSDConfigs[j]
-			dst = appendConsulScrapeWork(dst, sdc, cfg.baseDir, sc.swc)
+			var okLocal bool
+			dst, okLocal = appendConsulScrapeWork(dst, sdc, cfg.baseDir, sc.swc)
+			if ok {
+				ok = okLocal
+			}
+		}
+		if ok {
+			continue
+		}
+		swsPrev := swsPrevByJob[sc.swc.jobName]
+		if len(swsPrev) > 0 {
+			logger.Errorf("there were errors when discovering consul targets for job %q, so preserving the previous targets", sc.swc.jobName)
+			dst = append(dst[:dstLen], swsPrev...)
 		}
 	}
 	return dst
 }
 
 // getDNSSDScrapeWork returns `dns_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getDNSSDScrapeWork() []ScrapeWork {
+func (cfg *Config) getDNSSDScrapeWork(prev []ScrapeWork) []ScrapeWork {
+	swsPrevByJob := getSWSByJob(prev)
 	var dst []ScrapeWork
 	for i := range cfg.ScrapeConfigs {
 		sc := &cfg.ScrapeConfigs[i]
+		dstLen := len(dst)
+		ok := true
 		for j := range sc.DNSSDConfigs {
 			sdc := &sc.DNSSDConfigs[j]
-			dst = appendDNSScrapeWork(dst, sdc, sc.swc)
+			var okLocal bool
+			dst, okLocal = appendDNSScrapeWork(dst, sdc, sc.swc)
+			if ok {
+				ok = okLocal
+			}
+		}
+		if ok {
+			continue
+		}
+		swsPrev := swsPrevByJob[sc.swc.jobName]
+		if len(swsPrev) > 0 {
+			logger.Errorf("there were errors when discovering dns targets for job %q, so preserving the previous targets", sc.swc.jobName)
+			dst = append(dst[:dstLen], swsPrev...)
 		}
 	}
 	return dst
 }
 
 // getEC2SDScrapeWork returns `ec2_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getEC2SDScrapeWork() []ScrapeWork {
+func (cfg *Config) getEC2SDScrapeWork(prev []ScrapeWork) []ScrapeWork {
+	swsPrevByJob := getSWSByJob(prev)
 	var dst []ScrapeWork
 	for i := range cfg.ScrapeConfigs {
 		sc := &cfg.ScrapeConfigs[i]
+		dstLen := len(dst)
+		ok := true
 		for j := range sc.EC2SDConfigs {
 			sdc := &sc.EC2SDConfigs[j]
-			dst = appendEC2ScrapeWork(dst, sdc, sc.swc)
+			var okLocal bool
+			dst, okLocal = appendEC2ScrapeWork(dst, sdc, sc.swc)
+			if ok {
+				ok = okLocal
+			}
+		}
+		if ok {
+			continue
+		}
+		swsPrev := swsPrevByJob[sc.swc.jobName]
+		if len(swsPrev) > 0 {
+			logger.Errorf("there were errors when discovering ec2 targets for job %q, so preserving the previous targets", sc.swc.jobName)
+			dst = append(dst[:dstLen], swsPrev...)
 		}
 	}
 	return dst
 }
 
 // getGCESDScrapeWork returns `gce_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getGCESDScrapeWork() []ScrapeWork {
+func (cfg *Config) getGCESDScrapeWork(prev []ScrapeWork) []ScrapeWork {
+	swsPrevByJob := getSWSByJob(prev)
 	var dst []ScrapeWork
 	for i := range cfg.ScrapeConfigs {
 		sc := &cfg.ScrapeConfigs[i]
+		dstLen := len(dst)
+		ok := true
 		for j := range sc.GCESDConfigs {
 			sdc := &sc.GCESDConfigs[j]
-			dst = appendGCEScrapeWork(dst, sdc, sc.swc)
+			var okLocal bool
+			dst, okLocal = appendGCEScrapeWork(dst, sdc, sc.swc)
+			if ok {
+				ok = okLocal
+			}
+		}
+		if ok {
+			continue
+		}
+		swsPrev := swsPrevByJob[sc.swc.jobName]
+		if len(swsPrev) > 0 {
+			logger.Errorf("there were errors when discovering gce targets for job %q, so preserving the previous targets", sc.swc.jobName)
+			dst = append(dst[:dstLen], swsPrev...)
 		}
 	}
 	return dst
 }
 
 // getFileSDScrapeWork returns `file_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getFileSDScrapeWork(swsPrev []ScrapeWork) []ScrapeWork {
+func (cfg *Config) getFileSDScrapeWork(prev []ScrapeWork) []ScrapeWork {
 	// Create a map for the previous scrape work.
 	swsMapPrev := make(map[string][]ScrapeWork)
-	for i := range swsPrev {
-		sw := &swsPrev[i]
+	for i := range prev {
+		sw := &prev[i]
 		filepath := promrelabel.GetLabelValueByName(sw.Labels, "__vm_filepath")
 		if len(filepath) == 0 {
 			logger.Panicf("BUG: missing `__vm_filepath` label")
@@ -341,49 +424,49 @@ type scrapeWorkConfig struct {
 	sampleLimit          int
 }
 
-func appendKubernetesScrapeWork(dst []ScrapeWork, sdc *kubernetes.SDConfig, baseDir string, swc *scrapeWorkConfig) []ScrapeWork {
+func appendKubernetesScrapeWork(dst []ScrapeWork, sdc *kubernetes.SDConfig, baseDir string, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
 	targetLabels, err := kubernetes.GetLabels(sdc, baseDir)
 	if err != nil {
 		logger.Errorf("error when discovering kubernetes targets for `job_name` %q: %s; skipping it", swc.jobName, err)
-		return dst
+		return dst, false
 	}
-	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "kubernetes_sd_config")
+	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "kubernetes_sd_config"), true
 }
 
-func appendConsulScrapeWork(dst []ScrapeWork, sdc *consul.SDConfig, baseDir string, swc *scrapeWorkConfig) []ScrapeWork {
+func appendConsulScrapeWork(dst []ScrapeWork, sdc *consul.SDConfig, baseDir string, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
 	targetLabels, err := consul.GetLabels(sdc, baseDir)
 	if err != nil {
 		logger.Errorf("error when discovering consul targets for `job_name` %q: %s; skipping it", swc.jobName, err)
-		return dst
+		return dst, false
 	}
-	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "consul_sd_config")
+	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "consul_sd_config"), true
 }
 
-func appendDNSScrapeWork(dst []ScrapeWork, sdc *dns.SDConfig, swc *scrapeWorkConfig) []ScrapeWork {
+func appendDNSScrapeWork(dst []ScrapeWork, sdc *dns.SDConfig, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
 	targetLabels, err := dns.GetLabels(sdc)
 	if err != nil {
 		logger.Errorf("error when discovering dns targets for `job_name` %q: %s; skipping it", swc.jobName, err)
-		return dst
+		return dst, false
 	}
-	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "dns_sd_config")
+	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "dns_sd_config"), true
 }
 
-func appendEC2ScrapeWork(dst []ScrapeWork, sdc *ec2.SDConfig, swc *scrapeWorkConfig) []ScrapeWork {
+func appendEC2ScrapeWork(dst []ScrapeWork, sdc *ec2.SDConfig, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
 	targetLabels, err := ec2.GetLabels(sdc)
 	if err != nil {
 		logger.Errorf("error when discovering ec2 targets for `job_name` %q: %s; skipping it", swc.jobName, err)
-		return dst
+		return dst, false
 	}
-	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "ec2_sd_config")
+	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "ec2_sd_config"), true
 }
 
-func appendGCEScrapeWork(dst []ScrapeWork, sdc *gce.SDConfig, swc *scrapeWorkConfig) []ScrapeWork {
+func appendGCEScrapeWork(dst []ScrapeWork, sdc *gce.SDConfig, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
 	targetLabels, err := gce.GetLabels(sdc)
 	if err != nil {
 		logger.Errorf("error when discovering gce targets for `job_name` %q: %s; skippint it", swc.jobName, err)
-		return dst
+		return dst, false
 	}
-	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "gce_sd_config")
+	return appendScrapeWorkForTargetLabels(dst, swc, targetLabels, "gce_sd_config"), true
 }
 
 func appendScrapeWorkForTargetLabels(dst []ScrapeWork, swc *scrapeWorkConfig, targetLabels []map[string]string, sectionName string) []ScrapeWork {
@@ -519,6 +602,8 @@ func appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfig, target string, ex
 		AuthConfig:           swc.authConfig,
 		MetricRelabelConfigs: swc.metricRelabelConfigs,
 		SampleLimit:          swc.sampleLimit,
+
+		jobNameOriginal: swc.jobName,
 	})
 	return dst, nil
 }
diff --git a/lib/promscrape/scraper.go b/lib/promscrape/scraper.go
index 900313694..ef1dd8999 100644
--- a/lib/promscrape/scraper.go
+++ b/lib/promscrape/scraper.go
@@ -84,11 +84,11 @@ func runScraper(configFile string, pushData func(wr *prompbmarshal.WriteRequest)
 	scs := newScrapeConfigs(pushData)
 	scs.add("static_configs", 0, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getStaticScrapeWork() })
 	scs.add("file_sd_configs", *fileSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getFileSDScrapeWork(swsPrev) })
-	scs.add("kubernetes_sd_configs", *kubernetesSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getKubernetesSDScrapeWork() })
-	scs.add("consul_sd_configs", *consulSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getConsulSDScrapeWork() })
-	scs.add("dns_sd_configs", *dnsSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getDNSSDScrapeWork() })
-	scs.add("ec2_sd_configs", *ec2SDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getEC2SDScrapeWork() })
-	scs.add("gce_sd_configs", *gceSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getGCESDScrapeWork() })
+	scs.add("kubernetes_sd_configs", *kubernetesSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getKubernetesSDScrapeWork(swsPrev) })
+	scs.add("consul_sd_configs", *consulSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getConsulSDScrapeWork(swsPrev) })
+	scs.add("dns_sd_configs", *dnsSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getDNSSDScrapeWork(swsPrev) })
+	scs.add("ec2_sd_configs", *ec2SDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getEC2SDScrapeWork(swsPrev) })
+	scs.add("gce_sd_configs", *gceSDCheckInterval, func(cfg *Config, swsPrev []ScrapeWork) []ScrapeWork { return cfg.getGCESDScrapeWork(swsPrev) })
 
 	sighupCh := procutil.NewSighupChan()
 
diff --git a/lib/promscrape/scrapework.go b/lib/promscrape/scrapework.go
index 76f863aaa..85bc1cc76 100644
--- a/lib/promscrape/scrapework.go
+++ b/lib/promscrape/scrapework.go
@@ -66,6 +66,9 @@ type ScrapeWork struct {
 
 	// The maximum number of metrics to scrape after relabeling.
 	SampleLimit int
+
+	// The original 'job_name'
+	jobNameOriginal string
 }
 
 // key returns unique identifier for the given sw.
diff --git a/vendor/github.com/klauspost/compress/zstd/blockenc.go b/vendor/github.com/klauspost/compress/zstd/blockenc.go
index 4f0eba22f..c584f6aab 100644
--- a/vendor/github.com/klauspost/compress/zstd/blockenc.go
+++ b/vendor/github.com/klauspost/compress/zstd/blockenc.go
@@ -444,9 +444,9 @@ func fuzzFseEncoder(data []byte) int {
 }
 
 // encode will encode the block and append the output in b.output.
-func (b *blockEnc) encode(raw bool) error {
+func (b *blockEnc) encode(raw, rawAllLits bool) error {
 	if len(b.sequences) == 0 {
-		return b.encodeLits(raw)
+		return b.encodeLits(rawAllLits)
 	}
 	// We want some difference
 	if len(b.literals) > (b.size - (b.size >> 5)) {
diff --git a/vendor/github.com/klauspost/compress/zstd/encoder.go b/vendor/github.com/klauspost/compress/zstd/encoder.go
index bf42bb1cf..c56d2241f 100644
--- a/vendor/github.com/klauspost/compress/zstd/encoder.go
+++ b/vendor/github.com/klauspost/compress/zstd/encoder.go
@@ -280,7 +280,7 @@ func (e *Encoder) nextBlock(final bool) error {
 			// If we got the exact same number of literals as input,
 			// assume the literals cannot be compressed.
 			if len(src) != len(blk.literals) || len(src) != e.o.blockSize {
-				err = blk.encode(e.o.noEntropy)
+				err = blk.encode(e.o.noEntropy, !e.o.allLitEntropy)
 			}
 			switch err {
 			case errIncompressible:
@@ -491,7 +491,7 @@ func (e *Encoder) EncodeAll(src, dst []byte) []byte {
 		if len(blk.literals) != len(src) || len(src) != e.o.blockSize {
 			// Output directly to dst
 			blk.output = dst
-			err = blk.encode(e.o.noEntropy)
+			err = blk.encode(e.o.noEntropy, !e.o.allLitEntropy)
 		}
 
 		switch err {
@@ -528,7 +528,7 @@ func (e *Encoder) EncodeAll(src, dst []byte) []byte {
 			// If we got the exact same number of literals as input,
 			// assume the literals cannot be compressed.
 			if len(blk.literals) != len(todo) || len(todo) != e.o.blockSize {
-				err = blk.encode(e.o.noEntropy)
+				err = blk.encode(e.o.noEntropy, !e.o.allLitEntropy)
 			}
 
 			switch err {
diff --git a/vendor/github.com/klauspost/compress/zstd/encoder_options.go b/vendor/github.com/klauspost/compress/zstd/encoder_options.go
index 3fc03097a..dfac14ddd 100644
--- a/vendor/github.com/klauspost/compress/zstd/encoder_options.go
+++ b/vendor/github.com/klauspost/compress/zstd/encoder_options.go
@@ -12,16 +12,18 @@ type EOption func(*encoderOptions) error
 
 // options retains accumulated state of multiple options.
 type encoderOptions struct {
-	concurrent   int
-	level        EncoderLevel
-	single       *bool
-	pad          int
-	blockSize    int
-	windowSize   int
-	crc          bool
-	fullZero     bool
-	noEntropy    bool
-	customWindow bool
+	concurrent      int
+	level           EncoderLevel
+	single          *bool
+	pad             int
+	blockSize       int
+	windowSize      int
+	crc             bool
+	fullZero        bool
+	noEntropy       bool
+	allLitEntropy   bool
+	customWindow    bool
+	customALEntropy bool
 }
 
 func (o *encoderOptions) setDefault() {
@@ -207,6 +209,10 @@ func WithEncoderLevel(l EncoderLevel) EOption {
 				o.windowSize = 16 << 20
 			}
 		}
+		if !o.customALEntropy {
+			o.allLitEntropy = l > SpeedFastest
+		}
+
 		return nil
 	}
 }
@@ -221,6 +227,18 @@ func WithZeroFrames(b bool) EOption {
 	}
 }
 
+// WithAllLitEntropyCompression will apply entropy compression if no matches are found.
+// Disabling this will skip incompressible data faster, but in cases with no matches but
+// skewed character distribution compression is lost.
+// Default value depends on the compression level selected.
+func WithAllLitEntropyCompression(b bool) EOption {
+	return func(o *encoderOptions) error {
+		o.customALEntropy = true
+		o.allLitEntropy = b
+		return nil
+	}
+}
+
 // WithNoEntropyCompression will always skip entropy compression of literals.
 // This can be useful if content has matches, but unlikely to benefit from entropy
 // compression. Usually the slight speed improvement is not worth enabling this.
diff --git a/vendor/github.com/klauspost/compress/zstd/snappy.go b/vendor/github.com/klauspost/compress/zstd/snappy.go
index 356956ba2..690428cd2 100644
--- a/vendor/github.com/klauspost/compress/zstd/snappy.go
+++ b/vendor/github.com/klauspost/compress/zstd/snappy.go
@@ -178,7 +178,7 @@ func (r *SnappyConverter) Convert(in io.Reader, w io.Writer) (int64, error) {
 				r.err = ErrSnappyCorrupt
 				return written, r.err
 			}
-			err = r.block.encode(false)
+			err = r.block.encode(false, false)
 			switch err {
 			case errIncompressible:
 				r.block.popOffsets()
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 0c6854843..316166fce 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -90,7 +90,7 @@ github.com/jmespath/go-jmespath
 github.com/jstemmer/go-junit-report
 github.com/jstemmer/go-junit-report/formatter
 github.com/jstemmer/go-junit-report/parser
-# github.com/klauspost/compress v1.10.9
+# github.com/klauspost/compress v1.10.10
 github.com/klauspost/compress/flate
 github.com/klauspost/compress/fse
 github.com/klauspost/compress/gzip


Test patch:
diff --git a/app/vmalert/main_test.go b/app/vmalert/main_test.go
index 853cdcf3f..0bd574cba 100644
--- a/app/vmalert/main_test.go
+++ b/app/vmalert/main_test.go
@@ -51,3 +51,55 @@ func TestGetAlertURLGenerator(t *testing.T) {
 		t.Errorf("unexpected url want %s, got %s", exp, fn(testAlert))
 	}
 }
+
+func TestGetTLSConfig(t *testing.T) {
+	var certFile, keyFile, CAFile, serverName string
+	var insecureSkipVerify bool
+	serverName = "test"
+	insecureSkipVerify = true
+	tlsCfg, err := getTLSConfig(&certFile, &keyFile, &CAFile, &serverName, &insecureSkipVerify)
+	if err != nil {
+		t.Errorf("unexpected error %s", err)
+	}
+	if tlsCfg == nil {
+		t.Errorf("expected tlsConfig to be set, got nil")
+	}
+	if tlsCfg.ServerName != serverName {
+		t.Errorf("unexpected ServerName, want %s, got %s", serverName, tlsCfg.ServerName)
+	}
+	if tlsCfg.InsecureSkipVerify != insecureSkipVerify {
+		t.Errorf("unexpected InsecureSkipVerify, want %v, got %v", insecureSkipVerify, tlsCfg.InsecureSkipVerify)
+	}
+	certFile = "/path/to/nonexisting/cert/file"
+	_, err = getTLSConfig(&certFile, &keyFile, &CAFile, &serverName, &insecureSkipVerify)
+	if err == nil {
+		t.Errorf("expected keypair error, got nil")
+	}
+	certFile = ""
+	CAFile = "/path/to/nonexisting/cert/file"
+	_, err = getTLSConfig(&certFile, &keyFile, &CAFile, &serverName, &insecureSkipVerify)
+	if err == nil {
+		t.Errorf("expected read error, got nil")
+	}
+}
+
+func TestGetTransport(t *testing.T) {
+	var certFile, keyFile, CAFile, serverName string
+	var insecureSkipVerify bool
+	URL := "http://victoriametrics.com"
+	tr, err := getTransport(&URL, &certFile, &keyFile, &CAFile, &serverName, &insecureSkipVerify)
+	if err != nil {
+		t.Errorf("unexpected error %s", err)
+	}
+	if tr != nil {
+		t.Errorf("expected Transport to be nil, got %v", tr)
+	}
+	URL = "https://victoriametrics.com"
+	tr, err = getTransport(&URL, &certFile, &keyFile, &CAFile, &serverName, &insecureSkipVerify)
+	if err != nil {
+		t.Errorf("unexpected error %s", err)
+	}
+	if tr.TLSClientConfig == nil {
+		t.Errorf("expected TLSClientConfig to be set, got nil")
+	}
+}
diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index f6f8dfb01..a7c015d9b 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -1480,7 +1480,7 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
-	t.Run(`label_replace(mismatch)`, func(t *testing.T) {
+	t.Run(`label_replace(nonexisting_src)`, func(t *testing.T) {
 		t.Parallel()
 		q := `label_replace(time(), "__name__", "x${1}y", "foo", ".+")`
 		r := netstorage.Result{
@@ -1491,6 +1491,21 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`label_replace(mismatch)`, func(t *testing.T) {
+		t.Parallel()
+		q := `label_replace(label_set(time(), "foo", "foobar"), "__name__", "x${1}y", "foo", "bar(.+)")`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{1000, 1200, 1400, 1600, 1800, 2000},
+			Timestamps: timestampsExpected,
+		}
+		r.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("foo"),
+			Value: []byte("foobar"),
+		}}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`label_replace(match)`, func(t *testing.T) {
 		t.Parallel()
 		q := `label_replace(time(), "__name__", "x${1}y", "foo", ".*")`
diff --git a/lib/promrelabel/config_test.go b/lib/promrelabel/config_test.go
index 206c34519..3a1ae1477 100644
--- a/lib/promrelabel/config_test.go
+++ b/lib/promrelabel/config_test.go
@@ -11,8 +11,8 @@ func TestLoadRelabelConfigsSuccess(t *testing.T) {
 	if err != nil {
 		t.Fatalf("cannot load relabel configs from %q: %s", path, err)
 	}
-	if len(prcs) != 7 {
-		t.Fatalf("unexpected number of relabel configs loaded from %q; got %d; want %d", path, len(prcs), 7)
+	if len(prcs) != 9 {
+		t.Fatalf("unexpected number of relabel configs loaded from %q; got %d; want %d", path, len(prcs), 9)
 	}
 }
 
@@ -117,6 +117,36 @@ func TestParseRelabelConfigsFailure(t *testing.T) {
 			},
 		})
 	})
+	t.Run("keep_if_equal-missing-source-labels", func(t *testing.T) {
+		f([]RelabelConfig{
+			{
+				Action: "keep_if_equal",
+			},
+		})
+	})
+	t.Run("keep_if_equal-single-source-label", func(t *testing.T) {
+		f([]RelabelConfig{
+			{
+				Action:       "keep_if_equal",
+				SourceLabels: []string{"foo"},
+			},
+		})
+	})
+	t.Run("drop_if_equal-missing-source-labels", func(t *testing.T) {
+		f([]RelabelConfig{
+			{
+				Action: "drop_if_equal",
+			},
+		})
+	})
+	t.Run("drop_if_equal-single-source-label", func(t *testing.T) {
+		f([]RelabelConfig{
+			{
+				Action:       "drop_if_equal",
+				SourceLabels: []string{"foo"},
+			},
+		})
+	})
 	t.Run("drop-missing-source-labels", func(t *testing.T) {
 		f([]RelabelConfig{
 			{
diff --git a/lib/promrelabel/relabel_test.go b/lib/promrelabel/relabel_test.go
index 2cbfbee8b..f26a51372 100644
--- a/lib/promrelabel/relabel_test.go
+++ b/lib/promrelabel/relabel_test.go
@@ -336,6 +336,92 @@ func TestApplyRelabelConfigs(t *testing.T) {
 			},
 		})
 	})
+	t.Run("keep_if_equal-miss", func(t *testing.T) {
+		f([]ParsedRelabelConfig{
+			{
+				Action:       "keep_if_equal",
+				SourceLabels: []string{"foo", "bar"},
+			},
+		}, nil, true, nil)
+		f([]ParsedRelabelConfig{
+			{
+				Action:       "keep_if_equal",
+				SourceLabels: []string{"xxx", "bar"},
+			},
+		}, []prompbmarshal.Label{
+			{
+				Name:  "xxx",
+				Value: "yyy",
+			},
+		}, true, []prompbmarshal.Label{})
+	})
+	t.Run("keep_if_equal-hit", func(t *testing.T) {
+		f([]ParsedRelabelConfig{
+			{
+				Action:       "keep_if_equal",
+				SourceLabels: []string{"xxx", "bar"},
+			},
+		}, []prompbmarshal.Label{
+			{
+				Name:  "xxx",
+				Value: "yyy",
+			},
+			{
+				Name:  "bar",
+				Value: "yyy",
+			},
+		}, true, []prompbmarshal.Label{
+			{
+				Name:  "bar",
+				Value: "yyy",
+			},
+			{
+				Name:  "xxx",
+				Value: "yyy",
+			},
+		})
+	})
+	t.Run("drop_if_equal-miss", func(t *testing.T) {
+		f([]ParsedRelabelConfig{
+			{
+				Action:       "drop_if_equal",
+				SourceLabels: []string{"foo", "bar"},
+			},
+		}, nil, true, nil)
+		f([]ParsedRelabelConfig{
+			{
+				Action:       "drop_if_equal",
+				SourceLabels: []string{"xxx", "bar"},
+			},
+		}, []prompbmarshal.Label{
+			{
+				Name:  "xxx",
+				Value: "yyy",
+			},
+		}, true, []prompbmarshal.Label{
+			{
+				Name:  "xxx",
+				Value: "yyy",
+			},
+		})
+	})
+	t.Run("drop_if_equal-hit", func(t *testing.T) {
+		f([]ParsedRelabelConfig{
+			{
+				Action:       "drop_if_equal",
+				SourceLabels: []string{"xxx", "bar"},
+			},
+		}, []prompbmarshal.Label{
+			{
+				Name:  "xxx",
+				Value: "yyy",
+			},
+			{
+				Name:  "bar",
+				Value: "yyy",
+			},
+		}, true, []prompbmarshal.Label{})
+	})
 	t.Run("keep-miss", func(t *testing.T) {
 		f([]ParsedRelabelConfig{
 			{
diff --git a/lib/promrelabel/testdata/relabel_configs_valid.yml b/lib/promrelabel/testdata/relabel_configs_valid.yml
index 7ebe8ef3a..5b4a8828e 100644
--- a/lib/promrelabel/testdata/relabel_configs_valid.yml
+++ b/lib/promrelabel/testdata/relabel_configs_valid.yml
@@ -18,3 +18,7 @@
 - action: labelmap_all
   regex: "\\."
   replacement: ":"
+- action: keep_if_equal
+  source_labels: [foo, bar]
+- action: drop_if_equal
+  source_labels: [foo, bar]
diff --git a/lib/promscrape/config_test.go b/lib/promscrape/config_test.go
index 02560cae2..fafdc17dc 100644
--- a/lib/promscrape/config_test.go
+++ b/lib/promscrape/config_test.go
@@ -451,7 +451,8 @@ scrape_configs:
 					Value: "rty",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 		{
 			ScrapeURL:       "http://host2:80/abc/de",
@@ -489,7 +490,8 @@ scrape_configs:
 					Value: "rty",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 		{
 			ScrapeURL:       "http://localhost:9090/abc/de",
@@ -527,7 +529,8 @@ scrape_configs:
 					Value: "test",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 	})
 }
@@ -579,7 +582,8 @@ scrape_configs:
 					Value: "foo",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 	})
 	f(`
@@ -628,7 +632,8 @@ scrape_configs:
 					Value: "xxx",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 	})
 	f(`
@@ -700,6 +705,7 @@ scrape_configs:
 			AuthConfig: &promauth.Config{
 				Authorization: "Bearer xyz",
 			},
+			jobNameOriginal: "foo",
 		},
 		{
 			ScrapeURL:       "https://aaa:443/foo/bar?p=x%26y&p=%3D",
@@ -740,6 +746,7 @@ scrape_configs:
 			AuthConfig: &promauth.Config{
 				Authorization: "Bearer xyz",
 			},
+			jobNameOriginal: "foo",
 		},
 		{
 			ScrapeURL:       "http://1.2.3.4:80/metrics",
@@ -774,6 +781,7 @@ scrape_configs:
 				TLSServerName:         "foobar",
 				TLSInsecureSkipVerify: true,
 			},
+			jobNameOriginal: "qwer",
 		},
 	})
 	f(`
@@ -846,7 +854,8 @@ scrape_configs:
 					Value: "http://foo.bar:1234/metrics",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 	})
 	f(`
@@ -907,7 +916,8 @@ scrape_configs:
 					Value: "https",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 	})
 	f(`
@@ -949,7 +959,8 @@ scrape_configs:
 					Value: "3",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "foo",
 		},
 	})
 
@@ -997,6 +1008,7 @@ scrape_configs:
 			},
 			AuthConfig:           &promauth.Config{},
 			MetricRelabelConfigs: prcs,
+			jobNameOriginal:      "foo",
 		},
 	})
 	f(`
@@ -1037,6 +1049,7 @@ scrape_configs:
 			AuthConfig: &promauth.Config{
 				Authorization: "Basic eHl6OnNlY3JldC1wYXNz",
 			},
+			jobNameOriginal: "foo",
 		},
 	})
 	f(`
@@ -1075,6 +1088,7 @@ scrape_configs:
 			AuthConfig: &promauth.Config{
 				Authorization: "Bearer secret-pass",
 			},
+			jobNameOriginal: "foo",
 		},
 	})
 	snakeoilCert, err := tls.LoadX509KeyPair("testdata/ssl-cert-snakeoil.pem", "testdata/ssl-cert-snakeoil.key")
@@ -1119,6 +1133,7 @@ scrape_configs:
 			AuthConfig: &promauth.Config{
 				TLSCertificate: &snakeoilCert,
 			},
+			jobNameOriginal: "foo",
 		},
 	})
 	f(`
@@ -1179,7 +1194,8 @@ scrape_configs:
 					Value: "qwe",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "aaa",
 		},
 	})
 	f(`
@@ -1233,7 +1249,8 @@ scrape_configs:
 					Value: "snmp",
 				},
 			},
-			AuthConfig: &promauth.Config{},
+			AuthConfig:      &promauth.Config{},
+			jobNameOriginal: "snmp",
 		},
 	})
 }