基于ELK 7.50搭建elastalert 监控报警和权限控制

ELK+监控报警全步骤

需求： 公司要求对出在windows服务器上的日志进行日志分析并根据关键字进行报警，并配置kibana权限控制。下面为详细步骤

环境： centos 7.6 elk版本7.50 （因为7.50版本自带xpack功能，可以满足kibana角色权限控制）

1. windows字符集改成utf8

#创建目录（有就不用创建）

C:\WINDOWS\SHELLNEW

#创建一个文本文档(txt) 复制到该目录：

#命名为：UTF8.txt

#文件 -> 另存为…

#选择编码格式为：UTF-8
 
WIN + R  ->regedit

#按以下路径找到ShellNew项：

HKEY_CLASSES_ROOT\.txt\ShellNew

#新建 -> 字符串

#命名为：FileName

#双击 FileName这项，输入：UTF8.txt

#按以下路径找到Notepad项：HKEY_CURRENT_USER\Software\Microsoft\Notepad

#更改以下两项值为：1（如果不存在，自行创建：右键 -> 新建 -> DWORD）

fSavePageSettings 

fSaveWindowPositions

2. 下载安装包并安装

1.1 filebeat

https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.6.0-windows-x86_64.zip

#下载安装包，并解压至c:\filebeat

#进入c:\filebeat 修改配置文件filebeat.yml

############################################
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - c:\work\*
#  include_lines: ['Errors']
  tags: "ca"
- type: log
  enabled: true 
  paths:
    - d:\work\*
 # json.keys_under_root: true   #如果日志文件本来就是json格式，需要下面2行参数
 # json.overwrite_keys: true
  tags: "json"
  
output.redis:   
  hosts: ["192.168.2.23:6379"]
  key: "all"

setup.template.name: "nginx"
setup.template.pattern: "nginx_*"
setup.template.enabled: false
setup.template.overwrite: true
############################################


#启动powershell--> 以管理员身份运行

PowerShell.exe -ExecutionPolicy UnRestricted -File .\install-service-filebeat.ps1 #这是以windows服务形式启动

cd c:\filebeat

.\filebeat.exe -e -c filebeat.yml #这是以cmd形式启动，可以改成bat文件进行运行


#这边因为filebeat拉取数据直接给的redis，所以不需要配置elasticsearch的账号密码

1.2. redis

#安装redis

yum install -y redis

#修改配置文件

vim /etc/redis.conf

daemonize yes
bind 192.168.2.23
port 6379

#启动redis

systemctl start redis

netstat -ltnp |grep 6379

#测试登录
redis-cli -h 192.168.2.23
192.168.2.23:6379>

1.3 jdk

#下载jdk 8 为其他服务提供支持

wget https://download.oracle.com/otn/java/jdk/8u231-b11/5b13a193868b4bf28bcb45c792fce896/jdk-8u231-linux-x64.rpm

#安装jdk

rpm -ivh jdk-8u231-linux-x64.rpm

1.4. elasticsearch

# 安装es 7.5 需要安装openjdk 11(es 7.0以上对jdk版本要求升高了)

#下载安装包

wget https://download.java.net/java/GA/jdk11/13/GPL/openjdk-11.0.1_linux-x64_bin.tar.gz

#将安装包解压到/opt下

tar xf openjdk-11.0.1_linux-x64_bin.tar.gz -C /opt/

#修改配置文件

cd /usr/share/elasticsearch/bin

vi elasticsearch

添加以下下几行内容

#在后面句子后面添加
#   ES_JAVA_OPTS="-Xms8g -Xmx8g" ./bin/elasticsearch  
#配置自己的jdk11，但是并不影响整个系统的jdk环境变量，共存
export JAVA_HOME=/opt/jdk-11.0.1
export PATH=$JAVA_HOME/bin:$PATH

------------------

#在后面句子前面添加# manual parsing to find out, if process should be detached  
#添加jdk判断
    if [ -x "$JAVA_HOME/bin/java" ]; then
            JAVA="/opt/jdk-11.0.1/bin/java"
    else
            JAVA=`which java`
    fi

