Logstash not using ingest pipeline with datastreams

[rzisk (Ryan)]

Hello,

I recently got our Elasticsearch instances upgraded to the 8 branch am working on getting all of our Beats upgraded to 8 as well. So far there have been various gotchas with getting Logstash to play well with datastreams, it seems. I finally got indexing working by adding the "action" field manually. However there seems to be one last hurdle. I have loaded the templates and ingest pipelines for Winlogbeat but the events do not seem to be getting processed through the pipeline. I was able to manually test with an event already in Elasticsearch to confirm everything parsed correctly but it does not seem to be happening at ingest time.

Winlogbeat config:

logging:
  level: warning
output:
  logstash:
    backoff:
      init: 30s
      max: 300s
    enable: true
    hosts:
    - <logstash_url>
    loadbalance: true
    ssl:
      enable: true
path:
  config: C:\Program Files\Winlogbeat
  data: C:\Program Files\Winlogbeat\data
  home: C:\Program Files\Winlogbeat
  logs: C:\Program Files\Winlogbeat\logs
processors:
- add_host_metadata:
    netinfo:
      enabled: true
- add_locale: null
winlogbeat:
  event_logs:
  - ignore_older: 72h
    name: Application
  - ignore_older: 72h
    name: System
  - name: Microsoft-Windows-TerminalServices-LocalSessionManager/Operational
  - name: Microsoft-Windows-User Profile Service/Operational
  - name: Microsoft-Windows-TerminalServices-RDPClient/Operational
  - name: Security
  - ignore_older: 72h
    name: Microsoft-Windows-Sysmon/Operational
  - event_id: 400, 403, 600, 800
    name: Windows PowerShell
  - event_id: 4103, 4104, 4105, 4106
    name: Microsoft-Windows-PowerShell/Operational

Logstash output to Elasticsearch

output {
    if [@metadata][pipeline] {
        elasticsearch {
            hosts => ["<elastic_node>"]
            data_stream => "auto"
            action => "create"
            index => "%{[@metadata][beat]}-%{[@metadata][version]}"
            pipeline => "%{[@metadata][pipeline]}"
            user => "logstash_internal"
        }
   } else {
        elasticsearch {
            hosts => ["<elastic_node>"]
            data_stream => "auto"
            action => "create"
            index => "%{[@metadata][beat]}-%{[@metadata][version]}"
            user => "logstash_internal"
        }
    }
}

Am I missing some piece of configuration that would have changed from 7 to 8?

[rzisk (Ryan)]

I was able to confirm that the pipeline field isn't in the "@metadata" field. So I need to figure out where that is supposed to be added.

 "@metadata" => {
           "beat" => "winlogbeat",
        "version" => "8.2.3",
           "type" => "_doc",
          "input" => {
            "beats" => {
                "host" => {
                    "ip" => <ip>
                }
            }
        }
    }

[rzisk (Ryan)]

Adding the below filter to Logstash will get things into the correct ingest pipeline, but it doesn't seem like it should be needed.

filter {
 if [@metadata][beat] == "winlogbeat" and [@metadata][version] =~ /^8/ {
   mutate { add_field => { "[@metadata][pipeline]" => "winlogbeat-%{[@metadata][version]}-routing" } }
 }
}