Is there a programmatic way to bring up benthos in an already existing app

[ghstahl]

My existing app has a cli and it doesn't look like its compatible with benthos. Here is my little hack in my main, but I would like more of something like.

benthosInstance := service.New(options)
benthosInstance.Start()

//much later
benthosInstance.Stop()

My Hack

	originalArgs := os.Args

	// cancel context
	ctx, cancel := context.WithCancel(ctx)
	// benthos thinks its the only one, so lets replace the args and then set them back when it launches.
	benthosOSArgsS := os.Getenv("BENTHOS_OS_ARGS")
	fmt.Println("BENTHOS_OS_ARGS", benthosOSArgsS)
	// split them
	benthosOSArgs := strings.Split(benthosOSArgsS, ",")
	os.Args = []string{
		originalArgs[0],
	}
	os.Args = append(os.Args, benthosOSArgs...)
	log.Info().Interface("os.Args", os.Args).Msg("os.Args")
	os.Args[0] = "benthos"
	go func() {
		service.RunCLI(context.Background())
	}()
	time.Sleep(5 * time.Second)
	os.Args = originalArgs

[mihaitodor]

Hey @ghstahl, I'm not sure if that os.Args hack does what you expect it to, since Go doesn't seem to expose a mechanism to overwrite it globally, but I could be wrong. See here and here.

I think you're looking for the github.com/benthosdev/benthos/v4/public/service/servicetest package which provides RunCLIWithArgs().

[mihaitodor]

I think the first issue is that you have http and logger in that yaml which need to go in the "general service wide configuration" (see here).

LE: My bad. The streams created with the StreamBuilder API are not equivalent to the Streams Mode pipelines in terms of config. See the docs here:

Streams built with a stream builder have the HTTP server for exposing metrics and ready checks disabled by default, which is the only deviation away from a standard Benthos default configuration. In order to enable the server set the configuration field http.enabled to true explicitly, or use SetHTTPMux in order to provide an explicit HTTP multiplexer for registering those endpoints.

[ghstahl]

That is what you have in your streambuilder example.

err = builderTwo.SetYAML(`
http:
  enabled: false

input:
  generate:
    
    interval: 1ms
    mapping: 'root = "hello world two"'

  processors:
    - sleep:
        duration: 500ms
    - mapping: 'root = content().capitalize()'

output:
  stdout: {}
`)

The errors are coming from the benthos linter.

The lint problem it has are with the following;

1. it doesn't like - mapping: "deadletterit(@,content())"
2. It doesn't like my custom output
   cloudevents_grpc:

BTW: I removed the logger section in the yaml

[mihaitodor]

Thanks for pointing that out! I raised it with @Jeffail on Discord since I kinda' expected the StreamBuilder API to do the same linting as benthos streams foobar.yaml, but you're right and it seems to accept the http and logger sections for each stream, so I'll see what he replies.

Otherwise, I think you might not be importing your custom components when running that stream. For example, the following code works for me:

package main

import (
	"context"
	"sync"

	"github.com/benthosdev/benthos/v4/public/bloblang"
	_ "github.com/benthosdev/benthos/v4/public/components/io"
	_ "github.com/benthosdev/benthos/v4/public/components/pure"
	"github.com/benthosdev/benthos/v4/public/service"
)

// Add Bloblang `foobar()` function
func init() {
	crazyObjectSpec := bloblang.NewPluginSpec().
		Param(bloblang.NewStringParam("key"))

	err := bloblang.RegisterFunctionV2("foobar", crazyObjectSpec, func(args *bloblang.ParsedParams) (bloblang.Function, error) {
		key, err := args.GetString("key")
		if err != nil {
			return nil, err
		}

		return func() (interface{}, error) {
			obj := map[string]interface{}{}
			obj[key] = "foobar"
			return obj, nil
		}, nil
	})
	if err != nil {
		panic(err)
	}
}

// Add `foobar` processor
func init() {
	configSpec := service.NewConfigSpec()

	constructor := func(conf *service.ParsedConfig, mgr *service.Resources) (service.Processor, error) {
		return newfoobarProcessor(mgr.Logger(), mgr.Metrics()), nil
	}

	err := service.RegisterProcessor("foobar", configSpec, constructor)
	if err != nil {
		panic(err)
	}
}

type foobarProcessor struct {
	logger *service.Logger
}

func newfoobarProcessor(logger *service.Logger, metrics *service.Metrics) *foobarProcessor {
	return &foobarProcessor{
		logger: logger,
	}
}

func (r *foobarProcessor) Process(ctx context.Context, m *service.Message) (service.MessageBatch, error) {
	m.SetBytes([]byte(`{"foo": "bar"}`))
	return []*service.Message{m}, nil
}

func (r *foobarProcessor) Close(ctx context.Context) error {
	return nil
}

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

	builder := service.NewStreamBuilder()

	err := builder.SetYAML(`
input:
  generate:
    count: 1
    interval: 1ms
    mapping: 'root = ""'

  processors:
    - foobar: {}
    - mapping: |
        root = this
        root.data = foobar("test")

output:
  stdout: {}
`)
	panicOnErr(err)

	stream, err := builder.Build()
	panicOnErr(err)

	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()
		panicOnErr(stream.Run(context.Background()))
	}()

	wg.Wait()
}

It outputs {"data":{"test":"foobar"},"foo":"bar"}.

[ghstahl]

Duh! my registration was after I was making call to that streamBuilder

[mihaitodor]

I think the github.com/fluffy-bunny/fluffy-dozm-di dependency injection abstraction is making the registration process a bit complex, but my debugger is showing me that service.Register() in https://github.com/fluffy-bunny/benthos-cloudevents-fluffycore/blob/41f19198482291b38f8404f4b74b844bce1bb38e/internal/services/bloblang/bloblang_registration.go never gets called.