# 修改JDK11支持的垃圾回收器

vim /etc/elasticsearch/jvm.options

#-XX:+UseConcMarkSweepGC  #注释这个
-XX:+UseG1GC                #添加这个

#修改启动配置

systemctl edit elasticsearch 

[Service]
LimitMEMLOCK=infinity
systemctl daemon-reload
systemctl restart elasticsearch

# 修改配置文件
grep "^[a-Z]" /etc/elasticsearch/elasticsearch.yml 

node.name: node-1
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
bootstrap.memory_lock: true
network.host: 0.0.0.0
http.port: 9200

cluster.initial_master_nodes: ["node-1"]  #将注释取消，修改为这个   
discovery.type: single-node   #或者是添加这个参数，意思是单节模式

#启动elasticsearch

systemctl start elasticsearch

########################
#如果启动失败，可以换成./elasticsearch 启动，但是这边是有坑的，不能使用root用户启动
#把elasticsearch用户改成普通可登陆用户（删除，重新创建）
#要把相关的文件chown 改成 elasticsearch 
#把启动文件加入到bin/elasticsearch中

#[Service]里面添加下面三行
LimitMEMLOCK=infinity
systemctl daemon-reload
systemctl restart elasticsearch

#配置elasticsearch.yml 文件下面修改为
xpack.security.enabled: true
discovery.type: single-node
node.max_local_storage_nodes: 2

#启动 elasticsearch 

screen -S elasticsearch

cd /usr/share/elasticsearch/bin/

./elasticsearch

ctrl +a +d 

#验证是否启动

netstat -ltnp |grep 9200

=========================

#在web中验证是否启动成功，需要安装es-head

#修改ES配置文件支持跨域

http.cors.enabled: true 

http.cors.allow-origin: "*"

#方法1（通用）

# 使用docker安装 es-head

yum install docker -y

#下载es-head 

docker pull alivv/elasticsearch-head

#docker运行镜像

docker run --name es-head -p 9100:9100 -dit elivv/elasticsearch-head

#方法2 （通用）

#安装各种依赖包

yum install nodejs npm openssl screen -y

#安装node和npm

node -v

npm  -v

npm install -g cnpm --registry=https://registry.npm.taobao.org

cd /opt/

#拉取git代码

git clone git://github.com/mobz/elasticsearch-head.git

#安装cnpm

cd elasticsearch-head/

cnpm install

#启动es-head

screen -S es-head

cnpm run start

Ctrl+A+D

#方法3 （仅适用于Google浏览器）

#右上角 --》 更多工具--》扩展程序 

#下载下来ElasticSearch Head0.1.4

#将下载下来的包改成es-head-0.1.4_0.crx

1.5 开启xpack功能

#因为我们要实现kibana权限控制功能，那么第一步是要给es设置安全密码

vim /etc/elasticsearch/elasticsearch.yml

#开启自带的xpack的验证功能,在6.8版本以后，已经是自带xpack功能了，不需要花钱了
xpack.security.enabled: true
 
#配置单节点模式
discovery.type: single-node   #开启这个要关闭cluster那个选项

#执行程序

cd /usr/share/elasticsearch/bin 

./elasticsearch-setup-passwords interactive #输入y 输入密码，最少6位

#记住各自系统服务的账号和密码

#记住elasticsearch的账号和密码，因为在logstash、elastalert、kibana中都需要配置elasticsearch账号和密码

Changed password for user [apm_system]
Changed password for user [kibana]
Changed password for user [logstash_system]
Changed password for user [beats_system]
Changed password for user [remote_monitoring_user]
Changed password for user [elastic]

#重启es

#登录es，发现已经需要输入账号密码了

1.6. logstash

#下载logstash安装包

wget https://artifacts.elastic.co/downloads/logstash/logstash-7.5.0.rpm

#安装logstash

