基于AWS

96818

前言

　　运维故障排障速度往往与监控系统体系颗粒度成正比，监控到位才能快速排障

　　在部署这套系统之前，平台所有系统日志都由Graylog+Zabbix，针对日志出现的错误关键字进行告警，这种做法在运维工作开展过程中暴露出多个不足点，不详述；在考虑多方面原因后，最终对日志告警系统进行更换，选用的方案是：ELK + Kafka+  Filebeat + Elastalert
　　本文主要以两个需求为主轴做介绍

• 非工作时间服务器异常登录告警
  
• 系统日志出现错误关键字告警
  

架构

服务选型


name
version
info




Amazon Elasticsearch Service
v6.2
AWK官网部署教程


Logstash
v6.2.3
选用与ES相同版本


Filebeat
v6.2.3
选用与ES相同版本


Confluent(Kafka)
v4.0
这里推荐 Confluent 的版本，Confluent 是 kafka 作者 Neha Narkhede 从 Linkedin 出来之后联合 LinkedIn 前员工创建的大数据公司，专注于 kafka 的企业应用。


Elastalert
v0.1.29
原先考虑采用X-Pack但由于AWS目前还不支持

部署

　　本文采用的操作系统 ：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  "kafkaNode:9092"
consumer_threads => 3
topics => ["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 => [""]
index => "%{[fields][out_topic]}_%{+YYYYMMdd}"
document_type => "%{[@metadata][type]}"
}
}

　　/etc/logstash/conf.d/secure_log.conf

input {
kafka {
bootstrap_servers => "kafkaNode:9092"
consumer_threads => 3
topics => ["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 => [""]
index => "%{[fields][out_topic]}_%{+YYYYMMdd}"
document_type => "%{[@metadata][type]}"
}
}

　　Kafka


# 导入
rpm --import https://packages.confluent.io/rpm/4.0/archive.key
cat  time_start and \
login_time < time_end else True
# 合并两种类型payload
data = self.post_static_payload
data.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: BlacklistV2
required: [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-alpine
ENV SITE_PACKAGES /usr/local/lib/python2.7/site-packages/elastalert
WORKDIR /opt/elastalert
RUN 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/localtime
COPY ./ /opt/elastalert
CMD ["/opt/elastalert/start.sh"]
　　start.sh

#!/bin/sh
SITE_PATH=/usr/local/lib/python2.7/site-packages/elastalert
CONFIG=/opt/elastalert/config/config.yaml
MODULES=/opt/elastalert/modules
if [ -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错误日志

2018-04-25T09:14:55.590+0800    INFO    kafka/log.go:36    client/metadata fetching metadata for [[[system_log] kafkaNode:9092]] from broker %!s(MISSING)
2018-04-25T09:14:55.591+0800    INFO    kafka/log.go:36    producer/broker/[[0]] starting up
2018-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)
2018-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)
2018-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)]
2018-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)
2018-04-25T09:14:55.592+0800    INFO    kafka/log.go:36    producer/broker/[[0]] shut down
2018-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 = 0x1d
ctime = Wed Apr 11 14:45:58 CST 2018
mZxid = 0x1d
mtime = Wed Apr 11 14:45:58 CST 2018
pZxid = 0x1d
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x162b374170d0000
dataLength = 188
numChildren = 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-cat.cn/2018/04/29/aws-elk/