ELK 日志堆栈（Elasticsearch B.V. 官方称“Elastic Stack”），是由 ElasticSearch、Kibana 以及 Logstash 等组件组成的工具集，被业界广泛用于进行日志处理等任务。其中 ElasticSearch 组件本身也是一个强大不可替代的开源搜索引擎。
ElasticSearch 及其周边组件的功能非常强大，该堆栈可配置的灵活部分也很多。比如分布式集群、索引生命周期管理以及索引副本等功能。为简单起见，本文章将介绍在单台服务器配置 ELK 堆栈，以实现将 Python 应用程序的日志输出接入 ElasticSearch 的步骤。关于 ElasticSearch 的更多信息，请访问：https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html
安装 ELK 堆栈
Elastic 官方提供了多种在机器上安装 ELK 堆栈的方式。我们以一台安装有 Ubuntu 22.04 LTS 系统的服务器为例，演示如何通过 apt 包管理器安装 ELK 日志堆栈的方法。对于其他操作系统或安装方式，请参阅官方文档以了解更多信息。
今天我们的示例将会安装 ELK 堆栈的全部组件：ElasticSearch，本身是一个性能强大且易于拓展的数据和搜索引擎；Kibana，可视化数据并提供图形界面的组件；以及 Logstash，从任何方式收集日志，并输出至任意位置的可配置的数据流管线；这些组件将驱动整个 Python 应用程序的日志收集与处理流程。
安装 Java Runtime Environment（JRE）
ELK 堆栈主要由 Java 编写，因此第一步骤是要在系统中安装 Java 运行时环境（Java Runtime Environment，JRE）。我们以开源的 OpenJDK 为例，使用以下命令进行安装：
sudo apt install default-jre在 Ubuntu 22.04 LTS 系统中，这将会安装 openjdk-11-jre 包。
安装配置 ElasticSearch
ElasticSearch 通过 apt 包管理器的安装非常简单，使用以下命令进行安装：
sudo apt install elasticsearch安装完成后，注意安装脚本给出的超级用户的密码，以及最后一部分提示：
Reset the password of the elastic built-in superuser with
'/usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic'.
Generate an enrollment token for Kibana instances with
'/usr/share/elasticsearch/bin/elasticsearch-create-enrollment-token -s kibana'.
Generate an enrollment token for Elasticsearch nodes with
'/usr/share/elasticsearch/bin/elasticsearch-create-enrollment-token -s node'.ElasticSearch 默认的超级用户的用户名是 elastic，密码是安装日志上显示的密码。如果忘记，可以通过执行上方日志的命令将其重置。
ElasticSearch 安装完毕后，使用 root 权限编辑 /etc/elasticsearch/elasticsearch.yml 文件，并修改或添加以下配置：
network.host: localhost
http.host: localhost安装配置 Kibana
使用以下命令在服务器上安装 Kibana：
sudo apt install kibanaKibana 安装好后，在服务器本地浏览器访问 http://localhost:5601/，以执行 Kibana 的初始化配置。配置过程中，可能需要填写 ElasticSearch 的配对令牌（enrollment token），该令牌的内容可以通过执行上一小节安装 ElasticSearch 后打印的提示命令获得。
如果用户是通过 SSH 连接到远程服务器进行配置的，可以考虑在 SSH 连接参数后追加
-L 5601:localhost:5601，通过 SSH 端口转发，从而在本地客户端浏览器中访问localhost:5601以达到相同效果
通过超级用户 elastic 以及对应密码登录 Kibana 面板后，我们鼓励新增两个单独的用户分别用作日志管理与日志推送，以实现权限控制。登录 Kibana 面板后，点击左侧的“Management” > “Stack Management”，然后首先新建用户的权限。我们以两个用户名称为 user 和 python-logger 为例：
| 权限名称 | 集群权限 | 索引权限 |
|---|---|---|
user | kibana_admin | 对于 *（全部索引），拥有 all 权限 |
python-logger | monitormanagement | 对于 *（全部索引），拥有 all 权限 |
添加完用户权限后，我们实际添加名为 user 和 python-logger 的用户。添加过程中，要记得赋予用户同名的用户权限。事实上，这些用户名和权限名称都是可以自定义的，这里仅为示例。
添加完成后，编辑 /etc/kibana/kibana.yml 文件，添加或修改以下配置项：
elasticsearch.hosts: ['https://localhost:9200']不想每次登录 Kibana 都输入用户名密码？
如果你不希望每次登录 Kibana 面板时都输入用户名密码，可以在 /etc/kibana/kibana.yml 文件中追加以下配置项。注意，务必锁好 Kibana 面板的访问控制，否则任意用户都可以看到你的日志（并有权限篡改它们！）：
xpack.security.authc.providers:
anonymous.anonymous1:
order: 0
credentials:
username: "你的用户名"
password: "你的密码"安装配置 Logstash
Logstash 是控制日志从哪里收集、输出到哪里的一个“管道”管理器。我们通过以下命令安装 Logstash：
sudo apt install logstashLogstash 的配置位于 /etc/logstash/conf.d/ 目录中，如果该目录中没有文件，则新建任何后缀名为 .conf 的文件，均可以添加一个“日志管道”。作为示例，我们在该目录中创建一个名为 01-python.conf 的文件，具体内容如下：
input {
tcp {
port => 12345
codec => json
tags => "python-service"
}
}
output {
if "python-service" in [tags] {
elasticsearch {
index => "python-service-%{+YYYY-MM-dd}"
ssl_enabled => true
ssl_verification_mode => none
user => "python-logger"
password => "对应密码"
}
}
}在上述文件中，我们定义了一个从 TCP 12345 端口接受 JSON 格式的日志输入，将其输出到本地 ElasticSearch 的日志管线。在 ElasticSearch 中，每个日志都需要在一个索引（Index）中。业界惯常的索引命名习惯是按天分割索引文件，也就是上边的 index => "python-service-%{+YYYY-MM-dd}" 一行。这将创建名称例如 python-service-2023-12-31、python-service-2024-01-01 这样的索引。
添加好 Logstash 的配置文件后，我们通过执行以下命令应用更改：
sudo systemctl restart logstash可以通过以下命令查看 Logstash 本身的运行日志，以诊断任何报错：
sudo journalctl -u logstash理论上重新启动 Logstash 后，我们可以通过 sudo netstat -unltp 注意到 Logstash 已经在监听 TCP 12345 端口。我们可以使用下列示例 Python 脚本向 Logstash 发送内容，并登录 Kibana 进行查看：
import logging
import logstash # 需要 pip install python-logstash
import sys
host = 'localhost'
test_logger = logging.getLogger('python-logstash-logger')
test_logger.setLevel(logging.INFO)
test_logger.addHandler(logstash.TCPLogstashHandler(host, 12345, version=1))
test_logger.error('Hello, world! This is an error')
test_logger.info('Hello, world! This is an info')
test_logger.warning('Hello, world! This is a warning')同样地，请为服务器端的 TCP 12345 端口（或者其他你喜欢的端口数字）设置恰当的访问权限控制，以避免互联网上的任意用户向日志索引中追加内容。
Logstash 本身的配置文件功能很多，可以从多种数据源及不同数据格式中接收日志，并输出到指定位置（不仅包括 ElasticSearch）。关于 Logstash 配置文档的更多信息，请查看：https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html
配置 Python 应用程序
到目前为止，ELK 堆栈的全部内容已经配置完成了。现在我们开始配置 Python 应用程序。如果你使用的是 Django，则需要在 LOGGING 配置项中进行类似这样的配置：
LOGGING = {
...
'handlers': {
'logstash': {
'level': 'DEBUG',
'class': 'logstash.LogstashHandler',
'host': 'localhost',
'port': 12345,
'version': 1, # Version of logstash event schema. Default value: 0 (for backward compatibility of the library)
'message_type': 'logstash', # 'type' field in logstash message. Default value: 'logstash'.
'fqdn': False, # Fully qualified domain name. Default value: false.
'tags': ['tag1', 'tag2'], # list of tags. Default: None.
},
},
'loggers': {
'django.request': {
'handlers': ['logstash'],
'level': 'DEBUG',
'propagate': True,
},
},
...
}如果你使用其他框架或直接调用 logger 进行日志输出，配置方式可能像这样：
import logging
import logstash
import sys
host = 'localhost'
test_logger = logging.getLogger('python-logstash-logger')
test_logger.setLevel(logging.INFO)
test_logger.addHandler(logstash.LogstashHandler(host, 12345, version=1))配置日志自动清除
默认情况下 ElasticSearch 不会删除任何日志信息。这可能会导致日志量增大时，服务器磁盘空间不足。这里给出了一个我使用 Python 编写的自动清除 xx 天前索引的脚本，可以通过定时运行（如 crontab）脚本的方式，实现日志的按天滚动清除：
代码详见
from datetime import datetime, timedelta
from typing import List
from dataclasses import dataclass
import requests
from requests.auth import HTTPBasicAuth
@dataclass(frozen=True)
class IndexConfig(object):
delete_threshold: timedelta
delete_no_suffix: bool = False
ELASTIC_USERNAME = "管理用户的用户名"
ELASTIC_PASSWORD = "对应密码"
ELASTIC_AUTH = HTTPBasicAuth(username=ELASTIC_USERNAME, password=ELASTIC_PASSWORD)
ELASTIC_API_BASEURL = "https://localhost:9200"
ELASTIC_INDEX_PREFIXES_CONFIG = {
"python-service": IndexConfig(delete_threshold=timedelta(days=7)) # 在这里设置要自动删除的索引名称，以及间隔几天自动清除旧索引
}
def get_all_indexes() -> List[str]:
return requests.get(f"{ELASTIC_API_BASEURL}/{index_prefix}*", auth=ELASTIC_AUTH, verify=False).json()
def delete_index(index_name: str):
print(f"Delete index: {index_name}")
return requests.delete(f"{ELASTIC_API_BASEURL}/{index_name}", auth=ELASTIC_AUTH, verify=False).json()
for index_prefix, index_config in ELASTIC_INDEX_PREFIXES_CONFIG.items():
for index in get_all_indexes():
index_suffix = index[len(index_prefix)+1:]
if not index_suffix:
if index_config.delete_no_suffix:
delete_index(index)
continue
try:
index_datetime = datetime.fromisoformat(index_suffix)
except ValueError:
print(f"Warning: index_datetime invalid parsing: {index_suffix}.")
continue
if datetime.now() - index_datetime >= index_config.delete_threshold:
delete_index(index)上述脚本需要 Python 的第三方库 requests。
其他登陆方式
Kibana 可以配置更多的登陆方式，比如使用 Google 单点登录（Google SSO），但这些功能属于 ELK 的高级订阅层级支持的功能。如果要了解更多信息，可以参考 https://www.elastic.co/guide/en/cloud/current/ec-securing-clusters-oidc-op.html。