Forked Pipelines

[BeMoore]

Hi all,

I'm working on some forked pipelines and im getting a little confused on how this should be working... and this is sort of a continuation from this thread Single data source split to different indexes

So the examples taken from here

pipeline.id: intake
  queue.type: persisted
  config.string: |
    input { beats { port => 5044 } }
    output { pipeline { send_to => ["internal-es", "partner-s3"] } }
- pipeline.id: buffered-es
  queue.type: persisted
  config.string: |
    input { pipeline { address => "internal-es" } }
    # Index the full event
    output { elasticsearch { } }
- pipeline.id: partner
  queue.type: persisted
  config.string: |
    input { pipeline { address => "partner-s3" } }
    filter {
      # Remove the sensitive data
      mutate { remove_field => 'sensitive-data' }
    }
    output { s3 { } } # Output to partner's bucket

So the issue thats confusing me is this:

config.string: |
input { beats { port => 5044 } }
output { pipeline { send_to => ["internal-es", "partner-s3"] } }

specifically the input line.

i can't find anything that points me to what sort of inputs i can declare here. i'm trying the piplines.yml below and forking the data to two separate pipelines

- pipline.id: main 
  path.config: "C:\logstash-7.5.2\logstash-7.5.2\config\conf.d\syslog.conf"
  config.string: output { pipeline { send_to => [fortigate],[mswinevent] } }
  pipeline.workers: 1
  -pipline.id: fortigate 
  config.string: input { pipeline { address => fortigate } }
  path.config: "C:\logstash-7.5.2\logstash-7.5.2\config\conf.d\fortigate\fortigate.conf"
  pipeline.workers: 1
  - pipline.id: mswinevent
  config.string: input { pipeline { address => mswinevent } }
  path.config: "C:\logstash-7.5.2\logstash-7.5.2\config\conf.d\mswinevent\mswinevent.conf"
  pipeline.workers: 1

the logic to this is:
I retain a full copy of all the data via the syslog inbound channel but its split to fortigate and mswinevents for further processing to separate indexes and will drop anything not applicable

main ---> ES/syslog.conf
|--->fortigate---> fortigate.conf -----> ES/fortigate.index
|---> mswinevent----> ES/mswinevent.index

I've been playing about with the inputs/outputs and the only thing that happens is that the syslog data is processed and nothing hits fortigate or mswinevents pipelines.

so in the syslog.conf

input {
  udp {
    port => 5000
    type => syslog
  }
}
filter {
}
output {
  elasticsearch {
    hosts => ["http://192.168.170.155:9200"]
	index => "syslog"
    pipeline => "fortigate"
	}
}

i've tried:

    pipeline => "[fortigate]"
    pipeline => "%{fortigate}"
    pipeline => "${[fortigate]}"
    pipeline => "fortigate"

none of this works..... is there and easier way of either splitting the feeds at the pipeline.yml stage or ingesting via the logstash syslog.cong, processing then spitting it out to ES and another pipeline?

[Badger]

The pipeline option on an elasticsearch output refers to an elasticsearch ingestion pipeline, not to a logstash pipeline.

I have not tested whether you can have both a path.config and a config.string for a pipeline in pipelines.yml, but I would expect one or the other will be ignored.