Merge branch 'warpstreamlabs:main' into etcd-component

.github/workflows/cross_build.yml

@@ -40,7 +40,7 @@ jobs:
         check-latest: true
 
     - name: GoReleaser
-      uses: goreleaser/goreleaser-action@v5
+      uses: goreleaser/goreleaser-action@v6
       with:
         version: latest
         args: release --snapshot --timeout 120m

.github/workflows/deploy-docsite.yml

@@ -0,0 +1,46 @@
+# https://github.com/actions/starter-workflows/blob/main/pages/static.yml
+
+# Added Stage 'Build-Docsite'
+# Changed 'Upload-Artifact' path
+
+name: Deploy Doc Site
+
+on:
+  workflow_dispatch:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+  pages: write
+  id-token: write
+
+# Allow only one concurrent deployment, skipping runs queued between the run in-progress and latest queued.
+# However, do NOT cancel in-progress runs as we want to allow these production deployments to complete.
+concurrency:
+  group: "pages"
+  cancel-in-progress: false
+
+jobs:
+  deploy:
+    environment:
+      name: github-pages
+      url: ${{ steps.deployment.outputs.page_url }}
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Setup Pages
+        uses: actions/configure-pages@v5
+      - name: Build-Docsite
+        uses: actions/setup-node@v4
+        with:
+          node-version: 20.x
+      - run: yarn --cwd ./website install && yarn --cwd ./website build
+      - name: Upload artifact
+        uses: actions/upload-pages-artifact@v3
+        with:
+          path: './website/build'
+      - name: Deploy to GitHub Pages
+        id: deployment
+        uses: actions/deploy-pages@v4

.github/workflows/docker_edge.yml

@@ -41,7 +41,7 @@ jobs:
       uses: docker/setup-buildx-action@v3
 
     - name: Build and push
-      uses: docker/build-push-action@v5
+      uses: docker/build-push-action@v6
       with:
         context: ./
         file: ./resources/docker/Dockerfile
@@ -51,7 +51,7 @@ jobs:
         tags: warpstreamlabs/bento:edge,ghcr.io/${{ github.repository_owner }}/bento:edge
 
     - name: Build and push CGO
-      uses: docker/build-push-action@v5
+      uses: docker/build-push-action@v6
       with:
         context: ./
         file: ./resources/docker/Dockerfile.cgo

.github/workflows/release.yml

@@ -33,7 +33,7 @@ jobs:
       run: ./resources/scripts/release_notes.sh > ./release_notes.md
 
     - name: GoReleaser
-      uses: goreleaser/goreleaser-action@v5
+      uses: goreleaser/goreleaser-action@v6
       with:
         version: latest
         args: release --release-notes=./release_notes.md --timeout 120m
@@ -88,7 +88,7 @@ jobs:
           type=semver,suffix=-cgo,pattern={{major}}
 
     - name: Build and push CGO
-      uses: docker/build-push-action@v5
+      uses: docker/build-push-action@v6
       with:
         context: ./
         file: ./resources/docker/Dockerfile.cgo
@@ -107,7 +107,7 @@ jobs:
           type=semver,pattern={{major}}
 
     - name: Build and push
-      uses: docker/build-push-action@v5
+      uses: docker/build-push-action@v6
       with:
         context: ./
         file: ./resources/docker/Dockerfile

.github/workflows/test.yml

@@ -56,7 +56,20 @@ jobs:
         version: v1.57.1
         args: --timeout 30m cmd/... internal/... public/...
 
+  build-docsite:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Build Doc Site
+      uses: actions/setup-node@v4
+      with:
+        node-version: 20.x
+    - run: yarn --cwd ./website install && yarn --cwd ./website build
+
     # TODO: Remove this after https://github.com/golangci/golangci-lint/issues/3094 is addressed
+    # Issue: https://github.com/warpstreamlabs/bento/issues/53
     # - name: Get govulncheck
     #   run: go install golang.org/x/vuln/cmd/govulncheck@latest
 

CHANGELOG.old.md

@@ -3,6 +3,19 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## 4.30.0 - 2024-06-13
+
+### Added
+
+- Go API: New APIs for capturing synchronous responses from downstream components. (@Jeffail)
+- Go API: Ability to customise the overall configuration schema of a stream builder. (@Jeffail)
+- New `sin`, `cos`, `tan` and `pi` bloblang methods. (@mfamador)
+- Field `proxy_url` added to the `websocket` input and output. (@mihaitodor)
+
+### Fixed
+
+- The `websocket` input and output now obey the `HTTP_PROXY`, `HTTPS_PROXY` and `NO_PROXY` environment variables. (@mihaitodor)
+
 ## 4.29.0 - 2024-06-10
 
 ### Added

