Repository: redpanda-data/connect

Issue:
Create a multiline Codec 
I would like to read a file and emit messages which are according to a codec which is based on a regular expression

an example of such a codec code be "regex:^%{TIMESTAMP_ISO8601}"

The goal is to parse logfiles.


Discussion:
@Jeffail: The motivation for this is here: https://www.elastic.co/guide/en/logstash/current/plugins-codecs-multiline.html, where we want to express a pattern by which we would assume a line is a continuation of the previous if it matches, otherwise we flush all buffered data and begins a new one.@kong62: Any progress?@rockwotj: You can use a Scanner for that. Either the newline or regex ones work: https://docs.redpanda.com/redpanda-connect/components/scanners/re_match/@kong62: > You can use a Scanner for that. Either the newline or regex ones work: https://docs.redpanda.com/redpanda-connect/components/scanners/re_match/

work good. thank you

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 170a89f65..b5e3b0bdf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,8 @@ All notable changes to this project will be documented in this file.
 - Field `cors` added to the `http_server` input and output, for supporting CORS requests when custom servers are used.
 - Field `server_side_encryption` added to the `aws_s3` output.
 - Field `use_histogram_timing` added to the `prometheus` metrics exporter.
+- New duration string and back off field types added to plugin config builders.
+- Experimental field `multipart` added to the `http_client` output.
 
 ### Changed
 
diff --git a/internal/codec/reader.go b/internal/codec/reader.go
index b3b257bc0..0c377b073 100644
--- a/internal/codec/reader.go
+++ b/internal/codec/reader.go
@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 	"sync"
@@ -33,6 +34,7 @@ var ReaderDocs = docs.FieldCommon(
 	"gzip", "Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc.",
 	"lines", "Consume the file in segments divided by linebreaks.",
 	"multipart", "Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch.",
+	"regex:(?m)^\\d\\d:\\d\\d:\\d\\d", "Consume the file in segments divided by regular expression.",
 	"tar", "Parse the file as a tar archive, and consume each file of the archive as a message.",
 )
 