rpm -ivh logstash-7.5.0.rpm

#修改配置文件

vim /etc/logstash/conf.d/redis.conf

input {
  redis {
    host => "192.168.2.23"
    port => "6379"
    db => "0"
    key => "all"
    data_type => "list"
  }
}

#filter {
#  mutate {
#    convert => ["upstream_time", "float"]
#    convert => ["request_time", "float"]
#  }
#if "ca" in [message]{   
#   grok {
#    match => { "message" => "%{TIMESTAMP_ISO8601:DATE_time}\s*%{USER:server_name}\S+\s*%{INT:level}\,(?<SNO>(.*))\,(?<excute_time>(.*))\;\s\S+%{GREEDYDATA:message_value}"
#   }
#        }
#    }
#}

filter {
  if "ca" in [tags]{
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:DATE_time}\s*%{USER:server_name}\S+\s*%{INT:level}\,(?<SNO>(.*))\,(?<excute_time>(.*))\;\s\S+%{GREEDYDATA:message_value}" }   
  }
}
  if "json" in [tags]{
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:DATE_time}\s*\S+\<(?<MODULE>(.*))\>\s\S+\:\<(?<lv_num>(.*))\,(?<lv_SNO>(.*))\>\s\S+%{GREEDYDATA:message_value}" }
  }
}
}
output {
   stdout {}
   if "ca" in [tags] {
      elasticsearch {
        hosts => "http://192.168.2.23:9200"
        manage_template => false
        index => "ca-%{+yyyy.MM}"
        user => "elastic"
        password => "123456"
      }
    }
   if "json" in [tags] {
      elasticsearch {
        hosts => "http://192.168.2.23:9200"
        manage_template => false
        index => "km-%{+yyyy.MM}"
        user => "elastic"
        password => "123456"
      }
    }
#if "rrors" in [message] {     #mail插件，可以用来报警发邮件
#email {
#        port           =>    "25"
#        address        =>    "smtp.qq.com"
#        username       =>    ""
#        password       =>    "xxxxxxxxxxxxxxxxxxxxxx"
#        authentication =>    "plain"
#        use_tls        =>    false
#        from           =>    ""
#        subject        =>    "日志中有error信息"
#        to             =>    ""
#        via            =>    "smtp"
#        body           =>    "错误日志： \n  %{message} "
#    }
#}

}


#安装screen后台运行程序

yum install screen -y 

#创建一个logstash的后台程序

screen -S logstash  

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis.conf

#退出screen后台程序

ctrl + a + d

#查看screen后台程序

screen -ls

#进入指定的后台

screen -r scrren.id

1.6.1 将kibana的收集时间替换为系统日志时间

那如何解决上诉问题呢？请看下面??????

filter {
###替换@timestamp时间为日志真实时间######

#方法1
    grok {
        match => {  "message" => "(?<timestamp>%{TIMESTAMP_ISO8601})"  }
    }
    date {
        match => [ "timestamp", "ISO8601" ]
    }
    mutate {
         remove_field => [ "timestamp" ]
    }
}


#方法2 （自己写的实际的列子）

  if "json" in [tags]{
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:times}\s*\S+\<(?<MODULE>(.*))\>\s\S+\:\<(?<lv_num>(.*))\,(?<lv_SNO>(.*))\>\s\S+%{GREEDYDATA:message_value}" }
  }
  date {
        match => ["times", "ISO8601"]
        locale => "en"                      #下面这两行是定义时区，比中国时区多了八个小时
        timezone => "+00:00"                #一般不用这两种
        remove_field => "times"
    }
}
}

1.7. kibana

#下载rpm包

wget https://artifacts.elastic.co/downloads/kibana/kibana-7.5.0-x86_64.rpm

#安装kibana

rpm -ivh kibana-7.5.0-x86_64.rpm

#编辑配置文件

