Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6872 (2024-08-24 21:41:15)

RegisterMetricNames does not create new per-day index entries for new dates for existing timeseries
### Describe Your Changes

Once the timeseries is in tsidCache, new entries won't be created in per-day index because the RegisterMetricNames() code does consider different dates for the same timeseries. So this case has been added.

The same bug exists for AddRows() but it is not manifested because the index entries are finally created in updatePerDateData().

RegisterMetricNames also updated to increase the newTimeseriesCreated counter because it actually creates new time series in index.

A unit tests has been added that check all possible data patterns (different metric names and dates) and code branches in both RegisterMetricNames and AddRows. The total number of new unit tests is around 100 which increaded the running time of storage tests by 50%. The PR #6871 should help here.

### Checklist

The following checks are **mandatory**:

- [x] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).


hagen1778: >Once the timeseries is in tsidCache, new entries won't be created in per-day index because the RegisterMetricNames() code does consider different dates for the same timeseries. So this case has been added.

@rtm0 do I understand it right that this bug is related only to Graphite https://graphite.readthedocs.io/en/stable/tags.html#adding-series-to-the-tagdb handler?
rtm0: > @rtm0 do I understand it right that this bug is related only to Graphite https://graphite.readthedocs.io/en/stable/tags.html#adding-series-to-the-tagdb handler?

