app/vmagent: expose vmagent_remotewrite_global_rows_pushed_before_relabel and vmagent_remotewrite_rows_pushed_after_relabel_total metrics

app/vmagent/remotewrite/remotewrite.go

@@ -274,6 +274,8 @@ func PushWithAuthToken(at *auth.Token, wr *prompbmarshal.WriteRequest) {
 		rctx = getRelabelCtx()
 	}
 	tss := wr.Timeseries
+	rowsCount := getRowsCount(tss)
+	globalRowsPushedBeforeRelabel.Add(rowsCount)
 	maxSamplesPerBlock := *maxRowsPerBlock
 	// Allow up to 10x of labels per each block on average.
 	maxLabelsPerBlock := 10 * maxSamplesPerBlock
@@ -298,9 +300,10 @@ func PushWithAuthToken(at *auth.Token, wr *prompbmarshal.WriteRequest) {
 			tss = nil
 		}
 		if rctx != nil {
-			tssBlockLen := len(tssBlock)
+			rowsCountBeforeRelabel := getRowsCount(tssBlock)
 			tssBlock = rctx.applyRelabeling(tssBlock, labelsGlobal, pcsGlobal)
-			globalRelabelMetricsDropped.Add(tssBlockLen - len(tssBlock))
+			rowsCountAfterRelabel := getRowsCount(tssBlock)
+			rowsDroppedByGlobalRelabel.Add(rowsCountBeforeRelabel - rowsCountAfterRelabel)
 		}
 		sortLabelsIfNeeded(tssBlock)
 		tssBlock = limitSeriesCardinality(tssBlock)
@@ -414,7 +417,10 @@ func labelsToString(labels []prompbmarshal.Label) string {
 	return string(b)
 }
 
-var globalRelabelMetricsDropped = metrics.NewCounter("vmagent_remotewrite_global_relabel_metrics_dropped_total")
+var (
+	globalRowsPushedBeforeRelabel = metrics.NewCounter("vmagent_remotewrite_global_rows_pushed_before_relabel")
+	rowsDroppedByGlobalRelabel    = metrics.NewCounter("vmagent_remotewrite_global_relabel_metrics_dropped_total")
+)
 
 type remoteWriteCtx struct {
 	idx        int
@@ -423,7 +429,8 @@ type remoteWriteCtx struct {
 	pss        []*pendingSeries
 	pssNextIdx uint64
 
-	relabelMetricsDropped *metrics.Counter
+	rowsPushedAfterRelabel *metrics.Counter
+	rowsDroppedByRelabel   *metrics.Counter
 }
 
 func newRemoteWriteCtx(argIdx int, at *auth.Token, remoteWriteURL *url.URL, maxInmemoryBlocks int, sanitizedURL string) *remoteWriteCtx {
@@ -467,7 +474,8 @@ func newRemoteWriteCtx(argIdx int, at *auth.Token, remoteWriteURL *url.URL, maxI
 		c:   c,
 		pss: pss,
 
-		relabelMetricsDropped: metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_relabel_metrics_dropped_total{path=%q, url=%q}`, queuePath, sanitizedURL)),
+		rowsPushedAfterRelabel: metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_rows_pushed_after_relabel_total{path=%q, url=%q}`, queuePath, sanitizedURL)),
+		rowsDroppedByRelabel:   metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_relabel_metrics_dropped_total{path=%q, url=%q}`, queuePath, sanitizedURL)),
 	}
 }
 
@@ -483,7 +491,8 @@ func (rwctx *remoteWriteCtx) MustStop() {
 	rwctx.fq.MustClose()
 	rwctx.fq = nil
 
-	rwctx.relabelMetricsDropped = nil
+	rwctx.rowsPushedAfterRelabel = nil
+	rwctx.rowsDroppedByRelabel = nil
 }
 
 func (rwctx *remoteWriteCtx) Push(tss []prompbmarshal.TimeSeries) {
@@ -499,12 +508,15 @@ func (rwctx *remoteWriteCtx) Push(tss []prompbmarshal.TimeSeries) {
 		// and https://github.com/VictoriaMetrics/VictoriaMetrics/issues/599
 		v = tssRelabelPool.Get().(*[]prompbmarshal.TimeSeries)
 		tss = append(*v, tss...)
-		tssLen := len(tss)
+		rowsCountBeforeRelabel := getRowsCount(tss)
 		tss = rctx.applyRelabeling(tss, nil, pcs)
-		rwctx.relabelMetricsDropped.Add(tssLen - len(tss))
+		rowsCountAfterRelabel := getRowsCount(tss)
+		rwctx.rowsDroppedByRelabel.Add(rowsCountBeforeRelabel - rowsCountAfterRelabel)
 	}
 	pss := rwctx.pss
 	idx := atomic.AddUint64(&rwctx.pssNextIdx, 1) % uint64(len(pss))
+	rowsCount := getRowsCount(tss)
+	rwctx.rowsPushedAfterRelabel.Add(rowsCount)
 	pss[idx].Push(tss)
 	if rctx != nil {
 		*v = prompbmarshal.ResetTimeSeries(tss)
@@ -519,3 +531,11 @@ var tssRelabelPool = &sync.Pool{
 		return &a
 	},
 }
+
+func getRowsCount(tss []prompbmarshal.TimeSeries) int {
+	rowsCount := 0
+	for _, ts := range tss {
+		rowsCount += len(ts.Samples)
+	}
+	return rowsCount
+}

docs/CHANGELOG.md

@@ -16,6 +16,7 @@ The following tip changes can be tested by building VictoriaMetrics components f
 ## tip
 
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add ability to specify filters for Availability Zones in [ec2_sd_config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#ec2_sd_config) via `az_filters` section. This section can contain AZ-specific set of filters in the same way as the existing `filters` section, which is used for filtering EC2 instances. The list of supported AZ-specific filters is available [here](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeAvailabilityZones.html).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): expose `vmagent_remotewrite_global_rows_pushed_before_relabel` and `vmagent_remotewrite_rows_pushed_after_relabel_total` metrics at `http://vmagent:8429/metrics`, which can be used for monitoring the rate of rows (aka samples) pushed to remote storage before and after the relabeling via `-remoteWrite.relabelConfig` and `-remoteWrite.urlRelabelConfig`. See [relabeling docs](https://docs.victoriametrics.com/vmagent.html#relabeling) for details.
 
 * BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl.html): properly import InfluxDB measurements if they contain `db` tag. Previously this could result in incomplete import of measurmenet tags. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/2536). Thanks to @mback2k for the bugfix.
 * BUGFIX: [vmui](https://docs.victoriametrics.com/#vmui): do not reset the selected relative time range when entering new query. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2402#issuecomment-1115817302).