[ ~]# grep "^[a-Z]" /etc/kibana/kibana.yml
server.port: 5601
server.host: "192.168.2.23"
elasticsearch.hosts: ["http://172.16.1.45:9200"]
kibana.index: ".kibana"
elasticsearch.username: "kibana"
elasticsearch.password: "123456"

#启动kibana

systemctl daemon-reload  #有的时候，会提示找不到服务，就重新reload一下

systemctl start kibana

1.8. 安装elastalert报警

#安装elastalert 首先环境需要替换成python3.6

#安装python3.6

 yum install python36 -y
 yum install python36-pip -y

#测试，显示python3.6已经安装成功，但是默认还是python2
python --version
 python3 --version
 
 #将默认python 修改为python3
 
 cd /usr/bin
 
 ls python*
python   python2.7         python2-config  python3.6         python3.6m         python3.6m-x86_64-config  python.bak

#重新创建软链接指向python3.6

rm -f python && ln -s python3.6 python

#修改完成后，测试版本

# python --version
Python 3.6.8

# python2 --version
Python 2.7.5

#但是这个时候，因为将python默认改成了3.6，所以系统里面依赖python2的都需要修改配置文件

#将第一行"#!/usr/bin/python" 改为 "#!/usr/bin/python2"即可。
vi /usr/bin/yum

#将第一行"#!/usr/bin/python" 改为 "#!/usr/bin/python2"即可。
vi /usr/libexec/urlgrabber-ext-down

#如果某些服务报错，需要修改python环境


#安装依赖包
yum install gcc libffi-devel python-devel openssl-devel4 -y

#安装pip

yum install python2-pip -y

yum install python3-pip-9.0.3-5.el7.noarch

pip install elastalert

#从github上拉取代码

git clone https://github.com/Yelp/elastalert.git

#安装模块

 pip install "setuptools>=11.3"
 
#安装各种包

cd elastalert

python setup.py install 

echo $? #验证一下是否正确

#如果es是7.0版本以上，如果提示报错，那边把以前的卸载掉，重新安装

pip uninstall elasticsearch

pip3 install 'elasticsearch>=7.0.0'

#这个时候已经多出来elastalert 几个命令了

#创建报警索引index_status

#编辑配置文件
cp config.yaml.example config.yaml

#配置文件如下：

# grep -Ev "^$|^#" config.yaml  

rules_folder: example_rules
run_every:
  minutes: 1
buffer_time:
  minutes: 15
es_host: 192.168.2.23
es_port: 9200
es_username: "elastic"
es_password: "123456"
writeback_index: elastalert_status
writeback_alias: elastalert_alerts
alert_time_limit:
  days: 2

#创建报错索引

elastalert-create-index

提示成功

1.9. 配置elastic报警规则

配置example_rules下面的rules


#这是一个样本

# grep -Ev "^$|^#" example_rules/test_rule.yaml 
name: 你有一封来自elastalert的日志报警
type: frequency
index: ca*
num_events: 5
timeframe:
  minutes: 4
filter:
- term:
    level: "3"
alert:
- command
- email
new_style_string_format: true
command: ["/opt/test.sh", "elk nginx warning - freq 500 exceed, domain: {match[domain]}"]
alert_text: "报警来自 http://192.168.2.23:9100"
smtp_host: smtp.qq.com
smtp_port: 25
smtp_auth_file: /usr/local/elastalert/example_rules/smtp_auth_file.yaml
email_reply_to: 
from_addr: 
email:
- ""

#再启动一个km的报警规则

#[_server elastalert]# grep -Ev "^$|^#" example_rules/km_test.yaml
es_host: 192.168.2.23
es_port: 9200
name: 你有一封来自elastalert的km日志报警
type: frequency
index: km*
num_events: 5
timeframe:
  minutes: 4
filter:
- term:
    lv_num: "3"
alert:
- command
- email
new_style_string_format: true
command: ["/opt/test.sh", "elk nginx warning - freq 500 exceed, domain: {match[domain]}"]
alert_text: "报警来自 http://192.168.2.23:9100"
smtp_host: smtp.qq.com
smtp_port: 25
smtp_auth_file: /usr/local/elastalert/example_rules/smtp_auth_file.yaml   #创建此文件，并根据需求授权
email_reply_to: 
from_addr: 
email:
- ""