CONTRIBUTING.md

@@ -51,4 +51,4 @@ The APIs for adding new components (inputs, outputs, processors, caches, etc) ha
 
 The core components within Bento (inputs, processors, conditions and outputs) are all easily pluggable. If you are interested in adding new components please raise a ticket and we can discuss whether it's a good fit for the project.
 
-If not then it's still easy to build your own version of Bento with custom components. For guidance take a look at [this example repo](https://github.com/benthosdev/benthos-plugin-example).
+If not then it's still easy to build your own version of Bento with custom components.

README.md

@@ -129,7 +129,7 @@ And then run it with `make lint`.
 
 ## Plugins
 
-It's pretty easy to write your own custom plugins for Bento in Go, for information check out [the API docs][godoc-url], and for inspiration there's an [example repo][plugin-repo] demonstrating a variety of plugin implementations.
+It's pretty easy to write your own custom plugins for Bento in Go, for information check out [the API docs][godoc-url].
 
 ## Extra Plugins
 

internal/api/config.go

@@ -33,7 +33,7 @@ func NewConfig() Config {
 	return Config{
 		Address:        "0.0.0.0:4195",
 		Enabled:        true,
-		RootPath:       "/benthos",
+		RootPath:       "/bento",
 		DebugEndpoints: false,
 		CertFile:       "",
 		KeyFile:        "",

internal/api/docs.go

@@ -17,7 +17,7 @@ func Spec() docs.FieldSpecs {
 		docs.FieldString(fieldAddress, "The address to bind to.").HasDefault("0.0.0.0:4195"),
 		docs.FieldString(
 			fieldRootPath, "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`.",
-		).HasDefault("/benthos"),
+		).HasDefault("/bento"),
 		docs.FieldBool(
 			fieldDebugEndpoints, "Whether to register a few extra endpoints that can be useful for debugging performance or behavioral problems.",
 		).HasDefault(false),
@@ -48,14 +48,14 @@ func DocsMarkdown() ([]byte, error) {
 http:
   address: 0.0.0.0:4195
   enabled: true
-  root_path: /benthos
+  root_path: /bento
   debug_endpoints: false
 `,
 		AdvancedConfig: `
 http:
   address: 0.0.0.0:4195
   enabled: true
-  root_path: /benthos
+  root_path: /bento
   debug_endpoints: false
   cert_file: ""
   key_file: ""

internal/bloblang/query/methods_strings.go

@@ -2011,7 +2011,7 @@ root.description = this.description.trim_prefix("foobar_")`,
 			`{"description":"unchanged","name":"blobton"}`,
 		),
 	).Param(ParamString("prefix", "The leading prefix substring to trim from the string.")).
-		AtVersion("4.12.0"),
+		AtVersion("1.0.0"),
 	func(args *ParsedParams) (simpleMethod, error) {
 		prefix, err := args.FieldString("prefix")
 		if err != nil {
@@ -2043,7 +2043,7 @@ root.description = this.description.trim_suffix("_foobar")`,
 			`{"description":"unchanged","name":"blobton"}`,
 		),
 	).Param(ParamString("suffix", "The trailing suffix substring to trim from the string.")).
-		AtVersion("4.12.0"),
+		AtVersion("1.0.0"),
 	func(args *ParsedParams) (simpleMethod, error) {
 		suffix, err := args.FieldString("suffix")
 		if err != nil {

internal/docs/config.go

@@ -40,7 +40,7 @@ var labelField = FieldString(
 		return "label field should be omitted when pointing to a resource", true
 	}
 	return "", false
-}).AtVersion("3.44.0").LinterFunc(func(ctx LintContext, line, col int, v any) []Lint {
+}).AtVersion("1.0.0").LinterFunc(func(ctx LintContext, line, col int, v any) []Lint {
 	l, _ := v.(string)
 	if l == "" {
 		return nil

internal/httpclient/auth_oauth2.go

@@ -82,7 +82,7 @@ func oAuth2FieldSpec() *service.ConfigField {
 			Description("A list of optional requested permissions.").
 			Default([]any{}).
 			Advanced().
-			Version("3.45.0"),
+			Version("1.0.0"),
 
 		service.NewAnyMapField(ao2FieldEndpointParams).
 			Description("A list of optional endpoint parameters, values should be arrays of strings.").
@@ -92,7 +92,7 @@ func oAuth2FieldSpec() *service.ConfigField {
 				"bar": []string{"woof"},
 			}).
 			Default(map[string]any{}).
-			Version("4.21.0").
+			Version("1.0.0").
 			Optional().
 			LintRule(`
 root = if this.type() == "object" {

internal/httpclient/config.go

@@ -54,7 +54,7 @@ func ConfigField(defaultVerb string, forOutput bool, extraChildren ...*service.C
 			Description("EXPERIMENTAL: Optionally set a level at which the request and response payload of each request made will be logged.").
 			Advanced().
 			Default("").
-			Version("4.12.0"),
+			Version("1.0.0"),
 	}
 	innerFields = append(innerFields, AuthFieldSpecsExpanded()...)
 

internal/httpserver/cors.go

@@ -49,7 +49,7 @@ func ServerCORSFieldSpec() docs.FieldSpec {
 	return docs.FieldObject(fieldCORS, "Adds Cross-Origin Resource Sharing headers.").WithChildren(
 		docs.FieldBool(fieldCORSEnabled, "Whether to allow CORS requests.").HasDefault(false),
 		docs.FieldString(fieldCORSAllowedOrigins, "An explicit list of origins that are allowed for CORS requests.").Array().HasDefault([]any{}),
-	).AtVersion("3.63.0").Advanced()
+	).AtVersion("1.0.0").Advanced()
 }
 
 func CORSConfigFromParsed(pConf *docs.ParsedConfig) (conf CORSConfig, err error) {

internal/impl/amqp09/input.go

@@ -56,7 +56,7 @@ You can access these metadata fields using [function interpolation](/docs/config
 			Example([]string{"amqp://guest:guest@127.0.0.1:5672/"}).
 			Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
 			Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
-			Version("3.58.0"),
+			Version("1.0.0"),
 		service.NewStringField(queueField).
 			Description("An AMQP queue to consume from."),
 		service.NewObjectField(queueDeclareField,
@@ -101,7 +101,7 @@ You can access these metadata fields using [function interpolation](/docs/config
 			Description("A list of regular expression patterns whereby if a message that has failed to be delivered by Bento has an error that matches it will be dropped (or delivered to a dead-letter queue if one exists). By default failed messages are nacked with requeue enabled.").
 			Example([]string{"^reject me please:.+$"}).
 			Advanced().
-			Version("3.64.0").
+			Version("1.0.0").
 			Default([]any{}),
 		service.NewIntField(prefetchCountField).
 			Description("The maximum number of pending messages to have consumed at a time.").

internal/impl/amqp09/output.go

@@ -34,7 +34,7 @@ The fields 'key', 'exchange' and 'type' can be dynamically set using function in
 				Example([]string{"amqp://guest:guest@127.0.0.1:5672/"}).
 				Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
 				Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
-				Version("3.58.0"),
+				Version("1.0.0"),
 			service.NewInterpolatedStringField(exchangeField).
 				Description("An AMQP exchange to publish to."),
 			service.NewObjectField(exchangeDeclareField,

internal/impl/amqp1/input.go

@@ -39,25 +39,25 @@ func amqp1InputSpec() *service.ConfigSpec {
 				Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
 				Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
 				Optional().
-				Version("4.23.0"),
+				Version("1.0.0"),
 			service.NewStringField(sourceAddrField).
 				Description("The source address to consume from.").
 				Example("/foo").
 				Example("queue:/bar").
 				Example("topic:/baz"),
 			service.NewBoolField(azureRenewLockField).
 				Description("Experimental: Azure service bus specific option to renew lock if processing takes more then configured lock time").
-				Version("3.45.0").
+				Version("1.0.0").
 				Default(false).
 				Advanced(),
 			service.NewBoolField(getMessageHeaderField).
 				Description("Read additional message header fields into `amqp_*` metadata properties.").
-				Version("4.25.0").
+				Version("1.0.0").
 				Default(false).Advanced(),
 			service.NewIntField(creditField).
 				Description("Specifies the maximum number of unacknowledged messages the sender can transmit. Once this limit is reached, no more messages will arrive until messages are acknowledged and settled.").
 				LintRule(`root = if this < 1 { [ "`+creditField+` must be at least 1" ] }`).
-				Version("4.26.0").
+				Version("1.0.0").
 				Default(64).
 				Advanced(),
 			service.NewTLSToggledField(tlsField),

internal/impl/amqp1/output.go

@@ -39,7 +39,7 @@ This output benefits from sending multiple messages in flight in parallel for im
 				Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
 				Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
 				Optional().
-				Version("4.23.0"),
+				Version("1.0.0"),
 			service.NewStringField(targetAddrField).
 				Description("The target address to write to.").
 				Example("/foo").

internal/impl/aws/cache_dynamodb.go

@@ -26,7 +26,7 @@ func dynCacheConfig() *service.ConfigSpec {
 
 	spec := service.NewConfigSpec().
 		Stable().
-		Version("3.36.0").
+		Version("1.0.0").
 		Summary(`Stores key/value pairs as a single document in a DynamoDB table. The key is stored as a string value and used as the table hash key. The value is stored as
 a binary value using the ` + "`data_key`" + ` field name.`).
 		Description(`A prefix can be specified to allow multiple cache types to share a single DynamoDB table. An optional TTL duration (` + "`ttl`" + `) and field

internal/impl/aws/cache_s3.go

@@ -24,7 +24,7 @@ func s3CacheConfig() *service.ConfigSpec {
 
 	spec := service.NewConfigSpec().
 		Stable().
-		Version("3.36.0").
+		Version("1.0.0").
 		Summary(`Stores each item in an S3 bucket as a file, where an item ID is the path of the item within the bucket.`).
 		Description(`It is not possible to atomically upload S3 objects exclusively when the target does not already exist, therefore this cache is not suitable for deduplication.`).
 		Field(service.NewStringField("bucket").

internal/impl/aws/config/config.go

@@ -30,7 +30,7 @@ func SessionFields() []*service.ConfigField {
 				Default("").Advanced(),
 			service.NewBoolField("from_ec2_role").
 				Description("Use the credentials of a host EC2 machine configured to assume [an IAM role associated with the instance](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_use_switch-role-ec2.html).").
-				Default(false).Version("4.2.0"),
+				Default(false).Version("1.0.0"),
 			service.NewStringField("role").
 				Description("A role ARN to assume.").
 				Default("").Advanced(),

internal/impl/aws/input_kinesis.go

@@ -78,7 +78,7 @@ func kinesisInputConfigFromParsed(pConf *service.ParsedConfig) (conf kiConfig, e
 func kinesisInputSpec() *service.ConfigSpec {
 	spec := service.NewConfigSpec().
 		Stable().
-		Version("3.36.0").
+		Version("1.0.0").
 		Categories("Services", "AWS").
 		Summary("Receive messages from one or more Kinesis streams.").
 		Description(`

internal/impl/aws/input_sqs.go

@@ -90,7 +90,7 @@ You can access these metadata fields using
 				Advanced(),
 			service.NewBoolField(sqsiFieldResetVisibility).
 				Description("Whether to set the visibility timeout of the consumed message to zero once it is nacked. Disabling honors the preset visibility timeout specified for the queue.").
-				Version("3.58.0").
+				Version("1.0.0").
 				Default(true).
 				Advanced(),
 			service.NewIntField(sqsiFieldMaxNumberOfMessages).

internal/impl/aws/metrics_cloudwatch.go

@@ -39,7 +39,7 @@ func cwmConfigFromParsed(pConf *service.ParsedConfig) (conf cwmConfig, err error
 func cwMetricsSpec() *service.ConfigSpec {
 	return service.NewConfigSpec().
 		Stable().
-		Version("3.36.0").
+		Version("1.0.0").
 		Summary(`Send metrics to AWS CloudWatch using the PutMetricData endpoint.`).
 		Description(`
 ### Timing Metrics

internal/impl/aws/output_dynamodb.go

@@ -70,7 +70,7 @@ func ddboConfigFromParsed(pConf *service.ParsedConfig) (conf ddboConfig, err err
 func ddboOutputSpec() *service.ConfigSpec {
 	return service.NewConfigSpec().
 		Stable().
-		Version("3.36.0").
+		Version("1.0.0").
 		Categories("Services", "AWS").
 		Summary(`Inserts items into a DynamoDB table.`).
 		Description(`

internal/impl/aws/output_kinesis.go

@@ -57,7 +57,7 @@ func koConfigFromParsed(pConf *service.ParsedConfig) (conf koConfig, err error)
 func koOutputSpec() *service.ConfigSpec {
 	return service.NewConfigSpec().
 		Stable().
-		Version("3.36.0").
+		Version("1.0.0").
 		Categories("Services", "AWS").
 		Summary(`Sends messages to a Kinesis stream.`).
 		Description(`