add headers to http processor (#1552)

* add headers to http processor * Update pkg/plugin/processor/builtin/impl/webhook/http.go Co-authored-by: Lovro Mažgon <[email protected]> * Update pkg/plugin/processor/builtin/impl/webhook/http.go Co-authored-by: Lovro Mažgon <[email protected]> * add tests, handle lower case header name, re-generate * fix tests and code * Update pkg/plugin/processor/builtin/impl/webhook/http.go Co-authored-by: Lovro Mažgon <[email protected]> --------- Co-authored-by: Haris Osmanagić <[email protected]> Co-authored-by: Lovro Mažgon <[email protected]>
ConduitIO · May 14, 2024 · 734fa82 · 734fa82
1 parent 57b29be
commit 734fa82
Show file tree

Hide file tree

Showing 4 changed files with 150 additions and 5 deletions.
diff --git a/pkg/plugin/processor/builtin/impl/webhook/http.go b/pkg/plugin/processor/builtin/impl/webhook/http.go
@@ -44,8 +44,10 @@ type httpConfig struct {
 	URL string `json:"request.url" validate:"required"`
 	// Method is the HTTP request method to be used.
 	Method string `json:"request.method" default:"GET"`
-	// The value of the `Content-Type` header.
+	// Deprecated: use `headers.Content-Type` instead.
 	ContentType string `json:"request.contentType" default:"application/json"`
+	// Headers to add to the request, use `headers.*` to specify the header and its value (e.g. `headers.Authorization: "Bearer key"`).
+	Headers map[string]string `json:"headers"`
 
 	// Maximum number of retries for an individual record when backing off following an error.
 	BackoffRetryCount float64 `json:"backoffRetry.count" default:"0" validate:"gt=-1"`
@@ -72,6 +74,29 @@ type httpConfig struct {
 	ResponseStatusRef string `json:"response.status"`
 }
 
+func (c *httpConfig) parseHeaders() error {
+	if c.Headers == nil {
+		c.Headers = make(map[string]string)
+	}
+
+	if c.ContentType == "" {
+		return nil // Nothing to replace in headers
+	}
+
+	for name, _ := range c.Headers {
+		if strings.ToLower(name) == "content-type" {
+			return cerrors.Errorf("Configuration error, cannot provide both \"request.contentType\" and \"headers.Content-Type\", use \"headers.Content-Type\" only.")
+		}
+	}
+
+	c.Headers["Content-Type"] = c.ContentType
+	// the ContentType field is deprecated,
+	// so we're preparing for completely removing it in a later release
+	c.ContentType = ""
+
+	return nil
+}
+
 type httpProcessor struct {
 	sdk.UnimplementedProcessor
 
@@ -108,6 +133,11 @@ func (p *httpProcessor) Configure(ctx context.Context, m map[string]string) erro
 		return cerrors.Errorf("failed parsing configuration: %w", err)
 	}
 
+	err = p.config.parseHeaders()
+	if err != nil {
+		return err
+	}
+
 	if p.config.ResponseBodyRef == p.config.ResponseStatusRef {
 		return cerrors.New("invalid configuration: response.body and response.status set to same field")
 	}
@@ -293,8 +323,10 @@ func (p *httpProcessor) buildRequest(ctx context.Context, r opencdc.Record) (*ht
 		return nil, cerrors.Errorf("error creating HTTP request: %w", err)
 	}
 
-	// todo make it possible to add more headers, e.g. auth headers etc.
-	req.Header.Set("Content-Type", p.config.ContentType)
+	// set header values
+	for key, val := range p.config.Headers {
+		req.Header.Set(key, val)
+	}
 
 	return req, nil
 }

diff --git a/pkg/plugin/processor/builtin/impl/webhook/http_config_test.go b/pkg/plugin/processor/builtin/impl/webhook/http_config_test.go
@@ -0,0 +1,101 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// 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 webhook
+
+import (
+	"github.com/matryer/is"
+	"testing"
+)
+
+func TestHTTPConfig_ValidateHeaders(t *testing.T) {
+	testCases := []struct {
+		name       string
+		input      httpConfig
+		wantConfig httpConfig
+		wantErr    string
+	}{
+		{
+			name: "ContentType field present, header present",
+			input: httpConfig{
+				ContentType: "application/json",
+				Headers: map[string]string{
+					"Content-Type": "application/json",
+				},
+			},
+			wantErr: `Configuration error, cannot provide both "request.contentType" and "headers.Content-Type", use "headers.Content-Type" only.`,
+		},
+		{
+			name: "ContentType field present, header present, different case",
+			input: httpConfig{
+				ContentType: "application/json",
+				Headers: map[string]string{
+					"content-type": "application/json",
+				},
+			},
+			wantErr: `Configuration error, cannot provide both "request.contentType" and "headers.Content-Type", use "headers.Content-Type" only.`,
+		},
+		{
+			name: "ContentType field presents, header not present",
+			input: httpConfig{
+				ContentType: "application/json",
+			},
+			wantConfig: httpConfig{
+				Headers: map[string]string{
+					"Content-Type": "application/json",
+				},
+			},
+		},
+		{
+			name: "ContentType field not present, header present",
+			input: httpConfig{
+				Headers: map[string]string{
+					"Content-Type": "application/json",
+				},
+			},
+			wantConfig: httpConfig{
+				Headers: map[string]string{
+					"Content-Type": "application/json",
+				},
+			},
+		},
+		{
+			name: "ContentType field not present, header present, different case",
+			input: httpConfig{
+				Headers: map[string]string{
+					"content-type": "application/json",
+				},
+			},
+			wantConfig: httpConfig{
+				Headers: map[string]string{
+					"content-type": "application/json",
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			is := is.New(t)
+			err := tc.input.parseHeaders()
+
+			if tc.wantErr == "" {
+				is.Equal(tc.wantConfig, tc.input)
+			} else {
+				is.True(err != nil)
+				is.Equal(tc.wantErr, err.Error())
+			}
+		})
+	}
+}
diff --git a/pkg/plugin/processor/builtin/impl/webhook/http_paramgen.go b/pkg/plugin/processor/builtin/impl/webhook/http_paramgen.go
diff --git a/pkg/plugin/processor/builtin/internal/exampleutil/specs/webhook.http.json b/pkg/plugin/processor/builtin/internal/exampleutil/specs/webhook.http.json
@@ -40,6 +40,12 @@
         "type": "duration",
         "validations": []
       },
+      "headers.*": {
+        "default": "",
+        "description": "Headers to add to the request, use `headers.*` to specify the header and its value (e.g. `headers.Authorization: \"Bearer key\"`).",
+        "type": "string",
+        "validations": []
+      },
       "request.body": {
         "default": "",
         "description": "Specifies which field from the input record should be used as the body in\nthe HTTP request.\n\nFor more information about the format, see [Referencing fields](https://conduit.io/docs/processors/referencing-fields).",
@@ -48,7 +54,7 @@
       },
       "request.contentType": {
         "default": "application/json",
-        "description": "The value of the `Content-Type` header.",
+        "description": "Deprecated: use `headers.Content-Type` instead.",
         "type": "string",
         "validations": []
       },