前言
运维故障排障速度往往与监控系统体系颗粒度成正比，监控到位才能快速排障
在部署这套系统之前，平台所有系统日志都由Graylog+Zabbix，针对日志出现的错误关键字进行告警，这种做法在运维工作开展过程中暴露出多个不足点，不详述；在考虑多方面原因后，最终对日志告警系统进行更换，选用的方案是：ELK+Kafka+Filebeat+Elastalert
本文主要以两个需求为主轴做介绍
非工作时间服务器异常登录告警系统日志出现错误关键字告警
架构
服务选型
部署
本文采用的操作系统 ：CentOS release 6.6
Filebeat
# 下载源$ curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.2.3-x86_64.rpm# 安装$ sudo rpm -vi filebeat-6.2.3-x86_64.rpm
Logstash
# 导入Yum源$ rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch$ cat <<EOF > /etc/yum.repos.d/logstash.repo[logstash-6.x]name=Elastic repository for 6.x packagesbaseurl=https://artifacts.elastic.co/packages/6.x/yumgpgcheck=1gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearchenabled=1autorefresh=1type=rpm-mdEOF# 安装yum install logstash -y
Elastalert
# pip直接安装$ pip install elastalert# 如果出现依赖包报错，以下为常用开发所需依赖包$ yum install -y zlib openssl openssl-devel gcc gcc-c++ Xvfb libXfont Xorg libffi libffi-devel python-cffi python-devel libxslt-devel libxml2-devel zlib-devel bzip2-devel xz-libs wget
配置
Filebeat
/etc/filebeat/filebeat.yml
filebeat.config:prospectors:path: /etc/filebeat/conf/*.ymlreload.enabled: truereload.period: 10soutput.kafka:# kafkaNode为Kafaka服务所在服务器 hosts: ["kafkaNode:9092"]# 索引取fields.out_topictopic: "%{[fields][out_topic]}"partition.round_robin:reachable_only: false
/etc/filebeat/conf/base.yml
# 收集系统日志- type: logpaths: - /var/log/messages- /var/log/syslog*exclude_files: [".gz$"]exclude_lines: ["ssh_host_dsa_key"]tags: ["system_log"]scan_frequency: 1sfields:# 新增字段用于辨别来源客户端server_name: client01# 索引out_topic: "system_log"multiline:pattern: "^\\s"match: after# 收集登录日志- type: logpaths:- /var/log/secure*- /var/log/auth.log*tags: ["system_secure"]exclude_files: [".gz$"]scan_frequency: 1sfields:server_name: client01out_topic: "system_secure"multiline:pattern: "^\\s"match: after
Logstash
/etc/logstash/conf.d/system_log.conf
input {kafka {bootstrap_servers => "kafkaNode:9092"consumer_threads => 3topics => ["system_log"]auto_offset_reset => "latest"codec => "json"}}filter {# 排除logstash日志if [source] == "/var/log/logstash-stdout.log" {drop {}}if [fields][out_topic] == "system_log" {date {match => [ "[system][syslog][timestamp]", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]}grok {match => { "message" => ["%{SYSLOGTIMESTAMP:[system][syslog][timestamp]} %{SYSLOGHOST:[system][syslog][hostname]} %{DATA:[system][syslog][program]}(?:\[%{POSINT:[system][syslog][pid]}\])?: %{GREEDYMULTILINE:[system][syslog][message]}"] }pattern_definitions => { "GREEDYMULTILINE" => "(.|\n)*" }remove_field => "message"}}}output {elasticsearch {hosts => ["<亚马逊ES地址>"]index => "%{[fields][out_topic]}_%{+YYYYMMdd}"document_type => "%{[@metadata][type]}"}}
/etc/logstash/conf.d/secure_log.conf
input {kafka {bootstrap_servers => "kafkaNode:9092"consumer_threads => 3topics => ["system_secure"]auto_offset_reset => "latest"codec => "json"}}filter {if [fields][out_topic] == "system_secure" {grok {match => { "message" => ["%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} %{DATA:[system][auth][ssh][method]} for (invalid user )?%{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]} port %{NUMBER:[system][auth][ssh][port]} ssh2(: %{GREEDYDATA:[system][auth][ssh][signature]})?","%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} user %{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]}","%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: Did not receive identification string from %{IPORHOST:[system][auth][ssh][dropped_ip]}","%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sudo(?:\[%{POSINT:[system][auth][pid]}\])?: \s*%{DATA:[system][auth][user]} :( %{DATA:[system][auth][sudo][error]} ;)? TTY=%{DATA:[system][auth][sudo][tty]} ; PWD=%{DATA:[system][auth][sudo][pwd]} ; USER=%{DATA:[system][auth][sudo][user]} ; COMMAND=%{GREEDYDATA:[system][auth][sudo][command]}","%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} groupadd(?:\[%{POSINT:[system][auth][pid]}\])?: new group: name=%{DATA:system.auth.groupadd.name}, GID=%{NUMBER:system.auth.groupadd.gid}","%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} useradd(?:\[%{POSINT:[system][auth][pid]}\])?: new user: name=%{DATA:[system][auth][user][add][name]}, UID=%{NUMBER:[system][auth][user][add][uid]}, GID=%{NUMBER:[system][auth][user][add][gid]}, home=%{DATA:[system][auth][user][add][home]}, shell=%{DATA:[system][auth][user][add][shell]}$","%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} %{DATA:[system][auth][program]}(?:\[%{POSINT:[system][auth][pid]}\])?: %{GREEDYMULTILINE:[system][auth][message]}"] }pattern_definitions => {"GREEDYMULTILINE"=> "(.|\n)*"}remove_field => "message"}}}output {elasticsearch {hosts => ["<亚马逊ES地址>"]index => "%{[fields][out_topic]}_%{+YYYYMMdd}"document_type => "%{[@metadata][type]}"}}
Kafka
# 导入rpm --import https://packages.confluent.io/rpm/4.0/archive.keycat <<EOF > /etc/yum.repos.d/confluent.repo[Confluent.dist]name=Confluent repository (dist)baseurl=https://packages.confluent.io/rpm/4.0/6gpgcheck=1gpgkey=https://packages.confluent.io/rpm/4.0/archive.keyenabled=1[Confluent]name=Confluent repositorybaseurl=https://packages.confluent.io/rpm/4.0gpgcheck=1gpgkey=https://packages.confluent.io/rpm/4.0/archive.keyenabled=1EOFyum install confluent-platform-oss-2.11
Elastalert
Elastalert可以部署到任何一台能够读取到ES的服务器上；配置文件中
modules.eagle_post.EagleAlerterblacklist_v2经过修改，后面会介绍到
rules/system_log.yaml
es_host: <亚马逊ES地址>es_port: 80name: system log ruletype: blacklist_v2index: system_log*timeframe:minutes: 1# 监控keycompare_key: system.syslog.message# 出现下面任意关键字将告警，按需添加blacklist_v2:- "ERROR"- "error"alert: "modules.eagle_post.EagleAlerter"eagle_post_url: "<eagle>"eagle_post_all_values: Falseeagle_post_payload:server: "fields.server_name"info: "system.syslog.message"source: "source"
rules/system_log.yaml
es_host: <亚马逊ES地址>es_port: 80name: system secure ruletype: frequencyindex: system_secure*num_events: 1timeframe:minutes: 1filter:- query:wildcard:system.auth.user : "*"alert: "modules.eagle_post.EagleAlerter"eagle_post_url: "<eagle>"eagle_post_all_values: False# 非工作时间eagle_time_start: "09:00"eagle_time_end: "18:00"eagle_post_payload:user: "system.auth.user"server: "fields.server_name"ip: "system.auth.ssh.ip"event: "system.auth.ssh.event"
Elastalert
自定义type与alert
为了能够将告警接入到Eagle（自研统一接口平台）在尝试使用http_post做告警类型过程中，发现无法传入ES结果作为POST参数，所以对其进行简单修改，新增类型，实现能够无缝接入Eagle
Alert
moudules/eagle_post.py
将文件夹保存到
site-packages/elastalert
import jsonimport requestsimport dateutil.parserimport datetimefrom elastalert.alerts import Alerterfrom elastalert.util import EAExceptionfrom elastalert.util import elastalert_loggerfrom elastalert.util import lookup_es_keyclass EagleAlerter(Alerter):def __init__(self, rule):super(EagleAlerter, self).__init__(rule)# 设定时间有效范围self.post_time_start = self.rule.get('eagle_time_start','00:00')self.post_time_end = self.rule.get('eagle_time_end','00:00')# post链接self.post_url = self.rule.get('eagle_post_url','')self.post_payload = self.rule.get('eagle_post_payload', {})self.post_static_payload = self.rule.get('eagle_post_static_payload', {})self.post_all_values = self.rule.get('eagle_post_all_values', False)self.post_lock = Falsedef alert(self, matches):if not self.post_url:elastalert_logger.info('Please input eagle url!')return Falsefor match in matches:# 获取所有payloadpayload = match if self.post_all_values else {}# 构建字典for post_key, es_key in self.post_payload.items():payload[post_key] = lookup_es_key(match, es_key)# 获取当前时间login_time = datetime.datetime.now().time()# 获取时间限制time_start = dateutil.parser.parse(self.post_time_start).time()time_end = dateutil.parser.parse(self.post_time_end).time()# 如果在时间范围内，将不做告警self.post_lock = False if login_time > time_start and \login_time < time_end else True# 合并两种类型payloaddata = self.post_static_payloaddata.update(payload)# 发送告警if self.post_lock:myRequests = requests.Session()myRequests.post(url=self.post_url,data=data,verify=False)elastalert_logger.info("[-] eagle alert sent.")else:elastalert_logger.info("[*] nothing to do.")def get_info(self):return {'type': 'http_post'}
type
在使用blaklist过程发现改类型是全匹配，为了方便编写配置文件，所以对其做了简单修改
elastalert/ruletypes.py
# 新增class BlacklistV2Rule(CompareRule):required_options = frozenset(['compare_key', 'blacklist_v2'])def __init__(self, rules, args=None):super(BlacklistV2Rule, self).__init__(rules, args=None)self.expand_entries('blacklist_v2')def compare(self, event):term = lookup_es_key(event, self.rules['compare_key'])# 循环配置文件， 这种做法对性能有一定的损耗，在没找到更合适的解决方案前，就采取这种方式for i in self.rules['blacklist_v2']:if i in term:return True return False
elastalert/config.py
# 新增rules_mapping = {'frequency': ruletypes.FrequencyRule,'any': ruletypes.AnyRule,'spike': ruletypes.SpikeRule,'blacklist': ruletypes.BlacklistRule,'blacklist_v2': ruletypes.BlacklistV2Rule,'whitelist': ruletypes.WhitelistRule,'change': ruletypes.ChangeRule,'flatline': ruletypes.FlatlineRule,'new_term': ruletypes.NewTermsRule,'cardinality': ruletypes.CardinalityRule,'metric_aggregation': ruletypes.MetricAggregationRule,'percentage_match': ruletypes.PercentageMatchRule,}
elastalert/schema.yaml
# 新增- title: BlacklistV2required: [blacklist_v2, compare_key]properties:type: {enum: [blacklist_v2]}compare_key: {'items': {'type': 'string'},'type': ['string', 'array']}blacklist: {type: array, items: {type: string}}
打进Docker
做了个简单DockerFile做参考
FROM python:2.7-alpineENV SITE_PACKAGES /usr/local/lib/python2.7/site-packages/elastalertWORKDIR /opt/elastalertRUN apk update &&apk add gcc ca-certificates openssl-dev openssl libffi-dev gcc musl-dev tzdata openntpd && \ pip install elastalert && cp -rf /usr/share/zoneinfo/Asia/Taipei /etc/localtimeCOPY ./ /opt/elastalertCMD ["/opt/elastalert/start.sh"]
start.sh
#!/bin/shSITE_PATH=/usr/local/lib/python2.7/site-packages/elastalertCONFIG=/opt/elastalert/config/config.yamlMODULES=/opt/elastalert/modulesif [ -n "${MODULES}" ]then\cp -rf ${MODULES} ${SITE_PATH}echo "[-] Copy ${MODULES} to ${SITE_PATH}"fi\cp -rf elastalert/* ${SITE_PATH}/echo "[-] Copy elastalert/* to ${SITE_PATH}"python -m elastalert.elastalert --verbose --config ${CONFIG}
基础工作准备就绪，加入Bee容器管理平台完成自动构建。
实现效果
碰到的坑
Zookeeper
问题描述
老版
Kafaka依赖Zookeeper，默认安装时注册地址为:localhost，导致问题的现象：
filebeat错误日志
-04-25T09:14:55.590+0800 INFO kafka/log.go:36 client/metadata fetching metadata for [[[system_log] kafkaNode:9092]] from broker %!s(MISSING)-04-25T09:14:55.591+0800 INFO kafka/log.go:36 producer/broker/[[0]] starting up-04-25T09:14:55.591+0800 INFO kafka/log.go:36 producer/broker/[[0 %!d(string=system_log) 0]] state change to [open] on %!s(MISSING)/%!d(MISSING)-04-25T09:14:55.591+0800 INFO kafka/log.go:36 producer/leader/[[system_log %!s(int32=0) %!s(int32=0)]]/%!d(MISSING) selected broker %!d(MISSING)-04-25T09:14:55.591+0800 INFO kafka/log.go:36 producer/leader/[[system_secure %!s(int32=0) %!s(int=3)]]/%!d(MISSING) state change to [retrying-%!d(MISSING)]-04-25T09:14:55.591+0800 INFO kafka/log.go:36 producer/leader/[[system_secure %!s(int32=0) %!s(int32=0)]]/%!d(MISSING) abandoning broker %!d(MISSING)-04-25T09:14:55.592+0800 INFO kafka/log.go:36 producer/broker/[[0]] shut down-04-25T09:14:55.592+0800 INFO kafka/log.go:36 Failed to connect to broker [[localhost:9092 dial tcp [::1]:9092: getsockopt: connection refused]]: %!s(MISSING)
日志出现两个地址，一个是kafka地址，另外出现一个localhost地址。
这是因为filebeat已经跟kafaka建立了连接，但是从kafaka到zookeeper这一段找不到
解决方法
# get /brokers/ids/0 {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://localhost:9092"],"jmx_port":-1,"host":"localhost","timestamp":"1523429158364","port":9092,"version":4}cZxid = 0x1dctime = Wed Apr 11 14:45:58 CST mZxid = 0x1dmtime = Wed Apr 11 14:45:58 CST pZxid = 0x1dcversion = 0dataVersion = 0aclVersion = 0ephemeralOwner = 0x162b374170d0000dataLength = 188numChildren = 0# 发现注册地址是localhost,修改之set /brokers/ids/0 {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://kafkaNode:9092"],"jmx_port":9999,"host":"kafkaNode","timestamp":"1523429158364","port":9092,"version":4}
修改完重启，问题解决。
博文链接：http://a-//04/29/aws-elk/
