Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6039 (2024-03-29 06:05:38)

fix: [vmalert] shutdown remote write follow batch size limit
### Describe Your Changes

During shutdown period of `vmalert`, remotewrite client retrieve all pending time series from buffer queue, compose them into **1** batch and execute remote write. 

This **final batch** may exceed the limit of `-remoteWrite.maxBatchSize`, and be **rejected** by the receiver (gateway, vmcluster or others). Example: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6025

Related codes: https://github.com/VictoriaMetrics/VictoriaMetrics/blob/ac9c2a796fcabe39a476b5194d016d1bb482b943/app/vmalert/remotewrite/client.go#L153

---

### CHANGELOG 
#### Fixed
- BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert.html): respect `-remoteWrite.maxBatchSize` at shutdown period. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6025).

---

### Checklist

The following checks are mandatory:

- [X] I have read the [Contributing Guidelines](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/CONTRIBUTING.md)
- [X] All commits are signed and include `Signed-off-by` line. Use `git commit -s` to include `Signed-off-by` your commits. See this [doc](https://git-scm.com/book/en/v2/Git-Tools-Signing-Your-Work) about how to sign your commits.
- [X] Tests are passing locally. Use `make test` to run all tests locally.
- [X] Linting is passing locally. Use `make check-all` to run all linters locally.
- [x] Include a link to the GitHub issue in the commit message, if issue exists.
- [x] Mention the change in the [Changelog](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/CHANGELOG.md). Explain what has changed and why. If there is a related issue or documentation change - link them as well.
- [ ] After your pull request is merged, please add a message to the issue with instructions for how to test the fix or try the feature you added. Here is an [example](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4048#issuecomment-1546453726)
- [X] Do not close the original issue before the change is released. Please note, in some cases Github can automatically close the issue once PR is merged. Re-open the issue in such case.
- [X] If the change somehow affects public interfaces (a new flag was added or updated, or some behavior has changed) - add the corresponding change to documentation.


Note: tests and linting for some (existing) codes are failed. I believe it's not relevent to this pr, and could be fixed in another pr.


jiekun: # PR Notes
- Unit tests added. But it's a littly bit  [flaky](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6021). Would like to have some input from maintainers.
- New module imported for monkey patch during unit testing. Also need some advice (or test guideline/standard).

ping @Haleygo (who has done something related to the integration testing) for help.
Haleygo: Thanks for the pull request!
>New module imported for monkey patch during unit testing. Need some advice (or test guideline/standard).

I don't think we need to introduce external package though. We can mock a remotewrite server like [faultyRWServer](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/app/vmalert/remotewrite/client_test.go#L166) here, which will fail write request with particular order, and increase `-remoteWrite.retryMinInterval` to control retry attempts. The test result can be verified by checking server's acceptedRows like https://github.com/VictoriaMetrics/VictoriaMetrics/blob/ac9c2a796fcabe39a476b5194d016d1bb482b943/app/vmalert/remotewrite/client_test.go#L77-L80
jiekun: @Haleygo Thank you for the quick review!

I've altered it to a `batchCntRWServer` as suggested.

---
I would recommand to introduce a mock package (maybe in the future) tho, which may help reducing the complexity, and make unit test focus on a single unit.

Take this PR as an example, once we use a mocked HTTP server, the test actually covers both `run` method and many other lines in `flush` method and bla bla bla.

I think the maintainers have done a lot of work on testing, which is **quite impressive**! I believe that in the future, we could possibly add more tests but specific to certain types, such as unit tests + integration tests + and end-to-end tests (in different CI pipeline) (happy to contribute).
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6039?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 56.74%. Comparing base [(`8aaa828`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/8aaa828ba3f49a43cb8de89a0bcae160f7988c14?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) to head [(`33a3e8c`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6039?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Report is 414 commits behind head on master.


<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #6039      +/-   ##
==========================================
- Coverage   60.37%   56.74%   -3.63%     
==========================================
  Files         411      524     +113     
  Lines       76609    72342    -4267     
==========================================
- Hits        46253    41053    -5200     
- Misses      27794    28385     +591     
- Partials     2562     2904     +342     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6039?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go
index ed8f56ff4..6ae13e81c 100644
--- a/app/vmagent/remotewrite/remotewrite.go
+++ b/app/vmagent/remotewrite/remotewrite.go
@@ -50,8 +50,9 @@ var (
 		"By default the data is replicated across all the -remoteWrite.url . See https://docs.victoriametrics.com/vmagent.html#sharding-among-remote-storages")
 	shardByURLLabels = flagutil.NewArrayString("remoteWrite.shardByURL.labels", "Optional list of labels, which must be used for sharding outgoing samples "+
 		"among remote storage systems if -remoteWrite.shardByURL command-line flag is set. By default all the labels are used for sharding in order to gain "+
-		"even distribution of series over the specified -remoteWrite.url systems")
-	tmpDataPath = flag.String("remoteWrite.tmpDataPath", "vmagent-remotewrite-data", "Path to directory for storing pending data, which isn't sent to the configured -remoteWrite.url . "+
+		"even distribution of series over the specified -remoteWrite.url systems. See also -remoteWrite.shardByURL.inverseLabels.")
+	shardByURLLabelsInverse = flag.Bool("remoteWrite.shardByURL.inverseLabels", false, "Inverse the behavior of -remoteWrite.shardByURL.labels so that series are sharded using all labels except the ones specified in -remoteWrite.shardByURL.labels.")
+	tmpDataPath             = flag.String("remoteWrite.tmpDataPath", "vmagent-remotewrite-data", "Path to directory for storing pending data, which isn't sent to the configured -remoteWrite.url . "+
 		"See also -remoteWrite.maxDiskUsagePerURL and -remoteWrite.disableOnDiskQueue")
 	keepDanglingQueues = flag.Bool("remoteWrite.keepDanglingQueues", false, "Keep persistent queues contents at -remoteWrite.tmpDataPath in case there are no matching -remoteWrite.url. "+
 		"Useful when -remoteWrite.url is changed temporarily and persistent queue files will be needed later on.")
@@ -538,15 +539,7 @@ func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmar
 		tssByURL := make([][]prompbmarshal.TimeSeries, len(rwctxs))
 		tmpLabels := promutils.GetLabels()
 		for _, ts := range tssBlock {
-			hashLabels := ts.Labels
-			if len(shardByURLLabelsMap) > 0 {
-				hashLabels = tmpLabels.Labels[:0]
-				for _, label := range ts.Labels {
-					if _, ok := shardByURLLabelsMap[label.Name]; ok {
-						hashLabels = append(hashLabels, label)
-					}
-				}
-			}
+			hashLabels := extractShardingLabels(tmpLabels.Labels, ts.Labels, *shardByURLLabelsInverse)
 			h := getLabelsHash(hashLabels)
 			idx := h % uint64(len(tssByURL))
 			tssByURL[idx] = append(tssByURL[idx], ts)
@@ -592,6 +585,20 @@ func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmar
 	return !anyPushFailed.Load()
 }
 
+func extractShardingLabels(dst, src []prompbmarshal.Label, inverse bool) []prompbmarshal.Label {
+	if len(shardByURLLabelsMap) < 1 {
+		return src
+	}
+	dst = dst[:0]
+	for _, label := range src {
+		_, ok := shardByURLLabelsMap[label.Name]
+		if ok && !inverse || !ok && inverse {
+			dst = append(dst, label)
+		}
+	}
+	return dst
+}
+
 // sortLabelsIfNeeded sorts labels if -sortLabels command-line flag is set.
 func sortLabelsIfNeeded(tss []prompbmarshal.TimeSeries) {
 	if !*sortLabels {
diff --git a/app/vmalert/remotewrite/client.go b/app/vmalert/remotewrite/client.go
index de9ac6abc..80682a72b 100644
--- a/app/vmalert/remotewrite/client.go
+++ b/app/vmalert/remotewrite/client.go
@@ -151,12 +151,22 @@ func (c *Client) run(ctx context.Context) {
 	ticker := time.NewTicker(c.flushInterval)
 	wr := &prompbmarshal.WriteRequest{}
 	shutdown := func() {
+		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
+		logger.Infof("shutting down remote write client and flushing remained series")
+
+		shutdownFlushCnt := 0
 		for ts := range c.input {
 			wr.Timeseries = append(wr.Timeseries, ts)
+			if len(wr.Timeseries) >= c.maxBatchSize {
+				shutdownFlushCnt += len(wr.Timeseries)
+				c.flush(lastCtx, wr)
+			}
 		}
-		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
-		logger.Infof("shutting down remote write client and flushing remained %d series", len(wr.Timeseries))
+		// flush the last batch. `flush` will re-check and avoid flushing empty batch.
+		shutdownFlushCnt += len(wr.Timeseries)
 		c.flush(lastCtx, wr)
+
+		logger.Infof("shutting down remote write client flushed %d series", shutdownFlushCnt)
 		cancel()
 	}
 	c.wg.Add(1)
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 9d40dc9ad..3b373e8b4 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -54,11 +54,14 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * FEATURE: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): expose `vm_streamaggr_flush_timeouts_total` and `vm_streamaggr_dedup_flush_timeouts_total` [counters](https://docs.victoriametrics.com/keyconcepts/#counter) at [`/metrics` page](https://docs.victoriametrics.com/#monitoring), which can be used for detecting flush timeouts for stream aggregation states. Expose also `vm_streamaggr_flush_duration_seconds` and `vm_streamaggr_dedup_flush_duration_seconds` [histograms](https://docs.victoriametrics.com/keyconcepts/#histogram) for monitoring the real flush durations of stream aggregation states.
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): improve trace display for better visual separation of branches. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5926).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): use the provided `-remoteWrite.tlsServerName` as `Host` header in requests to `-remoteWrite.url`. This allows sending data to https remote storage by IP address instead of hostname. Thanks to @minor-fixes for initial idea and [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5802).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add `-remoteWrite.shardByURL.inverseLabels` cmd-line flag to enhance sharding logic across configured `-remoteWrite.url` URLs when `-remoteWrite.shardByURL.labels` is set. Thanks to @edma2 for the idea and [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5938).
 * FEATURE: optimize [`/api/v1/labels`](https://docs.victoriametrics.com/url-examples/#apiv1labels) and [`/api/v1/label/.../values`](https://docs.victoriametrics.com/url-examples/#apiv1labelvalues) when `match[]` filters contains metric name. For example, `/api/v1/label/instance/values?match[]=up` now works much faster than before. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5055).
 * FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl.html): support client-side TLS configuration for [native protocol](https://docs.victoriametrics.com/vmctl/#migrating-data-from-victoriametrics). See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5748). Thanks to @khushijain21 for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5824).
 * FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl.html): support client-side TLS configuration for VictoriaMetrics destination specified via `--vm-*` cmd-line flags used in [InfluxDB](https://docs.victoriametrics.com/vmctl/#migrating-data-from-influxdb-1x), [Remote Read protocol](https://docs.victoriametrics.com/vmctl/#migrating-data-by-remote-read-protocol), [OpenTSDB](https://docs.victoriametrics.com/vmctl/#migrating-data-from-opentsdb), [Prometheus](https://docs.victoriametrics.com/vmctl/#migrating-data-from-prometheus) and [Promscale](https://docs.victoriametrics.com/vmctl/#migrating-data-from-promscale) migration modes.
 * FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl.html): split [explore phase](https://docs.victoriametrics.com/vmctl/#migrating-data-from-victoriametrics) in `vm-native` mode by time intervals when [--vm-native-step-interval](https://docs.victoriametrics.com/vmctl/#using-time-based-chunking-of-migration) is specified. This should reduce probability of exceeding complexity limits for number of selected series during explore phase. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5369).
 * FEATURE: [graphite](https://docs.victoriametrics.com/#graphite-render-api-usage): add support for `aggregateSeriesLists`, `diffSeriesLists`, `multiplySeriesLists` and `sumSeriesLists` functions. Thanks to @rbizos for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5809).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): added command line argument that enables OpenTelementry metric names and labels sanitization.
+
 
 * BUGFIX: prevent from automatic deletion of newly registered time series when it is queried immediately after the addition. The probability of this bug has been increased significantly after [v1.99.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.99.0) because of optimizations related to registering new time series. See [this](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5948) and [this](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5959) issue.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent.html): properly set `Host` header in requests to scrape targets if it is specified via [`headers` option](https://docs.victoriametrics.com/sd_configs/#http-api-client-options). Thanks to @fholzer for [the bugreport](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5969) and [the fix](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5970).
@@ -72,6 +75,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * BUGFIX: properly wait for force merge to be completed during the shutdown. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5944) for the details.
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert.html): set correct `endsAt` value in notifications sent to the Alertmanager. Previously, a rule with evaluation intervals lower than 10s could never be triggered. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5995) for details.
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert.html): properly account for `-rule.resendDelay` for alerting rules that are constantly switching state from inactive to firing. Before, notifications for such rules could have been skipped if state change happened more often than `-rule.resendDelay`. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6028) for details.
+* BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert.html): respect `-remoteWrite.maxBatchSize` at shutdown period. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6025).
 
 ## [v1.99.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.99.0)
 
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 9508806c0..de59e0e3d 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -1557,6 +1557,9 @@ VictoriaMetrics supports data ingestion via [OpenTelemetry protocol for metrics]
 VictoriaMetrics expects `protobuf`-encoded requests at `/opentelemetry/v1/metrics`.
 Set HTTP request header `Content-Encoding: gzip` when sending gzip-compressed data to `/opentelemetry/v1/metrics`.
 
