[IoT] Mqtt broker - Kafka - ELK 스택 셋팅

구조도

구성품

• IoT devices: 라즈베리파이 등 MQTT Client
• MQTT Broker: EMQ (https://www.emqx.io)
• Kafka producer: Kafka Junction (https://github.com/brycemcd/MQTT-to-Kafka)
• Kafka consumer: Logstash + kafka input plugin + elasticsearch output plugin
• Kafka (+ Zookeeper)
• Elastic stack

프로세스

1. IoT 기기에서 MQTT 프로토콜로 MQTT Broker 에 hello-kafka 토픽에 데이터 전송(=pub).
   

2. Kafka Producer 인 Kafka bridge 에서 MQTT Broker (종류 중 Emqx 사용함)의 hello-kafka 토픽을 sub 하고 있는 상태이며,
   pub 된 데이터를 kafka cluster 의 hello-kafka-cluster 토픽으로 전달(=produce).

3. kafka cluster의 hello-kafka-cluster 토픽에 Kakfa consumer (종류 중 Logstash 사용함)를 걸어놓았으므로,
   kafka cluster 의 해당 토픽으로부터 데이터를 꺼내서 Elasticsearch 로 던져줌(=consume).

4. Kibana 의 Timelion 그래프로 Elasticsearch 에 쌓인 IoT 데이터 1초 단위 확인.
   
   

Kibana

Timelion 쿼리 예: .es(metric=avg:column1, interval='1s')

Docker 셋팅

#

# https://hub.docker.com/r/wurstmeister/kafka/

# https://better-coding.com/building-apache-kafka-cluster-using-docker-compose-and-virtualbox/

# https://gist.github.com/everpeace/7a317860cab6c7fb39d5b0c13ec2543e

# https://github.com/zoidbergwill/docker-compose-kafka/blob/master/docker-compose.yml

# https://www.kaaproject.org/kafka-docker/

#

version: '3.3'

services:

  emq:

    image: emqx/emqx:latest

    ports:

      - "1883:1883"

      - "8083:8083"

      - "18083:18083"

  kafka1:

    image: wurstmeister/kafka

    healthcheck:

      test: ps augwwx | egrep [S]upportedKafka

    depends_on:

      - zk1

      - zk2

      - zk3

    volumes:

      - "./kafka1/mqtt.properties:/opt/kafka/config/mqtt.properties"

      - "./kafka-connect/kafka-connect-mqtt-1.0-SNAPSHOT.jar:/opt/kafka/libs/kafka-connect-mqtt-1.0-SNAPSHOT.jar"

      - "./kafka-connect/kafka-connect-mqtt-1.2.1.jar:/opt/kafka/libs/kafka-connect-mqtt-1.2.1.jar"

      - "./kafka-connect/org.eclipse.paho.client.mqttv3-1.0.2.jar:/opt/kafka/libs/org.eclipse.paho.client.mqttv3-1.0.2.jar"

    ports:

      - "9091:9092"

    environment:

      KAFKA_BROKER_ID: 1

      KAFKA_CREATE_TOPICS: hello-mqtt:1:2,hello-mqtt-kafka:1:2,test-topic:1:2

      KAFKA_ZOOKEEPER_CONNECT:  zk1:2181,zk2:2181,zk3:2181

      KAFKA_ADVERTISED_HOST_NAME: 192.168.35.106

      KAFKA_ADVERTISED_PORT: 9091

      KAFKA_LOG_RETENTION_HOURS: "168"

      KAFKA_LOG_RETENTION_BYTES: "100000000"

  kafka2:

    image: wurstmeister/kafka

    healthcheck:

      test: ps augwwx | egrep [S]upportedKafka

    depends_on:

      - zk1

      - zk2

      - zk3

    volumes:

      - "./kafka2/mqtt.properties:/opt/kafka/config/mqtt.properties"

      - "./kafka-connect/kafka-connect-mqtt-1.0-SNAPSHOT.jar:/opt/kafka/libs/kafka-connect-mqtt-1.0-SNAPSHOT.jar"

      - "./kafka-connect/kafka-connect-mqtt-1.2.1.jar:/opt/kafka/libs/kafka-connect-mqtt-1.2.1.jar"

      - "./kafka-connect/org.eclipse.paho.client.mqttv3-1.0.2.jar:/opt/kafka/libs/org.eclipse.paho.client.mqttv3-1.0.2.jar"

    ports:

      - "9092:9092"

    environment:

      KAFKA_BROKER_ID: 2

      KAFKA_ZOOKEEPER_CONNECT:  zk1:2181,zk2:2181,zk3:2181

      KAFKA_ADVERTISED_HOST_NAME: 192.168.35.106

      KAFKA_ADVERTISED_PORT: 9092

      KAFKA_LOG_RETENTION_HOURS: "168"

      KAFKA_LOG_RETENTION_BYTES: "100000000"

  kafka3:

    image: wurstmeister/kafka

    healthcheck:

      test: ps augwwx | egrep [S]upportedKafka

    depends_on:

      - zk1

      - zk2

      - zk3

    volumes:

      - "./kafka3/mqtt.properties:/opt/kafka/config/mqtt.properties"

      - "./kafka-connect/kafka-connect-mqtt-1.0-SNAPSHOT.jar:/opt/kafka/libs/kafka-connect-mqtt-1.0-SNAPSHOT.jar"

      - "./kafka-connect/kafka-connect-mqtt-1.2.1.jar:/opt/kafka/libs/kafka-connect-mqtt-1.2.1.jar"

      - "./kafka-connect/org.eclipse.paho.client.mqttv3-1.0.2.jar:/opt/kafka/libs/org.eclipse.paho.client.mqttv3-1.0.2.jar"

    ports:

      - "9093:9092"

    environment:

      KAFKA_BROKER_ID: 3

      KAFKA_ZOOKEEPER_CONNECT:  zk1:2181,zk2:2181,zk3:2181

      KAFKA_ADVERTISED_HOST_NAME: 192.168.35.106

      KAFKA_ADVERTISED_PORT: 9093

      KAFKA_LOG_RETENTION_HOURS: "168"

      KAFKA_LOG_RETENTION_BYTES: "100000000"

  zk1:

    image: elevy/zookeeper:latest

    environment:

      MYID: 1

      SERVERS: zk1,zk2,zk3

    ports:

      - "2181:2181"

      - "2888"

      - "3888"

    healthcheck:

      test: echo stat | nc localhost 2181

      interval: 10s

      timeout: 10s

      retries: 3

  zk2:

    image: elevy/zookeeper:latest

    environment:

      MYID: 2

      SERVERS: zk1,zk2,zk3

    ports:

      - "2182:2181"

      - "2888"

      - "3888"

    healthcheck:

      test: echo stat | nc localhost 2181

      interval: 10s

      timeout: 10s

      retries: 3

  zk3:

    image: elevy/zookeeper:latest

    environment:

      MYID: 3

      SERVERS: zk1,zk2,zk3

    ports:

      - "2183:2181"

      - "2888"

      - "3888"

    healthcheck:

      test: echo stat | nc localhost 2181

      interval: 10s

      timeout: 10s

      retries: 3

  kafka-manager:

    image: sheepkiller/kafka-manager:latest

    depends_on:

      - kafka1

      - kafka2

      - kafka3

      - zk1

      - zk2

      - zk3

    links:

      - zk1

      - zk2

      - zk3

    ports:

      - "9000:9000"

    environment:

      ZK_HOSTS: zk1:2181,zk2:2181,zk3:2181

      APPLICATION_SECRET: letmein

      KM_ARGS: -Djava.net.preferIPv4Stack=true

  kafka-bridge:

    build: ./mqtt-to-kafka

    # NOTE: adding -u unbuffers STDIN/STDOUT and makes print statements

    # available in the logs

    command: ["python", "-u", "message_intercept.py"]

    volumes:

      - ./mqtt-to-kafka:/app

    depends_on:

      - kafka1

      - kafka2

      - kafka3

      - emq

    environment:

      - KAFKA_HOSTS="kafka1,kafka2,kafka3"

    restart: unless-stopped

  elasticsearch:

    build:

      context: elk/elasticsearch/

      args:

        ELK_VERSION: $ELK_VERSION

    volumes:

      - ./elk/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml:ro

      - ./elk/elasticsearch/data:/usr/share/elasticsearch/data

    ports:

      - "9200:9200"

      - "9300:9300"

    environment:

      ES_JAVA_OPTS: "-Xmx256m -Xms256m"

    healthcheck:

      test: echo stat | nc localhost 9200

      interval: 10s

      timeout: 10s

      retries: 3

  logstash:

    build:

      context: elk/logstash/

      args:

        ELK_VERSION: $ELK_VERSION

    volumes:

      - ./elk/logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml:ro

      - ./elk/logstash/pipeline:/usr/share/logstash/pipeline:ro

    ports:

      - "5000:5000"

      - "9600:9600"

    environment:

      LS_JAVA_OPTS: "-Xmx256m -Xms256m"

    depends_on:

      - elasticsearch

  kibana:

    build:

      context: elk/kibana/

      args:

        ELK_VERSION: $ELK_VERSION

    volumes:

      - ./elk/kibana/config/:/usr/share/kibana/config:ro

    ports:

      - "5601:5601"

    depends_on:

      - elasticsearch

    healthcheck:

      test: echo stat | nc localhost 5601

      interval: 10s

      timeout: 10s

      retries: 3

  grafana:

    image: grafana/grafana

    ports:

      - "3000:3000"

    depends_on:

      - elasticsearch

    healthcheck:

      test: echo stat | nc localhost 3000 

      interval: 10s

      timeout: 10s

      retries: 3

Docker 셋팅 사례

emq-kafka-elk.zip

다운받은 파일을 특정 폴더를 만들어 압축을 풀어주시고
콘솔에서 docker-compose up -d 명령 실행.

단일 노드(=호스트)에 docker-compose 를 사용하여 docker container 들로
서비스들 띄워지도록 셋팅함.

Web UI

• EMQ
• localhost:18083
• Kafka manager
• localhost:9000
• Kibana
• localhost:5601
• Grafana
• localhost:3000