Logstash 与Elasticsearch整合使用示例

环境准备

192.168.4.99

Zookeeper

docker run -d --name monitor-zk -p 2181:2181 -p 2888:2888 -p 3888:3888 jplock/zookeeper

Kafka

docker run -d \
--name monitor-kafka \
-p 9092:9092 \
-e KAFKA_ADVERTISED_HOST_NAME=192.168.4.99 \
-e ZOOKEEPER_IP=192.168.4.99 \
ches/kafka

# Kafka Test（可以忽略）
docker run --rm --interactive ches/kafka  kafka-console-producer.sh --topic m1 --broker-list 192.168.4.99:9092
docker run --rm ches/kafka kafka-console-consumer.sh --topic test --from-beginning --zookeeper 192.168.4.99:2181

Elasticsearch

docker run -d \
--name monitor-es \
-p 9200:9200 -p 9300:9300 \
-v /opt/monitor/es/data:/usr/share/elasticsearch/data \
-v /opt/monitor/es/plugins:/usr/share/elasticsearch/plugins \
elasticsearch:2.0.0 \
-Des.node.name="Node01" \
-Des.network.host=::0

Logstash

docker run -d \
--name monitor-logstash \
-p 7001-7005:7001-7005 \
-v /opt/monitor/logstash/config/:/config-dir \
-v /opt/monitor/logstash/input_test/:/input_test \ # 用于测试File输入的目录映射
logstash:2.0.0 \
logstash -f /config-dir/config.conf # 配置文件，后面会介绍

Logstash 配置

Logstash

config.conf

input {
kafka { # 这个用于kafka输入
type => 'key_trace_log'
topic_id => 'format_key_trace_topic'
zk_connect => '192.168.4.99:2181'
}
log4j { # 这个用于Log4j输入
type => 'tp_log_test1'
mode => 'server'
host => '0.0.0.0'
port => 7001
}
tcp { # 这个用于Logback输入
type => 'tp_log_test2'
host => '0.0.0.0'
port => 7002
codec => 'json_lines'
}
file { # 这个用于File输入
type => 'tp_log_test3'
path => ['/input_test/logs/*.log'] # 这是文件所在的目录
start_position => 'beginning'
}
}
filter {
if [type] == 'key_trace_log' {
csv { # Kafka输入的个message，这里假设是用tab分隔的，对应的字段如下
separator => '	'
columns => ['level','time','job_code','task_code','clue_value','node_code','stage','message']
}
}
if [type] == 'tp_log_test3' {
grok { # 对于File输入的数据需要做正则解析，下面正则解析的内容类似：2015-11-24 11:28:58  [ main:4234 ] - [ DEBUG ]  logstash slf4j test loop 109
pattern => '(?<datetime>.{19}) *\[ (?<class>.*):(?<line>\d*) \] - \[ (?<level>.*) \] (?<msg>.*)'
}
}
}
output {
if [type] == 'key_trace_log' {
elasticsearch {
hosts => ['192.168.4.99:9200']
index => 'monitor_key_trace_log'
document_type => 'default'
}
websocket {
host => '0.0.0.0'
port => 7005
}
}
if [type] == 'tp_log_test1' {
elasticsearch {
hosts => ['192.168.4.99:9200']
index => 'tp_log'
document_type => 'test1'
}
}
if [type] == 'tp_log_test2' {
elasticsearch {
hosts => ['192.168.4.99:9200']
index => 'tp_log'
document_type => 'test2'
}
}
if [type] == 'tp_log_test3' {
elasticsearch {
hosts => ['192.168.4.99:9200']
index => 'tp_log'
document_type => 'test3'
}
}
stdout { codec => rubydebug }
}

Logback

<dependency>
<groupId>net.logstash.logback</groupId>
<artifactId>logstash-logback-encoder</artifactId>
<version>4.5.1</version>
</dependency>

<appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
<destination>192.168.4.99:7002</destination>
<encoder class="net.logstash.logback.encoder.LogstashEncoder" />
<keepAliveDuration>5 minutes</keepAliveDuration>
</appender>

<root level="TRACE">
<appender-ref ref="logstash"/>
</root>

Log4j

log4j.appender.SOCKET=org.apache.log4j.net.SocketAppender
log4j.appender.SOCKET.port=7001
log4j.appender.SOCKET.remoteHost=192.168.4.99