最近需要收集linux系统日志，然后新版本rsyslog也支持直接吐到kafka，所以计划采用rsyslog加上传统的elk集群，大概的架构计划如下：

elasticsearch 8.0.0
logstash 8.0.0
kibanna 8.0.0
rsyslog 8.24.0
kafka 3.2.0

es
https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.0.0-x86_64.rpm
logstash
https://artifacts.elastic.co/downloads/logstash/logstash-8.0.0-x86_64.rpm
kibana
https://artifacts.elastic.co/downloads/kibana/kibana-8.0.0-x86_64.rpm
kafka
https://www.apache.org/dyn/closer.cgi?path=/kafka/3.2.0/kafka_2.12-3.2.0.tgz

elasticsearch部署

rpm -ivh elasticsearch-8.0.0-x86_64.rpm

✅ Elasticsearch security features have been automatically configured!
✅ Authentication is enabled and cluster connections are encrypted.




ℹ️  Password for the elastic user (reset with `bin/elasticsearch-reset-password -u elastic`):密钥




ℹ️  HTTP CA certificate SHA-256 fingerprint:
  9da4a430e857626b3e126a4bdf32724560bb0c51f093ef29232703ce43712548




ℹ️  Configure Kibana to use this cluster:
• Run Kibana and click the configuration link in the terminal when Kibana starts.
• Copy the following enrollment token and paste it into Kibana in your browser (valid for the next 30 minutes):
  eyJ2ZXIiOiI4LjAuMSIsImFkciI6WyIxMC4xMTMuMjIuMTY1OjkyMDEiXSwiZmdyIjoiOWRhNGE0MzBlODU3NjI2YjNlMTI2YTRiZGYzMjcyNDU2MGJiMGM1MWYwOTNlZjI5MjMyNzAzY2U0MzcxMjU0OCIsImtleSI6Ik9OR21UbjhCcnVQeTczamd4S3FtOkkwS1ZTV1VmUmNpYXpLQzFTMG1TV2cifQ==




ℹ️ Configure other nodes to join this cluster:
• Copy the following enrollment token and start new Elasticsearch nodes with `bin/elasticsearch --enrollment-token <token>` (valid for the next 30 minutes):
  eyJ2ZXIiOiI4LjAuMSIsImFkciI6WyIxMC4xMTMuMjIuMTY1OjkyMDEiXSwiZmdyIjoiOWRhNGE0MzBlODU3NjI2YjNlMTI2YTRiZGYzMjcyNDU2MGJiMGM1MWYwOTNlZjI5MjMyNzAzY2U0MzcxMjU0OCIsImtleSI6Ik45R21UbjhCcnVQeTczamd4S3FsOkZjcVZCNFF5VHVhcFZVUmFUVHBLS2cifQ==




  If you're running in Docker, copy the enrollment token and run:
  `docker run -e "ENROLLMENT_TOKEN=<token>" docker.elastic.co/elasticsearch/elasticsearch:8.0.1`

rpm -ivh elasticsearch-8.0.0-x86_64.rpm
/usr/share/elasticsearch/bin/elasticsearch-reconfigure-node --enrollment-token  加上述token
最后的话，就是配置文件了，其实要注意，有些是会自动生成的，自己配的话，有时候会重复
elasticsearch.yml
cluster.name: es
node.name: node-1
path.data: data/elasticsearch/data
path.logs: data/elasticsearch/logs
network.host: 0.0.0.0
discovery.seed_hosts: ["ip1","ip2", "ip3"]
cluster.initial_master_nodes: ["node-1"]
xpack.security.enabled: true
xpack.security.enrollment.enabled: true
xpack.security.http.ssl:
  enabled: true
  keystore.path: certs/http.p12
xpack.security.transport.ssl:
  enabled: true
  verification_mode: certificate
  keystore.path: certs/transport.p12
  truststore.path: certs/transport.p12
http.host: [_local_, _site_]

systemctl daemon-reload
systemctl start elasticsearch

kafka部署

tar -xvf
vim data/kafka/config/server.properties
broker.id=0 #需要修改
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=ip1:2181,ip2:2181,ip3:2181 #需要配置
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0


#启动
kafka-server-start.sh -daemon ../config/server.properties

查看topic
/data/kafka/bin/kafka-topics.sh --list --bootstrap-server ip1:9092
查看group
/data/kafka/bin/kafka-consumer-groups.sh --list  --bootstrap-server ip:9092

logstash部署

rpm -ivh logstash-8.0.0-x86_64.rpm
修改配置文件
vim etc/logsgtash/conf.d/test.conf
input {
  kafka {
        codec => "json"
        group_id => "logs"
        topics => ["system_logs"]
        bootstrap_servers => "ip1:9092,ip2:9092,ip3:9092"
        auto_offset_reset => "latest"
  }
 
}
 
 
output {
  elasticsearch {
    hosts => ["https://ip1:9200"]
    index => "sys-log-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "passwd"
    cacert => '/etc/logstash/http_ca.crt'
  }
}


#注意是https 密钥文件的话，是从es拷贝过来的

kibana部署

rpm -ivh kibana-8.0.0-x86_64.rpm

/etc/rsyslog.conf
#添加如下几行，另外需要在服务端部署 yum install rsyslog-kafka
module(load="omkafka")
module(load="imudp")
input(type="imudp" port="514" ruleset="tokafka")
*.* @ip:514 #服务端ip

/etc/ryslog.d/system.conf

#这个配置是将系统日志json化
template(name="json" type="list" option.json="on") {
   constant(value="{")
      constant(value="\"timestamp\":\"")
      property(name="timereported" dateFormat="rfc3339")
      constant(value="\",\"host\":\"")
      property(name="hostname")
      constant(value="\",\"facility\":\"")
      property(name="syslogfacility-text")
      constant(value="\",\"severity\":\"")
      property(name="syslogseverity-text")
      constant(value="\",\"syslog-tag\":\"")
      property(name="syslogtag")
      constant(value="\",\"message\":\"")
      property(name="msg" position.from="2")
   constant(value="\"}")
}
ruleset(name="tokafka") {
  action(
    type="omkafka"
    broker="ip1:9092,ip3:9092,ip2:9092"
    confParam=["compression.codec=snappy", "socket.keepalive.enable=true"]
    topic="system_logs"
    partitions.auto="on"
    template="json"
    queue.workerThreads="4"
    queue.size="360000"
    queue.dequeueBatchSize="100"
  )
}