Kafka, Logstash, Opensearch 설치 및 명령어

KwangYong·2024년 3월 4일

TFT

목록 보기

1/5

install

👀 가상환경에 Kafka, Logstash, Opensearch 설치

sudo apt-get install apt-transport-https ca-certificates curl gnupg-agent software-properties-common
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -
sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable"
sudo apt-get update && sudo apt-get upgrade
sudo apt-get install docker-ce docker-ce-cli containerd.io
sudo apt-get install docker-compose
docker-compose -v

mkdir /data

vi /data/logstash.yml

input {

        kafka {

                codec => plain

                bootstrap_servers =>  "kafka:9092"

                group_id => "logstash"

                topics => ["test_input"]

                consumer_threads => 1

        }

}

filter {

}



output {

        stdout {

                codec => plain

        }

        opensearch {

                hosts => "opensearch:9200"

                index => "kafka-test-%{+YYYY-MM-dd}"

        }

}


닫은 후 

vi /data/docker-compose.yml


version: '3'
services:
  opensearch:
    image: opensearchproject/opensearch:2.9.0
    container_name: opensearch
    restart: always
    environment:
      - cluster.name=opensearch-cluster
      - node.name=opensearch
      - discovery.seed_hosts=opensearch
        # - cluster.initial_cluster_manager_nodes=opensearch
      - bootstrap.memory_lock=true
      - "OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m"
      - discovery.type=single-node
    ulimits:
      memlock:
        soft: 0
        hard: 0
      nofile:
        soft: 65535
        hard: 65535
    volumes:
      - /data/opensearch/data:/usr/share/opensearch/data
    ports:
      - 9200:9200
      - 9600:9600

  logstash-oss:
     image: opensearchproject/logstash-oss-with-opensearch-output-plugin:7.13.2
     container_name: logstash-oss
     volumes:
      - /data/logstash/data:/usr/share/logstash/data
      - /data/logstash/pipeline/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
     ports:
       - 5044:5044
     restart: always
     environment:
       - ES_HOST=opensearch
     depends_on:
       - opensearch
     links:
       - opensearch

  zookeeper:
     image: confluentinc/cp-zookeeper:7.1.2
     hostname: zookeeper
     container_name: zookeeper
     restart: always
     ports:
       - "2181:2181"
     volumes:
       - /data/zookeeper/config/:/etc/kafka/
       - /data/zookeeper/log/:/var/lib/zookeeper/log/
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
  kafka:
     image: confluentinc/cp-kafka:7.1.2
     hostname: kafka
     container_name: kafka
     depends_on:
       - zookeeper
     ports:
       - "29092:29092"
       - "9092:9092"
       - "9101:9101"
     restart: always
     volumes: 
       - /data/kafka/topic/:/var/lib/kafka/data
       - /data/kafka/config/:/etc/kafka/       
     environment:
       KAFKA_BROKER_ID: 1
       KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://kafka:9092
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
       KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 1
       KAFKA_JMX_PORT: 9101
       KAFKA_JMX_HOSTNAME: kafka

  kafka-ui:
    image: provectuslabs/kafka-ui
    container_name: kafka-ui
    ports:
      - "8080:8080"
    restart: always
    environment:
      - KAFKA_CLUSTERS_0_NAME=kafka
      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
      - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2181

  opensearch-dashboards:
    container_name: opensearch-dashboards
    image: opensearchproject/opensearch-dashboards:2.9.0
    ports:
      - "5601:5601"
    environment:
      - OPENSEARCH_HOSTS=http://opensearch:9200
      - "DISABLE_SECURITY_DASHBOARDS_PLUGIN=true"
 
    depends_on:
      - opensearch
  
  redis-container:
    image: redis:7.0
    container_name: redis
    ports: 
      - "6379:6379"
    volumes:
      - ./redis/data:/data
      - ./redis/conf/redis.conf:/usr/local/conf/redis.conf
    restart: always
    command: redis-server /usr/local/conf/redis.conf

  mysql:
    image: mysql:8.0
    container_name: mysql
    ports:
      - 3306:3306 # HOST:CONTAINER
    environment:
      MYSQL_ROOT_PASSWORD: admin
    command:
      - --character-set-server=utf8mb4
      - --collation-server=utf8mb4_unicode_ci
    volumes:
      - ./mysql/data:/var/lib/mysql
      
      



cd /data
docker-compose up -d

👀 virtual box 포트포워딩

[가상이미지 마우스 오른쪽버튼] -> [설정] -> [네트워크] -> [고급] -> [포트 포워딩]

👀 opensearch ssl 플러그인 제거

Test 용도이므로 보안 관련 SSL 제거
1. opensearch 진입 후
2. cd plugin
3. rm -rf opensearch-security
4. cd ..
5. cd config
6. vi opensearch.yml에서 security 관련 코드 모두 제거

👀 kafka 토픽에 메세지 전송

    docker exec -it kafka kafka-console-producer --topic test_input --bootstrap-server kafka:9092

