部署 ELK 日志系统 Elasticsearch+Logstash+Kibana

118次阅读

共计 9989 个字符，预计需要花费 25 分钟才能阅读完成。

安装 ELK

安装 Java 1.8 环境

解压源码安装包：

tar xf jdk-8u121-linux-x64.tar.gz
ll
mkdir /work/opt -p
mv jdk1.8.0_121 /work/opt/jdk
ll /work/opt/jdk/
chown -R root.root /work/opt

vim /etc/profile : // 添加
export JAVA_HOME=/work/opt/jdk
export JAVA_BIN=$JAVA_HOME/bin
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=$JAVA_HOME/jre/lib/ext:$JAVA_HOME/lib/tools.jar
export PATH=$JAVA_BIN:$PATH

安装 elasticsearch-5.3.0

tar xf elasticsearch-5.3.0.tar.gz
mv elasticsearch-5.3.0 /work/opt/elasticsearch

Ubuntu@ip-172-31-1-79:/work/opt/elasticsearch/config$ egrep -v ‘#|^$’ elasticsearch.yml
cluster.name: lvnian-elk
node.name: lvnian-elk-node1
path.data: /data #由于是普通用户启动 elasticsearch，所以这个目录的属主需要改为普通用户
path.logs: /work/opt/elasticsearch/logs #由于是普通用户启动 elasticsearch，所以这个目录的属主需要改为普通用户
bootstrap.memory_lock: false
network.host: 0.0.0.0
http.port: 9200
http.cors.enabled: true
http.cors.allow-origin: “*”
ubuntu@ip-172-31-1-79:/work/opt/elasticsearch/config$
nohup /work/opt/elasticsearch/bin/elasticsearch >> /tmp/elasticsearch.log &

##ES 5.1.1 安装 head：
（5.1.1 版本的 elasticsearch 没有提供直接插件安装方法，但在该 github 上该插件作者给出了方法）

下载二进制源码包：
wget https://nodejs.org/dist/v4.2.2/node-v6.2.0-linux-x64.tar.gz
解压：
tar xf node-v6.2.0-linux-x64.tar.gz -C /work/opt/
设置环境变量：
vim /etc/profile：
export NODE_HOME=/work/opt/node/
export PATH=$PATH:$NODE_HOME/bin

root@ip-172-31-1-79:/work/source# node –version
v6.10.1

npm config set registry https://registry.npm.taobao.org // 设置代理镜像源，加速下载
cd /home/stt/node-v4.2.2-linux-x64/lib/node_modules
npm install grunt // 显示的 2 条 warn 可以忽略

测试 grunt 是否生效：

$ grunt -version
grunt-cli v1.2.0
grunt v1.0.1

安装 head 插件：

下载：git clone git://github.com/mobz/elasticsearch-head.git
cd /home/stt/elasticsearch-head
npm install (提示：如果遇到网络的瓶颈，将预先下载的源码包放在对应的位置效果一样，目录为 /tmp/phantomjs/，确定位置后可自行创建并上传包)
修改 elasticsearch-head/ _site/app.js
// 把 localhost 改为 ip
找到：
this.base_uri = this.config.base_uri || this.prefs.get(“app-base_uri”) || “http://localhost:9200”;
改为：
this.base_uri = this.config.base_uri || this.prefs.get(“app-base_uri”) || “http://192.168.8.116:9200”;
修改 elasticsearch-head/Gruntfile.js
connect: {
server: {
options: {
hostname: “0.0.0.0”, // 增加这个配置
port: 9100,
base: ‘.’,
keepalive: true
}
}
}

启动服务：（后台运行）
grunt server & // 需要在 /home/stt/elasticsearch-head 下执行，因为我的 grunt 没有进行全局的安装

## 安装 Logstash(这个软件在你需要读取的日志服务上安装，用 logstash 读取你的日志，上传给 elasticsearch 的):
也需要安装 java1.8 环境

