1、下载Logstash

wget https://artifacts.elastic.co/downloads/logstash/logstash-7.17.7-linux-x86_64.tar.gz

2、解压到/usr/local/

mkdir -p /usr/local/logstash7.17
tar -zxf logstash-7.17.7-linux-x86_64.tar.gz -C /usr/local/logstash7.17/ --strip-components=1 #--strip-components选项表示从目录级别上去除指定的前缀，以实现更加控制解压的效果

3、logstash数据写到kafka中

cd /usr/local/logstash7.17/config/
cp logstash-sample.conf logstash.conf
input{ stdin{ }
}
output{ kafka{ topic_id => "logstash_k"bootstrap_servers => "192.168.0.191:9092"  # kafka的地址jaas_path => "/usr/local/logstash7.17/config/logstash.conf"security_protocol => "SASL_PLAINTEXT"sasl_mechanism => "SCRAM-SHA-256"codec => "json"}stdout{ codec => rubydebug}
}

4、启动、停止

cd /usr/local/logstash7.17
mkdir -p /data/logstash7/
nohup ./bin/logstash -f config/logstash.conf &>> /data/logstash7/logstash-server-`date "+%Y%m%d"`.log & echo $! > /data/logstash7/logstash.pid #启动
cat /data/logstash7/logstash.pid | xargs -I {} kill {} #停止

logstash配置详解

Logstash 是一个强大的日志处理工具，可以用于数据收集，转换和存储。以下是 Logstash 配置文件的基本结构和一个示例配置：

# Logstash 配置文件示例# 设置 Logstash 的工作区域
# path 指定了 Logstash 的工作目录
# node.name 是集群中此节点的名称
# cluster.name 是集群的名称
# pipeline.workers 是执行输入和过滤器的工作线程数
# pipeline.batch.size 是每个批次的事件数
# pipeline.batch.delay 是等待批次填满的时间pipeline.workers: 2
pipeline.batch.size: 125
pipeline.batch.delay: 50# 输入配置
# 使用 file 插件监控日志文件的变化
input {file {path => "/var/log/system.log"start_position => "beginning"}
}# 过滤器配置
# 使用 grok 插件解析日志格式
filter {grok {match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{DATA:program}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:message}" }add_field => [ "received_at", "%{@timestamp}" ]add_field => [ "received_from", "%{host}" ]}
}# 输出配置
# 将处理后的事件发送到 Elasticsearch
output {elasticsearch {hosts => ["localhost:9200"]index => "logstash-%{+YYYY.MM.dd}"}
}

在这个配置中，我们定义了 Logstash 的工作区域和线程设置，然后配置了一个文件输入来监控系统日志文件的变化。接着，我们使用了 grok 插件来解析日志消息，并添加了两个额外字段。最后，我们配置了 Elasticsearch 作为输出，将数据发送到 Elasticsearch 索引中。