Repository: redpanda-data/connect

Issue:
Elasticsearch Upsert
hi, what are your thoughts in adding option to [upsert](https://www.elastic.co/guide/en/elasticsearch/reference/7.4/docs-update.html#doc_as_upsert) instead of index messages to elasticsearch? We are currently dealing with events that is processed across several services, which in turn become separate kafka messages. We would like to combine them into one elasticsearch doc but can't define interval or parts count to wait for the completion of the event before indexing it to elasticsearch. If there's an upsert mechanism, we could just throw message at elasticsearch at let them eventually become consistent.

thanks

Hints from issue comments:
@Jeffail: Hey @rucciva, this sounds reasonable. We'd need to fork the current output to instead perform updates. I'm not 100% familiar with elasticsearch but I imagine the configuration options we'd want to expose wouldn't have much overlap with the current index way of doing things. Therefore it might be worth separating the config options so that users aren't overwhelmed.@mihaitodor: Should be fixed in [v4.58.0](https://github.com/redpanda-data/connect/releases/tag/v4.58.0) in `elasticsearch_v8` via #3481.

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b32e0868f..04e6285bd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+### Added
+
+- The service-wide `http` config now supports basic authentication.
+
 ## 4.3.0 - 2022-06-23
 
 ### Added
diff --git a/cmd/tools/benthos_docs_gen/main.go b/cmd/tools/benthos_docs_gen/main.go
index 1f72d2ff5..e6a7d080b 100644
--- a/cmd/tools/benthos_docs_gen/main.go
+++ b/cmd/tools/benthos_docs_gen/main.go
@@ -8,6 +8,7 @@ import (
 	"path"
 	"path/filepath"
 
+	"github.com/benthosdev/benthos/v4/internal/api"
 	"github.com/benthosdev/benthos/v4/internal/bundle"
 	tdocs "github.com/benthosdev/benthos/v4/internal/cli/test/docs"
 	"github.com/benthosdev/benthos/v4/internal/component/buffer"
@@ -74,6 +75,9 @@ func main() {
 	// Unit test docs
 	doTestDocs(docsDir)
 
+	// HTTP docs
+	doHTTP(docsDir)
+
 	// Logger docs
 	doLogger(docsDir)
 
@@ -169,6 +173,15 @@ func doTestDocs(dir string) {
 	create("test docs", filepath.Join(dir, "..", "configuration", "unit_testing.md"), mdSpec)
 }
 
+func doHTTP(dir string) {
+	mdSpec, err := api.DocsMarkdown()
+	if err != nil {
+		panic(fmt.Sprintf("Failed to generate docs for http: %v", err))
+	}
+
+	create("http docs", filepath.Join(dir, "http", "about.md"), mdSpec)
+}
+
 func doLogger(dir string) {
 	mdSpec, err := log.DocsMarkdown()
 	if err != nil {
diff --git a/internal/api/api.go b/internal/api/api.go
index ab44098fc..d8ee009f3 100644
--- a/internal/api/api.go
+++ b/internal/api/api.go
@@ -28,6 +28,7 @@ type Config struct {
 	CertFile       string              `json:"cert_file" yaml:"cert_file"`
 	KeyFile        string              `json:"key_file" yaml:"key_file"`
 	CORS           httpdocs.ServerCORS `json:"cors" yaml:"cors"`
+	BasicAuth      httpdocs.BasicAuth  `json:"basic_auth" yaml:"basic_auth"`
 }
 
 // NewConfig creates a new API config with default values.
@@ -40,6 +41,7 @@ func NewConfig() Config {
 		CertFile:       "",
 		KeyFile:        "",
 		CORS:           httpdocs.NewServerCORS(),
+		BasicAuth:      httpdocs.NewBasicAuth(),
 	}
 }
 
@@ -105,6 +107,10 @@ func New(
 		}
 	}
 
+	if err := conf.BasicAuth.Validate(); err != nil {
+		return nil, err
+	}
+
 	t := &Type{
 		conf:      conf,
 		endpoints: map[string]string{},
@@ -247,12 +253,13 @@ func (t *Type) RegisterEndpoint(path, desc string, handlerFunc http.HandlerFunc)
 	defer t.handlersMut.Unlock()
 
 	if _, exists := t.handlers[path]; !exists {
-		wrapHandler := func(w http.ResponseWriter, r *http.Request) {
+		wrapHandler := t.conf.BasicAuth.WrapHandler(func(w http.ResponseWriter, r *http.Request) {
 			t.handlersMut.RLock()
 			h := t.handlers[path]
 			t.handlersMut.RUnlock()
 			h(w, r)
-		}
+		})
+
 		t.mux.HandleFunc(path, wrapHandler)
 		t.mux.HandleFunc(t.conf.RootPath+path, wrapHandler)
 	}
diff --git a/internal/api/docs.go b/internal/api/docs.go
index a3f9c5b87..83c5d1c09 100644
--- a/internal/api/docs.go
+++ b/internal/api/docs.go
@@ -1,8 +1,13 @@
 package api
 
 import (
+	"bytes"
+	"text/template"
+
 	"github.com/benthosdev/benthos/v4/internal/docs"
 	httpdocs "github.com/benthosdev/benthos/v4/internal/http/docs"
+
+	_ "embed"
 )
 
 // Spec returns a field spec for the API configuration fields.
@@ -19,5 +24,52 @@ func Spec() docs.FieldSpecs {
 		docs.FieldString("cert_file", "An optional certificate file for enabling TLS.").Advanced().HasDefault(""),
 		docs.FieldString("key_file", "An optional key file for enabling TLS.").Advanced().HasDefault(""),
 		httpdocs.ServerCORSFieldSpec(),
+		httpdocs.BasicAuthFieldSpec(),
 	}
 }
+
+//go:embed docs.md
+var httpDocs string
+
+type templateContext struct {
+	Fields         []docs.FieldSpecCtx
+	CommonConfig   string
+	AdvancedConfig string
+}
+
+// DocsMarkdown returns a markdown document for the http documentation.
+func DocsMarkdown() ([]byte, error) {
+	httpDocsTemplate := docs.FieldsTemplate(false) + httpDocs
+
+	var buf bytes.Buffer
+	err := template.Must(template.New("http").Parse(httpDocsTemplate)).Execute(&buf, templateContext{
+		Fields: docs.FieldObject("", "").WithChildren(Spec()...).FlattenChildrenForDocs(),
+		CommonConfig: `
+http:
+  address: 0.0.0.0:4195
+  enabled: true
+  root_path: /benthos
+  debug_endpoints: false
+`,
+		AdvancedConfig: `
+http:
+  address: 0.0.0.0:4195
+  enabled: true
+  root_path: /benthos
+  debug_endpoints: false
+  cert_file: ""
+  key_file: ""
+  cors:
+    enabled: false
+    allowed_origins: []
+  basic_auth:
+    enabled: false
+    username: ""
+    password_hash: ""
+    algorithm: "sha256"
+    salt: ""
+`,
+	})
+
+	return buf.Bytes(), err
+}
diff --git a/internal/api/docs.md b/internal/api/docs.md
new file mode 100644
index 000000000..59b7b690c
--- /dev/null
+++ b/internal/api/docs.md
@@ -0,0 +1,105 @@
+---
+title: HTTP
+---
+
+<!--
+     THIS FILE IS AUTOGENERATED!
+
+     To make changes please edit the contents of:
+     internal/api/docs.md
+-->
+
+When Benthos runs it kicks off an HTTP server that provides a few generally useful endpoints and is also where configured components such as the [`http_server` input][inputs.http_server] [and output][outputs.http_server] can register their own endpoints if they don't require their own host/port.
+
+The configuration for this server lives under the `http` namespace, with the following default values:
+
+{{if eq .CommonConfig .AdvancedConfig -}}
+```yaml
+# Config fields, showing default values
+{{.CommonConfig -}}
+```
+{{else}}
+import Tabs from '@theme/Tabs';
+
+<Tabs defaultValue="common" values={{`{[
+  { label: 'Common', value: 'common', },
+  { label: 'Advanced', value: 'advanced', },
+]}`}}>
+
+import TabItem from '@theme/TabItem';
+
+<TabItem value="common">
+
+```yaml
+# Common config fields, showing default values
+{{.CommonConfig -}}
+```
+
+</TabItem>
+<TabItem value="advanced">
+
+```yaml
+# All config fields, showing default values
+{{.AdvancedConfig -}}
+```
+
+</TabItem>
+</Tabs>
+{{end -}}
+
+The field `enabled` can be set to `false` in order to disable the server.
+
+The field `root_path` specifies a general prefix for all endpoints, this can help isolate the service endpoints when using a reverse proxy with other shared services. All endpoints will still be registered at the root as well as behind the prefix, e.g. with a `root_path` set to `/foo` the endpoint `/version` will be accessible from both `/version` and `/foo/version`.
+
+## Enabling HTTPS
+
+By default Benthos will serve traffic over HTTP. In order to enforce TLS and serve traffic exclusively over HTTPS you must provide a `cert_file` and `key_file` path in your config, which point to a file containing a certificate and a matching private key for the server respectively.
+
+If the certificate is signed by a certificate authority, the `cert_file` should be the concatenation of the server's certificate, any intermediates, and the CA's certificate.
+
+## Enabling Basic Authentication
+
+By default Benthos does not do any sort of authentication for the service-wide HTTP server. However, it's possible to configure basic authentication with the [`basic_auth`](##basic_auth) field.
+
+## Endpoints
+
+The following endpoints will be generally available when the HTTP server is enabled:
+
+- `/version` provides version info.
+- `/ping` can be used as a liveness probe as it always returns a 200.
+- `/ready` can be used as a readiness probe as it serves a 200 only when both the input and output are connected, otherwise a 503 is returned.
+- `/metrics`, `/stats` both provide metrics when the metrics type is either [`json_api`][metrics.json_api] or [`prometheus`][metrics.prometheus].
+- `/endpoints` provides a JSON object containing a list of available endpoints, including those registered by configured components.
+
+## CORS
+
+In order to serve Cross-Origin Resource Sharing headers, which instruct browsers to allow CORS requests, set the subfield `cors.enabled` to `true`.
+
+### allowed_origins
+
+A list of allowed origins to connect from. The literal value `*` can be specified as a wildcard. Note `cors.enabled` must be set to `true` for this list to take effect.
+
+## Debug Endpoints
+
+The field `debug_endpoints` when set to `true` prompts Benthos to register a few extra endpoints that can be useful for debugging performance or behavioral problems:
+
+- `/debug/config/json` returns the loaded config as JSON.
+- `/debug/config/yaml` returns the loaded config as YAML.
+- `/debug/pprof/block` responds with a pprof-formatted block profile.
+- `/debug/pprof/heap` responds with a pprof-formatted heap profile.
+- `/debug/pprof/mutex` responds with a pprof-formatted mutex profile.
+- `/debug/pprof/profile` responds with a pprof-formatted cpu profile.
+- `/debug/pprof/symbol` looks up the program counters listed in the request, responding with a table mapping program counters to function names.
+- `/debug/pprof/trace` responds with the execution trace in binary form. Tracing lasts for duration specified in seconds GET parameter, or for 1 second if not specified.
+- `/debug/stack` returns a snapshot of the current service stack trace.
+
+## Fields
+
+The schema of the `http` section is as follows:
+
+{{template "field_docs" . -}}
+
+[inputs.http_server]: /docs/components/inputs/http_server
+[outputs.http_server]: /docs/components/outputs/http_server
+[metrics.json_api]: /docs/components/metrics/json_api
+[metrics.prometheus]: /docs/components/metrics/prometheus
diff --git a/internal/http/docs/basic_auth.go b/internal/http/docs/basic_auth.go
new file mode 100644
index 000000000..62cc74f2b
--- /dev/null
+++ b/internal/http/docs/basic_auth.go
@@ -0,0 +1,154 @@
+package docs
+
+import (
+	"crypto/md5"
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"net/http"
+
+	"golang.org/x/crypto/bcrypt"
+	"golang.org/x/crypto/scrypt"
+
+	"github.com/benthosdev/benthos/v4/internal/docs"
+)
+
+const (
+	scryptN      = 32768
+	scryptR      = 8
+	scryptP      = 1
+	scryptKeyLen = 32
+)
+
+// BasicAuth contains the configuration fields for the HTTP BasicAuth
+type BasicAuth struct {
+	Enabled      bool   `json:"enabled" yaml:"enabled"`
+	Username     string `json:"username" yaml:"username"`
+	PasswordHash string `json:"password_hash" yaml:"password_hash"`
+	Realm        string `json:"realm" yaml:"realm"`
+	Algorithm    string `json:"algorithm" yaml:"algorithm"`
+	Salt         string `json:"salt" yaml:"salt"`
+}
+
+// NewBasicAuth returns a BasicAuth with default values
+func NewBasicAuth() BasicAuth {
+	return BasicAuth{
+		Enabled:      false,
+		Username:     "",
+		PasswordHash: "",
+		Realm:        "restricted",
+		Algorithm:    "sha256",
+		Salt:         "",
+	}
+}
+
+// Validate confirms that the BasicAuth is properly configured
+func (b BasicAuth) Validate() error {
+	if !b.Enabled {
+		return nil
+	}
+
+	if b.Username == "" || b.PasswordHash == "" {
+		return errors.New("both username and password_hash are required")
+	}
+
+	if !(b.Algorithm == "md5" || b.Algorithm == "sha256" || b.Algorithm == "bcrypt" || b.Algorithm == "scrypt") {
+		return errors.New("algorithm should be one of md5, sha256, bcrypt, or scrypt")
+	}
+
+	if b.Algorithm == "scrypt" && b.Salt == "" {
+		return errors.New("salt is required for scrypt")
+	}
+
+	if b.Algorithm == "scrypt" {
+		if _, err := base64.StdEncoding.DecodeString(b.Salt); err != nil {
+			return fmt.Errorf("invalid salt : %w", err)
+		}
+	}
+
+	return nil
+}
+
+// WrapHandler wraps the provided HTTP handler with middleware that enforces
+// BasicAuth if it's enabled.
+func (b BasicAuth) WrapHandler(next http.HandlerFunc) http.HandlerFunc {
+	if !b.Enabled {
+		return next
+	}
+
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		user, pass, ok := r.BasicAuth()
+		if !ok {
+			user = ""
+			pass = ""
+		}
+
+		if ok, err := b.matches(user, pass); !ok || err != nil {
+			if err != nil {
+				http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+				return
+			}
+
+			w.Header().Set("WWW-Authenticate", fmt.Sprintf(`Basic realm=%q, charset="UTF-8"`, b.Realm))
+			http.Error(w, "Unauthorized", http.StatusUnauthorized)
+			return
+		}
+
+		next.ServeHTTP(w, r)
+	})
+}
+
+// BasicAuthFieldSpec returns the spec for an HTTP BasicAuth component
+func BasicAuthFieldSpec() docs.FieldSpec {
+	return docs.FieldObject("basic_auth", "Allows you to enforce and customise basic authentication for requests to the HTTP server.").WithChildren(
+		docs.FieldBool("enabled", "Enable basic authentication").HasDefault(false),
+		docs.FieldString("realm", "Custom realm name").HasDefault("restricted"),
+		docs.FieldString("username", "Username required to authenticate.").HasDefault(""),
+		docs.FieldString("password_hash", "Hashed password required to authenticate. (base64 encoded)").HasDefault(""),
+		docs.FieldString("algorithm", "Encryption algorithm used to generate `password_hash`.", "md5", "sha256", "bcrypt", "scrypt").HasDefault("sha256"),
+		docs.FieldString("salt", "Salt for scrypt algorithm. (base64 encoded)").HasDefault(""),
+	).Advanced()
+}
+
+func (b BasicAuth) matches(user, pass string) (bool, error) {
+	expectedPassHash, err := base64.StdEncoding.DecodeString(b.PasswordHash)
+	if err != nil {
+		return false, err
+	}
+
+	userMatch := (subtle.ConstantTimeCompare([]byte(user), []byte(b.Username)) == 1)
+	passMatch := b.compareHashAndPassword(expectedPassHash, []byte(pass))
+
+	return (userMatch && passMatch), nil
+}
+
+func (b BasicAuth) compareHashAndPassword(hashedPassword, password []byte) bool {
+	switch b.Algorithm {
+	case "md5":
+		v := md5.Sum(password)
+		return (subtle.ConstantTimeCompare(hashedPassword, v[:]) == 1)
+	case "sha256":
+		v := sha256.Sum256(password)
+		return (subtle.ConstantTimeCompare(hashedPassword, v[:]) == 1)
+	case "bcrypt":
+		if err := bcrypt.CompareHashAndPassword(hashedPassword, password); err != nil {
+			return false
+		}
+		return true
+	case "scrypt":
+		salt, err := base64.StdEncoding.DecodeString(b.Salt)
+		if err != nil {
+			return false
+		}
+
+		v, err := scrypt.Key(password, salt, scryptN, scryptR, scryptP, scryptKeyLen)
+		if err != nil {
+			return false
+		}
+		return (subtle.ConstantTimeCompare(hashedPassword, v) == 1)
+	default:
+		return false
+	}
+}
diff --git a/internal/impl/elasticsearch/output.go b/internal/impl/elasticsearch/output.go
index c76266f1e..a9e9ee400 100644
--- a/internal/impl/elasticsearch/output.go
+++ b/internal/impl/elasticsearch/output.go
@@ -58,7 +58,7 @@ false for connections to succeed.`),
 		Config: docs.FieldComponent().WithChildren(
 			docs.FieldString("urls", "A list of URLs to connect to. If an item of the list contains commas it will be expanded into multiple URLs.", []string{"http://localhost:9200"}).Array(),
 			docs.FieldString("index", "The index to place messages.").IsInterpolated(),
-			docs.FieldString("action", "The action to take on the document.").IsInterpolated().HasOptions("create", "index", "update", "delete").Advanced(),
+			docs.FieldString("action", "The action to take on the document.").IsInterpolated().HasOptions("create", "index", "update", "upsert", "delete").Advanced(),
 			docs.FieldString("pipeline", "An optional pipeline id to preprocess incoming documents.").IsInterpolated().Advanced(),
 			docs.FieldString("id", "The ID for indexed messages. Interpolation should be used in order to create a unique ID for each message.").IsInterpolated(),
 			docs.FieldString("type", "The document mapping type. This field is required for versions of elasticsearch earlier than 6.0.0, but are invalid for versions 7.0.0 or later.").Optional().IsInterpolated(),
@@ -378,6 +378,17 @@ func (e *Elasticsearch) buildBulkableRequest(p *pendingBulkIndex) (elastic.Bulka
 			r = r.Type(p.Type)
 		}
 		return r, nil
+	case "upsert":
+		r := elastic.NewBulkUpdateRequest().
+			Index(p.Index).
+			Routing(p.Routing).
+			Id(p.ID).
+			DocAsUpsert(true).
+			Doc(p.Doc)
+		if p.Type != "" {
+			r = r.Type(p.Type)
+		}
+		return r, nil
 	case "delete":
 		r := elastic.NewBulkDeleteRequest().
 			Index(p.Index).
diff --git a/internal/impl/kafka/input_kafka_franz.go b/internal/impl/kafka/input_kafka_franz.go
index f569dbaa8..eb7a32843 100644
--- a/internal/impl/kafka/input_kafka_franz.go
+++ b/internal/impl/kafka/input_kafka_franz.go
@@ -73,7 +73,6 @@ This input adds the following metadata fields to each message:
 			Advanced()).
 		Field(service.NewTLSToggledField("tls")).
 		Field(saslField())
-
 }
 
 func init() {
diff --git a/internal/impl/kafka/logger.go b/internal/impl/kafka/logger.go
index 5bbb9b81d..3098d746b 100644
--- a/internal/impl/kafka/logger.go
+++ b/internal/impl/kafka/logger.go
@@ -28,6 +28,6 @@ func (k *kgoLogger) Log(level kgo.LogLevel, msg string, keyvals ...interface{})
 	case kgo.LogLevelInfo:
 		tmpL.Info(msg)
 	case kgo.LogLevelDebug:
-		tmpL.Debug(msg)
+		tmpL.Trace(msg)
 	}
 }
diff --git a/internal/log/docs.go b/internal/log/docs.go
index 6f72d16a5..0f69a4247 100644
--- a/internal/log/docs.go
+++ b/internal/log/docs.go
@@ -2,7 +2,7 @@ package log
 
 import (
 	"bytes"
-	"html/template"
+	"text/template"
 
 	"github.com/benthosdev/benthos/v4/internal/docs"
 
diff --git a/public/service/logger.go b/public/service/logger.go
index 437fdc5b2..b4b0a1f88 100644
--- a/public/service/logger.go
+++ b/public/service/logger.go
@@ -17,6 +17,22 @@ func newReverseAirGapLogger(l log.Modular) *Logger {
 	return &Logger{l}
 }
 
+// Tracef logs a trace message using fmt.Sprintf when args are specified.
+func (l *Logger) Tracef(template string, args ...interface{}) {
+	if l == nil {
+		return
+	}
+	l.m.Tracef(template, args...)
+}
+
+// Trace logs a trace message.
+func (l *Logger) Trace(message string) {
+	if l == nil {
+		return
+	}
+	l.m.Traceln(message)
+}
+
 // Debugf logs a debug message using fmt.Sprintf when args are specified.
 func (l *Logger) Debugf(template string, args ...interface{}) {
 	if l == nil {
diff --git a/website/docs/components/http/about.md b/website/docs/components/http/about.md
index a6edf13fe..cac96421a 100644
--- a/website/docs/components/http/about.md
+++ b/website/docs/components/http/about.md
@@ -2,11 +2,45 @@
 title: HTTP
 ---
 
+<!--
+     THIS FILE IS AUTOGENERATED!
+
+     To make changes please edit the contents of:
+     internal/api/docs.md
+-->
+
 When Benthos runs it kicks off an HTTP server that provides a few generally useful endpoints and is also where configured components such as the [`http_server` input][inputs.http_server] [and output][outputs.http_server] can register their own endpoints if they don't require their own host/port.
 
 The configuration for this server lives under the `http` namespace, with the following default values:
 
+
+import Tabs from '@theme/Tabs';
+
+<Tabs defaultValue="common" values={[
+  { label: 'Common', value: 'common', },
+  { label: 'Advanced', value: 'advanced', },
+]}>
+
+import TabItem from '@theme/TabItem';
+
+<TabItem value="common">
+
+```yaml
+# Common config fields, showing default values
+
+http:
+  address: 0.0.0.0:4195
+  enabled: true
+  root_path: /benthos
+  debug_endpoints: false
+```
+
+</TabItem>
+<TabItem value="advanced">
+
 ```yaml
+# All config fields, showing default values
+
 http:
   address: 0.0.0.0:4195
   enabled: true
@@ -17,8 +51,16 @@ http:
   cors:
     enabled: false
     allowed_origins: []
+  basic_auth:
+    enabled: false
+    username: ""
+    password_hash: ""
+    algorithm: "sha256"
+    salt: ""
 ```
 
+</TabItem>
+</Tabs>
 The field `enabled` can be set to `false` in order to disable the server.
 
 The field `root_path` specifies a general prefix for all endpoints, this can help isolate the service endpoints when using a reverse proxy with other shared services. All endpoints will still be registered at the root as well as behind the prefix, e.g. with a `root_path` set to `/foo` the endpoint `/version` will be accessible from both `/version` and `/foo/version`.
@@ -29,6 +71,10 @@ By default Benthos will serve traffic over HTTP. In order to enforce TLS and ser
 
 If the certificate is signed by a certificate authority, the `cert_file` should be the concatenation of the server's certificate, any intermediates, and the CA's certificate.
 
+## Enabling Basic Authentication
+
+By default Benthos does not do any sort of authentication for the service-wide HTTP server. However, it's possible to configure basic authentication with the [`basic_auth`](##basic_auth) field.
+
 ## Endpoints
 
 The following endpoints will be generally available when the HTTP server is enabled:
@@ -61,6 +107,149 @@ The field `debug_endpoints` when set to `true` prompts Benthos to register a few
 - `/debug/pprof/trace` responds with the execution trace in binary form. Tracing lasts for duration specified in seconds GET parameter, or for 1 second if not specified.
 - `/debug/stack` returns a snapshot of the current service stack trace.
 
+## Fields
+
+The schema of the `http` section is as follows:
+
+### `enabled`
+
+Whether to enable to HTTP server.
+
+
+Type: `bool`  
+Default: `true`  
+
+### `address`
+
+The address to bind to.
+
+
+Type: `string`  
+Default: `"0.0.0.0:4195"`  
+
+### `root_path`
+
+Specifies a general prefix for all endpoints, this can help isolate the service endpoints when using a reverse proxy with other shared services. All endpoints will still be registered at the root as well as behind the prefix, e.g. with a root_path set to `/foo` the endpoint `/version` will be accessible from both `/version` and `/foo/version`.
+
+
+Type: `string`  
+Default: `"/benthos"`  
+
+### `debug_endpoints`
+
+Whether to register a few extra endpoints that can be useful for debugging performance or behavioral problems.
+
+
+Type: `bool`  
+Default: `false`  
+
+### `cert_file`
+
+An optional certificate file for enabling TLS.
+
+
+Type: `string`  
+Default: `""`  
+
+### `key_file`
+
+An optional key file for enabling TLS.
+
+
+Type: `string`  
+Default: `""`  
+
+### `cors`
+
+Adds Cross-Origin Resource Sharing headers.
+
+
+Type: `object`  
+Requires version 3.63.0 or newer  
+
+### `cors.enabled`
+
+Whether to allow CORS requests.
+
+
+Type: `bool`  
+Default: `false`  
+
+### `cors.allowed_origins`
+
+An explicit list of origins that are allowed for CORS requests.
+
+
+Type: list of `string`  
+Default: `[]`  
+
+### `basic_auth`
+
+Allows you to enforce and customise basic authentication for requests to the HTTP server.
+
+
+Type: `object`  
+
+### `basic_auth.enabled`
+
+Enable basic authentication
+
+
+Type: `bool`  
+Default: `false`  
+
+### `basic_auth.realm`
+
+Custom realm name
+
+
+Type: `string`  
+Default: `"restricted"`  
+
+### `basic_auth.username`
+
+Username required to authenticate.
+
+
+Type: `string`  
+Default: `""`  
+
+### `basic_auth.password_hash`
+
+Hashed password required to authenticate. (base64 encoded)
+
+
+Type: `string`  
+Default: `""`  
+
+### `basic_auth.algorithm`
+
+Encryption algorithm used to generate `password_hash`.
+
+
+Type: `string`  
+Default: `"sha256"`  
+
+```yml
+# Examples
+
+algorithm: md5
+
+algorithm: sha256
+
+algorithm: bcrypt
+
+algorithm: scrypt
+```
+
+### `basic_auth.salt`
+
+Salt for scrypt algorithm. (base64 encoded)
+
+
+Type: `string`  
+Default: `""`  
+
 [inputs.http_server]: /docs/components/inputs/http_server
 [outputs.http_server]: /docs/components/outputs/http_server
 [metrics.json_api]: /docs/components/metrics/json_api
diff --git a/website/docs/components/logger/about.md b/website/docs/components/logger/about.md
index d2d6d2f2a..76428275e 100644
--- a/website/docs/components/logger/about.md
+++ b/website/docs/components/logger/about.md
@@ -55,7 +55,7 @@ Set the minimum severity level for emitting logs.
 
 
 Type: `string`  
-Default: `&#34;INFO&#34;`  
+Default: `"INFO"`  
 Options: `OFF`, `FATAL`, `ERROR`, `WARN`, `INFO`, `DEBUG`, `TRACE`, `ALL`, `NONE`.
 
 ### `format`
@@ -64,7 +64,7 @@ Set the format of emitted logs.
 
 
 Type: `string`  
-Default: `&#34;logfmt&#34;`  
+Default: `"logfmt"`  
 Options: `json`, `logfmt`.
 
 ### `add_timestamp`
@@ -81,7 +81,7 @@ A map of key/value pairs to add to each structured log.
 
 
 Type: map of `string`  
-Default: `{&#34;@service&#34;:&#34;benthos&#34;}`  
+Default: `{"@service":"benthos"}`  
 
 ### `file`
 
@@ -96,7 +96,7 @@ The file path to write logs to, if the file does not exist it will be created. L
 
 
 Type: `string`  
-Default: `&#34;&#34;`  
+Default: `""`  
 
 ### `file.rotate`
 
diff --git a/website/docs/components/outputs/elasticsearch.md b/website/docs/components/outputs/elasticsearch.md
index dc424d872..e07e6974a 100644
--- a/website/docs/components/outputs/elasticsearch.md
+++ b/website/docs/components/outputs/elasticsearch.md
@@ -157,7 +157,7 @@ This field supports [interpolation functions](/docs/configuration/interpolation#
 
 Type: `string`  
 Default: `"index"`  
-Options: `create`, `index`, `update`, `delete`.
+Options: `create`, `index`, `update`, `upsert`, `delete`.
 
 ### `pipeline`
 
diff --git a/website/src/pages/support.js b/website/src/pages/support.js
index 3cd935317..6a0f736e2 100644
--- a/website/src/pages/support.js
+++ b/website/src/pages/support.js
@@ -48,7 +48,7 @@ function Support() {
             <div className="row margin-bottom--lg">
               <div className="col col--9 col--offset-1">
                 <p>
-Benthos has a helpful and conveniently global community so if you have quick questions, are in the market for ideas, or just want to make some friends, it's worth trying your luck with our <a href="#community-support">community spaces</a>. However, for organisations that want quicker action, guaranteed attention or some one-on-one consulting time there are some <a href="#paid-support">paid services available</a> that are worth considering.
+Benthos has a helpful and conveniently global community so if you have quick questions, are in the market for ideas, or just want to make some friends, it's worth trying your luck with our <a href="#community-support">community spaces</a>. However, for organisations that want quicker action, guaranteed attention or some one-on-one consulting time there are some <a href="#paid-services">paid services available</a> that are worth considering.
                 </p>
               </div>
             </div>
@@ -103,6 +103,11 @@ Benthos has a helpful and conveniently global community so if you have quick que
                 <p>This especially applies to configuration files you're showing us, please try and reduce them down into specifically the components you're asking about.</p>
                 <h3>Reproduce Your Issue</h3>
                 <p>It's common for users to point out an issue they're seemingly having with Benthos and it then turns out to be an unrelated issue with cached builds, configs not being saved, docker images not being pulled fresh, etc. These encounters may sound rare but anecdotally they occur more often in our support channels than actual bugs, and we've spent considerable measures of time trying to reproduce issues that don't exist (yet). By attempting to isolate and reproduce issues yourself you can significantly reduce the burden on us.</p>
+                <h3>Provide Context</h3>
+                <p>Things that a typical community member cannot do include mind-reading, assuming control of your PC (I hope) and time travelling into the past in order to obtain context of the problem on your behalf. In order to work around these limitations please make sure that when you are asking us for help regarding a problem you've had that at the very least you are able to give us any error messages/logs that were emitted by Benthos when it happened.</p>
+                <h3>Never Direct Message</h3>
+                <p>Unless you are explicitly instructed to do so please never direct message maintainers or community members for support. If you aren't receiving responses in a public support channel it is because they currently do not have time to address your issue, please be patient or consider a <a href="#paid-services">paid support option instead</a>. Direct messages only serve to add extra pressure on volunteers, and answering questions via direct messages denies other users the opportunity to read the answer themselves.</p>
+                <p>If there is sensitive information within your question that you do not want to expose publicly then please take the time to scrub that information from the material you are sharing.</p>
               </div>
             </div>
           </div>



Test patch:
diff --git a/internal/api/api_test.go b/internal/api/api_test.go
index 27226ede7..dec46225c 100644
--- a/internal/api/api_test.go
+++ b/internal/api/api_test.go
@@ -85,3 +85,85 @@ func TestAPIEnableCORSNoHeaders(t *testing.T) {
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "must specify at least one allowed origin")
 }
+
+func TestAPIBasicAuth(t *testing.T) {
+	secret256 := "K7gNU3sdo+OL0wNhqoVWhr3g6s1xYv72ol/pe/Unols="
+	secretMD5 := "Xr4ilOzQ4PCOq3aQ0qbuaQ=="
+	secretBcrypt := "JDJhJDEwJHRDMWs2NHc5VjA0R3JIdEE2QzQ2SC5na3o3WWNUUWlQc1RNbVNCUU8yOG8uSWg4YUUvaUcu"
+	secretScrypt := "eJ01V0+YCJxHraokLp79ijKS0z1jzVQ3vFtGx5UEEiY="
+	type testCase struct {
+		name         string
+		enabled      bool
+		algorithm    string
+		correctUser  string
+		correctPass  string
+		givenUser    string
+		givenPass    string
+		expectedErr  assert.ErrorAssertionFunc
+		expectedCode int
+	}
+
+	tests := []testCase{
+		{name: "validAuth", enabled: true, algorithm: "sha256",
+			correctUser: "myuser", correctPass: secret256,
+			givenUser: "myuser", givenPass: "secret",
+			expectedErr: assert.NoError, expectedCode: http.StatusOK},
+		{name: "validAuthMD5", enabled: true, algorithm: "md5",
+			correctUser: "myuser", correctPass: secretMD5,
+			givenUser: "myuser", givenPass: "secret",
+			expectedErr: assert.NoError, expectedCode: http.StatusOK},
+		{name: "validAuthBcrypt", enabled: true, algorithm: "bcrypt",
+			correctUser: "myuser", correctPass: secretBcrypt,
+			givenUser: "myuser", givenPass: "secret",
+			expectedErr: assert.NoError, expectedCode: http.StatusOK},
+		{name: "validAuthScrypt", enabled: true, algorithm: "scrypt",
+			correctUser: "myuser", correctPass: secretScrypt,
+			givenUser: "myuser", givenPass: "secret",
+			expectedErr: assert.NoError, expectedCode: http.StatusOK},
+		{name: "invalidAuth", enabled: true, algorithm: "sha256",
+			correctUser: "myuser", correctPass: secret256,
+			givenUser: "myuser", givenPass: "wrong",
+			expectedErr: assert.NoError, expectedCode: http.StatusUnauthorized},
+		{name: "noAuthGiven", enabled: true, algorithm: "sha256",
+			correctUser: "myuser", correctPass: secret256,
+			givenUser: "", givenPass: "", expectedErr: assert.NoError,
+			expectedCode: http.StatusUnauthorized},
+		{name: "disabledAuthWrong", enabled: false, algorithm: "sha256",
+			correctUser: "myuser", correctPass: secret256,
+			givenUser: "myuser", givenPass: "wrong",
+			expectedErr: assert.NoError, expectedCode: http.StatusOK},
+		{name: "disabledAuthNoneGiven", enabled: false, algorithm: "sha256",
+			correctUser: "myuser", correctPass: secret256,
+			givenUser: "", givenPass: "",
+			expectedErr: assert.NoError, expectedCode: http.StatusOK},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(tc testCase) func(t *testing.T) {
+			return func(t *testing.T) {
+				conf := api.NewConfig()
+				conf.BasicAuth.Enabled = tc.enabled
+				conf.BasicAuth.Algorithm = tc.algorithm
+				conf.BasicAuth.Username = tc.correctUser
+				conf.BasicAuth.PasswordHash = tc.correctPass
+				conf.BasicAuth.Salt = "EzrwNJYw2wkErVVV1P36FQ=="
+
+				s, err := api.New("", "", conf, nil, log.Noop(), metrics.Noop())
+				if ok := tc.expectedErr(t, err); !(ok && err == nil) {
+					return
+				}
+
+				handler := s.Handler()
+
+				request, _ := http.NewRequest("GET", "/version", http.NoBody)
+				if tc.givenUser != "" || tc.givenPass != "" {
+					request.SetBasicAuth(tc.givenUser, tc.givenPass)
+				}
+				response := httptest.NewRecorder()
+				handler.ServeHTTP(response, request)
+
+				assert.Equal(t, tc.expectedCode, response.Code)
+			}
+		}(tc))
+	}
+}