tar xf logstash-5.3.0.tar.gz
mkdir /work/opt
mv logstash-5.3.0 /work/opt/
cd /work/opt/

vim /work/opt/logstash-5.3.0/conf/central.conf #(处理基于 FILE 方式输入的日志信息，这里是简单的举个例子，日后继续学习补充)

input {
file {
path => “/tmp/*.log”
}
}
output {
elasticsearch {
hosts => “192.168.8.116:9200”
index => “nginx-access”
}
stdout {
codec => rubydebug
}
}

## 安装 Kibana：
解压源码包：

tar zxf kibana-5.1.1-linux-x86_64.tar.gz -C /home/stt/server/
vim config/kibana.yml // 修改
server.port: 5601 // 打开注释而已，不用可以去效果，请使用默认端口
server.host: “0.0.0.0” // 打开监听地址，让别的机器也能访问这个 kibana
elasticsearch.url: “http://127.0.0.1:9200” // 这个 url 要根据实质情况，添加访问 elasticsearch 的 url
启动服务：（后台运行）
/home/stt/server/kibana-5.1.1-linux-x86_64/bin/kibana &

安装 nginx 反向代理，
apt-gei install nginx

nginx 放心代理配置文件如下：

## 文件名 kibana.conf
upstream backend {
server 172.31.6.155:5601;
}
server {
listen 80;
server_name kibana.lvnian.co;
access_log /tmp/kibana-access.log;
error_log /tmp/kibana-error.log;
location / {
#设置主机头和客户端真实地址，以便服务器获取客户端真实 IP
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
#禁用缓存
proxy_buffering off;
#反向代理的地址
proxy_pass http://backend;
}
}

logstash 读取 nginx 的访问日志以及 error 日志，上传到 logstash 的配置文件文件。
用下面命令运行即可

nohup /work/opt/logstash-5.3.0/bin/logstash -f /work/opt/logstash-5.3.0/conf/elk-nginx-log.conf &

文件名：elk-nginx-log.conf
input {
file {
path => “/data/log/nginx/*.log”
start_position => beginning
}
}
filter {
if [path] =~ “access” {
mutate {replace => { type => “nginx_access”} }
ruby {
init => “@kname = [‘http_x_forwarded_for’,’time_local’,’request’,’status’,’body_bytes_sent’,’request_body’,’content_length’,’http_referer’,’http_user_agent’,’http_cookie’,’remote_addr’,’hostname’,’upstream_addr’,’upstream_response_time’,’request_time’]”
code => ”
new_event = LogStash::Event.new(Hash[@kname.zip(event.get(‘message’).split(‘|’))])
new_event.remove(‘@timestamp’)
event.append(new_event)
”
}
if [request] {
ruby {
init => “@kname = [‘method’,’uri’,’verb’]”
code => ”
new_event = LogStash::Event.new(Hash[@kname.zip(event.get(‘request’).split(‘ ‘))])
new_event.remove(‘@timestamp’)
event.append(new_event)
”
}
if [uri] {
ruby {
init => “@kname = [‘url_path’,’url_args’]”
code => ”
new_event = LogStash::Event.new(Hash[@kname.zip(event.get(‘uri’).split(‘?’))])
new_event.remove(‘@timestamp’)
event.append(new_event)
”
}
kv {
prefix => “url_”
source => “url_args”
field_split => “& “
remove_field => [“url_args”,”uri”,”request”]
}
}
}
mutate {
convert => [
“body_bytes_sent” , “integer”,
“content_length”, “integer”,
“upstream_response_time”, “float”,
“request_time”, “float”
]
}
date {
match => [“time_local”, “dd/MMM/yyyy:hh:mm:ss Z”]
locale => “en”
}
}
else if [path] =~ “error” {
mutate {replace => { type => “nginx_error”} }
grok {
match => {“message” => “(?<datetime>\d\d\d\d/\d\d/\d\d \d\d:\d\d:\d\d) \[(?<errtype>\w+)\] \S+: \*\d+ (?<errmsg>[^,]+), (?<errinfo>.*)$” }
}
mutate {
rename => [“host”, “fromhost”]
gsub => [“errmsg”, “too large body: \d+ bytes”, “too large body”]
}
if [errinfo]
{
ruby {
code => ”
new_event = LogStash::Event.new(Hash[event.get(‘errinfo’).split(‘, ‘).map{|l| l.split(‘: ‘)}])
new_event.remove(‘@timestamp’)
event.append(new_event)
”
}
}
grok {
# match => {“request” => ‘”%{WORD:verb} %{URIPATH:urlpath}(?:\?%{NGX_URIPARAM:urlparam})?(?: HTTP/%{NUMBER:httpversion})”‘ }
match => {“request” => ‘”%{WORD:verb} %{URIPATH:urlpath}?(?: HTTP/%{NUMBER:httpversion})”‘ }
patterns_dir => [“/etc/logstash/patterns”]
# remove_field => [“message”, “errinfo”, “request”]
}
}
else {
mutate {replace => { type => “random_logs”} }
}
}
output {
elasticsearch {
hosts => “172.31.1.79:9200”
#index => “logstash-nginx”
index => “logstash-%{type}-%{+YYYY.MM.dd}”
document_type => “%{type}”
flush_size => 20000
idle_flush_time => 10
sniffing => true
template_overwrite => true
}
# stdout {
# codec => rubydebug
# }
}

监控指点时间内 502 日志超过阀值的微信报警
#!/usr/bin/env Python
# -*- coding:utf-8 -*-
#Author: gaogd

import weixin_alter as weixin
from elasticsearch import Elasticsearch
import json,time ,datetime

now_day=time.strftime(‘%Y.%m.%d’,time.localtime(time.time()))

es = Elasticsearch([{‘host’:’172.3.11.179′,’port’:9200}])
index=’logstash-nginx_access-%s’%now_day

body ={
“query”: {
“bool”: {
“filter”: [
{“term”: { “status”: “502”}},
{
“range” : {
“@timestamp” : {
“gt”: “now-15m”,
“lt”: “now”
},
},
},
]
}
},
“aggs”: {
“group_by_source”: {“terms”: {“field”: “params.source”}}
},
“size”: 0
}

res=es.search(index=index, body=body)
# res=es.search(index=index, q=QueryCondition)
num= res[‘hits’][‘total’]
# print ‘502 错误量：’,num
if int(num) > 20:
with open(‘./alter’,’r’) as f:
time=f.read()
if len(time) == 0 or int(time) == 0 or int(time) >= 15:
print u’502 告警！最近 1 分钟内日志出现 502 错误的有：%s 次 \n %s’ % (num,datetime.datetime.now())
content = u’502 告警!!! \n 最近 1 分钟内前端日志出现 502 错误 \n 一分钟内有：%s 次 ’ % (num)
weixin.WeixinSend(str(content))
with open(‘./alter’,’w+’) as f:
f.write(‘1’)
elif 0 < int(time) <=15:
time=int(int(time)+1)
print ‘———->’,time
with open(‘./alter’, ‘w+’) as f:
f.write(str(time))
exit()
else:
with open(‘./alter’,’r’) as f:
num1=f.read()
if int(num1) > 0:
print u’502 告警恢复！\n 日志 502 错误已经恢复 \n 最近 1 分钟内日志出现 502 错误少于 20 次 \n 现在错误次数为：%s 次 \n %s’ % (num,datetime.datetime.now())
content = u’502 告警恢复！！！\n 日志 502 错误已经恢复最近 1 分钟内日志出现 502 错误少于 20 次 \n 现在错误次数为：%s 次 ’ % num
weixin.WeixinSend(str(content))
with open(‘./alter’, ‘w+’) as f:
f.write(‘0’)