Fixes for metrics encode/decode

Fixes a number of issues:
* delimited protos now follow the escaping scheme
* NewDecoder now supports decoding text protos
* Text decoder now refuses to decode openmetrics, which is unsupported

Fixes open-telemetry/opentelemetry-collector-contrib#39700

Signed-off-by: Owen Williams <owen.williams@grafana.com>

Author: ywwg

expfmt/decode.go

@@ -15,6 +15,7 @@ package expfmt
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -23,6 +24,7 @@ import (
 
 	dto "github.com/prometheus/client_model/go"
 	"google.golang.org/protobuf/encoding/protodelim"
+	"google.golang.org/protobuf/encoding/prototext"
 
 	"github.com/prometheus/common/model"
 )
@@ -72,12 +74,17 @@ func ResponseFormat(h http.Header) Format {
 
 // NewDecoder returns a new decoder based on the given input format.
 // If the input format does not imply otherwise, a text format decoder is returned.
-func NewDecoder(r io.Reader, format Format) Decoder {
+func NewDecoder(r io.Reader, format Format) (Decoder, error) {
 	switch format.FormatType() {
 	case TypeProtoDelim:
-		return &protoDecoder{r: bufio.NewReader(r)}
+		return &protoDecoder{r: bufio.NewReader(r)}, nil
+	case TypeProtoText, TypeProtoCompact:
+		return &prototextDecoder{r: r}, nil
 	}
-	return &textDecoder{r: r}
+	if format.FormatType() == TypeOpenMetrics {
+		return nil, errors.New("cannot decode openmetrics")
+	}
+	return &textDecoder{r: r}, nil
 }
 
 // protoDecoder implements the Decoder interface for protocol buffers.
@@ -115,6 +122,44 @@ func (d *protoDecoder) Decode(v *dto.MetricFamily) error {
 	return nil
 }
 
+// prototextDecoder implements the Decoder interface for protocol buffers that
+// are encoded in text format.
+type prototextDecoder struct {
+	r io.Reader
+}
+
+// Decode implements the Decoder interface.
+func (d *prototextDecoder) Decode(v *dto.MetricFamily) error {
+	opts := prototext.UnmarshalOptions{}
+	b, err := io.ReadAll(d.r)
+	if err != nil {
+		return err
+	}
+	if err := opts.Unmarshal(b, v); err != nil {
+		return err
+	}
+	if !model.IsValidMetricName(model.LabelValue(v.GetName())) {
+		return fmt.Errorf("invalid metric name %q", v.GetName())
+	}
+	for _, m := range v.GetMetric() {
+		if m == nil {
+			continue
+		}
+		for _, l := range m.GetLabel() {
+			if l == nil {
+				continue
+			}
+			if !model.LabelValue(l.GetValue()).IsValid() {
+				return fmt.Errorf("invalid label value %q", l.GetValue())
+			}
+			if !model.LabelName(l.GetName()).IsValid() {
+				return fmt.Errorf("invalid label name %q", l.GetName())
+			}
+		}
+	}
+	return nil
+}
+
 // textDecoder implements the Decoder interface for the text protocol.
 type textDecoder struct {
 	r    io.Reader

expfmt/decode_test.go

@@ -408,7 +408,8 @@ func TestProtoMultiMessageDecoder(t *testing.T) {
 	require.NoErrorf(t, err, "Reading file failed: %v", err)
 
 	buf := bytes.NewReader(data)
-	decoder := NewDecoder(buf, FmtProtoDelim)
+	decoder, err := NewDecoder(buf, FmtProtoDelim)
+	require.NoError(t, err)
 	var metrics []*dto.MetricFamily
 	for {
 		var mf dto.MetricFamily
@@ -557,7 +558,8 @@ func TestTextDecoderWithBufioReader(t *testing.T) {
 
 	var decoded bool
 	r := bufio.NewReader(strings.NewReader(example))
-	dec := NewDecoder(r, FmtText)
+	dec, err := NewDecoder(r, FmtText)
+	require.NoError(t, err)
 	for {
 		var mf dto.MetricFamily
 		if err := dec.Decode(&mf); err != nil {

expfmt/encode.go

@@ -153,7 +153,7 @@ func NewEncoder(w io.Writer, format Format, options ...EncoderOption) Encoder {
 	case TypeProtoDelim:
 		return encoderCloser{
 			encode: func(v *dto.MetricFamily) error {
-				_, err := protodelim.MarshalTo(w, v)
+				_, err := protodelim.MarshalTo(w, model.EscapeMetricFamily(v, escapingScheme))
 				return err
 			},
 			close: func() error { return nil },

expfmt/encode_test.go

@@ -344,8 +344,21 @@ func TestEscapedEncode(t *testing.T) {
 		t.Errorf("expected the output bytes buffer to be non-empty")
 	}
 
-	buff.Reset()
+	dec, err := NewDecoder(bytes.NewReader(out), FmtProtoDelim)
+	if err != nil {
+		t.Errorf("unexpected error calling NewDecoder: %s", err.Error())
+	}
+	var gotFamily dto.MetricFamily
+	err = dec.Decode(&gotFamily)
+	if err != nil {
+		t.Errorf("unexpected error during proto decode: %s", err.Error())
+	}
+	expectEscapedName := "foo_metric"
+	if gotFamily.GetName() != expectEscapedName {
+		t.Errorf("incorrect encoded metric name, want %v, got %v", expectEscapedName, gotFamily.GetName())
+	}
 
+	buff.Reset()
 	compactEncoder := NewEncoder(&buff, FmtProtoCompact)
 	err = compactEncoder.Encode(metric)
 	if err != nil {
@@ -356,9 +369,19 @@ func TestEscapedEncode(t *testing.T) {
 	if len(out) == 0 {
 		t.Errorf("expected the output bytes buffer to be non-empty")
 	}
+	dec, err = NewDecoder(bytes.NewReader(out), FmtProtoCompact)
+	if err != nil {
+		t.Errorf("unexpected error calling NewDecoder: %s", err.Error())
+	}
+	err = dec.Decode(&gotFamily)
+	if err != nil {
+		t.Errorf("unexpected error during proto decode: %s", err.Error())
+	}
+	if gotFamily.GetName() != "foo_metric" {
+		t.Errorf("incorrect encoded metricname , want foo_metric, got %v", gotFamily.GetName())
+	}
 
 	buff.Reset()
-
 	protoTextEncoder := NewEncoder(&buff, FmtProtoText)
 	err = protoTextEncoder.Encode(metric)
 	if err != nil {
@@ -369,9 +392,19 @@ func TestEscapedEncode(t *testing.T) {
 	if len(out) == 0 {
 		t.Errorf("expected the output bytes buffer to be non-empty")
 	}
+	dec, err = NewDecoder(bytes.NewReader(out), FmtProtoText)
+	if err != nil {
+		t.Errorf("unexpected error calling NewDecoder: %s", err.Error())
+	}
+	err = dec.Decode(&gotFamily)
+	if err != nil {
+		t.Errorf("unexpected error during proto decode: %s", err.Error())
+	}
+	if gotFamily.GetName() != "foo_metric" {
+		t.Errorf("incorrect encoded metric name, want foo_metic, got %v", gotFamily.GetName())
+	}
 
 	buff.Reset()
-
 	textEncoder := NewEncoder(&buff, FmtText)
 	err = textEncoder.Encode(metric)
 	if err != nil {
@@ -392,3 +425,116 @@ foo_metric{dotted_label_name="my.label.value"} 8
 		t.Errorf("expected TextEncoder to return %s, but got %s instead", expected, string(out))
 	}
 }
+
+func TestDottedEncode(t *testing.T) {
+	//nolint:staticcheck
+	model.NameValidationScheme = model.UTF8Validation
+	metric := &dto.MetricFamily{
+		Name: proto.String("foo.metric"),
+		Type: dto.MetricType_COUNTER.Enum(),
+		Metric: []*dto.Metric{
+			{
+				Counter: &dto.Counter{
+					Value: proto.Float64(1.234),
+				},
+			},
+			{
+				Label: []*dto.LabelPair{
+					{
+						Name:  proto.String("dotted.label.name"),
+						Value: proto.String("my.label.value"),
+					},
+				},
+				Counter: &dto.Counter{
+					Value: proto.Float64(8),
+				},
+			},
+		},
+	}
+
+	scenarios := []struct {
+		format           Format
+		expectMetricName string
+		expectLabelName  string
+	}{
+		{
+			format:           FmtProtoDelim,
+			expectMetricName: "foo_metric",
+			expectLabelName:  "dotted_label_name",
+		},
+		{
+			format:           FmtProtoDelim.WithEscapingScheme(model.NoEscaping),
+			expectMetricName: "foo.metric",
+			expectLabelName:  "dotted.label.name",
+		},
+		{
+			format:           FmtProtoDelim.WithEscapingScheme(model.DotsEscaping),
+			expectMetricName: "foo_dot_metric",
+			expectLabelName:  "dotted_dot_label_dot_name",
+		},
+		{
+			format:           FmtProtoCompact,
+			expectMetricName: "foo_metric",
+			expectLabelName:  "dotted_label_name",
+		},
+		{
+			format:           FmtProtoCompact.WithEscapingScheme(model.NoEscaping),
+			expectMetricName: "foo.metric",
+			expectLabelName:  "dotted.label.name",
+		},
+		{
+			format:           FmtProtoCompact.WithEscapingScheme(model.ValueEncodingEscaping),
+			expectMetricName: "U__foo_2e_metric",
+			expectLabelName:  "U__dotted_2e_label_2e_name",
+		},
+		{
+			format:           FmtProtoText,
+			expectMetricName: "foo_metric",
+			expectLabelName:  "dotted_label_name",
+		},
+		{
+			format:           FmtProtoText.WithEscapingScheme(model.NoEscaping),
+			expectMetricName: "foo.metric",
+			expectLabelName:  "dotted.label.name",
+		},
+		{
+			format:           FmtText,
+			expectMetricName: "foo_metric",
+			expectLabelName:  "dotted_label_name",
+		},
+		{
+			format:           FmtText.WithEscapingScheme(model.NoEscaping),
+			expectMetricName: "foo.metric",
+			expectLabelName:  "dotted.label.name",
+		},
+		// common library does not support open metrics parsing so we do not test
+		// that here.
+	}
+
+	for i, scenario := range scenarios {
+		out := bytes.NewBuffer(make([]byte, 0))
+		enc := NewEncoder(out, scenario.format)
+		err := enc.Encode(metric)
+		if err != nil {
+			t.Errorf("%d. error: %s", i, err)
+			continue
+		}
+
+		dec, err := NewDecoder(bytes.NewReader(out.Bytes()), scenario.format)
+		if err != nil {
+			t.Errorf("unexpected error calling NewDecoder: %s", err.Error())
+		}
+		var gotFamily dto.MetricFamily
+		err = dec.Decode(&gotFamily)
+		if err != nil {
+			t.Errorf("%v: unexpected error during decode: %s", scenario.format, err.Error())
+		}
+		if gotFamily.GetName() != scenario.expectMetricName {
+			t.Errorf("%v: incorrect encoded metric name, want %v, got %v", scenario.format, scenario.expectMetricName, gotFamily.GetName())
+		}
+		lName := gotFamily.GetMetric()[1].Label[0].GetName()
+		if lName != scenario.expectLabelName {
+			t.Errorf("%v: incorrect encoded label name, want %v, got %v", scenario.format, scenario.expectLabelName, lName)
+		}
+	}
+}

expfmt/text_parse.go

@@ -574,7 +574,14 @@ func (p *TextParser) readingType() stateFn {
 	if p.readTokenUntilNewline(false); p.err != nil {
 		return nil // Unexpected end of input.
 	}
-	metricType, ok := dto.MetricType_value[strings.ToUpper(p.currentToken.String())]
+	typString := strings.ToUpper(p.currentToken.String())
+	// OpenMetrics uses UNKNOWN for untyped metrics instead of UNTYPED, so we add
+	// a special case for that here.
+	if typString == "UNKNOWN" {
+		p.parseError("found metric type UNKNOWN, is the input OpenMetrics instead of Prometheus format?")
+		return nil
+	}
+	metricType, ok := dto.MetricType_value[typString]
 	if !ok {
 		p.parseError(fmt.Sprintf("unknown metric type %q", p.currentToken.String()))
 		return nil