docker exec -it kafka는 Docker에서 실행 중인 'kafka'라는 이름의 컨테이너에 대한 인터랙티브 터미널 세션을 시작합니다.
kafka-console-producer는 Kafka의 콘솔 프로듀서 도구를 실행합니다.
-topic test_input은 메시지를 보낼 토픽을 지정합니다. 여기서는 'test_input'이라는 이름의 토픽입니다.
-bootstrap-server kafka:9092는 Kafka 클러스터에 연결하기 위한 부트스트랩 서버 주소와 포트를 지정합니다. 여기서는 'kafka:9092'를 사용합니다.

이 명령을 실행한 후에는 터미널에 메시지를 입력할 수 있으며, 각 줄이 입력될 때마다 해당 메시지가 'test_input' 토픽으로 전송됩니다.

👀 kafka 컨테이너에서 kafka 토픽과 메세지 확인

Kafka 컨테이너에 접근: 먼저, Kafka가 실행 중인 Docker 컨테이너로 접근합니다. 여기서 kafka는 Kafka 컨테이너의 이름입니다.
```
docker exec -it kafka /bin/bash
```
토픽리스트 확인 : Kafka 컨테이너 내에서 Kafka 토픽 리스트를 확인. 이 명령어는 Kafka 브로커에 접속하여 사용 가능한 토픽의 목록을 보여줍니다.
```
kafka-topics --list --bootstrap-server localhost:9092
```
특정 토픽의 메시지 확인: 특정 토픽 (예: test_input)에 대한 메시지를 확인하고 싶다면, kafka-console-consumer 명령어를 사용할 수 있습니다. 예를 들어,
```
 kafka-console-consumer --bootstrap-server localhost:9092 --topic test_input --from-beginning
```
명령을 사용하여 test_input 토픽의 메시지를 처음부터 확인할 수 있습니다.

이러한 절차를 통해 Kafka에서 test_input 토픽의 존재 여부와 메시지의 상태를 확인할 수 있습니다.

👀 Logstash 로그 분석

토픽 구독: Logstash가 Kafka의 test_input 토픽에 성공적으로 구독했습니다.

  [Consumer clientId=logstash-0, groupId=logstash] Subscribed to topic(s): test_input

그룹 코디네이터 발견: Kafka 그룹 코디네이터를 발견하고, 그룹에 가입하기 위해 재시도합니다.

[Consumer clientId=logstash-0, groupId=logstash] Discovered group coordinator kafka:9092 (id: 2147483646 rack: null)
[Consumer clientId=logstash-0, groupId=logstash] (Re-)joining group

그룹 가입 실패 및 재시도: 첫 번째 그룹 가입 시도가 실패했으나, Logstash는 재시도하여 성공적으로 그룹에 가입했습니다.

[Consumer clientId=logstash-0, groupId=logstash] Join group failed with org.apache.kafka.common.errors.MemberIdRequiredException: The group member needs to have a valid member id before actually entering a consumer group
[Consumer clientId=logstash-0, groupId=logstash] Successfully joined group with generation 9

파티션 할당: Logstash가 test_input-0 파티션을 할당받았습니다.

[Consumer clientId=logstash-0, groupId=logstash] Adding newly assigned partitions: test_input-0

오프셋 설정: Logstash가 해당 파티션의 오프셋을 설정했습니다. 이는 Logstash가 Kafka로부터 메시지를 읽고 있다는 것을 의미합니다.

[Consumer clientId=logstash-0, groupId=logstash] Setting offset for partition test_input-0 to the committed offset FetchPosition{offset=6, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[kafka:9092 (id: 1 rack: null)], epoch=0}}

이 로그들은 Logstash가 Kafka로부터 메시지를 성공적으로 수신하고 있음을 나타냅니다.

👀 Opensearch Dashboards

<인덱스 패턴 생성>
대시보드의 왼쪽 사이드바에서 "dashboard Management"를 찾아 클릭합니다.
"Index Patterns" 섹션을 선택합니다.
"Create index pattern"을 클릭합니다.
인덱스 패턴 이름으로 kafka-test-*를 입력하고 "Next step"을 클릭합니다.
시간 필터를 선택할 필요가 있으면, 데이터에 해당하는 시간 필드를 선택합니다. 시간 필터가 필요 없다면, 이 단계를 건너뛸 수 있습니다.
"Create index pattern"을 클릭하여 인덱스 패턴을 생성합니다.

<데이터 탐색>
왼쪽 사이드바에서 "Discover"를 선택합니다.
상단의 인덱스 패턴 드롭다운 메뉴에서 kafka-test-* 인덱스 패턴을 선택합니다.
이 화면에서 해당 인덱스의 문서들을 볼 수 있습니다. 여기서는 시간 범위를 조정하거나 특정 필드를 검색하여 데이터를 더 자세히 볼 수 있습니다.