Data Prepper requires OpenSearch is available or will fail

[dlvenable]

Is your feature request related to a problem? Please describe.

A Data Prepper pipeline with the OpenSearch sink requires that OpenSearch be running and reachable in order to run. This is a rather confusing user experience. It is unclear if OpenSearch is unreachable in general, or just starting up.

Describe the solution you'd like

Data Prepper pipelines with OpenSearch as a sink should be able to start and wait for the OpenSearch sink rather than fail.

One approach that could work well is to have a concept of an un-started pipeline. Data Prepper can create the different sinks and then wait for them to be ready before starting up the sources. In this way, Data Prepper won't accept data in the source until the sink is ready.

This would probably require adding a field to Sink:

boolean isReady();

Describe alternatives you've considered (Optional)

Data Prepper could just run and let the buffer fill up until the sink is ready. This is simpler than the concept proposed above.

Additional context

Here is a forum post with a user of Data Prepper who was help up because of the current behavior.

https://discuss.opendistrocommunity.dev/t/data-preeper-plugin/8319

[chenqi0805]

I think we need to

• add connection healthcheck into initialize() before doing other steps that modifies the state of opensearch backend
• move initialize method out of the opensearch sink constructor and add the infra at the data-prepper-core to support its invocation before execute API: https://github.com/opensearch-project/data-prepper/blob/b29b3487c115ed7aba7719f5d48053e363d42ea3/data-prepper-plugins/opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java#L89

[dlvenable]

Definitely moving the connection initialization out of the constructor is critical.

Other sinks may have similar issues as well. Data Prepper Core should be able to allow a Sink to try repeatedly to initialize or setup. And the different Sinks should be able to do this in parallel. So we may need something a little different from void initialize().

[kkondaka]

Buffering until sink is ready has the downside that the sink might not ever be ready and the buffer space is taken forever. I think it is more safer to drop everything while the sink is not ready. Basically tail dropping vs head dropping. If we are ok with head-dropping then initialize() can be done in doOutput() if it's not already done.

[dlvenable]

I think we should probably rename initialize to start to better match the existing Source plugin.

When Data Prepper is starting up, it can perform the following steps:

1. Create the sinks and sources.
2. Call start() on the sinks.
3. Wait for all sinks to report ready.
4. Call start(Buffer) on the source.

If not all of the sinks are ready after some timeout, then Data Prepper can fail. This could be a configurable option in the data-prepper-config.yaml file.

For example:

source_start_timeout: PT30S