From e9b4bb1534f39252df9b249495c3211ceba204db Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Thu, 4 Aug 2022 18:12:13 +0300
Subject: [PATCH] push.go: expose `metrics_push_*` metrics related to metrics
 push process

Updates https://github.com/VictoriaMetrics/metrics/issues/35
---
 metrics.go |  1 +
 push.go    | 22 +++++++++++++++++++++-
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/metrics.go b/metrics.go
index 57dcd3f..087026d 100644
--- a/metrics.go
+++ b/metrics.go
@@ -99,6 +99,7 @@ func WritePrometheus(w io.Writer, exposeProcessMetrics bool) {
 func WriteProcessMetrics(w io.Writer) {
 	writeGoMetrics(w)
 	writeProcessMetrics(w)
+	writePushMetrics(w)
 }
 
 // WriteFDMetrics writes `process_max_fds` and `process_open_fds` metrics to w.
diff --git a/push.go b/push.go
index c592ae6..5dd2033 100644
--- a/push.go
+++ b/push.go
@@ -112,6 +112,12 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
 	c := &http.Client{
 		Timeout: interval,
 	}
+	pushesTotal := pushMetrics.GetOrCreateCounter(fmt.Sprintf(`metrics_push_total{url=%q}`, pushURLRedacted))
+	pushErrorsTotal := pushMetrics.GetOrCreateCounter(fmt.Sprintf(`metrics_push_errors_total{url=%q}`, pushURLRedacted))
+	bytesPushedTotal := pushMetrics.GetOrCreateCounter(fmt.Sprintf(`metrics_push_bytes_pushed_total{url=%q}`, pushURLRedacted))
+	pushDuration := pushMetrics.GetOrCreateHistogram(fmt.Sprintf(`metrics_push_duration_seconds{url=%q}`, pushURLRedacted))
+	pushBlockSize := pushMetrics.GetOrCreateHistogram(fmt.Sprintf(`metrics_push_block_size_bytes{url=%q}`, pushURLRedacted))
+	pushMetrics.GetOrCreateFloatCounter(fmt.Sprintf(`metrics_push_interval_seconds{url=%q}`, pushURLRedacted)).Set(interval.Seconds())
 	go func() {
 		ticker := time.NewTicker(interval)
 		var bb bytes.Buffer
@@ -136,15 +142,22 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
 			if err := zw.Close(); err != nil {
 				panic(fmt.Errorf("BUG: cannot flush metrics to gzip writer: %s", err))
 			}
+			pushesTotal.Inc()
+			blockLen := bb.Len()
+			bytesPushedTotal.Add(blockLen)
+			pushBlockSize.Update(float64(blockLen))
 			req, err := http.NewRequest("GET", pushURL, &bb)
 			if err != nil {
-				log.Printf("ERROR: metrics.push: cannot initialize request for metrics push to %q: %s", pushURLRedacted, err)
+				panic(fmt.Errorf("BUG: metrics.push: cannot initialize request for metrics push to %q: %w", pushURLRedacted, err))
 			}
 			req.Header.Set("Content-Type", "text/plain")
 			req.Header.Set("Content-Encoding", "gzip")
+			startTime := time.Now()
 			resp, err := c.Do(req)
+			pushDuration.UpdateDuration(startTime)
 			if err != nil {
 				log.Printf("ERROR: metrics.push: cannot push metrics to %q: %s", pushURLRedacted, err)
+				pushErrorsTotal.Inc()
 				continue
 			}
 			if resp.StatusCode/100 != 2 {
@@ -152,6 +165,7 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
 				_ = resp.Body.Close()
 				log.Printf("ERROR: metrics.push: unexpected status code in response from %q: %d; expecting 2xx; response body: %q",
 					pushURLRedacted, resp.StatusCode, body)
+				pushErrorsTotal.Inc()
 				continue
 			}
 			_ = resp.Body.Close()
@@ -160,6 +174,12 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
 	return nil
 }
 
+var pushMetrics = NewSet()
+
+func writePushMetrics(w io.Writer) {
+	pushMetrics.WritePrometheus(w)
+}
+
 func addExtraLabels(dst, src []byte, extraLabels string) []byte {
 	for len(src) > 0 {
 		var line []byte