@@ -230,11 +232,11 @@ func partReader(codec string, conf ReaderConfig) (ReaderConstructor, bool, error
 	if strings.HasPrefix(codec, "csv:") {
 		by := strings.TrimPrefix(codec, "csv:")
 		if by == "" {
-			return nil, false, errors.New("custom delimiter codec requires a non-empty delimiter")
+			return nil, false, errors.New("csv codec requires a non-empty delimiter")
 		}
 		byRunes := []rune(by)
 		if len(byRunes) != 1 {
-			return nil, false, errors.New("custom delimiter codec requires a single character delimiter")
+			return nil, false, errors.New("csv codec requires a single character delimiter")
 		}
 		byRune := byRunes[0]
 		return func(path string, r io.ReadCloser, fn ReaderAckFn) (Reader, error) {
@@ -250,6 +252,15 @@ func partReader(codec string, conf ReaderConfig) (ReaderConstructor, bool, error
 			return newChunkerReader(conf, r, chunkSize, fn)
 		}, true, nil
 	}
+	if strings.HasPrefix(codec, "regex:") {
+		by := strings.TrimPrefix(codec, "regex:")
+		if by == "" {
+			return nil, false, errors.New("regex codec requires a non-empty delimiter")
+		}
+		return func(path string, r io.ReadCloser, fn ReaderAckFn) (Reader, error) {
+			return newRexExpSplitReader(conf, r, by, fn)
+		}, true, nil
+	}
 	return nil, false, nil
 }
 
@@ -803,3 +814,113 @@ func (m *multipartReader) Next(ctx context.Context) ([]types.Part, ReaderAckFn,
 func (m *multipartReader) Close(ctx context.Context) error {
 	return m.child.Close(ctx)
 }
+
+//------------------------------------------------------------------------------
+
+type regexReader struct {
+	buf       *bufio.Scanner
+	r         io.ReadCloser
+	sourceAck ReaderAckFn
+
+	mut      sync.Mutex
+	finished bool
+	pending  int32
+}
+
+func newRexExpSplitReader(conf ReaderConfig, r io.ReadCloser, regex string, ackFn ReaderAckFn) (Reader, error) {
+	scanner := bufio.NewScanner(r)
+	if conf.MaxScanTokenSize != bufio.MaxScanTokenSize {
+		scanner.Buffer([]byte{}, conf.MaxScanTokenSize)
+	}
+
+	compiled, err := regexp.Compile(regex)
+
+	if err != nil {
+		return nil, err
+	}
+
+	scanner.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+		if atEOF && len(data) == 0 {
+			return 0, nil, nil
+		}
+
+		loc := compiled.FindAllIndex(data, 2)
+		if loc == nil {
+			if atEOF {
+				return len(data), data, nil
+			}
+			return 0, nil, nil
+		}
+
+		if len(loc) == 1 {
+			if atEOF {
+				if loc[0][0] == 0 {
+					return len(data), data, nil
+				}
+				return loc[0][0], data[0:loc[0][0]], nil
+			}
+			return 0, nil, nil
+		}
+		if loc[0][0] == 0 {
+			return loc[1][0], data[0:loc[1][0]], nil
+		}
+		return loc[0][0], data[0:loc[0][0]], nil
+	})
+
+	return &regexReader{
+		buf:       scanner,
+		r:         r,
+		sourceAck: ackOnce(ackFn),
+	}, nil
+}
+
+func (a *regexReader) ack(ctx context.Context, err error) error {
+	a.mut.Lock()
+	a.pending--
+	doAck := a.pending == 0 && a.finished
+	a.mut.Unlock()
+
+	if err != nil {
+		return a.sourceAck(ctx, err)
+	}
+	if doAck {
+		return a.sourceAck(ctx, nil)
+	}
+	return nil
+}
+
+func (a *regexReader) Next(ctx context.Context) ([]types.Part, ReaderAckFn, error) {
+	scanned := a.buf.Scan()
+
+	a.mut.Lock()
+	defer a.mut.Unlock()
+
+	if scanned {
+		a.pending++
+
+		bytesCopy := make([]byte, len(a.buf.Bytes()))
+		copy(bytesCopy, a.buf.Bytes())
+		return []types.Part{message.NewPart(bytesCopy)}, a.ack, nil
+	}
+	err := a.buf.Err()
+	if err == nil {
+		err = io.EOF
+		a.finished = true
+	} else {
+		_ = a.sourceAck(ctx, err)
+	}
+	return nil, nil, err
+}
+
+func (a *regexReader) Close(ctx context.Context) error {
+	a.mut.Lock()
+	defer a.mut.Unlock()
+
+	if !a.finished {
+		_ = a.sourceAck(ctx, errors.New("service shutting down"))
+	}
+	if a.pending == 0 {
+		_ = a.sourceAck(ctx, nil)
+	}
+	return a.r.Close()
+}
diff --git a/internal/http/client.go b/internal/http/client.go
index aa92bdcbb..7754b619e 100644
--- a/internal/http/client.go
+++ b/internal/http/client.go
@@ -28,6 +28,16 @@ import (
 	"github.com/Jeffail/benthos/v3/lib/util/throttle"
 )
 