#测试创建的规则语法

elastalert-test-rule example_rules/test_rule.yaml  #如果语法正确会看到hit [num] 字样

#启动规则，正式拉取报警信息

screen -S elastalert1

python -m elastalert.elastalert --verbose --rule example_rules/test_rule.yaml 

ctrl +a +d

#再拉取一个guiz
screen -S elastalert2

python -m elastalert.elastalert --verbose --rule example_rules/km_test.yaml

ctrl +a +d

#测试，在服务器日志上插入报警信息，查看是否可以双报警

1.10. 配置短信报警（配置command模块）

vim python3.py 
from  datetime import datetime
import hashlib
import base64
import requests
import json
class YunTongXin():
    # 生产环境的base_url
    base_url = 'https://app.cloopen.com:8883'
    timestamp = None
    def __init__(self,accountSid,authToken,appId,templateId,notice=''):
        self.accountSid = accountSid  # 开发者主账户 ACCOUNT SID
        self.authToken = authToken # 账户授权令牌
        self.appId = appId # 应用id
        self.templateId = templateId # 模版id
        self.notice = notice # 提示信息

    # 构造请求url
    def gen_request_url(self,sig):
        self.url = self.base_url + '/2013-12-26/Accounts/{}/SMS/TemplateSMS?sig={}'.format(self.accountSid,sig)
        return self.url

    # 构造请求头
    def gen_request_header(self,timestamp):
        authorization = self.gen_authorization(timestamp)
        return {
            "Accept":"application/json",
            "Content-Type":"application/json;charset=utf-8",
            "Authorization":authorization
        }

    # 构建请求体
    def gen_request_body(self,phone,code):
        return {
            "to":phone,
            "appId":self.appId,
            "templateId":self.templateId,
            "datas":[code,"3"]
        }

    # 获取 Authorization
    def gen_authorization(self,timestamp):
        return self.base64_encode(self.accountSid+':'+timestamp)

    # base64加密
    def base64_encode(self,raw):
        return base64.b64encode(raw.encode('utf-8')).decode()


    # 生成签名文档
    def gen_sig(self,timestamp):
        return self.md5(self.accountSid+self.authToken+timestamp)

    # 生成时间戳
    def gen_timestamp(self):
        return  datetime.now().strftime('%Y%m%d%H%M%S')

    # md5加密
    def md5(self,raw):
        md5 = hashlib.md5()
        md5.update(raw.encode('utf-8'))
        return md5.hexdigest().upper()

    # 请求云通信接口
    def request_yuntongxin_api(self,url,header,body):
        response = requests.post(url,headers=header,data=body)
        return response.text

    # 运行
    def run(self,phone,code):
        # 获取时间戳
        timestamp = self.gen_timestamp()
        # 生成签名
        sig = self.gen_sig(timestamp)
        # 请求url
        url = self.gen_request_url(sig)
        # 请求头
        header = self.gen_request_header(timestamp)
        # 请求体
        body = self.gen_request_body(phone,code)
        #请求云通信接口
        data = self.request_yuntongxin_api(url,header,json.dumps(body))
        return data


if __name__ == '__main__':
    config = {
        "accountSid":"8a216da86eb206c4016exxxxxxx", # 主账户id,控制台首页获取
        "authToken":"e46b476182d94dc094e8xxxxxx", # 令牌
        "appId":"8a216da86eb206c4016ec46cxxxxxx", # 应用id
        "templateId":"1" # 模版id
    }
    phone = '199xxxxx'  # 手机号，如果是多个手机号用英文的,分割 比如说13200000000,13300000000
    code = '123' # 验证码
    yun = YunTongXin(**config)
    res = yun.run(phone,code)
    print(res)
    
    
  #将脚本内容放入command模块中，后置传参第一个位置