[BUG] pull-based ingestion with Kafka is "missing _source field" with OpenTelemetry as a source

[timojohlo]

Describe the bug

I am trying out the pull-based ingestion with Kafka described here.
The ingestion works with sample log lines, but when I try to ingest data that is pushed from OpenTelemetry to Kafka and then ingested by OpenSearch, I am getting the following error in OpenSearch:

[ERROR][o.o.i.p.DropIngestionErrorStrategy] [opensearch-cluster-master-2] Error processing update from kafka: java.lang.IllegalArgumentException: Missing _source field. Invalid message

The OpenTelemetry config for the kafka exporter consists of:

    exporters:
      kafka/test:
        brokers:
        - kafka:9092
        encoding: otlp_json
        topic: otel-logs

I believe, this _source field has to be set up by OpenSearch and is not something that has to be configured in the OpenTelemetry config?

Related component

Plugins

To Reproduce

1. Install OpenSearch opensearch-3.2.1 Instance in kubernetes cluster
   helm install opensearch-kafka opensearch/opensearch --values values-opensearch.yaml --set "opensearch.opensearchPlugins[0]=ingestion-kafka"
2. Install OpenSearch dashboard opensearch-dashboards-3.2.2 in kubernetes cluster
   helm install opensearch-kafka-dashboards opensearch/opensearch-dashboards
3. Set up Kafka kafka-32.4.3 in kubernetes cluster

helm install kafka \
  --set replicaCount=3 \
  --set kraft.enabled=true \
  --set service.ports.client=9092 \
  --set listeners.client.protocol=PLAINTEXT \
  --set controller.replicaCount=3 \
  --set "persistence.enabled=false" \
  --set "provisioning.enabled=true" \
  --set "provisioning.topics[0].partitions=3" \
  --set "provisioning.topics[0].replicationFactor=3" \
  --set "provisioning.topics[0].name=otel-logs

4. Install OpenTelemetry in kubernetes cluster
   helm install opentelemetry-operator open-telemetry/opentelemetry-operator
5. Follow the description to set up the pull-based index here
6. Check error logs with kubectl e.g.
   kubectl logs opensearch-cluster-master-2

Expected behavior

Logs from OpenTelemetry can be pulled into OpenSearch.

Additional Details

Plugins

• ingestion-kafka

Screenshots
not applicable

Host/Environment (please complete the following information):

• helm chart opensearch-dashboards-3.2.2
• helm chart opensearch-3.2.1
• helm chart opentelemetry-collector-contrib:0.133.0
• helm chart opentelemetry-operator:0.93.0

Additional context
no other context.

[andrross]

@timojohlo The messages consumed from Kafka must conform to this format: https://docs.opensearch.org/latest/api-reference/document-apis/pull-based-ingestion/#message-format I don't believe this feature has been designed to consume messages directly generated by the OTel collector. Is that what you're trying to do here?

FYI @varunbharadwaj @yupeng9

[timojohlo]

Thank you @andrross!

Yes, this is what I am trying to do.
In the docs you linked, it says only _source is required. From my understanding _source is created on OpenSearch-side when data is ingested as JSON, which should be the case for OpenTelemetry logs. So I am wondering what exactly is missing here?

[varunbharadwaj]

Hi @timojohlo
The pull-based ingestion flow works differently than traditional push-based OpenSearch flows. It expects messages to be written to Kafka in the format specified in the docs (as mentioned by Andrew). The message payload (fields you want to index) go into the _source field, and they will be indexed. I'm not exactly sure how the current message looks like, but we might have to map it to the expected format for it to work.

[andrross]

From my understanding _source is created on OpenSearch-side when data is ingested as JSON

_source is a field you must write into the Kafka payload. It is the document to index into OpenSearch. It is not created by OpenSearch. This feature as implemented cannot natively ingest OpenTelemetry data so as @varunbharadwaj mentioned there will need to be something to do the conversion.

[timojohlo]

Ok, understood. Thank you for the clarification @andrross @varunbharadwaj!
I will look into ways of converting and close this issue for now.

Possible solution in the future could be this for ingestion as well:

OpenSearch should expose native OTLP/gRPC and OTLP/HTTP endpoints.

open-telemetry/opentelemetry-collector-contrib#39707 (comment)

[andrross]

@timojohlo Have you looked at DataPrepper? It does offer native endpoints for OpenTelemetry. For example: https://docs.opensearch.org/latest/data-prepper/pipelines/configuration/sources/otel-logs-source/

[ps48]

@timojohlo Thanks for checking this mode of OTel Logs ingestions. I have a couple of questions:

1. Did you try out adding an OTel proccessor to convert your docs schema into the one required by OpenSearch Kafka endpoints?
2. We do have other alternatives in Data prepper Otel log source as mentioned above by @andrross and OTLP source
3. Moreover, you can use the OpenSearch exporter itself from OTel collector contrib to send logs directly into OpenSearch

[timojohlo]

@andrross
Yes, we are currently evaluating several options for scaling our OTel infrastructure (queing, buffering) with DataPrepper being one of the options.

@ps48 Thanks for reaching out, much appreciated!

Did you try out adding an OTel proccessor to convert your docs schema into the one required by OpenSearch Kafka endpoints?

OpenSearch requires the _source field on top-level and AFAIK the processor can not manipulate or add top-level fields, only sub-properties of body, attributes, resources such as body.name, attributes.id etc. can be added?

We do have other alternatives in Data prepper Otel log source as mentioned above by @andrross and OTLP source

Thank you, I will look into it.

Moreover, you can use the OpenSearch exporter itself from OTel collector contrib to send logs directly into OpenSearch

Yes, we are already using this extensively here https://github.com/cloudoperators/greenhouse-extensions/tree/main/logs, but we are missing functionalities such as buffering and queuing where Kafka seems like a good fit.

[andrross]

@ps48 Can you look into whether its possible to make the OTel collector output a compatible format? If this isn't easy to do, then @varunbharadwaj I think we should investigate how to make this work on the OpenSearch side. This feels like an obvious use case for pull-based ingestion.