分布式日志系统 ELK

Elasticsearch LogStash、Kibana 安装地址：[https://www.elastic.co/products]

Elasticsearch

Elasticsearch 是一个具有强大索引功能的文档存储库，并且可以通过 REST API 来搜索数据。基于 Apache Lucene，使用 Java 编写

ElasticSearch不仅仅只提供对这些被索引文档的强大搜索功能。快速、分布式、水平扩展，支持实时文档存储和分析，支持数百台服务器和 PB 级索引数据。

//端口：9200

Kibana

Kibana 是 Elasticsearch 中专门提供强有力的可视化查询Web应用程序。使用Kibana，能非常简单地为 Elasticsearch 中索引的数据创建查询、图表和仪表盘。

需要配置elasticsearch服务地址，比如：

//端口：5601
elasticsearch.url："http://localhost:9200"

Logstash

Logstash是一个具备实时处理能力的开源的数据收集引擎。可以动态地从不同的来源收集数据，将数据处理（过滤、变形）过之后统一输出到某个特定地址，为将来更多样化的数据分析做准备。 logstash采用input,filter,output的三段配置法。input配置输入源，filter配置对输入源中的信息怎样进行处理，而output配置输出位置。

vi /var/logstash/config/logstash.yml
http.host: "0.0.0.0"
path.config: /usr/share/logstash/pipeline
xpack.monitoring.elasticsearch.hosts: http://localhost:9200
xpack.security.enabled: false

//新增配置信息
vi /var/logstash/pipeline/log4j_to_es.config 
input {
  tcp {
        host => "localhost"
        port => 4567
        codec => json_lines
      }
}

filter {
}

output {
  stdout { codec => rubydebug }
  elasticsearch {
    action => "index"          #The operation on ES
    hosts  => "localhost:9200"   #ElasticSearch host, can be array.
    index  => "applog"         #The index to write data to.
  }
}

vi /var/kibana/config/kibana.yml
server.name: kibana
server.host: "0"
elasticsearch.hosts: [ "http://localhost:9200" ]
xpack.monitoring.ui.container.elasticsearch.enabled: true

常用的命令

1. -f：通过这个命令可以指定Logstash的配置文件，根据配置文件配置logstash
2. -e：后面跟着字符串，该字符串可以被当做logstash的配置（如果是“” 则默认使用stdin作为输入，stdout作为输出）
3. -l：日志输出的地址（默认就是stdout直接在控制台中输出）
4. -t：测试配置文件是否正确，然后退出。

5. 账号问题

   groupadd elk useradd elk -g elk userdel -r elk groupdel elk chown -R elk:elk /opt

Docker ELK

sysctl -w vm.max_map_count=262144 docker kill elk docker rm elk

docker run -p 5601:5601 -p 9200:9200 -p 4567:4567 -v /var/logstash:/var/logstash/ -e ES_MIN_MEM=1024m -e ES_MAX_MEM=20148m -d –restart=always –name elk sebp/elk

//进入docker 中 docker exec -it elk /bin/bash

cd /opt/logstash/data/ ls -lah rm -rf .lock nohup /opt/logstash/bin/logstash -f /var/logstash/pipeline/log4j_to_es.config>/var/logstash/logstash.log &

Docker Single

docker kill elasticsearch docker rm elasticsearch docker run -p 9200:9200 -e “http.host=0.0.0.0” -e “transport.host=127.0.0.1” –name elasticsearch –network host -d docker.io/elasticsearch:7.1.1

docker kill kibana docker rm kibana docker run -p 5601:5601 -v /var/kibana/config:/usr/share/kibana/config:ro –name kibana –network host -d docker.io/kibana:7.1.1

docker kill logstash docker rm logstash docker run -p 4567:4567 -v /var/logstash/pipeline:/usr/share/logstash/pipeline:ro -v /var/logstash/config:/usr/share/logstash/config:ro –name logstash –network host -d docker.io/logstash:7.1.1

Dockerfile elasticsearch kibana logstash

#!/bin/bash nohup /opt/elasticsearch/bin/elasticsearch -Des.insecure.allow.root=true > /var/log/elasticsearch.log & nohup /opt/kibana/bin/kibana > /var/log/kibana.log & nohup /opt/logstash/bin/logstash -f /temp/logstash/log4j_to_es.config>/var/log/logstash.log &

nginx config

log_format main ‘{“@timestamp”:”$time_iso8601”,’ ‘“@source”:”$server_addr”,’ ‘“hostname”:”$hostname”,’ ‘“ip”:”$remote_addr”,’ ‘“client”:”$remote_addr”,’ ‘“request_method”:”$request_method”,’ ‘“scheme”:”$scheme”,’ ‘“domain”:”$server_name”,’ ‘“referer”:”$http_referer”,’ ‘“request”:”$request_uri”,’ ‘“args”:”$args”,’ ‘“size”:$body_bytes_sent,’ ‘“status”: $status,’ ‘“responsetime”:$request_time,’ ‘“upstreamtime”:”$upstream_response_time”,’ ‘“upstreamaddr”:”$upstream_addr”,’ ‘“http_user_agent”:”$http_user_agent”,’ ‘“https”:”$https”’ ‘}’;

input { file { ## 修改你环境nginx日志路径 path => “/var/logs/xxx/access/*.log” ignore_older => 0 codec => json } }

filter { mutate { convert => [ “status”,”integer” ] convert => [ “size”,”integer” ] convert => [ “upstreatime”,”float” ] convert => [“[geoip][coordinates]”, “float”] remove_field => “message” }

grok {

patterns_dir => [ “/etc/logstash/patterns.d” ]

match => { “message” => “%{NGINXACCESS}”}

}

date {
    match => [ "timestamp" ,"dd/MMM/YYYY:HH:mm:ss Z" ]
}
geoip {
  source => "client"  ##日志格式里的ip来源，这里是client这个字段（client":"$remote_addr"）
  target => "geoip"
  database =>"/usr/share/GeoIP/GeoLite2-City.mmdb"   ##### 下载GeoIP库
  add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
  add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
}
mutate {
  remove_field => "timestamp"
}
if "_geoip_lookup_failure" in [tags] { drop { } } ### 如果解析的地址是内网IP geoip解析将会失败，会生成_geoip_lookup_failure字段，这段话的意思是如果内网地址 drop掉这个字段。 }

output { elasticsearch { hosts => [“xxx:9200”,”xxxx:9200”,”xxxx:9200”] index => “logstash-nginx-test-xxxx_%{+YYYY-MM}” user => xxxx password => xxxx } stdout { codec => rubydebug } }