Fix token passthrough for HEC

exporter/splunkhecexporter/client.go

@@ -90,6 +90,13 @@ func (c *client) pushMetricsData(
 		req.Header.Set("Content-Encoding", "gzip")
 	}
 
+	if md.ResourceMetrics().Len() != 0 {
+		accessToken, found := md.ResourceMetrics().At(0).Resource().Attributes().Get(splunk.HecTokenLabel)
+		if found {
+			req.Header.Set("Authorization", splunk.HECTokenHeader+" "+accessToken.StringVal())
+		}
+	}
+
 	resp, err := c.client.Do(req)
 	if err != nil {
 		return err
@@ -136,6 +143,18 @@ func (c *client) pushLogData(ctx context.Context, ld pdata.Logs) error {
 
 	// Callback when each batch is to be sent.
 	send := func(ctx context.Context, buf *bytes.Buffer, headers map[string]string) (err error) {
+		localHeaders := headers
+		if ld.ResourceLogs().Len() != 0 {
+			accessToken, found := ld.ResourceLogs().At(0).Resource().Attributes().Get(splunk.HecTokenLabel)
+			if found {
+				localHeaders = map[string]string{}
+				for k, v := range headers {
+					localHeaders[k] = v
+				}
+				localHeaders["Authorization"] = splunk.HECTokenHeader + " " + accessToken.StringVal()
+			}
+		}
+
 		shouldCompress := buf.Len() >= minCompressionLen && !c.config.DisableCompression
 
 		if shouldCompress {
@@ -150,10 +169,10 @@ func (c *client) pushLogData(ctx context.Context, ld pdata.Logs) error {
 				return fmt.Errorf("failed flushing compressed data to gzip writer: %v", err)
 			}
 
-			return c.postEvents(ctx, gzipBuffer, headers, shouldCompress)
+			return c.postEvents(ctx, gzipBuffer, localHeaders, shouldCompress)
 		}
 
-		return c.postEvents(ctx, buf, headers, shouldCompress)
+		return c.postEvents(ctx, buf, localHeaders, shouldCompress)
 	}
 
 	return c.pushLogDataInBatches(ctx, ld, send)

exporter/splunkhecexporter/factory.go

@@ -21,10 +21,12 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	conventions "go.opentelemetry.io/collector/model/semconv/v1.5.0"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchperresourceattr"
 )
 
 const (
@@ -34,6 +36,18 @@ const (
 	defaultHTTPTimeout = 10 * time.Second
 )
 
+// TODO: Find a place for this to be shared.
+type baseMetricsExporter struct {
+	component.Component
+	consumer.Metrics
+}
+
+// TODO: Find a place for this to be shared.
+type baseLogsExporter struct {
+	component.Component
+	consumer.Logs
+}
+
 // NewFactory creates a factory for Splunk HEC exporter.
 func NewFactory() component.ExporterFactory {
 	return exporterhelper.NewFactory(
@@ -112,7 +126,7 @@ func createMetricsExporter(
 		return nil, err
 	}
 
-	return exporterhelper.NewMetricsExporter(
+	exporter, err := exporterhelper.NewMetricsExporter(
 		expCfg,
 		set,
 		exp.pushMetricsData,
@@ -122,6 +136,16 @@ func createMetricsExporter(
 		exporterhelper.WithQueue(expCfg.QueueSettings),
 		exporterhelper.WithStart(exp.start),
 		exporterhelper.WithShutdown(exp.stop))
+	if err != nil {
+		return nil, err
+	}
+
+	wrapped := &baseMetricsExporter{
+		Component: exporter,
+		Metrics:   batchperresourceattr.NewBatchPerResourceMetrics(splunk.HecTokenLabel, exporter),
+	}
+
+	return wrapped, nil
 }
 
 func createLogsExporter(
@@ -140,7 +164,7 @@ func createLogsExporter(
 		return nil, err
 	}
 
-	return exporterhelper.NewLogsExporter(
+	logsExporter, err := exporterhelper.NewLogsExporter(
 		expCfg,
 		set,
 		exp.pushLogData,
@@ -150,4 +174,15 @@ func createLogsExporter(
 		exporterhelper.WithQueue(expCfg.QueueSettings),
 		exporterhelper.WithStart(exp.start),
 		exporterhelper.WithShutdown(exp.stop))
+
+	if err != nil {
+		return nil, err
+	}
+
+	wrapped := &baseLogsExporter{
+		Component: logsExporter,
+		Logs:      batchperresourceattr.NewBatchPerResourceLogs(splunk.HecTokenLabel, logsExporter),
+	}
+
+	return wrapped, nil
 }

exporter/splunkhecexporter/go.mod

@@ -5,7 +5,9 @@ go 1.17
 require (
 	github.com/census-instrumentation/opencensus-proto v0.3.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.39.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.39.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk v0.39.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchperresourceattr v0.39.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/opencensus v0.39.0
 	github.com/stretchr/testify v1.7.0
 	go.opentelemetry.io/collector v0.39.0
@@ -48,4 +50,6 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/splun
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/opencensus => ../../pkg/translator/opencensus
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchperresourceattr => ../../pkg/batchperresourceattr
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal => ../../internal/coreinternal

exporter/splunkhecexporter/logdata_to_splunk.go

@@ -82,6 +82,8 @@ func mapLogRecordToSplunkEvent(res pdata.Resource, lr pdata.LogRecord, config *C
 			sourcetype = v.StringVal()
 		case indexKey:
 			index = v.StringVal()
+		case splunk.HecTokenLabel:
+			// ignore
 		default:
 			fields[k] = convertAttributeValue(v, logger)
 		}
@@ -98,6 +100,8 @@ func mapLogRecordToSplunkEvent(res pdata.Resource, lr pdata.LogRecord, config *C
 			sourcetype = v.StringVal()
 		case indexKey:
 			index = v.StringVal()
+		case splunk.HecTokenLabel:
+			// ignore
 		default:
 			fields[k] = convertAttributeValue(v, logger)
 		}

exporter/splunkhecexporter/logdata_to_splunk_test.go

@@ -85,6 +85,27 @@ func Test_mapLogRecordToSplunkEvent(t *testing.T) {
 					"myhost", "myapp", "myapp-type"),
 			},
 		},
+		{
+			name: "with_hec_token",
+			logRecordFn: func() pdata.LogRecord {
+				logRecord := pdata.NewLogRecord()
+				logRecord.Body().SetStringVal("mylog")
+				logRecord.Attributes().InsertString(splunk.HecTokenLabel, "mytoken")
+				logRecord.SetTimestamp(ts)
+				return logRecord
+			},
+			logResourceFn: pdata.NewResource,
+			configDataFn: func() *Config {
+				config := createDefaultConfig().(*Config)
+				config.Source = "source"
+				config.SourceType = "sourcetype"
+				return config
+			},
+			wantSplunkEvents: []*splunk.Event{
+				commonLogSplunkEvent("mylog", ts, map[string]interface{}{},
+					"unknown", "source", "sourcetype"),
+			},
+		},
 		{
 			name: "non-string attribute",
 			logRecordFn: func() pdata.LogRecord {

exporter/splunkhecexporter/metricdata_to_splunk.go

@@ -86,6 +86,8 @@ func metricDataToSplunk(logger *zap.Logger, data pdata.Metrics, config *Config)
 				sourceType = v.StringVal()
 			case indexKey:
 				index = v.StringVal()
+			case splunk.HecTokenLabel:
+				// ignore
 			default:
 				commonFields[k] = v.AsString()
 			}

exporter/splunkhecexporter/tracedata_to_splunk.go

@@ -85,6 +85,8 @@ func traceDataToSplunk(logger *zap.Logger, data pdata.Traces, config *Config) ([
 				sourceType = v.StringVal()
 			case indexKey:
 				index = v.StringVal()
+			case splunk.HecTokenLabel:
+				// ignore
 			default:
 				commonFields[k] = v.AsString()
 			}

receiver/splunkhecreceiver/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/splunkhecexporter v0.39.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.39.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk v0.39.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchperresourceattr v0.39.0
 	github.com/stretchr/testify v1.7.0
 	go.opentelemetry.io/collector v0.39.0
 	go.opentelemetry.io/collector/model v0.39.0

receiver/splunkhecreceiver/receiver.go

@@ -24,6 +24,7 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
+	"strings"
 	"sync"
 	"time"
 
@@ -367,10 +368,11 @@ func (r *splunkReceiver) consumeLogs(ctx context.Context, events []*splunk.Event
 
 func (r *splunkReceiver) createResourceCustomizer(req *http.Request) func(resource pdata.Resource) {
 	if r.config.AccessTokenPassthrough {
-		accessToken := req.Header.Get(splunk.HECTokenHeader)
-		if accessToken != "" {
+		accessToken := req.Header.Get("Authorization")
+		if strings.HasPrefix(accessToken, splunk.HECTokenHeader+" ") {
+			accessTokenValue := accessToken[len(splunk.HECTokenHeader)+1:]
 			return func(resource pdata.Resource) {
-				resource.Attributes().InsertString(splunk.HecTokenLabel, accessToken)
+				resource.Attributes().InsertString(splunk.HecTokenLabel, accessTokenValue)
 			}
 		}
 	}