Add signalfx metric token passthrough (#325)

Adding an `access_token_passthrough` configuration option and feature to the SignalFx metric receiver and exporter to allow continued association of datapoints with an [SFx access token](https://docs.signalfx.com/en/latest/admin-guide/tokens.html).  This will provide the option having collector-proxied datapoints keep the initial access token set by metric clients instead of requiring all datapoints being tied to the access token sourced from the SFx exporter configuration.

**Testing:** Updated relevant unit tests and tested local deployment with multiple clients and batch processor.

**Documentation:** Updated readme for both receiver and exporter.

Author: rmfitzpatrick

exporter/signalfxexporter/README.md

@@ -26,6 +26,7 @@ be used instead. If path is not specified, `/v2/datapoint` is used.
 If `realm` is set, this option is derived and will be `https://api.{realm}.signalfx.com/`. If a value is explicitly
 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.
 
 Note: Either `realm` or both `ingest_url` and `api_url` should be explicitly set.
 
@@ -35,6 +36,7 @@ Example:
 exporters:
   signalfx:
     access_token: <replace_with_actual_access_token>
+    access_token_passthrough: true
     headers:
       added-entry: "added value"
       dot.test: test

exporter/signalfxexporter/config.go

@@ -22,6 +22,8 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/config/configmodels"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 // Config defines configuration for SignalFx exporter.
@@ -58,6 +60,8 @@ type Config struct {
 
 	// Whether to log dimension updates being sent to SignalFx.
 	LogDimensionUpdates bool `mapstructure:"log_dimension_updates"`
+
+	splunk.AccessTokenPassthroughConfig `mapstructure:",squash"`
 }
 
 func (cfg *Config) getOptionsFromConfig() (*exporterOptions, error) {

exporter/signalfxexporter/config_test.go

@@ -25,6 +25,8 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 func TestLoadConfig(t *testing.T) {
@@ -60,6 +62,9 @@ func TestLoadConfig(t *testing.T) {
 			"dot.test":    "test",
 		},
 		Timeout: 2 * time.Second,
+		AccessTokenPassthroughConfig: splunk.AccessTokenPassthroughConfig{
+			AccessTokenPassthrough: false,
+		},
 	}
 	assert.Equal(t, &expectedCfg, e1)
 

exporter/signalfxexporter/dpclient.go

@@ -31,22 +31,25 @@ import (
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 // sfxDPClient sends the data to the SignalFx backend.
 type sfxDPClient struct {
-	ingestURL *url.URL
-	headers   map[string]string
-	client    *http.Client
-	logger    *zap.Logger
-	zippers   sync.Pool
+	ingestURL              *url.URL
+	headers                map[string]string
+	client                 *http.Client
+	logger                 *zap.Logger
+	zippers                sync.Pool
+	accessTokenPassthrough bool
 }
 
 func (s *sfxDPClient) pushMetricsData(
 	ctx context.Context,
 	md consumerdata.MetricsData,
 ) (droppedTimeSeries int, err error) {
-
+	accessToken := s.retrieveAccessToken(md)
 	sfxDataPoints, numDroppedTimeseries, err := metricDataToSignalFxV2(s.logger, md)
 	if err != nil {
 		return exporterhelper.NumTimeSeries(md), consumererror.Permanent(err)
@@ -66,6 +69,10 @@ func (s *sfxDPClient) pushMetricsData(
 		req.Header.Set(k, v)
 	}
 
+	if s.accessTokenPassthrough && accessToken != "" {
+		req.Header.Set(splunk.SFxAccessTokenHeader, accessToken)
+	}
+
 	if compressed {
 		req.Header.Set("Content-Encoding", "gzip")
 	}
@@ -98,7 +105,7 @@ func buildHeaders(config *Config) (map[string]string, error) {
 	}
 
 	if config.AccessToken != "" {
-		headers["X-Sf-Token"] = config.AccessToken
+		headers[splunk.SFxAccessTokenHeader] = config.AccessToken
 	}
 
 	// Add any custom headers from the config. They will override the pre-defined
@@ -122,6 +129,16 @@ func (s *sfxDPClient) encodeBody(dps []*sfxpb.DataPoint) (bodyReader io.Reader,
 	return s.getReader(body)
 }
 
+func (s *sfxDPClient) retrieveAccessToken(md consumerdata.MetricsData) string {
+	accessToken := ""
+	if labels := md.Resource.GetLabels(); labels != nil {
+		accessToken = labels[splunk.SFxAccessTokenLabel]
+		// Drop internally passed access token in all cases
+		delete(labels, splunk.SFxAccessTokenLabel)
+	}
+	return accessToken
+}
+
 // avoid attempting to compress things that fit into a single ethernet frame
 func (s *sfxDPClient) getReader(b []byte) (io.Reader, bool, error) {
 	var err error

exporter/signalfxexporter/exporter.go

@@ -81,6 +81,7 @@ func New(
 		zippers: sync.Pool{New: func() interface{} {
 			return gzip.NewWriter(nil)
 		}},
+		accessTokenPassthrough: config.AccessTokenPassthrough,
 	}
 
 	dimClient := dimensions.NewDimensionClient(

exporter/signalfxexporter/exporter_test.go

@@ -144,6 +144,103 @@ func TestConsumeMetricsData(t *testing.T) {
 	}
 }
 
+func TestConsumeMetricsDataWithAccessTokenPassthrough(t *testing.T) {
+	fromHeaders := "AccessTokenFromClientHeaders"
+	fromLabels := "AccessTokenFromLabel"
+
+	newMetricData := func(includeToken bool) consumerdata.MetricsData {
+		md := consumerdata.MetricsData{
+			Node: &commonpb.Node{
+				ServiceInfo: &commonpb.ServiceInfo{Name: "test_signalfx"},
+			},
+			Resource: &resourcepb.Resource{
+				Type: "test",
+				Labels: map[string]string{
+					"com.splunk.signalfx.access_token": fromLabels,
+				},
+			},
+			Metrics: []*metricspb.Metric{
+				metricstestutils.Gauge(
+					"test_gauge",
+					[]string{"k0", "k1"},
+					metricstestutils.Timeseries(
+						time.Now(),
+						[]string{"v0", "v1"},
+						metricstestutils.Double(time.Now(), 123))),
+			},
+		}
+		if !includeToken {
+			delete(md.Resource.Labels, "com.splunk.signalfx.access_token")
+		}
+		return md
+	}
+
+	tests := []struct {
+		name                   string
+		accessTokenPassthrough bool
+		includedInMetricData   bool
+		expectedToken          string
+	}{
+		{
+			name:                   "passthrough access token and included in md",
+			accessTokenPassthrough: true,
+			includedInMetricData:   true,
+			expectedToken:          fromLabels,
+		},
+		{
+			name:                   "passthrough access token and not included in md",
+			accessTokenPassthrough: true,
+			includedInMetricData:   false,
+			expectedToken:          fromHeaders,
+		},
+		{
+			name:                   "don't passthrough access token and included in md",
+			accessTokenPassthrough: false,
+			includedInMetricData:   true,
+			expectedToken:          fromHeaders,
+		},
+		{
+			name:                   "don't passthrough access token and not included in md",
+			accessTokenPassthrough: false,
+			includedInMetricData:   false,
+			expectedToken:          fromHeaders,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				assert.Equal(t, "test", r.Header.Get("test_header_"))
+				assert.Equal(t, tt.expectedToken, r.Header.Get("x-sf-token"))
+				w.WriteHeader(http.StatusAccepted)
+			}))
+			defer server.Close()
+
+			serverURL, err := url.Parse(server.URL)
+			assert.NoError(t, err)
+
+			dpClient := &sfxDPClient{
+				ingestURL: serverURL,
+				headers: map[string]string{
+					"test_header_": "test",
+					"X-Sf-Token":   fromHeaders,
+				},
+				client: &http.Client{
+					Timeout: 1 * time.Second,
+				},
+				logger: zap.NewNop(),
+				zippers: sync.Pool{New: func() interface{} {
+					return gzip.NewWriter(nil)
+				}},
+				accessTokenPassthrough: tt.accessTokenPassthrough,
+			}
+
+			numDroppedTimeSeries, err := dpClient.pushMetricsData(context.Background(), newMetricData(tt.includedInMetricData))
+			assert.Equal(t, 0, numDroppedTimeSeries)
+			assert.NoError(t, err)
+		})
+	}
+}
+
 func generateLargeBatch(t *testing.T) *consumerdata.MetricsData {
 	md := &consumerdata.MetricsData{
 		Node: &commonpb.Node{

exporter/signalfxexporter/factory.go

@@ -21,6 +21,8 @@ import (
 	"go.opentelemetry.io/collector/config/configerror"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 const (
@@ -47,6 +49,9 @@ func (f *Factory) CreateDefaultConfig() configmodels.Exporter {
 			NameVal: typeStr,
 		},
 		Timeout: defaultHTTPTimeout,
+		AccessTokenPassthroughConfig: splunk.AccessTokenPassthroughConfig{
+			AccessTokenPassthrough: true,
+		},
 	}
 }
 

exporter/signalfxexporter/go.mod

@@ -5,6 +5,7 @@ go 1.14
 require (
 	github.com/census-instrumentation/opencensus-proto v0.2.1
 	github.com/golang/protobuf v1.3.5
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.0.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver v0.0.0
 	github.com/signalfx/com_signalfx_metrics_protobuf v0.0.0-20190530013331-054be550cb49
 	github.com/stretchr/testify v1.5.1

exporter/signalfxexporter/testdata/config.yaml

@@ -14,6 +14,7 @@ exporters:
     headers:
       added-entry: "added value"
       dot.test: test
+    access_token_passthrough: false
 
 service:
   pipelines:

internal/common/splunk/common.go

@@ -0,0 +1,25 @@
+// Copyright 2020, 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 splunk
+
+const (
+	SFxAccessTokenHeader = "X-Sf-Token"
+	SFxAccessTokenLabel  = "com.splunk.signalfx.access_token"
+)
+
+type AccessTokenPassthroughConfig struct {
+	// Whether to associate datapoints with an organization access token received in request.
+	AccessTokenPassthrough bool `mapstructure:"access_token_passthrough"`
+}

receiver/signalfxreceiver/README.md

@@ -12,12 +12,14 @@ Example:
 receivers:
   signalfx:
     endpoint: localhost:7276
+    access_token_passthrough: true
     tls:
       cert_file: /test.crt
       key_file: /test.key
 ```
 
 * `endpoint`: Address and port that the SignalFx receiver should bind to. Note that this must be 0.0.0.0:<port> instead of localhost if you want to receive spans from sources exporting to IPs other than localhost on the same host. For example, when the collector is deployed as a k8s deployment and exposed using a service.
+* `access_token_passthrough`: (default = `false`) Whether to preserve incoming access token (`X-Sf-Token` header value) as `"com.splunk.signalfx.access_token"` metric resource label.  Can be used in tandem with identical configuration option for [SignalFx exporter](../../exporter/signalfxexporter/README.md) to preserve datapoint origin.
 * `tls`: This is an optional object used to specify if TLS should be used for incoming connections.
     * `cert_file`: Specifies the certificate file to use for TLS connection.  Note: Both `key_file` and `cert_file` are required for TLS connection.
     * `key_file`: Specifies the key file to use for TLS connection. Note: Both `key_file` and `cert_file` are required for TLS connection.

receiver/signalfxreceiver/config.go

@@ -17,6 +17,8 @@ package signalfxreceiver
 import (
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/config/configtls"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 // Config defines configuration for the SignalFx receiver.
@@ -26,4 +28,6 @@ type Config struct {
 	// Configures the receiver to use TLS.
 	// The default value is nil, which will cause the receiver to not use TLS.
 	TLSCredentials *configtls.TLSSetting `mapstructure:"tls, omitempty"`
+
+	splunk.AccessTokenPassthroughConfig `mapstructure:",squash"`
 }

receiver/signalfxreceiver/config_test.go

@@ -23,6 +23,8 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/config/configtls"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 func TestLoadConfig(t *testing.T) {
@@ -51,6 +53,9 @@ func TestLoadConfig(t *testing.T) {
 				NameVal:  "signalfx/allsettings",
 				Endpoint: "localhost:8080",
 			},
+			AccessTokenPassthroughConfig: splunk.AccessTokenPassthroughConfig{
+				AccessTokenPassthrough: true,
+			},
 		})
 
 	r2 := cfg.Receivers["signalfx/tls"].(*Config)
@@ -64,5 +69,8 @@ func TestLoadConfig(t *testing.T) {
 				CertFile: "/test.crt",
 				KeyFile:  "/test.key",
 			},
+			AccessTokenPassthroughConfig: splunk.AccessTokenPassthroughConfig{
+				AccessTokenPassthrough: false,
+			},
 		})
 }

receiver/signalfxreceiver/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/golang/protobuf v1.3.5
 	github.com/gorilla/mux v1.7.3
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter v0.0.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.0.0
 	github.com/signalfx/com_signalfx_metrics_protobuf v0.0.0-20190530013331-054be550cb49
 	github.com/stretchr/testify v1.5.1
 	go.opencensus.io v0.22.3

receiver/signalfxreceiver/receiver.go

@@ -25,6 +25,7 @@ import (
 	"time"
 	"unsafe"
 
+	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
 	"github.com/golang/protobuf/proto"
 	"github.com/gorilla/mux"
 	sfxpb "github.com/signalfx/com_signalfx_metrics_protobuf"
@@ -35,6 +36,8 @@ import (
 	"go.opentelemetry.io/collector/obsreport"
 	"go.opentelemetry.io/collector/translator/conventions"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 const (
@@ -211,6 +214,18 @@ func (r *sfxReceiver) handleReq(resp http.ResponseWriter, req *http.Request) {
 
 	md, _ := SignalFxV2ToMetricsData(r.logger, msg.Datapoints)
 
+	if r.config.AccessTokenPassthrough {
+		if accessToken := req.Header.Get(splunk.SFxAccessTokenHeader); accessToken != "" {
+			if md.Resource == nil {
+				md.Resource = &resourcepb.Resource{}
+			}
+			if md.Resource.Labels == nil {
+				md.Resource.Labels = make(map[string]string, 1)
+			}
+			md.Resource.Labels[splunk.SFxAccessTokenLabel] = accessToken
+		}
+	}
+
 	err = r.nextConsumer.ConsumeMetricsData(ctx, *md)
 	obsreport.EndMetricsReceiveOp(
 		ctx,