@hagen1778, RegisterMetricNames() is currently used for Graphite only. So yes, you are right. I understand that it is a very narrow use case and the fix is of low importance. But this code is in the codebase and therefore needs to be maintained. Also it turns out that this code can be used for restoring per-day index after it was disabled and then enabled again (something I've been working on and is almost ready, I will send a pr once a few other pending ones a merged in)

 I'd also say that the unit test in this pr is even more important since it covers many different data patterns and code branches in both RegisterMetricNames() and AddRows(), since it allows to intoduce changes to these functions safely.

rtm0: > Looks good! Thanks for adding well described tests!

Thank you!

> Could you please add a note to the [CHANGELOG](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/changelog/CHANGELOG.md#tip) with description of the bugfix? See other bugfixes from contributors for examples.

Done.

Patch:
diff --git a/docs/README.md b/docs/README.md
index 12e52f697..9fc91b084 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -1202,8 +1202,8 @@ Using the delete API is not recommended in the following cases, since it brings
 * 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.
   [Forced merge](#forced-merge) may be used for freeing up disk space occupied by old data.
-  Note that VictoriaMetrics doesn't delete entries from inverted index (aka `indexdb`) for the deleted time series.
-  Inverted index is cleaned up once per the configured [retention](#retention).
+  Note that VictoriaMetrics doesn't delete entries from [IndexDB](#indexdb) for the deleted time series.
+  IndexDB is cleaned up once per the configured [retention](#retention).
 
 It's better to use the `-retentionPeriod` command-line flag for efficient pruning of old data.
 
@@ -1905,7 +1905,46 @@ See more details in [monitoring docs](#monitoring).
 
 See [this article](https://valyala.medium.com/how-victoriametrics-makes-instant-snapshots-for-multi-terabyte-time-series-data-e1f3fb0e0282) for more details.
 
-See also [how to work with snapshots](#how-to-work-with-snapshots).
+See also [how to work with snapshots](#how-to-work-with-snapshots) and [IndexDB](#indexdb).
+
+## IndexDB
+
+VictoriaMetrics identifies
+[time series](https://docs.victoriametrics.com/keyconcepts/#time-series) by
+`TSID` (time series ID) and stores
+[raw samples](https://docs.victoriametrics.com/keyconcepts/#raw-samples) sorted
+by TSID (see [Storage](#storage)). Thus, the TSID is a primary index and could
+be used for searching and retrieving raw samples. However, the TSID is never
+exposed to the clients, i.e. it is for internal use only.
+
+Instead, VictoriaMetrics maintains an **inverted index** that enables searching
+the raw samples by metric name, label name, and label value by mapping these
+values to the corresponding TSIDs.
+
+VictoriaMetrics uses two types of inverted indexes:
+
+-   Global index. Searches using this index is performed across the entire
+    retention period.
+-   Per-day index. This index stores mappings similar to ones in global index
+    but also includes the date in each mapping. This speeds up data retrieval
+    for queries within a shorter time range (which is often just the last day).
+
+When the search query is executed, VictoriaMetrics decides which index to use
+based on the time range of the query:
+
+-   Per-day index is used if the search time range is 40 days or less.
+-   Global index is used for search queries with a time range greater than 40
+    days.
+
+Mappings are added to the indexes during the data ingestion:
+
+-   In global index each mapping is created only once per retention period.
+-   In the per-day index each mapping is be created for each unique date that
+    has been seen in the samples for the corresponding time series.
+
+IndexDB respects [retention period](#retention) and once it is over, the indexes
+are dropped. For the new retention period, the indexes are gradually populated
+again as the new samples arrive.
 
 ## Retention
 
@@ -1969,8 +2008,8 @@ For example, the following config sets 3 days retention for time series with `te
 Important notes:
 
 - The data outside the configured retention isn't deleted instantly - it is deleted eventually during [background merges](https://docs.victoriametrics.com/#storage).
-- The `-retentionFilter` doesn't remove old data from `indexdb` (aka inverted index) until the configured [-retentionPeriod](#retention).
-  So the `indexdb` size can grow big under [high churn rate](https://docs.victoriametrics.com/faq/#what-is-high-churn-rate)
+- The `-retentionFilter` doesn't remove old data from [IndexDB](#indexdb) until the configured [-retentionPeriod](#retention).
+  So the IndexDB size can grow big under [high churn rate](https://docs.victoriametrics.com/faq/#what-is-high-churn-rate)
   even for small retentions configured via `-retentionFilter`.
 
 It is safe updating `-retentionFilter` during VictoriaMetrics restarts - the updated retention filters are applied eventually
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index c237595fd..fa9e588fc 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -18,16 +18,13 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 ## tip
 
-**Update note 1: The `external_labels` field in vmalert-tool [test file](https://docs.victoriametrics.com/vmalert-tool/#test-file-format) will be deprecated soon. Please use `-external.label` command-line flag instead, in the same way as vmalert uses it. This change is done for the sake of consistency between vmalert and vmalert-tool configuration. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6735).**
-
-**Update note 2: starting from this release, [vmagent](https://docs.victoriametrics.com/vmagent/) and [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) will no longer add the default port 80/443 for scrape URLs without a specified port. 
-The value of `instance` label for those scrape targets will be changed from `<address>:<80|443>` to `<address>`, generating new time series. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6792) for details.**
+**Update note: The `external_labels` field in vmalert-tool [test file](https://docs.victoriametrics.com/vmalert-tool/#test-file-format) will be deprecated soon. Please use `-external.label` command-line flag instead, in the same way as vmalert uses it. This change is done for the sake of consistency between vmalert and vmalert-tool configuration. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6735).**
 
 * FEATURE: add `/influx/health` health-check handler for Influx endpoints. This is needed as some clients use the health endpoint to determine if the server is healthy and ready for data ingestion. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6653) for the details.
 * FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl/): add `--vm-backoff-retries`, `--vm-backoff-factor`, `--vm-backoff-min-duration` and `--vm-native-backoff-retries`, `--vm-native-backoff-factor`, `--vm-native-backoff-min-duration` command-line flags. These flags allow to change backoff policy config for import requests to VictoriaMetrics. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6622).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): allow overriding the `sample_limit` option at [scrape_configs](https://docs.victoriametrics.com/sd_configs/#scrape_configs) when a label `__sample_limit__` is specified for target. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6665). Thanks to @zoglam for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6666).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): reduce memory usage when scraping targets with big response body. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6759).
-* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/) and [Single-node VictoriaMetrics](https://docs.victoriametrics.com/): stop adding default port 80/443 for scrape URLs without a port. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6792).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/) and [Single-node VictoriaMetrics](https://docs.victoriametrics.com/): stop adding default port 80/443 for scrape URLs without a port. The value in `instance` label will still carry port for backward compatibility. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6792).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add flags `-remoteWrite.retryMinInterval` and `-remoteWrite.retryMaxTime` for adjusting remote-write requests retry policy. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5486). Thanks to @yorik for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6289).
 * FEATURE: [vmalert](https://docs.victoriametrics.com/vmalert/): add command-line flag `-notifier.headers` to allow configuring additional headers for all requests sent to the corresponding `-notifier.url`.
 * FEATURE: [vmalert-tool](https://docs.victoriametrics.com/vmalert-tool/): add `-external.label` and `-external.url` command-line flags, in the same way as these flags are supported by vmalert. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6735).
@@ -49,6 +46,7 @@ The value of `instance` label for those scrape targets will be changed from `<ad
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert/), [vmctl](https://docs.victoriametrics.com/vmctl/) and snapshot API: verify correctness of URLs provided via cmd-line flags before executing HTTP requests. See [this](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6740) issue for details.
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert): reduce memory usage when parsing responses with big number of metrics in response. The memory usage was increased in [v1.102.0-rc1](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.102.0-rc1) after attempt to reduce CPU usage for heavy loaded vmalerts.
 * BUGFIX: all VictoriaMetrics components: forcefully set owner/group for release tars to 1000:1000. This helps to avoid unpacking [issues](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6788) on systems with limitations around UID:GID configuration. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6846).
+* BUGFIX: [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): fix metric names registering in the per-day index for new dates for existing time series when making calls to `/tags/tagSeries` and `/tags/tagMultiSeries` handlers of [Grpahite API](https://docs.victoriametrics.com/#graphite-api-usage). See [this](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6872/) for details.
 
 ## [v1.102.1](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.102.1)
 
diff --git a/lib/promrelabel/scrape_url.go b/lib/promrelabel/scrape_url.go
index 7533d7547..cedbe6caf 100644
--- a/lib/promrelabel/scrape_url.go
+++ b/lib/promrelabel/scrape_url.go
@@ -38,6 +38,13 @@ func GetScrapeURL(labels *promutils.Labels, extraParams map[string][]string) (st
 		address = address[:n]
 	}
 
+	// If port is missing, typically it should be 80/443. This WAS written in a label and used as scrapeURL.
+	// However, adding the port by default can cause some issues, see: https://github.com/prometheus/prometheus/pull/9523#issuecomment-2059314966
+	// After https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6792:
+	// - don't add the default port to scrapeURL.
+	// - continue adding the default port to the label value for backward compatibility and avoid generating new time series.
+	addressMustWithPort := addMissingPort(address, scheme == "https")
+
 	if !strings.HasPrefix(metricsPath, "/") {
 		metricsPath = "/" + metricsPath
 	}
@@ -51,7 +58,7 @@ func GetScrapeURL(labels *promutils.Labels, extraParams map[string][]string) (st
 	}
 	paramsStr := url.Values(params).Encode()
 	scrapeURL := buildScrapeURL(scheme, address, metricsPath, optionalQuestion, paramsStr)
-	return scrapeURL, address
+	return scrapeURL, addressMustWithPort
 }
 
 func getParamsFromLabels(labels *promutils.Labels, extraParams map[string][]string) map[string][]string {
@@ -89,4 +96,25 @@ func buildScrapeURL(scheme, address, metricsPath, optionalQuestion, paramsStr st
 	return s
 }
 
+func addMissingPort(addr string, isTLS bool) string {
+	if strings.Contains(addr, ":") {
+		return addr
+	}
+	if isTLS {
+		return concatTwoStrings(addr, ":443")
+	}
+	return concatTwoStrings(addr, ":80")
+}
+
+func concatTwoStrings(x, y string) string {
+	bb := bbPool.Get()
+	b := bb.B[:0]
+	b = append(b, x...)
+	b = append(b, y...)
+	s := bytesutil.InternBytes(b)
+	bb.B = b
+	bbPool.Put(bb)
+	return s
+}
+
 var bbPool bytesutil.ByteBufferPool
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index 2995326b1..b3553d33d 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -40,6 +40,8 @@ const (
 
 // Storage represents TSDB storage.
 type Storage struct {
+	rowsAddedTotal atomic.Uint64
+
 	tooSmallTimestampRows atomic.Uint64
 	tooBigTimestampRows   atomic.Uint64
 
@@ -552,7 +554,7 @@ func (m *Metrics) Reset() {
 
 // UpdateMetrics updates m with metrics from s.
 func (s *Storage) UpdateMetrics(m *Metrics) {
-	m.RowsAddedTotal = rowsAddedTotal.Load()
+	m.RowsAddedTotal += s.rowsAddedTotal.Load()
 	m.DedupsDuringMerge = dedupsDuringMerge.Load()
 	m.SnapshotsCount += uint64(s.mustGetSnapshotsCount())
 
@@ -1606,8 +1608,6 @@ func (s *Storage) ForceMergePartitions(partitionNamePrefix string) error {
 	return s.tb.ForceMergePartitions(partitionNamePrefix)
 }
 
-var rowsAddedTotal atomic.Uint64
-
 // AddRows adds the given mrs to s.
 //
 // The caller should limit the number of concurrent AddRows calls to the number
@@ -1629,7 +1629,7 @@ func (s *Storage) AddRows(mrs []MetricRow, precisionBits uint8) {
 			mrs = nil
 		}
 		s.add(ic.rrs, ic.tmpMrs, mrsBlock, precisionBits)
-		rowsAddedTotal.Add(uint64(len(mrsBlock)))
+		s.rowsAddedTotal.Add(uint64(len(mrsBlock)))
 	}
 	putMetricRowsInsertCtx(ic)
 }
@@ -1674,6 +1674,7 @@ func (s *Storage) RegisterMetricNames(qt *querytracer.Tracer, mrs []MetricRow) {
 	mn := GetMetricName()
 	defer PutMetricName(mn)
 
+	var newSeriesCount uint64
 	var seriesRepopulated uint64
 
 	idb := s.idb()
@@ -1708,6 +1709,18 @@ func (s *Storage) RegisterMetricNames(qt *querytracer.Tracer, mrs []MetricRow) {
 				genTSID.generation = generation
 				s.putSeriesToCache(mr.MetricNameRaw, &genTSID, date)
 				seriesRepopulated++
+			} else if !s.dateMetricIDCache.Has(generation, date, genTSID.TSID.MetricID) {
+				if !is.hasDateMetricIDNoExtDB(date, genTSID.TSID.MetricID) {
+					if err := mn.UnmarshalRaw(mr.MetricNameRaw); err != nil {
+						if firstWarn == nil {
+							firstWarn = fmt.Errorf("cannot unmarshal MetricNameRaw %q: %w", mr.MetricNameRaw, err)
+						}
+						continue
+					}
+					mn.sortTags()
+					is.createPerDayIndexes(date, &genTSID.TSID, mn)
+				}
+				s.dateMetricIDCache.Set(generation, date, genTSID.TSID.MetricID)
 			}
 			continue
 		}
@@ -1758,8 +1771,10 @@ func (s *Storage) RegisterMetricNames(qt *querytracer.Tracer, mrs []MetricRow) {
 		createAllIndexesForMetricName(is, mn, &genTSID.TSID, date)
 		genTSID.generation = generation
 		s.putSeriesToCache(mr.MetricNameRaw, &genTSID, date)
+		newSeriesCount++
 	}
 
+	s.newTimeseriesCreated.Add(newSeriesCount)
 	s.timeseriesRepopulated.Add(seriesRepopulated)
 
 	// There is no need in pre-filling idbNext here, since RegisterMetricNames() is rarely called.


Test patch:
diff --git a/lib/promrelabel/scrape_url_test.go b/lib/promrelabel/scrape_url_test.go
index 7bdbdb5a1..8660bb9d0 100644
--- a/lib/promrelabel/scrape_url_test.go
+++ b/lib/promrelabel/scrape_url_test.go
@@ -24,35 +24,39 @@ func TestGetScrapeURL(t *testing.T) {
 	f(`{foo="bar"}`, "", "")
 
 	// __address__ without port
-	f(`{__address__="foo"}`, "http://foo/metrics", "foo")
+	f(`{__address__="foo"}`, "http://foo/metrics", "foo:80")
 
 	// __address__ with explicit port
 	f(`{__address__="foo:1234"}`, "http://foo:1234/metrics", "foo:1234")
 
 	// explicit __scheme__
-	f(`{__address__="foo",__scheme__="https"}`, "https://foo/metrics", "foo")
+	f(`{__address__="foo",__scheme__="https"}`, "https://foo/metrics", "foo:443")
 	f(`{__address__="foo:1234",__scheme__="https"}`, "https://foo:1234/metrics", "foo:1234")
 
 	// explicit __metrics_path__
-	f(`{__address__="foo",__metrics_path__="abc"}`, "http://foo/abc", "foo")
-	f(`{__address__="foo",__metrics_path__="/abc"}`, "http://foo/abc", "foo")
-	f(`{__address__="foo",__metrics_path__="/ab/c?d=ef&aa=bb"}`, "http://foo/ab/c?d=ef&aa=bb", "foo")
+	f(`{__address__="foo",__metrics_path__="abc"}`, "http://foo/abc", "foo:80")
+	f(`{__address__="foo",__metrics_path__="/abc"}`, "http://foo/abc", "foo:80")
+	f(`{__address__="foo",__metrics_path__="/ab/c?d=ef&aa=bb"}`, "http://foo/ab/c?d=ef&aa=bb", "foo:80")
 
 	// explitit __param_*
-	f(`{__address__="foo",__param_x="y"}`, "http://foo/metrics?x=y", "foo")
-	f(`{__address__="foo",__param_x="y",__param_y="aa"}`, "http://foo/metrics?x=y&y=aa", "foo")
-	f(`{__address__="foo",__param_x="y",__metrics_path__="?abc=de"}`, "http://foo/?abc=de&x=y", "foo")
-	f(`{__address__="foo",__param_abc="y",__metrics_path__="?abc=de"}`, "http://foo/?abc=de&abc=y", "foo")
+	f(`{__address__="foo",__param_x="y"}`, "http://foo/metrics?x=y", "foo:80")
+	f(`{__address__="foo",__param_x="y",__param_y="aa"}`, "http://foo/metrics?x=y&y=aa", "foo:80")
+	f(`{__address__="foo",__param_x="y",__metrics_path__="?abc=de"}`, "http://foo/?abc=de&x=y", "foo:80")
+	f(`{__address__="foo",__param_abc="y",__metrics_path__="?abc=de"}`, "http://foo/?abc=de&abc=y", "foo:80")
 
 	// __address__ with metrics path and/or scheme
-	f(`{__address__="foo/bar/baz?abc=de"}`, "http://foo/bar/baz?abc=de", "foo")
+	f(`{__address__="foo/bar/baz?abc=de"}`, "http://foo/bar/baz?abc=de", "foo:80")
 	f(`{__address__="foo:784/bar/baz?abc=de"}`, "http://foo:784/bar/baz?abc=de", "foo:784")
 	f(`{__address__="foo:784/bar/baz?abc=de",__param_xx="yy"}`, "http://foo:784/bar/baz?abc=de&xx=yy", "foo:784")
 	f(`{__address__="foo:784/bar/baz?abc=de",__param_xx="yy",__scheme__="https"}`, "https://foo:784/bar/baz?abc=de&xx=yy", "foo:784")
-	f(`{__address__="http://foo/bar/baz?abc=de",__param_xx="yy"}`, "http://foo/bar/baz?abc=de&xx=yy", "foo")
-	f(`{__address__="https://foo/bar/baz?abc=de",__param_xx="yy"}`, "https://foo/bar/baz?abc=de&xx=yy", "foo")
-
-	// __address__ with 80 or 443 as port
+	f(`{__address__="http://foo/bar/baz?abc=de",__param_xx="yy"}`, "http://foo/bar/baz?abc=de&xx=yy", "foo:80")
+	f(`{__address__="https://foo/bar/baz?abc=de",__param_xx="yy"}`, "https://foo/bar/baz?abc=de&xx=yy", "foo:443")
+
+	// __address__ already carry 80/443 port
+	f(`{__address__="foo:80"}`, "http://foo:80/metrics", "foo:80")
+	f(`{__address__="foo:443"}`, "http://foo:443/metrics", "foo:443")
+	f(`{__address__="http://foo"}`, "http://foo/metrics", "foo:80")
+	f(`{__address__="https://foo"}`, "https://foo/metrics", "foo:443")
 	f(`{__address__="http://foo:80"}`, "http://foo:80/metrics", "foo:80")
 	f(`{__address__="https://foo:443"}`, "https://foo:443/metrics", "foo:443")
 }
diff --git a/lib/promscrape/config_test.go b/lib/promscrape/config_test.go
index a73c5b5fd..84c1bd72f 100644
--- a/lib/promscrape/config_test.go
+++ b/lib/promscrape/config_test.go
@@ -229,7 +229,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "host1",
+				"instance": "host1:80",
 				"job":      "abc",
 			}),
 			jobNameOriginal: "abc",
@@ -240,7 +240,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "host2",
+				"instance": "host2:443",
 				"job":      "abc",
 			}),
 			jobNameOriginal: "abc",
@@ -464,7 +464,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "host1",
+				"instance": "host1:80",
 				"job":      "foo",
 				"qwe":      "rty",
 			}),
@@ -476,7 +476,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "host2",
+				"instance": "host2:80",
 				"job":      "foo",
 				"qwe":      "rty",
 			}),
@@ -695,7 +695,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "s",
+				"instance": "s:80",
 				"job":      "aa",
 			}),
 			jobNameOriginal: "aa",
@@ -717,7 +717,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "s",
+				"instance": "s:80",
 				"job":      "aa",
 			}),
 			jobNameOriginal: "aa",
@@ -739,7 +739,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "s",
+				"instance": "s:80",
 				"job":      "aa",
 			}),
 			jobNameOriginal: "aa",
@@ -761,7 +761,7 @@ scrape_configs:
 			ScrapeTimeout:  defaultScrapeTimeout,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "s",
+				"instance": "s:80",
 				"job":      "aa",
 			}),
 			jobNameOriginal: "aa",
