feat: add suggested changes, tests, and stdin support

Signed-off-by: François Gouteroux <francois.gouteroux@gmail.com>

cmd/promtool/main.go

@@ -81,6 +81,7 @@ func main() {
 	var (
 		httpRoundTripper   = api.DefaultRoundTripper
 		serverURL          *url.URL
+		remoteWriteURL     *url.URL
 		httpConfigFilePath string
 	)
 
@@ -180,12 +181,12 @@ func main() {
 
 	pushCmd := app.Command("push", "Push to a Prometheus server.")
 	pushCmd.Flag("http.config.file", "HTTP client configuration file for promtool to connect to Prometheus.").PlaceHolder("<filename>").ExistingFileVar(&httpConfigFilePath)
-	pushMetricsCmd := pushCmd.Command("metrics", "Push metrics to a prometheus remote write.")
+	pushMetricsCmd := pushCmd.Command("metrics", "Push metrics to a prometheus remote write (for testing purpose only).")
-	pushMetricsCmd.Arg("remote-write-url", "Prometheus remote write url to push metrics.").Required().URLVar(&serverURL)
+	pushMetricsCmd.Arg("remote-write-url", "Prometheus remote write url to push metrics.").Required().URLVar(&remoteWriteURL)
 	metricFiles := pushMetricsCmd.Arg(
 		"metric-files",
-		"The metric files to push.",
+		"The metric files to push, default is read from standard input (STDIN).",
-	).Required().ExistingFiles()
+	).ExistingFiles()
 	metricJobLabel := pushMetricsCmd.Flag("job-label", "Job label to attach to metrics.").Default("promtool").String()
 	pushMetricsTimeout := pushMetricsCmd.Flag("timeout", "The time to wait for pushing metrics.").Default("30s").Duration()
 	pushMetricsHeaders := pushMetricsCmd.Flag("header", "Prometheus remote write header.").StringMap()
@@ -314,7 +315,7 @@ func main() {
 		os.Exit(CheckMetrics(*checkMetricsExtended))
 
 	case pushMetricsCmd.FullCommand():
-		os.Exit(PushMetrics(serverURL, httpRoundTripper, *pushMetricsHeaders, *pushMetricsTimeout, *metricJobLabel, *metricFiles...))
+		os.Exit(PushMetrics(remoteWriteURL, httpRoundTripper, *pushMetricsHeaders, *pushMetricsTimeout, *metricJobLabel, *metricFiles...))
 
 	case queryInstantCmd.FullCommand():
 		os.Exit(QueryInstant(serverURL, httpRoundTripper, *queryInstantExpr, *queryInstantTime, p))

cmd/promtool/metrics.go

@@ -21,22 +21,18 @@ import (
 	"net/http"
 	"net/url"
 	"os"
-	"sort"
 	"time"
 
 	"github.com/golang/snappy"
-	dto "github.com/prometheus/client_model/go"
 	config_util "github.com/prometheus/common/config"
-	"github.com/prometheus/common/expfmt"
 	"github.com/prometheus/common/model"
 
-	"github.com/prometheus/prometheus/prompb"
 	"github.com/prometheus/prometheus/storage/remote"
+	"github.com/prometheus/prometheus/util/fmtutil"
 )
 
-// Push metrics to a prometheus remote write.
+// Push metrics to a prometheus remote write (for testing purpose only).
 func PushMetrics(url *url.URL, roundTripper http.RoundTripper, headers map[string]string, timeout time.Duration, jobLabel string, files ...string) int {
-	// remote write should respect specification: https://prometheus.io/docs/concepts/remote_write_spec/
 	failed := false
 
 	addressURL, err := url.Parse(url.String())
@@ -67,18 +63,36 @@ func PushMetrics(url *url.URL, roundTripper http.RoundTripper, headers map[strin
 		headers:      headers,
 	}
 
-	for _, f := range files {
+	// add empty string to avoid matching filename
+	if len(files) == 0 {
+		files = append(files, "")
+	}
+
+	for _, file := range files {
 		var data []byte
 		var err error
-		data, err = os.ReadFile(f)
+
+		// if file is an empty string it is a stdin
+		if file == "" {
+			data, err = io.ReadAll(os.Stdin)
+			if err != nil {
+				fmt.Fprintln(os.Stderr, err)
+				failed = true
+				break
+			}
+
+			fmt.Printf("Parsing stdin\n")
+		} else {
+			data, err = os.ReadFile(file)
 			if err != nil {
 				fmt.Fprintln(os.Stderr, err)
 				failed = true
 				continue
 			}
 
-		fmt.Printf("Parsing metric file %s\n", f)
+			fmt.Printf("Parsing metric file %s\n", file)
-		metricsData, err := parseMetricsTextAndFormat(bytes.NewReader(data), jobLabel)
+		}
+		metricsData, err := fmtutil.ParseMetricsTextAndFormat(bytes.NewReader(data), jobLabel)
 		if err != nil {
 			fmt.Fprintln(os.Stderr, err)
 			failed = true
@@ -100,7 +114,7 @@ func PushMetrics(url *url.URL, roundTripper http.RoundTripper, headers map[strin
 			failed = true
 			continue
 		}
-		fmt.Printf("Successfully pushed metric file %s\n", f)
+		fmt.Printf("Successfully pushed metric file %s\n", file)
 	}
 
 	if failed {
@@ -121,114 +135,3 @@ func (s *setHeadersTransport) RoundTrip(req *http.Request) (*http.Response, erro
 	}
 	return s.RoundTripper.RoundTrip(req)
 }
-
-var MetricMetadataTypeValue = map[string]int32{
-	"UNKNOWN":        0,
-	"COUNTER":        1,
-	"GAUGE":          2,
-	"HISTOGRAM":      3,
-	"GAUGEHISTOGRAM": 4,
-	"SUMMARY":        5,
-	"INFO":           6,
-	"STATESET":       7,
-}
-
-// formatMetrics convert metric family to a writerequest
-func formatMetrics(mf map[string]*dto.MetricFamily, jobLabel string) (*prompb.WriteRequest, error) {
-	wr := &prompb.WriteRequest{}
-
-	// build metric list
-	sortedMetricNames := make([]string, 0, len(mf))
-	for metric := range mf {
-		sortedMetricNames = append(sortedMetricNames, metric)
-	}
-	// sort metrics name in lexicographical order
-	sort.Strings(sortedMetricNames)
-
-	for _, metricName := range sortedMetricNames {
-		// Set metadata writerequest
-		mtype := MetricMetadataTypeValue[mf[metricName].Type.String()]
-		metadata := prompb.MetricMetadata{
-			MetricFamilyName: mf[metricName].GetName(),
-			Type:             prompb.MetricMetadata_MetricType(mtype),
-			Help:             mf[metricName].GetHelp(),
-		}
-		wr.Metadata = append(wr.Metadata, metadata)
-
-		for _, metric := range mf[metricName].Metric {
-			var timeserie prompb.TimeSeries
-
-			// build labels map
-			labels := make(map[string]string, len(metric.Label)+2)
-			labels[model.MetricNameLabel] = metricName
-			labels[model.JobLabel] = jobLabel
-
-			for _, label := range metric.Label {
-				labelname := label.GetName()
-				if labelname == model.JobLabel {
-					labelname = fmt.Sprintf("%s%s", model.ExportedLabelPrefix, labelname)
-				}
-				labels[labelname] = label.GetValue()
-			}
-
-			// build labels name list
-			sortedLabelNames := make([]string, 0, len(labels))
-			for label := range labels {
-				sortedLabelNames = append(sortedLabelNames, label)
-			}
-			// sort labels name in lexicographical order
-			sort.Strings(sortedLabelNames)
-
-			for _, label := range sortedLabelNames {
-				timeserie.Labels = append(timeserie.Labels, prompb.Label{
-					Name:  label,
-					Value: labels[label],
-				})
-			}
-
-			timeserie.Samples = []prompb.Sample{
-				{
-					Timestamp: time.Now().UnixNano() / int64(time.Millisecond),
-					Value:     getMetricsValue(metric),
-				},
-			}
-
-			wr.Timeseries = append(wr.Timeseries, timeserie)
-		}
-	}
-	return wr, nil
-}
-
-// parseMetricsTextReader consumes an io.Reader and returns the MetricFamily
-func parseMetricsTextReader(input io.Reader) (map[string]*dto.MetricFamily, error) {
-	var parser expfmt.TextParser
-	mf, err := parser.TextToMetricFamilies(input)
-	if err != nil {
-		return nil, err
-	}
-	return mf, nil
-}
-
-// getMetricsValue return the value of a timeserie without the need to give value type
-func getMetricsValue(m *dto.Metric) float64 {
-	switch {
-	case m.Gauge != nil:
-		return m.GetGauge().GetValue()
-	case m.Counter != nil:
-		return m.GetCounter().GetValue()
-	case m.Untyped != nil:
-		return m.GetUntyped().GetValue()
-	default:
-		return 0.
-	}
-}
-
-// parseMetricsTextAndFormat return the data in the expected prometheus metrics write request format
-func parseMetricsTextAndFormat(input io.Reader, jobLabel string) (*prompb.WriteRequest, error) {
-	mf, err := parseMetricsTextReader(input)
-	if err != nil {
-		return nil, err
-	}
-
-	return formatMetrics(mf, jobLabel)
-}

docs/command-line/promtool.md

@@ -390,7 +390,7 @@ Push to a Prometheus server.
 
 ##### `promtool push metrics`
 
-Push metrics to a prometheus remote write.
+Push metrics to a prometheus remote write (for testing purpose only).
 
 
 
@@ -410,7 +410,7 @@ Push metrics to a prometheus remote write.
 | Argument | Description | Required |
 | --- | --- | --- |
 | remote-write-url | Prometheus remote write url to push metrics. | Yes |
-| metric-files | The metric files to push. | Yes |
+| metric-files | The metric files to push, default is read from standard input (STDIN). |  |
 
 
 

util/fmtutil/format.go

@@ -0,0 +1,142 @@
+// Copyright 2023 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fmtutil
+
+import (
+	"fmt"
+	"io"
+	"sort"
+	"time"
+
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
+	"github.com/prometheus/common/model"
+
+	"github.com/prometheus/prometheus/prompb"
+)
+
+var MetricMetadataTypeValue = map[string]int32{
+	"UNKNOWN":        0,
+	"COUNTER":        1,
+	"GAUGE":          2,
+	"HISTOGRAM":      3,
+	"GAUGEHISTOGRAM": 4,
+	"SUMMARY":        5,
+	"INFO":           6,
+	"STATESET":       7,
+}
+
+// FormatMetrics convert metric family to a writerequest.
+func FormatMetrics(mf map[string]*dto.MetricFamily, jobLabel string) (*prompb.WriteRequest, error) {
+	wr := &prompb.WriteRequest{}
+
+	// build metric list
+	sortedMetricNames := make([]string, 0, len(mf))
+	for metric := range mf {
+		sortedMetricNames = append(sortedMetricNames, metric)
+	}
+	// sort metrics name in lexicographical order
+	sort.Strings(sortedMetricNames)
+
+	for _, metricName := range sortedMetricNames {
+		// Set metadata writerequest
+		mtype := MetricMetadataTypeValue[mf[metricName].Type.String()]
+		metadata := prompb.MetricMetadata{
+			MetricFamilyName: mf[metricName].GetName(),
+			Type:             prompb.MetricMetadata_MetricType(mtype),
+			Help:             mf[metricName].GetHelp(),
+		}
+		wr.Metadata = append(wr.Metadata, metadata)
+
+		for _, metric := range mf[metricName].Metric {
+			var timeserie prompb.TimeSeries
+
+			// build labels map
+			labels := make(map[string]string, len(metric.Label)+2)
+			labels[model.MetricNameLabel] = metricName
+			labels[model.JobLabel] = jobLabel
+
+			for _, label := range metric.Label {
+				labelname := label.GetName()
+				if labelname == model.JobLabel {
+					labelname = fmt.Sprintf("%s%s", model.ExportedLabelPrefix, labelname)
+				}
+				labels[labelname] = label.GetValue()
+			}
+
+			// build labels name list
+			sortedLabelNames := make([]string, 0, len(labels))
+			for label := range labels {
+				sortedLabelNames = append(sortedLabelNames, label)
+			}
+			// sort labels name in lexicographical order
+			sort.Strings(sortedLabelNames)
+
+			for _, label := range sortedLabelNames {
+				timeserie.Labels = append(timeserie.Labels, prompb.Label{
+					Name:  label,
+					Value: labels[label],
+				})
+			}
+
+			timestamp := metric.GetTimestampMs()
+			if timestamp == 0 {
+				timestamp = time.Now().UnixNano() / int64(time.Millisecond)
+			}
+
+			timeserie.Samples = []prompb.Sample{
+				{
+					Timestamp: timestamp,
+					Value:     getMetricsValue(metric),
+				},
+			}
+
+			wr.Timeseries = append(wr.Timeseries, timeserie)
+		}
+	}
+	return wr, nil
+}
+
+// getMetricsValue return the value of a timeserie without the need to give value type
+func getMetricsValue(m *dto.Metric) float64 {
+	switch {
+	case m.Gauge != nil:
+		return m.GetGauge().GetValue()
+	case m.Counter != nil:
+		return m.GetCounter().GetValue()
+	case m.Untyped != nil:
+		return m.GetUntyped().GetValue()
+	default:
+		return 0.
+	}
+}
+
+// ParseMetricsTextReader consumes an io.Reader and returns the MetricFamily.
+func ParseMetricsTextReader(input io.Reader) (map[string]*dto.MetricFamily, error) {
+	var parser expfmt.TextParser
+	mf, err := parser.TextToMetricFamilies(input)
+	if err != nil {
+		return nil, err
+	}
+	return mf, nil
+}
+
+// ParseMetricsTextAndFormat return the data in the expected prometheus metrics write request format.
+func ParseMetricsTextAndFormat(input io.Reader, jobLabel string) (*prompb.WriteRequest, error) {
+	mf, err := ParseMetricsTextReader(input)
+	if err != nil {
+		return nil, err
+	}
+	return FormatMetrics(mf, jobLabel)
+}

util/fmtutil/format_test.go

@@ -0,0 +1,71 @@
+// Copyright 2023 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fmtutil
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/prometheus/prometheus/prompb"
+)
+
+var writeRequestFixture = &prompb.WriteRequest{
+	Metadata: []prompb.MetricMetadata{
+		{
+			MetricFamilyName: "test_metric1",
+			Type:             2,
+			Help:             "this is a test metric",
+		},
+	},
+	Timeseries: []prompb.TimeSeries{
+		{
+			Labels: []prompb.Label{
+				{Name: "__name__", Value: "test_metric1"},
+				{Name: "b", Value: "c"},
+				{Name: "baz", Value: "qux"},
+				{Name: "d", Value: "e"},
+				{Name: "foo", Value: "bar"},
+				{Name: "job", Value: "promtool"},
+			},
+			Samples: []prompb.Sample{{Value: 1, Timestamp: 1}},
+		},
+		{
+			Labels: []prompb.Label{
+				{Name: "__name__", Value: "test_metric1"},
+				{Name: "b", Value: "c"},
+				{Name: "baz", Value: "qux"},
+				{Name: "d", Value: "e"},
+				{Name: "foo", Value: "bar"},
+				{Name: "job", Value: "promtool"},
+			},
+			Samples: []prompb.Sample{{Value: 2, Timestamp: 1}},
+		},
+	},
+}
+
+func TestParseMetricsTextAndFormat(t *testing.T) {
+	input := bytes.NewReader([]byte(`
+	# HELP test_metric1 this is a test metric
+	# TYPE test_metric1 gauge
+	test_metric1{b="c",baz="qux",d="e",foo="bar"} 1 1
+	test_metric1{b="c",baz="qux",d="e",foo="bar"} 2 1
+	`))
+
+	expected, err := ParseMetricsTextAndFormat(input, "promtool")
+	require.NoError(t, err)
+
+	require.Equal(t, writeRequestFixture, expected)
+}