Repository: VictoriaMetrics/VictoriaMetrics
Pull request #598 (2020-07-02 04:08:00)

`vmselect` support current query list and the ability to stop a specific query manually for single node version
same PR as #575  but for master branch, will be updated follow the review of #575 

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598?src=pr&el=h1) Report
> Merging [#598](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598?src=pr&el=desc) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/8da3f773ae76dee897c8f66973f6207a4c1c03c7&el=desc) will **decrease** coverage by `0.15%`.
> The diff coverage is `55.88%`.

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

```diff
@@            Coverage Diff             @@
##           master     #598      +/-   ##
==========================================
- Coverage   68.08%   67.93%   -0.16%     
==========================================
  Files         154      154              
  Lines       21257    21321      +64     
==========================================
+ Hits        14473    14484      +11     
- Misses       5622     5680      +58     
+ Partials     1162     1157       -5     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [app/vmselect/promql/exec.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9leGVjLmdv) | `67.55% <55.88%> (-7.45%)` | :arrow_down: |
| [lib/mergeset/block\_stream\_reader.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L2Jsb2NrX3N0cmVhbV9yZWFkZXIuZ28=) | `50.96% <0.00%> (-11.62%)` | :arrow_down: |
| [lib/storage/merge.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvbWVyZ2UuZ28=) | `82.64% <0.00%> (-1.66%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `67.73% <0.00%> (-1.54%)` | :arrow_down: |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `75.86% <0.00%> (-0.28%)` | :arrow_down: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `71.78% <0.00%> (+0.23%)` | :arrow_up: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `92.09% <0.00%> (+2.74%)` | :arrow_up: |
| [lib/fasttime/fasttime.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598/diff?src=pr&el=tree#diff-bGliL2Zhc3R0aW1lL2Zhc3R0aW1lLmdv) | `76.92% <0.00%> (+23.07%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598?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/598?src=pr&el=footer). Last update [8da3f77...0e9b5b9](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/598?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

syepes: It would also be interesting to have the Top N or last X queries appended locality to a file on the server.
This would enable the tracking/auditing of the executed queries.

Patch:
diff --git a/README.md b/README.md
index 99bd79d9e..1df438249 100644
--- a/README.md
+++ b/README.md
@@ -79,6 +79,7 @@ See [features available for enterprise customers](https://github.com/VictoriaMet
   * [HTTP OpenTSDB /api/put requests](#sending-opentsdb-data-via-http-apiput-requests) if `-opentsdbHTTPListenAddr` is set.
   * [/api/v1/import](#how-to-import-time-series-data).
   * [Arbitrary CSV data](#how-to-import-csv-data).
+* Supports metrics' relabeling. See [these docs](#relabeling) for details.
 * Ideally works with big amounts of time series data from Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
 * Has open source [cluster version](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/cluster).
 * See also technical [Articles about VictoriaMetrics](https://github.com/VictoriaMetrics/VictoriaMetrics/wiki/Articles).
@@ -111,6 +112,7 @@ See [features available for enterprise customers](https://github.com/VictoriaMet
 * [How to delete time series](#how-to-delete-time-series)
 * [How to export time series](#how-to-export-time-series)
 * [How to import time series data](#how-to-import-time-series-data)
+* [Relabeling](#relabeling)
 * [Federation](#federation)
 * [Capacity planning](#capacity-planning)
 * [High availability](#high-availability)
@@ -650,7 +652,7 @@ The delete API is intended mainly for the following cases:
 It isn't recommended using delete API for the following cases, since it brings non-zero overhead:
 
 * Regular cleanups for unneeded data. Just prevent writing unneeded data into VictoriaMetrics.
-  This can be done with relabeling in [vmagent](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/app/vmagent/README.md).
+  This can be done with [relabeling](#relabeling).
   See [this article](https://www.robustperception.io/relabelling-can-discard-targets-timeseries-and-alerts) for details.
 * Reducing disk space usage by deleting unneeded time series. This doesn't work as expected, since the deleted
   time series occupy disk space until the next merge operation, which can never occur when deleting too old data.
@@ -724,6 +726,22 @@ Note that it could be required to flush response cache after importing historica
 Each request to `/api/v1/import` can load up to a single vCPU core on VictoriaMetrics. Import speed can be improved by splitting the original file into smaller parts
 and importing them concurrently. Note that the original file must be split on newlines.
 
+
+### Relabeling
+
+VictoriaMetrics supports Prometheus-compatible relabeling for all the ingested metrics if `-relabelConfig` command-line flag points
+to a file containing a list of [relabel_config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config) entries.
+
+Additionally VictoriaMetrics provides the following extra actions for relabeling rules:
+
+* `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.
+
+See also [relabeling in vmagent](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/app/vmagent/README.md#relabeling).
+
+
 ### Federation
 
 VictoriaMetrics exports [Prometheus-compatible federation data](https://prometheus.io/docs/prometheus/latest/federation/)
@@ -769,6 +787,8 @@ The required resources for query path:
   The higher number of scanned time series and lower `step` argument results in the higher RAM usage.
 
 * CPU cores: a CPU core per 30 millions of scanned data points per second.
+  This means that heavy queries that touch big number of time series (over 10K) and/or big number data points (over 100M)
+  usually require more CPU resources than tiny queries that touch a few time series with small number of data points.
 
 * Network usage: depends on the frequency and the type of incoming requests. Typical Grafana dashboards usually
   require negligible network bandwidth.
@@ -957,7 +977,7 @@ The most interesting metrics are:
   of tweaking these flag values arises.
 
 * It is recommended upgrading to the latest available release from [this page](https://github.com/VictoriaMetrics/VictoriaMetrics/releases),
-  since the issue could be already fixed there.
+  since the encountered issue could be already fixed there.
 
 * If VictoriaMetrics works slowly and eats more than a CPU core per 100K ingested data points per second,
   then it is likely you have too many active time series for the current amount of RAM.
diff --git a/app/victoria-metrics/self_scraper.go b/app/victoria-metrics/self_scraper.go
index 5d42d20a6..ed12efeb4 100644
--- a/app/victoria-metrics/self_scraper.go
+++ b/app/victoria-metrics/self_scraper.go
@@ -85,7 +85,6 @@ func selfScraper(scrapeInterval time.Duration) {
 			mr.Timestamp = currentTimestamp
 			mr.Value = r.Value
 		}
-		logger.Infof("writing %d rows at timestamp %d", len(mrs), currentTimestamp)
 		vmstorage.AddRows(mrs)
 	}
 }
diff --git a/app/vmagent/README.md b/app/vmagent/README.md
index 48370d627..ea2801ba5 100644
--- a/app/vmagent/README.md
+++ b/app/vmagent/README.md
@@ -147,14 +147,20 @@ The following scrape types in [scrape_config](https://prometheus.io/docs/prometh
 * `dns_sd_configs` - for scraping targets discovered from DNS records (SRV, A and AAAA).
   See [dns_sd_config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#dns_sd_config) for details.
 
+File feature requests at [our issue tracker](https://github.com/VictoriaMetrics/VictoriaMetrics/issues) if you need other service discovery mechanisms to be supported by `vmagent`.
+
+`vmagent` also support the following additional options in `scrape_config` section:
+
+* `disable_compression: true` - for disabling response compression on a per-job basis. By default `vmagent` requests compressed responses from scrape targets
+  in order to save network bandwidth.
+* `disable_keepalive: true` - for disabling [HTTP keep-alive connections](https://en.wikipedia.org/wiki/HTTP_persistent_connection) on a per-job basis.
+  By default `vmagent` uses keep-alive connections to scrape targets in order to reduce overhead on connection re-establishing.
+
 Note that `vmagent` doesn't support `refresh_interval` option these scrape configs. Use the corresponding `-promscrape.*CheckInterval`
 command-line flag instead. For example, `-promscrape.consulSDCheckInterval=60s` sets `refresh_interval` for all the `consul_sd_configs`
 entries to 60s. Run `vmagent -help` in order to see default values for `-promscrape.*CheckInterval` flags.
 
 
-File feature requests at [our issue tracker](https://github.com/VictoriaMetrics/VictoriaMetrics/issues) if you need other service discovery mechanisms to be supported by `vmagent`.
-
-
 ### Adding labels to metrics
 
 Labels can be added to metrics via the following mechanisms:
diff --git a/app/vmagent/remotewrite/relabel.go b/app/vmagent/remotewrite/relabel.go
index 7b739a71c..6adede560 100644
--- a/app/vmagent/remotewrite/relabel.go
+++ b/app/vmagent/remotewrite/relabel.go
@@ -16,7 +16,7 @@ var (
 	unparsedLabelsGlobal = flagutil.NewArray("remoteWrite.label", "Optional label in the form 'name=value' to add to all the metrics before sending them to -remoteWrite.url. "+
 		"Pass multiple -remoteWrite.label flags in order to add multiple flags to metrics before sending them to remote storage")
 	relabelConfigPathGlobal = flag.String("remoteWrite.relabelConfig", "", "Optional path to file with relabel_config entries. These entries are applied to all the metrics "+
-		"before sending them to -remoteWrite.url. See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config for details")
+		"before sending them to -remoteWrite.url. See https://victoriametrics.github.io/vmagent.html#relabeling for details")
 	relabelConfigPaths = flagutil.NewArray("remoteWrite.urlRelabelConfig", "Optional path to relabel config for the corresponding -remoteWrite.url")
 )
 
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index ec65ea508..cccd0f460 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -144,7 +144,6 @@ var (
 	alertsSent       = metrics.NewCounter(`vmalert_alerts_sent_total`)
 	alertsSendErrors = metrics.NewCounter(`vmalert_alerts_send_errors_total`)
 
-	remoteWriteSent   = metrics.NewCounter(`vmalert_remotewrite_sent_total`)
 	remoteWriteErrors = metrics.NewCounter(`vmalert_remotewrite_errors_total`)
 )
 
@@ -255,7 +254,6 @@ func (e *executor) exec(ctx context.Context, rule Rule, returnSeries bool, inter
 	}
 
 	if len(tss) > 0 && e.rw != nil {
-		remoteWriteSent.Add(len(tss))
 		for _, ts := range tss {
 			if err := e.rw.Push(ts); err != nil {
 				remoteWriteErrors.Inc()
diff --git a/app/vmalert/remotewrite/remotewrite.go b/app/vmalert/remotewrite/remotewrite.go
index acb919ceb..caff50a5c 100644
--- a/app/vmalert/remotewrite/remotewrite.go
+++ b/app/vmalert/remotewrite/remotewrite.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/metrics"
 	"github.com/golang/snappy"
 )
 
@@ -61,7 +62,7 @@ const (
 	defaultConcurrency   = 4
 	defaultMaxBatchSize  = 1e3
 	defaultMaxQueueSize  = 1e5
-	defaultFlushInterval = time.Second
+	defaultFlushInterval = 5 * time.Second
 	defaultWriteTimeout  = 30 * time.Second
 )
 
@@ -85,6 +86,9 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 	if cfg.WriteTimeout == 0 {
 		cfg.WriteTimeout = defaultWriteTimeout
 	}
+	if cfg.Transport == nil {
+		cfg.Transport = http.DefaultTransport.(*http.Transport).Clone()
+	}
 	c := &Client{
 		c: &http.Client{
 			Timeout:   cfg.WriteTimeout,
@@ -95,6 +99,7 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 		baPass:        cfg.BasicAuthPass,
 		flushInterval: cfg.FlushInterval,
 		maxBatchSize:  cfg.MaxBatchSize,
+		maxQueueSize:  cfg.MaxQueueSize,
 		doneCh:        make(chan struct{}),
 		input:         make(chan prompbmarshal.TimeSeries, cfg.MaxQueueSize),
 	}
@@ -137,14 +142,11 @@ func (c *Client) Close() error {
 
 func (c *Client) run(ctx context.Context) {
 	ticker := time.NewTicker(c.flushInterval)
-	wr := prompbmarshal.WriteRequest{}
+	wr := &prompbmarshal.WriteRequest{}
 	shutdown := func() {
 		for ts := range c.input {
 			wr.Timeseries = append(wr.Timeseries, ts)
 		}
-		if len(wr.Timeseries) < 1 {
-			return
-		}
 		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
 		c.flush(lastCtx, wr)
 		cancel()
@@ -163,44 +165,82 @@ func (c *Client) run(ctx context.Context) {
 				return
 			case <-ticker.C:
 				c.flush(ctx, wr)
-				wr = prompbmarshal.WriteRequest{}
-			case ts := <-c.input:
+			case ts, ok := <-c.input:
+				if !ok {
+					continue
+				}
 				wr.Timeseries = append(wr.Timeseries, ts)
 				if len(wr.Timeseries) >= c.maxBatchSize {
 					c.flush(ctx, wr)
-					wr = prompbmarshal.WriteRequest{}
 				}
 			}
 		}
 	}()
 }
 
-func (c *Client) flush(ctx context.Context, wr prompbmarshal.WriteRequest) {
+var (
+	sentRows     = metrics.NewCounter(`vmalert_remotewrite_sent_rows_total`)
+	sentBytes    = metrics.NewCounter(`vmalert_remotewrite_sent_bytes_total`)
+	droppedRows  = metrics.NewCounter(`vmalert_remotewrite_dropped_rows_total`)
+	droppedBytes = metrics.NewCounter(`vmalert_remotewrite_dropped_bytes_total`)
+)
+
+// flush is a blocking function that marshals WriteRequest and sends
+// it to remote write endpoint. Flush performs limited amount of retries
+// if request fails.
+func (c *Client) flush(ctx context.Context, wr *prompbmarshal.WriteRequest) {
 	if len(wr.Timeseries) < 1 {
 		return
 	}
+	defer prompbmarshal.ResetWriteRequest(wr)
+
 	data, err := wr.Marshal()
 	if err != nil {
 		logger.Errorf("failed to marshal WriteRequest: %s", err)
 		return
 	}
-	req, err := http.NewRequest("POST", c.addr, bytes.NewReader(snappy.Encode(nil, data)))
+
+	const attempts = 5
+	b := snappy.Encode(nil, data)
+	for i := 0; i < attempts; i++ {
+		err := c.send(ctx, b)
+		if err == nil {
+			sentRows.Add(len(wr.Timeseries))
+			sentBytes.Add(len(b))
+			return
+		}
+
+		logger.Errorf("attempt %d to send request failed: %s", i+1, err)
+		// sleeping to avoid remote db hammering
+		time.Sleep(time.Second)
+		continue
+	}
+
+	droppedRows.Add(len(wr.Timeseries))
+	droppedBytes.Add(len(b))
+	logger.Errorf("all %d attempts to send request failed - dropping %d timeseries",
+		attempts, len(wr.Timeseries))
+}
+
+func (c *Client) send(ctx context.Context, data []byte) error {
+	r := bytes.NewReader(data)
+	req, err := http.NewRequest("POST", c.addr, r)
 	if err != nil {
-		logger.Errorf("failed to create new HTTP request: %s", err)
-		return
+		return fmt.Errorf("failed to create new HTTP request: %s", err)
 	}
 	if c.baPass != "" {
 		req.SetBasicAuth(c.baUser, c.baPass)
 	}
 	resp, err := c.c.Do(req.WithContext(ctx))
 	if err != nil {
-		logger.Errorf("error getting response from %s:%s", req.URL, err)
-		return
+		return fmt.Errorf("error while sending request to %s: %s; Data len %d(%d)",
+			req.URL, err, len(data), r.Size())
 	}
 	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode != http.StatusNoContent {
 		body, _ := ioutil.ReadAll(resp.Body)
-		logger.Errorf("unexpected response code %d for %s. Response body %s", resp.StatusCode, req.URL, body)
-		return
+		return fmt.Errorf("unexpected response code %d for %s. Response body %q",
+			resp.StatusCode, req.URL, body)
 	}
+	return nil
 }
diff --git a/app/vminsert/common/insert_ctx.go b/app/vminsert/common/insert_ctx.go
index a81b060fe..3eda96d34 100644
--- a/app/vminsert/common/insert_ctx.go
+++ b/app/vminsert/common/insert_ctx.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net/http"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -17,11 +18,14 @@ type InsertCtx struct {
 
 	mrs            []storage.MetricRow
 	metricNamesBuf []byte
+
+	relabelCtx relabel.Ctx
 }
 
 // Reset resets ctx for future fill with rowsLen rows.
 func (ctx *InsertCtx) Reset(rowsLen int) {
-	for _, label := range ctx.Labels {
+	for i := range ctx.Labels {
+		label := &ctx.Labels[i]
 		label.Name = nil
 		label.Value = nil
 	}
@@ -37,6 +41,8 @@ func (ctx *InsertCtx) Reset(rowsLen int) {
 	}
 	ctx.mrs = ctx.mrs[:0]
 	ctx.metricNamesBuf = ctx.metricNamesBuf[:0]
+
+	ctx.relabelCtx.Reset()
 }
 
 func (ctx *InsertCtx) marshalMetricNameRaw(prefix []byte, labels []prompb.Label) []byte {
@@ -118,6 +124,11 @@ func (ctx *InsertCtx) AddLabel(name, value string) {
 	ctx.Labels = labels
 }
 
+// ApplyRelabeling applies relabeling to ic.Labels.
+func (ctx *InsertCtx) ApplyRelabeling() {
+	ctx.Labels = ctx.relabelCtx.ApplyRelabeling(ctx.Labels)
+}
+
 // FlushBufs flushes buffered rows to the underlying storage.
 func (ctx *InsertCtx) FlushBufs() error {
 	if err := vmstorage.AddRows(ctx.mrs); err != nil {
diff --git a/app/vminsert/csvimport/request_handler.go b/app/vminsert/csvimport/request_handler.go
index 4d2644175..3d72edf61 100644
--- a/app/vminsert/csvimport/request_handler.go
+++ b/app/vminsert/csvimport/request_handler.go
@@ -36,6 +36,11 @@ func insertRows(rows []parser.Row) error {
 			tag := &r.Tags[j]
 			ctx.AddLabel(tag.Key, tag.Value)
 		}
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
 		ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value)
 	}
 	rowsInserted.Add(len(rows))
diff --git a/app/vminsert/graphite/request_handler.go b/app/vminsert/graphite/request_handler.go
index fe1690e3f..a6f58f1f6 100644
--- a/app/vminsert/graphite/request_handler.go
+++ b/app/vminsert/graphite/request_handler.go
@@ -36,6 +36,11 @@ func insertRows(rows []parser.Row) error {
 			tag := &r.Tags[j]
 			ctx.AddLabel(tag.Key, tag.Value)
 		}
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
 		ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value)
 	}
 	rowsInserted.Add(len(rows))
diff --git a/app/vminsert/influx/request_handler.go b/app/vminsert/influx/request_handler.go
index cb25d1205..1395fb8b4 100644
--- a/app/vminsert/influx/request_handler.go
+++ b/app/vminsert/influx/request_handler.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
@@ -59,6 +60,7 @@ func insertRows(db string, rows []parser.Row) error {
 	ic := &ctx.Common
 	ic.Reset(rowsLen)
 	rowsTotal := 0
+	hasRelabeling := relabel.HasRelabeling()
 	for i := range rows {
 		r := &rows[i]
 		ic.Labels = ic.Labels[:0]
@@ -73,22 +75,35 @@ func insertRows(db string, rows []parser.Row) error {
 		if len(db) > 0 && !hasDBLabel {
 			ic.AddLabel("db", db)
 		}
-		ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf[:0], ic.Labels)
 		ctx.metricGroupBuf = append(ctx.metricGroupBuf[:0], r.Measurement...)
 		skipFieldKey := len(r.Fields) == 1 && *skipSingleField
 		if len(ctx.metricGroupBuf) > 0 && !skipFieldKey {
 			ctx.metricGroupBuf = append(ctx.metricGroupBuf, *measurementFieldSeparator...)
 		}
 		metricGroupPrefixLen := len(ctx.metricGroupBuf)
+		ctx.metricNameBuf = ctx.metricNameBuf[:0]
+		if !hasRelabeling {
+			ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf, ic.Labels)
+		}
+		labelsLen := len(ic.Labels)
 		for j := range r.Fields {
 			f := &r.Fields[j]
 			if !skipFieldKey {
 				ctx.metricGroupBuf = append(ctx.metricGroupBuf[:metricGroupPrefixLen], f.Key...)
 			}
 			metricGroup := bytesutil.ToUnsafeString(ctx.metricGroupBuf)
-			ic.Labels = ic.Labels[:0]
+			ic.Labels = ic.Labels[:labelsLen]
 			ic.AddLabel("", metricGroup)
-			ic.WriteDataPoint(ctx.metricNameBuf, ic.Labels[:1], r.Timestamp, f.Value)
+			ic.ApplyRelabeling() // this must be called even if !hasRelabeling in order to remove labels with empty values
+			if len(ic.Labels) == 0 {
+				// Skip metric without labels.
+				continue
+			}
+			labels := ic.Labels
+			if !hasRelabeling {
+				labels = labels[labelsLen : labelsLen+1]
+			}
+			ic.WriteDataPoint(ctx.metricNameBuf, labels, r.Timestamp, f.Value)
 		}
 		rowsTotal += len(r.Fields)
 	}
diff --git a/app/vminsert/main.go b/app/vminsert/main.go
index 34cf9804c..bcbf7ec05 100644
--- a/app/vminsert/main.go
+++ b/app/vminsert/main.go
@@ -14,6 +14,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/prompush"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/promremotewrite"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	graphiteserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/graphite"
@@ -46,6 +47,7 @@ var (
 
 // Init initializes vminsert.
 func Init() {
+	relabel.Init()
 	storage.SetMaxLabelsPerTimeseries(*maxLabelsPerTimeseries)
 
 	writeconcurrencylimiter.Init()
diff --git a/app/vminsert/opentsdb/request_handler.go b/app/vminsert/opentsdb/request_handler.go
index d18f1d03f..4a61e09b6 100644
--- a/app/vminsert/opentsdb/request_handler.go
+++ b/app/vminsert/opentsdb/request_handler.go
@@ -36,6 +36,11 @@ func insertRows(rows []parser.Row) error {
 			tag := &r.Tags[j]
 			ctx.AddLabel(tag.Key, tag.Value)
 		}
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
 		ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value)
 	}
 	rowsInserted.Add(len(rows))
diff --git a/app/vminsert/opentsdbhttp/request_handler.go b/app/vminsert/opentsdbhttp/request_handler.go
index 0d7d5bd75..e70d1805e 100644
--- a/app/vminsert/opentsdbhttp/request_handler.go
+++ b/app/vminsert/opentsdbhttp/request_handler.go
@@ -42,6 +42,11 @@ func insertRows(rows []parser.Row) error {
 			tag := &r.Tags[j]
 			ctx.AddLabel(tag.Key, tag.Value)
 		}
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
 		ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value)
 	}
 	rowsInserted.Add(len(rows))
diff --git a/app/vminsert/prompush/push.go b/app/vminsert/prompush/push.go
index b4734b7cb..15790be1e 100644
--- a/app/vminsert/prompush/push.go
+++ b/app/vminsert/prompush/push.go
@@ -1,13 +1,8 @@
 package prompush
 
 import (
-	"runtime"
-	"sync"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -21,8 +16,8 @@ const maxRowsPerBlock = 10000
 
 // Push pushes wr to storage.
 func Push(wr *prompbmarshal.WriteRequest) {
-	ctx := getPushCtx()
-	defer putPushCtx(ctx)
+	ctx := common.GetInsertCtx()
+	defer common.PutInsertCtx(ctx)
 
 	tss := wr.Timeseries
 	for len(tss) > 0 {
@@ -34,80 +29,39 @@ func Push(wr *prompbmarshal.WriteRequest) {
 		} else {
 			tss = nil
 		}
-		ctx.push(tssBlock)
+		push(ctx, tssBlock)
 	}
 }
 
-func (ctx *pushCtx) push(tss []prompbmarshal.TimeSeries) {
+func push(ctx *common.InsertCtx, tss []prompbmarshal.TimeSeries) {
 	rowsLen := 0
 	for i := range tss {
 		rowsLen += len(tss[i].Samples)
 	}
-	ic := &ctx.Common
-	ic.Reset(rowsLen)
+	ctx.Reset(rowsLen)
 	rowsTotal := 0
-	labels := ctx.labels[:0]
 	for i := range tss {
 		ts := &tss[i]
-		labels = labels[:0]
+		ctx.Labels = ctx.Labels[:0]
 		for j := range ts.Labels {
 			label := &ts.Labels[j]
-			labels = append(labels, prompb.Label{
-				Name:  bytesutil.ToUnsafeBytes(label.Name),
-				Value: bytesutil.ToUnsafeBytes(label.Value),
-			})
+			ctx.AddLabel(label.Name, label.Value)
+		}
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
 		}
 		var metricNameRaw []byte
 		for i := range ts.Samples {
 			r := &ts.Samples[i]
-			metricNameRaw = ic.WriteDataPointExt(metricNameRaw, labels, r.Timestamp, r.Value)
+			metricNameRaw = ctx.WriteDataPointExt(metricNameRaw, ctx.Labels, r.Timestamp, r.Value)
 		}
 		rowsTotal += len(ts.Samples)
 	}
-	ctx.labels = labels
 	rowsInserted.Add(rowsTotal)
 	rowsPerInsert.Update(float64(rowsTotal))
-	if err := ic.FlushBufs(); err != nil {
+	if err := ctx.FlushBufs(); err != nil {
 		logger.Errorf("cannot flush promscrape data to storage: %s", err)
 	}
 }
-
-type pushCtx struct {
-	Common common.InsertCtx
-	labels []prompb.Label
-}
-
-func (ctx *pushCtx) reset() {
-	ctx.Common.Reset(0)
-
-	for i := range ctx.labels {
-		label := &ctx.labels[i]
-		label.Name = nil
-		label.Value = nil
-	}
-	ctx.labels = ctx.labels[:0]
-}
-
-func getPushCtx() *pushCtx {
-	select {
-	case ctx := <-pushCtxPoolCh:
-		return ctx
-	default:
-		if v := pushCtxPool.Get(); v != nil {
-			return v.(*pushCtx)
-		}
-		return &pushCtx{}
-	}
-}
-
-func putPushCtx(ctx *pushCtx) {
-	ctx.reset()
-	select {
-	case pushCtxPoolCh <- ctx:
-	default:
-		pushCtxPool.Put(ctx)
-	}
-}
-
-var pushCtxPool sync.Pool
-var pushCtxPoolCh = make(chan *pushCtx, runtime.GOMAXPROCS(-1))
diff --git a/app/vminsert/promremotewrite/request_handler.go b/app/vminsert/promremotewrite/request_handler.go
index 7fab4bcfb..4f7b2e80e 100644
--- a/app/vminsert/promremotewrite/request_handler.go
+++ b/app/vminsert/promremotewrite/request_handler.go
@@ -34,10 +34,17 @@ func insertRows(timeseries []prompb.TimeSeries) error {
 	rowsTotal := 0
 	for i := range timeseries {
 		ts := &timeseries[i]
+		// Make a shallow copy of ts.Labels before calling ctx.ApplyRelabeling, since ctx.ApplyRelabeling may modify labels.
+		ctx.Labels = append(ctx.Labels[:0], ts.Labels...)
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
 		var metricNameRaw []byte
 		for i := range ts.Samples {
 			r := &ts.Samples[i]
-			metricNameRaw = ctx.WriteDataPointExt(metricNameRaw, ts.Labels, r.Timestamp, r.Value)
+			metricNameRaw = ctx.WriteDataPointExt(metricNameRaw, ctx.Labels, r.Timestamp, r.Value)
 		}
 		rowsTotal += len(ts.Samples)
 	}
diff --git a/app/vminsert/relabel/relabel.go b/app/vminsert/relabel/relabel.go
new file mode 100644
index 000000000..62ce3afe5
--- /dev/null
+++ b/app/vminsert/relabel/relabel.go
@@ -0,0 +1,141 @@
+package relabel
+
+import (
+	"flag"
+	"fmt"
+	"sync/atomic"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
+)
+
+var relabelConfig = flag.String("relabelConfig", "", "Optional path to a file with relabeling rules, which are applied to all the ingested metrics. "+
+	"See https://victoriametrics.github.io/#relabeling for details")
+
+// Init must be called after flag.Parse and before using the relabel package.
+func Init() {
+	prcs, err := loadRelabelConfig()
+	if err != nil {
+		logger.Fatalf("cannot load relabelConfig: %s", err)
+	}
+	prcsGlobal.Store(&prcs)
+	if len(*relabelConfig) == 0 {
+		return
+	}
+	sighupCh := procutil.NewSighupChan()
+	go func() {
+		for range sighupCh {
+			logger.Infof("received SIGHUP; reloading -relabelConfig=%q...", *relabelConfig)
+			prcs, err := loadRelabelConfig()
+			if err != nil {
+				logger.Errorf("cannot load the updated relabelConfig: %s; preserving the previous config", err)
+				continue
+			}
+			prcsGlobal.Store(&prcs)
+			logger.Infof("successfully reloaded -relabelConfig=%q", *relabelConfig)
+		}
+	}()
+}
+
+var prcsGlobal atomic.Value
+
+func loadRelabelConfig() ([]promrelabel.ParsedRelabelConfig, error) {
+	if len(*relabelConfig) == 0 {
+		return nil, nil
+	}
+	prcs, err := promrelabel.LoadRelabelConfigs(*relabelConfig)
+	if err != nil {
+		return nil, fmt.Errorf("error when reading -relabelConfig=%q: %w", *relabelConfig, err)
+	}
+	return prcs, nil
+}
+
+// HasRelabeling returns true if there is global relabeling.
+func HasRelabeling() bool {
+	prcs := prcsGlobal.Load().(*[]promrelabel.ParsedRelabelConfig)
+	return len(*prcs) > 0
+}
+
+// Ctx holds relabeling context.
+type Ctx struct {
+	// tmpLabels is used during ApplyRelabeling call.
+	tmpLabels []prompbmarshal.Label
+}
+
+// Reset resets ctx.
+func (ctx *Ctx) Reset() {
+	labels := ctx.tmpLabels
+	for i := range labels {
+		label := &labels[i]
+		label.Name = ""
+		label.Value = ""
+	}
+	ctx.tmpLabels = ctx.tmpLabels[:0]
+}
+
+// ApplyRelabeling applies relabeling to the given labels and returns the result.
+//
+// The returned labels are valid until the next call to ApplyRelabeling.
+func (ctx *Ctx) ApplyRelabeling(labels []prompb.Label) []prompb.Label {
+	// Remove labels with empty values, since such labels have no sense.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/600 .
+	hasEmptyValues := false
+	for _, label := range labels {
+		if len(label.Value) == 0 {
+			hasEmptyValues = true
+			break
+		}
+	}
+	if hasEmptyValues {
+		dst := labels[:0]
+		for _, label := range labels {
+			if len(label.Value) == 0 {
+				continue
+			}
+			dst = append(dst, label)
+		}
+		labels = dst
+	}
+
+	prcs := prcsGlobal.Load().(*[]promrelabel.ParsedRelabelConfig)
+	if len(*prcs) == 0 {
+		// There are no relabeling rules.
+		return labels
+	}
+	// Convert src to prompbmarshal.Label format suitable for relabeling.
+	tmpLabels := ctx.tmpLabels[:0]
+	for _, label := range labels {
+		name := bytesutil.ToUnsafeString(label.Name)
+		if len(name) == 0 {
+			name = "__name__"
+		}
+		value := bytesutil.ToUnsafeString(label.Value)
+		tmpLabels = append(tmpLabels, prompbmarshal.Label{
+			Name:  name,
+			Value: value,
+		})
+	}
+
+	// Apply relabeling
+	tmpLabels = promrelabel.ApplyRelabelConfigs(tmpLabels, 0, *prcs, true)
+	ctx.tmpLabels = tmpLabels
+
+	// Return back labels to the desired format.
+	dst := labels[:0]
+	for _, label := range tmpLabels {
+		name := bytesutil.ToUnsafeBytes(label.Name)
+		if label.Name == "__name__" {
+			name = nil
+		}
+		value := bytesutil.ToUnsafeBytes(label.Value)
+		dst = append(dst, prompb.Label{
+			Name:  name,
+			Value: value,
+		})
+	}
+	return dst
+}
diff --git a/app/vminsert/vmimport/request_handler.go b/app/vminsert/vmimport/request_handler.go
index ecaf20957..53171b4cc 100644
--- a/app/vminsert/vmimport/request_handler.go
+++ b/app/vminsert/vmimport/request_handler.go
@@ -44,6 +44,11 @@ func insertRows(rows []parser.Row) error {
 			tag := &r.Tags[j]
 			ic.AddLabelBytes(tag.Key, tag.Value)
 		}
+		ic.ApplyRelabeling()
+		if len(ic.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
 		ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf[:0], ic.Labels)
 		values := r.Values
 		timestamps := r.Timestamps
diff --git a/app/vmselect/main.go b/app/vmselect/main.go
index c59ea186c..da971bce9 100644
--- a/app/vmselect/main.go
+++ b/app/vmselect/main.go
@@ -1,6 +1,7 @@
 package vmselect
 
 import (
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
@@ -23,7 +24,8 @@ var (
 	maxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
 		"It shouldn't be high, since a single request can saturate all the CPU cores. See also -search.maxQueueDuration")
 	maxQueueDuration  = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests limit is reached")
-	resetCacheAuthKey = flag.String("search.resetCacheAuthKey", "", "Optional authKey for resetting rollup cache via /internal/resetRollupResultCache call")
+	resetCacheAuthKey = flag.String("search.resetCacheAuthKey", "", "Optional authKey for resetting rollup cache via /internal/resetRollupResultCache call. Will be replaced by internalAuthKey.")
+	internalAuthKey   = flag.String("internalAuthKey", "", "Optional authKey for operation calls with prefix /internal. Use value of search.resetCacheAuthKey fif not set.")
 )
 
 func getDefaultMaxConcurrentRequests() int {
@@ -66,6 +68,8 @@ var (
 	})
 )
 
+var internalHandlerPrefix = "/internal"
+
 // RequestHandler handles remote read API requests for Prometheus
 func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	startTime := time.Now()
@@ -96,15 +100,13 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	}
 
 	path := strings.Replace(r.URL.Path, "//", "/", -1)
-	if path == "/internal/resetRollupResultCache" {
-		if len(*resetCacheAuthKey) > 0 && r.FormValue("authKey") != *resetCacheAuthKey {
-			sendPrometheusError(w, r, fmt.Errorf("invalid authKey=%q for %q", r.FormValue("authKey"), path))
-			return true
-		}
-		promql.ResetRollupResultCache()
-		return true
+	if strings.HasPrefix(path, internalHandlerPrefix) {
+		return internalHandler(w, r)
 	}
+	return apiHandler(startTime, w, r, path)
+}
 
+func apiHandler(startTime time.Time, w http.ResponseWriter, r *http.Request, path string) bool {
 	if strings.HasPrefix(path, "/api/v1/label/") {
 		s := r.URL.Path[len("/api/v1/label/"):]
 		if strings.HasSuffix(s, "/values") {
@@ -236,6 +238,78 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	}
 }
 
+func internalHandler(w http.ResponseWriter, r *http.Request) bool {
+	path := r.URL.Path
+	p := path[len(internalHandlerPrefix):]
+	authKey := *internalAuthKey
+	if authKey == "" {
+		authKey = *resetCacheAuthKey
+	}
+	if authKey != "" && r.FormValue("authKey") != authKey {
+		sendPrometheusError(w, r, fmt.Errorf("invalid authKey=%q for %q", r.FormValue("authKey"), path))
+		return true
+	}
+
+	switch p {
+	case "/resetRollupResultCache":
+		promql.ResetRollupResultCache()
+		return true
+	case "/query/list":
+		queryListRequests.Inc()
+		w.Header().Set("Content-Type", "application/json")
+		data, err := json.Marshal(promql.GetAllRunningQueries())
+		if err != nil {
+			queryListErrors.Inc()
+			sendPrometheusError(w, r, err)
+			return true
+		}
+		fmt.Fprintf(w, `{"status":"success","data": %v}`, string(data))
+		return true
+	case "/query/info":
+		queryInfoRequests.Inc()
+		w.Header().Set("Content-Type", "application/json")
+		pid := strings.TrimSpace(r.URL.Query().Get("pid"))
+		if pid == "" {
+			queryInfoErrors.Inc()
+			sendPrometheusError(w, r, fmt.Errorf("pid not set"))
+			return true
+		}
+		info, err := promql.GetQueryInfo(pid)
+		if err != nil {
+			queryInfoErrors.Inc()
+			sendPrometheusError(w, r, err)
+			return true
+		}
+		data, err := json.Marshal(info)
+		if err != nil {
+			queryInfoErrors.Inc()
+			sendPrometheusError(w, r, err)
+			return true
+		}
+		fmt.Fprintf(w, `{"status":"success","data": %v}`, string(data))
+		return true
+	case "/query/kill":
+		queryKillRequests.Inc()
+		w.Header().Set("Content-Type", "application/json")
+		pid := strings.TrimSpace(r.URL.Query().Get("pid"))
+		if pid == "" {
+			queryKillErrors.Inc()
+			sendPrometheusError(w, r, fmt.Errorf("pid not set"))
+			return true
+		}
+		err := promql.CancelRunningQuery(pid)
+		if err != nil {
+			queryKillErrors.Inc()
+			sendPrometheusError(w, r, err)
+			return true
+		}
+		fmt.Fprintf(w, "%s", `{"status":"success","data":{}}`)
+		return true
+	default:
+		return false
+	}
+}
+
 func sendPrometheusError(w http.ResponseWriter, r *http.Request, err error) {
 	logger.Warnf("error in %q: %s", r.RequestURI, err)
 
@@ -286,4 +360,13 @@ var (
 	rulesRequests    = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/rules"}`)
 	alertsRequests   = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/alerts"}`)
 	metadataRequests = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/metadata"}`)
+
+	queryListRequests = metrics.NewCounter(`vm_http_requests_total{path="/-/{}/query/list"}`)
+	queryListErrors   = metrics.NewCounter(`vm_http_request_errors_total{path="/-/{}/query/list"}`)
+
+	queryKillRequests = metrics.NewCounter(`vm_http_requests_total{path="/-/{}/query/kill"}`)
+	queryKillErrors   = metrics.NewCounter(`vm_http_request_errors_total{path="/-/{}/query/kill"}`)
+
+	queryInfoRequests = metrics.NewCounter(`vm_http_requests_total{path="/-/{}/query/info"}`)
+	queryInfoErrors   = metrics.NewCounter(`vm_http_request_errors_total{path="/-/{}/query/info"}`)
 )
diff --git a/app/vmselect/prometheus/prometheus.go b/app/vmselect/prometheus/prometheus.go
index eb106f9b0..cf95f4099 100644
--- a/app/vmselect/prometheus/prometheus.go
+++ b/app/vmselect/prometheus/prometheus.go
@@ -791,8 +791,8 @@ func queryRangeHandler(w http.ResponseWriter, query string, start, end, step int
 		return fmt.Errorf("cannot execute query: %w", err)
 	}
 	queryOffset := getLatencyOffsetMilliseconds()
-	if ct-end < queryOffset {
-		result = adjustLastPoints(result)
+	if ct-queryOffset < end {
+		result = adjustLastPoints(result, ct-queryOffset, end)
 	}
 
 	// Remove NaN values as Prometheus does.
@@ -837,40 +837,24 @@ func removeNaNValuesInplace(tss []netstorage.Result) {
 
 var queryRangeDuration = metrics.NewSummary(`vm_request_duration_seconds{path="/api/v1/query_range"}`)
 
-// adjustLastPoints substitutes the last point values with the previous
-// point values, since the last points may contain garbage.
-func adjustLastPoints(tss []netstorage.Result) []netstorage.Result {
-	if len(tss) == 0 {
-		return nil
-	}
-
-	// Search for the last non-NaN value across all the timeseries.
-	lastNonNaNIdx := -1
+// adjustLastPoints substitutes the last point values on the time range (start..end]
+// with the previous point values, since these points may contain garbage.
+func adjustLastPoints(tss []netstorage.Result, start, end int64) []netstorage.Result {
 	for i := range tss {
-		values := tss[i].Values
-		j := len(values) - 1
-		for j >= 0 && math.IsNaN(values[j]) {
+		ts := &tss[i]
+		values := ts.Values
+		timestamps := ts.Timestamps
+		j := len(timestamps) - 1
+		for j >= 0 && timestamps[j] > start {
 			j--
 		}
-		if j > lastNonNaNIdx {
-			lastNonNaNIdx = j
+		j++
+		if j <= 0 {
+			continue
 		}
-	}
-	if lastNonNaNIdx == -1 {
-		// All timeseries contain only NaNs.
-		return nil
-	}
-
-	// Substitute the last two values starting from lastNonNaNIdx
-	// with the previous values for each timeseries.
-	for i := range tss {
-		values := tss[i].Values
-		for j := 0; j < 2; j++ {
-			idx := lastNonNaNIdx + j
-			if idx <= 0 || idx >= len(values) || math.IsNaN(values[idx-1]) {
-				continue
-			}
-			values[idx] = values[idx-1]
+		for j < len(timestamps) && timestamps[j] <= end {
+			values[j] = values[j-1]
+			j++
 		}
 	}
 	return tss
diff --git a/app/vmselect/promql/exec.go b/app/vmselect/promql/exec.go
index f42337477..4590790e8 100644
--- a/app/vmselect/promql/exec.go
+++ b/app/vmselect/promql/exec.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"math"
 	"sort"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -13,10 +14,96 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/VictoriaMetrics/metricsql"
+
+	"github.com/google/uuid"
 )
 
 var logSlowQueryDuration = flag.Duration("search.logSlowQueryDuration", 5*time.Second, "Log queries with execution time exceeding this value. Zero disables slow query logging")
 
+type query struct {
+	q       *string
+	ec      *EvalConfig
+	stopCh  chan error
+	startAt time.Time
+}
+
+type queriesMap struct {
+	mu sync.Mutex
+	m  map[string]query
+}
+
+func newQueriesMap() *queriesMap {
+	var qm queriesMap
+	qm.m = make(map[string]query)
+
+	return &qm
+}
+
+func (qm *queriesMap) Add(q query) string {
+	qm.mu.Lock()
+	c := uuid.New().String()
+	qm.m[c] = q
+	qm.mu.Unlock()
+
+	return c
+}
+
+func (qm *queriesMap) Delete(c string) {
+	qm.mu.Lock()
+	delete(qm.m, c)
+	qm.mu.Unlock()
+}
+
+var runningQueries = newQueriesMap()
+
+const truncateQueryLength = 16
+
+// GetAllRunningQueries get all the running queries' list
+func GetAllRunningQueries() map[string]map[string]string {
+	all := make(map[string]map[string]string)
+	runningQueries.mu.Lock()
+	for c, rq := range runningQueries.m {
+		m := make(map[string]string)
+		if len(*rq.q) > truncateQueryLength {
+			m["query"] = (*rq.q)[:truncateQueryLength] + "..."
+		} else {
+			m["query"] = *rq.q
+		}
+		m["cost"] = time.Since(rq.startAt).String()
+		all[c] = m
+	}
+	runningQueries.mu.Unlock()
+
+	return all
+}
+
+// GetQueryInfo get all the running queries' info
+func GetQueryInfo(c string) (map[string]string, error) {
+	if rq, ok := runningQueries.m[c]; ok {
+		m := make(map[string]string)
+		m["query"] = *rq.q
+		m["start"] = strconv.FormatInt(rq.ec.Start, 10)
+		m["end"] = strconv.FormatInt(rq.ec.End, 10)
+		m["step"] = strconv.FormatInt(rq.ec.Step, 10)
+		m["cost"] = time.Since(rq.startAt).String()
+
+		return m, nil
+	}
+	return nil, fmt.Errorf("query of qid {%v} is not running", c)
+}
+
+// CancelRunningQuery cancel the given query's execution
+func CancelRunningQuery(c string) error {
+	runningQueries.mu.Lock()
+	defer runningQueries.mu.Unlock()
+	if rq, ok := runningQueries.m[c]; ok {
+		rq.stopCh <- fmt.Errorf("cancel query manully")
+		return nil
+	}
+	return fmt.Errorf("query of qid {%v} is not running", c)
+
+}
+
 var slowQueries = metrics.NewCounter(`vm_slow_queries_total`)
 
 // Exec executes q for the given ec.
@@ -33,16 +120,40 @@ func Exec(ec *EvalConfig, q string, isFirstPointOnly bool) ([]netstorage.Result,
 		}()
 	}
 
+	stopCh := make(chan error, 1)
+	resultCh := make(chan []netstorage.Result)
+	c := runningQueries.Add(query{
+		q:       &q,
+		ec:      ec,
+		startAt: time.Now(),
+		stopCh:  stopCh,
+	})
+	defer runningQueries.Delete(c)
+
+	go exec(ec, q, isFirstPointOnly, stopCh, resultCh)
+
+	select {
+	case err := <-stopCh:
+		logger.Infof(err.Error())
+		return nil, err
+	case result := <-resultCh:
+		return result, nil
+	}
+}
+
+func exec(ec *EvalConfig, q string, isFirstPointOnly bool, stopCh chan error, resultCh chan []netstorage.Result) {
 	ec.validate()
 
 	e, err := parsePromQLWithCache(q)
 	if err != nil {
-		return nil, err
+		stopCh <- err
+		return
 	}
 
 	rv, err := evalExpr(ec, e)
 	if err != nil {
-		return nil, err
+		stopCh <- err
+		return
 	}
 
 	if isFirstPointOnly {
@@ -56,9 +167,10 @@ func Exec(ec *EvalConfig, q string, isFirstPointOnly bool) ([]netstorage.Result,
 	maySort := maySortResults(e, rv)
 	result, err := timeseriesToResult(rv, maySort)
 	if err != nil {
-		return nil, err
+		stopCh <- err
+		return
 	}
-	return result, err
+	resultCh <- result
 }
 
 func maySortResults(e metricsql.Expr, tss []*timeseries) bool {
diff --git a/app/vmselect/promql/transform.go b/app/vmselect/promql/transform.go
index 3a25d1dc6..e2843d3c1 100644
--- a/app/vmselect/promql/transform.go
+++ b/app/vmselect/promql/transform.go
@@ -72,6 +72,7 @@ var transformFuncs = map[string]transformFunc{
 	"":                   transformUnion, // empty func is a synonim to union
 	"keep_last_value":    transformKeepLastValue,
 	"keep_next_value":    transformKeepNextValue,
+	"interpolate":        transformInterpolate,
 	"start":              newTransformFuncZeroArgs(transformStart),
 	"end":                newTransformFuncZeroArgs(transformEnd),
 	"step":               newTransformFuncZeroArgs(transformStep),
@@ -764,6 +765,52 @@ func transformKeepNextValue(tfa *transformFuncArg) ([]*timeseries, error) {
 	return rvs, nil
 }
 
+func transformInterpolate(tfa *transformFuncArg) ([]*timeseries, error) {
+	args := tfa.args
+	if err := expectTransformArgsNum(args, 1); err != nil {
+		return nil, err
+	}
+	rvs := args[0]
+	for _, ts := range rvs {
+		values := ts.Values
+		if len(values) == 0 {
+			continue
+		}
+		prevValue := nan
+		var nextValue float64
+		for i := 0; i < len(values); i++ {
+			if !math.IsNaN(values[i]) {
+				continue
+			}
+			if i > 0 {
+				prevValue = values[i-1]
+			}
+			j := i + 1
+			for j < len(values) {
+				if !math.IsNaN(values[j]) {
+					break
+				}
+				j++
+			}
+			if j >= len(values) {
+				nextValue = prevValue
+			} else {
+				nextValue = values[j]
+			}
+			if math.IsNaN(prevValue) {
+				prevValue = nextValue
+			}
+			delta := (nextValue - prevValue) / float64(j-i+1)
+			for i < j {
+				prevValue += delta
+				values[i] = prevValue
+				i++
+			}
+		}
+	}
+	return rvs, nil
+}
+
 func newTransformFuncRunning(rf func(a, b float64, idx int) float64) transformFunc {
 	return func(tfa *transformFuncArg) ([]*timeseries, error) {
 		args := tfa.args
diff --git a/app/vmstorage/main.go b/app/vmstorage/main.go
index 6c7d7b11a..c400257c1 100644
--- a/app/vmstorage/main.go
+++ b/app/vmstorage/main.go
@@ -429,6 +429,10 @@ func registerStorageMetrics() {
 		return float64(m().AddRowsConcurrencyCurrent)
 	})
 
+	metrics.NewGauge(`vm_search_delays_total`, func() float64 {
+		return float64(m().SearchDelays)
+	})
+
 	metrics.NewGauge(`vm_slow_row_inserts_total`, func() float64 {
 		return float64(m().SlowRowInserts)
 	})
diff --git a/docs/MetricsQL.md b/docs/MetricsQL.md
index 127b75dea..0e905fc6b 100644
--- a/docs/MetricsQL.md
+++ b/docs/MetricsQL.md
@@ -80,6 +80,7 @@ This functionality can be tried at [an editable Grafana dashboard](http://play-g
   Use `limitk(1, q)` if you need retaining all the labels from `q`.
 - `keep_last_value(q)` - fills missing data (gaps) in `q` with the previous non-empty value.
 - `keep_next_value(q)` - fills missing data (gaps) in `q` with the next non-empty value.
+- `interpolate(q)` - fills missing data (gaps) in `q` with linearly interpolated values.
 - `distinct_over_time(m[d])` - returns distinct number of values for `m` data points over `d` duration.
 - `distinct(q)` - returns a time series with the number of unique values for each timestamp in `q`.
 - `sum2_over_time(m[d])` - returns sum of squares for all the `m` values over `d` duration.
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 99bd79d9e..1df438249 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -79,6 +79,7 @@ See [features available for enterprise customers](https://github.com/VictoriaMet
   * [HTTP OpenTSDB /api/put requests](#sending-opentsdb-data-via-http-apiput-requests) if `-opentsdbHTTPListenAddr` is set.
   * [/api/v1/import](#how-to-import-time-series-data).
   * [Arbitrary CSV data](#how-to-import-csv-data).
+* Supports metrics' relabeling. See [these docs](#relabeling) for details.
 * Ideally works with big amounts of time series data from Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
 * Has open source [cluster version](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/cluster).
 * See also technical [Articles about VictoriaMetrics](https://github.com/VictoriaMetrics/VictoriaMetrics/wiki/Articles).
@@ -111,6 +112,7 @@ See [features available for enterprise customers](https://github.com/VictoriaMet
 * [How to delete time series](#how-to-delete-time-series)
 * [How to export time series](#how-to-export-time-series)
 * [How to import time series data](#how-to-import-time-series-data)
+* [Relabeling](#relabeling)
 * [Federation](#federation)
 * [Capacity planning](#capacity-planning)
 * [High availability](#high-availability)
@@ -650,7 +652,7 @@ The delete API is intended mainly for the following cases:
 It isn't recommended using delete API for the following cases, since it brings non-zero overhead:
 
 * Regular cleanups for unneeded data. Just prevent writing unneeded data into VictoriaMetrics.
-  This can be done with relabeling in [vmagent](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/app/vmagent/README.md).
+  This can be done with [relabeling](#relabeling).
   See [this article](https://www.robustperception.io/relabelling-can-discard-targets-timeseries-and-alerts) for details.
 * Reducing disk space usage by deleting unneeded time series. This doesn't work as expected, since the deleted
   time series occupy disk space until the next merge operation, which can never occur when deleting too old data.
@@ -724,6 +726,22 @@ Note that it could be required to flush response cache after importing historica
 Each request to `/api/v1/import` can load up to a single vCPU core on VictoriaMetrics. Import speed can be improved by splitting the original file into smaller parts
 and importing them concurrently. Note that the original file must be split on newlines.
 
+
+### Relabeling
+
+VictoriaMetrics supports Prometheus-compatible relabeling for all the ingested metrics if `-relabelConfig` command-line flag points
+to a file containing a list of [relabel_config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config) entries.
+
+Additionally VictoriaMetrics provides the following extra actions for relabeling rules:
+
+* `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.
+
+See also [relabeling in vmagent](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/app/vmagent/README.md#relabeling).
+
+
 ### Federation
 
 VictoriaMetrics exports [Prometheus-compatible federation data](https://prometheus.io/docs/prometheus/latest/federation/)
@@ -769,6 +787,8 @@ The required resources for query path:
   The higher number of scanned time series and lower `step` argument results in the higher RAM usage.
 
 * CPU cores: a CPU core per 30 millions of scanned data points per second.
+  This means that heavy queries that touch big number of time series (over 10K) and/or big number data points (over 100M)
+  usually require more CPU resources than tiny queries that touch a few time series with small number of data points.
 
 * Network usage: depends on the frequency and the type of incoming requests. Typical Grafana dashboards usually
   require negligible network bandwidth.
@@ -957,7 +977,7 @@ The most interesting metrics are:
   of tweaking these flag values arises.
 
 * It is recommended upgrading to the latest available release from [this page](https://github.com/VictoriaMetrics/VictoriaMetrics/releases),
-  since the issue could be already fixed there.
+  since the encountered issue could be already fixed there.
 
 * If VictoriaMetrics works slowly and eats more than a CPU core per 100K ingested data points per second,
   then it is likely you have too many active time series for the current amount of RAM.
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 48370d627..ea2801ba5 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -147,14 +147,20 @@ The following scrape types in [scrape_config](https://prometheus.io/docs/prometh
 * `dns_sd_configs` - for scraping targets discovered from DNS records (SRV, A and AAAA).
   See [dns_sd_config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#dns_sd_config) for details.
 
+File feature requests at [our issue tracker](https://github.com/VictoriaMetrics/VictoriaMetrics/issues) if you need other service discovery mechanisms to be supported by `vmagent`.
+
+`vmagent` also support the following additional options in `scrape_config` section:
+
+* `disable_compression: true` - for disabling response compression on a per-job basis. By default `vmagent` requests compressed responses from scrape targets
+  in order to save network bandwidth.
+* `disable_keepalive: true` - for disabling [HTTP keep-alive connections](https://en.wikipedia.org/wiki/HTTP_persistent_connection) on a per-job basis.
+  By default `vmagent` uses keep-alive connections to scrape targets in order to reduce overhead on connection re-establishing.
+
 Note that `vmagent` doesn't support `refresh_interval` option these scrape configs. Use the corresponding `-promscrape.*CheckInterval`
 command-line flag instead. For example, `-promscrape.consulSDCheckInterval=60s` sets `refresh_interval` for all the `consul_sd_configs`
 entries to 60s. Run `vmagent -help` in order to see default values for `-promscrape.*CheckInterval` flags.
 
 
-File feature requests at [our issue tracker](https://github.com/VictoriaMetrics/VictoriaMetrics/issues) if you need other service discovery mechanisms to be supported by `vmagent`.
-
-
 ### Adding labels to metrics
 
 Labels can be added to metrics via the following mechanisms:
diff --git a/go.mod b/go.mod
index 53c77fbe1..0fe64d02d 100644
--- a/go.mod
+++ b/go.mod
@@ -9,10 +9,11 @@ require (
 	// like https://github.com/valyala/fasthttp/commit/996610f021ff45fdc98c2ce7884d5fa4e7f9199b
 	github.com/VictoriaMetrics/fasthttp v1.0.1
 	github.com/VictoriaMetrics/metrics v1.11.3
-	github.com/VictoriaMetrics/metricsql v0.2.3
+	github.com/VictoriaMetrics/metricsql v0.2.4
 	github.com/aws/aws-sdk-go v1.32.13
 	github.com/cespare/xxhash/v2 v2.1.1
 	github.com/golang/snappy v0.0.1
+	github.com/google/uuid v1.1.1
 	github.com/klauspost/compress v1.10.10
 	github.com/valyala/fastjson v1.5.3
 	github.com/valyala/fastrand v1.0.0
diff --git a/go.sum b/go.sum
index 894de115a..f71db1bad 100644
--- a/go.sum
+++ b/go.sum
@@ -53,8 +53,8 @@ github.com/VictoriaMetrics/metrics v1.11.2 h1:t/ceLP6SvagUqypCKU7cI7+tQn54+TIV/t
 github.com/VictoriaMetrics/metrics v1.11.2/go.mod h1:LU2j9qq7xqZYXz8tF3/RQnB2z2MbZms5TDiIg9/NHiQ=
 github.com/VictoriaMetrics/metrics v1.11.3 h1:eSfXc0CrquKa1VTNUvhP+dhNjLUZHQGTFfp19mYCQWE=
 github.com/VictoriaMetrics/metrics v1.11.3/go.mod h1:LU2j9qq7xqZYXz8tF3/RQnB2z2MbZms5TDiIg9/NHiQ=
-github.com/VictoriaMetrics/metricsql v0.2.3 h1:xGscDmLoeIV7+8qX/mdHnOY0vu4m+wHIVGMoy/nBovY=
-github.com/VictoriaMetrics/metricsql v0.2.3/go.mod h1:UIjd9S0W1UnTWlJdM0wLS+2pfuPqjwqKoK8yTos+WyE=
+github.com/VictoriaMetrics/metricsql v0.2.4 h1:240YwT8B4KITVFE7EOrf1rVvsY+5fYsAzyb+bI6/q50=
+github.com/VictoriaMetrics/metricsql v0.2.4/go.mod h1:UIjd9S0W1UnTWlJdM0wLS+2pfuPqjwqKoK8yTos+WyE=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8=
 github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/aws/aws-sdk-go v1.32.13 h1:zzyXF7SUxJcJa3hTcYCl1/Ey+kh2N8TjK5tWnL0wieo=
@@ -128,6 +128,8 @@ github.com/google/pprof v0.0.0-20200229191704-1ebb73c60ed3/go.mod h1:ZgVRPoUq/hf
 github.com/google/pprof v0.0.0-20200430221834-fc25d7d30c6d/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/pprof v0.0.0-20200507031123-427632fa3b1c/go.mod h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5 h1:sjZBwGj9Jlw33ImPtvFviGYvseOtDM7hkSKB7+Tv3SM=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
diff --git a/lib/fs/reader_at.go b/lib/fs/reader_at.go
index 6a39865fb..ef30eb019 100644
--- a/lib/fs/reader_at.go
+++ b/lib/fs/reader_at.go
@@ -14,7 +14,8 @@ import (
 )
 
 var disableMmap = flag.Bool("fs.disableMmap", is32BitPtr, "Whether to use pread() instead of mmap() for reading data files. "+
-	"By default mmap() is used for 64-bit arches and pread() is used for 32-bit arches, since they cannot data files bigger than 2^32 bytes in memory")
+	"By default mmap() is used for 64-bit arches and pread() is used for 32-bit arches, since they cannot read data files bigger than 2^32 bytes in memory. "+
+	"mmap() is usually faster for reading small data chunks than pread()")
 
 const is32BitPtr = (^uintptr(0) >> 32) == 0
 
diff --git a/lib/promrelabel/config.go b/lib/promrelabel/config.go
index f0134c0ac..8904ee033 100644
--- a/lib/promrelabel/config.go
+++ b/lib/promrelabel/config.go
@@ -90,7 +90,7 @@ func parseRelabelConfig(dst []ParsedRelabelConfig, rc *RelabelConfig) ([]ParsedR
 			return dst, fmt.Errorf("missing `source_labels` for `action=replace_all`")
 		}
 		if targetLabel == "" {
-			return dst, fmt.Errorf("missing `target_label` for `action=replace`")
+			return dst, fmt.Errorf("missing `target_label` for `action=replace_all`")
 		}
 	case "keep_if_equal":
 		if len(sourceLabels) < 2 {
diff --git a/lib/promscrape/client.go b/lib/promscrape/client.go
index cf1b227db..7c861d855 100644
--- a/lib/promscrape/client.go
+++ b/lib/promscrape/client.go
@@ -14,19 +14,24 @@ import (
 var (
 	maxScrapeSize = flag.Int("promscrape.maxScrapeSize", 16*1024*1024, "The maximum size of scrape response in bytes to process from Prometheus targets. "+
 		"Bigger responses are rejected")
-	disableCompression = flag.Bool("promscrape.disableCompression", false, "Whether to disable sending 'Accept-Encoding: gzip' request headers to scrape targets. "+
-		"This may reduce CPU usage on scrape targets at the cost of higher network bandwidth utilization")
-	disableKeepAlive = flag.Bool("promscrape.disableKeepAlive", false, "Whether to disable HTTP keep-alive connections when scraping targets. This may be useful when targets "+
-		"has no support for HTTP keep-alive connection. Note that disabling HTTP keep-alive may increase load on both vmagent and scrape targets")
+	disableCompression = flag.Bool("promscrape.disableCompression", false, "Whether to disable sending 'Accept-Encoding: gzip' request headers to all the scrape targets. "+
+		"This may reduce CPU usage on scrape targets at the cost of higher network bandwidth utilization. "+
+		"It is possible to set 'disable_compression: true' individually per each 'scrape_config' section in '-promscrape.config' for fine grained control")
+	disableKeepAlive = flag.Bool("promscrape.disableKeepAlive", false, "Whether to disable HTTP keep-alive connections when scraping all the targets. "+
+		"This may be useful when targets has no support for HTTP keep-alive connection. "+
+		"It is possible to set `disable_keepalive: true` individually per each 'scrape_config` section in '-promscrape.config' for fine grained control. "+
+		"Note that disabling HTTP keep-alive may increase load on both vmagent and scrape targets")
 )
 
 type client struct {
 	hc *fasthttp.HostClient
 
-	scrapeURL  string
-	host       string
-	requestURI string
-	authHeader string
+	scrapeURL          string
+	host               string
+	requestURI         string
+	authHeader         string
+	disableCompression bool
+	disableKeepAlive   bool
 }
 
 func newClient(sw *ScrapeWork) *client {
@@ -61,36 +66,39 @@ func newClient(sw *ScrapeWork) *client {
 	return &client{
 		hc: hc,
 
-		scrapeURL:  sw.ScrapeURL,
-		host:       host,
-		requestURI: requestURI,
-		authHeader: sw.AuthConfig.Authorization,
+		scrapeURL:          sw.ScrapeURL,
+		host:               host,
+		requestURI:         requestURI,
+		authHeader:         sw.AuthConfig.Authorization,
+		disableCompression: sw.DisableCompression,
+		disableKeepAlive:   sw.DisableKeepAlive,
 	}
 }
 
 func (c *client) ReadData(dst []byte) ([]byte, error) {
+	deadline := time.Now().Add(c.hc.ReadTimeout)
 	req := fasthttp.AcquireRequest()
 	req.SetRequestURI(c.requestURI)
 	req.SetHost(c.host)
-	if !*disableCompression {
+	if !*disableCompression || c.disableCompression {
 		req.Header.Set("Accept-Encoding", "gzip")
 	}
-	if *disableKeepAlive {
+	if *disableKeepAlive || c.disableKeepAlive {
 		req.SetConnectionClose()
 	}
 	if c.authHeader != "" {
 		req.Header.Set("Authorization", c.authHeader)
 	}
 	resp := fasthttp.AcquireResponse()
-	err := doRequestWithPossibleRetry(c.hc, req, resp)
+	err := doRequestWithPossibleRetry(c.hc, req, resp, deadline)
 	statusCode := resp.StatusCode()
-	if statusCode == fasthttp.StatusMovedPermanently || statusCode == fasthttp.StatusFound {
+	if err == nil && (statusCode == fasthttp.StatusMovedPermanently || statusCode == fasthttp.StatusFound) {
 		// Allow a single redirect.
 		// It is expected that the redirect is made on the same host.
 		// Otherwise it won't work.
 		if location := resp.Header.Peek("Location"); len(location) > 0 {
 			req.URI().UpdateBytes(location)
-			err = c.hc.Do(req, resp)
+			err = c.hc.DoDeadline(req, resp, deadline)
 			statusCode = resp.StatusCode()
 		}
 	}
@@ -137,18 +145,19 @@ var (
 	scrapesGunzipFailed = metrics.NewCounter(`vm_promscrape_scrapes_gunzip_failed_total`)
 )
 
-func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response) error {
+func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response, deadline time.Time) 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)
+	// Use DoDeadline instead of Do even if hc.ReadTimeout is already set in order to guarantee the given deadline
+	// across multiple retries.
+	err := hc.DoDeadline(req, resp, deadline)
 	if err == nil {
 		return nil
 	}
 	if err != fasthttp.ErrConnectionClosed {
 		return err
 	}
-	// Retry request if the server closed the keep-alive connection during the first attempt.
+	// Retry request if the server closes the keep-alive connection unless deadline exceeds.
 	attempts++
 	if attempts > 3 {
 		return fmt.Errorf("the server closed 3 subsequent connections: %w", err)
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index 39071ad69..7b72d09c6 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -76,6 +76,10 @@ type ScrapeConfig struct {
 	MetricRelabelConfigs []promrelabel.RelabelConfig `yaml:"metric_relabel_configs"`
 	SampleLimit          int                         `yaml:"sample_limit"`
 
+	// These options are supported only by lib/promscrape.
+	DisableCompression bool `yaml:"disable_compression"`
+	DisableKeepAlive   bool `yaml:"disable_keepalive"`
+
 	// This is set in loadConfig
 	swc *scrapeWorkConfig
 }
@@ -404,6 +408,8 @@ func getScrapeWorkConfig(sc *ScrapeConfig, baseDir string, globalCfg *GlobalConf
 		relabelConfigs:       relabelConfigs,
 		metricRelabelConfigs: metricRelabelConfigs,
 		sampleLimit:          sc.SampleLimit,
+		disableCompression:   sc.DisableCompression,
+		disableKeepAlive:     sc.DisableKeepAlive,
 	}
 	return swc, nil
 }
@@ -422,6 +428,8 @@ type scrapeWorkConfig struct {
 	relabelConfigs       []promrelabel.ParsedRelabelConfig
 	metricRelabelConfigs []promrelabel.ParsedRelabelConfig
 	sampleLimit          int
+	disableCompression   bool
+	disableKeepAlive     bool
 }
 
 func appendKubernetesScrapeWork(dst []ScrapeWork, sdc *kubernetes.SDConfig, baseDir string, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
@@ -602,6 +610,8 @@ func appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfig, target string, ex
 		AuthConfig:           swc.authConfig,
 		MetricRelabelConfigs: swc.metricRelabelConfigs,
 		SampleLimit:          swc.sampleLimit,
+		DisableCompression:   swc.disableCompression,
+		DisableKeepAlive:     swc.disableKeepAlive,
 
 		jobNameOriginal: swc.jobName,
 	})
diff --git a/lib/promscrape/scrapework.go b/lib/promscrape/scrapework.go
index 396124edb..3d2e57647 100644
--- a/lib/promscrape/scrapework.go
+++ b/lib/promscrape/scrapework.go
@@ -67,6 +67,12 @@ type ScrapeWork struct {
 	// The maximum number of metrics to scrape after relabeling.
 	SampleLimit int
 
+	// Whether to disable response compression when querying ScrapeURL.
+	DisableCompression bool
+
+	// Whether to disable HTTP keep-alive when querying ScrapeURL.
+	DisableKeepAlive bool
+
 	// The original 'job_name'
 	jobNameOriginal string
 }
@@ -76,9 +82,9 @@ type ScrapeWork struct {
 // it can be used for comparing for equality two ScrapeWork objects.
 func (sw *ScrapeWork) key() string {
 	key := fmt.Sprintf("ScrapeURL=%s, ScrapeInterval=%s, ScrapeTimeout=%s, HonorLabels=%v, HonorTimestamps=%v, Labels=%s, "+
-		"AuthConfig=%s, MetricRelabelConfigs=%s, SampleLimit=%d",
+		"AuthConfig=%s, MetricRelabelConfigs=%s, SampleLimit=%d, DisableCompression=%v, DisableKeepAlive=%v",
 		sw.ScrapeURL, sw.ScrapeInterval, sw.ScrapeTimeout, sw.HonorLabels, sw.HonorTimestamps, sw.LabelsString(),
-		sw.AuthConfig.String(), sw.metricRelabelConfigsString(), sw.SampleLimit)
+		sw.AuthConfig.String(), sw.metricRelabelConfigsString(), sw.SampleLimit, sw.DisableCompression, sw.DisableKeepAlive)
 	return key
 }
 
diff --git a/lib/protoparser/graphite/streamparser.go b/lib/protoparser/graphite/streamparser.go
index 825ffdcb7..cdca2a3ed 100644
--- a/lib/protoparser/graphite/streamparser.go
+++ b/lib/protoparser/graphite/streamparser.go
@@ -75,7 +75,7 @@ func (ctx *streamContext) Read(r io.Reader) bool {
 	currentTimestamp := int64(fasttime.UnixTimestamp())
 	for i := range rows {
 		r := &rows[i]
-		if r.Timestamp == 0 {
+		if r.Timestamp == 0 || r.Timestamp == -1 {
 			r.Timestamp = currentTimestamp
 		}
 	}
diff --git a/lib/protoparser/promremotewrite/streamparser.go b/lib/protoparser/promremotewrite/streamparser.go
index 5f7268180..19e75d2e2 100644
--- a/lib/protoparser/promremotewrite/streamparser.go
+++ b/lib/protoparser/promremotewrite/streamparser.go
@@ -65,7 +65,7 @@ var (
 	readCalls       = metrics.NewCounter(`vm_protoparser_read_calls_total{type="promremotewrite"}`)
 	readErrors      = metrics.NewCounter(`vm_protoparser_read_errors_total{type="promremotewrite"}`)
 	rowsRead        = metrics.NewCounter(`vm_protoparser_rows_read_total{type="promremotewrite"}`)
-	unmarshalErrors = metrics.NewCounter(`vm_protoparser_unmarshal_errors{type="promremotewrite"}`)
+	unmarshalErrors = metrics.NewCounter(`vm_protoparser_unmarshal_errors_total{type="promremotewrite"}`)
 )
 
 func getPushCtx() *pushCtx {
diff --git a/lib/storage/index_db.go b/lib/storage/index_db.go
index c138144d4..006b32af0 100644
--- a/lib/storage/index_db.go
+++ b/lib/storage/index_db.go
@@ -746,6 +746,7 @@ func (db *indexDB) SearchTagKeys(maxTagKeys int) ([]string, error) {
 
 	keys := make([]string, 0, len(tks))
 	for key := range tks {
+		// Do not skip empty keys, since they are converted to __name__
 		keys = append(keys, key)
 	}
 
@@ -813,6 +814,11 @@ func (db *indexDB) SearchTagValues(tagKey []byte, maxTagValues int) ([]string, e
 
 	tagValues := make([]string, 0, len(tvs))
 	for tv := range tvs {
+		if len(tv) == 0 {
+			// Skip empty values, since they have no any meaning.
+			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/600
+			continue
+		}
 		tagValues = append(tagValues, tv)
 	}
 
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index 298f6738a..475bd1ec9 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -327,6 +327,8 @@ type Metrics struct {
 	AddRowsConcurrencyCapacity     uint64
 	AddRowsConcurrencyCurrent      uint64
 
+	SearchDelays uint64
+
 	SlowRowInserts         uint64
 	SlowPerDayIndexInserts uint64
 	SlowMetricNameLoads    uint64
@@ -385,6 +387,8 @@ func (s *Storage) UpdateMetrics(m *Metrics) {
 	m.AddRowsConcurrencyCapacity = uint64(cap(addRowsConcurrencyCh))
 	m.AddRowsConcurrencyCurrent = uint64(len(addRowsConcurrencyCh))
 
+	m.SearchDelays += atomic.LoadUint64(&searchDelays)
+
 	m.SlowRowInserts += atomic.LoadUint64(&s.slowRowInserts)
 	m.SlowPerDayIndexInserts += atomic.LoadUint64(&s.slowPerDayIndexInserts)
 	m.SlowMetricNameLoads += atomic.LoadUint64(&s.slowMetricNameLoads)
@@ -793,8 +797,26 @@ func nextRetentionDuration(retentionMonths int) time.Duration {
 	return deadline.Sub(t)
 }
 
+var (
+	searchTSIDsCondLock sync.Mutex
+	searchTSIDsCond     = sync.NewCond(&searchTSIDsCondLock)
+
+	searchDelays uint64
+)
+
 // searchTSIDs returns sorted TSIDs for the given tfss and the given tr.
 func (s *Storage) searchTSIDs(tfss []*TagFilters, tr TimeRange, maxMetrics int) ([]TSID, error) {
+	// Make sure that there are enough resources for processing the ingested data via Storage.AddRows
+	// before starting the query.
+	// This should prevent from data ingestion starvation when provessing heavy queries.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/291 .
+	searchTSIDsCondLock.Lock()
+	for len(addRowsConcurrencyCh) >= cap(addRowsConcurrencyCh) {
+		atomic.AddUint64(&searchDelays, 1)
+		searchTSIDsCond.Wait()
+	}
+	searchTSIDsCondLock.Unlock()
+
 	// Do not cache tfss -> tsids here, since the caching is performed
 	// on idb level.
 	tsids, err := s.idb().searchTSIDs(tfss, tr, maxMetrics)
@@ -854,12 +876,13 @@ func (s *Storage) DeleteMetrics(tfss []*TagFilters) (int, error) {
 	if err != nil {
 		return deletedCount, fmt.Errorf("cannot delete tsids: %w", err)
 	}
-	// Do not reset MetricName -> TSID cache (tsidCache), since the obtained
-	// entries must be checked against deleted metricIDs.
-	// See Storage.add for details.
-	//
+	// Reset MetricName->TSID cache in order to prevent from adding new data points
+	// to deleted time series in Storage.add.
+	s.tsidCache.Reset()
+
 	// Do not reset MetricID -> MetricName cache, since it must be used only
 	// after filtering out deleted metricIDs.
+
 	return deletedCount, nil
 }
 
@@ -998,7 +1021,6 @@ func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
 	// goroutines call AddRows.
 	select {
 	case addRowsConcurrencyCh <- struct{}{}:
-		defer func() { <-addRowsConcurrencyCh }()
 	default:
 		// Sleep for a while until giving up
 		atomic.AddUint64(&s.addRowsConcurrencyLimitReached, 1)
@@ -1006,7 +1028,6 @@ func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
 		select {
 		case addRowsConcurrencyCh <- struct{}{}:
 			timerpool.Put(t)
-			defer func() { <-addRowsConcurrencyCh }()
 		case <-t.C:
 			timerpool.Put(t)
 			atomic.AddUint64(&s.addRowsConcurrencyLimitTimeout, 1)
@@ -1022,17 +1043,23 @@ func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) error {
 	rr.rows, err = s.add(rr.rows, mrs, precisionBits)
 	putRawRows(rr)
 
+	// Notify blocked goroutines at Storage.searchTSIDs that they may proceed with their work.
+	<-addRowsConcurrencyCh
+	searchTSIDsCond.Signal()
+
 	return err
 }
 
 var (
-	addRowsConcurrencyCh = make(chan struct{}, runtime.GOMAXPROCS(-1)*2)
+	// Limit the concurrency for data ingestion to GOMAXPROCS, since this operation
+	// is CPU bound, so there is no sense in running more than GOMAXPROCS concurrent
+	// goroutines on data ingestion path.
+	addRowsConcurrencyCh = make(chan struct{}, runtime.GOMAXPROCS(-1))
 	addRowsTimeout       = 30 * time.Second
 )
 
 func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]rawRow, error) {
 	idb := s.idb()
-	dmis := idb.getDeletedMetricIDs()
 	rowsLen := len(rows)
 	if n := rowsLen + len(mrs) - cap(rows); n > 0 {
 		rows = append(rows[:cap(rows)], make([]rawRow, n)...)
@@ -1058,7 +1085,8 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
 		if mr.Timestamp < minTimestamp {
 			// Skip rows with too small timestamps outside the retention.
 			if firstWarn == nil {
-				firstWarn = fmt.Errorf("cannot insert row with too small timestamp %d outside the retention; minimum allowed timestamp is %d",
+				firstWarn = fmt.Errorf("cannot insert row with too small timestamp %d outside the retention; minimum allowed timestamp is %d; "+
+					"probably you need updating -retentionPeriod command-line flag",
 					mr.Timestamp, minTimestamp)
 			}
 			atomic.AddUint64(&s.tooSmallTimestampRows, 1)
@@ -1067,7 +1095,8 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
 		if mr.Timestamp > maxTimestamp {
 			// Skip rows with too big timestamps significantly exceeding the current time.
 			if firstWarn == nil {
-				firstWarn = fmt.Errorf("cannot insert row with too big timestamp %d exceeding the current time; maximum allowd timestamp is %d",
+				firstWarn = fmt.Errorf("cannot insert row with too big timestamp %d exceeding the current time; maximum allowd timestamp is %d; "+
+					"propbably you need updating -retentionPeriod command-line flag",
 					mr.Timestamp, maxTimestamp)
 			}
 			atomic.AddUint64(&s.tooBigTimestampRows, 1)
@@ -1084,8 +1113,11 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
 			r.TSID = prevTSID
 			continue
 		}
-		if s.getTSIDFromCache(&r.TSID, mr.MetricNameRaw) && !dmis.Has(r.TSID.MetricID) {
+		if s.getTSIDFromCache(&r.TSID, mr.MetricNameRaw) {
 			// Fast path - the TSID for the given MetricName has been found in cache and isn't deleted.
+			// There is no need in checking whether r.TSID.MetricID is deleted, since tsidCache doesn't
+			// contain MetricName->TSID entries for deleted time series.
+			// See Storage.DeleteMetrics code for details.
 			prevTSID = r.TSID
 			prevMetricNameRaw = mr.MetricNameRaw
 			continue
@@ -1130,8 +1162,11 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
 				r.TSID = prevTSID
 				continue
 			}
-			if s.getTSIDFromCache(&r.TSID, mr.MetricNameRaw) && !dmis.Has(r.TSID.MetricID) {
+			if s.getTSIDFromCache(&r.TSID, mr.MetricNameRaw) {
 				// Fast path - the TSID for the given MetricName has been found in cache and isn't deleted.
+				// There is no need in checking whether r.TSID.MetricID is deleted, since tsidCache doesn't
+				// contain MetricName->TSID entries for deleted time series.
+				// See Storage.DeleteMetrics code for details.
 				prevTSID = r.TSID
 				prevMetricNameRaw = mr.MetricNameRaw
 				continue
diff --git a/vendor/github.com/VictoriaMetrics/metricsql/transform.go b/vendor/github.com/VictoriaMetrics/metricsql/transform.go
index de86e7a0b..d19222637 100644
--- a/vendor/github.com/VictoriaMetrics/metricsql/transform.go
+++ b/vendor/github.com/VictoriaMetrics/metricsql/transform.go
@@ -51,6 +51,7 @@ var transformFuncs = map[string]bool{
 	"":                   true, // empty func is a synonim to union
 	"keep_last_value":    true,
 	"keep_next_value":    true,
+	"interpolate":        true,
 	"start":              true,
 	"end":                true,
 	"step":               true,
diff --git a/vendor/github.com/google/uuid/.travis.yml b/vendor/github.com/google/uuid/.travis.yml
new file mode 100644
index 000000000..d8156a60b
--- /dev/null
+++ b/vendor/github.com/google/uuid/.travis.yml
@@ -0,0 +1,9 @@
+language: go
+
+go:
+  - 1.4.3
+  - 1.5.3
+  - tip
+
+script:
+  - go test -v ./...
diff --git a/vendor/github.com/google/uuid/CONTRIBUTING.md b/vendor/github.com/google/uuid/CONTRIBUTING.md
new file mode 100644
index 000000000..04fdf09f1
--- /dev/null
+++ b/vendor/github.com/google/uuid/CONTRIBUTING.md
@@ -0,0 +1,10 @@
+# How to contribute
+
+We definitely welcome patches and contribution to this project!
+
+### Legal requirements
+
+In order to protect both you and ourselves, you will need to sign the
+[Contributor License Agreement](https://cla.developers.google.com/clas).
+
+You may have already signed it for other Google projects.
diff --git a/vendor/github.com/google/uuid/CONTRIBUTORS b/vendor/github.com/google/uuid/CONTRIBUTORS
new file mode 100644
index 000000000..b4bb97f6b
--- /dev/null
+++ b/vendor/github.com/google/uuid/CONTRIBUTORS
@@ -0,0 +1,9 @@
+Paul Borman <borman@google.com>
+bmatsuo
+shawnps
+theory
+jboverfelt
+dsymonds
+cd1
+wallclockbuilder
+dansouza
diff --git a/vendor/github.com/google/uuid/LICENSE b/vendor/github.com/google/uuid/LICENSE
new file mode 100644
index 000000000..5dc68268d
--- /dev/null
+++ b/vendor/github.com/google/uuid/LICENSE
@@ -0,0 +1,27 @@
+Copyright (c) 2009,2014 Google Inc. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/vendor/github.com/google/uuid/README.md b/vendor/github.com/google/uuid/README.md
new file mode 100644
index 000000000..9d92c11f1
--- /dev/null
+++ b/vendor/github.com/google/uuid/README.md
@@ -0,0 +1,19 @@
+# uuid ![build status](https://travis-ci.org/google/uuid.svg?branch=master)
+The uuid package generates and inspects UUIDs based on
+[RFC 4122](http://tools.ietf.org/html/rfc4122)
+and DCE 1.1: Authentication and Security Services. 
+
+This package is based on the github.com/pborman/uuid package (previously named
+code.google.com/p/go-uuid).  It differs from these earlier packages in that
+a UUID is a 16 byte array rather than a byte slice.  One loss due to this
+change is the ability to represent an invalid UUID (vs a NIL UUID).
+
+###### Install
+`go get github.com/google/uuid`
+
+###### Documentation 
+[![GoDoc](https://godoc.org/github.com/google/uuid?status.svg)](http://godoc.org/github.com/google/uuid)
+
+Full `go doc` style documentation for the package can be viewed online without
+installing this package by using the GoDoc site here: 
+http://godoc.org/github.com/google/uuid
diff --git a/vendor/github.com/google/uuid/dce.go b/vendor/github.com/google/uuid/dce.go
new file mode 100644
index 000000000..fa820b9d3
--- /dev/null
+++ b/vendor/github.com/google/uuid/dce.go
@@ -0,0 +1,80 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"encoding/binary"
+	"fmt"
+	"os"
+)
+
+// A Domain represents a Version 2 domain
+type Domain byte
+
+// Domain constants for DCE Security (Version 2) UUIDs.
+const (
+	Person = Domain(0)
+	Group  = Domain(1)
+	Org    = Domain(2)
+)
+
+// NewDCESecurity returns a DCE Security (Version 2) UUID.
+//
+// The domain should be one of Person, Group or Org.
+// On a POSIX system the id should be the users UID for the Person
+// domain and the users GID for the Group.  The meaning of id for
+// the domain Org or on non-POSIX systems is site defined.
+//
+// For a given domain/id pair the same token may be returned for up to
+// 7 minutes and 10 seconds.
+func NewDCESecurity(domain Domain, id uint32) (UUID, error) {
+	uuid, err := NewUUID()
+	if err == nil {
+		uuid[6] = (uuid[6] & 0x0f) | 0x20 // Version 2
+		uuid[9] = byte(domain)
+		binary.BigEndian.PutUint32(uuid[0:], id)
+	}
+	return uuid, err
+}
+
+// NewDCEPerson returns a DCE Security (Version 2) UUID in the person
+// domain with the id returned by os.Getuid.
+//
+//  NewDCESecurity(Person, uint32(os.Getuid()))
+func NewDCEPerson() (UUID, error) {
+	return NewDCESecurity(Person, uint32(os.Getuid()))
+}
+
+// NewDCEGroup returns a DCE Security (Version 2) UUID in the group
+// domain with the id returned by os.Getgid.
+//
+//  NewDCESecurity(Group, uint32(os.Getgid()))
+func NewDCEGroup() (UUID, error) {
+	return NewDCESecurity(Group, uint32(os.Getgid()))
+}
+
+// Domain returns the domain for a Version 2 UUID.  Domains are only defined
+// for Version 2 UUIDs.
+func (uuid UUID) Domain() Domain {
+	return Domain(uuid[9])
+}
+
+// ID returns the id for a Version 2 UUID. IDs are only defined for Version 2
+// UUIDs.
+func (uuid UUID) ID() uint32 {
+	return binary.BigEndian.Uint32(uuid[0:4])
+}
+
+func (d Domain) String() string {
+	switch d {
+	case Person:
+		return "Person"
+	case Group:
+		return "Group"
+	case Org:
+		return "Org"
+	}
+	return fmt.Sprintf("Domain%d", int(d))
+}
diff --git a/vendor/github.com/google/uuid/doc.go b/vendor/github.com/google/uuid/doc.go
new file mode 100644
index 000000000..5b8a4b9af
--- /dev/null
+++ b/vendor/github.com/google/uuid/doc.go
@@ -0,0 +1,12 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package uuid generates and inspects UUIDs.
+//
+// UUIDs are based on RFC 4122 and DCE 1.1: Authentication and Security
+// Services.
+//
+// A UUID is a 16 byte (128 bit) array.  UUIDs may be used as keys to
+// maps or compared directly.
+package uuid
diff --git a/vendor/github.com/google/uuid/go.mod b/vendor/github.com/google/uuid/go.mod
new file mode 100644
index 000000000..fc84cd79d
--- /dev/null
+++ b/vendor/github.com/google/uuid/go.mod
@@ -0,0 +1,1 @@
+module github.com/google/uuid
diff --git a/vendor/github.com/google/uuid/hash.go b/vendor/github.com/google/uuid/hash.go
new file mode 100644
index 000000000..b17461631
--- /dev/null
+++ b/vendor/github.com/google/uuid/hash.go
@@ -0,0 +1,53 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"crypto/md5"
+	"crypto/sha1"
+	"hash"
+)
+
+// Well known namespace IDs and UUIDs
+var (
+	NameSpaceDNS  = Must(Parse("6ba7b810-9dad-11d1-80b4-00c04fd430c8"))
+	NameSpaceURL  = Must(Parse("6ba7b811-9dad-11d1-80b4-00c04fd430c8"))
+	NameSpaceOID  = Must(Parse("6ba7b812-9dad-11d1-80b4-00c04fd430c8"))
+	NameSpaceX500 = Must(Parse("6ba7b814-9dad-11d1-80b4-00c04fd430c8"))
+	Nil           UUID // empty UUID, all zeros
+)
+
+// NewHash returns a new UUID derived from the hash of space concatenated with
+// data generated by h.  The hash should be at least 16 byte in length.  The
+// first 16 bytes of the hash are used to form the UUID.  The version of the
+// UUID will be the lower 4 bits of version.  NewHash is used to implement
+// NewMD5 and NewSHA1.
+func NewHash(h hash.Hash, space UUID, data []byte, version int) UUID {
+	h.Reset()
+	h.Write(space[:])
+	h.Write(data)
+	s := h.Sum(nil)
+	var uuid UUID
+	copy(uuid[:], s)
+	uuid[6] = (uuid[6] & 0x0f) | uint8((version&0xf)<<4)
+	uuid[8] = (uuid[8] & 0x3f) | 0x80 // RFC 4122 variant
+	return uuid
+}
+
+// NewMD5 returns a new MD5 (Version 3) UUID based on the
+// supplied name space and data.  It is the same as calling:
+//
+//  NewHash(md5.New(), space, data, 3)
+func NewMD5(space UUID, data []byte) UUID {
+	return NewHash(md5.New(), space, data, 3)
+}
+
+// NewSHA1 returns a new SHA1 (Version 5) UUID based on the
+// supplied name space and data.  It is the same as calling:
+//
+//  NewHash(sha1.New(), space, data, 5)
+func NewSHA1(space UUID, data []byte) UUID {
+	return NewHash(sha1.New(), space, data, 5)
+}
diff --git a/vendor/github.com/google/uuid/marshal.go b/vendor/github.com/google/uuid/marshal.go
new file mode 100644
index 000000000..7f9e0c6c0
--- /dev/null
+++ b/vendor/github.com/google/uuid/marshal.go
@@ -0,0 +1,37 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import "fmt"
+
+// MarshalText implements encoding.TextMarshaler.
+func (uuid UUID) MarshalText() ([]byte, error) {
+	var js [36]byte
+	encodeHex(js[:], uuid)
+	return js[:], nil
+}
+
+// UnmarshalText implements encoding.TextUnmarshaler.
+func (uuid *UUID) UnmarshalText(data []byte) error {
+	id, err := ParseBytes(data)
+	if err == nil {
+		*uuid = id
+	}
+	return err
+}
+
+// MarshalBinary implements encoding.BinaryMarshaler.
+func (uuid UUID) MarshalBinary() ([]byte, error) {
+	return uuid[:], nil
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler.
+func (uuid *UUID) UnmarshalBinary(data []byte) error {
+	if len(data) != 16 {
+		return fmt.Errorf("invalid UUID (got %d bytes)", len(data))
+	}
+	copy(uuid[:], data)
+	return nil
+}
diff --git a/vendor/github.com/google/uuid/node.go b/vendor/github.com/google/uuid/node.go
new file mode 100644
index 000000000..d651a2b06
--- /dev/null
+++ b/vendor/github.com/google/uuid/node.go
@@ -0,0 +1,90 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"sync"
+)
+
+var (
+	nodeMu sync.Mutex
+	ifname string  // name of interface being used
+	nodeID [6]byte // hardware for version 1 UUIDs
+	zeroID [6]byte // nodeID with only 0's
+)
+
+// NodeInterface returns the name of the interface from which the NodeID was
+// derived.  The interface "user" is returned if the NodeID was set by
+// SetNodeID.
+func NodeInterface() string {
+	defer nodeMu.Unlock()
+	nodeMu.Lock()
+	return ifname
+}
+
+// SetNodeInterface selects the hardware address to be used for Version 1 UUIDs.
+// If name is "" then the first usable interface found will be used or a random
+// Node ID will be generated.  If a named interface cannot be found then false
+// is returned.
+//
+// SetNodeInterface never fails when name is "".
+func SetNodeInterface(name string) bool {
+	defer nodeMu.Unlock()
+	nodeMu.Lock()
+	return setNodeInterface(name)
+}
+
+func setNodeInterface(name string) bool {
+	iname, addr := getHardwareInterface(name) // null implementation for js
+	if iname != "" && addr != nil {
+		ifname = iname
+		copy(nodeID[:], addr)
+		return true
+	}
+
+	// We found no interfaces with a valid hardware address.  If name
+	// does not specify a specific interface generate a random Node ID
+	// (section 4.1.6)
+	if name == "" {
+		ifname = "random"
+		randomBits(nodeID[:])
+		return true
+	}
+	return false
+}
+
+// NodeID returns a slice of a copy of the current Node ID, setting the Node ID
+// if not already set.
+func NodeID() []byte {
+	defer nodeMu.Unlock()
+	nodeMu.Lock()
+	if nodeID == zeroID {
+		setNodeInterface("")
+	}
+	nid := nodeID
+	return nid[:]
+}
+
+// SetNodeID sets the Node ID to be used for Version 1 UUIDs.  The first 6 bytes
+// of id are used.  If id is less than 6 bytes then false is returned and the
+// Node ID is not set.
+func SetNodeID(id []byte) bool {
+	if len(id) < 6 {
+		return false
+	}
+	defer nodeMu.Unlock()
+	nodeMu.Lock()
+	copy(nodeID[:], id)
+	ifname = "user"
+	return true
+}
+
+// NodeID returns the 6 byte node id encoded in uuid.  It returns nil if uuid is
+// not valid.  The NodeID is only well defined for version 1 and 2 UUIDs.
+func (uuid UUID) NodeID() []byte {
+	var node [6]byte
+	copy(node[:], uuid[10:])
+	return node[:]
+}
diff --git a/vendor/github.com/google/uuid/node_js.go b/vendor/github.com/google/uuid/node_js.go
new file mode 100644
index 000000000..24b78edc9
--- /dev/null
+++ b/vendor/github.com/google/uuid/node_js.go
@@ -0,0 +1,12 @@
+// Copyright 2017 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build js
+
+package uuid
+
+// getHardwareInterface returns nil values for the JS version of the code.
+// This remvoves the "net" dependency, because it is not used in the browser.
+// Using the "net" library inflates the size of the transpiled JS code by 673k bytes.
+func getHardwareInterface(name string) (string, []byte) { return "", nil }
diff --git a/vendor/github.com/google/uuid/node_net.go b/vendor/github.com/google/uuid/node_net.go
new file mode 100644
index 000000000..0cbbcddbd
--- /dev/null
+++ b/vendor/github.com/google/uuid/node_net.go
@@ -0,0 +1,33 @@
+// Copyright 2017 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build !js
+
+package uuid
+
+import "net"
+
+var interfaces []net.Interface // cached list of interfaces
+
+// getHardwareInterface returns the name and hardware address of interface name.
+// If name is "" then the name and hardware address of one of the system's
+// interfaces is returned.  If no interfaces are found (name does not exist or
+// there are no interfaces) then "", nil is returned.
+//
+// Only addresses of at least 6 bytes are returned.
+func getHardwareInterface(name string) (string, []byte) {
+	if interfaces == nil {
+		var err error
+		interfaces, err = net.Interfaces()
+		if err != nil {
+			return "", nil
+		}
+	}
+	for _, ifs := range interfaces {
+		if len(ifs.HardwareAddr) >= 6 && (name == "" || name == ifs.Name) {
+			return ifs.Name, ifs.HardwareAddr
+		}
+	}
+	return "", nil
+}
diff --git a/vendor/github.com/google/uuid/sql.go b/vendor/github.com/google/uuid/sql.go
new file mode 100644
index 000000000..f326b54db
--- /dev/null
+++ b/vendor/github.com/google/uuid/sql.go
@@ -0,0 +1,59 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"database/sql/driver"
+	"fmt"
+)
+
+// Scan implements sql.Scanner so UUIDs can be read from databases transparently
+// Currently, database types that map to string and []byte are supported. Please
+// consult database-specific driver documentation for matching types.
+func (uuid *UUID) Scan(src interface{}) error {
+	switch src := src.(type) {
+	case nil:
+		return nil
+
+	case string:
+		// if an empty UUID comes from a table, we return a null UUID
+		if src == "" {
+			return nil
+		}
+
+		// see Parse for required string format
+		u, err := Parse(src)
+		if err != nil {
+			return fmt.Errorf("Scan: %v", err)
+		}
+
+		*uuid = u
+
+	case []byte:
+		// if an empty UUID comes from a table, we return a null UUID
+		if len(src) == 0 {
+			return nil
+		}
+
+		// assumes a simple slice of bytes if 16 bytes
+		// otherwise attempts to parse
+		if len(src) != 16 {
+			return uuid.Scan(string(src))
+		}
+		copy((*uuid)[:], src)
+
+	default:
+		return fmt.Errorf("Scan: unable to scan type %T into UUID", src)
+	}
+
+	return nil
+}
+
+// Value implements sql.Valuer so that UUIDs can be written to databases
+// transparently. Currently, UUIDs map to strings. Please consult
+// database-specific driver documentation for matching types.
+func (uuid UUID) Value() (driver.Value, error) {
+	return uuid.String(), nil
+}
diff --git a/vendor/github.com/google/uuid/time.go b/vendor/github.com/google/uuid/time.go
new file mode 100644
index 000000000..e6ef06cdc
--- /dev/null
+++ b/vendor/github.com/google/uuid/time.go
@@ -0,0 +1,123 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"encoding/binary"
+	"sync"
+	"time"
+)
+
+// A Time represents a time as the number of 100's of nanoseconds since 15 Oct
+// 1582.
+type Time int64
+
+const (
+	lillian    = 2299160          // Julian day of 15 Oct 1582
+	unix       = 2440587          // Julian day of 1 Jan 1970
+	epoch      = unix - lillian   // Days between epochs
+	g1582      = epoch * 86400    // seconds between epochs
+	g1582ns100 = g1582 * 10000000 // 100s of a nanoseconds between epochs
+)
+
+var (
+	timeMu   sync.Mutex
+	lasttime uint64 // last time we returned
+	clockSeq uint16 // clock sequence for this run
+
+	timeNow = time.Now // for testing
+)
+
+// UnixTime converts t the number of seconds and nanoseconds using the Unix
+// epoch of 1 Jan 1970.
+func (t Time) UnixTime() (sec, nsec int64) {
+	sec = int64(t - g1582ns100)
+	nsec = (sec % 10000000) * 100
+	sec /= 10000000
+	return sec, nsec
+}
+
+// GetTime returns the current Time (100s of nanoseconds since 15 Oct 1582) and
+// clock sequence as well as adjusting the clock sequence as needed.  An error
+// is returned if the current time cannot be determined.
+func GetTime() (Time, uint16, error) {
+	defer timeMu.Unlock()
+	timeMu.Lock()
+	return getTime()
+}
+
+func getTime() (Time, uint16, error) {
+	t := timeNow()
+
+	// If we don't have a clock sequence already, set one.
+	if clockSeq == 0 {
+		setClockSequence(-1)
+	}
+	now := uint64(t.UnixNano()/100) + g1582ns100
+
+	// If time has gone backwards with this clock sequence then we
+	// increment the clock sequence
+	if now <= lasttime {
+		clockSeq = ((clockSeq + 1) & 0x3fff) | 0x8000
+	}
+	lasttime = now
+	return Time(now), clockSeq, nil
+}
+
+// ClockSequence returns the current clock sequence, generating one if not
+// already set.  The clock sequence is only used for Version 1 UUIDs.
+//
+// The uuid package does not use global static storage for the clock sequence or
+// the last time a UUID was generated.  Unless SetClockSequence is used, a new
+// random clock sequence is generated the first time a clock sequence is
+// requested by ClockSequence, GetTime, or NewUUID.  (section 4.2.1.1)
+func ClockSequence() int {
+	defer timeMu.Unlock()
+	timeMu.Lock()
+	return clockSequence()
+}
+
+func clockSequence() int {
+	if clockSeq == 0 {
+		setClockSequence(-1)
+	}
+	return int(clockSeq & 0x3fff)
+}
+
+// SetClockSequence sets the clock sequence to the lower 14 bits of seq.  Setting to
+// -1 causes a new sequence to be generated.
+func SetClockSequence(seq int) {
+	defer timeMu.Unlock()
+	timeMu.Lock()
+	setClockSequence(seq)
+}
+
+func setClockSequence(seq int) {
+	if seq == -1 {
+		var b [2]byte
+		randomBits(b[:]) // clock sequence
+		seq = int(b[0])<<8 | int(b[1])
+	}
+	oldSeq := clockSeq
+	clockSeq = uint16(seq&0x3fff) | 0x8000 // Set our variant
+	if oldSeq != clockSeq {
+		lasttime = 0
+	}
+}
+
+// Time returns the time in 100s of nanoseconds since 15 Oct 1582 encoded in
+// uuid.  The time is only defined for version 1 and 2 UUIDs.
+func (uuid UUID) Time() Time {
+	time := int64(binary.BigEndian.Uint32(uuid[0:4]))
+	time |= int64(binary.BigEndian.Uint16(uuid[4:6])) << 32
+	time |= int64(binary.BigEndian.Uint16(uuid[6:8])&0xfff) << 48
+	return Time(time)
+}
+
+// ClockSequence returns the clock sequence encoded in uuid.
+// The clock sequence is only well defined for version 1 and 2 UUIDs.
+func (uuid UUID) ClockSequence() int {
+	return int(binary.BigEndian.Uint16(uuid[8:10])) & 0x3fff
+}
diff --git a/vendor/github.com/google/uuid/util.go b/vendor/github.com/google/uuid/util.go
new file mode 100644
index 000000000..5ea6c7378
--- /dev/null
+++ b/vendor/github.com/google/uuid/util.go
@@ -0,0 +1,43 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"io"
+)
+
+// randomBits completely fills slice b with random data.
+func randomBits(b []byte) {
+	if _, err := io.ReadFull(rander, b); err != nil {
+		panic(err.Error()) // rand should never fail
+	}
+}
+
+// xvalues returns the value of a byte as a hexadecimal digit or 255.
+var xvalues = [256]byte{
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 255, 255, 255, 255, 255, 255,
+	255, 10, 11, 12, 13, 14, 15, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 10, 11, 12, 13, 14, 15, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+}
+
+// xtob converts hex characters x1 and x2 into a byte.
+func xtob(x1, x2 byte) (byte, bool) {
+	b1 := xvalues[x1]
+	b2 := xvalues[x2]
+	return (b1 << 4) | b2, b1 != 255 && b2 != 255
+}
diff --git a/vendor/github.com/google/uuid/uuid.go b/vendor/github.com/google/uuid/uuid.go
new file mode 100644
index 000000000..524404cc5
--- /dev/null
+++ b/vendor/github.com/google/uuid/uuid.go
@@ -0,0 +1,245 @@
+// Copyright 2018 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+)
+
+// A UUID is a 128 bit (16 byte) Universal Unique IDentifier as defined in RFC
+// 4122.
+type UUID [16]byte
+
+// A Version represents a UUID's version.
+type Version byte
+
+// A Variant represents a UUID's variant.
+type Variant byte
+
+// Constants returned by Variant.
+const (
+	Invalid   = Variant(iota) // Invalid UUID
+	RFC4122                   // The variant specified in RFC4122
+	Reserved                  // Reserved, NCS backward compatibility.
+	Microsoft                 // Reserved, Microsoft Corporation backward compatibility.
+	Future                    // Reserved for future definition.
+)
+
+var rander = rand.Reader // random function
+
+// Parse decodes s into a UUID or returns an error.  Both the standard UUID
+// forms of xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx and
+// urn:uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx are decoded as well as the
+// Microsoft encoding {xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx} and the raw hex
+// encoding: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx.
+func Parse(s string) (UUID, error) {
+	var uuid UUID
+	switch len(s) {
+	// xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+	case 36:
+
+	// urn:uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+	case 36 + 9:
+		if strings.ToLower(s[:9]) != "urn:uuid:" {
+			return uuid, fmt.Errorf("invalid urn prefix: %q", s[:9])
+		}
+		s = s[9:]
+
+	// {xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx}
+	case 36 + 2:
+		s = s[1:]
+
+	// xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+	case 32:
+		var ok bool
+		for i := range uuid {
+			uuid[i], ok = xtob(s[i*2], s[i*2+1])
+			if !ok {
+				return uuid, errors.New("invalid UUID format")
+			}
+		}
+		return uuid, nil
+	default:
+		return uuid, fmt.Errorf("invalid UUID length: %d", len(s))
+	}
+	// s is now at least 36 bytes long
+	// it must be of the form  xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+	if s[8] != '-' || s[13] != '-' || s[18] != '-' || s[23] != '-' {
+		return uuid, errors.New("invalid UUID format")
+	}
+	for i, x := range [16]int{
+		0, 2, 4, 6,
+		9, 11,
+		14, 16,
+		19, 21,
+		24, 26, 28, 30, 32, 34} {
+		v, ok := xtob(s[x], s[x+1])
+		if !ok {
+			return uuid, errors.New("invalid UUID format")
+		}
+		uuid[i] = v
+	}
+	return uuid, nil
+}
+
+// ParseBytes is like Parse, except it parses a byte slice instead of a string.
+func ParseBytes(b []byte) (UUID, error) {
+	var uuid UUID
+	switch len(b) {
+	case 36: // xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+	case 36 + 9: // urn:uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+		if !bytes.Equal(bytes.ToLower(b[:9]), []byte("urn:uuid:")) {
+			return uuid, fmt.Errorf("invalid urn prefix: %q", b[:9])
+		}
+		b = b[9:]
+	case 36 + 2: // {xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx}
+		b = b[1:]
+	case 32: // xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+		var ok bool
+		for i := 0; i < 32; i += 2 {
+			uuid[i/2], ok = xtob(b[i], b[i+1])
+			if !ok {
+				return uuid, errors.New("invalid UUID format")
+			}
+		}
+		return uuid, nil
+	default:
+		return uuid, fmt.Errorf("invalid UUID length: %d", len(b))
+	}
+	// s is now at least 36 bytes long
+	// it must be of the form  xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+	if b[8] != '-' || b[13] != '-' || b[18] != '-' || b[23] != '-' {
+		return uuid, errors.New("invalid UUID format")
+	}
+	for i, x := range [16]int{
+		0, 2, 4, 6,
+		9, 11,
+		14, 16,
+		19, 21,
+		24, 26, 28, 30, 32, 34} {
+		v, ok := xtob(b[x], b[x+1])
+		if !ok {
+			return uuid, errors.New("invalid UUID format")
+		}
+		uuid[i] = v
+	}
+	return uuid, nil
+}
+
+// MustParse is like Parse but panics if the string cannot be parsed.
+// It simplifies safe initialization of global variables holding compiled UUIDs.
+func MustParse(s string) UUID {
+	uuid, err := Parse(s)
+	if err != nil {
+		panic(`uuid: Parse(` + s + `): ` + err.Error())
+	}
+	return uuid
+}
+
+// FromBytes creates a new UUID from a byte slice. Returns an error if the slice
+// does not have a length of 16. The bytes are copied from the slice.
+func FromBytes(b []byte) (uuid UUID, err error) {
+	err = uuid.UnmarshalBinary(b)
+	return uuid, err
+}
+
+// Must returns uuid if err is nil and panics otherwise.
+func Must(uuid UUID, err error) UUID {
+	if err != nil {
+		panic(err)
+	}
+	return uuid
+}
+
+// String returns the string form of uuid, xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+// , or "" if uuid is invalid.
+func (uuid UUID) String() string {
+	var buf [36]byte
+	encodeHex(buf[:], uuid)
+	return string(buf[:])
+}
+
+// URN returns the RFC 2141 URN form of uuid,
+// urn:uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx,  or "" if uuid is invalid.
+func (uuid UUID) URN() string {
+	var buf [36 + 9]byte
+	copy(buf[:], "urn:uuid:")
+	encodeHex(buf[9:], uuid)
+	return string(buf[:])
+}
+
+func encodeHex(dst []byte, uuid UUID) {
+	hex.Encode(dst, uuid[:4])
+	dst[8] = '-'
+	hex.Encode(dst[9:13], uuid[4:6])
+	dst[13] = '-'
+	hex.Encode(dst[14:18], uuid[6:8])
+	dst[18] = '-'
+	hex.Encode(dst[19:23], uuid[8:10])
+	dst[23] = '-'
+	hex.Encode(dst[24:], uuid[10:])
+}
+
+// Variant returns the variant encoded in uuid.
+func (uuid UUID) Variant() Variant {
+	switch {
+	case (uuid[8] & 0xc0) == 0x80:
+		return RFC4122
+	case (uuid[8] & 0xe0) == 0xc0:
+		return Microsoft
+	case (uuid[8] & 0xe0) == 0xe0:
+		return Future
+	default:
+		return Reserved
+	}
+}
+
+// Version returns the version of uuid.
+func (uuid UUID) Version() Version {
+	return Version(uuid[6] >> 4)
+}
+
+func (v Version) String() string {
+	if v > 15 {
+		return fmt.Sprintf("BAD_VERSION_%d", v)
+	}
+	return fmt.Sprintf("VERSION_%d", v)
+}
+
+func (v Variant) String() string {
+	switch v {
+	case RFC4122:
+		return "RFC4122"
+	case Reserved:
+		return "Reserved"
+	case Microsoft:
+		return "Microsoft"
+	case Future:
+		return "Future"
+	case Invalid:
+		return "Invalid"
+	}
+	return fmt.Sprintf("BadVariant%d", int(v))
+}
+
+// SetRand sets the random number generator to r, which implements io.Reader.
+// If r.Read returns an error when the package requests random data then
+// a panic will be issued.
+//
+// Calling SetRand with nil sets the random number generator to the default
+// generator.
+func SetRand(r io.Reader) {
+	if r == nil {
+		rander = rand.Reader
+		return
+	}
+	rander = r
+}
diff --git a/vendor/github.com/google/uuid/version1.go b/vendor/github.com/google/uuid/version1.go
new file mode 100644
index 000000000..199a1ac65
--- /dev/null
+++ b/vendor/github.com/google/uuid/version1.go
@@ -0,0 +1,44 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import (
+	"encoding/binary"
+)
+
+// NewUUID returns a Version 1 UUID based on the current NodeID and clock
+// sequence, and the current time.  If the NodeID has not been set by SetNodeID
+// or SetNodeInterface then it will be set automatically.  If the NodeID cannot
+// be set NewUUID returns nil.  If clock sequence has not been set by
+// SetClockSequence then it will be set automatically.  If GetTime fails to
+// return the current NewUUID returns nil and an error.
+//
+// In most cases, New should be used.
+func NewUUID() (UUID, error) {
+	nodeMu.Lock()
+	if nodeID == zeroID {
+		setNodeInterface("")
+	}
+	nodeMu.Unlock()
+
+	var uuid UUID
+	now, seq, err := GetTime()
+	if err != nil {
+		return uuid, err
+	}
+
+	timeLow := uint32(now & 0xffffffff)
+	timeMid := uint16((now >> 32) & 0xffff)
+	timeHi := uint16((now >> 48) & 0x0fff)
+	timeHi |= 0x1000 // Version 1
+
+	binary.BigEndian.PutUint32(uuid[0:], timeLow)
+	binary.BigEndian.PutUint16(uuid[4:], timeMid)
+	binary.BigEndian.PutUint16(uuid[6:], timeHi)
+	binary.BigEndian.PutUint16(uuid[8:], seq)
+	copy(uuid[10:], nodeID[:])
+
+	return uuid, nil
+}
diff --git a/vendor/github.com/google/uuid/version4.go b/vendor/github.com/google/uuid/version4.go
new file mode 100644
index 000000000..84af91c9f
--- /dev/null
+++ b/vendor/github.com/google/uuid/version4.go
@@ -0,0 +1,38 @@
+// Copyright 2016 Google Inc.  All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package uuid
+
+import "io"
+
+// New creates a new random UUID or panics.  New is equivalent to
+// the expression
+//
+//    uuid.Must(uuid.NewRandom())
+func New() UUID {
+	return Must(NewRandom())
+}
+
+// NewRandom returns a Random (Version 4) UUID.
+//
+// The strength of the UUIDs is based on the strength of the crypto/rand
+// package.
+//
+// A note about uniqueness derived from the UUID Wikipedia entry:
+//
+//  Randomly generated UUIDs have 122 random bits.  One's annual risk of being
+//  hit by a meteorite is estimated to be one chance in 17 billion, that
+//  means the probability is about 0.00000000006 (6 × 10−11),
+//  equivalent to the odds of creating a few tens of trillions of UUIDs in a
+//  year and having one duplicate.
+func NewRandom() (UUID, error) {
+	var uuid UUID
+	_, err := io.ReadFull(rander, uuid[:])
+	if err != nil {
+		return Nil, err
+	}
+	uuid[6] = (uuid[6] & 0x0f) | 0x40 // Version 4
+	uuid[8] = (uuid[8] & 0x3f) | 0x80 // Variant is 10
+	return uuid, nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 968954876..2065828a1 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -16,7 +16,7 @@ github.com/VictoriaMetrics/fasthttp/fasthttputil
 github.com/VictoriaMetrics/fasthttp/stackless
 # github.com/VictoriaMetrics/metrics v1.11.3
 github.com/VictoriaMetrics/metrics
-# github.com/VictoriaMetrics/metricsql v0.2.3
+# github.com/VictoriaMetrics/metricsql v0.2.4
 github.com/VictoriaMetrics/metricsql
 github.com/VictoriaMetrics/metricsql/binaryop
 # github.com/aws/aws-sdk-go v1.32.13
@@ -80,6 +80,8 @@ github.com/golang/protobuf/ptypes/duration
 github.com/golang/protobuf/ptypes/timestamp
 # github.com/golang/snappy v0.0.1
 github.com/golang/snappy
+# github.com/google/uuid v1.1.1
+github.com/google/uuid
 # github.com/googleapis/gax-go/v2 v2.0.5
 github.com/googleapis/gax-go/v2
 # github.com/jmespath/go-jmespath v0.3.0


Test patch:
diff --git a/app/victoria-metrics/main_test.go b/app/victoria-metrics/main_test.go
index 7f13ebfc7..6eb2c13c9 100644
--- a/app/victoria-metrics/main_test.go
+++ b/app/victoria-metrics/main_test.go
@@ -373,7 +373,7 @@ func checkMetricsResult(got, want []Metric) error {
 		want = removeIfFoundMetrics(r, want)
 	}
 	if len(want) > 0 {
-		return fmt.Errorf("exptected metrics %+v not found in %+v", want, got)
+		return fmt.Errorf("expected metrics %+v not found in %+v", want, got)
 	}
 	return nil
 }
diff --git a/app/victoria-metrics/testdata/graphite/max_lookback_set.json b/app/victoria-metrics/testdata/graphite/max_lookback_set.json
index 23ca9c947..b477d6cff 100644
--- a/app/victoria-metrics/testdata/graphite/max_lookback_set.json
+++ b/app/victoria-metrics/testdata/graphite/max_lookback_set.json
@@ -16,6 +16,8 @@
 	      ["{TIME_S-120s}","3"],
 	      ["{TIME_S-60s}","2"],
 	      ["{TIME_S-30s}","1"],
-	      ["{TIME_S-20s}","1"]
+	      ["{TIME_S-20s}","1"],
+	      ["{TIME_S-10s}","1"],
+	      ["{TIME_S-0s}","1"]
       ]}]}}
 }
diff --git a/app/vmalert/config/config_test.go b/app/vmalert/config/config_test.go
index 793a9f7c7..9e653480b 100644
--- a/app/vmalert/config/config_test.go
+++ b/app/vmalert/config/config_test.go
@@ -70,13 +70,13 @@ func TestParseBad(t *testing.T) {
 
 func TestRule_Validate(t *testing.T) {
 	if err := (&Rule{}).Validate(); err == nil {
-		t.Errorf("exptected empty name error")
+		t.Errorf("expected empty name error")
 	}
 	if err := (&Rule{Alert: "alert"}).Validate(); err == nil {
-		t.Errorf("exptected empty expr error")
+		t.Errorf("expected empty expr error")
 	}
 	if err := (&Rule{Alert: "alert", Expr: "test>0"}).Validate(); err != nil {
-		t.Errorf("exptected valid rule; got %s", err)
+		t.Errorf("expected valid rule; got %s", err)
 	}
 }
 
diff --git a/app/vmalert/datasource/vm_test.go b/app/vmalert/datasource/vm_test.go
index cc1382815..708f76a04 100644
--- a/app/vmalert/datasource/vm_test.go
+++ b/app/vmalert/datasource/vm_test.go
@@ -29,7 +29,7 @@ func TestVMSelectQuery(t *testing.T) {
 			t.Errorf("expected %s:%s as basic auth got %s:%s", basicAuthName, basicAuthPass, name, pass)
 		}
 		if r.URL.Query().Get("query") != query {
-			t.Errorf("exptected %s in query param, got %s", query, r.URL.Query().Get("query"))
+			t.Errorf("expected %s in query param, got %s", query, r.URL.Query().Get("query"))
 		}
 		switch c {
 		case 0:
@@ -76,7 +76,7 @@ func TestVMSelectQuery(t *testing.T) {
 		t.Fatalf("unexpected %s", err)
 	}
 	if len(m) != 1 {
-		t.Fatalf("exptected 1 metric  got %d in %+v", len(m), m)
+		t.Fatalf("expected 1 metric  got %d in %+v", len(m), m)
 	}
 	expected := Metric{
 		Labels:    []Label{{Value: "vm_rows", Name: "__name__"}},
diff --git a/app/vmalert/main_test.go b/app/vmalert/main_test.go
index 853cdcf3f..3dc45c0c7 100644
--- a/app/vmalert/main_test.go
+++ b/app/vmalert/main_test.go
@@ -41,7 +41,7 @@ func TestGetAlertURLGenerator(t *testing.T) {
 	}
 	_, err = getAlertURLGenerator(nil, "foo?{{invalid}}", true)
 	if err == nil {
-		t.Errorf("exptected tempalte validation error got nil")
+		t.Errorf("expected tempalte validation error got nil")
 	}
 	fn, err = getAlertURLGenerator(u, "foo?query={{$value}}", true)
 	if err != nil {
diff --git a/app/vmalert/notifier/alertmanager_test.go b/app/vmalert/notifier/alertmanager_test.go
index 18f2efcf4..f7bdb47d4 100644
--- a/app/vmalert/notifier/alertmanager_test.go
+++ b/app/vmalert/notifier/alertmanager_test.go
@@ -52,16 +52,16 @@ func TestAlertManager_Send(t *testing.T) {
 				t.Errorf("expected 1 alert in array got %d", len(a))
 			}
 			if a[0].GeneratorURL != "0/0" {
-				t.Errorf("exptected 0/0 as generatorURL got %s", a[0].GeneratorURL)
+				t.Errorf("expected 0/0 as generatorURL got %s", a[0].GeneratorURL)
 			}
 			if a[0].Labels["alertname"] != "alert0" {
-				t.Errorf("exptected alert0 as alert name got %s", a[0].Labels["alertname"])
+				t.Errorf("expected alert0 as alert name got %s", a[0].Labels["alertname"])
 			}
 			if a[0].StartsAt.IsZero() {
-				t.Errorf("exptected non-zero start time")
+				t.Errorf("expected non-zero start time")
 			}
 			if a[0].EndAt.IsZero() {
-				t.Errorf("exptected non-zero end time")
+				t.Errorf("expected non-zero end time")
 			}
 		}
 	})
diff --git a/app/vmalert/remotewrite/remotewrite_test.go b/app/vmalert/remotewrite/remotewrite_test.go
new file mode 100644
index 000000000..7dab24712
--- /dev/null
+++ b/app/vmalert/remotewrite/remotewrite_test.go
@@ -0,0 +1,102 @@
+package remotewrite
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/golang/snappy"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+)
+
+func TestClient_Push(t *testing.T) {
+	testSrv := newRWServer()
+	cfg := Config{
+		Addr:         testSrv.URL,
+		MaxBatchSize: 100,
+	}
+	client, err := NewClient(context.Background(), cfg)
+	if err != nil {
+		t.Fatalf("failed to create client: %s", err)
+	}
+	const rowsN = 1e4
+	var sent int
+	for i := 0; i < rowsN; i++ {
+		s := prompbmarshal.TimeSeries{
+			Samples: []prompbmarshal.Sample{{
+				Value:     rand.Float64(),
+				Timestamp: time.Now().Unix(),
+			}},
+		}
+		err := client.Push(s)
+		if err == nil {
+			sent++
+		}
+	}
+	if sent == 0 {
+		t.Fatalf("0 series sent")
+	}
+	if err := client.Close(); err != nil {
+		t.Fatalf("failed to close client: %s", err)
+	}
+	got := testSrv.accepted()
+	if got != sent {
+		t.Fatalf("expected to have %d series; got %d", sent, got)
+	}
+}
+
+func newRWServer() *rwServer {
+	rw := &rwServer{}
+	rw.Server = httptest.NewServer(http.HandlerFunc(rw.handler))
+	return rw
+}
+
+type rwServer struct {
+	// WARN: ordering of fields is important for alignment!
+	// see https://golang.org/pkg/sync/atomic/#pkg-note-BUG
+	acceptedRows uint64
+	*httptest.Server
+}
+
+func (rw *rwServer) accepted() int {
+	return int(atomic.LoadUint64(&rw.acceptedRows))
+}
+
+func (rw *rwServer) err(w http.ResponseWriter, err error) {
+	w.WriteHeader(http.StatusBadRequest)
+	w.Write([]byte(err.Error()))
+}
+
+func (rw *rwServer) handler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		rw.err(w, fmt.Errorf("bad method %q", r.Method))
+		return
+	}
+	data, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		rw.err(w, fmt.Errorf("body read err: %s", err))
+		return
+	}
+	defer func() { _ = r.Body.Close() }()
+
+	b, err := snappy.Decode(nil, data)
+	if err != nil {
+		rw.err(w, fmt.Errorf("decode err: %s", err))
+		return
+	}
+	wr := &prompb.WriteRequest{}
+	if err := wr.Unmarshal(b); err != nil {
+		rw.err(w, fmt.Errorf("unmarhsal err: %s", err))
+		return
+	}
+	atomic.AddUint64(&rw.acceptedRows, uint64(len(wr.Timeseries)))
+	w.WriteHeader(http.StatusNoContent)
+}
diff --git a/app/vmselect/prometheus/prometheus_test.go b/app/vmselect/prometheus/prometheus_test.go
index 69c2c1237..32bfdd585 100644
--- a/app/vmselect/prometheus/prometheus_test.go
+++ b/app/vmselect/prometheus/prometheus_test.go
@@ -113,3 +113,130 @@ func TestGetTimeError(t *testing.T) {
 	f("-292273086-05-16T16:47:07Z")
 	f("292277025-08-18T07:12:54.999999998Z")
 }
+
+func TestAdjustLastPoints(t *testing.T) {
+	f := func(tss []netstorage.Result, start, end int64, tssExpected []netstorage.Result) {
+		t.Helper()
+		tss = adjustLastPoints(tss, start, end)
+		for i, ts := range tss {
+			for j, value := range ts.Values {
+				expectedValue := tssExpected[i].Values[j]
+				if math.IsNaN(expectedValue) {
+					if !math.IsNaN(value) {
+						t.Fatalf("unexpected result; got %v; want nan", value)
+					}
+				} else if expectedValue != value {
+					t.Fatalf("unexpected result; got %v; want %v", value, expectedValue)
+				}
+			}
+			if !reflect.DeepEqual(ts.Timestamps, tssExpected[i].Timestamps) {
+				t.Fatalf("unexpected result; got %v; want %v", tss, tssExpected)
+			}
+		}
+
+	}
+
+	nan := math.NaN()
+
+	f(nil, 300, 500, nil)
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+	}, 400, 500, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, 4},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+	}, 300, 500, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 3, 3},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, nan, nan, nan, nan},
+		},
+	}, 500, 300, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, nan, nan, nan, nan},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400},
+			Values:     []float64{1, 2, 3, 4},
+		},
+	}, 400, 500, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, 4},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400},
+			Values:     []float64{1, 2, 3, 4},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300},
+			Values:     []float64{1, 2, nan},
+		},
+	}, 300, 600, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 3, 3},
+		},
+		{
+			Timestamps: []int64{100, 200, 300},
+			Values:     []float64{1, 2, nan},
+		},
+	})
+}
diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index a7c015d9b..236ff35ef 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -4103,6 +4103,44 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1}
 		f(q, resultExpected)
 	})
+	t.Run(`interpolate()`, func(t *testing.T) {
+		t.Parallel()
+		q := `interpolate(label_set(time() < 1300 default time() > 1700, "__name__", "foobar", "x", "y"))`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{1000, 1200, 1400, 1600, 1800, 2000},
+			Timestamps: timestampsExpected,
+		}
+		r1.MetricName.MetricGroup = []byte("foobar")
+		r1.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("x"),
+			Value: []byte("y"),
+		}}
+		resultExpected := []netstorage.Result{r1}
+		f(q, resultExpected)
+	})
+	t.Run(`interpolate(tail)`, func(t *testing.T) {
+		t.Parallel()
+		q := `interpolate(time() < 1300)`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{1000, 1200, 1200, 1200, 1200, 1200},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r1}
+		f(q, resultExpected)
+	})
+	t.Run(`interpolate(head)`, func(t *testing.T) {
+		t.Parallel()
+		q := `interpolate(time() > 1500)`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{1600, 1600, 1600, 1600, 1800, 2000},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r1}
+		f(q, resultExpected)
+	})
 	t.Run(`distinct_over_time([500s])`, func(t *testing.T) {
 		t.Parallel()
 		q := `distinct_over_time((time() < 1700)[500s])`
@@ -5595,6 +5633,7 @@ func TestExecError(t *testing.T) {
 	f(`median("foo", "bar")`)
 	f(`keep_last_value()`)
 	f(`keep_next_value()`)
+	f(`interpolate()`)
 	f(`distinct_over_time()`)
 	f(`distinct()`)
 	f(`alias()`)
diff --git a/lib/promscrape/config_test.go b/lib/promscrape/config_test.go
index e3c3e1be2..6310b09eb 100644
--- a/lib/promscrape/config_test.go
+++ b/lib/promscrape/config_test.go
@@ -1201,6 +1201,9 @@ scrape_configs:
 	f(`
 scrape_configs:
   - job_name: 'snmp'
+    sample_limit: 100
+    disable_keepalive: true
+    disable_compression: true
     static_configs:
       - targets:
         - 192.168.1.2  # SNMP device.
@@ -1249,8 +1252,11 @@ scrape_configs:
 					Value: "snmp",
 				},
 			},
-			AuthConfig:      &promauth.Config{},
-			jobNameOriginal: "snmp",
+			AuthConfig:         &promauth.Config{},
+			SampleLimit:        100,
+			DisableKeepAlive:   true,
+			DisableCompression: true,
+			jobNameOriginal:    "snmp",
 		},
 	})
 }
diff --git a/lib/protoparser/graphite/parser_test.go b/lib/protoparser/graphite/parser_test.go
index ca8bcccb1..6c299d0f7 100644
--- a/lib/protoparser/graphite/parser_test.go
+++ b/lib/protoparser/graphite/parser_test.go
@@ -1,7 +1,9 @@
 package graphite
 
 import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"reflect"
+	"strings"
 	"testing"
 )
 
@@ -84,6 +86,13 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 			Value:  1123,
 		}},
 	})
+	f("aaa 1123 -1", &Rows{
+		Rows: []Row{{
+			Metric:    "aaa",
+			Value:     1123,
+			Timestamp: -1,
+		}},
+	})
 
 	// Timestamp bigger than 1<<31
 	f("aaa 1123 429496729600", &Rows{
@@ -161,3 +170,56 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 		},
 	})
 }
+
+func Test_streamContext_Read(t *testing.T) {
+	f := func(s string, rowsExpected *Rows) {
+		t.Helper()
+		ctx := &streamContext{}
+		ctx.Read(strings.NewReader(s))
+		if len(ctx.Rows.Rows) != len(rowsExpected.Rows) {
+			t.Fatalf("different len of expected rows;\ngot\n%+v;\nwant\n%+v", ctx.Rows, rowsExpected.Rows)
+		}
+		if !reflect.DeepEqual(ctx.Rows.Rows, rowsExpected.Rows) {
+			t.Fatalf("unexpected rows;\ngot\n%+v;\nwant\n%+v", ctx.Rows.Rows, rowsExpected.Rows)
+		}
+	}
+
+	// Full line without tags
+	f("aaa 1123 345", &Rows{
+		Rows: []Row{{
+			Metric:    "aaa",
+			Value:     1123,
+			Timestamp: 345 * 1000,
+		}},
+	})
+	// Full line with tags
+	f("aaa;x=y 1123 345", &Rows{
+		Rows: []Row{{
+			Metric: "aaa",
+			Tags: []Tag{{
+				Key:   "x",
+				Value: "y",
+			}},
+			Value:     1123,
+			Timestamp: 345 * 1000,
+		}},
+	})
+	// missing timestamp.
+	// Note that this test may be flaky due to timing issues. TODO: fix it
+	f("aaa 1123", &Rows{
+		Rows: []Row{{
+			Metric:    "aaa",
+			Value:     1123,
+			Timestamp: int64(fasttime.UnixTimestamp()) * 1000,
+		}},
+	})
+	// -1 timestamp. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/610
+	// Note that this test may be flaky due to timing issues. TODO: fix it.
+	f("aaa 1123 -1", &Rows{
+		Rows: []Row{{
+			Metric:    "aaa",
+			Value:     1123,
+			Timestamp: int64(fasttime.UnixTimestamp()) * 1000,
+		}},
+	})
+}