+// MultipartExpressions represents three dynamic expressions that define a
+// multipart message part in an HTTP request. Specifying one or more of these
+// can be used as a way of creating HTTP requests that overrides the default
+// behaviour.
+type MultipartExpressions struct {
+	ContentDisposition *field.Expression
+	ContentType        *field.Expression
+	Body               *field.Expression
+}
+
 // Client is a component able to send and receive Benthos messages over HTTP.
 type Client struct {
 	client *http.Client
@@ -38,6 +48,7 @@ type Client struct {
 
 	url               *field.Expression
 	headers           map[string]*field.Expression
+	multipart         []MultipartExpressions
 	host              *field.Expression
 	metaInsertFilter  *metadata.IncludeFilter
 	metaExtractFilter *metadata.IncludeFilter
@@ -215,6 +226,13 @@ func OptSetLogger(log log.Modular) func(*Client) {
 	}
 }
 
+// OptSetMultiPart sets the multipart to request.
+func OptSetMultiPart(multipart []MultipartExpressions) func(*Client) {
+	return func(t *Client) {
+		t.multipart = multipart
+	}
+}
+
 // OptSetStats sets the metrics aggregator to use.
 func OptSetStats(stats metrics.Type) func(*Client) {
 	return func(t *Client) {
@@ -295,8 +313,25 @@ func (h *Client) waitForAccess(ctx context.Context) bool {
 func (h *Client) CreateRequest(sendMsg, refMsg types.Message) (req *http.Request, err error) {
 	var overrideContentType string
 	var body io.Reader
-
-	if sendMsg != nil && sendMsg.Len() == 1 {
+	if len(h.multipart) > 0 {
+		buf := &bytes.Buffer{}
+		writer := multipart.NewWriter(buf)
+		for _, v := range h.multipart {
+			var part io.Writer
+			mh := make(textproto.MIMEHeader)
+			mh.Set("Content-Type", v.ContentType.String(0, refMsg))
+			mh.Set("Content-Disposition", v.ContentDisposition.String(0, refMsg))
+			if part, err = writer.CreatePart(mh); err != nil {
+				return
+			}
+			if _, err = io.Copy(part, bytes.NewReader([]byte(v.Body.String(0, refMsg)))); err != nil {
+				return
+			}
+		}
+		writer.Close()
+		overrideContentType = writer.FormDataContentType()
+		body = buf
+	} else if sendMsg != nil && sendMsg.Len() == 1 {
 		if msgBytes := sendMsg.Get(0).Get(); len(msgBytes) > 0 {
 			body = bytes.NewBuffer(msgBytes)
 		}
diff --git a/lib/output/http_client.go b/lib/output/http_client.go
index 743e88e70..7e29e8b47 100644
--- a/lib/output/http_client.go
+++ b/lib/output/http_client.go
@@ -41,6 +41,13 @@ these propagated responses.`,
 			docs.FieldAdvanced("propagate_response", "Whether responses from the server should be [propagated back](/docs/guides/sync_responses) to the input."),
 			docs.FieldCommon("max_in_flight", "The maximum number of messages to have in flight at a given time. Increase this to improve throughput."),
 			batch.FieldSpec(),
+			docs.FieldAdvanced(
+				"multipart", "EXPERIMENTAL: Create explicit multipart HTTP requests by specifying an array of parts to add to the request, each part specified consists of content headers and a data field that can be populated dynamically. If this field is populated it will override the default request creation behaviour.",
+			).Array().HasType(docs.FieldTypeObject).HasDefault([]interface{}{}).WithChildren(
+				docs.FieldInterpolatedString("content_type", "The content type of the individual message part.", "application/bin").HasDefault(""),
+				docs.FieldInterpolatedString("content_disposition", "The content disposition of the individual message part.", `form-data; name="bin"; filename='${! meta("AttachmentName") }`).HasDefault(""),
+				docs.FieldInterpolatedString("body", "The body of the individual message part.", `${! json("data.part1") }`).HasDefault(""),
+			).AtVersion("3.63.0"),
 		),
 		Categories: []Category{
 			CategoryNetwork,
diff --git a/lib/output/writer/http_client.go b/lib/output/writer/http_client.go
index deb5c04ba..34c5632bd 100644
--- a/lib/output/writer/http_client.go
+++ b/lib/output/writer/http_client.go
@@ -2,9 +2,11 @@ package writer
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/Jeffail/benthos/v3/internal/http"
+	"github.com/Jeffail/benthos/v3/internal/interop"
 	"github.com/Jeffail/benthos/v3/lib/log"
 	"github.com/Jeffail/benthos/v3/lib/message/batch"
 	"github.com/Jeffail/benthos/v3/lib/message/roundtrip"
@@ -15,14 +17,25 @@ import (
 
 //------------------------------------------------------------------------------
 
+// HTTPClientMultipartExpression represents dynamic expressions that define a
+// multipart message part in an HTTP request. Specifying one or more of these
+// can be used as a way of creating HTTP requests that overrides the default
+// behaviour.
+type HTTPClientMultipartExpression struct {
+	ContentDisposition string `json:"content_disposition" yaml:"content_disposition"`
+	ContentType        string `json:"content_type" yaml:"content_type"`
+	Body               string `json:"body" yaml:"body"`
+}
+
 // HTTPClientConfig contains configuration fields for the HTTPClient output
 // type.
 type HTTPClientConfig struct {
 	client.Config     `json:",inline" yaml:",inline"`
-	BatchAsMultipart  bool               `json:"batch_as_multipart" yaml:"batch_as_multipart"`
-	MaxInFlight       int                `json:"max_in_flight" yaml:"max_in_flight"`
-	PropagateResponse bool               `json:"propagate_response" yaml:"propagate_response"`
-	Batching          batch.PolicyConfig `json:"batching" yaml:"batching"`
+	BatchAsMultipart  bool                            `json:"batch_as_multipart" yaml:"batch_as_multipart"`
+	MaxInFlight       int                             `json:"max_in_flight" yaml:"max_in_flight"`
+	PropagateResponse bool                            `json:"propagate_response" yaml:"propagate_response"`
+	Batching          batch.PolicyConfig              `json:"batching" yaml:"batching"`
+	Multipart         []HTTPClientMultipartExpression `json:"multipart" yaml:"multipart"`
 }
 
 // NewHTTPClientConfig creates a new HTTPClientConfig with default values.
@@ -63,14 +76,35 @@ func NewHTTPClient(
 		conf:      conf,
 		closeChan: make(chan struct{}),
 	}
-	var err error
-	if h.client, err = http.NewClient(
-		conf.Config,
+
+	opts := []func(*http.Client){
 		http.OptSetLogger(h.log),
 		http.OptSetManager(mgr),
 		// TODO: V4 Remove this
 		http.OptSetStats(metrics.Namespaced(h.stats, "client")),
-	); err != nil {
+	}
+
+	if len(conf.Multipart) > 0 {
+		parts := make([]http.MultipartExpressions, len(conf.Multipart))
+		for i, p := range conf.Multipart {
+			var exprPart http.MultipartExpressions
+			var err error
+			if exprPart.ContentDisposition, err = interop.NewBloblangField(mgr, p.ContentDisposition); err != nil {
+				return nil, fmt.Errorf("failed to parse multipart %v field content_disposition: %v", i, err)
+			}
+			if exprPart.ContentType, err = interop.NewBloblangField(mgr, p.ContentType); err != nil {
+				return nil, fmt.Errorf("failed to parse multipart %v field content_type: %v", i, err)
+			}
+			if exprPart.Body, err = interop.NewBloblangField(mgr, p.Body); err != nil {
+				return nil, fmt.Errorf("failed to parse multipart %v field data: %v", i, err)
+			}
+			parts[i] = exprPart
+		}
+		opts = append(opts, http.OptSetMultiPart(parts))
+	}
+
+	var err error
+	if h.client, err = http.NewClient(conf.Config, opts...); err != nil {
 		return nil, err
 	}
 	return &h, nil
diff --git a/public/service/config.go b/public/service/config.go
index 7f8be2edc..cdddb4f00 100644
--- a/public/service/config.go
+++ b/public/service/config.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 
 	"github.com/Jeffail/benthos/v3/internal/bloblang/query"
 	"github.com/Jeffail/benthos/v3/internal/bundle"
@@ -29,6 +30,15 @@ func NewStringField(name string) *ConfigField {
 	}
 }
 
+// NewDurationField describes a new duration string type config field, allowing
+// users to define a time interval with strings of the form 60s, 3m, etc.
+func NewDurationField(name string) *ConfigField {
+	// TODO: Add linting rule for duration
+	return &ConfigField{
+		field: docs.FieldString(name, ""),
+	}
+}
+
 // NewStringEnumField describes a new string type config field that can have one
 // of a discrete list of values.
 func NewStringEnumField(name string, options ...string) *ConfigField {
@@ -502,6 +512,28 @@ func (p *ParsedConfig) FieldString(path ...string) (string, error) {
 	return str, nil
 }
 
+// FieldDuration accesses a duration string field from the parsed config by its
+// name. If the field is not found or is not a valid duration string an error is
+// returned.
+//
+// This method is not valid when the configuration spec was built around a
+// config constructor.
+func (p *ParsedConfig) FieldDuration(path ...string) (time.Duration, error) {
+	v, exists := p.field(path...)
+	if !exists {
+		return 0, fmt.Errorf("field '%v' was not found in the config", p.fullDotPath(path...))
+	}
+	str, ok := v.(string)
+	if !ok {
+		return 0, fmt.Errorf("expected field '%v' to be a string, got %T", p.fullDotPath(path...), v)
+	}
+	d, err := time.ParseDuration(str)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse '%v' as a duration string: %w", p.fullDotPath(path...), err)
+	}
+	return d, nil
+}
+
 // FieldStringList accesses a field that is a list of strings from the parsed
 // config by its name and returns the value. Returns an error if the field is
 // not found, or is not a list of strings.
diff --git a/public/service/config_backoff.go b/public/service/config_backoff.go
new file mode 100644
index 000000000..e06169fef
--- /dev/null
+++ b/public/service/config_backoff.go
@@ -0,0 +1,54 @@
+package service
+
+import (
+	"github.com/cenkalti/backoff/v4"
+)
+
+// NewBackOffField defines a new object type config field that describes an
+// exponential back off policy, often used for timing retry attempts. It is then
+// possible to extract a *backoff.ExponentialBackOff from the resulting parsed
+// config with the method FieldBackOff.
+//
+// It is possible to configure a back off policy that has no upper bound (no
+// maximum elapsed time set). In cases where this would be problematic the field
+// allowUnbounded should be set `false` in order to add linting rules that
+// ensure an upper bound is set.
+func NewBackOffField(name string, allowUnbounded bool) *ConfigField {
+	maxElapsedTime := NewDurationField("max_elapsed_time").
+		Description("The maximum overall period of time to spend on retry attempts before the request is aborted.").
+		Default("15m").Example("1m").Example("1h")
+	if allowUnbounded {
+		maxElapsedTime.field.Description += " Setting this value to a zeroed duration (such as `0s`) will result in unbounded retries."
+	}
+
+	// TODO: Add linting rule to ensure we aren't unbounded if necessary.
+	return NewObjectField(name,
+		NewDurationField("initial_interval").
+			Description("The initial period to wait between retry attempts.").
+			Default("500ms").Example("50ms").Example("1s"),
+		NewDurationField("max_interval").
+			Description("The maximum period to wait between retry attempts").
+			Default("60s").Example("5s").Example("1m"),
+		maxElapsedTime,
+	)
+}
+
+// FieldBackOff accesses a field from a parsed config that was defined with
+// NewBackoffField and returns a *tls.Config, or an error if the configuration was
+// invalid.
+func (p *ParsedConfig) FieldBackOff(path ...string) (*backoff.ExponentialBackOff, error) {
+	b := backoff.NewExponentialBackOff()
+
+	var err error
+	if b.InitialInterval, err = p.FieldDuration(append(path, "initial_interval")...); err != nil {
+		return nil, err
+	}
+	if b.MaxInterval, err = p.FieldDuration(append(path, "max_interval")...); err != nil {
+		return nil, err
+	}
+	if b.MaxElapsedTime, err = p.FieldDuration(append(path, "max_elapsed_time")...); err != nil {
+		return nil, err
+	}
+
+	return b, nil
+}
diff --git a/website/docs/components/inputs/aws_s3.md b/website/docs/components/inputs/aws_s3.md
index a49f4fd83..8f9347478 100644
--- a/website/docs/components/inputs/aws_s3.md
+++ b/website/docs/components/inputs/aws_s3.md
@@ -235,6 +235,7 @@ Default: `"all-bytes"`
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/azure_blob_storage.md b/website/docs/components/inputs/azure_blob_storage.md
index e5c5affd8..64b570b48 100644
--- a/website/docs/components/inputs/azure_blob_storage.md
+++ b/website/docs/components/inputs/azure_blob_storage.md
@@ -159,6 +159,7 @@ Default: `"all-bytes"`
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/file.md b/website/docs/components/inputs/file.md
index 793105a38..fedf9c215 100644
--- a/website/docs/components/inputs/file.md
+++ b/website/docs/components/inputs/file.md
@@ -92,6 +92,7 @@ Default: `"lines"`
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/gcp_cloud_storage.md b/website/docs/components/inputs/gcp_cloud_storage.md
index f1c884768..c2914cd08 100644
--- a/website/docs/components/inputs/gcp_cloud_storage.md
+++ b/website/docs/components/inputs/gcp_cloud_storage.md
@@ -120,6 +120,7 @@ Default: `"all-bytes"`
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/http_client.md b/website/docs/components/inputs/http_client.md
index cc34ece85..61152395e 100644
--- a/website/docs/components/inputs/http_client.md
+++ b/website/docs/components/inputs/http_client.md
@@ -717,6 +717,7 @@ Requires version 3.42.0 or newer
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/sftp.md b/website/docs/components/inputs/sftp.md
index cd5ce2077..3d64b95db 100644
--- a/website/docs/components/inputs/sftp.md
+++ b/website/docs/components/inputs/sftp.md
@@ -164,6 +164,7 @@ Default: `"all-bytes"`
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/socket.md b/website/docs/components/inputs/socket.md
index 4972822d1..dffa9c1e2 100644
--- a/website/docs/components/inputs/socket.md
+++ b/website/docs/components/inputs/socket.md
@@ -100,6 +100,7 @@ Requires version 3.42.0 or newer
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/socket_server.md b/website/docs/components/inputs/socket_server.md
index 7a8893d0c..30d5f898e 100644
--- a/website/docs/components/inputs/socket_server.md
+++ b/website/docs/components/inputs/socket_server.md
@@ -101,6 +101,7 @@ Requires version 3.42.0 or newer
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/inputs/stdin.md b/website/docs/components/inputs/stdin.md
index dbf2a7512..12b4c96af 100644
--- a/website/docs/components/inputs/stdin.md
+++ b/website/docs/components/inputs/stdin.md
@@ -76,6 +76,7 @@ Requires version 3.42.0 or newer
 | `gzip` | Decompress a gzip file, this codec should precede another codec, e.g. `gzip/all-bytes`, `gzip/tar`, `gzip/csv`, etc. |
 | `lines` | Consume the file in segments divided by linebreaks. |
 | `multipart` | Consumes the output of another codec and batches messages together. A batch ends when an empty message is consumed. For example, the codec `lines/multipart` could be used to consume multipart messages where an empty line indicates the end of each batch. |
+| `regex:(?m)^\d\d:\d\d:\d\d` | Consume the file in segments divided by regular expression. |
 | `tar` | Parse the file as a tar archive, and consume each file of the archive as a message. |
 
 
diff --git a/website/docs/components/outputs/http_client.md b/website/docs/components/outputs/http_client.md
index 819713808..32b5904df 100644
--- a/website/docs/components/outputs/http_client.md
+++ b/website/docs/components/outputs/http_client.md
@@ -111,6 +111,7 @@ output:
       period: ""
       check: ""
       processors: []
+    multipart: []
 ```
 
 </TabItem>
@@ -764,4 +765,58 @@ processors:
   - merge_json: {}
 ```
 
+### `multipart`
+
+EXPERIMENTAL: Create explicit multipart HTTP requests by specifying an array of parts to add to the request, each part specified consists of content headers and a data field that can be populated dynamically. If this field is populated it will override the default request creation behaviour.
+
+
+Type: `array`  
+Default: `[]`  
+Requires version 3.63.0 or newer  
+
+### `multipart[].content_type`
+
+The content type of the individual message part.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
+
+
+Type: `string`  
+Default: `""`  
+
+```yaml
+# Examples
+
+content_type: application/bin
+```
+
+### `multipart[].content_disposition`
+
+The content disposition of the individual message part.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
+
+
+Type: `string`  
+Default: `""`  
+
+```yaml
+# Examples
+
+content_disposition: form-data; name="bin"; filename='${! meta("AttachmentName") }
+```
+
+### `multipart[].body`
+
+The body of the individual message part.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
+
+
+Type: `string`  
+Default: `""`  
+
+```yaml
+# Examples
+
+body: ${! json("data.part1") }
+```
+
 



Test patch:
diff --git a/internal/codec/reader_test.go b/internal/codec/reader_test.go
index 7eaf86ce1..4a54747d5 100644
--- a/internal/codec/reader_test.go
+++ b/internal/codec/reader_test.go
@@ -742,3 +742,30 @@ func TestMultipartLinesReader(t *testing.T) {
 	data = []byte("")
 	testReaderSuite(t, "lines/multipart", "", data)
 }
+
+func TestRegexpSplitReader(t *testing.T) {
+	data := []byte("foo\nbar\nbaz")
+	testReaderSuite(t, "regex:(?m)^", "", data, "foo\n", "bar\n", "baz")
+
+	data = []byte("foo\nbar\nsplit\nbaz\nsplitsplit")
+	testReaderSuite(t, "regex:split", "", data, "foo\nbar\n", "split\nbaz\n", "split", "split")
+
+	data = []byte("split")
+	testReaderSuite(t, "regex:\\n", "", data, "split")
+	testReaderSuite(t, "regex:split", "", data, "split")
+
+	data = []byte("foo\nbar\nsplit\nbaz\nsplitsplit")
+	testReaderSuite(t, "regex:\\n", "", data, "foo", "\nbar", "\nsplit", "\nbaz", "\nsplitsplit")
+
+	data = []byte("foo\nbar\nsplit\nbaz")
+	testReaderSuite(t, "regex:\\n", "", data, "foo", "\nbar", "\nsplit", "\nbaz")
+
+	data = []byte("20:20:22 ERROR\nCode\n20:20:21 INFO\n20:20:21 INFO\n20:20:22 ERROR\nCode\n")
+	testReaderSuite(t, "regex:\\n\\d", "", data, "20:20:22 ERROR\nCode", "\n20:20:21 INFO", "\n20:20:21 INFO", "\n20:20:22 ERROR\nCode\n")
+
+	data = []byte("20:20:22 ERROR\nCode\n20:20:21 INFO\n20:20:21 INFO\n20:20\n20:20:22 ERROR\nCode\n2022")
+	testReaderSuite(t, "regex:(?m)^\\d\\d:\\d\\d:\\d\\d", "", data, "20:20:22 ERROR\nCode\n", "20:20:21 INFO\n", "20:20:21 INFO\n20:20\n", "20:20:22 ERROR\nCode\n2022")
+
+	data = []byte("")
+	testReaderSuite(t, "regex:split", "", data)
+}
diff --git a/lib/output/writer/http_client_test.go b/lib/output/writer/http_client_test.go
index a2ff03506..623681323 100644
--- a/lib/output/writer/http_client_test.go
+++ b/lib/output/writer/http_client_test.go
@@ -1,6 +1,7 @@
 package writer
 
 import (
+	"encoding/json"
 	"fmt"
 	"io"
 	"mime"
@@ -296,3 +297,174 @@ func TestHTTPClientMultipart(t *testing.T) {
 		t.Error(err)
 	}
 }
+func TestHTTPOutputClientMultipartBody(t *testing.T) {
+	nTestLoops := 1000
+	resultChan := make(chan types.Message, 1)
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		msg := message.New(nil)
+		defer func() {
+			resultChan <- msg
+		}()
+
+		mediaType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
+		if err != nil {
+			t.Errorf("Bad media type: %v -> %v", r.Header.Get("Content-Type"), err)
+			return
+		}
+
+		if strings.HasPrefix(mediaType, "multipart/") {
+			mr := multipart.NewReader(r.Body, params["boundary"])
+			for {
+				p, err := mr.NextPart()
+
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				msgBytes, err := io.ReadAll(p)
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				msg.Append(message.NewPart(msgBytes))
+			}
+		}
+	}))
+	defer ts.Close()
+
+	conf := NewHTTPClientConfig()
+	conf.URL = ts.URL + "/testpost"
+	conf.Multipart = []HTTPClientMultipartExpression{
+		{
+			ContentDisposition: `form-data; name="text"`,
+			ContentType:        "text/plain",
+			Body:               "PART-A"},
+		{
+			ContentDisposition: `form-data; name="file1"; filename="a.txt"`,
+			ContentType:        "text/plain",
+			Body:               "PART-B"},
+	}
+	h, err := NewHTTPClient(conf, types.NoopMgr(), log.Noop(), metrics.Noop())
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < nTestLoops; i++ {
+		if err = h.Write(message.New([][]byte{[]byte("test")})); err != nil {
+			t.Error(err)
+		}
+		select {
+		case resMsg := <-resultChan:
+			if resMsg.Len() != len(conf.Multipart) {
+				t.Errorf("Wrong # parts: %v != %v", resMsg.Len(), 2)
+				return
+			}
+			if exp, actual := "PART-A", string(resMsg.Get(0).Get()); exp != actual {
+				t.Errorf("Wrong result, %v != %v", exp, actual)
+				return
+			}
+			if exp, actual := "PART-B", string(resMsg.Get(1).Get()); exp != actual {
+				t.Errorf("Wrong result, %v != %v", exp, actual)
+				return
+			}
+		case <-time.After(time.Second):
+			t.Errorf("Action timed out")
+			return
+		}
+	}
+
+	h.CloseAsync()
+	if err = h.WaitForClose(time.Second); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestHTTPOutputClientMultipartHeaders(t *testing.T) {
+	resultChan := make(chan types.Message, 1)
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		msg := message.New(nil)
+		defer func() {
+			resultChan <- msg
+		}()
+
+		mediaType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
+		if err != nil {
+			t.Errorf("Bad media type: %v -> %v", r.Header.Get("Content-Type"), err)
+			return
+		}
+
+		if strings.HasPrefix(mediaType, "multipart/") {
+			mr := multipart.NewReader(r.Body, params["boundary"])
+			for {
+				p, err := mr.NextPart()
+
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				a, err := json.Marshal(p.Header)
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				msg.Append(message.NewPart(a))
+			}
+		}
+	}))
+	defer ts.Close()
+
+	conf := NewHTTPClientConfig()
+	conf.URL = ts.URL + "/testpost"
+	conf.Multipart = []HTTPClientMultipartExpression{
+		{
+			ContentDisposition: `form-data; name="text"`,
+			ContentType:        "text/plain",
+			Body:               "PART-A"},
+		{
+			ContentDisposition: `form-data; name="file1"; filename="a.txt"`,
+			ContentType:        "text/plain",
+			Body:               "PART-B"},
+	}
+	h, err := NewHTTPClient(conf, types.NoopMgr(), log.Noop(), metrics.Noop())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = h.Write(message.New([][]byte{[]byte("test")})); err != nil {
+		t.Error(err)
+	}
+	select {
+	case resMsg := <-resultChan:
+		for i := range conf.Multipart {
+			if resMsg.Len() != len(conf.Multipart) {
+				t.Errorf("Wrong # parts: %v != %v", resMsg.Len(), 2)
+				return
+			}
+			mp := make(map[string][]string)
+			err := json.Unmarshal(resMsg.Get(i).Get(), &mp)
+			if err != nil {
+				t.Error(err)
+			}
+			if exp, actual := conf.Multipart[i].ContentDisposition, mp["Content-Disposition"]; exp != actual[0] {
+				t.Errorf("Wrong result, %v != %v", exp, actual)
+				return
+			}
+			if exp, actual := conf.Multipart[i].ContentType, mp["Content-Type"]; exp != actual[0] {
+				t.Errorf("Wrong result, %v != %v", exp, actual)
+				return
+			}
+		}
+	case <-time.After(time.Second):
+		t.Errorf("Action timed out")
+		return
+
+	}
+	h.CloseAsync()
+	if err = h.WaitForClose(time.Second); err != nil {
+		t.Error(err)
+	}
+}
diff --git a/public/service/config_backoff_test.go b/public/service/config_backoff_test.go
new file mode 100644
index 000000000..dd73d26ce
--- /dev/null
+++ b/public/service/config_backoff_test.go
@@ -0,0 +1,30 @@
+package service
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigBackOff(t *testing.T) {
+	spec := NewConfigSpec().
+		Field(NewBackOffField("a", true))
+
+	parsedConfig, err := spec.ParseYAML(`
+a:
+  max_interval: 300s
+`, nil)
+	require.NoError(t, err)
+
+	_, err = parsedConfig.FieldBackOff("b")
+	require.Error(t, err)
+
+	bConf, err := parsedConfig.FieldBackOff("a")
+	require.NoError(t, err)
+
+	assert.Equal(t, time.Millisecond*500, bConf.InitialInterval)
+	assert.Equal(t, time.Second*300, bConf.MaxInterval)
+	assert.Equal(t, time.Minute*15, bConf.MaxElapsedTime)
+}