@@ -853,7 +853,7 @@ scrape_configs:
 			HonorTimestamps: true,
 			DenyRedirects:   true,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "foo.bar",
+				"instance": "foo.bar:443",
 				"job":      "foo",
 				"x":        "y",
 			}),
@@ -869,7 +869,7 @@ scrape_configs:
 			HonorTimestamps: true,
 			DenyRedirects:   true,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "aaa",
+				"instance": "aaa:443",
 				"job":      "foo",
 				"x":        "y",
 			}),
@@ -882,7 +882,7 @@ scrape_configs:
 			ScrapeTimeout:  8 * time.Second,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "1.2.3.4",
+				"instance": "1.2.3.4:80",
 				"job":      "qwer",
 			}),
 			jobNameOriginal: "qwer",
@@ -893,7 +893,7 @@ scrape_configs:
 			ScrapeTimeout:  8 * time.Second,
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
-				"instance": "foobar",
+				"instance": "foobar:80",
 				"job":      "asdf",
 			}),
 			jobNameOriginal: "asdf",
@@ -1109,7 +1109,7 @@ scrape_configs:
 			MaxScrapeSize:  maxScrapeSize.N,
 			Labels: promutils.NewLabelsFromMap(map[string]string{
 				"foo":      "bar",
-				"instance": "pp",
+				"instance": "pp:80",
 				"job":      "yyy",
 			}),
 			ExternalLabels: promutils.NewLabelsFromMap(map[string]string{
diff --git a/lib/storage/storage_test.go b/lib/storage/storage_test.go
index a789a2db1..26a36b5c3 100644
--- a/lib/storage/storage_test.go
+++ b/lib/storage/storage_test.go
@@ -1449,3 +1449,415 @@ func testCountAllMetricNames(s *Storage, tr TimeRange) int {
 	}
 	return len(names)
 }
+
+// testCountAllMetricIDs is a test helper function that counts the IDs of
+// all time series within the given time range.
+func testCountAllMetricIDs(s *Storage, tr TimeRange) int {
+	tfsAll := NewTagFilters()
+	if err := tfsAll.Add([]byte("__name__"), []byte(".*"), false, true); err != nil {
+		panic(fmt.Sprintf("unexpected error in TagFilters.Add: %v", err))
+	}
+	ids, err := s.idb().searchMetricIDs(nil, []*TagFilters{tfsAll}, tr, 1e9, noDeadline)
+	if err != nil {
+		panic(fmt.Sprintf("seachMetricIDs() failed unexpectedly: %s", err))
+	}
+	return len(ids)
+}
+
+func TestStorageRegisterMetricNamesForVariousDataPatternsConcurrently(t *testing.T) {
+	testStorageVariousDataPatternsConcurrently(t, true, func(s *Storage, mrs []MetricRow) {
+		s.RegisterMetricNames(nil, mrs)
+	})
+}
+
+func TestStorageAddRowsForVariousDataPatternsConcurrently(t *testing.T) {
+	testStorageVariousDataPatternsConcurrently(t, false, func(s *Storage, mrs []MetricRow) {
+		s.AddRows(mrs, defaultPrecisionBits)
+	})
+}
+
+// testStorageVariousDataPatternsConcurrently tests different concurrency use
+// cases when ingesting data of different patterns.
+//
+// The function is intended to be used by other tests that define which
+// operation (AddRows or RegisterMetricNames) is tested.
+func testStorageVariousDataPatternsConcurrently(t *testing.T, registerOnly bool, op func(s *Storage, mrs []MetricRow)) {
+	defer testRemoveAll(t)
+
+	const concurrency = 4
+
+	t.Run("serial", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, registerOnly, op, 1, false)
+	})
+	t.Run("concurrentRows", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, registerOnly, op, concurrency, true)
+	})
+	t.Run("concurrentBatches", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, registerOnly, op, concurrency, false)
+	})
+}
+
+// testStorageVariousDataPatterns tests the ingestion of different combinations
+// of metric names and dates.
+//
+// The function is intended to be used by other tests that define the
+// concurrency and the operation (AddRows or RegisterMetricNames) under test.
+func testStorageVariousDataPatterns(t *testing.T, registerOnly bool, op func(s *Storage, mrs []MetricRow), concurrency int, splitBatches bool) {
+	f := func(t *testing.T, sameBatchMetricNames, sameRowMetricNames, sameBatchDates, sameRowDates bool) {
+		batches, wantCounts := testGenerateMetricRowBatches(&batchOptions{
+			numBatches:           4,
+			numRowsPerBatch:      100,
+			registerOnly:         registerOnly,
+			sameBatchMetricNames: sameBatchMetricNames,
+			sameRowMetricNames:   sameRowMetricNames,
+			sameBatchDates:       sameBatchDates,
+			sameRowDates:         sameRowDates,
+		})
+		strict := concurrency == 1
+
+		s := MustOpenStorage(t.Name(), 0, 0, 0)
+
+		testDoConcurrently(s, op, concurrency, splitBatches, batches)
+		s.DebugFlush()
+		assertCounts(t, s, wantCounts, strict)
+
+		// Rotate indexDB to test the case when TSIDs from tsidCache have the
+		// generation that is older than the generation of the current indexDB.
+		s.mustRotateIndexDB(time.Now())
+		testDoConcurrently(s, op, concurrency, splitBatches, batches)
+		s.DebugFlush()
+		assertCounts(t, s, wantCounts, strict)
+
+		// Empty the tsidCache to test the case when tsid is retrived from the
+		// index that belongs to the current generation indexDB.
+		s.resetAndSaveTSIDCache()
+		testDoConcurrently(s, op, concurrency, splitBatches, batches)
+		s.DebugFlush()
+		assertCounts(t, s, wantCounts, strict)
+
+		// Empty the tsidCache and rotate indexDB to test the case when tsid is
+		// retrived from the index that belongs to the previous generation
+		// indexDB.
+		s.resetAndSaveTSIDCache()
+		s.mustRotateIndexDB(time.Now())
+		testDoConcurrently(s, op, concurrency, splitBatches, batches)
+		s.DebugFlush()
+		assertCounts(t, s, wantCounts, strict)
+
+		s.MustClose()
+	}
+
+	t.Run("sameBatchMetrics/sameRowMetrics/sameBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric 1971-01-01, metric 1971-01-01
+		// Batch2: metric 1971-01-01, metric 1971-01-01
+		t.Parallel()
+		f(t, true, true, true, true)
+	})
+
+	t.Run("sameBatchMetrics/sameRowMetrics/sameBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric 1971-01-01, metric 1971-01-02
+		// Batch2: metric 1971-01-01, metric 1971-01-02
+		t.Parallel()
+		f(t, true, true, true, false)
+	})
+
+	t.Run("sameBatchMetrics/sameRowMetrics/diffBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric 1971-01-01, metric 1971-01-01
+		// Batch2: metric 1971-01-02, metric 1971-01-02
+		t.Parallel()
+		f(t, true, true, false, true)
+	})
+
+	t.Run("sameBatchMetrics/sameRowMetrics/diffBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric 1971-01-01, metric 1971-01-02
+		// Batch2: metric 1971-01-03, metric 1971-01-04
+		t.Parallel()
+		f(t, true, true, false, false)
+	})
+
+	t.Run("sameBatchMetrics/diffRowMetrics/sameBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric_row0 1971-01-01, metric_row1 1971-01-01
+		// Batch2: metric_row0 1971-01-01, metric_row1 1971-01-01
+		t.Parallel()
+		f(t, true, false, true, true)
+	})
+
+	t.Run("sameBatchMetrics/diffRowMetrics/sameBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric_row0 1971-01-01, metric_row1 1971-01-02
+		// Batch2: metric_row0 1971-01-01, metric_row1 1971-01-02
+		t.Parallel()
+		f(t, true, false, true, false)
+	})
+
+	t.Run("sameBatchMetrics/diffRowMetrics/diffBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric_row0 1971-01-01, metric_row1 1971-01-01
+		// Batch2: metric_row0 1971-01-02, metric_row1 1971-01-02
+		t.Parallel()
+		f(t, true, false, false, true)
+	})
+
+	t.Run("sameBatchMetrics/diffRowMetrics/diffBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric_row0 1971-01-01, metric_row1 1971-01-02
+		// Batch2: metric_row0 1971-01-03, metric_row1 1971-01-04
+		t.Parallel()
+		f(t, true, false, false, false)
+	})
+
+	t.Run("diffBatchMetrics/sameRowMetrics/sameBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0 1971-01-01, metric_batch0 1971-01-01
+		// Batch2: metric_batch1 1971-01-01, metric_batch1 1971-01-01
+		t.Parallel()
+		f(t, false, true, true, true)
+	})
+
+	t.Run("diffBatchMetrics/sameRowMetrics/sameBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0 1971-01-01, metric_batch0 1971-01-02
+		// Batch2: metric_batch1 1971-01-01, metric_batch1 1971-01-02
+		t.Parallel()
+		f(t, false, true, true, false)
+	})
+
+	t.Run("diffBatchMetrics/sameRowMetrics/diffBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0 1971-01-01, metric_batch0 1971-01-01
+		// Batch2: metric_batch1 1971-01-02, metric_batch1 1971-01-02
+		t.Parallel()
+		f(t, false, true, false, true)
+	})
+
+	t.Run("diffBatchMetrics/sameRowMetrics/diffBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0 1971-01-01, metric_batch0 1971-01-02
+		// Batch2: metric_batch1 1971-01-03, metric_batch1 1971-01-04
+		t.Parallel()
+		f(t, false, true, false, false)
+	})
+
+	t.Run("diffBatchMetrics/diffRowMetrics/sameBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0_row0 1971-01-01, metric_batch0_row1 1971-01-01
+		// Batch2: metric_batch1_row0 1971-01-01, metric_batch1_row1 1971-01-01
+		t.Parallel()
+		f(t, false, false, true, true)
+	})
+
+	t.Run("diffBatchMetrics/diffRowMetrics/sameBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0_row0 1971-01-01, metric_batch0_row1 1971-01-02
+		// Batch2: metric_batch1_row0 1971-01-01, metric_batch1_row1 1971-01-02
+		t.Parallel()
+		f(t, false, false, true, false)
+	})
+
+	t.Run("diffBatchMetrics/diffRowMetrics/diffBatchDates/sameRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0_row0 1971-01-01, metric_batch0_row1 1971-01-01
+		// Batch2: metric_batch1_row0 1971-01-02, metric_batch1_row1 1971-01-02
+		t.Parallel()
+		f(t, false, false, false, true)
+	})
+
+	t.Run("diffBatchMetrics/diffRowMetrics/diffBatchDates/diffRowDates", func(t *testing.T) {
+		// Batch1: metric_batch0_row0 1971-01-01, metric_batch0_row1 1971-01-02
+		// Batch2: metric_batch1_row0 1971-01-03, metric_batch1_row1 1971-01-04
+		t.Parallel()
+		f(t, false, false, false, false)
+	})
+}
+
+// testDoConcurrently performs some storage operation on metric rows
+// concurrently.
+//
+// The function accepts metric rows organized in batches. The number of
+// goroutines is specified with concurrency arg. If splitBatches is false, then
+// each batch is processed in a separate goroutine. Otherwise, rows from a
+// single batch are spread across multiple goroutines and next batch won't be
+// processed until all records of the current batch are processed.
+func testDoConcurrently(s *Storage, op func(s *Storage, mrs []MetricRow), concurrency int, splitBatches bool, mrsBatches [][]MetricRow) {
+	if concurrency < 1 {
+		panic(fmt.Sprintf("Unexpected concurrency: got %d, want >= 1", concurrency))
+	}
+
+	var wg sync.WaitGroup
+	mrsCh := make(chan []MetricRow)
+	for range concurrency {
+		wg.Add(1)
+		go func() {
+			for mrs := range mrsCh {
+				op(s, mrs)
+			}
+			wg.Done()
+		}()
+	}
+
+	n := 1
+	if splitBatches {
+		n = concurrency
+	}
+	for _, batch := range mrsBatches {
+		step := len(batch) / n
+		if step == 0 {
+			step = 1
+		}
+		for begin := 0; begin < len(batch); begin += step {
+			limit := begin + step
+			if limit > len(batch) {
+				limit = len(batch)
+			}
+			mrsCh <- batch[begin:limit]
+		}
+	}
+	close(mrsCh)
+	wg.Wait()
+}
+
+type counts struct {
+	metrics          *Metrics
+	timeRangeCounts  map[TimeRange]int
+	dateTSDBStatuses map[uint64]*TSDBStatus
+}
+
+// assertCounts retrieves various counts from storage and compares them with
+// the wanted ones.
+//
+// Some counts can be greater than wanted values because duplicate metric IDs
+// can be created when rows are inserted concurrently. In this case `strict`
+// arg can be set to false in order to replace strict equality comparison with
+// `greater or equal`.
+func assertCounts(t *testing.T, s *Storage, want *counts, strict bool) {
+	t.Helper()
+
+	var gotMetrics Metrics
+	s.UpdateMetrics(&gotMetrics)
+	gotCnt, wantCnt := gotMetrics.NewTimeseriesCreated, want.metrics.NewTimeseriesCreated
+	if strict {
+		if gotCnt != wantCnt {
+			t.Errorf("unexpected Metrics.NewTimeseriesCreated: got %d, want %d", gotCnt, wantCnt)
+		}
+	} else {
+		if gotCnt < wantCnt {
+			t.Errorf("unexpected Metrics.NewTimeseriesCreated: got %d, want >= %d", gotCnt, wantCnt)
+		}
+	}
+
+	for tr, want := range want.timeRangeCounts {
+		if got := testCountAllMetricNames(s, tr); got != want {
+			t.Errorf("%v: unexpected metric name count: got %d, want %d", &tr, got, want)
+		}
+		got := testCountAllMetricIDs(s, tr)
+		if strict {
+			if got != want {
+				t.Errorf("%v: unexpected metric ID count: got %d, want %d", &tr, got, want)
+			}
+		} else {
+			if got < want {
+				t.Errorf("%v: unexpected metric ID count: got %d, want >= %d", &tr, got, want)
+			}
+		}
+
+	}
+
+	for date, wantStatus := range want.dateTSDBStatuses {
+		dt := time.UnixMilli(int64(date) * msecPerDay).UTC()
+		gotStatus, err := s.GetTSDBStatus(nil, nil, date, "", 10, 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("GetTSDBStatus(%v) failed unexpectedly: %v", dt, err)
+		}
+		got, want := gotStatus.TotalSeries, wantStatus.TotalSeries
+		if strict {
+			if got != want {
+				t.Errorf("%v: unexpected TSDBStatus.TotalSeries: got %d, want %d", dt, got, want)
+			}
+		} else {
+			if got < want {
+				t.Errorf("%v: unexpected TSDBStatus.TotalSeries: got %d, want >= %d", dt, got, want)
+			}
+		}
+	}
+}
+
+type batchOptions struct {
+	numBatches           int
+	numRowsPerBatch      int
+	registerOnly         bool
+	sameBatchMetricNames bool
+	sameRowMetricNames   bool
+	sameBatchDates       bool
+	sameRowDates         bool
+}
+
+// testGenerateMetricRowBatches generates metric rows batches of various
+// combinations of metric names and dates. The function also returns the counts
+// that the storage is expected to report once the generated batch is ingested
+// into the storage.
+func testGenerateMetricRowBatches(opts *batchOptions) ([][]MetricRow, *counts) {
+	if opts.numBatches <= 0 {
+		panic(fmt.Sprintf("unexpected number of batches: got %d, want > 0", opts.numBatches))
+	}
+	if opts.numRowsPerBatch <= 0 {
+		panic(fmt.Sprintf("unexpected number of rows per batch: got %d, want > 0", opts.numRowsPerBatch))
+	}
+
+	rng := rand.New(rand.NewSource(1))
+
+	batches := make([][]MetricRow, opts.numBatches)
+	metricName := "metric"
+	startTime := time.Date(1971, 1, 1, 0, 0, 0, 0, time.UTC)
+	endTime := time.Date(1971, 1, 1, 23, 59, 59, 999, time.UTC)
+	days := time.Duration(0)
+	trNames := make(map[TimeRange]map[string]bool)
+	names := make(map[string]bool)
+	for batch := range opts.numBatches {
+		batchMetricName := metricName
+		if !opts.sameBatchMetricNames {
+			batchMetricName += fmt.Sprintf("_batch%d", batch)
+		}
+		var rows []MetricRow
+		for row := range opts.numRowsPerBatch {
+			rowMetricName := batchMetricName
+			if !opts.sameRowMetricNames {
+				rowMetricName += fmt.Sprintf("_row%d", row)
+			}
+			mn := MetricName{
+				MetricGroup: []byte(rowMetricName),
+			}
+			tr := TimeRange{
+				MinTimestamp: startTime.Add(days * 24 * time.Hour).UnixMilli(),
+				MaxTimestamp: endTime.Add(days * 24 * time.Hour).UnixMilli(),
+			}
+			rows = append(rows, MetricRow{
+				MetricNameRaw: mn.marshalRaw(nil),
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+			if !opts.sameRowDates {
+				days++
+			}
+
+			if trNames[tr] == nil {
+				trNames[tr] = make(map[string]bool)
+			}
+			names[rowMetricName] = true
+			trNames[tr][rowMetricName] = true
+		}
+		batches[batch] = rows
+		if opts.sameBatchDates {
+			days = 0
+		} else if opts.sameRowDates {
+			days++
+		}
+	}
+
+	allTimeseries := len(names)
+	want := counts{
+		metrics: &Metrics{
+			NewTimeseriesCreated: uint64(allTimeseries),
+		},
+		timeRangeCounts:  make(map[TimeRange]int),
+		dateTSDBStatuses: make(map[uint64]*TSDBStatus),
+	}
+	for tr, names := range trNames {
+		count := len(names)
+		date := uint64(tr.MinTimestamp / msecPerDay)
+		want.timeRangeCounts[tr] = count
+		want.dateTSDBStatuses[date] = &TSDBStatus{
+			TotalSeries: uint64(count),
+		}
+	}
+	return batches, &want
+}