Declarative streaming ETL for mundane tasks, written in Go

TLDR: Please share your use cases here or send them privately to [email protected], we will use these to help steer the roadmap towards Benthos version 4.

Hey everyone,

Benthos has been at major version 3 for over a year now, and I consider it to be a pretty cool achievement that given all the dope features added we've managed to keep both the Benthos config spec and APIs fully backwards compatible.

However, eventually it would be nice to cut a new major release and prune all of the dead weight that has accumulated during this time. In order to do that I want to be sure that we've fully thought through any potential breaking changes that could be bundled along with it, and that requires a roadmap.

Therefore I'm hoping to hear from you on how you use Benthos, what works well for you and what doesn't, and any features you'd hope to see eventually that would make your lives easier.

Some features that might come into play are:

• Improved plugin APIs (need more use cases to help shape this)
• HTTP poll APIs for the dynamic components as well as streams mode, allowing Benthos to pull configs
• Configuration file reloading
• Expanded logging, metrics and tracing options

Please let us know how you use (or intend to use) Benthos, either by posting here or getting in touch privately at [email protected]. You can also find us on the discord channel at https://discord.gg/6VaWjzP, more chat options can be found on the community page.

Once planning starts I'm going to be looking into organising public roadmap planning sessions.

If you're using Benthos and find something about it frustrating, but not worthy of raising an issue, please give me a brief description here. It could be anything: features, documentation, performance, me.

Also, if the thing that annoys you is already posted then give it a thumbs up.

I've been tinkering with the Lambda binary and had a few things I wanted to run by folks and get thoughts on. I'm happy to split this into multiple issues for multiple threads if we need it. To demo things, I've put together https://github.com/kconwayatlassian/benthos-ext that implements each of the ideas below:

• Ability to both produce to one or more outputs in addition to returning the processed message. This enables us to use any combination of outputs and brokers but still return the message as the Lambda response.

• Offering the Lambda function as a code export instead of only a binary for custom builds outside of Benthos and code reuse in things that adapt Lambda to other runtimes.

• Adding some test coverage of the Lambda functionality.

I'm looking for feedback on 1) if any of the ideas have value beyond my team that would justify having them in Benthos and 2) feedback on whether the demo code would be acceptable as a Benthos contribution or how I'd need to refactor in order to meet the project standards/style.

Is it possible to promote properties of a JSON input to attributes in gcp_pubsub?

https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage

I would like to promote things like event timestamps to attributes.time.

It looks like the gcp_pubsub output puts the data into Message Body on the GCP side.

We are using the attributes to sort the data in Cloud Dataflow:

read = PubsubIO
    .readMessagesWithAttributes()
    .withTimestampAttribute("time")
    .fromTopic(defaultOptions.getTopic());

EDIT: In the pubsub GO lib the attributes would go in the attributes field of the message struct here: https://github.com/googleapis/google-cloud-go/blob/pubsub/v1.2.0/pubsub/message.go#L36

You could likely pull off fields from the JSON object using jmespath and store them in an additional input to the gcp_pubsub output.

Likely this would require code updates. But I wanted to confirm it was not possible via some other means and whether the parsing of attributes is possible.

Thanks!

Using Service Bus from Azure (which uses AMQP 1.0)

Messages are received correctly, but after I while I get this message:

Failed to read message: link detached, reason: *Error{Condition: amqp:link:detach-forced, Description: The link 'G19:151274741:MB9xjRjuCfMcXqGvOKbKyGeG9fJypgtW-sjO4ErGrubNtxbFpGcCxx' is force detached. Code: consumer(link11331556). Details: InnerMessageReceiver was closed. TrackingId:e33664c70000930b00ace7e45f28340d_G19_B11, SystemTracker:queue-name-dev:Queue:queue-name, Timestamp:2020-08-03T16:01:10, Info: map[]}

I will analyze it more tomorrow and update the ticket if I find anything helpful

My config:

input:
  broker:
    inputs:
    - amqp_1:
        url: "myqueueurl"
        source_address: "my-queue-name"
    batching:
      count: 180
      period: "50s"
// processors etc.....

I have the following config and benthos 3.11.0:

input: kinesis_balanced: stream: what-a-great-stream start_from_oldest: false dynamodb_table: table-for-benthos dynamodb_billing_mode: PAY_PER_REQUEST region: us-east-1 credentials: profile: what-a-great-profile output: file: path: outfile.txt delimiter: ""

Using the same AWS profile, on us-east-1, I'm able to put records on this stream, get records from it, and when I scan dynamodb table-for-benthos I see leases are being updated by benthos as long as it's running. The file, outfile.txt is getting created, and if I swap the input to stdin it works.

I'm not getting any errors, and I'm not getting any output from kinesis.

I'm so close to living the dream, I can taste it.

Hey everyone it's been a while since we last updated the minimum Go version for Benthos and I'm considering updating to 1.18. The obvious immediate benefit would be to use generics in our own packages but it would also mean being able to use libraries that themselves use generics without having annoying build toggles.

Like before I'm pinning this issue, if setting the minimum version to 1.18 would be an issue for you or your team then let me know. None of this is blocking us from implementing anything critical so I'd rather delay than cause problems downstream.

Fixes #797.

