# 修改Filebeat和ELK配置 #### 修改Filebeat配置 Filebeat目录和文件结构如下所示： ``` |_ filebeat |_ log |_ docker-compose.yml |_ filebeat.yml ``` 1. 登录SHUBAO组件所在节点，切换**root** 用户，执行以下命令创建Filebeat相关目录和配置文件。 用户可自定义配置文件所在路径，此处以在根目录下创建filebeat目录为例进行说明。 ``` mkdir -p /filebeat/log touch /filebeat/filebeat.yml touch /filebeat/docker-compose.yml ``` 其中，log目录用于存储Filebeat收集到的日志数据以及其它元数据信息，即使容器Filebeat被删除并重新启动，会继续读取该目录中的内容（如registry），并在收集数据时更新该目录，这样可确保Filebeat只收集新的日志数据，避免将已经发送过的数据再次发送。 2. 执行以下命令为log目录配置权限。 ``` chmod 777 /filebeat/log ``` 3. 编辑filebeat.yml文件，在文件中增加如下配置。 ``` filebeat.inputs: - type: filestream # id需要替换为数字，不同SHUBAO组件所在节点之间相互唯一。例如SHUBAO组件所在节点有3个，则id可分别配置为site_log_filebeat_01，site_log_filebeat_02，site_log_filebeat_03 id: site_log_filebeat_id enabled: true paths: - /usr/share/filebeat/access_log/access_public.log fields: # index_name为在ElasticSearch存储和Kibana可视化显示时的指定索引名称，可自定义，若需要区分日志来源，则不同SHUBAO组件所在节点之间配置为不一样。 es_index_name: index_name fields_under_root: true close_inactive: 1m clean_inactive: 72h output.logstash: # 填写ELK组件所服务器的IP和Logstash对外开放的端口号（在ELK的logstash.conf文件中配置，默认为5044），需自行配置防火墙端口，开放访问权限。 hosts: ["ELK_LOGSTASH_IP:5044"] ``` 4. 编辑docker-compose.yml文件，在文件中增加如下配置。 ``` version: '3' services: filebeat: image: docker.elastic.co/beats/filebeat:8.7.1 network_mode: host restart: always container_name: filebeat volumes: - ./filebeat.yml:/usr/share/filebeat/filebeat.yml # 如果需要把Filebeat收集到的日志数据保存到本地log目录下，则取消本行代码注释 # - ./log:/usr/share/filebeat/data # filebeat读取本地日志，该本地日志地址需要根据实际待监控路径的地址进行配置，相对路径或绝对路径均可。 # 多地址监听：若在一台服务器中需要监听多个地址，则需在filebeat.yml的path和docker-compose.yml的volumes中增加新的配置，配置方式与该例子相同。 - /opt/roma/logs/apic-shubao/run/access_public.log:/usr/share/filebeat/access_log/access_public.log ``` 5. 重复步骤[1]-[4]，登录其他SHUBAO组件所在节点修改Filebeat配置。 #### 修改ELK配置 ELK目录和文件结构如下所示： ``` |_ ELK |_ elasticsearch |_ elasticsearch.yml |_ kibana |_ kibana.yml |_ logstash |_ log |_ logstash.conf |_ logstash.yml |_ python_elk |_ send_msg.py |_ docker-compose.yml ``` 1. 登录ELK所部署的服务器，切换**root** 用户，执行以下命令创建ELK相关目录和配置文件。 用户可自定义配置文件所在路径，此处以在根目录下创建ELK目录为例进行说明。 ``` mkdir -p /ELK/logstash/log /ELK/elasticsearch /ELK/kibana /ELK/python_elk touch /ELK/logstash/logstash.yml touch /ELK/logstash/logstash.conf touch /ELK/elasticsearch/elasticsearch.yml touch /ELK/kibana/kibana.yml touch /ELK/python_elk/send_msg.py touch /ELK/docker-compose.yml ``` 其中，log目录用于挂载Logstash的数据目录，存储Logstash处理后的数据和元数据信息，即使容器Logstash被删除，日志数据也不会丢失，并且可以轻松的备份和恢复数据。 2. 执行以下命令为log目录配置权限。 ``` chmod 777 /ELK/logstash/log ``` 3. 编辑logstash.conf文件，在文件中增加如下配置。 ``` input{ beats { port => 5044 } } # 过滤解析规则 filter { grok { match => { "message" => "%{DATA:my_remote_addr} %{DATA:request_id} %{DATA:api_id} %{DATA:user_name} %{DATA:app_id} \[%{HTTPDATE:timestamp}\] %{NUMBER:request_time:float} %{DATA:request_method} %{URIPROTO:scheme}://%{DATA:host} %{DATA:router_uri} %{DATA:server_protocol} %{NUMBER:status:int} %{NUMBER:bytes_sent:int} %{NUMBER:request_length:int} \"%{DATA:http_user_agent}\" \"%{DATA:http_x_forwarded_for}\" \"%{DATA:upstream_addr}\" %{DATA:upstream_uri} \"%{DATA:upstream_status}\" \"%{DATA:upstream_connect_time}\" \"%{DATA:upstream_header_time}\" \"%{DATA:upstream_response_time}\" %{WORD}-%{WORD}-%{INT} %{DATA:all_upstream_response_time} %{DATA:errorType} %{DATA:auth_type} %{DATA:access_model1} %{DATA:access_model2} %{DATA:inner_time} %{DATA:proxy_protocol_vni} %{DATA:proxy_protocol_vpce_id} \"%{DATA:proxy_protocol_addr}\" %{NUMBER:body_bytes_sent} %{DATA:api_name} %{DATA:app_name} %{DATA:provider_app_id} %{DATA:provider_app_name} \"%{DATA:custom_data_log1}\" \"%{DATA:custom_data_log2}\" \"%{DATA:custom_data_log3}\" \"%{DATA:custom_data_log4}\" \"%{DATA:custom_data_log5}\" \"%{DATA:custom_data_log6}\" \"%{DATA:custom_data_log7}\" \"%{DATA:custom_data_log8}\" \"%{DATA:custom_data_log9}\" \"%{DATA:custom_data_log10}\" %{WORD:response_source}?" } } date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] target => "@timestamp" } if [upstream_response_time] =~ /^\d+(\.\d+)?$/ { mutate { add_field => { "duration" => "%{upstream_response_time}" } convert => { "upstream_response_time" => "float" } } } } output{ elasticsearch { # ElasticSearch对外开放的端口默认为9200，如果修改了端口，此处需同步修改 hosts => ["http://localhost:9200"] index => "site_access_log_%{[es_index_name]}_%{+YYYY-MM-dd}" # username和password自定义，需与部署ELK时设置的用户名和密码保持一致 user => "username" password => "password" } stdout { codec => rubydebug } # 当状态异常或调用超时，触发异常告警通知 if ([status] == 504 or [status] == 502 or [status] == 404) or ([duration] and [upstream_response_time] >= 60.0) { # http信息可根据实际需要修改 http { retry_failed => "false" http_method => "post" # 配置告警信息监听地址，端口默认为9202 url => "http://localhost:9202" format => "message" content_type => "application/x-www-form-urlencoded" # 自定义告警信息内容，可根据实际需要修改 message => "ip: %{my_remote_addr}, 告警时间: %{timestamp}, 接口: %{router_uri}, 状态码: %{status}, 调用时长: %{upstream_response_time}" } } } ``` 4. 编辑logstash.yml文件，在文件中增加如下配置。 ``` path.config: /usr/share/logstash/pipeline/*.conf xpack.monitoring.enabled: true # ElasticSearch对外开放的端口默认为9200，如果修改了端口，此处需同步修改 xpack.monitoring.elasticsearch.hosts: ["http://localhost:9200"] ``` 5. 编辑elasticsearch.yml文件，在文件中增加如下配置。 ``` cluster.name: "elasticsearch" network.host: "127.0.0.1" http.cors.enabled: true http.cors.allow-origin: "*" discovery.type: "single-node" xpack.security.enabled: true ``` 6. 编辑kibana.yml文件，在文件中增加如下配置。 ``` # 0.0.0.0表示所有客户端地址均可通过用户名及密码访问Kibana可视化界面，如果仅允许指定客户端IP访问，则修改为指定的客户端IP server.host: "0.0.0.0" # Kibana端口号，默认为5601，如果需要修改端口号，则取消本行代码注释，并修改实际使用的端口号 # server.port: 5601 # ElasticSearch对外开放的端口默认为9200，如果修改了端口，此处需同步修改 elasticsearch.hosts: [ "http://localhost:9200" ] monitoring.ui.container.elasticsearch.enabled: true # username和password自定义，需与部署ELK时设置的用户名和密码保持一致 elasticsearch.username: 'username' elasticsearch.password: 'password' ``` 7. 编辑docker-compose.yml文件，在文件中增加如下配置。 ``` version: '3' services: logstash: image: docker.elastic.co/logstash/logstash:8.7.1 restart: always container_name: logstash network_mode: host volumes: - ./logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf - ./logstash/logstash.yml:/usr/share/logstash/config/logstash.yml # 如果需要把Logstash处理后的日志数据保存到本地log目录下，则取消本行代码注释 # - ./logstash/log/:/usr/share/logstash/data/:rw elasticsearch: image: docker.elastic.co/elasticsearch/elasticsearch:8.7.1 restart: always container_name: elasticsearch # 如果ElasticSearch占用资源过高，导致容器无法成功启动，可取消此处的代码注释，根据服务器情况对资源使用进行限制 # deploy: # resources: # limits: # memory: 4g # cpus: '0.8' network_mode: host volumes: - ./elasticsearch/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml kibana: image: docker.elastic.co/kibana/kibana:8.7.1 restart: always container_name: kibana network_mode: host volumes: - ./kibana/kibana.yml:/usr/share/kibana/config/kibana.yml python_elk: # python_image_id为python镜像的镜像ID image: python_image_id restart: always container_name: python_elk network_mode: host volumes: - ./python_elk/send_msg.py:/send_msg.py build: . command: python3 -u /send_msg.py ``` 8. 编辑send_msg.py文件，在文件中增加如下配置。 以告警通知发送到一个API为例进行说明，且该API使用Token鉴权。 ``` from http.server import BaseHTTPRequestHandler, HTTPServer import json import requests import os class RequestHandler(BaseHTTPRequestHandler): def do_POST(self): content_length = int(self.headers['Content-Length']) content = self.rfile.read(content_length).decode('utf-8') print("\n\n\n\n\nPath: {}\nHeaders: {}Body: {}\n".format(str(self.path), str(self.headers), content)) # url_address为获取Token的URL地址，appid和appsecret为获取Token所需的鉴权信息。 token_get_url = "url_address" params = { "appid": "appid", "appsecret": "appsecret" } response = requests.get(token_get_url, params=params) if response.status_code == 200: response_json = json.loads(response.content.decode('utf-8')) access_token = response_json['d']['access_token'] # 配置要接收告警通知的API信息，msg_send_url为API的URL地址，identity、ucs_type和wid为API的相关请求参数，可根据API实际情况增删和修改参数 msg_send_url = "msg_send_url" identity = "identity" ucs_type = ucs_type wid = "wid" payload = "access_token={}&numbers[0]={}&ucs_type={}&content[msgtype]=text&content[text][content]={}&wid={}".format(access_token, identity, ucs_type, content, wid) print("content: {}".format(content)) print("payload: {}".format(payload)) ans = os.system('curl {} -d "{}"'.format(msg_send_url, payload)) print("os.system: ", ans) # 告警信息监听端口默认为9202，如果修改了端口，此处需同步修改 def run_server(port=9202): with HTTPServer(("localhost", port), RequestHandler) as httpd: print("Server listening on port {}...", port) httpd.serve_forever() if __name__ == "__main__": run_server() ```