Add SAPM AccessTokenPassthrough (#349)

Adding the `access_token_passthrough` config option and functionality to SAPM receiver and exporter similar to that of the SignalFx metric receiver and exporter from f361e4d.  These changes add a `com.splunk.signalfx.access_token` attribute with a value taken from `X-Sf-Token` headers, if provided, that will be used for subsequent trace submissions in the SAPM exporter by its client.

**Testing:** Added and modified unit tests and e2e run configuration.

**Documentation:** Updated relevant readmes.

Author: rmfitzpatrick

exporter/sapmexporter/README.md

@@ -15,13 +15,18 @@ idle HTTP connection the exporter can keep open.
 - `num_workers` (default = 8): NumWorkers is the number of workers that should be used to
 export traces. Exporter can make as many requests in parallel as the number of workers. Note
 that this will likely be removed in future in favour of processors handling parallel exporting.
+- `access_token_passthrough`: (default = `true`) Whether to use `"com.splunk.signalfx.access_token"`
+trace resource attribute, if any, as SFx access token.  In either case this attribute will be deleted
+during final translation.  Intended to be used in tandem with identical configuration option for
+[SAPM receiver](../../receiver/sapmreceiver/README.md) to preserve trace origin.
 
 Example:
 
 ```yaml
 exporters:
   sapm:
     access_token: YOUR_ACCESS_TOKEN
+    access_token_passthrough: true
     endpoint: https://ingest.YOUR_SIGNALFX_REALM.signalfx.com/v2/trace
     max_connections: 100
     num_workers: 8

exporter/sapmexporter/config.go

@@ -20,6 +20,8 @@ import (
 
 	sapmclient "github.com/signalfx/sapm-proto/client"
 	"go.opentelemetry.io/collector/config/configmodels"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 const (
@@ -47,6 +49,8 @@ type Config struct {
 
 	// Disable GZip compression.
 	DisableCompression bool `mapstructure:"disable_compression"`
+
+	splunk.AccessTokenPassthroughConfig `mapstructure:",squash"`
 }
 
 func (c *Config) validate() error {

exporter/sapmexporter/config_test.go

@@ -22,6 +22,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configmodels"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 func TestLoadConfig(t *testing.T) {
@@ -50,5 +52,27 @@ func TestLoadConfig(t *testing.T) {
 			AccessToken:      "abcd1234",
 			NumWorkers:       3,
 			MaxConnections:   45,
+			AccessTokenPassthroughConfig: splunk.AccessTokenPassthroughConfig{
+				AccessTokenPassthrough: false,
+			},
 		})
 }
+
+func TestInvalidConfig(t *testing.T) {
+	invalid := Config{
+		AccessToken:    "abcd1234",
+		NumWorkers:     3,
+		MaxConnections: 45,
+	}
+	noEndpointErr := invalid.validate()
+	require.Error(t, noEndpointErr)
+
+	invalid = Config{
+		Endpoint:       ":123:456",
+		AccessToken:    "abcd1234",
+		NumWorkers:     3,
+		MaxConnections: 45,
+	}
+	invalidURLErr := invalid.validate()
+	require.Error(t, invalidURLErr)
+}

exporter/sapmexporter/exporter.go

@@ -25,53 +25,115 @@ import (
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/translator/trace/jaeger"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/splunk"
 )
 
 // sapmExporter is a wrapper struct of SAPM exporter
 type sapmExporter struct {
 	client *sapmclient.Client
 	logger *zap.Logger
+	config *Config
 }
 
 func (se *sapmExporter) Shutdown(context.Context) error {
 	se.client.Stop()
 	return nil
 }
 
-func newSAPMTraceExporter(cfg *Config, params component.ExporterCreateParams) (component.TraceExporter, error) {
+func newSAPMExporter(cfg *Config, params component.ExporterCreateParams) (sapmExporter, error) {
 	err := cfg.validate()
 	if err != nil {
-		return nil, err
+		return sapmExporter{}, err
 	}
 
 	client, err := sapmclient.New(cfg.clientOptions()...)
 	if err != nil {
-		return nil, err
+		return sapmExporter{}, err
 	}
-	se := sapmExporter{
+	return sapmExporter{
 		client: client,
 		logger: params.Logger,
+		config: cfg,
+	}, err
+}
+
+func newSAPMTraceExporter(cfg *Config, params component.ExporterCreateParams) (component.TraceExporter, error) {
+	se, err := newSAPMExporter(cfg, params)
+	if err != nil {
+		return nil, err
 	}
+
 	return exporterhelper.NewTraceExporter(
 		cfg,
 		se.pushTraceData,
 		exporterhelper.WithShutdown(se.Shutdown))
 }
 
-// pushTraceData exports traces in SAPM proto and returns number of dropped spans and error if export failed
-func (se *sapmExporter) pushTraceData(ctx context.Context, td pdata.Traces) (droppedSpansCount int, err error) {
-	batches, err := jaeger.InternalTracesToJaegerProto(td)
-	if err != nil {
-		return td.SpanCount(), consumererror.Permanent(err)
+// tracesByAccessToken takes a pdata.Traces struct and will iterate through its ResourceSpans' attributes,
+// regrouping by any SFx access token label value if Config.AccessTokenPassthrough is enabled.  It will delete any
+// set token label in any case to prevent serialization.
+// It returns a map of newly constructed pdata.Traces keyed by access token, defaulting to empty string.
+func (se *sapmExporter) tracesByAccessToken(td pdata.Traces) map[string]pdata.Traces {
+	tracesByToken := make(map[string]pdata.Traces, 1)
+	resourceSpans := td.ResourceSpans()
+	for i := 0; i < resourceSpans.Len(); i++ {
+		resourceSpan := resourceSpans.At(i)
+		if resourceSpan.IsNil() {
+			// Invalid trace so nothing to export
+			continue
+		}
+
+		accessToken := ""
+		if !resourceSpan.Resource().IsNil() {
+			attrs := resourceSpan.Resource().Attributes()
+			attributeValue, ok := attrs.Get(splunk.SFxAccessTokenLabel)
+			if ok {
+				attrs.Delete(splunk.SFxAccessTokenLabel)
+				if se.config.AccessTokenPassthrough {
+					accessToken = attributeValue.StringVal()
+				}
+			}
+		}
+
+		traceForToken, ok := tracesByToken[accessToken]
+		if !ok {
+			traceForToken = pdata.NewTraces()
+			tracesByToken[accessToken] = traceForToken
+		}
+
+		// Append ResourceSpan to trace for this access token
+		traceForTokenSize := traceForToken.ResourceSpans().Len()
+		traceForToken.ResourceSpans().Resize(traceForTokenSize + 1)
+		traceForToken.ResourceSpans().At(traceForTokenSize).InitEmpty()
+		resourceSpan.CopyTo(traceForToken.ResourceSpans().At(traceForTokenSize))
 	}
-	err = se.client.Export(ctx, batches)
-	if err != nil {
-		if sendErr, ok := err.(*sapmclient.ErrSend); ok {
-			if sendErr.Permanent {
-				return 0, consumererror.Permanent(sendErr)
+
+	return tracesByToken
+}
+
+// pushTraceData exports traces in SAPM proto by associated SFx access token and returns number of dropped spans
+// and the last experienced error if any translation or export failed
+func (se *sapmExporter) pushTraceData(ctx context.Context, td pdata.Traces) (droppedSpansCount int, err error) {
+	traces := se.tracesByAccessToken(td)
+	droppedSpansCount = 0
+	for accessToken, trace := range traces {
+		batches, translateErr := jaeger.InternalTracesToJaegerProto(trace)
+		if translateErr != nil {
+			droppedSpansCount += trace.SpanCount()
+			err = consumererror.Permanent(translateErr)
+			continue
+		}
+
+		exportErr := se.client.ExportWithAccessToken(ctx, batches, accessToken)
+		if exportErr != nil {
+			if sendErr, ok := exportErr.(*sapmclient.ErrSend); ok {
+				if sendErr.Permanent {
+					err = consumererror.Permanent(sendErr)
+				}
 			}
+			droppedSpansCount += trace.SpanCount()
 		}
-		return td.SpanCount(), err
 	}
-	return 0, nil
+	return
 }