I tested the output like so (might be worth adding some integration tests for this too, since only Postgres is covered under lib/test/integration:

localhost> docker run --rm -it -e ACCEPT_EULA=Y -e SA_PASSWORD=ins4n3lyStrongP4ssword -p1433:1433 microsoft/mssql-server-linux

container> /opt/mssql-tools/bin/sqlcmd -l 30 -S localhost -h-1 -V1 -U sa -P ins4n3lyStrongP4ssword -Q "create table footable(data varchar(50) not null);"

input:
  stdin:
    codec: lines
output:
  sql:
    driver: mssql
    data_source_name: "sqlserver://sa:ins4n3lyStrongP4ssword@localhost:1433?database=master"
    query: "insert into footable(data) values(?);"
    args:
      - ${! json("data") }

localhost> ./target/bin/benthos -c sql.yaml
{"data":"foo"}

container> /opt/mssql-tools/bin/sqlcmd -l 30 -S localhost -h-1 -V1 -U sa -P ins4n3lyStrongP4ssword -Q "select * from footable;"
foo

(1 rows affected)

Hey, @Jeffail , I have a little something for benthos.

I used github.com/influxdata/go-syslog to parse a message and I decided to implement just rfc5424 support. Shouldn't be a real problem to include another rfc under processor's wing but it is quite good for a starter. I am diabolically sad that rfc3164 (one of the popular) can't be parsed because the library hasn't implemented it yet, but then again I can't hold a grudge for longer than a few minutes, so..

I think, there is no need to provide functionality via config letting users not to create some of final JSON structure (not all of us need those field and those might be better off in the metadata), but every use case is unique and it's not fair to throw everyone in the same barrel, especially if it's limited space inside. So, overall, it should be easy to unset some of those JSON fields afterwards.

I also used bestEffort option which implies processor won't crash during the parsing if there would be some problem in an original message as long as it is bound, in a certain degree, to rfc.

Currently the input of type Kafka will only get the message from a given topic, is there any plan to enable also getting the keys?

My use case is very simply that I need to back up the _schemas topic, but then in order to restore I need to be able to produce those messages with the same key they were originally produced with.

Would you accept a PR for this?

I want to loop insert data from array into database. just like this:

input:
  generate:
    mapping: |
      root = {"sqldata":["delete from b.tax where month_of_salary='202208'","delete from a.tax where month_of_salary='202208'"]}
    interval: 0s
    count: 1
pipeline:
  processors:
    - while:
        at_least_once: false
        max_loops: 0
        check: this.sqldata.length() > 0
        processors:
          - sql_raw:
              driver: clickhouse
              dsn: clickhouse://[username[:password]@][netloc][:port]/dbname[?param1=value1&...&paramN=valueN]
              query: this.sqldata[i]
          
output:
  drop: {}

the processor of while is seem not appropriate and i can't get index of array.How to loop through array and execute sql statement?

Currently, this is implicitly set to 0, which deviates from the default value (2) that is specified in the standard library docs here.

Guess some people who are currently using conn_max_idle: 0 implicitly might bump into performance issues, not sure...

Hello,

We have developed a plugin[1] for franz-go that adds tracing to Kafka records and exports them to OpenTelemetry. This plugin allows tracing context propagation from upstream input into Kafka records and from Kafka records to downstream output.

We are interested in knowing if the Benthos team would like to incorporate this capability into the project. We can contribute to this feature and would be happy to work with the team. Please let us know your thoughts on this.

[1] https://github.com/twmb/franz-go/pull/294

We now purposefully unmarshal the Mongo query via the bson library. This adds implicit support for ObjectIds ($oid), Timestamps ($ts), Regexes ($regex), etc. in queries.

This may offer a workaround for #1516 as well (from what I've understood).

First golang contribution, feel free to nitpick 😅

I was looking at using benthos to replace a custom rolled solution, however the use-case requires strict ordering (within a hash key). Therefore I was looking at how benthos handles that on the Kinesis output.

Due to the indiscriminate usage of PutRecords with (possibly) equal ParitionKey's, it's possible for some in-between records to fail, be retried and be appended at the end of a batch rather than their rightful position. Not sure if I explained this well, but the first half of this article describes the issue (and solution) in more detail.

From what I've seen, it'd require doing the groupBy around here and possibly do multiple batches in parallel.

This issue is meant to track https://github.com/sijms/go-ora/issues/287 which has all the details. In essence, one can't currently SELECT columns of type TIMESTAMP directly, since they won't be deserialised correctly in sqlRowToMap and they'll end up containing an empty object ({}) instead of the actual timestamp value. I guess a workaround is to call the TO_CHAR() function on the TIMESTAMP column in the SELECT statement until this is addressed somehow.

Can we provide ability to add ClientOptions when building NewClient in output_pubsub flow? I am mainly looking at the option of setting the endpoint to a region specific pubsub endpoint. If providing the ability to some of those supported options is not recommended or going to be a breaking change, can we provide option to add specific ones as config like endpoint etc.. ?

File that needs to be changed: https://github.com/benthosdev/benthos/blob/48c7bb7a0718bdcef453cc156f60f2ffafb48730/internal/impl/gcp/output_pubsub.go#L101

Google Pubsub library doc related to Client Option: https://pkg.go.dev/cloud.google.com/go/pubsub#NewClient https://pkg.go.dev/google.golang.org/api/option#ClientOption