Add ability to optionally translate metrics in signalfx exporter

This commit provides ability to translate metrics in siganlfx exporter. If an optional "send_compatible_metrics" configuration flag is enabled signalfx exporter translates metrics before sending them out according to `translation_rules` configuration. If "translation_rules" field is not set, default translation rules are used that defined in exporter/signalfxexporter/translation/constants.go.

Only dimension renaming is added in this commit to validate the approach. Dimension translation applied to metric dimensions as well as to metadata properties update. Other translation rules will be added in the following commits

exporter/signalfxexporter/README.md

@@ -27,6 +27,11 @@ If `realm` is set, this option is derived and will be `https://api.{realm}.signa
 set, the value of `realm` will not be used in determining `api_url`. The explicit value will be used instead.
 - `log_dimension_updates` (default = `false`): Whether or not to log dimension updates.
 - `access_token_passthrough`: (default = `true`) Whether to use `"com.splunk.signalfx.access_token"` metric resource label, if any, as SFx access token.  In either case this label will be dropped during final translation.  Intended to be used in tandem with identical configuration option for [SignalFx receiver](../../receiver/signalfxreceiver/README.md) to preserve datapoint origin.
+- `send_compatible_metrics` (default = `false`): Whether metrics must be translated to a format 
+backward-compatible with SignalFx naming conventions.
+- `translation_rules`: Set of rules on how to translate metrics to a SignalFx compatible format
+If not provided explicitly, the rules defined in `translations/config/default.yaml` are used.
+Used only when `send_compatible_metrics` set to `true`.
 
 Note: Either `realm` or both `ingest_url` and `api_url` should be explicitly set.
 

exporter/signalfxexporter/config.go

@@ -23,6 +23,7 @@ import (
 
 	"go.opentelemetry.io/collector/config/configmodels"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter/translation"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
@@ -62,6 +63,14 @@ type Config struct {
 	LogDimensionUpdates bool `mapstructure:"log_dimension_updates"`
 
 	splunk.AccessTokenPassthroughConfig `mapstructure:",squash"`
+
+	// SendCompatibleMetrics specifies if metrics must be sent in a format backward-compatible with
+	// SignalFx naming conventions, "false" by default.
+	SendCompatibleMetrics bool `mapstructure:"send_compatible_metrics"`
+
+	// TranslationRules defines a set of rules how to translate metrics to a SignalFx compatible format
+	// If not provided explicitly, the rules defined in translations/config/default.yaml are used.
+	TranslationRules []translation.Rule `mapstructure:"translation_rules"`
 }
 
 func (cfg *Config) getOptionsFromConfig() (*exporterOptions, error) {
@@ -83,12 +92,21 @@ func (cfg *Config) getOptionsFromConfig() (*exporterOptions, error) {
 		cfg.Timeout = 5 * time.Second
 	}
 
+	var metricTranslator *translation.MetricTranslator
+	if cfg.SendCompatibleMetrics {
+		metricTranslator, err = translation.NewMetricTranslator(cfg.TranslationRules)
+		if err != nil {
+			return nil, fmt.Errorf("invalid \"translation_rules\": %v", err)
+		}
+	}
+
 	return &exporterOptions{
-		ingestURL:    ingestURL,
-		apiURL:       apiURL,
-		httpTimeout:  cfg.Timeout,
-		token:        cfg.AccessToken,
-		logDimUpdate: cfg.LogDimensionUpdates,
+		ingestURL:        ingestURL,
+		apiURL:           apiURL,
+		httpTimeout:      cfg.Timeout,
+		token:            cfg.AccessToken,
+		logDimUpdate:     cfg.LogDimensionUpdates,
+		metricTranslator: metricTranslator,
 	}, nil
 }
 

exporter/signalfxexporter/config_test.go

@@ -26,6 +26,7 @@ import (
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter/translation"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
@@ -65,6 +66,15 @@ func TestLoadConfig(t *testing.T) {
 		AccessTokenPassthroughConfig: splunk.AccessTokenPassthroughConfig{
 			AccessTokenPassthrough: false,
 		},
+		SendCompatibleMetrics: true,
+		TranslationRules: []translation.Rule{
+			{
+				Action: translation.ActionRenameDimensionKeys,
+				Mapping: map[string]string{
+					"k8s.cluster.name": "kubernetes_cluster",
+				},
+			},
+		},
 	}
 	assert.Equal(t, &expectedCfg, e1)
 
@@ -75,13 +85,15 @@ func TestLoadConfig(t *testing.T) {
 
 func TestConfig_getOptionsFromConfig(t *testing.T) {
 	type fields struct {
-		ExporterSettings configmodels.ExporterSettings
-		AccessToken      string
-		Realm            string
-		IngestURL        string
-		APIURL           string
-		Timeout          time.Duration
-		Headers          map[string]string
+		ExporterSettings      configmodels.ExporterSettings
+		AccessToken           string
+		Realm                 string
+		IngestURL             string
+		APIURL                string
+		Timeout               time.Duration
+		Headers               map[string]string
+		SendCompatibleMetrics bool
+		TranslationRules      []translation.Rule
 	}
 	tests := []struct {
 		name    string
@@ -171,17 +183,34 @@ func TestConfig_getOptionsFromConfig(t *testing.T) {
 			want:    nil,
 			wantErr: true,
 		},
+		{
+			name: "Test invalid translation rules",
+			fields: fields{
+				Realm:                 "us0",
+				AccessToken:           "access_token",
+				SendCompatibleMetrics: true,
+				TranslationRules: []translation.Rule{
+					{
+						Action: translation.ActionRenameDimensionKeys,
+					},
+				},
+			},
+			want:    nil,
+			wantErr: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			cfg := &Config{
-				ExporterSettings: tt.fields.ExporterSettings,
-				AccessToken:      tt.fields.AccessToken,
-				Realm:            tt.fields.Realm,
-				IngestURL:        tt.fields.IngestURL,
-				APIURL:           tt.fields.APIURL,
-				Timeout:          tt.fields.Timeout,
-				Headers:          tt.fields.Headers,
+				ExporterSettings:      tt.fields.ExporterSettings,
+				AccessToken:           tt.fields.AccessToken,
+				Realm:                 tt.fields.Realm,
+				IngestURL:             tt.fields.IngestURL,
+				APIURL:                tt.fields.APIURL,
+				Timeout:               tt.fields.Timeout,
+				Headers:               tt.fields.Headers,
+				SendCompatibleMetrics: tt.fields.SendCompatibleMetrics,
+				TranslationRules:      tt.fields.TranslationRules,
 			}
 			got, err := cfg.getOptionsFromConfig()
 			if (err != nil) != tt.wantErr {

exporter/signalfxexporter/dimensions/dimclient.go

@@ -30,6 +30,8 @@ import (
 	"time"
 
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter/translation"
 )
 
 // DimensionClient sends updates to dimensions to the SignalFx API
@@ -70,6 +72,7 @@ type DimensionClient struct {
 	TotalSuccessfulUpdates       int64
 	logUpdates                   bool
 	logger                       *zap.Logger
+	metricTranslator             *translation.MetricTranslator
 }
 
 type queuedDimension struct {
@@ -84,6 +87,7 @@ type DimensionClientOptions struct {
 	Logger                *zap.Logger
 	SendDelay             int
 	PropertiesMaxBuffered int
+	MetricTranslator      *translation.MetricTranslator
 }
 
 // NewDimensionClient returns a new client
@@ -107,17 +111,18 @@ func NewDimensionClient(ctx context.Context, options DimensionClientOptions) *Di
 	sender := NewReqSender(ctx, client, 20, map[string]string{"client": "dimension"})
 
 	return &DimensionClient{
-		ctx:           ctx,
-		Token:         options.Token,
-		APIURL:        options.APIURL,
-		sendDelay:     time.Duration(options.SendDelay) * time.Second,
-		delayedSet:    make(map[DimensionKey]*DimensionUpdate),
-		delayedQueue:  make(chan *queuedDimension, options.PropertiesMaxBuffered),
-		requestSender: sender,
-		client:        client,
-		now:           time.Now,
-		logger:        options.Logger,
-		logUpdates:    options.LogUpdates,
+		ctx:              ctx,
+		Token:            options.Token,
+		APIURL:           options.APIURL,
+		sendDelay:        time.Duration(options.SendDelay) * time.Second,
+		delayedSet:       make(map[DimensionKey]*DimensionUpdate),
+		delayedQueue:     make(chan *queuedDimension, options.PropertiesMaxBuffered),
+		requestSender:    sender,
+		client:           client,
+		now:              time.Now,
+		logger:           options.Logger,
+		logUpdates:       options.LogUpdates,
+		metricTranslator: options.MetricTranslator,
 	}
 }
 

exporter/signalfxexporter/dimensions/kubernetesmetadata.go

@@ -21,30 +21,46 @@ import (
 
 	"go.opentelemetry.io/collector/component/componenterror"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter/translation"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/collection"
 )
 
 var propNameSanitizer = strings.NewReplacer(
 	".", "_",
 	"/", "_")
 
-func getDimensionUpdateFromMetadata(metadata collection.KubernetesMetadataUpdate) *DimensionUpdate {
-	properties, tags := getPropertiesAndTags(metadata)
+func getDimensionUpdateFromMetadata(
+	metadata collection.KubernetesMetadataUpdate,
+	metricTranslator *translation.MetricTranslator,
+) *DimensionUpdate {
+
+	translateDimension := func(dim string) string {
+		res := dim
+		if metricTranslator != nil {
+			res = metricTranslator.TranslateDimension(res)
+		}
+		return propNameSanitizer.Replace(res)
+	}
+
+	properties, tags := getPropertiesAndTags(metadata, translateDimension)
 
 	return &DimensionUpdate{
-		Name:       propNameSanitizer.Replace(metadata.ResourceIDKey),
+		Name:       translateDimension(metadata.ResourceIDKey),
 		Value:      string(metadata.ResourceID),
 		Properties: properties,
 		Tags:       tags,
 	}
 }
 
-func getPropertiesAndTags(kmu collection.KubernetesMetadataUpdate) (map[string]*string, map[string]bool) {
+func getPropertiesAndTags(
+	kmu collection.KubernetesMetadataUpdate,
+	translate func(string) string,
+) (map[string]*string, map[string]bool) {
 	properties := map[string]*string{}
 	tags := map[string]bool{}
 
 	for label, val := range kmu.MetadataToAdd {
-		key := propNameSanitizer.Replace(label)
+		key := translate(label)
 		if key == "" {
 			continue
 		}
@@ -58,7 +74,7 @@ func getPropertiesAndTags(kmu collection.KubernetesMetadataUpdate) (map[string]*
 	}
 
 	for label, val := range kmu.MetadataToRemove {
-		key := propNameSanitizer.Replace(label)
+		key := translate(label)
 		if key == "" {
 			continue
 		}
@@ -71,7 +87,7 @@ func getPropertiesAndTags(kmu collection.KubernetesMetadataUpdate) (map[string]*
 	}
 
 	for label, val := range kmu.MetadataToUpdate {
-		key := propNameSanitizer.Replace(label)
+		key := translate(label)
 		if key == "" {
 			continue
 		}
@@ -92,7 +108,7 @@ func getPropertiesAndTags(kmu collection.KubernetesMetadataUpdate) (map[string]*
 func (dc *DimensionClient) PushKubernetesMetadata(metadata []*collection.KubernetesMetadataUpdate) error {
 	var errs []error
 	for _, m := range metadata {
-		dimensionUpdate := getDimensionUpdateFromMetadata(*m)
+		dimensionUpdate := getDimensionUpdateFromMetadata(*m, dc.metricTranslator)
 
 		if dimensionUpdate.Name == "" || dimensionUpdate.Value == "" {
 			atomic.AddInt64(&dc.TotalInvalidDimensions, int64(1))