[webhookeventreceiver] add option to include headers as attributes

Author: tredman

.chloggen/webhookeventreceiver-add-option-to-include-headers-as-attributes.yaml

@@ -0,0 +1,30 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receiver/webhookeventreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add option to include headers as log attributes
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: []
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Adds new `convert_headers_to_attributes` option. If set, all headers, with the exception of the
+  required header (if also enabled) will be added as log attributes. Header names are normalized
+  to snake_case and then prefixed with the namespace `header`.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/webhookeventreceiver/README.md

@@ -32,6 +32,7 @@ The following settings are optional:
     * `key` (required if `required_header` config option is set): Represents the key portion of the required header.
     * `value` (required if `required_header` config option is set): Represents the value portion of the required header.
 * `split_logs_at_newline` (default: false): If true, the receiver will create a separate log record for each line in the request body.
+* `convert_headers_to_attributes` (optional): add all request headers (excluding `required_header` if also set) log attributes
 
 ### Split logs at newline example
 

receiver/webhookeventreceiver/config.go

@@ -20,13 +20,14 @@ var (
 
 // Config defines configuration for the Generic Webhook receiver.
 type Config struct {
-	confighttp.ServerConfig `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
-	ReadTimeout             string                   `mapstructure:"read_timeout"`          // wait time for reading request headers in ms. Default is 500ms.
-	WriteTimeout            string                   `mapstructure:"write_timeout"`         // wait time for writing request response in ms. Default is 500ms.
-	Path                    string                   `mapstructure:"path"`                  // path for data collection. Default is /events
-	HealthPath              string                   `mapstructure:"health_path"`           // path for health check api. Default is /health_check
-	RequiredHeader          RequiredHeader           `mapstructure:"required_header"`       // optional setting to set a required header for all requests to have
-	SplitLogsAtNewLine      bool                     `mapstructure:"split_logs_at_newline"` // optional setting to split logs into multiple log records
+	confighttp.ServerConfig    `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
+	ReadTimeout                string                   `mapstructure:"read_timeout"`                  // wait time for reading request headers in ms. Default is 500ms.
+	WriteTimeout               string                   `mapstructure:"write_timeout"`                 // wait time for writing request response in ms. Default is 500ms.
+	Path                       string                   `mapstructure:"path"`                          // path for data collection. Default is /events
+	HealthPath                 string                   `mapstructure:"health_path"`                   // path for health check api. Default is /health_check
+	RequiredHeader             RequiredHeader           `mapstructure:"required_header"`               // optional setting to set a required header for all requests to have
+	SplitLogsAtNewLine         bool                     `mapstructure:"split_logs_at_newline"`         // optional setting to split logs into multiple log records
+	ConvertHeadersToAttributes bool                     `mapstructure:"convert_headers_to_attributes"` // optional to convert all headers to attributes
 }
 
 type RequiredHeader struct {

receiver/webhookeventreceiver/receiver.go

@@ -191,7 +191,7 @@ func (er *eventReceiver) handleReq(w http.ResponseWriter, r *http.Request, _ htt
 
 	// send body into a scanner and then convert the request body into a log
 	sc := bufio.NewScanner(bodyReader)
-	ld, numLogs := reqToLog(sc, r.URL.Query(), er.cfg, er.settings)
+	ld, numLogs := reqToLog(sc, r.Header, r.URL.Query(), er.cfg, er.settings)
 	consumerErr := er.logConsumer.ConsumeLogs(ctx, ld)
 
 	_ = bodyReader.Close()

receiver/webhookeventreceiver/req_to_log.go

@@ -5,7 +5,10 @@ package webhookeventreceiver // import "github.com/open-telemetry/opentelemetry-
 
 import (
 	"bufio"
+	"net/http"
+	"net/textproto"
 	"net/url"
+	"strings"
 	"time"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -16,6 +19,7 @@ import (
 )
 
 func reqToLog(sc *bufio.Scanner,
+	headers http.Header,
 	query url.Values,
 	cfg *Config,
 	settings receiver.Settings,
@@ -45,6 +49,9 @@ func reqToLog(sc *bufio.Scanner,
 	scopeLog.Scope().SetVersion(settings.BuildInfo.Version)
 	scopeLog.Scope().Attributes().PutStr("source", settings.ID.String())
 	scopeLog.Scope().Attributes().PutStr("receiver", metadata.Type.String())
+	if cfg.ConvertHeadersToAttributes {
+		appendHeaders(cfg, scopeLog, headers)
+	}
 
 	for sc.Scan() {
 		logRecord := scopeLog.LogRecords().AppendEmpty()
@@ -64,3 +71,17 @@ func appendMetadata(resourceLog plog.ResourceLogs, query url.Values) {
 		}
 	}
 }
+
+// append headers as attributes
+func appendHeaders(config *Config, scopeLog plog.ScopeLogs, headers http.Header) {
+	for k := range headers {
+		// Skip the required header used for authentication
+		if k == textproto.CanonicalMIMEHeaderKey(config.RequiredHeader.Key) {
+			continue
+		}
+		// store headers with "header" namespace and normalize key to snake_case
+		normalizedHeader := strings.ReplaceAll(k, "-", "_")
+		normalizedHeader = strings.ToLower(normalizedHeader)
+		scopeLog.Scope().Attributes().PutStr("header."+normalizedHeader, strings.Join(headers.Values(k), ";"))
+	}
+}

receiver/webhookeventreceiver/req_to_log_test.go

@@ -8,6 +8,8 @@ import (
 	"bytes"
 	"io"
 	"log"
+	"net/http"
+	"net/textproto"
 	"net/url"
 	"testing"
 
@@ -23,10 +25,12 @@ func TestReqToLog(t *testing.T) {
 	defaultConfig := createDefaultConfig().(*Config)
 
 	tests := []struct {
-		desc  string
-		sc    *bufio.Scanner
-		query url.Values
-		tt    func(t *testing.T, reqLog plog.Logs, reqLen int, settings receiver.Settings)
+		desc    string
+		sc      *bufio.Scanner
+		headers http.Header
+		query   url.Values
+		config  *Config
+		tt      func(t *testing.T, reqLog plog.Logs, reqLen int, settings receiver.Settings)
 	}{
 		{
 			desc: "Valid query valid event",
@@ -112,11 +116,103 @@ func TestReqToLog(t *testing.T) {
 				require.Equal(t, 2, scopeLogsScope.Attributes().Len())
 			},
 		},
+		{
+			desc: "Headers not added by default",
+			headers: http.Header{
+				textproto.CanonicalMIMEHeaderKey("X-Foo"): []string{"1"},
+				textproto.CanonicalMIMEHeaderKey("X-Bar"): []string{"2"},
+			},
+			sc: func() *bufio.Scanner {
+				reader := io.NopCloser(bytes.NewReader([]byte("this is a: log")))
+				return bufio.NewScanner(reader)
+			}(),
+			tt: func(t *testing.T, reqLog plog.Logs, reqLen int, _ receiver.Settings) {
+				require.Equal(t, 1, reqLen)
+
+				attributes := reqLog.ResourceLogs().At(0).Resource().Attributes()
+				require.Equal(t, 0, attributes.Len())
+
+				scopeLogsScope := reqLog.ResourceLogs().At(0).ScopeLogs().At(0).Scope()
+				require.Equal(t, 2, scopeLogsScope.Attributes().Len()) // expect no additional attributes even though headers are set
+			},
+		},
+		{
+			desc: "Headers added if ConvertHeadersToAttributes enabled",
+			headers: http.Header{
+				textproto.CanonicalMIMEHeaderKey("X-Foo"): []string{"1"},
+				textproto.CanonicalMIMEHeaderKey("X-Bar"): []string{"2"},
+			},
+			config: &Config{
+				Path:                       defaultPath,
+				HealthPath:                 defaultHealthPath,
+				ReadTimeout:                defaultReadTimeout,
+				WriteTimeout:               defaultWriteTimeout,
+				ConvertHeadersToAttributes: true,
+			},
+			sc: func() *bufio.Scanner {
+				reader := io.NopCloser(bytes.NewReader([]byte("this is a: log")))
+				return bufio.NewScanner(reader)
+			}(),
+			tt: func(t *testing.T, reqLog plog.Logs, reqLen int, _ receiver.Settings) {
+				require.Equal(t, 1, reqLen)
+
+				attributes := reqLog.ResourceLogs().At(0).Resource().Attributes()
+				require.Equal(t, 0, attributes.Len())
+
+				scopeLogsScope := reqLog.ResourceLogs().At(0).ScopeLogs().At(0).Scope()
+				require.Equal(t, 4, scopeLogsScope.Attributes().Len()) // expect no additional attributes even though headers are set
+				v, exists := scopeLogsScope.Attributes().Get("header.x_foo")
+				require.True(t, exists)
+				require.Equal(t, "1", v.AsString())
+				v, exists = scopeLogsScope.Attributes().Get("header.x_bar")
+				require.True(t, exists)
+				require.Equal(t, "2", v.AsString())
+			},
+		},
+		{
+			desc: "Required header skipped",
+			headers: http.Header{
+				textproto.CanonicalMIMEHeaderKey("X-Foo"):             []string{"1"},
+				textproto.CanonicalMIMEHeaderKey("X-Bar"):             []string{"2"},
+				textproto.CanonicalMIMEHeaderKey("X-Required-Header"): []string{"password"},
+			},
+			config: &Config{
+				Path:                       defaultPath,
+				HealthPath:                 defaultHealthPath,
+				ReadTimeout:                defaultReadTimeout,
+				WriteTimeout:               defaultWriteTimeout,
+				RequiredHeader:             RequiredHeader{Key: "X-Required-Header", Value: "password"},
+				ConvertHeadersToAttributes: true,
+			},
+			sc: func() *bufio.Scanner {
+				reader := io.NopCloser(bytes.NewReader([]byte("this is a: log")))
+				return bufio.NewScanner(reader)
+			}(),
+			tt: func(t *testing.T, reqLog plog.Logs, reqLen int, _ receiver.Settings) {
+				require.Equal(t, 1, reqLen)
+
+				attributes := reqLog.ResourceLogs().At(0).Resource().Attributes()
+				require.Equal(t, 0, attributes.Len())
+
+				scopeLogsScope := reqLog.ResourceLogs().At(0).ScopeLogs().At(0).Scope()
+				require.Equal(t, 4, scopeLogsScope.Attributes().Len()) // expect no additional attributes even though headers are set
+				_, exists := scopeLogsScope.Attributes().Get("header.x_foo")
+				require.True(t, exists)
+				_, exists = scopeLogsScope.Attributes().Get("header.x_bar")
+				require.True(t, exists)
+				_, exists = scopeLogsScope.Attributes().Get("header.x_required_header")
+				require.False(t, exists)
+			},
+		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.desc, func(t *testing.T) {
-			reqLog, reqLen := reqToLog(test.sc, test.query, defaultConfig, receivertest.NewNopSettings(metadata.Type))
+			testConfig := defaultConfig
+			if test.config != nil {
+				testConfig = test.config
+			}
+			reqLog, reqLen := reqToLog(test.sc, test.headers, test.query, testConfig, receivertest.NewNopSettings(metadata.Type))
 			test.tt(t, reqLog, reqLen, receivertest.NewNopSettings(metadata.Type))
 		})
 	}