章
目
录
今天咱就来讲讲怎么用Logstash收集和处理FastAPI应用的日志。Logstash可是个厉害的数据处理“小能手”，它能从好多地方收集数据，对数据进行处理和转换，最后把数据送到像Elasticsearch这样的存储系统里。下面就一步步教大家具体怎么操作。
一、认识Logstash
Logstash是一款开源的数据收集引擎，就像一个数据“搬运工”，支持各种各样的数据源，像日志文件、系统消息队列，甚至数据库里的数据它都能收集。收集到数据后，还能对数据进行各种处理和转换，然后把处理好的数据发送到不同的目标，比如Elasticsearch、Kafka，或者通过邮件通知相关人员。
二、具体操作步骤
2.1 安装必要的包
要实现FastAPI应用和Logstash之间的通信，首先得安装python-logstash-async这个包，它能帮我们异步发送日志，这样应用就不会因为发送日志而卡顿。安装命令很简单，在终端输入下面这条命令就行：
pip install python-logstash-async
2.2 配置FastAPI应用
安装好包之后，就得在FastAPI应用里配置日志处理器，让它能把日志发送给Logstash。下面这段Python代码就是具体的配置过程：
from fastapi import FastAPI, Request
from logstash_async.handler import AsynchronousLogstashHandler
import logging
app = FastAPI()
# 设置Logstash服务的地址和端口，这里假设Logstash在本地运行，端口是6000
host = 'localhost'
port = 6000
# 获取一个日志记录器对象，名字是fastapi_logger
logger = logging.getLogger("fastapi_logger")
# 设置日志记录级别为INFO，意味着只记录INFO级别及以上的日志信息
logger.setLevel(logging.INFO)
# 创建一个异步Logstash处理器，用于将日志发送到指定的Logstash服务
async_handler = AsynchronousLogstashHandler(host, port, database_path=None)
# 将异步处理器添加到日志记录器中
logger.addHandler(async_handler)
# 定义一个中间件，用来捕获每次请求和响应的日志信息
@app.middleware("http")
async def log_requests(request: Request, call_next):
# 记录请求的方法和URL
logger.info(f"Request: {request.method} {request.url}")
# 调用下一个中间件或路由处理函数，并获取响应
response = await call_next(request)
# 记录响应的状态码和响应体
logger.info(f"Response: {response.status_code} {response.body}")
return response
@app.get("/")
async def root():
return {"message": "Hello World"}
这段代码里，先引入了需要的库，然后设置了Logstash的地址和端口，创建了日志记录器和异步处理器，最后定义了一个中间件来记录请求和响应的日志。
2.3 配置Logstash
接下来要配置Logstash，让它能接收FastAPI应用发送过来的日志。在Logstash的配置文件logstash.conf里，我们这么写：
input {
tcp {
port => 6000
codec => json
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "fastapi_logs"
}
}
这里设置了Logstash通过TCP协议监听6000端口，并且使用JSON编解码器来解析接收到的数据。输出部分配置了将处理后的数据发送到本地的Elasticsearch服务，索引名字是fastapi_logs。
2.4 运行Logstash
配置好Logstash后，就可以启动Logstash服务了。在终端输入下面这条命令：
logstash -f logstash.conf
这个命令会让Logstash按照logstash.conf里的配置来运行。
2.5 运行FastAPI应用
最后，启动FastAPI应用。在终端输入：
uvicorn main:app --reload
这样FastAPI应用就启动起来了，并且会自动重新加载代码，方便开发调试。
当你访问FastAPI应用时，应用产生的响应日志就会被发送到Logstash，然后Logstash再把这些日志存储到Elasticsearch里。通过Kibana，你就能实时查看和分析这些日志了。
三、这种方案的优势
使用python-logstash-async包来异步发送日志，最大的好处就是应用不会因为发送日志而被阻塞，这样就能提高应用的性能。而且Logstash的配置很灵活，能帮我们处理和转换日志数据，让日志数据更适合存储和分析。
四、更多拓展案例
4.1 多源数据收集
Logstash的本事可不止处理FastAPI应用的日志。它还能从多个不同的数据源收集数据，像日志文件、Redis、Kafka这些都不在话下。
4.2 数据转换和过滤
Logstash有很多过滤器插件，比如Grok、JSON，用这些插件可以对收集到的数据进行解析和转换，让数据变得更“规整”，方便后续处理。
4.3 输出到多个目标
Logstash支持把处理好的数据输出到多个不同的目标，除了Elasticsearch，还能输出到Kafka、文件等等，满足不同的业务需求。
按照上面这些步骤，就能轻松用Logstash收集和处理FastAPI应用的日志啦。大家在实际项目中可以根据自己的需求灵活调整配置，希望对大家有所帮助！
