From 5f13ba8631cac3afc0eb1aa17fa8836990188192 Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Thu, 4 Nov 2021 15:00:51 +0200
Subject: [PATCH] app/vmagent/remotewrite: send data to remote storage systems
 in parallel

This should improve data ingestion speed many `-remoteWrite.url` options are configured
---
 app/vmagent/remotewrite/remotewrite.go | 23 ++++++++++++++++++-----
 docs/CHANGELOG.md                      |  1 +
 2 files changed, 19 insertions(+), 5 deletions(-)

diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go
index 434fe02284..0214f239cc 100644
--- a/app/vmagent/remotewrite/remotewrite.go
+++ b/app/vmagent/remotewrite/remotewrite.go
@@ -301,11 +301,7 @@ func PushWithAuthToken(at *auth.Token, wr *prompbmarshal.WriteRequest) {
 		}
 		sortLabelsIfNeeded(tssBlock)
 		tssBlock = limitSeriesCardinality(tssBlock)
-		if len(tssBlock) > 0 {
-			for _, rwctx := range rwctxs {
-				rwctx.Push(tssBlock)
-			}
-		}
+		pushBlockToRemoteStorages(rwctxs, tssBlock)
 		if rctx != nil {
 			rctx.reset()
 		}
@@ -315,6 +311,23 @@ func PushWithAuthToken(at *auth.Token, wr *prompbmarshal.WriteRequest) {
 	}
 }
 
+func pushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmarshal.TimeSeries) {
+	if len(tssBlock) == 0 {
+		// Nothing to push
+		return
+	}
+	// Push block to remote storages in parallel in order to reduce the time needed for sending the data to multiple remote storage systems.
+	var wg sync.WaitGroup
+	for _, rwctx := range rwctxs {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			rwctx.Push(tssBlock)
+		}()
+		wg.Wait()
+	}
+}
+
 // sortLabelsIfNeeded sorts labels if -sortLabels command-line flag is set.
 func sortLabelsIfNeeded(tss []prompbmarshal.TimeSeries) {
 	if !*sortLabels {
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 573df4a08c..1a34e532b0 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -11,6 +11,7 @@ sort: 15
 * FEATURE: automatically detect timestamp precision (ns, us, ms or s) for the data ingested into VictoriaMetrics via [InfluxDB line protocol](https://docs.victoriametrics.com/#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf).
 * FEATURE: vmagent: add ability to protect `/config` page with auth key via `-configAuthKey` command-line flag. This page may contain sensitive information such as passwords, so it may be good to restrict access to this page. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1764).
 * FEATURE: vmagent: add `-promscrape.maxResponseHeadersSize` command-line flag for tuning the maximum HTTP response headers size for Prometheus scrape targets.
+* FEATURE: vmagent: send data to multiple configured remote storage systems in parallel (e.g. when multiple `-remoteWrite.url` flag values are specified). This should improve data ingestion speed.
 * FEATURE: add [label_graphite_group](https://docs.victoriametrics.com/MetricsQL.html#label_graphite_group) function for extracting the given groups from Graphite metric names.
 * FEATURE: add [limit_offset](https://docs.victoriametrics.com/MetricsQL.html#limit_offset) function, which can be used for implementing simple paging over big number of time series. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1778).