push.go: push metrics to remote storage in gzip-compressed form

This should reduce the needed network bandwidth by ~10x for reading the metrics at remote storage side
This commit is contained in:
Aliaksandr Valialkin 2022-07-25 10:17:02 +03:00
parent f790ba580c
commit b305bc011e
No known key found for this signature in database
GPG Key ID: A72BEC6CD3D0DED1

24
push.go
View File

@ -9,6 +9,8 @@ import (
"net/http" "net/http"
"net/url" "net/url"
"time" "time"
"compress/gzip"
) )
// InitPushProcessMetrics sets up periodic push for 'process_*' metrics to the given pushURL with the given interval. // InitPushProcessMetrics sets up periodic push for 'process_*' metrics to the given pushURL with the given interval.
@ -86,6 +88,9 @@ func (s *Set) InitPush(pushURL string, interval time.Duration, extraLabels strin
// //
// It is OK calling InitPushExt multiple times with different pushURL - // It is OK calling InitPushExt multiple times with different pushURL -
// in this case metrics are pushed to all the provided pushURL urls. // in this case metrics are pushed to all the provided pushURL urls.
//
// It is OK calling InitPushExt multiple times with different writeMetrics -
// in this case all the metrics generated by writeMetrics callbacks are writte to pushURL.
func InitPushExt(pushURL string, interval time.Duration, extraLabels string, writeMetrics func(w io.Writer)) error { func InitPushExt(pushURL string, interval time.Duration, extraLabels string, writeMetrics func(w io.Writer)) error {
if interval <= 0 { if interval <= 0 {
return fmt.Errorf("interval must be positive; got %s", interval) return fmt.Errorf("interval must be positive; got %s", interval)
@ -111,15 +116,30 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
ticker := time.NewTicker(interval) ticker := time.NewTicker(interval)
var bb bytes.Buffer var bb bytes.Buffer
var tmpBuf []byte var tmpBuf []byte
zw := gzip.NewWriter(&bb)
for range ticker.C { for range ticker.C {
bb.Reset() bb.Reset()
writeMetrics(&bb) writeMetrics(&bb)
if len(extraLabels) > 0 { if len(extraLabels) > 0 {
tmpBuf = addExtraLabels(tmpBuf[:0], bb.Bytes(), extraLabels) tmpBuf = addExtraLabels(tmpBuf[:0], bb.Bytes(), extraLabels)
bb.Reset() bb.Reset()
bb.Write(tmpBuf) if _, err := bb.Write(tmpBuf); err != nil {
panic(fmt.Errorf("BUG: cannot write %d bytes to bytes.Buffer: %s", len(tmpBuf), err))
}
} }
resp, err := c.Post(pushURL, "text/plain", &bb) tmpBuf = append(tmpBuf[:0], bb.Bytes()...)
bb.Reset()
zw.Reset(&bb)
if _, err := zw.Write(tmpBuf); err != nil {
panic(fmt.Errorf("BUG: cannot write %d bytes to gzip writer: %s", len(tmpBuf), err))
}
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)
}
req.Header.Set("Content-Type", "text/plain")
req.Header.Set("Content-Encoding", "gzip")
resp, err := c.Do(req)
if err != nil { if err != nil {
log.Printf("ERROR: metrics.push: cannot push metrics to %q: %s", pushURLRedacted, err) log.Printf("ERROR: metrics.push: cannot push metrics to %q: %s", pushURLRedacted, err)
continue continue