Integration with web architecture

[gaoshenga]

Hello, how do you integrate benthos with the web framework?

[mihaitodor]

Hey @gaoshenga, I'm not sure I have enough information to answer your question. Which web framework are you referring to? I'm guessing you might be interested in building a custom http_server input which you can do by importing Benthos as a library and adding your own input plugin as shown in this repo.

[gaoshenga]

Hello @mihaitodor, thank you for your help. I'm ready to integrate the benthos plug-in with the GIN framework. I want to start the benthos instance by calling the bentho method to read the local configuration file. Instead of running a command like the benthos command "benthos -c./config.yaml" to start.

[mihaitodor]

OK, have a look at the StreamBuilderConfig example here which I think covers what you’re looking for.

[gaoshenga]

Thank you. I tried the case you provided just now. This case is the result of just one run. I want to execute a local file through benthos, and then launch a benthos instance. Is there any other way to do it?

[mihaitodor]

Not sure I follow... I think that example does exactly what you're asking for, assuming the input that you configure for Benthos terminates (basically stream.Run() returns) after processing all its data, like, for example, the file or the generate ones.

[gaoshenga]

Thank you very much. Your answer has given me a deeper understanding. It seems impossible to monitor the state of the stream with StreamBuilderConfig. It would be nice if we could monitor the state of streams like the streams rest api.

[mihaitodor]

Not sure what issue you bumped into, but here's a slightly modified version of that example I linked above:

package main

import (
	"context"

	"github.com/benthosdev/benthos/v4/public/service"

	// Import only required Benthos components, switch with `components/all` for
	// all standard components.
	_ "github.com/benthosdev/benthos/v4/public/components/io"
	_ "github.com/benthosdev/benthos/v4/public/components/pure"
        _ "github.com/benthosdev/benthos/v4/public/components/prometheus"
)

func main() {
	panicOnErr := func(err error) {
		if err != nil {
			panic(err)
		}
	}

	builder := service.NewStreamBuilder()

	// Set the full Benthos configuration of the stream.
	err := builder.SetYAML(`
http:
  address: 0.0.0.0:4195
  enabled: true

input:
  generate:
    count: 0
    interval: 1s
    mapping: 'root = "hello world"'

  processors:
    - mapping: 'root = content().uppercase()'

output:
  stdout: {}

logger:
  level: none

metrics:
  prometheus: {}
`)
	panicOnErr(err)

	// Build a stream with our configured components.
	stream, err := builder.Build()
	panicOnErr(err)

	// And run it, blocking until it gracefully terminates once the generate
	// input has generated a message and it has flushed through the stream.
	err = stream.Run(context.Background())
	panicOnErr(err)

}

If I run this, I'm able to do curl http://localhost:4195/metrics and it spews out the Prometheus metrics. Note that I also added the prometheus component import from here https://github.com/benthosdev/benthos/blob/main/public/components/all/package.go, but you can always add all of them by importing github.com/benthosdev/benthos/v4/public/components/all if you don't wish to cherry-pick only what you need.

[gaoshenga]

I see. I want to create a web project that can configure multiple processing data streams through the front page, each stream being a separate configuration file. You can start, stop, and monitor the status of each stream dynamically via benthos. This is my first experience with benthos. Thank you for your wonderful answer.