[Logstash] multi pipeline 구성하기 + kafka 메세지를 logstash input으로 받기

728x90

개요

Spring boot 환경에서 Docker로 ELK 사용하기 에서 Docker 환경에서 ELK를 실행하였습니다.

기존에는 Logstash는 아래와 같이 단일 파이프라인으로 구성했었는데 멀티 파이프라인으로 수정해보도록 하겠습니다.

input {
  tcp {
    port => 5000
    codec => json_lines
  }
}

output {
  elasticsearch {
    hosts => "elasticsearch:9200"
    user => "elastic"
    password => "elastic123!@#"
    index => "logstash-%{+YYYY.MM.dd}"
  }
}

Logstash 멀티 파이프라인 설정을 통해 여러 독립적인 파이프라인을 한 Logstash 인스턴스에서 실행할 수 있습니다. 이 기능은 각각 다른 소스에서 오는 데이터를 처리하거나, 다양한 처리 과정을 별도로 관리해야 할 때 유용합니다. 멀티 파이프라인을 구성하는 방법은 다음과 같습니다.

Logstash 멀티 파이프라인 구성 방법

1. Logstash Kafka Input 플러그인 설정

kafka로부터 메세지를 받기 위해서 Logstash kafka input 플러그인이 필요합니다.

docker ps

docker ps 명령어를 입력하면 docker의 container를 확인할 수 있습니다. 여기서 Logstash의 container id를 확인합니다.

확인한 container id로 아래의 명령어를 실행하면 container의 shell에 접속할 수 있습니다.

docker exec -it <container_name_or_id> /bin/bash

그리고 아래의 명령을 실행하면 설치된 Logstash 플러그인 목록을 조회하고, 그 중에서 Kafka input 플러그인이 있는지 확인할 수 있습니다.

bin/logstash-plugin list | grep logstash-input-kafka
bin/logstash-plugin list | grep logstash-integration-kafka

둘 중에 하나라도 존재하면 kafka input 플러그인이 설치되있는 상태입니다. 만약 설치가 안되어있다면 아래의 명령어를 통해 설치를 합니다.

bin/logstash-plugin install logstash-input-kafka

2. 파이프라인 구성 파일 준비

먼저, 각 파이프라인에 대한 구성 파일을 준비합니다. 예를 들어, pipeline1.conf와 pipeline2.conf라는 두 개의 구성 파일을 가지고 있다고 가정합니다. 이 파일들은 Logstash 컨테이너 내부에 위치해야 합니다. 저는 pipeline1.conf는 was의 로그를 처리하고 pipeline2.conf는 kafka의 메세지를 처리하게 구성하였습니다.

pipeline1.conf

input {
  tcp {
    port => 5001
    codec => json_lines
  }
}

output {
  elasticsearch {
    hosts => "elasticsearch:9200"
    user => "elastic"
    password => "elastic123!@#"
    index => "waslog-%{+YYYY.MM.dd}"
  }
}

pipeline2.conf

input {
  kafka {
    bootstrap_servers => "host.docker.internal:9092"
    topics => "event" 
    consumer_threads => 1
    codec => "json"
  }
}

output {
  elasticsearch {
    hosts => "elasticsearch:9200"
    user => "elastic"
    password => "elastic123!@#"
    index => "kafka-event-%{+YYYY.MM.dd}"
  }
}

host.docker.internal은 Docker 호스트를 가리킵니다. ELK는 docker에서 실행되고 kafka는 docker 외부에서 실행되기 때문에 사용합니다.

3. pipelines.yml 파일 생성

Logstash의 멀티 파이프라인 설정을 위한 pipelines.yml 파일을 생성합니다. 이 파일은 위에서 작성한 파이프라인 구성 파일들을 참조합니다.

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

4. docker-compose.yml 파일 수정

docker-compose.yml 파일을 수정하여 Logstash 컨테이너가 멀티 파이프라인 구성을 사용하도록 설정합니다. 이 과정에서 필요한 구성 파일들을 Logstash 컨테이너 내에 마운트해야 합니다.

기존에 logstash 설정은 아래와 같이 되어있습니다.

# Logstash 설정
  logstash:
    container_name: logstash
    image: docker.elastic.co/logstash/logstash:7.10.1
    environment:
      - XPACK_MONITORING_ENABLED=false
    ports:
      - 5001:5001
      - 9600:9600
    volumes:
      - ./logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    depends_on: 
      - elasticsearch
    networks:
      - elk

아래와 같이 수정해줍니다.

# Logstash 설정
  logstash:
    container_name: logstash
    image: docker.elastic.co/logstash/logstash:7.10.1
    environment:
      - XPACK_MONITORING_ENABLED=false
    ports:
      - 5001:5001
      - 9600:9600
    volumes:
      - ./logstash/config/pipelines.yml:/usr/share/logstash/config/pipelines.yml
      - ./logstash/pipeline/pipeline1.conf:/usr/share/logstash/pipeline/pipeline1.conf
      - ./logstash/pipeline/pipeline2.conf:/usr/share/logstash/pipeline/pipeline2.conf
    depends_on: 
      - elasticsearch
    networks:
      - elk

volumes 섹션에서는 파이프라인 구성 파일들과 pipelines.yml 파일을 Logstash 컨테이너에 마운트합니다.

5. Docker Compose 실행

docker compose를 실행합니다.

docker-compose up -d

6. Kibana 인덱스 패턴 생성

Kibana 홈페이지에 접속해서 Manage 아이콘을 클릭합니다.

좌측의 Kibana > Index Patterns을 클릭하고 Create index pattern 버튼을 클릭합니다.

Index pattern의 이름을 입력하고 Next step 버튼을 클릭합니다.

Time field를 선택 후 Create index pattern 버튼을 클릭합니다.

각 index 별로 로그를 확인할 수 있습니다.

참고자료

https://investment-engineer.tistory.com/6