Repository: VictoriaMetrics/VictoriaMetrics
Pull request #761 (2020-09-10 19:39:00)

vmalert: add Group name as label to generated alerts and timeseries #611

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

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

```diff
@@            Coverage Diff             @@
##           master     #761      +/-   ##
==========================================
+ Coverage   68.16%   68.19%   +0.03%     
==========================================
  Files         164      164              
  Lines       22265    22266       +1     
==========================================
+ Hits        15176    15185       +9     
+ Misses       5850     5844       -6     
+ Partials     1239     1237       -2     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [app/vmalert/alerting.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvYWxlcnRpbmcuZ28=) | `75.75% <100.00%> (+0.10%)` | :arrow_up: |
| [lib/storage/merge.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvbWVyZ2UuZ28=) | `82.64% <0.00%> (-1.66%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `85.62% <0.00%> (-1.31%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `67.78% <0.00%> (-0.28%)` | :arrow_down: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `71.44% <0.00%> (+0.23%)` | :arrow_up: |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `75.38% <0.00%> (+0.53%)` | :arrow_up: |
| [lib/storage/part.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydC5nbw==) | `74.48% <0.00%> (+1.37%)` | :arrow_up: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/761/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `62.71% <0.00%> (+1.69%)` | :arrow_up: |

------

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


Patch:
diff --git a/README.md b/README.md
index 223b987fb..6f9421ca7 100644
--- a/README.md
+++ b/README.md
@@ -551,6 +551,9 @@ It should return something like the following:
 Extra labels may be added to all the imported metrics by passing `extra_label=name=value` query args.
 For example, `/api/v1/import/prometheus?extra_label=foo=bar` would add `{foo="bar"}` label to all the imported metrics.
 
+If timestamp is missing in `<metric> <value> <timestamp>` Prometheus exposition format line, then the current timestamp is used during data ingestion.
+It can be overriden by passing unix timestamp in *milliseconds* via `timestamp` query arg. For example, `/api/v1/import/prometheus?timestamp=1594370496905`.
+
 VictoriaMetrics accepts arbitrary number of lines in a single request to `/api/v1/import/prometheus`, i.e. it supports data streaming.
 
 VictoriaMetrics also may scrape Prometheus targets - see [these docs](#how-to-scrape-prometheus-exporters-such-as-node-exporter).
diff --git a/app/vmagent/prometheusimport/request_handler.go b/app/vmagent/prometheusimport/request_handler.go
index b5ef3b833..aac41d944 100644
--- a/app/vmagent/prometheusimport/request_handler.go
+++ b/app/vmagent/prometheusimport/request_handler.go
@@ -23,9 +23,13 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
+	defaultTimestamp, err := parserCommon.GetTimestamp(req)
+	if err != nil {
+		return err
+	}
 	return writeconcurrencylimiter.Do(func() error {
 		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		return parser.ParseStream(req.Body, isGzipped, func(rows []parser.Row) error {
+		return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
 			return insertRows(rows, extraLabels)
 		})
 	})
diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go
index 2acb2d457..498093d2d 100644
--- a/app/vmagent/remotewrite/remotewrite.go
+++ b/app/vmagent/remotewrite/remotewrite.go
@@ -85,7 +85,7 @@ func Init() {
 	for i, remoteWriteURL := range *remoteWriteURLs {
 		urlLabelValue := fmt.Sprintf("secret-url-%d", i+1)
 		if *showRemoteWriteURL {
-			urlLabelValue = remoteWriteURL
+			urlLabelValue = fmt.Sprintf("%d:%s", i+1, remoteWriteURL)
 		}
 		rwctx := newRemoteWriteCtx(i, remoteWriteURL, maxInmemoryBlocks, urlLabelValue)
 		rwctxs = append(rwctxs, rwctx)
@@ -192,7 +192,7 @@ type remoteWriteCtx struct {
 
 func newRemoteWriteCtx(argIdx int, remoteWriteURL string, maxInmemoryBlocks int, urlLabelValue string) *remoteWriteCtx {
 	h := xxhash.Sum64([]byte(remoteWriteURL))
-	path := fmt.Sprintf("%s/persistent-queue/%016X", *tmpDataPath, h)
+	path := fmt.Sprintf("%s/persistent-queue/%d_%016X", *tmpDataPath, argIdx+1, h)
 	fq := persistentqueue.MustOpenFastQueue(path, remoteWriteURL, maxInmemoryBlocks, maxPendingBytesPerURL.N)
 	_ = metrics.GetOrCreateGauge(fmt.Sprintf(`vmagent_remotewrite_pending_data_bytes{path=%q, url=%q}`, path, urlLabelValue), func() float64 {
 		return float64(fq.GetPendingBytes())
diff --git a/app/vmalert/alerting.go b/app/vmalert/alerting.go
index 11538e829..08960f031 100644
--- a/app/vmalert/alerting.go
+++ b/app/vmalert/alerting.go
@@ -26,6 +26,7 @@ type AlertingRule struct {
 	Labels      map[string]string
 	Annotations map[string]string
 	GroupID     uint64
+	GroupName   string
 
 	// guard status fields
 	mu sync.RWMutex
@@ -56,6 +57,7 @@ func newAlertingRule(group *Group, cfg config.Rule) *AlertingRule {
 		Labels:      cfg.Labels,
 		Annotations: cfg.Annotations,
 		GroupID:     group.ID(),
+		GroupName:   group.Name,
 		alerts:      make(map[uint64]*notifier.Alert),
 		metrics:     &alertingRuleMetrics{},
 	}
@@ -242,6 +244,9 @@ func (ar *AlertingRule) newAlert(m datasource.Metric, start time.Time) (*notifie
 		Start:   start,
 		Expr:    ar.Expr,
 	}
+	// label defined here to make override possible by
+	// time series labels.
+	a.Labels[alertGroupNameLabel] = ar.GroupName
 	for _, l := range m.Labels {
 		// drop __name__ to be consistent with Prometheus alerting
 		if l.Name == "__name__" {
@@ -336,15 +341,18 @@ func (ar *AlertingRule) newAlertAPI(a notifier.Alert) *APIAlert {
 }
 
 const (
-	// AlertMetricName is the metric name for synthetic alert timeseries.
+	// alertMetricName is the metric name for synthetic alert timeseries.
 	alertMetricName = "ALERTS"
-	// AlertForStateMetricName is the metric name for 'for' state of alert.
+	// alertForStateMetricName is the metric name for 'for' state of alert.
 	alertForStateMetricName = "ALERTS_FOR_STATE"
 
-	// AlertNameLabel is the label name indicating the name of an alert.
+	// alertNameLabel is the label name indicating the name of an alert.
 	alertNameLabel = "alertname"
-	// AlertStateLabel is the label name indicating the state of an alert.
+	// alertStateLabel is the label name indicating the state of an alert.
 	alertStateLabel = "alertstate"
+
+	// alertGroupNameLabel defines the label name attached for generated time series.
+	alertGroupNameLabel = "alertgroup"
 )
 
 // alertToTimeSeries converts the given alert with the given timestamp to timeseries
diff --git a/app/vmalert/config/config.go b/app/vmalert/config/config.go
index 6c38d9e67..f027dad59 100644
--- a/app/vmalert/config/config.go
+++ b/app/vmalert/config/config.go
@@ -1,6 +1,7 @@
 package config
 
 import (
+	"crypto/md5"
 	"fmt"
 	"hash/fnv"
 	"io/ioutil"
@@ -24,11 +25,30 @@ type Group struct {
 	Interval    time.Duration `yaml:"interval,omitempty"`
 	Rules       []Rule        `yaml:"rules"`
 	Concurrency int           `yaml:"concurrency"`
+	// Checksum stores the hash of yaml definition for this group.
+	// May be used to detect any changes like rules re-ordering etc.
+	Checksum string
 
 	// Catches all undefined fields and must be empty after parsing.
 	XXX map[string]interface{} `yaml:",inline"`
 }
 
+// UnmarshalYAML implements the yaml.Unmarshaler interface.
+func (g *Group) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	type group Group
+	if err := unmarshal((*group)(g)); err != nil {
+		return err
+	}
+	b, err := yaml.Marshal(g)
+	if err != nil {
+		return fmt.Errorf("failed to marshal group configuration for checksum: %s", err)
+	}
+	h := md5.New()
+	h.Write(b)
+	g.Checksum = fmt.Sprintf("%x", h.Sum(nil))
+	return nil
+}
+
 // Validate check for internal Group or Rule configuration errors
 func (g *Group) Validate(validateAnnotations, validateExpressions bool) error {
 	if g.Name == "" {
@@ -101,7 +121,7 @@ func (r *Rule) Name() string {
 }
 
 // HashRule hashes significant Rule fields into
-// unique hash value
+// unique hash that supposed to define Rule uniqueness
 func HashRule(r Rule) uint64 {
 	h := fnv.New64a()
 	h.Write([]byte(r.Expr))
@@ -112,16 +132,7 @@ func HashRule(r Rule) uint64 {
 		h.Write([]byte("alerting"))
 		h.Write([]byte(r.Alert))
 	}
-	type item struct {
-		key, value string
-	}
-	var kv []item
-	for k, v := range r.Labels {
-		kv = append(kv, item{key: k, value: v})
-	}
-	sort.Slice(kv, func(i, j int) bool {
-		return kv[i].key < kv[j].key
-	})
+	kv := sortMap(r.Labels)
 	for _, i := range kv {
 		h.Write([]byte(i.key))
 		h.Write([]byte(i.value))
@@ -204,3 +215,18 @@ func checkOverflow(m map[string]interface{}, ctx string) error {
 	}
 	return nil
 }
+
+type item struct {
+	key, value string
+}
+
+func sortMap(m map[string]string) []item {
+	var kv []item
+	for k, v := range m {
+		kv = append(kv, item{key: k, value: v})
+	}
+	sort.Slice(kv, func(i, j int) bool {
+		return kv[i].key < kv[j].key
+	})
+	return kv
+}
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index e94dec112..f2a1b5bd7 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -24,6 +24,7 @@ type Group struct {
 	Rules       []Rule
 	Interval    time.Duration
 	Concurrency int
+	Checksum    string
 
 	doneCh     chan struct{}
 	finishedCh chan struct{}
@@ -53,6 +54,7 @@ func newGroup(cfg config.Group, defaultInterval time.Duration, labels map[string
 		File:        cfg.File,
 		Interval:    cfg.Interval,
 		Concurrency: cfg.Concurrency,
+		Checksum:    cfg.Checksum,
 		doneCh:      make(chan struct{}),
 		finishedCh:  make(chan struct{}),
 		updateCh:    make(chan *Group),
@@ -156,6 +158,7 @@ func (g *Group) updateWith(newGroup *Group) error {
 		newRules = append(newRules, nr)
 	}
 	g.Concurrency = newGroup.Concurrency
+	g.Checksum = newGroup.Checksum
 	g.Rules = newRules
 	return nil
 }
diff --git a/app/vmalert/manager.go b/app/vmalert/manager.go
index 2c749b830..3abdd1340 100644
--- a/app/vmalert/manager.go
+++ b/app/vmalert/manager.go
@@ -93,31 +93,53 @@ func (m *manager) update(ctx context.Context, path []string, validateTpl, valida
 		groupsRegistry[ng.ID()] = ng
 	}
 
+	type updateItem struct {
+		old *Group
+		new *Group
+	}
+	var toUpdate []updateItem
+
 	m.groupsMu.Lock()
 	for _, og := range m.groups {
 		ng, ok := groupsRegistry[og.ID()]
 		if !ok {
-			// old group is not present in new list
-			// and must be stopped and deleted
+			// old group is not present in new list,
+			// so must be stopped and deleted
 			og.close()
 			delete(m.groups, og.ID())
 			og = nil
 			continue
 		}
-		og.updateCh <- ng
 		delete(groupsRegistry, ng.ID())
+		if og.Checksum != ng.Checksum {
+			toUpdate = append(toUpdate, updateItem{old: og, new: ng})
+		}
 	}
-
 	for _, ng := range groupsRegistry {
 		m.startGroup(ctx, ng, restore)
 	}
 	m.groupsMu.Unlock()
+
+	if len(toUpdate) > 0 {
+		var wg sync.WaitGroup
+		for _, item := range toUpdate {
+			wg.Add(1)
+			go func(old *Group, new *Group) {
+				old.updateCh <- new
+				wg.Done()
+			}(item.old, item.new)
+		}
+		wg.Wait()
+	}
 	return nil
 }
 
 func (g *Group) toAPI() APIGroup {
+	g.mu.RLock()
+	defer g.mu.RUnlock()
+
 	ag := APIGroup{
-		// encode as strings to avoid rounding
+		// encode as string to avoid rounding
 		ID:          fmt.Sprintf("%d", g.ID()),
 		Name:        g.Name,
 		File:        g.File,
diff --git a/app/vminsert/prometheusimport/request_handler.go b/app/vminsert/prometheusimport/request_handler.go
index fa7d1068a..c499600cd 100644
--- a/app/vminsert/prometheusimport/request_handler.go
+++ b/app/vminsert/prometheusimport/request_handler.go
@@ -23,9 +23,13 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
+	defaultTimestamp, err := parserCommon.GetTimestamp(req)
+	if err != nil {
+		return err
+	}
 	return writeconcurrencylimiter.Do(func() error {
 		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-		return parser.ParseStream(req.Body, isGzipped, func(rows []parser.Row) error {
+		return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
 			return insertRows(rows, extraLabels)
 		})
 	})
diff --git a/app/vmselect/graphite/graphite.go b/app/vmselect/graphite/graphite.go
index 4c36d7487..25c63f6eb 100644
--- a/app/vmselect/graphite/graphite.go
+++ b/app/vmselect/graphite/graphite.go
@@ -197,7 +197,7 @@ func MetricsIndexHandler(startTime time.Time, w http.ResponseWriter, r *http.Req
 }
 
 // metricsFind searches for label values that match the given query.
-func metricsFind(tr storage.TimeRange, label, query string, delimiter byte, deadline netstorage.Deadline) ([]string, error) {
+func metricsFind(tr storage.TimeRange, label, query string, delimiter byte, deadline searchutils.Deadline) ([]string, error) {
 	expandTail := strings.HasSuffix(query, "*")
 	for strings.HasSuffix(query, "*") {
 		query = query[:len(query)-1]
diff --git a/app/vmselect/netstorage/netstorage.go b/app/vmselect/netstorage/netstorage.go
index 802d800c6..63fa24ae1 100644
--- a/app/vmselect/netstorage/netstorage.go
+++ b/app/vmselect/netstorage/netstorage.go
@@ -8,8 +8,8 @@ import (
 	"runtime"
 	"sort"
 	"sync"
-	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
@@ -53,7 +53,7 @@ func (r *Result) reset() {
 type Results struct {
 	tr        storage.TimeRange
 	fetchData bool
-	deadline  Deadline
+	deadline  searchutils.Deadline
 
 	packedTimeseries []packedTimeseries
 	sr               *storage.Search
@@ -458,11 +458,11 @@ func DeleteSeries(sq *storage.SearchQuery) (int, error) {
 }
 
 // GetLabels returns labels until the given deadline.
-func GetLabels(deadline Deadline) ([]string, error) {
+func GetLabels(deadline searchutils.Deadline) ([]string, error) {
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	labels, err := vmstorage.SearchTagKeys(*maxTagKeysPerSearch, deadline.deadline)
+	labels, err := vmstorage.SearchTagKeys(*maxTagKeysPerSearch, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during labels search: %w", err)
 	}
@@ -482,7 +482,7 @@ func GetLabels(deadline Deadline) ([]string, error) {
 
 // GetLabelValues returns label values for the given labelName
 // until the given deadline.
-func GetLabelValues(labelName string, deadline Deadline) ([]string, error) {
+func GetLabelValues(labelName string, deadline searchutils.Deadline) ([]string, error) {
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
@@ -491,7 +491,7 @@ func GetLabelValues(labelName string, deadline Deadline) ([]string, error) {
 	}
 
 	// Search for tag values
-	labelValues, err := vmstorage.SearchTagValues([]byte(labelName), *maxTagValuesPerSearch, deadline.deadline)
+	labelValues, err := vmstorage.SearchTagValues([]byte(labelName), *maxTagValuesPerSearch, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during label values search for labelName=%q: %w", labelName, err)
 	}
@@ -505,11 +505,11 @@ func GetLabelValues(labelName string, deadline Deadline) ([]string, error) {
 // GetTagValueSuffixes returns tag value suffixes for the given tagKey and the given tagValuePrefix.
 //
 // It can be used for implementing https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find
-func GetTagValueSuffixes(tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, deadline Deadline) ([]string, error) {
+func GetTagValueSuffixes(tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, deadline searchutils.Deadline) ([]string, error) {
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	suffixes, err := vmstorage.SearchTagValueSuffixes(tr, []byte(tagKey), []byte(tagValuePrefix), delimiter, *maxTagValueSuffixesPerSearch, deadline.deadline)
+	suffixes, err := vmstorage.SearchTagValueSuffixes(tr, []byte(tagKey), []byte(tagValuePrefix), delimiter, *maxTagValueSuffixesPerSearch, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during search for suffixes for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s: %w",
 			tagKey, tagValuePrefix, delimiter, tr.String(), err)
@@ -518,11 +518,11 @@ func GetTagValueSuffixes(tr storage.TimeRange, tagKey, tagValuePrefix string, de
 }
 
 // GetLabelEntries returns all the label entries until the given deadline.
-func GetLabelEntries(deadline Deadline) ([]storage.TagEntry, error) {
+func GetLabelEntries(deadline searchutils.Deadline) ([]storage.TagEntry, error) {
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	labelEntries, err := vmstorage.SearchTagEntries(*maxTagKeysPerSearch, *maxTagValuesPerSearch, deadline.deadline)
+	labelEntries, err := vmstorage.SearchTagEntries(*maxTagKeysPerSearch, *maxTagValuesPerSearch, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during label entries request: %w", err)
 	}
@@ -548,11 +548,11 @@ func GetLabelEntries(deadline Deadline) ([]storage.TagEntry, error) {
 }
 
 // GetTSDBStatusForDate returns tsdb status according to https://prometheus.io/docs/prometheus/latest/querying/api/#tsdb-stats
-func GetTSDBStatusForDate(deadline Deadline, date uint64, topN int) (*storage.TSDBStatus, error) {
+func GetTSDBStatusForDate(deadline searchutils.Deadline, date uint64, topN int) (*storage.TSDBStatus, error) {
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	status, err := vmstorage.GetTSDBStatusForDate(date, topN, deadline.deadline)
+	status, err := vmstorage.GetTSDBStatusForDate(date, topN, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during tsdb status request: %w", err)
 	}
@@ -560,11 +560,11 @@ func GetTSDBStatusForDate(deadline Deadline, date uint64, topN int) (*storage.TS
 }
 
 // GetSeriesCount returns the number of unique series.
-func GetSeriesCount(deadline Deadline) (uint64, error) {
+func GetSeriesCount(deadline searchutils.Deadline) (uint64, error) {
 	if deadline.Exceeded() {
 		return 0, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	n, err := vmstorage.GetSeriesCount(deadline.deadline)
+	n, err := vmstorage.GetSeriesCount(deadline.Deadline())
 	if err != nil {
 		return 0, fmt.Errorf("error during series count request: %w", err)
 	}
@@ -589,7 +589,7 @@ var ssPool sync.Pool
 // ProcessSearchQuery performs sq on storage nodes until the given deadline.
 //
 // Results.RunParallel or Results.Cancel must be called on the returned Results.
-func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadline) (*Results, error) {
+func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline searchutils.Deadline) (*Results, error) {
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
@@ -611,7 +611,7 @@ func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadli
 	defer vmstorage.WG.Done()
 
 	sr := getStorageSearch()
-	maxSeriesCount := sr.Init(vmstorage.Storage, tfss, tr, *maxMetricsPerSearch, deadline.deadline)
+	maxSeriesCount := sr.Init(vmstorage.Storage, tfss, tr, *maxMetricsPerSearch, deadline.Deadline())
 
 	m := make(map[string][]storage.BlockRef, maxSeriesCount)
 	orderedMetricNames := make([]string, 0, maxSeriesCount)
@@ -672,33 +672,3 @@ func setupTfss(tagFilterss [][]storage.TagFilter) ([]*storage.TagFilters, error)
 	}
 	return tfss, nil
 }
-
-// Deadline contains deadline with the corresponding timeout for pretty error messages.
-type Deadline struct {
-	deadline uint64
-
-	timeout  time.Duration
-	flagHint string
-}
-
-// NewDeadline returns deadline for the given timeout.
-//
-// flagHint must contain a hit for command-line flag, which could be used
-// in order to increase timeout.
-func NewDeadline(startTime time.Time, timeout time.Duration, flagHint string) Deadline {
-	return Deadline{
-		deadline: uint64(startTime.Add(timeout).Unix()),
-		timeout:  timeout,
-		flagHint: flagHint,
-	}
-}
-
-// Exceeded returns true if deadline is exceeded.
-func (d *Deadline) Exceeded() bool {
-	return fasttime.UnixTimestamp() > d.deadline
-}
-
-// String returns human-readable string representation for d.
-func (d *Deadline) String() string {
-	return fmt.Sprintf("%.3f seconds; the timeout can be adjusted with `%s` command-line flag", d.timeout.Seconds(), d.flagHint)
-}
diff --git a/app/vmselect/prometheus/prometheus.go b/app/vmselect/prometheus/prometheus.go
index d8fbbcf93..4c18563d0 100644
--- a/app/vmselect/prometheus/prometheus.go
+++ b/app/vmselect/prometheus/prometheus.go
@@ -26,7 +26,7 @@ import (
 )
 
 var (
-	latencyOffset = flag.Duration("search.latencyOffset", time.Second*30, "The time when data points become visible in query results after the colection. "+
+	latencyOffset = flag.Duration("search.latencyOffset", time.Second*30, "The time when data points become visible in query results after the collection. "+
 		"Too small value can result in incomplete last points for query results")
 	maxQueryLen = flagutil.NewBytes("search.maxQueryLen", 16*1024, "The maximum search query length in bytes")
 	maxLookback = flag.Duration("search.maxLookback", 0, "Synonim to -search.lookback-delta from Prometheus. "+
@@ -150,7 +150,7 @@ func ExportHandler(startTime time.Time, w http.ResponseWriter, r *http.Request)
 
 var exportDuration = metrics.NewSummary(`vm_request_duration_seconds{path="/api/v1/export"}`)
 
-func exportHandler(w http.ResponseWriter, matches []string, start, end int64, format string, maxRowsPerLine int, deadline netstorage.Deadline) error {
+func exportHandler(w http.ResponseWriter, matches []string, start, end int64, format string, maxRowsPerLine int, deadline searchutils.Deadline) error {
 	writeResponseFunc := WriteExportStdResponse
 	writeLineFunc := func(rs *netstorage.Result, resultsCh chan<- *quicktemplate.ByteBuffer) {
 		bb := quicktemplate.AcquireByteBuffer()
@@ -322,7 +322,7 @@ func LabelValuesHandler(startTime time.Time, labelName string, w http.ResponseWr
 	return nil
 }
 
-func labelValuesWithMatches(labelName string, matches []string, start, end int64, deadline netstorage.Deadline) ([]string, error) {
+func labelValuesWithMatches(labelName string, matches []string, start, end int64, deadline searchutils.Deadline) ([]string, error) {
 	if len(matches) == 0 {
 		logger.Panicf("BUG: matches must be non-empty")
 	}
@@ -485,7 +485,7 @@ func LabelsHandler(startTime time.Time, w http.ResponseWriter, r *http.Request)
 	return nil
 }
 
-func labelsWithMatches(matches []string, start, end int64, deadline netstorage.Deadline) ([]string, error) {
+func labelsWithMatches(matches []string, start, end int64, deadline searchutils.Deadline) ([]string, error) {
 	if len(matches) == 0 {
 		logger.Panicf("BUG: matches must be non-empty")
 	}
diff --git a/app/vmselect/promql/eval.go b/app/vmselect/promql/eval.go
index 1ab6b7c1b..2bbdb4d3f 100644
--- a/app/vmselect/promql/eval.go
+++ b/app/vmselect/promql/eval.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
@@ -88,7 +89,7 @@ type EvalConfig struct {
 	// QuotedRemoteAddr contains quoted remote address.
 	QuotedRemoteAddr string
 
-	Deadline netstorage.Deadline
+	Deadline searchutils.Deadline
 
 	MayCache bool
 
diff --git a/app/vmselect/searchutils/searchutils.go b/app/vmselect/searchutils/searchutils.go
index 123f68941..409aed7e8 100644
--- a/app/vmselect/searchutils/searchutils.go
+++ b/app/vmselect/searchutils/searchutils.go
@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/metricsql"
 )
 
@@ -97,18 +97,18 @@ func GetDuration(r *http.Request, argKey string, defaultValue int64) (int64, err
 const maxDurationMsecs = 100 * 365 * 24 * 3600 * 1000
 
 // GetDeadlineForQuery returns deadline for the given query r.
-func GetDeadlineForQuery(r *http.Request, startTime time.Time) netstorage.Deadline {
+func GetDeadlineForQuery(r *http.Request, startTime time.Time) Deadline {
 	dMax := maxQueryDuration.Milliseconds()
 	return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxQueryDuration")
 }
 
 // GetDeadlineForExport returns deadline for the given request to /api/v1/export.
-func GetDeadlineForExport(r *http.Request, startTime time.Time) netstorage.Deadline {
+func GetDeadlineForExport(r *http.Request, startTime time.Time) Deadline {
 	dMax := maxExportDuration.Milliseconds()
 	return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxExportDuration")
 }
 
-func getDeadlineWithMaxDuration(r *http.Request, startTime time.Time, dMax int64, flagHint string) netstorage.Deadline {
+func getDeadlineWithMaxDuration(r *http.Request, startTime time.Time, dMax int64, flagHint string) Deadline {
 	d, err := GetDuration(r, "timeout", 0)
 	if err != nil {
 		d = 0
@@ -117,7 +117,7 @@ func getDeadlineWithMaxDuration(r *http.Request, startTime time.Time, dMax int64
 		d = dMax
 	}
 	timeout := time.Duration(d) * time.Millisecond
-	return netstorage.NewDeadline(startTime, timeout, flagHint)
+	return NewDeadline(startTime, timeout, flagHint)
 }
 
 // GetBool returns boolean value from the given argKey query arg.
@@ -130,3 +130,38 @@ func GetBool(r *http.Request, argKey string) bool {
 		return true
 	}
 }
+
+// Deadline contains deadline with the corresponding timeout for pretty error messages.
+type Deadline struct {
+	deadline uint64
+
+	timeout  time.Duration
+	flagHint string
+}
+
+// NewDeadline returns deadline for the given timeout.
+//
+// flagHint must contain a hit for command-line flag, which could be used
+// in order to increase timeout.
+func NewDeadline(startTime time.Time, timeout time.Duration, flagHint string) Deadline {
+	return Deadline{
+		deadline: uint64(startTime.Add(timeout).Unix()),
+		timeout:  timeout,
+		flagHint: flagHint,
+	}
+}
+
+// Exceeded returns true if deadline is exceeded.
+func (d *Deadline) Exceeded() bool {
+	return fasttime.UnixTimestamp() > d.deadline
+}
+
+// Deadline returns deadline in unix timestamp seconds.
+func (d *Deadline) Deadline() uint64 {
+	return d.deadline
+}
+
+// String returns human-readable string representation for d.
+func (d *Deadline) String() string {
+	return fmt.Sprintf("%.3f seconds; the timeout can be adjusted with `%s` command-line flag", d.timeout.Seconds(), d.flagHint)
+}
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 223b987fb..6f9421ca7 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -551,6 +551,9 @@ It should return something like the following:
 Extra labels may be added to all the imported metrics by passing `extra_label=name=value` query args.
 For example, `/api/v1/import/prometheus?extra_label=foo=bar` would add `{foo="bar"}` label to all the imported metrics.
 
+If timestamp is missing in `<metric> <value> <timestamp>` Prometheus exposition format line, then the current timestamp is used during data ingestion.
+It can be overriden by passing unix timestamp in *milliseconds* via `timestamp` query arg. For example, `/api/v1/import/prometheus?timestamp=1594370496905`.
+
 VictoriaMetrics accepts arbitrary number of lines in a single request to `/api/v1/import/prometheus`, i.e. it supports data streaming.
 
 VictoriaMetrics also may scrape Prometheus targets - see [these docs](#how-to-scrape-prometheus-exporters-such-as-node-exporter).
diff --git a/lib/protoparser/common/timestamp.go b/lib/protoparser/common/timestamp.go
new file mode 100644
index 000000000..2e51abb91
--- /dev/null
+++ b/lib/protoparser/common/timestamp.go
@@ -0,0 +1,22 @@
+package common
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+)
+
+// GetTimestamp extracts unix timestamp in milliseconds from `timestamp` query arg.
+//
+// It returns 0 if there is no `timestamp` query arg.
+func GetTimestamp(req *http.Request) (int64, error) {
+	ts := req.URL.Query().Get("timestamp")
+	if len(ts) == 0 {
+		return 0, nil
+	}
+	timestamp, err := strconv.ParseInt(ts, 10, 64)
+	if err != nil {
+		return 0, fmt.Errorf("cannot parse `timestamp=%s` query arg: %w", ts, err)
+	}
+	return timestamp, nil
+}
diff --git a/lib/protoparser/prometheus/streamparser.go b/lib/protoparser/prometheus/streamparser.go
index 871960e44..fe17e6a36 100644
--- a/lib/protoparser/prometheus/streamparser.go
+++ b/lib/protoparser/prometheus/streamparser.go
@@ -17,7 +17,7 @@ import (
 // The callback can be called multiple times for streamed data from r.
 //
 // callback shouldn't hold rows after returning.
-func ParseStream(r io.Reader, isGzipped bool, callback func(rows []Row) error) error {
+func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback func(rows []Row) error) error {
 	if isGzipped {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -28,7 +28,7 @@ func ParseStream(r io.Reader, isGzipped bool, callback func(rows []Row) error) e
 	}
 	ctx := getStreamContext()
 	defer putStreamContext(ctx)
-	for ctx.Read(r) {
+	for ctx.Read(r, defaultTimestamp) {
 		if err := callback(ctx.Rows.Rows); err != nil {
 			return err
 		}
@@ -36,7 +36,7 @@ func ParseStream(r io.Reader, isGzipped bool, callback func(rows []Row) error) e
 	return ctx.Error()
 }
 
-func (ctx *streamContext) Read(r io.Reader) bool {
+func (ctx *streamContext) Read(r io.Reader, defaultTimestamp int64) bool {
 	readCalls.Inc()
 	if ctx.err != nil {
 		return false
@@ -55,11 +55,13 @@ func (ctx *streamContext) Read(r io.Reader) bool {
 	rows := ctx.Rows.Rows
 
 	// Fill missing timestamps with the current timestamp.
-	currentTimestamp := int64(time.Now().UnixNano() / 1e6)
+	if defaultTimestamp <= 0 {
+		defaultTimestamp = int64(time.Now().UnixNano() / 1e6)
+	}
 	for i := range rows {
 		r := &rows[i]
 		if r.Timestamp == 0 {
-			r.Timestamp = currentTimestamp
+			r.Timestamp = defaultTimestamp
 		}
 	}
 	return true
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index f9ffd9bb7..d038351d3 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -1120,6 +1120,11 @@ func (s *Storage) add(rows []rawRow, mrs []MetricRow, precisionBits uint8) ([]ra
 			// doesn't know how to work with them.
 			continue
 		}
+		if math.IsInf(mr.Value, 0) {
+			// Skip Inf values, since they may break precision for already stored data.
+			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/752
+			continue
+		}
 		if mr.Timestamp < minTimestamp {
 			// Skip rows with too small timestamps outside the retention.
 			if firstWarn == nil {


Test patch:
diff --git a/app/vmalert/config/config_test.go b/app/vmalert/config/config_test.go
index d46b4ba23..d01665980 100644
--- a/app/vmalert/config/config_test.go
+++ b/app/vmalert/config/config_test.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
+	"gopkg.in/yaml.v2"
 )
 
 func TestMain(m *testing.M) {
@@ -320,3 +321,36 @@ func TestHashRule(t *testing.T) {
 		}
 	}
 }
+
+func TestGroupChecksum(t *testing.T) {
+	data := `
+name: TestGroup
+rules:
+  - alert: ExampleAlertAlwaysFiring
+    expr: sum by(job) (up == 1)
+  - record: handler:requests:rate5m
+    expr: sum(rate(prometheus_http_requests_total[5m])) by (handler)
+`
+	var g Group
+	if err := yaml.Unmarshal([]byte(data), &g); err != nil {
+		t.Fatalf("failed to unmarshal: %s", err)
+	}
+	if g.Checksum == "" {
+		t.Fatalf("expected to get non-empty checksum")
+	}
+	newData := `
+name: TestGroup
+rules:
+  - record: handler:requests:rate5m
+    expr: sum(rate(prometheus_http_requests_total[5m])) by (handler)
+  - alert: ExampleAlertAlwaysFiring
+    expr: sum by(job) (up == 1)
+`
+	var ng Group
+	if err := yaml.Unmarshal([]byte(newData), &g); err != nil {
+		t.Fatalf("failed to unmarshal: %s", err)
+	}
+	if g.Checksum == ng.Checksum {
+		t.Fatalf("expected to get different checksums")
+	}
+}
diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index 48a276f23..6f2b1d660 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 )
 
@@ -21,7 +22,7 @@ func TestExecSuccess(t *testing.T) {
 			Start:    start,
 			End:      end,
 			Step:     step,
-			Deadline: netstorage.NewDeadline(time.Now(), time.Minute, ""),
+			Deadline: searchutils.NewDeadline(time.Now(), time.Minute, ""),
 		}
 		for i := 0; i < 5; i++ {
 			result, err := Exec(ec, q, false)
@@ -5896,7 +5897,7 @@ func TestExecError(t *testing.T) {
 			Start:    1000,
 			End:      2000,
 			Step:     100,
-			Deadline: netstorage.NewDeadline(time.Now(), time.Minute, ""),
+			Deadline: searchutils.NewDeadline(time.Now(), time.Minute, ""),
 		}
 		for i := 0; i < 4; i++ {
 			rv, err := Exec(ec, q, false)
diff --git a/lib/protoparser/prometheus/streamparser_test.go b/lib/protoparser/prometheus/streamparser_test.go
index 3a120d590..6ef7e11f2 100644
--- a/lib/protoparser/prometheus/streamparser_test.go
+++ b/lib/protoparser/prometheus/streamparser_test.go
@@ -8,11 +8,12 @@ import (
 )
 
 func TestParseStream(t *testing.T) {
+	const defaultTimestamp = 123
 	f := func(s string, rowsExpected []Row) {
 		t.Helper()
 		bb := bytes.NewBufferString(s)
 		var result []Row
-		err := ParseStream(bb, false, func(rows []Row) error {
+		err := ParseStream(bb, defaultTimestamp, false, func(rows []Row) error {
 			result = appendRowCopies(result, rows)
 			return nil
 		})
@@ -33,7 +34,7 @@ func TestParseStream(t *testing.T) {
 			t.Fatalf("unexpected error when closing gzip writer: %s", err)
 		}
 		result = nil
-		err = ParseStream(bb, true, func(rows []Row) error {
+		err = ParseStream(bb, defaultTimestamp, true, func(rows []Row) error {
 			result = appendRowCopies(result, rows)
 			return nil
 		})
@@ -67,6 +68,11 @@ func TestParseStream(t *testing.T) {
 			Timestamp: 4,
 		},
 	})
+	f("foo 23", []Row{{
+		Metric:    "foo",
+		Value:     23,
+		Timestamp: defaultTimestamp,
+	}})
 }
 
 func appendRowCopies(dst, src []Row) []Row {