[Logstash] Multi config, input, filter, output inside a Logstash server and Not conflict

Hướng dẫn chạy logstash trên docker

git clone https://github.com/deviantony/docker-elk.git
cd docker-elk/

Ở đây mình chỉ chạy logstash thôi nên sửa lại file docker-compose chút

version: '3.2'

services:

  logstash_openvpn:
    build:
      context: logstash/
      args:
        ELK_VERSION: $ELK_VERSION
    volumes:
      - type: bind
        source: ./logstash/config/logstash.yml
        target: /usr/share/logstash/config/logstash.yml
        read_only: true
      - type: bind
        source: ./logstash/config/pipelines.yml
        target: /usr/share/logstash/config/pipelines.yml
        read_only: true
      - type: bind
        source: ./logstash/pipeline
        target: /usr/share/logstash/pipeline
        read_only: true
    ports:
      - "5041:5044"
      - "3514:3514"
    environment:
      LS_JAVA_OPTS: "-Xmx256m -Xms256m"
    networks:
      - elk

networks:
  elk:
    driver: bridge

volumes:
  elasticsearch:

Ở đây mình custom port và “./logstash/config/pipelines.yml” để chạy multi file config pipeline trong đó có multi input, filter, output

Trong một lớp học elk của itforvn mình được chia sẽ làm sao để phân tách nhiều input, filter, output trong cùng 1 logstash thì có cách bạn có thể if else đặt thêm type cho kiểm nhận log:

https://thachnuida.com/2016/12/04/push-nginx-log-to-elasticsearch-by-using-logstash/

Nhưng mà mình thì muốn là tạo nhiều file rời rạc để quản lý thay vì viết 1 file quá dài

Chúng ta cùng tìm hiểu file pipelines.yml

# This file is where you define your pipelines. You can define multiple.
# For more information on multiple pipelines, see the documentation:
#   https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html

- pipeline.id: main
  path.config: "/usr/share/logstash/pipeline/logstash.conf"

- pipeline.id: syslog
  path.config: "/usr/share/logstash/pipeline/syslog.conf"

Ở đây mình chia thành 2 pipeline.id khách nhau và các config trong logstash.conf sẽ không hề ảnh hướng đến syslog.conf

file mặc định sẽ dưới đây

# This file is where you define your pipelines. You can define multiple.
# For more information on multiple pipelines, see the documentation:
#   https://www.elastic.co/guide/en/logstash/current/multiple-pipelines.html

- pipeline.id: main
  path.config: "/usr/share/logstash/pipeline/"

Các file trong thư mục /usr/share/logstash/pipeline/ thì chạy chung 1 pipeline.id nên nếu có nhiều file config thì chúng sẽ ghi đè lên nhau và rất dễ sung đột hoặc mất log cần thiết.

Điều này là rất cần thiết vì 1 container logstash khá là nặng:

Chúng ta cần tận dụng tối đa khả năng multiple của logstash

Example:

Ví dụ “conflict” điển hình là khi bạn đặt 2 bộ input/filter/output cho 2 loại log khác nhau vào cùng thư mục conf.d, Logstash sẽ chạy chúng trong một pipeline và đôi lúc event đi nhầm output (ví dụ firewall log vào index Java).​

Ví dụ conflict (Cách 1)

Giả sử bạn có 2 file cấu hình:

java.conf

input { beats { port => 5044 } }

filter {
  # parse java stacktrace...
}

output {
  elasticsearch { index => "java-stacktrace-%{+YYYY.MM.dd}" }
}

firewall.conf

input { udp { port => 5514 } }

filter {
  # parse cisco asa...
}

output {
  elasticsearch { index => "cisco-asa-%{+YYYY.MM.dd}" }
}

Khi chạy cùng lúc, bạn có thể thấy “firewall logs in the java stack trace index.. or vice versa” (log bị lẫn index) đúng như tình huống người dùng mô tả.​

Vì sao bị lẫn

Khi bạn đưa Logstash một thư mục config, nó load các file theo thứ tự alphabet; nhưng quan trọng hơn là việc bạn tách input/filter/output ra nhiều file hay gộp vào một file “không làm chúng thành pipeline riêng”—các stanza vẫn thuộc cùng một pipeline, nên output nào không có điều kiện có thể “ăn” mọi event.​

Nói ngắn gọn: nếu output { elasticsearch { index => ... } } không có if/tags/field để ràng buộc, thì event từ input A cũng có thể đi vào output B, gây “mix output”.

Nếu bạn không chia thành các file thì trong các input bạn phải add field để đánh dấu:

input {
  file {
    path => "/var/log/nginx/access.log"
    type => "nginx"                    # ← Gắn type = "nginx"
  }
  file {
    path => "/var/log/syslog"
    type => "syslog"                   # ← Gắn type = "syslog"
  }
}

Event từ /var/log/nginx/access.log sẽ có:

{
  "@timestamp": "2026-02-12T08:46:00.000Z",
  "message": "192.168.1.1 - - [12/Feb/2026:15:46:00 +0700] \"GET / HTTP/1.1\" 200 1234",
  "path": "/var/log/nginx/access.log",
  "type": "nginx",                     // ← Field tự động thêm
  "host": "your-server"
}

Sau đó bạn dùng if [type] == "..." để phân luồng

filter {
  if [type] == "nginx" {
    grok { match => { "message" => "%{COMBINEDAPACHELOG}" } }
  }
  else if [type] == "syslog" {
    grok { match => { "message" => "%{SYSLOGLINE}" } }
  }
}

output {
  if [type] == "nginx" {
    elasticsearch { index => "nginx-logs-%{+YYYY.MM.dd}" }
  }
  else if [type] == "syslog" {
    elasticsearch { index => "syslog-logs-%{+YYYY.MM.dd}" }
  }
}