Chaining Piplines Example

# pipelines.yml

- pipeline.id: pipeline-1
  path.config: "/etc/logstash/pipelines/p1.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-2
  path.config: "/etc/logstash/pipelines/p2.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-3
  path.config: "/etc/logstash/pipelines/p3.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-4
  path.config: "/etc/logstash/pipelines/p4.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-5
  path.config: "/etc/logstash/pipelines/p5.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-6
  path.config: "/etc/logstash/pipelines/p6.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-7
  path.config: "/etc/logstash/pipelines/p7.conf"
  pipeline.workers: 1
- pipeline.id: pipeline-8
  path.config: "/etc/logstash/pipelines/p8.conf"
  pipeline.workers: 1
  
  
# p1.conf

input {
    http {
        host => "127.0.0.1"
        port => 8080
    }
}
filter{
    mutate {
        add_field => {
            'rule_matched' => 'NONE'
        }
    }

    if [device] == 'device_0' and [network] == 'network_b' {
        mutate {
            update => {
                'rule_matched' => 'rule_0'
            }
        }
    }


    if [device] == 'device_1' and [network] == 'network_b' {
        mutate {
            update => {
                'rule_matched' => 'rule_1'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8081
      }
}

# p2.conf

input {
  tcp {
    host => "127.0.0.1"
    port => 8081
  }
}

filter{

    if [device] == 'device_2' and [network] == 'network_b' {
        mutate {
            update => {
                'rule_matched' => 'rule_2'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8082
      }
}

# p3.conf

input {
  tcp {
    host => "127.0.0.1"
    port => 8082
  }
}

filter{

    if [device] == 'device_3' and [network] == 'network_b' {
        mutate {
            update => {
                'rule_matched' => 'rule_3'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8083
      }
}

# p4.conf

input {
  tcp {
    host => "127.0.0.1"
    port => 8083
  }
}

filter{

    if [start_time] > '1491177600' and [end_time] < '1501632000' and [device] == 'device_4' {
        mutate {
            update => {
                'rule_matched' => 'rule_4'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8084
      }
}

# p5.conf
input {
  tcp {
    host => "127.0.0.1"
    port => 8084
  }
}

filter{

    if [start_time] > '1491177600' and [end_time] < '1501632000' and [device] == 'device_5' {
        mutate {
            update => {
                'rule_matched' => 'rule_5'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8085
      }
}

# p6.conf

input {
  tcp {
    host => "127.0.0.1"
    port => 8085
  }
}

filter{

    if [start_time] > '1491177600' and [end_time] < '1501632000' and [device] == 'device_6' {
        mutate {
            update => {
                'rule_matched' => 'rule_6'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8086
      }
}

# p7.conf

input {
  tcp {
    host => "127.0.0.1"
    port => 8086
  }
}

filter{

    if ([duration] > 700 and [network] == 'network_a') or [device] == 'device_7' {
        mutate {
            update => {
                'rule_matched' => 'rule_7'
            }
        }
    }


}

output {
      tcp {
        host => "127.0.0.1"
        port => 8087
      }
}

# p8.conf

input {
  tcp {
    host => "127.0.0.1"
    port => 8087
  }
}

filter{

    if ([duration] > 700 and [network] == 'network_a') or [device] == 'device_8' {
        mutate {
            update => {
                'rule_matched' => 'rule_8'
            }
        }
    }

    if ([duration] > 700 and [network] == 'network_a') or [device] == 'device_9' {
        mutate {
            update => {
                'rule_matched' => 'rule_9'
            }
        }
    }

    ruby { code => "event.set('time_taken', (Time.now.to_i - event.get('creation_time')))" }

}

output {
    file {
        path => "/tmp/test.log"
        codec => line { format => "%{time_taken} seconds, %{device}, %{rule_matched}"  }
    }
}