push.go: add InitPushWithOptions() function, which allows extending push options without breaking backwards compatibility and without the need to introduce new functions
Overrides https://github.com/VictoriaMetrics/metrics/pull/53 Overrides https://github.com/VictoriaMetrics/metrics/pull/55 Allows specifying authorization headers via PushOptions.Headers for https://github.com/VictoriaMetrics/metrics/issues/36 Allows disabling request body compression via PushOption.DisableCompression for https://github.com/VictoriaMetrics/metrics/pull/41
This commit is contained in:
parent
447d235cbb
commit
a9e3faa53c
150
push.go
150
push.go
@ -8,11 +8,49 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"compress/gzip"
|
"compress/gzip"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// PushOptions is the list of options, which may be applied to InitPushWithOptions().
|
||||||
|
type PushOptions struct {
|
||||||
|
// ExtraLabels is an optional comma-separated list of `label="value"` labels, which must be added to all the metrics before pushing them to pushURL.
|
||||||
|
ExtraLabels string
|
||||||
|
|
||||||
|
// Headers is an optional list of HTTP headers to add to every push request to pushURL.
|
||||||
|
//
|
||||||
|
// Every item in the list must have the form `Header: value`. For example, `Authorization: Custom my-top-secret`.
|
||||||
|
Headers []string
|
||||||
|
|
||||||
|
// Whether to disable HTTP request body compression before sending the metrics to pushURL.
|
||||||
|
//
|
||||||
|
// By default the compression is enabled.
|
||||||
|
DisableCompression bool
|
||||||
|
}
|
||||||
|
|
||||||
|
// InitPushWithOptions sets up periodic push for globally registered metrics to the given pushURL with the given interval.
|
||||||
|
//
|
||||||
|
// If pushProcessMetrics is set to true, then 'process_*' and `go_*` metrics are also pushed to pushURL.
|
||||||
|
//
|
||||||
|
// opts may contain additional configuration options if non-nil.
|
||||||
|
//
|
||||||
|
// The metrics are pushed to pushURL in Prometheus text exposition format.
|
||||||
|
// See https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md#text-based-format
|
||||||
|
//
|
||||||
|
// It is recommended pushing metrics to /api/v1/import/prometheus endpoint according to
|
||||||
|
// https://docs.victoriametrics.com/#how-to-import-data-in-prometheus-exposition-format
|
||||||
|
//
|
||||||
|
// It is OK calling InitPushWithOptions multiple times with different pushURL -
|
||||||
|
// in this case metrics are pushed to all the provided pushURL urls.
|
||||||
|
func InitPushWithOptions(pushURL string, interval time.Duration, pushProcessMetrics bool, opts *PushOptions) error {
|
||||||
|
writeMetrics := func(w io.Writer) {
|
||||||
|
WritePrometheus(w, pushProcessMetrics)
|
||||||
|
}
|
||||||
|
return initPushWithOptions(pushURL, interval, writeMetrics, opts)
|
||||||
|
}
|
||||||
|
|
||||||
// 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.
|
||||||
//
|
//
|
||||||
// extraLabels may contain comma-separated list of `label="value"` labels, which will be added
|
// extraLabels may contain comma-separated list of `label="value"` labels, which will be added
|
||||||
@ -38,7 +76,7 @@ func InitPushProcessMetrics(pushURL string, interval time.Duration, extraLabels
|
|||||||
// extraLabels may contain comma-separated list of `label="value"` labels, which will be added
|
// extraLabels may contain comma-separated list of `label="value"` labels, which will be added
|
||||||
// to all the metrics before pushing them to pushURL.
|
// to all the metrics before pushing them to pushURL.
|
||||||
//
|
//
|
||||||
// If pushProcessMetrics is set to true, then 'process_*' metrics are also pushed to pushURL.
|
// If pushProcessMetrics is set to true, then 'process_*' and `go_*` metrics are also pushed to pushURL.
|
||||||
//
|
//
|
||||||
// The metrics are pushed to pushURL in Prometheus text exposition format.
|
// The metrics are pushed to pushURL in Prometheus text exposition format.
|
||||||
// See https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md#text-based-format
|
// See https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md#text-based-format
|
||||||
@ -55,12 +93,31 @@ func InitPush(pushURL string, interval time.Duration, extraLabels string, pushPr
|
|||||||
return InitPushExt(pushURL, interval, extraLabels, writeMetrics)
|
return InitPushExt(pushURL, interval, extraLabels, writeMetrics)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// InitPushWithOptions sets up periodic push for metrics from s to the given pushURL with the given interval.
|
||||||
|
//
|
||||||
|
// opts may contain additional configuration options if non-nil.
|
||||||
|
//
|
||||||
|
// The metrics are pushed to pushURL in Prometheus text exposition format.
|
||||||
|
// See https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md#text-based-format
|
||||||
|
//
|
||||||
|
// It is recommended pushing metrics to /api/v1/import/prometheus endpoint according to
|
||||||
|
// https://docs.victoriametrics.com/#how-to-import-data-in-prometheus-exposition-format
|
||||||
|
//
|
||||||
|
// It is OK calling InitPushWithOptions multiple times with different pushURL -
|
||||||
|
// in this case metrics are pushed to all the provided pushURL urls.
|
||||||
|
func (s *Set) InitPushWithOptions(pushURL string, interval time.Duration, opts *PushOptions) error {
|
||||||
|
writeMetrics := func(w io.Writer) {
|
||||||
|
s.WritePrometheus(w)
|
||||||
|
}
|
||||||
|
return initPushWithOptions(pushURL, interval, writeMetrics, opts)
|
||||||
|
}
|
||||||
|
|
||||||
// InitPush sets up periodic push for metrics from s to the given pushURL with the given interval.
|
// InitPush sets up periodic push for metrics from s to the given pushURL with the given interval.
|
||||||
//
|
//
|
||||||
// extraLabels may contain comma-separated list of `label="value"` labels, which will be added
|
// extraLabels may contain comma-separated list of `label="value"` labels, which will be added
|
||||||
// to all the metrics before pushing them to pushURL.
|
// to all the metrics before pushing them to pushURL.
|
||||||
//
|
//
|
||||||
// / The metrics are pushed to pushURL in Prometheus text exposition format.
|
// The metrics are pushed to pushURL in Prometheus text exposition format.
|
||||||
// See https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md#text-based-format
|
// See https://github.com/prometheus/docs/blob/main/content/docs/instrumenting/exposition_formats.md#text-based-format
|
||||||
//
|
//
|
||||||
// It is recommended pushing metrics to /api/v1/import/prometheus endpoint according to
|
// It is recommended pushing metrics to /api/v1/import/prometheus endpoint according to
|
||||||
@ -90,14 +147,16 @@ func (s *Set) InitPush(pushURL string, interval time.Duration, extraLabels strin
|
|||||||
// 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 -
|
// It is OK calling InitPushExt multiple times with different writeMetrics -
|
||||||
// in this case all the metrics generated by writeMetrics callbacks are writte to pushURL.
|
// in this case all the metrics generated by writeMetrics callbacks are written 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 {
|
opts := &PushOptions{
|
||||||
return fmt.Errorf("interval must be positive; got %s", interval)
|
ExtraLabels: extraLabels,
|
||||||
}
|
|
||||||
if err := validateTags(extraLabels); err != nil {
|
|
||||||
return fmt.Errorf("invalid extraLabels=%q: %w", extraLabels, err)
|
|
||||||
}
|
}
|
||||||
|
return initPushWithOptions(pushURL, interval, writeMetrics, opts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func initPushWithOptions(pushURL string, interval time.Duration, writeMetrics func(w io.Writer), opts *PushOptions) error {
|
||||||
|
// validate pushURL
|
||||||
pu, err := url.Parse(pushURL)
|
pu, err := url.Parse(pushURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("cannot parse pushURL=%q: %w", pushURL, err)
|
return fmt.Errorf("cannot parse pushURL=%q: %w", pushURL, err)
|
||||||
@ -108,16 +167,53 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
|
|||||||
if pu.Host == "" {
|
if pu.Host == "" {
|
||||||
return fmt.Errorf("missing host in pushURL=%q", pushURL)
|
return fmt.Errorf("missing host in pushURL=%q", pushURL)
|
||||||
}
|
}
|
||||||
pushURLRedacted := pu.Redacted()
|
|
||||||
c := &http.Client{
|
// validate interval
|
||||||
Timeout: interval,
|
if interval <= 0 {
|
||||||
|
return fmt.Errorf("interval must be positive; got %s", interval)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// validate ExtraLabels
|
||||||
|
var extraLabels string
|
||||||
|
if opts != nil {
|
||||||
|
extraLabels = opts.ExtraLabels
|
||||||
|
}
|
||||||
|
if err := validateTags(extraLabels); err != nil {
|
||||||
|
return fmt.Errorf("invalid extraLabels=%q: %w", extraLabels, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// validate Headers
|
||||||
|
var headers http.Header
|
||||||
|
if opts != nil {
|
||||||
|
for _, h := range opts.Headers {
|
||||||
|
n := strings.IndexByte(h, ':')
|
||||||
|
if n < 0 {
|
||||||
|
return fmt.Errorf("missing `:` delimiter in the header %q", h)
|
||||||
|
}
|
||||||
|
name := strings.TrimSpace(h[:n])
|
||||||
|
value := strings.TrimSpace(h[n+1:])
|
||||||
|
headers.Add(name, value)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// validate DisableCompression
|
||||||
|
disableCompression := false
|
||||||
|
if opts != nil {
|
||||||
|
disableCompression = opts.DisableCompression
|
||||||
|
}
|
||||||
|
|
||||||
|
// Initialize metrics for the given pushURL
|
||||||
|
pushURLRedacted := pu.Redacted()
|
||||||
pushesTotal := pushMetrics.GetOrCreateCounter(fmt.Sprintf(`metrics_push_total{url=%q}`, pushURLRedacted))
|
pushesTotal := pushMetrics.GetOrCreateCounter(fmt.Sprintf(`metrics_push_total{url=%q}`, pushURLRedacted))
|
||||||
pushErrorsTotal := pushMetrics.GetOrCreateCounter(fmt.Sprintf(`metrics_push_errors_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))
|
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))
|
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))
|
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())
|
pushMetrics.GetOrCreateFloatCounter(fmt.Sprintf(`metrics_push_interval_seconds{url=%q}`, pushURLRedacted)).Set(interval.Seconds())
|
||||||
|
|
||||||
|
c := &http.Client{
|
||||||
|
Timeout: interval,
|
||||||
|
}
|
||||||
go func() {
|
go func() {
|
||||||
ticker := time.NewTicker(interval)
|
ticker := time.NewTicker(interval)
|
||||||
var bb bytes.Buffer
|
var bb bytes.Buffer
|
||||||
@ -133,14 +229,16 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
|
|||||||
panic(fmt.Errorf("BUG: cannot write %d bytes to bytes.Buffer: %s", len(tmpBuf), err))
|
panic(fmt.Errorf("BUG: cannot write %d bytes to bytes.Buffer: %s", len(tmpBuf), err))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
tmpBuf = append(tmpBuf[:0], bb.Bytes()...)
|
if !disableCompression {
|
||||||
bb.Reset()
|
tmpBuf = append(tmpBuf[:0], bb.Bytes()...)
|
||||||
zw.Reset(&bb)
|
bb.Reset()
|
||||||
if _, err := zw.Write(tmpBuf); err != nil {
|
zw.Reset(&bb)
|
||||||
panic(fmt.Errorf("BUG: cannot write %d bytes to gzip writer: %s", len(tmpBuf), err))
|
if _, err := zw.Write(tmpBuf); err != nil {
|
||||||
}
|
panic(fmt.Errorf("BUG: cannot write %d bytes to gzip writer: %s", len(tmpBuf), err))
|
||||||
if err := zw.Close(); err != nil {
|
}
|
||||||
panic(fmt.Errorf("BUG: cannot flush metrics to gzip writer: %s", err))
|
if err := zw.Close(); err != nil {
|
||||||
|
panic(fmt.Errorf("BUG: cannot flush metrics to gzip writer: %s", err))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
pushesTotal.Inc()
|
pushesTotal.Inc()
|
||||||
blockLen := bb.Len()
|
blockLen := bb.Len()
|
||||||
@ -150,8 +248,20 @@ func InitPushExt(pushURL string, interval time.Duration, extraLabels string, wri
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
panic(fmt.Errorf("BUG: metrics.push: cannot initialize request for metrics push to %q: %w", pushURLRedacted, err))
|
panic(fmt.Errorf("BUG: metrics.push: cannot initialize request for metrics push to %q: %w", pushURLRedacted, err))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Set the needed headers
|
||||||
|
for name, values := range headers {
|
||||||
|
for _, value := range values {
|
||||||
|
req.Header.Add(name, value)
|
||||||
|
}
|
||||||
|
}
|
||||||
req.Header.Set("Content-Type", "text/plain")
|
req.Header.Set("Content-Type", "text/plain")
|
||||||
req.Header.Set("Content-Encoding", "gzip")
|
|
||||||
|
if !disableCompression {
|
||||||
|
req.Header.Set("Content-Encoding", "gzip")
|
||||||
|
}
|
||||||
|
|
||||||
|
// Perform the request
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
resp, err := c.Do(req)
|
resp, err := c.Do(req)
|
||||||
pushDuration.UpdateDuration(startTime)
|
pushDuration.UpdateDuration(startTime)
|
||||||
|
Loading…
Reference in New Issue
Block a user