+VictoriaMetrics automatically does not sanitize metric names for the data ingested via OpenTelemetry protocol
+If you need accepting metric and label names as is with sanitizing, then pass `-opentelemetry.sanitizeMetrics=true` command-line flag to VictoriaMetrics.
+
 See [How to use OpenTelemetry metrics with VictoriaMetrics](https://docs.victoriametrics.com/guides/getting-started-with-opentelemetry/).
 
 ## JSON line format
diff --git a/docs/vmagent.md b/docs/vmagent.md
index f24ff2a71..c19b09077 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -191,7 +191,8 @@ among remote storage systems specified in `-remoteWrite.url`. Sometimes it may b
 set of labels for sharding. For example, it may be needed to route all the metrics with the same `instance` label
 to the same `-remoteWrite.url`. In this case you can specify comma-separated list of these labels in the `-remoteWrite.shardByURL.labels`
 command-line flag. For example, `-remoteWrite.shardByURL.labels=instance,__name__` would shard metrics with the same name and `instance`
-label to the same `-remoteWrite.url`.
+label to the same `-remoteWrite.url`. The sharding logic can be inverted by specifying `-remoteWrite.shardByURL.inverseLabels`,
+so that metrics are sharded using all labels except the ones specified in `-remoteWrite.shardByURL.labels`.
 
 See also [how to scrape big number of targets](#scraping-big-number-of-targets).
 
@@ -2105,8 +2106,10 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
      Empty values are set to default value.
   -remoteWrite.shardByURL
      Whether to shard outgoing series across all the remote storage systems enumerated via -remoteWrite.url . By default the data is replicated across all the -remoteWrite.url . See https://docs.victoriametrics.com/vmagent.html#sharding-among-remote-storages
+  -remoteWrite.shardByURL.inverseLabels
+      Inverse the behavior of -remoteWrite.shardByURL.labels so that series are sharded using all labels except the ones specified in -remoteWrite.shardByURL.labels.
   -remoteWrite.shardByURL.labels array
-     Optional list of labels, which must be used for sharding outgoing samples among remote storage systems if -remoteWrite.shardByURL command-line flag is set. By default all the labels are used for sharding in order to gain even distribution of series over the specified -remoteWrite.url systems
+     Optional list of labels, which must be used for sharding outgoing samples among remote storage systems if -remoteWrite.shardByURL command-line flag is set. By default all the labels are used for sharding in order to gain even distribution of series over the specified -remoteWrite.url systems. See also -remoteWrite.shardByURL.inverseLabels.
      Supports an array of values separated by comma or specified via multiple flags.
      Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
   -remoteWrite.showURL
diff --git a/lib/promrelabel/relabel.go b/lib/promrelabel/relabel.go
index ba0a896ab..2e306fe70 100644
--- a/lib/promrelabel/relabel.go
+++ b/lib/promrelabel/relabel.go
@@ -663,6 +663,11 @@ func SanitizeLabelName(name string) string {
 	return labelNameSanitizer.Transform(name)
 }
 
+// SanitizeLabelNameParts returns label name slice generated from metric name divided by unsupported characters
+func SanitizeLabelNameParts(name string) []string {
+	return unsupportedLabelNameChars.Split(name, -1)
+}
+
 var labelNameSanitizer = bytesutil.NewFastStringTransformer(func(s string) string {
 	return unsupportedLabelNameChars.ReplaceAllString(s, "_")
 })
diff --git a/lib/protoparser/opentelemetry/pb/pb.go b/lib/protoparser/opentelemetry/pb/pb.go
index 11d9c8291..883950e9f 100644
--- a/lib/protoparser/opentelemetry/pb/pb.go
+++ b/lib/protoparser/opentelemetry/pb/pb.go
@@ -190,6 +190,7 @@ func (sm *ScopeMetrics) unmarshalProtobuf(src []byte) (err error) {
 // Metric represents the corresponding OTEL protobuf message
 type Metric struct {
 	Name      string
+	Unit      string
 	Gauge     *Gauge
 	Sum       *Sum
 	Histogram *Histogram
@@ -198,6 +199,7 @@ type Metric struct {
 
 func (m *Metric) marshalProtobuf(mm *easyproto.MessageMarshaler) {
 	mm.AppendString(1, m.Name)
+	mm.AppendString(3, m.Unit)
 	switch {
 	case m.Gauge != nil:
 		m.Gauge.marshalProtobuf(mm.AppendMessage(5))
@@ -213,6 +215,7 @@ func (m *Metric) marshalProtobuf(mm *easyproto.MessageMarshaler) {
 func (m *Metric) unmarshalProtobuf(src []byte) (err error) {
 	// message Metric {
 	//   string name = 1;
+	//   string unit = 3;
 	//   oneof data {
 	//     Gauge gauge = 5;
 	//     Sum sum = 7;
@@ -233,6 +236,12 @@ func (m *Metric) unmarshalProtobuf(src []byte) (err error) {
 				return fmt.Errorf("cannot read metric name")
 			}
 			m.Name = strings.Clone(name)
+		case 3:
+			unit, ok := fc.String()
+			if !ok {
+				return fmt.Errorf("cannot read metric unit")
+			}
+			m.Unit = strings.Clone(unit)
 		case 5:
 			data, ok := fc.MessageData()
 			if !ok {
@@ -617,6 +626,7 @@ func (ndp *NumberDataPoint) unmarshalProtobuf(src []byte) (err error) {
 type Sum struct {
 	DataPoints             []*NumberDataPoint
 	AggregationTemporality AggregationTemporality
+	IsMonotonic            bool
 }
 
 // AggregationTemporality represents the corresponding OTEL protobuf enum
@@ -636,6 +646,7 @@ func (s *Sum) marshalProtobuf(mm *easyproto.MessageMarshaler) {
 		dp.marshalProtobuf(mm.AppendMessage(1))
 	}
 	mm.AppendInt64(2, int64(s.AggregationTemporality))
+	mm.AppendBool(3, s.IsMonotonic)
 }
 
 func (s *Sum) unmarshalProtobuf(src []byte) (err error) {
@@ -666,6 +677,12 @@ func (s *Sum) unmarshalProtobuf(src []byte) (err error) {
 				return fmt.Errorf("cannot read AggregationTemporality")
 			}
 			s.AggregationTemporality = AggregationTemporality(at)
+		case 3:
+			im, ok := fc.Bool()
+			if !ok {
+				return fmt.Errorf("cannot read IsMonotonic")
+			}
+			s.IsMonotonic = im
 		}
 	}
 	return nil
diff --git a/lib/protoparser/opentelemetry/stream/streamparser.go b/lib/protoparser/opentelemetry/stream/streamparser.go
index d0b18c7ce..92aeab67a 100644
--- a/lib/protoparser/opentelemetry/stream/streamparser.go
+++ b/lib/protoparser/opentelemetry/stream/streamparser.go
@@ -1,10 +1,13 @@
 package stream
 
 import (
+	"flag"
 	"fmt"
 	"io"
 	"strconv"
+	"strings"
 	"sync"
+	"unicode"
 
 	"github.com/VictoriaMetrics/metrics"
 
@@ -13,11 +16,72 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/pb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 )
 
+var (
+	// sanitizeMetrics controls sanitizing metric and label names ingested via OpenTelemetry protocol.
+	sanitizeMetrics = flag.Bool("opentelemetry.sanitizeMetrics", false, "Sanitize metric and label names for the ingested OpenTelemetry data")
+)
+
+// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/b8655058501bed61a06bb660869051491f46840b/pkg/translator/prometheus/normalize_name.go#L19
+var unitMap = []struct {
+	prefix string
+	units  map[string]string
+}{
+	{
+		units: map[string]string{
+			// Time
+			"d":   "days",
+			"h":   "hours",
+			"min": "minutes",
+			"s":   "seconds",
+			"ms":  "milliseconds",
+			"us":  "microseconds",
+			"ns":  "nanoseconds",
+
+			// Bytes
+			"By":   "bytes",
+			"KiBy": "kibibytes",
+			"MiBy": "mebibytes",
+			"GiBy": "gibibytes",
+			"TiBy": "tibibytes",
+			"KBy":  "kilobytes",
+			"MBy":  "megabytes",
+			"GBy":  "gigabytes",
+			"TBy":  "terabytes",
+
+			// SI
+			"m": "meters",
+			"V": "volts",
+			"A": "amperes",
+			"J": "joules",
+			"W": "watts",
+			"g": "grams",
+
+			// Misc
+			"Cel": "celsius",
+			"Hz":  "hertz",
+			"1":   "",
+			"%":   "percent",
+		},
+	}, {
+		prefix: "per",
+		units: map[string]string{
+			"s":  "second",
+			"m":  "minute",
+			"h":  "hour",
+			"d":  "day",
+			"w":  "week",
+			"mo": "month",
+			"y":  "year",
+		},
+	},
+}
+
 // ParseStream parses OpenTelemetry protobuf or json data from r and calls callback for the parsed rows.
 //
 // callback shouldn't hold tss items after returning.
@@ -58,10 +122,11 @@ func (wr *writeContext) appendSamplesFromScopeMetrics(sc *pb.ScopeMetrics) {
 			// skip metrics without names
 			continue
 		}
+		metricName := sanitizeMetricName(m)
 		switch {
 		case m.Gauge != nil:
 			for _, p := range m.Gauge.DataPoints {
-				wr.appendSampleFromNumericPoint(m.Name, p)
+				wr.appendSampleFromNumericPoint(metricName, p)
 			}
 		case m.Sum != nil:
 			if m.Sum.AggregationTemporality != pb.AggregationTemporalityCumulative {
@@ -69,11 +134,11 @@ func (wr *writeContext) appendSamplesFromScopeMetrics(sc *pb.ScopeMetrics) {
 				continue
 			}
 			for _, p := range m.Sum.DataPoints {
-				wr.appendSampleFromNumericPoint(m.Name, p)
+				wr.appendSampleFromNumericPoint(metricName, p)
 			}
 		case m.Summary != nil:
 			for _, p := range m.Summary.DataPoints {
-				wr.appendSamplesFromSummary(m.Name, p)
+				wr.appendSamplesFromSummary(metricName, p)
 			}
 		case m.Histogram != nil:
 			if m.Histogram.AggregationTemporality != pb.AggregationTemporalityCumulative {
@@ -81,11 +146,11 @@ func (wr *writeContext) appendSamplesFromScopeMetrics(sc *pb.ScopeMetrics) {
 				continue
 			}
 			for _, p := range m.Histogram.DataPoints {
-				wr.appendSamplesFromHistogram(m.Name, p)
+				wr.appendSamplesFromHistogram(metricName, p)
 			}
 		default:
 			rowsDroppedUnsupportedMetricType.Inc()
-			logger.Warnf("unsupported type for metric %q", m.Name)
+			logger.Warnf("unsupported type for metric %q", metricName)
 		}
 	}
 }
@@ -209,7 +274,7 @@ func (wr *writeContext) appendSampleWithExtraLabel(metricName, labelName, labelV
 func appendAttributesToPromLabels(dst []prompbmarshal.Label, attributes []*pb.KeyValue) []prompbmarshal.Label {
 	for _, at := range attributes {
 		dst = append(dst, prompbmarshal.Label{
-			Name:  at.Key,
+			Name:  sanitizeLabelName(at.Key),
 			Value: at.Value.FormatString(),
 		})
 	}
@@ -290,6 +355,74 @@ func (wr *writeContext) parseRequestToTss(req *pb.ExportMetricsServiceRequest) {
 	}
 }
 
+// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/b8655058501bed61a06bb660869051491f46840b/pkg/translator/prometheus/normalize_label.go#L26
+func sanitizeLabelName(labelName string) string {
+	if !*sanitizeMetrics {
+		return labelName
+	}
+	if len(labelName) == 0 {
+		return labelName
+	}
+	labelName = promrelabel.SanitizeLabelName(labelName)
+	if unicode.IsDigit(rune(labelName[0])) {
+		return "key_" + labelName
+	} else if strings.HasPrefix(labelName, "_") && !strings.HasPrefix(labelName, "__") {
+		return "key" + labelName
+	}
+	return labelName
+}
+
+// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/b8655058501bed61a06bb660869051491f46840b/pkg/translator/prometheus/normalize_name.go#L83
+func sanitizeMetricName(metric *pb.Metric) string {
+	if !*sanitizeMetrics {
+		return metric.Name
+	}
+	nameTokens := promrelabel.SanitizeLabelNameParts(metric.Name)
+	unitTokens := strings.SplitN(metric.Unit, "/", len(unitMap))
+	for i, u := range unitTokens {
+		unitToken := strings.TrimSpace(u)
+		if unitToken == "" || strings.ContainsAny(unitToken, "{}") {
+			continue
+		}
+		if unit, ok := unitMap[i].units[unitToken]; ok {
+			unitToken = unit
+		}
+		if unitToken != "" && !containsToken(nameTokens, unitToken) {
+			unitPrefix := unitMap[i].prefix
+			if unitPrefix != "" {
+				nameTokens = append(nameTokens, unitPrefix, unitToken)
+			} else {
+				nameTokens = append(nameTokens, unitToken)
+			}
+		}
+	}
+	if metric.Sum != nil && metric.Sum.IsMonotonic {
+		nameTokens = moveOrAppend(nameTokens, "total")
+	} else if metric.Unit == "1" && metric.Gauge != nil {
+		nameTokens = moveOrAppend(nameTokens, "ratio")
+	}
+	return strings.Join(nameTokens, "_")
+}
+
+func containsToken(tokens []string, value string) bool {
+	for _, token := range tokens {
+		if token == value {
+			return true
+		}
+	}
+	return false
+}
+
+func moveOrAppend(tokens []string, value string) []string {
+	for t := range tokens {
+		if tokens[t] == value {
+			tokens = append(tokens[:t], tokens[t+1:]...)
+			break
+		}
+	}
+	return append(tokens, value)
+}
+
 var wrPool sync.Pool
 
 func getWriteContext() *writeContext {


Test patch:
diff --git a/app/vmagent/remotewrite/remotewrite_test.go b/app/vmagent/remotewrite/remotewrite_test.go
new file mode 100644
index 000000000..b5634772a
--- /dev/null
+++ b/app/vmagent/remotewrite/remotewrite_test.go
@@ -0,0 +1,38 @@
+package remotewrite
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+)
+
+func TestExtractShardingLabels(t *testing.T) {
+	shardByURLLabelsMap = make(map[string]struct{})
+	shardByURLLabelsMap["instance"] = struct{}{}
+	shardByURLLabelsMap["job"] = struct{}{}
+	defer func() {
+		shardByURLLabelsMap = nil
+	}()
+
+	f := func(in, exp []prompbmarshal.Label, inverse bool) {
+		t.Helper()
+		var got []prompbmarshal.Label
+		got = extractShardingLabels(got, in, inverse)
+		if !reflect.DeepEqual(got, exp) {
+			t.Fatalf("expected to get \n%#v; \ngot \n%#v instead", exp, got)
+		}
+	}
+
+	f(nil, nil, true)
+	f(nil, nil, false)
+
+	f([]prompbmarshal.Label{{Name: "foo"}}, nil, false)
+	f([]prompbmarshal.Label{{Name: "foo"}}, []prompbmarshal.Label{{Name: "foo"}}, true)
+
+	f([]prompbmarshal.Label{{Name: "foo"}, {Name: "job"}}, []prompbmarshal.Label{{Name: "job"}}, false)
+	f([]prompbmarshal.Label{{Name: "foo"}, {Name: "job"}}, []prompbmarshal.Label{{Name: "foo"}}, true)
+
+	f([]prompbmarshal.Label{{Name: "foo"}, {Name: "instance"}, {Name: "job"}}, []prompbmarshal.Label{{Name: "instance"}, {Name: "job"}}, false)
+	f([]prompbmarshal.Label{{Name: "foo"}, {Name: "instance"}, {Name: "job"}}, []prompbmarshal.Label{{Name: "foo"}}, true)
+}
diff --git a/app/vmalert/remotewrite/client_test.go b/app/vmalert/remotewrite/client_test.go
index 76ab3fc92..c88dcb5ce 100644
--- a/app/vmalert/remotewrite/client_test.go
+++ b/app/vmalert/remotewrite/client_test.go
@@ -84,6 +84,70 @@ func TestClient_Push(t *testing.T) {
 	}
 }
 
+func TestClient_run_maxBatchSizeDuringShutdown(t *testing.T) {
+	batchSize := 20
+
+	testTable := []struct {
+		name     string // name of the test case
+		pushCnt  int    // how many time series is pushed to the client
+		batchCnt int    // the expected batch count sent by the client
+	}{
+		{
+			name:     "pushCnt % batchSize == 0",
+			pushCnt:  batchSize * 40,
+			batchCnt: 40,
+		},
+		{
+			name:     "pushCnt % batchSize != 0",
+			pushCnt:  batchSize*40 + 1,
+			batchCnt: 40 + 1,
+		},
+	}
+
+	for _, tt := range testTable {
+		t.Run(tt.name, func(t *testing.T) {
+			// run new server
+			bcServer := newBatchCntRWServer()
+
+			// run new client
+			rwClient, err := NewClient(context.Background(), Config{
+				MaxBatchSize: batchSize,
+
+				// Set everything to 1 to simplify the calculation.
+				Concurrency:   1,
+				MaxQueueSize:  1000,
+				FlushInterval: time.Minute,
+
+				// batch count server
+				Addr: bcServer.URL,
+			})
+			if err != nil {
+				t.Fatalf("new remote write client failed, err: %v", err)
+			}
+
+			// push time series to the client.
+			for i := 0; i < tt.pushCnt; i++ {
+				if err = rwClient.Push(prompbmarshal.TimeSeries{}); err != nil {
+					t.Fatalf("push time series to the client failed, err: %v", err)
+				}
+			}
+
+			// close the client so the rest ts will be flushed in `shutdown`
+			if err = rwClient.Close(); err != nil {
+				t.Fatalf("shutdown client failed, err: %v", err)
+			}
+
+			// finally check how many batches is sent.
+			if tt.batchCnt != bcServer.acceptedBatches() {
+				t.Errorf("client sent batch count incorrect, want: %d, get: %d", tt.batchCnt, bcServer.acceptedBatches())
+			}
+			if tt.pushCnt != bcServer.accepted() {
+				t.Errorf("client sent time series count incorrect, want: %d, get: %d", tt.pushCnt, bcServer.accepted())
+			}
+		})
+	}
+}
+
 func newRWServer() *rwServer {
 	rw := &rwServer{}
 	rw.Server = httptest.NewServer(http.HandlerFunc(rw.handler))
@@ -184,3 +248,27 @@ func (frw *faultyRWServer) handler(w http.ResponseWriter, r *http.Request) {
 		w.Write([]byte("server overloaded"))
 	}
 }
+
+type batchCntRWServer struct {
+	*rwServer
+
+	batchCnt atomic.Int64 // accepted batch count, which also equals to request count
+}
+
+func newBatchCntRWServer() *batchCntRWServer {
+	bc := &batchCntRWServer{
+		rwServer: &rwServer{},
+	}
+
+	bc.Server = httptest.NewServer(http.HandlerFunc(bc.handler))
+	return bc
+}
+
+func (bc *batchCntRWServer) handler(w http.ResponseWriter, r *http.Request) {
+	bc.batchCnt.Add(1)
+	bc.rwServer.handler(w, r)
+}
+
+func (bc *batchCntRWServer) acceptedBatches() int {
+	return int(bc.batchCnt.Load())
+}
diff --git a/lib/protoparser/opentelemetry/stream/streamparser_test.go b/lib/protoparser/opentelemetry/stream/streamparser_test.go
index 1d5281d4a..242cb39f3 100644
--- a/lib/protoparser/opentelemetry/stream/streamparser_test.go
+++ b/lib/protoparser/opentelemetry/stream/streamparser_test.go
@@ -15,8 +15,9 @@ import (
 )
 
 func TestParseStream(t *testing.T) {
-	f := func(samples []*pb.Metric, tssExpected []prompbmarshal.TimeSeries) {
+	f := func(samples []*pb.Metric, tssExpected []prompbmarshal.TimeSeries, sanitize bool) {
 		t.Helper()
+		*sanitizeMetrics = sanitize
 
 		checkSeries := func(tss []prompbmarshal.TimeSeries) error {
 			if len(tss) != len(tssExpected) {
@@ -86,10 +87,10 @@ func TestParseStream(t *testing.T) {
 	// Test all metric types
 	f(
 		[]*pb.Metric{
-			generateGauge("my-gauge"),
-			generateHistogram("my-histogram"),
-			generateSum("my-sum"),
-			generateSummary("my-summary"),
+			generateGauge("my-gauge", ""),
+			generateHistogram("my-histogram", ""),
+			generateSum("my-sum", "", false),
+			generateSummary("my-summary", ""),
 		},
 		[]prompbmarshal.TimeSeries{
 			newPromPBTs("my-gauge", 15000, 15.0, jobLabelValue, kvLabel("label1", "value1")),
@@ -106,16 +107,85 @@ func TestParseStream(t *testing.T) {
 			newPromPBTs("my-summary", 35000, 7.5, jobLabelValue, kvLabel("label6", "value6"), kvLabel("quantile", "0.1")),
 			newPromPBTs("my-summary", 35000, 10.0, jobLabelValue, kvLabel("label6", "value6"), kvLabel("quantile", "0.5")),
 			newPromPBTs("my-summary", 35000, 15.0, jobLabelValue, kvLabel("label6", "value6"), kvLabel("quantile", "1")),
-		})
+		},
+		false,
+	)
 
 	// Test gauge
 	f(
 		[]*pb.Metric{
-			generateGauge("my-gauge"),
+			generateGauge("my-gauge", ""),
 		},
 		[]prompbmarshal.TimeSeries{
 			newPromPBTs("my-gauge", 15000, 15.0, jobLabelValue, kvLabel("label1", "value1")),
 		},
+		false,
+	)
+
+	// Test gauge with unit and sanitization
+	f(
+		[]*pb.Metric{
+			generateGauge("my-gauge", "ms"),
+		},
+		[]prompbmarshal.TimeSeries{
+			newPromPBTs("my_gauge_milliseconds", 15000, 15.0, jobLabelValue, kvLabel("label1", "value1")),
+		},
+		true,
+	)
+
+	// Test gauge with unit inside metric
+	f(
+		[]*pb.Metric{
+			generateGauge("my-gauge-milliseconds", "ms"),
+		},
+		[]prompbmarshal.TimeSeries{
+			newPromPBTs("my_gauge_milliseconds", 15000, 15.0, jobLabelValue, kvLabel("label1", "value1")),
+		},
+		true,
+	)
+
+	// Test gauge with ratio suffix
+	f(
+		[]*pb.Metric{
+			generateGauge("my-gauge-milliseconds", "1"),
+		},
+		[]prompbmarshal.TimeSeries{
+			newPromPBTs("my_gauge_milliseconds_ratio", 15000, 15.0, jobLabelValue, kvLabel("label1", "value1")),
+		},
+		true,
+	)
+
+	// Test sum with total suffix
+	f(
+		[]*pb.Metric{
+			generateSum("my-sum", "ms", true),
+		},
+		[]prompbmarshal.TimeSeries{
+			newPromPBTs("my_sum_milliseconds_total", 150000, 15.5, jobLabelValue, kvLabel("label5", "value5")),
+		},
+		true,
+	)
+
+	// Test sum with total suffix, which exists in a metric name
+	f(
+		[]*pb.Metric{
+			generateSum("my-total-sum", "ms", true),
+		},
+		[]prompbmarshal.TimeSeries{
+			newPromPBTs("my_sum_milliseconds_total", 150000, 15.5, jobLabelValue, kvLabel("label5", "value5")),
+		},
+		true,
+	)
+
+	// Test sum with total and complex suffix
+	f(
+		[]*pb.Metric{
+			generateSum("my-total-sum", "m/s", true),
+		},
+		[]prompbmarshal.TimeSeries{
+			newPromPBTs("my_sum_meters_per_second_total", 150000, 15.5, jobLabelValue, kvLabel("label5", "value5")),
+		},
+		true,
 	)
 }
 
@@ -152,7 +222,7 @@ func attributesFromKV(k, v string) []*pb.KeyValue {
 	}
 }
 
-func generateGauge(name string) *pb.Metric {
+func generateGauge(name, unit string) *pb.Metric {
 	n := int64(15)
 	points := []*pb.NumberDataPoint{
 		{
@@ -163,13 +233,14 @@ func generateGauge(name string) *pb.Metric {
 	}
 	return &pb.Metric{
 		Name: name,
+		Unit: unit,
 		Gauge: &pb.Gauge{
 			DataPoints: points,
 		},
 	}
 }
 
-func generateHistogram(name string) *pb.Metric {
+func generateHistogram(name, unit string) *pb.Metric {
 	points := []*pb.HistogramDataPoint{
 		{
 
@@ -183,6 +254,7 @@ func generateHistogram(name string) *pb.Metric {
 	}
 	return &pb.Metric{
 		Name: name,
+		Unit: unit,
 		Histogram: &pb.Histogram{
 			AggregationTemporality: pb.AggregationTemporalityCumulative,
 			DataPoints:             points,
@@ -190,7 +262,7 @@ func generateHistogram(name string) *pb.Metric {
 	}
 }
 
-func generateSum(name string) *pb.Metric {
+func generateSum(name, unit string, isMonotonic bool) *pb.Metric {
 	d := float64(15.5)
 	points := []*pb.NumberDataPoint{
 		{
@@ -201,14 +273,16 @@ func generateSum(name string) *pb.Metric {
 	}
 	return &pb.Metric{
 		Name: name,
+		Unit: unit,
 		Sum: &pb.Sum{
 			AggregationTemporality: pb.AggregationTemporalityCumulative,
 			DataPoints:             points,
+			IsMonotonic:            isMonotonic,
 		},
 	}
 }
 
-func generateSummary(name string) *pb.Metric {
+func generateSummary(name, unit string) *pb.Metric {
 	points := []*pb.SummaryDataPoint{
 		{
 			Attributes:   attributesFromKV("label6", "value6"),
@@ -233,6 +307,7 @@ func generateSummary(name string) *pb.Metric {
 	}
 	return &pb.Metric{
 		Name: name,
+		Unit: unit,
 		Summary: &pb.Summary{
 			DataPoints: points,
 		},
diff --git a/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go b/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go
index e1f73aaf1..2915786cd 100644
--- a/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go
+++ b/lib/protoparser/opentelemetry/stream/streamparser_timing_test.go
@@ -10,10 +10,10 @@ import (
 
 func BenchmarkParseStream(b *testing.B) {
 	samples := []*pb.Metric{
-		generateGauge("my-gauge"),
-		generateHistogram("my-histogram"),
-		generateSum("my-sum"),
-		generateSummary("my-summary"),
+		generateGauge("my-gauge", ""),
+		generateHistogram("my-histogram", ""),
+		generateSum("my-sum", "", false),
+		generateSummary("my-summary", ""),
 	}
 	b.SetBytes(1)
 	b.ReportAllocs()