1, logstash 配置文件
[root@host1: ] cat /opt/logstash/kafka-to-tcp.yml
input {
kafka {
bootstrap_servers => "192.168.0.11:9092"
consumer_threads => 3
group_id => "logstash_123"
auto_offset_reset => "earliest"
topics => ["alertTopic1"]
codec => json { charset => "UTF-8" }
}
}
filter {
if ([ruleName] !~ ".*主机告警.*") {
drop {}
}
mutate {
remove_field => ["eventId","ruleId"]
gsub => [
"Msg" , "[\r|\n]" , ""
]
}
}
output {
file {
codec => json_lines { charset => "UTF-8" }
path => "/tmp/b.log"
}
tcp {
host => "192.168.0.11"
port => "514"
codec => plain {
format =>"%{TIME} 测试环境--ruleName:%{ruleName},Msg:%{Msg}\n"
}
}
}
2,调试并后台启动
- ./bin/logstash -f /xx/xx.yml
[root@host1: ] cat /usr/lib/systemd/system/logstashtcp.service
[Unit]
Description=Logstash
Requires=network.service
After=network.service
[Service]
LimitNOFILE=65536
LimitMEMLOCK=infinity
ExecStart=/opt/logstash/bin/logstash -f /opt/logstash/kafka-to-tcp.yml
ExecReload=/bin/kill -HUP $MAINPID
KillMode=mixed
SuccessExitStatus=143
Restart=on-failure
RestartSec=42s
[Install]
WantedBy=multi-user.target