Initial implementation of OpenMetrics text format

pkg/targets/constants/constants.go

@@ -11,6 +11,7 @@ const (
 	FormatAkumuli         = "akumuli"
 	FormatCrateDB         = "cratedb"
 	FormatPrometheus      = "prometheus"
+	FormatOpenMetrics     = "openmetrics"
 	FormatVictoriaMetrics = "victoriametrics"
 	FormatTimestream      = "timestream"
 	FormatQuestDB         = "questdb"
@@ -27,6 +28,7 @@ func SupportedFormats() []string {
 		FormatAkumuli,
 		FormatCrateDB,
 		FormatPrometheus,
+		FormatOpenMetrics,
 		FormatVictoriaMetrics,
 		FormatTimestream,
 		FormatQuestDB,

pkg/targets/initializers/target_initializers.go

@@ -2,6 +2,8 @@ package initializers
 
 import (
 	"fmt"
+	"strings"
+
 	"github.com/timescale/tsbs/pkg/targets"
 	"github.com/timescale/tsbs/pkg/targets/akumuli"
 	"github.com/timescale/tsbs/pkg/targets/cassandra"
@@ -10,13 +12,13 @@ import (
 	"github.com/timescale/tsbs/pkg/targets/crate"
 	"github.com/timescale/tsbs/pkg/targets/influx"
 	"github.com/timescale/tsbs/pkg/targets/mongo"
+	"github.com/timescale/tsbs/pkg/targets/openmetrics"
 	"github.com/timescale/tsbs/pkg/targets/prometheus"
 	"github.com/timescale/tsbs/pkg/targets/questdb"
 	"github.com/timescale/tsbs/pkg/targets/siridb"
 	"github.com/timescale/tsbs/pkg/targets/timescaledb"
 	"github.com/timescale/tsbs/pkg/targets/timestream"
 	"github.com/timescale/tsbs/pkg/targets/victoriametrics"
-	"strings"
 )
 
 func GetTarget(format string) targets.ImplementedTarget {
@@ -35,6 +37,8 @@ func GetTarget(format string) targets.ImplementedTarget {
 		return influx.NewTarget()
 	case constants.FormatMongo:
 		return mongo.NewTarget()
+	case constants.FormatOpenMetrics:
+		return openmetrics.NewTarget()
 	case constants.FormatPrometheus:
 		return prometheus.NewTarget()
 	case constants.FormatSiriDB:

pkg/targets/openmetrics/implemented_target.go

@@ -0,0 +1,33 @@
+package openmetrics
+
+import (
+	"github.com/blagojts/viper"
+	"github.com/spf13/pflag"
+	"github.com/timescale/tsbs/pkg/data/serialize"
+	"github.com/timescale/tsbs/pkg/data/source"
+	"github.com/timescale/tsbs/pkg/targets"
+	"github.com/timescale/tsbs/pkg/targets/constants"
+)
+
+func NewTarget() targets.ImplementedTarget {
+	return &openMetricsTarget{}
+}
+
+type openMetricsTarget struct {
+}
+
+func (t *openMetricsTarget) TargetSpecificFlags(flagPrefix string, flagSet *pflag.FlagSet) {
+	flagSet.String(flagPrefix+"url", "http://localhost:9091/", "Prometheus Pushgateway endpoint")
+}
+
+func (t *openMetricsTarget) TargetName() string {
+	return constants.FormatOpenMetrics
+}
+
+func (t *openMetricsTarget) Serializer() serialize.PointSerializer {
+	return &Serializer{}
+}
+
+func (t *openMetricsTarget) Benchmark(string, *source.DataSourceConfig, *viper.Viper) (targets.Benchmark, error) {
+	panic("not implemented")
+}

pkg/targets/openmetrics/serializer.go

@@ -0,0 +1,114 @@
+package openmetrics
+
+import (
+	"fmt"
+	"io"
+	"strconv"
+
+	"github.com/timescale/tsbs/pkg/data"
+	"github.com/timescale/tsbs/pkg/data/serialize"
+)
+
+// Serializer writes a Point in a text form for OpenMetrics
+// https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md#text-format
+type Serializer struct{}
+
+// This function writes output that looks like:
+// <measurement>_<field_name>{<tag key>="<tag value>",<tag key>="<tag value>"} <field value> <timestamp>\n
+//
+// For example:
+// foo_baz{tag0="bar1",tag1="bar2"} -1.0 100\n
+func (s *Serializer) Serialize(p *data.Point, w io.Writer) (err error) {
+	commonTags := make([]byte, 0, 1024)
+
+	tagKeys := p.TagKeys()
+	tagValues := p.TagValues()
+
+	if len(tagKeys) > 0 {
+		commonTags = append(commonTags, '{')
+	}
+
+	for i := 0; i < len(tagKeys); i++ {
+		if i > 0 {
+			commonTags = append(commonTags, ',')
+		}
+		commonTags = append(commonTags, tagKeys[i]...)
+		commonTags = append(commonTags, '=')
+		if tagValues[i] == nil {
+			commonTags = append(commonTags, '"')
+			commonTags = append(commonTags, '"')
+			continue
+		}
+		switch v := tagValues[i].(type) {
+		case string:
+			commonTags = append(commonTags, '"')
+			commonTags = append(commonTags, []byte(v)...)
+			commonTags = append(commonTags, '"')
+		case float32:
+			commonTags = append(commonTags, '"')
+			commonTags = append(commonTags, strconv.FormatFloat(float64(tagValues[i].(float32)), 'f', -1, 64)...)
+			commonTags = append(commonTags, '"')
+		case float64:
+			commonTags = append(commonTags, '"')
+			commonTags = append(commonTags, strconv.FormatFloat(float64(tagValues[i].(float64)), 'f', -1, 64)...)
+			commonTags = append(commonTags, '"')
+		default:
+			panic(fmt.Errorf("non-string tags not implemented for openmetrics %s", v))
+		}
+	}
+
+	if len(tagKeys) > 0 {
+		commonTags = append(commonTags, '}')
+	}
+
+	commonTags = append(commonTags, ' ')
+
+	buf := make([]byte, 0, 1024)
+
+	fieldKeys := p.FieldKeys()
+	fieldValues := p.FieldValues()
+	for i := 0; i < len(fieldKeys); i++ {
+		if fieldValues[i] == nil {
+			continue
+		}
+		buf = append(buf, p.MeasurementName()...)
+		buf = append(buf, '_')
+		buf = append(buf, fieldKeys[i]...)
+
+		buf = append(buf, commonTags...)
+
+		var value string
+		switch t := fieldValues[i].(type) {
+		case nil:
+			panic("logical error: nil field has to be skipped earlier")
+		case string:
+			value = fieldValues[i].(string)
+		case int:
+			value = strconv.FormatInt(int64(fieldValues[i].(int)), 10)
+		case int64:
+			value = strconv.FormatInt(fieldValues[i].(int64), 10)
+		case float32:
+			value = strconv.FormatFloat(float64(fieldValues[i].(float32)), 'f', -1, 64)
+		case float64:
+			value = strconv.FormatFloat(fieldValues[i].(float64), 'f', -1, 64)
+		default:
+			panic(fmt.Errorf("non-string tags not implemented for prometheus %s", t))
+		}
+		buf = append(buf, value...)
+
+		t := p.Timestamp()
+		if t != nil {
+			buf = append(buf, ' ')
+			if t.Nanosecond() == 0 {
+				buf = serialize.FastFormatAppend(p.Timestamp().UTC().Unix(), buf)
+			} else {
+				buf = serialize.FastFormatAppend(p.Timestamp().UTC().UnixNano(), buf)
+			}
+		}
+		buf = append(buf, '\n')
+	}
+
+	_, err = w.Write(buf)
+
+	return err
+}

pkg/targets/openmetrics/serializer_test.go

@@ -0,0 +1,45 @@
+package openmetrics
+
+import (
+	"testing"
+
+	"github.com/timescale/tsbs/pkg/data/serialize"
+)
+
+func TestInfluxSerializerSerialize(t *testing.T) {
+	cases := []serialize.SerializeCase{
+		{
+			Desc:       "a regular Point",
+			InputPoint: serialize.TestPointDefault(),
+			Output:     "cpu_usage_guest_nice{hostname=\"host_0\",region=\"eu-west-1\",datacenter=\"eu-west-1b\"} 38.24311829 1451606400\n",
+		},
+		{
+			Desc:       "a regular Point using int as value",
+			InputPoint: serialize.TestPointInt(),
+			Output:     "cpu_usage_guest{hostname=\"host_0\",region=\"eu-west-1\",datacenter=\"eu-west-1b\"} 38 1451606400\n",
+		},
+		{
+			Desc:       "a regular Point with multiple fields",
+			InputPoint: serialize.TestPointMultiField(),
+			Output: `cpu_big_usage_guest{hostname="host_0",region="eu-west-1",datacenter="eu-west-1b"} 5000000000 1451606400
+cpu_usage_guest{hostname="host_0",region="eu-west-1",datacenter="eu-west-1b"} 38 1451606400
+cpu_usage_guest_nice{hostname="host_0",region="eu-west-1",datacenter="eu-west-1b"} 38.24311829 1451606400
+`,
+		},
+		{
+			Desc:       "a Point with no tags",
+			InputPoint: serialize.TestPointNoTags(),
+			Output:     "cpu_usage_guest_nice 38.24311829 1451606400\n",
+		}, {
+			Desc:       "a Point with a nil tag",
+			InputPoint: serialize.TestPointWithNilTag(),
+			Output:     "cpu_usage_guest_nice{hostname=\"\"} 38.24311829 1451606400\n",
+		}, {
+			Desc:       "a Point with a nil field",
+			InputPoint: serialize.TestPointWithNilField(),
+			Output:     "cpu_usage_guest_nice 38.24311829 1451606400\n",
+		},
+	}
+
+	serialize.SerializerTest(t, cases, &Serializer{})
+}