整体思路

filebeat采集数据--->Kafka--->Logstash提取Kafka--->ES-->Kibana可视化

ElasticSearch

首次启动时，密码在控制台日志中, 建议保留首次运行时的日志

 elk 下载地址： 下载 Elastic 产品 | Elastic

-> 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`):
  mfIi*c0bb63oTA9C8w6L

->  HTTP CA certificate SHA-256 fingerprint:
  45528b7da49344772a9822de77d093ad35e459217e2faf7459a4b25997a95e35

启动报错： not all primary shards of [.geoip_databases] index are active

exception during geoip databases update
org.elasticsearch.ElasticsearchException: not all primary shards of [.geoip_databases] index are active at

org.elasticsearch.ingest.geoip. GeoIpDownloader.updateDatabases (GeoIpDownloader.java:134)  ~[ingest-geoip-8.3.2.jar:8.3.2] at org.elasticsearch.ingest.geoip. GeoIpDownloader. runDownloader (GeoIpDownloader.java:274) [ingest-geoip-8.3.2.jar:8.3.2]
        at org.elasticsearch.ingest.geoip. GeoIpDownloaderTaskExecutor.nodeOperation (GeoIpDownloaderTaskExecutor.java:102) [ingest-geoip-8.3.2.jar:8.3.2] at org. elasticsearch.ingest.geoip. GeoIpDownloaderTaskExecutor. nodeOperation(GeoIpDownloaderTaskExecutor.java:48) [ingest-geoip-8.3.2.jar:8.3.2]
        at org.elasticsearch.persistent.NodePersistentTasksExecutor$1.doRun(NodePersistentTasksExecutor.java:42) [elasticsearch-8.3.2.jar:?]
        at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:769) [elasticsearch-8.3.2.jar:?]
        at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26) [elasticsearch-8.3.2.jar:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]

启动时会去更新地图的一些数据库，这里直接禁掉即可，用到时再说，配置文件

conf/elasticsearch.yml  配置 

ingest.geoip.downloader.enabled: false

Kibana

连接Elastic

将生成的token复制到上面的框中 

C:\Tools\elasticsearch-8.3.2\bin>elasticsearch-create-enrollment-token.bat -s kibana
warning: ignoring JAVA_HOME=C:\Tools\Java\jdk1.8.0_301; using bundled JDK
eyJ2ZXIiOiI4LjMuMiIsImFkciI6WyIxMC4xMC42OC4xNzU6OTIwMCJdLCJmZ3IiOiI0NTUyOGI3ZGE0OTM0NDc3MmE5ODIyZGU3N2QwOTNhZDM1ZTQ1OTIxN2UyZmFmNzQ1OWE0YjI1OTk3YTk1ZTM1Iiwia2V5IjoiR2Q1eThZRUJzUXEwaW9IMmxaWkM6LWFlZW51SVdTZ1MwM0Z6Y1ZCcGlvQSJ9 

LogStash

本地实现收集日志配置 logstash.conf

运行完需要删除
C:\Tools\logstash-8.3.2\data\plugins\inputs\file 里以.sin开头的文件，这里面记录file的状态，不删，再次运行就没有输出了

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
  file {
    #文件的绝对路径 无论是window还是liunx系统都必须是反斜杠/
    # logstash.conf bin文件在同级目录  bin目录下启动命令 
    path => "C:/Tools/logstash-8.3.2/test.log"
    start_position => "beginning"
    #每隔3秒刷新
    stat_interval => 3
    #sincedb_path => "/dev/null"
    }
}
  #beats {
  #  port => 5044
 # }

output {
  elasticsearch {
    hosts => ["https://10.10.68.175:9200"]
    ssl  => true
    cacert => "config/certs/http_ca.crt"
  #  index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "mfIi*c0bb63oTA9C8w6L"
  }
}
 

filebeat

下载地址：Download Filebeat • Lightweight Log Analysis | Elastic