sumologicexporter: add graphite format (#2695)

 - Add support for graphite format to sumo logic exporter
 - Add sanitization for carbon2 format
 - Replace `=` with `:` for sanitization

**Link to tracking Issue:**
#1498

Author: sumo-drosiek

exporter/sumologicexporter/README.md

@@ -10,11 +10,16 @@ Empty string means no compression
 - `max_request_body_size` (optional): Max HTTP request body size in bytes before compression (if applied). By default `1_048_576` (1MB) is used.
 - `metadata_attributes` (optional): List of regexes for attributes which should be send as metadata
 - `log_format` (optional) (logs only): Format to use when sending logs to Sumo. (default `json`) (possible values: `json`, `text`)
-- `metric_format` (optional) (metrics only): Format of the metrics to be sent (default is `prometheus`) (possible values: `carbon2`, `prometheus`)
-  `carbon2` and `graphite` are going to be supported soon.
+- `metric_format` (optional) (metrics only): Format of the metrics to be sent (default is `prometheus`) (possible values: `carbon2`, `graphite`, `prometheus`).
+- `graphite_template` (default=`%{_metric_}`) (optional) (metrics only): Template for Graphite format.
+[Source templates](#source-templates) are going to be applied.
+Applied only if `metric_format` is set to `graphite`.
 - `source_category` (optional): Desired source category. Useful if you want to override the source category configured for the source.
+[Source templates](#source-templates) are going to be applied.
 - `source_name` (optional): Desired source name. Useful if you want to override the source name configured for the source.
+[Source templates](#source-templates) are going to be applied.
 - `source_host` (optional): Desired host name. Useful if you want to override the source host configured for the source.
+[Source templates](#source-templates) are going to be applied.
 - `timeout` (default = 5s): Is the timeout for every attempt to send data to the backend.
 Maximum connection timeout is 55s.
 - `retry_on_failure`
@@ -30,7 +35,15 @@ Maximum connection timeout is 55s.
     - `num_seconds` is the number of seconds to buffer in case of a backend outage
     - `requests_per_second` is the average number of requests per seconds.
 
-Example:
+## Source Templates
+
+You can specify a template with an attribute for `source_category`, `source_name`, `source_host` or `graphite_template` using `%{attr_name}`.
+
+For example, when there is an attribute `my_attr`: `my_value`, `metrics/%{my_attr}` would be expanded to `metrics/my_value`.
+
+For `graphite_template`, in addition to above, `%{_metric_}` is going to be replaced with metric name.
+
+## Example Configuration
 
 ```yaml
 exporters:

exporter/sumologicexporter/carbon_formatter.go

@@ -42,18 +42,27 @@ func carbon2TagString(record metricPair) string {
 		if k == "name" || k == "unit" {
 			k = fmt.Sprintf("_%s", k)
 		}
-		returnValue = append(returnValue, fmt.Sprintf("%s=%s", k, tracetranslator.AttributeValueToString(v, false)))
+		returnValue = append(returnValue, fmt.Sprintf(
+			"%s=%s",
+			sanitizeCarbonString(k),
+			sanitizeCarbonString(tracetranslator.AttributeValueToString(v, false)),
+		))
 	})
 
-	returnValue = append(returnValue, fmt.Sprintf("metric=%s", record.metric.Name()))
+	returnValue = append(returnValue, fmt.Sprintf("metric=%s", sanitizeCarbonString(record.metric.Name())))
 
 	if len(record.metric.Unit()) > 0 {
-		returnValue = append(returnValue, fmt.Sprintf("unit=%s", record.metric.Unit()))
+		returnValue = append(returnValue, fmt.Sprintf("unit=%s", sanitizeCarbonString(record.metric.Unit())))
 	}
 
 	return strings.Join(returnValue, " ")
 }
 
+// sanitizeCarbonString replaces problematic characters with underscore
+func sanitizeCarbonString(text string) string {
+	return strings.NewReplacer(" ", "_", "=", ":", "\n", "_").Replace(text)
+}
+
 // carbon2IntRecord converts IntDataPoint to carbon2 metric string
 // with additional information from metricPair.
 func carbon2IntRecord(record metricPair, dataPoint pdata.IntDataPoint) string {

exporter/sumologicexporter/config.go

@@ -46,19 +46,25 @@ type Config struct {
 	// The format of metrics you will be sending, either graphite or carbon2 or prometheus (Default is prometheus)
 	// Possible values are `carbon2` and `prometheus`
 	MetricFormat MetricFormatType `mapstructure:"metric_format"`
+	// Graphite template.
+	// Placeholders `%{attr_name}` will be replaced with attribute value for attr_name.
+	GraphiteTemplate string `mapstructure:"graphite_template"`
 
 	// List of regexes for attributes which should be send as metadata
 	MetadataAttributes []string `mapstructure:"metadata_attributes"`
 
 	// Sumo specific options
 	// Desired source category.
 	// Useful if you want to override the source category configured for the source.
+	// Placeholders `%{attr_name}` will be replaced with attribute value for attr_name.
 	SourceCategory string `mapstructure:"source_category"`
 	// Desired source name.
 	// Useful if you want to override the source name configured for the source.
+	// Placeholders `%{attr_name}` will be replaced with attribute value for attr_name.
 	SourceName string `mapstructure:"source_name"`
 	// Desired host name.
 	// Useful if you want to override the source host configured for the source.
+	// Placeholders `%{attr_name}` will be replaced with attribute value for attr_name.
 	SourceHost string `mapstructure:"source_host"`
 	// Name of the client
 	Client string `mapstructure:"client"`
@@ -124,4 +130,6 @@ const (
 	DefaultSourceHost string = ""
 	// DefaultClient defines default Client
 	DefaultClient string = "otelcol"
+	// DefaultGraphiteTemplate defines default template for Graphite
+	DefaultGraphiteTemplate string = "%{_metric_}"
 )

exporter/sumologicexporter/exporter.go

@@ -32,6 +32,7 @@ type sumologicexporter struct {
 	client              *http.Client
 	filter              filter
 	prometheusFormatter prometheusFormatter
+	graphiteFormatter   graphiteFormatter
 }
 
 func initExporter(cfg *Config) (*sumologicexporter, error) {
@@ -77,6 +78,11 @@ func initExporter(cfg *Config) (*sumologicexporter, error) {
 		return nil, err
 	}
 
+	gf, err := newGraphiteFormatter(cfg.GraphiteTemplate)
+	if err != nil {
+		return nil, err
+	}
+
 	httpClient, err := cfg.HTTPClientSettings.ToClient()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create HTTP Client: %w", err)
@@ -88,6 +94,7 @@ func initExporter(cfg *Config) (*sumologicexporter, error) {
 		client:              httpClient,
 		filter:              f,
 		prometheusFormatter: pf,
+		graphiteFormatter:   gf,
 	}
 
 	return se, nil
@@ -151,7 +158,15 @@ func (se *sumologicexporter) pushLogsData(ctx context.Context, ld pdata.Logs) er
 	if err != nil {
 		return consumererror.NewLogs(fmt.Errorf("failed to initialize compressor: %w", err), ld)
 	}
-	sdr := newSender(se.config, se.client, se.filter, se.sources, c, se.prometheusFormatter)
+	sdr := newSender(
+		se.config,
+		se.client,
+		se.filter,
+		se.sources,
+		c,
+		se.prometheusFormatter,
+		se.graphiteFormatter,
+	)
 
 	// Iterate over ResourceLogs
 	rls := ld.ResourceLogs()
@@ -246,7 +261,15 @@ func (se *sumologicexporter) pushMetricsData(ctx context.Context, md pdata.Metri
 	if err != nil {
 		return consumererror.NewMetrics(fmt.Errorf("failed to initialize compressor: %w", err), md)
 	}
-	sdr := newSender(se.config, se.client, se.filter, se.sources, c, se.prometheusFormatter)
+	sdr := newSender(
+		se.config,
+		se.client,
+		se.filter,
+		se.sources,
+		c,
+		se.prometheusFormatter,
+		se.graphiteFormatter,
+	)
 
 	// Iterate over ResourceMetrics
 	rms := md.ResourceMetrics()

exporter/sumologicexporter/factory.go

@@ -53,6 +53,7 @@ func createDefaultConfig() config.Exporter {
 		SourceName:         DefaultSourceName,
 		SourceHost:         DefaultSourceHost,
 		Client:             DefaultClient,
+		GraphiteTemplate:   DefaultGraphiteTemplate,
 
 		HTTPClientSettings: CreateDefaultHTTPClientSettings(),
 		RetrySettings:      exporterhelper.DefaultRetrySettings(),

exporter/sumologicexporter/factory_test.go

@@ -47,6 +47,7 @@ func TestCreateDefaultConfig(t *testing.T) {
 		SourceName:         "",
 		SourceHost:         "",
 		Client:             "otelcol",
+		GraphiteTemplate:   "%{_metric_}",
 
 		HTTPClientSettings: confighttp.HTTPClientSettings{
 			Timeout: 5 * time.Second,

exporter/sumologicexporter/fields.go

@@ -32,7 +32,7 @@ type fields struct {
 func newFields(attrMap pdata.AttributeMap) fields {
 	return fields{
 		orig:     attrMap,
-		replacer: strings.NewReplacer(",", "_", "=", "_", "\n", "_"),
+		replacer: strings.NewReplacer(",", "_", "=", ":", "\n", "_"),
 	}
 }
 

exporter/sumologicexporter/fields_test.go

@@ -32,7 +32,7 @@ func TestFieldsAsString(t *testing.T) {
 }
 
 func TestFieldsSanitization(t *testing.T) {
-	expected := "key1=value_1, key3=value_3, key__2=valu_e_2"
+	expected := "key1=value_1, key3=value_3, key:_2=valu_e:2"
 	flds := fieldsFromMap(map[string]string{
 		"key1":   "value,1",
 		"key3":   "value\n3",

exporter/sumologicexporter/graphite_formatter.go

@@ -0,0 +1,138 @@
+// Copyright 2021, OpenTelemetry 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 sumologicexporter
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"go.opentelemetry.io/collector/consumer/pdata"
+	tracetranslator "go.opentelemetry.io/collector/translator/trace"
+)
+
+type graphiteFormatter struct {
+	template sourceFormat
+	replacer *strings.Replacer
+}
+
+const (
+	graphiteMetricNamePlaceholder = "_metric_"
+)
+
+// newGraphiteFormatter creates new formatter for given SourceFormat template
+func newGraphiteFormatter(template string) (graphiteFormatter, error) {
+	r, err := regexp.Compile(sourceRegex)
+	if err != nil {
+		return graphiteFormatter{}, err
+	}
+
+	sf := newSourceFormat(r, template)
+
+	return graphiteFormatter{
+		template: sf,
+		replacer: strings.NewReplacer(`.`, `_`, ` `, `_`),
+	}, nil
+}
+
+// escapeGraphiteString replaces dot and space using replacer,
+// as dot is special character for graphite format
+func (gf *graphiteFormatter) escapeGraphiteString(value string) string {
+	return gf.replacer.Replace(value)
+}
+
+// format returns metric name basing on template for given fields nas metric name
+func (gf *graphiteFormatter) format(f fields, metricName string) string {
+	s := gf.template
+	labels := make([]interface{}, 0, len(s.matches))
+
+	for _, matchset := range s.matches {
+		if matchset == graphiteMetricNamePlaceholder {
+			labels = append(labels, gf.escapeGraphiteString(metricName))
+		} else {
+			attr, ok := f.orig.Get(matchset)
+			var value string
+			if ok {
+				value = tracetranslator.AttributeValueToString(attr, false)
+			} else {
+				value = ""
+			}
+			labels = append(labels, gf.escapeGraphiteString(value))
+		}
+	}
+
+	return fmt.Sprintf(s.template, labels...)
+}
+
+// intRecord converts IntDataPoint to graphite metric string
+// with additional information from fields
+func (gf *graphiteFormatter) intRecord(fs fields, name string, dataPoint pdata.IntDataPoint) string {
+	return fmt.Sprintf("%s %d %d",
+		gf.format(fs, name),
+		dataPoint.Value(),
+		dataPoint.Timestamp()/pdata.Timestamp(time.Second),
+	)
+}
+
+// doubleRecord converts DoubleDataPoint to graphite metric string
+// with additional information from fields
+func (gf *graphiteFormatter) doubleRecord(fs fields, name string, dataPoint pdata.DoubleDataPoint) string {
+	return fmt.Sprintf("%s %g %d",
+		gf.format(fs, name),
+		dataPoint.Value(),
+		dataPoint.Timestamp()/pdata.Timestamp(time.Second),
+	)
+}
+
+// metric2String returns stringified metricPair
+func (gf *graphiteFormatter) metric2String(record metricPair) string {
+	var nextLines []string
+	fs := newFields(record.attributes)
+	name := record.metric.Name()
+
+	switch record.metric.DataType() {
+	case pdata.MetricDataTypeIntGauge:
+		dps := record.metric.IntGauge().DataPoints()
+		nextLines = make([]string, 0, dps.Len())
+		for i := 0; i < dps.Len(); i++ {
+			nextLines = append(nextLines, gf.intRecord(fs, name, dps.At(i)))
+		}
+	case pdata.MetricDataTypeIntSum:
+		dps := record.metric.IntSum().DataPoints()
+		nextLines = make([]string, 0, dps.Len())
+		for i := 0; i < dps.Len(); i++ {
+			nextLines = append(nextLines, gf.intRecord(fs, name, dps.At(i)))
+		}
+	case pdata.MetricDataTypeDoubleGauge:
+		dps := record.metric.DoubleGauge().DataPoints()
+		nextLines = make([]string, 0, dps.Len())
+		for i := 0; i < dps.Len(); i++ {
+			nextLines = append(nextLines, gf.doubleRecord(fs, name, dps.At(i)))
+		}
+	case pdata.MetricDataTypeDoubleSum:
+		dps := record.metric.DoubleSum().DataPoints()
+		nextLines = make([]string, 0, dps.Len())
+		for i := 0; i < dps.Len(); i++ {
+			nextLines = append(nextLines, gf.doubleRecord(fs, name, dps.At(i)))
+		}
+	// Skip complex metrics
+	case pdata.MetricDataTypeDoubleHistogram:
+	case pdata.MetricDataTypeIntHistogram:
+	case pdata.MetricDataTypeDoubleSummary:
+	}
+
+	return strings.Join(nextLines, "\n")
+}