OpenSearch API source

The opensearch_api source accepts HTTP requests compatible with the OpenSearch Bulk API. Standard OpenSearch clients like Apache Flink OpenSearch Connector, Logstash, and opensearch-py can send data to Data Prepper through this source.

Unlike the OpenSearch Bulk API, this source writes asynchronously. Events are buffered for downstream processing rather than written directly to OpenSearch. When the zero buffer is used, events pass directly to the sink. Regardless of buffer type, the response is synthetic and does not reflect actual indexing results from OpenSearch.

Usage

The opensearch_api source supports the following endpoints:

POST /{path}/_bulk
POST /{path}/{index}/_bulk

Configuration

The following table lists the configuration options for the opensearch_api source.

Authentication

By default, the source runs without authentication. To make this explicit in your configuration, use the unauthenticated key:

source:
  opensearch_api:
    authentication:
      unauthenticated:

HTTP basic authentication

The following example configures HTTP basic authentication with a username and password:

source:
  opensearch_api:
    authentication:
      http_basic:
        username: my-user
        password: my_s3cr3t

Custom authentication

To provide custom authentication, create a plugin that implements the ArmeriaHttpAuthenticationProvider interface.

Event metadata attributes

For each action in the bulk request, the source creates a Data Prepper event with the following metadata attributes.

You can reference these attributes in processors and sinks using the getMetadata method or ${metadata_key} syntax.

Response format

On success, the source returns a JSON response with an empty items array:

{
  "took": 5,
  "errors": false,
  "items": []
}

The following table lists the response fields.

Response status codes

The following table lists the possible response status codes.

Buffer compatibility

The opensearch_api source works with the following buffer types.

Metrics

The opensearch_api source emits the following metrics.

Counters

The following table lists the counter metrics.

Timer

The following table lists the timer metric.

Distribution summary

The following table lists the distribution summary metric.

Examples

The following examples demonstrate common configurations for the opensearch_api source.

Basic pipeline with an OpenSearch sink

The following example configures a pipeline that receives bulk requests and writes events to an OpenSearch sink using metadata attributes:

opensearch-pipeline:
  source:
    opensearch_api:
      port: 9202
      path: "/opensearch"
  sink:
    - opensearch:
        hosts: ["https://opensearch-node:9200"]
        index: "${getMetadata(\"opensearch_index\")}"
        document_id: "${getMetadata(\"opensearch_id\")}"
        routing: "${getMetadata(\"opensearch_routing\")}"
        action: "${getMetadata(\"opensearch_action\")}"

SSL enabled

The following example enables SSL with HTTP basic authentication:

opensearch-pipeline:
  source:
    opensearch_api:
      port: 9202
      ssl: true
      ssl_certificate_file: "/certs/server.crt"
      ssl_key_file: "/certs/server.key"
      authentication:
        http_basic:
          username: admin
          password: admin
  sink:
    - opensearch:
        hosts: ["https://opensearch-node:9200"]
        index: "${getMetadata(\"opensearch_index\")}"
        action: "${getMetadata(\"opensearch_action\")}"

Sending data using cURL

The following example sends a bulk request to the opensearch_api source using cURL:

curl -XPOST "http://localhost:9202/opensearch/_bulk" \
  -H "Content-Type: application/json" \
  -d '{"index":{"_index":"movies","_id":"1"}}
{"title":"Rush","year":2013}
{"delete":{"_index":"movies","_id":"2"}}
'

Sending data using opensearch-py

The following example sends a bulk request using the opensearch-py client:

from opensearchpy import OpenSearch

client = OpenSearch(
    hosts=[{"host": "localhost", "port": 9202}],
    use_ssl=False,
    url_prefix="/opensearch"
)

client.bulk(body=[
    {"index": {"_index": "movies", "_id": "1"}},
    {"title": "Rush", "year": 2013},
])