Filebeat5+Kafka+ELK Docker是怎么搭建日志系统

docker pull jeygeethan/zookeeper-cluster

docker run --name docker-zk -d --restart=always \--net="host" \-p 2181:2181 \-p 2888:2888 \-p 3888:3888 \-v ~/dockerdata/zookeeper/lib:/var/lib/zookeeper \-v ~/dockerdata/zookeeper/log:/var/log/zookeeper \-v /etc/localtime:/etc/localtime:ro \jeygeethan/zookeeper-cluster 192.168.0.1,192.168.0.2,192.168.0.3 1 {1/2/3: 三个节点分别设置} 

运行 docker exec -it docker-zk bash

运行 bin/zkCli.sh

创建节点 create /nicholas "nicholas"查看节点 get /nicholas 显示创建成功，     在虚机B、C上执行get操作检查下新的节点是否已同步，可见则成功。

docker pull jeygeethan/kafka-cluster

docker run --name docker-kafka -d -p 9092:9092-e KAFKA_HOST=192.168.0.1-e KAFKA_PORT=9092-e ZOOKEEPER_CONNECT=192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181-e KAFKA_ID=0 {0/1/2: 三个节点分别设置，从0开始} -v ~/dockerdata/kafka/logs:/tmp/kafka-logsjeygeethan/kafka-cluster

进入docker  docker exec -it docker-kafka bash转换目录  cd /opt/kafka_2.11-0.9.0.1/bin创建Topic  ./kafka-topics.sh --create --topic TP_NIC --partitions 4 --replication-factor 2         --zookeeper 192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181查看Topic  ./kafka-topics.sh --describe --topic TP_NIC          --zookeeper 192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181在broker0（虚机A）上生产消息  ./kafka-console-producer.sh --topic=TP_NIC \          --broker-list=192.168.0.1:9092,192.168.0.2:9092,192.168.0.3:9092在broker1（虚机B）上消费消息  ./kafka-console-consumer.sh --topic=TP_NIC \          --zookeeper 192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181

docker pull sebp/elk:es240_l240_k460

chown -R 991:991 ~/dockerdata/es && chown -R 992:992 ~/dockerdata/logstash && chown -R 993:993 ~/dockerdata/kibana

vi ~/dockerdata/es/config/elasticsearch.yml编辑内容如下cluster.name: mm-clusternode.name: mm-node-01node.master: falsenode.data: true#restrict outside accessnetwork.host: 192.168.0.11transport.tcp.port: 9300http.port: 9200path.data: /etc/elasticsearch/datapath.work: /etc/elasticsearch/workpath.logs: /etc/elasticsearch/logspath.plugins: /etc/elasticsearch/pluginsbootstrap.mlockall: truediscovery.zen.ping.multicat.enabled: falsediscovery.zen.fd.ping_timeout: 100s#discovery.zen.fd.ping_retries: 6#discovery.zen.fd.ping_interval: 30sdiscovery.zen.ping.timeout: 100sdiscovery.zen.minimum_master_nodes: 1discovery.zen.ping.unicast.hosts: ["192.168.0.11", "192.168.0.12", "192.168.0.13"]gateway.recover_after_nodes: 2#action.auto_create_index: falseindex.number_of_replicas: 0index.number_of_shards: 2

vi ~/dockerdata/kibana/config/kibana.yml检查 elasticsearch.url: "http://localhost:9200" 对应上了即可

新增Kafka的input配置文件：vi ~/dockerdata/logstash/config/03-kafka-input.conf------------------------------------------input {    kafka {        zk_connect => "192.168.0.1:2181,192.168.0.2:2181,192.168.0.3:2181 "        #group_id => ""        topic_id => "syslog"        codec => "plain"        reset_beginning => false        consumer_threads => 5        decorate_events => true        add_field => { "[@metadata][type]" => "syslog" }    }}修改日志解析过滤配置文件：vi ~/dockerdata/logstash/config/10-syslog.conf------------------------------------------filter {  if [@metadata][type] in ["syslog","accesslog"] {    ruby {      code => "event['mlogsendts'] = event['@timestamp']"    }    mutate {      add_field => ["mlogsendts_string", "%{@timestamp}"]    }    json {      source => "message"      add_field => {        "mlogmsg" => "%{message}"      }      remove_field => [ "message"]    }    grok {      patterns_dir => ["/opt/logstash/patterns"]      match => { "mlogmsg" => "[%{MMLOGTS:mlogts}]\s[%{MMLOGWORDEXT:mlogcell}]\s[%{MMLOGWORDEXT:mlognode}]\s[%{MMLOGWORDEXT:mlogthread}]\s[%{MMLOGWORD:mloglevel}]\s[%{MMLOGWORDEXT:mlogclass}]\s%{GREEDYDATA}" }    }    grok {      match => { "source" => "%{GREEDYDATA}/%{GREEDYDATA:mlogfilename}.log" }    }    syslog_pri { }    date {      match => [ "mlogts", "yyyy-MM-dd HH:mm:ss.SSS" ]      timezone => "Asia/Shanghai"      target => "@timestamp"    }  }}这里最复杂的其实是两件事，① 用日志中的时间戳替换系统@timstamp（参见配置）②grok表达式将日志中的变量分段解析（找在线grok校验工具可以验证自己的正则，很费劲！！！）grok表达式默认支持各种格式的正则格式变量，大家自行官网搜索，这里我是自定义的一些正则变量，存放在：vi ~/dockerdata/logstash/patterns/mmlog.patterns下面的文件logstash默认会自动扫描的，所以文件名随便定义，只要正则对了就可以了。内容为：MMLOGTS \d{4}-\d{2}-\d{2}\s\d{2}\:\d{2}\:\d{2}.\d{3}MMLOGWORD \wMMLOGWORDEXT [^]]+MMLOGTHREAD \w(\w)\:\w-\wMMLOGCLASS [\w.]+\:\w\s

vi /usr/local/src/filebeat5/filebeat.yml编辑内容如下：################### Filebeat Configuration Example ###################################################### Filebeat ######################################filebeat.prospectors:# Each - is a prospector. Below are the prospector specific configurations- input_type: log  paths: ["/usr/local/src/logs/${appname}-${cellname}-${nodename}/sys-.log"]  encoding: utf-8  exclude_files: ['.\d{4}-\d{2}-\d{2}..log']  document_type: syslog  fields:    mlogapp: ${appname}  fields_under_root: true  scan_frequency: 1s  ignore_older: 30m# must set ignore_older to be greater than close_inactive.  close_inactive: 5m  close_removed: true  clean_removed: true  multiline:    pattern: ^[[[:digit:]]{4}-[[:digit:]]{2}-[[:digit:]]{2}[[:blank:]][[:digit:]]{2}:[[:digit:]]{2}:[[:digit:]]{2}.[[:digit:]]{3}]    negate: true    match: after    max_lines: 500output.kafka:  # initial brokers for reading cluster metadata  hosts: ["192.168.0.1:9092", "192.168.0.2:9092", "192.168.0.3:9092"]  # message topic selection + partitioning  topic: '%{[type]}'  partition.round_robin:    reachable_only: false  required_acks: 1  compression: gzip  max_message_bytes: 1000000############################# Logging #########################################logging.level: infologging.to_files: truelogging.to_syslog: falselogging.files:  path: /usr/local/logs/filebeat  name: filebeat.log  keepfiles: 7...----------------------*启动Filebeat5：**export appname="uss-web" && export cellname="cell01" && export nodename="node01" \&& cd /usr/local/src/filebeat/ \&& nohup ./filebeat -e > /usr/local/src/logs/filebeat/nohup.out 2>&1 &