How to setup data prepper so it will send data to OpenSearch data streams.

[FishyFishPat]

Hi there,

i am currently setting up OpenTelemetry nad OpenSearch integration and I have run into a problem. When I create a data stream template I get an error from data prepper upon startup and I can not seem to find an answer to my problem:

WARN  org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchSink - Failed to initialize OpenSearch sink, retrying: org.opensearch.client.opensearch._types.OpenSearchException: Request failed: [illegal_argument_exception] cannot create index with name [ss4o_metrics-otel], because it matches with template [metrics] that creates data streams only, use create data stream api instead

I have tried adding action:create in to data prepper configuration, but it doesn't work. I would really appreciate some help here.

Configuration:

pipelineConfig:
  enabled: true
  config:
    otel-logs-pipeline:
      workers: 8
      delay: 10
      source:
        otel_logs_source:
          ssl: false
      buffer:
        bounded_blocking:
          buffer_size: 256000
          batch_size: 12000
      sink:
        - opensearch:
            hosts: ["https://opensearch-cluster-masters:9200"]
            username: "admin"
            password: ""
            insecure: true
            index: "${/attributes/resource.attributes.k8s@namespace@name}-otel-logs"
            index_type: custom
            bulk_size: 10
    otel-trace-pipeline:
      workers: 8
      delay: "100"
      source:
        otel_trace_source:
          ssl: false
      buffer:
        bounded_blocking:
          buffer_size: 25600
          batch_size: 400
      sink:
        - pipeline:
            name: "raw-traces-pipeline"
        - pipeline:
            name: "otel-service-map-pipeline"
    raw-traces-pipeline:
      workers: 8
      delay: 3000
      source:
        pipeline:
          name: "otel-trace-pipeline"
      buffer:
        bounded_blocking:
          buffer_size: 25600
          batch_size: 400
      processor:
        - otel_trace_raw:
      sink:
        - opensearch:
            hosts: ["https://opensearch-cluster-masters:9200"]
            username: "admin"
            password: ""
            insecure: true
            index_type: trace-analytics-raw # fiksaj indexe
    otel-service-map-pipeline:
      workers: 8
      delay: 3000
      source:
        pipeline:
          name: "otel-trace-pipeline"
      processor:
        - service_map_stateful:
            window_duration: 180
      buffer:
        bounded_blocking:
          buffer_size: 25600
          batch_size: 400
      sink:
        - opensearch:
            hosts: ["https://opensearch-cluster-masters:9200"]
            username: "admin"
            password: ""
            insecure: true
            index_type: trace-analytics-service-map
            bulk_size: 10
    otel-metrics-pipeline:
      workers: 8
      delay: 3000
      source:
        otel_metrics_source:
          health_check_service: true
          ssl: false
      buffer:
        bounded_blocking:
          buffer_size: 32768
          batch_size: 2048
      processor:
        - otel_metrics:
            calculate_histogram_buckets: true
            calculate_exponential_histogram_buckets: true
            exponential_histogram_max_allowed_scale: 10
            flatten_attributes: false
      sink:
        - opensearch:
            hosts: ["https://opensearch-cluster-masters:9200"]
            username: "admin"
            password: ""
            insecure: true
            index: ss4o_metrics-otel-%{yyyy.MM.dd}
            index_type: custom
            bulk_size: 10
            template_type: index-template

Thank you.