feat(elasticsearch) add new elasticsearch_v8 output #3160
Conversation
This new output uses the latest official Elasticsearch go library.
ooesili
added
enhancement
outputs
|
|
| if os.Getenv("ELASTICSEARCH_DEBUG") != "" { | ||
| conf.clientOpts.Logger = &elastictransport.CurlLogger{ | ||
| Output: os.Stdout, | ||
| EnableRequestBody: true, | ||
| EnableResponseBody: true, | ||
| } | ||
| } |
There was a problem hiding this comment.
Can we wrap this logger with our service logger at trace level instead?
There was a problem hiding this comment.
There's not an easy way to do that, elastictransport.CurlLogger is a decorator around a custom roundtripper type interface. I could write my own struct for that but it doesn't really seem worth it for a debugger that I intended to really only be used by us during development.
There was a problem hiding this comment.
Can we make the env var more verbose then? REDPANDA_CONNECT_ELASTICSEARCH_DEBUG=1
| if os.Getenv("ELASTICSEARCH_DEBUG") != "" { | ||
| conf.clientOpts.Logger = &elastictransport.CurlLogger{ | ||
| Output: os.Stdout, | ||
| EnableRequestBody: true, | ||
| EnableResponseBody: true, | ||
| } | ||
| } |
There was a problem hiding this comment.
Can we make the env var more verbose then? REDPANDA_CONNECT_ELASTICSEARCH_DEBUG=1
| @@ -35,6 +35,7 @@ import ( | |||
| _ "github.com/redpanda-data/connect/v4/public/components/dgraph" | |||
| _ "github.com/redpanda-data/connect/v4/public/components/discord" | |||
| _ "github.com/redpanda-data/connect/v4/public/components/elasticsearch" | |||
| _ "github.com/redpanda-data/connect/v4/public/components/elasticsearchv8" | |||
There was a problem hiding this comment.
nit: up to you, but I would just put this in the same elasticsearch package. You could also use a v8 subdirectory if you do want to separate the packages.
| @@ -70,6 +70,7 @@ drop_on ,output ,drop_on ,0.0.0 ,certif | |||
| dynamic ,input ,dynamic ,0.0.0 ,community ,n ,n ,n | |||
| dynamic ,output ,dynamic ,0.0.0 ,community ,n ,n ,n | |||
| elasticsearch ,output ,elasticsearch ,0.0.0 ,community ,n ,n ,n | |||
| elasticsearch_v8 ,output ,elasticsearch_v8 ,4.46.0 ,certified ,n ,y ,y | |||
There was a problem hiding this comment.
| elasticsearch_v8 ,output ,elasticsearch_v8 ,4.46.0 ,certified ,n ,y ,y | |
| elasticsearch_v8 ,output ,elasticsearch_v8 ,4.47.0 ,certified ,n ,y ,y |
| action, err := batch.TryInterpolatedString(i, e.conf.actionStr) | ||
| if err != nil { | ||
| return fmt.Errorf("interpolating action: %w", err) | ||
| } |
There was a problem hiding this comment.
You should be using batch.InterpolationExecutor and reusing the executors across the batch to prevent N copies of the batch (TryInterpolatedString makes a shallow copy of the batch every time it's called).
This new output uses the latest official Elasticsearch Go library.
esutil.NewBulkIndexerbecause it has a it's own internal concurrency and flushing logic that we don't really need since the benthos engine provides that for us.createaction because it is not idempotent which I don't think would play well with the assumptions that Connect makes of it's plugins.