Observability：将 OpenTelemetry 添加到你的 Flask 应用程序

作者：来自 Elastic jessgarson

待办事项列表可以帮助管理与假期计划相关的所有购物和任务。使用 Flask，你可以轻松创建待办事项列表应用程序，并使用 Elastic 作为遥测后端，通过 OpenTelemetry 对其进行监控。

Flask 是一个轻量级的 Python Web 框架，可让你轻松创建应用程序。OpenTelemetry 是一个开源的、与供应商无关的可观察性框架，它提供跨不同服务和平台的统一监控功能，允许与各种后端系统无缝集成。

这篇文章将引导你使用 OpenTelemetry Python 的 Elastic Distribution 来监控 Flask 中内置的待办事项列表应用程序。本文中概述的示例的完整代码可以在此处找到。

如何将 Elastic 连接到 OpenTelemetry？

OpenTelemetry 的一大优点是它可以灵活地与你的应用程序集成。使用 Elastic 作为遥测后端。你有几个选择；你可以使用 OpenTelemetry 收集器（官方 OpenTelmetry 语言客户端）连接到 APM（AWS Lambda 收集器导出器）。我们的文档可让你详细了解将 OpenTelemetry 连接到 Elastic 的选项。

你将使用 OpenTelemetry Python 的 Elastic Distribution 作为本文中的示例。此库是 OpenTelemetry Python 的一个版本，具有附加功能并支持将 OpenTelemetry 与 Elastic 集成。需要注意的是，此包目前处于预览阶段，不应在生产环境中使用。

使用 Flask 创建待办事项列表应用程序

在监控应用程序之前，你必须先创建它。本文将指导你创建一个简单的待办事项列表应用程序来帮助你跟踪任务。完成应用程序后，它将如下所示：

开始之前，你需要创建一个虚拟环境。创建虚拟环境后，你需要安装所需的软件包。

pip install Flask Flask-SQLAlchemy

安装所需的软件包后，你必须导入必要的软件包和方法，配置 Flask 应用程序，并使用 SQLalachemy 设置 SQLite 数据库。之后，你将定义一个数据库模型来存储和列出任务项。你还需要初始化应用程序以使用 SQLalcamey，并通过创建所需的表来初始化数据库。

from flask import Flask, request, render_template_string, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import Mapped, mapped_column

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///tasks.db"
db = SQLAlchemy()

# Define a database model named Task for storing task data
class Task(db.Model):
   id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
   description: Mapped[str] = mapped_column(db.String(256), nullable=False)

# Initialize SQLAlchemy with the configured Flask application
db.init_app(app)

# Initialize the database within the application context
with app.app_context():
   db.create_all()  # Creates all tables

你现在可以设置 HTML 模板来创建待办事项列表应用程序的前端，包括用于添加任务的内联 CSS 和表单内容。你还将定义用于列出现有任务和删除任务的其他功能。

HOME_HTML = """
<!doctype html>
<html lang="en">
<head>
 <meta charset="UTF-8">
 <title>To-Do List</title>
 <style>
   body {
     font-family: Arial, sans-serif;
     background-color: #f4f4f9;
     margin: 40px auto;
     padding: 20px;
     max-width: 600px;
     box-shadow: 0 0 10px rgba(0,0,0,0.1);
   }
   h1 {
     color: #333;
   }
   form {
     margin-bottom: 20px;
   }
   input[type="text"] {
     padding: 10px;
     width: calc(100% - 22px);
     margin-bottom: 10px;
   }
   input[type="submit"] {
     background-color: #5cb85c;
     border: none;
     color: white;
     padding: 10px 20px;
     text-transform: uppercase;
     letter-spacing: 0.05em;
     cursor: pointer;
   }
   ul {
     list-style-type: none;
     padding: 0;
   }
   li {
     position: relative;
     padding: 8px;
     background-color: #fff;
     border-bottom: 1px solid #ddd;
   }
   .delete-button {
     position: absolute;
     right: 10px;
     top: 10px;
     background-color: #ff6347;
     color: white;
     border: none;
     padding: 5px 10px;
     border-radius: 5px;
     cursor: pointer;
   }
 </style>
</head>
<body>
 <h1>To-Do List</h1>
 <form action="/add" method="post">
   <input type="text" name="task" placeholder="Add new task">
   <input type="submit" value="Add Task">
 </form>
 <ul>
   {% for task in tasks %}
     <li>{{ task.description }} <button class="delete-button" onclick="location.href='/delete/{{ task.id }}'">Delete</button></li>
   {% endfor %}
 </ul>
</body>
</html>
"""

现在，你可以创建路由，以便在加载时在应用程序上显示待办事项列表任务、添加新任务和删除任务。

/ 路由允许你定义当有人访问应用程序主页时返回哪些数据；在这种情况下，你从数据库输入的所有待办事项列表任务都将显示在屏幕上。

对于添加新任务，当你在应用程序上填写表单以添加和提交新任务时，/add 路由会将此新任务保存在数据库中。保存任务后，它会将你送回主页，以便他们可以看到添加了新任务的列表。

你还将为 /delete 定义一个路由，它描述了当你删除任务时会发生什么（在本例中，当你单击任务旁边的删除按钮时）。然后，应用程序会从数据库中删除该任务。

最后，你将添加逻辑来运行应用程序。

# Define route for the home page to display tasks
@app.route("/", methods=["GET"])
def home():
   tasks = Task.query.all()  # Retrieve all tasks from the database
   return render_template_string(
       HOME_HTML, tasks=tasks
   )  # Render the homepage with tasks listed

# Define route to add new tasks from the form submission
@app.route("/add", methods=["POST"])
def add():
   task_description = request.form["task"]  # Extract task description from form data
   new_task = Task(description=task_description)  # Create new Task instance
   db.session.add(new_task)  # Add new task to database session
   db.session.commit()  # Commit changes to the database
   return redirect(url_for("home"))  # Redirect to the home page

# Define route to delete tasks based on task ID
@app.route("/delete/<int:task_id>", methods=["GET"])
def delete(task_id: int):
   task_to_delete = Task.query.get(task_id)  # Get task by ID
   if task_to_delete:
       db.session.delete(task_to_delete)  # Remove task from the database session
       db.session.commit()  # Commit the change to the database
   return redirect(url_for("home"))  # Redirect to the home page

# Check if the script is the main program and run the app
if __name__ == "__main__":
   app.run()  # Start the Flask application

要在本地运行你的应用程序，你可以在终端中运行以下命令。

flask run -p 5000

检测应用程序

检测是指向应用程序添加可观察性功能以收集遥测数据，例如跟踪、指标和日志。在数据中，你可以看到正在运行的依赖服务；例如，你可以看到正在构建的应用程序（在此示例中）使用 SQLite。你还可以跟踪请求在分布式系统中跨度移动的各种服务，并查看有关在分布式系统中运行的进程的定量信息。

自动检测与手动检测

你有两种检测应用程序的选项：自动检测和手动检测。

自动检测会修改应用程序类的字节码以将监控代码插入到应用程序中。使用自动检测，你可以轻松监控应用程序，而无需担心创建自定义监控。这是一种开始监控应用程序或向现有应用程序添加监控功能的好方法。

手动检测允许你向应用程序添加自定义代码段以收集和传输遥测数据。如果你正在寻找自定义或发现自动检测仅涵盖你需要的部分内容，则它很有用。

向你的待办事项列表应用程序添加自动检测

要向你的 Flask 应用程序添加自动检测，你无需添加任何其他监控代码。当你运行应用程序时，OpenTelemetry 将通过 Python 路径自动添加所需的代码。你可以按照以下步骤向你的应用程序添加自动检测。

步骤 1：安装所需的软件包

首先，你需要安装检测所需的软件包。

pip install elastic-opentelemetry opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-instrumentation-flask
opentelemetry-bootstrap --action=install

第 2 步：设置本地环境变量

现在，你可以设置环境变量以包含应用程序的服务名称、API 密钥和弹性主机端点。

export OTEL_RESOURCE_ATTRIBUTES=service.name=todo-list-app
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=ApiKey <your-api-key>"
export OTEL_EXPORTER_OTLP_ENDPOINT=https://<your-elastic-url>

步骤 3：运行应用程序

要使用 OpenTelemetry 运行你的应用程序，你需要在终端中运行以下命令：

opentelemetry-instrument flask run -p 5000

此时，如果你查看 Kibana，其中显示 “observability” 紧接着 “services”，你应该会看到你的服务列为你在环境变量中设置的服务名称。

如果你点击 service 名称，你应该会看到一个仪表板，其中包含待办事项列表应用程序的可观察性数据。下面的 “Transactions” 部分中的屏幕截图显示了你可以采取的操作，例如在使用 GET 方法加载应用程序时加载所有待办事项列表项，以及使用 POST 方法将新项目添加到待办事项列表中。

向你的待办事项列表应用程序添加手动检测

由于手动检测允许你根据自己的喜好自定义检测，因此你必须将自己的监控代码添加到你的应用程序中才能启动并运行。本节的完整代码示例可在此处找到。

首先，你需要通过设置以下环境变量来更新你的服务名称：

export OTEL_RESOURCE_ATTRIBUTES=service.name=mi-todo-list-app

现在你已经更新了 service 名称，你需要更新导入语句以包含更多功能，用于监控应用程序、解析环境变量以及确保标头格式正确。

import os
from flask import Flask, request, render_template_string, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import Mapped, mapped_column
from opentelemetry import trace, metrics
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

# Get environment variables
service_name = os.getenv("OTEL_RESOURCE_ATTRIBUTES", "service.name=todo-flask-app").split("=")[-1]
otlp_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
otlp_headers = os.getenv("OTEL_EXPORTER_OTLP_HEADERS")

if not otlp_endpoint or not otlp_headers:
   raise ValueError("OTEL_EXPORTER_OTLP_ENDPOINT and OTEL_EXPORTER_OTLP_HEADERS must be set in environment variables")

# Ensure headers are properly formatted for gRPC metadata
headers_dict = dict(item.split(":", 1) for item in otlp_headers.split(",") if ":" in item)

现在，你将需要配置你的应用程序以生成、批处理和发送跟踪数据到 Elastic，从而深入了解你的应用程序的运行和性能。

# Configure tracing provider and exporter
resource = Resource(attributes={
    "service.name": service_name
})
trace.set_tracer_provider(TracerProvider(resource=resource))
tracer_provider = trace.get_tracer_provider()

otlp_trace_exporter = OTLPSpanExporter(
    endpoint=otlp_endpoint,
    headers=headers_dict,
)
span_processor = BatchSpanProcessor(otlp_trace_exporter)
tracer_provider.add_span_processor(span_processor)

你现在可以设置用于捕获遥测数据的框架。你需要创建跟踪器、计量器和计数器。跟踪器为分布式跟踪创建跨度，帮助了解分布式系统的流程和性能问题。计量器和计数器捕获操作指标（如计数请求）对于生产环境中的性能监控和警报至关重要。你的指标配置可确保这些指标得到适当批处理并发送到 Elastic 进行分析。

# Create a tracer
tracer = trace.get_tracer(__name__)

# Configure metrics provider and exporter
otlp_metric_exporter = OTLPMetricExporter(
    endpoint=otlp_endpoint,
    headers=headers_dict,
)
metric_reader = PeriodicExportingMetricReader(otlp_metric_exporter)
meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
metrics.set_meter_provider(meter_provider)

# Create a meter
meter = metrics.get_meter(__name__)
requests_counter = meter.create_counter(
    name="requests_count",
    description="Number of requests received",
    unit="1",
)

你将需要使用 Flask 和 SQLite 来获取有关可观察性后端（即 Elastic）中的这两项服务的信息。

FlaskInstrumentor().instrument_app(app)
with app.app_context():
    SQLAlchemyInstrumentor().instrument(engine=db.engine)

现在，你可以为每个应用程序路由配备跟踪和指标收集功能。这样你就可以定义向每个路由（GET、POST 和 DELETE）添加哪些跟踪和指标，从而让你能够了解运营绩效，同时还可以收集有关用户交互和系统效率的宝贵数据。

# Define route for the home page to display tasks
@app.route("/", methods=["GET"])
def home():
   with app.app_context():
       with tracer.start_as_current_span("home-request"):
           requests_counter.add(1, {"method": "GET", "endpoint": "/"})
           tasks = Task.query.all()  # Retrieve all tasks from the database
           return render_template_string(
               HOME_HTML, tasks=tasks
           )  # Render the homepage with tasks listed

# Define route to add new tasks from the form submission
@app.route("/add", methods=["POST"])
def add():
   with app.app_context():
       with tracer.start_as_current_span("add-task"):
           requests_counter.add(1, {"method": "POST", "endpoint": "/add"})
           task_description = request.form["task"]  # Extract task description from form data
           new_task = Task(description=task_description)  # Create new Task instance
           db.session.add(new_task)  # Add new task to database session
           db.session.commit()  # Commit changes to the database
           return redirect(url_for("home"))  # Redirect to the home page

# Define route to delete tasks based on task ID
@app.route("/delete/<int:task_id>", methods=["GET"])
def delete(task_id: int):
   with app.app_context():
       with tracer.start_as_current_span("delete-task"):
           requests_counter.add(1, {"method": "GET", "endpoint": f"/delete/{task_id}"})
           task_to_delete = Task.query.get(task_id)  # Get task by ID
           if task_to_delete:
               db.session.delete(task_to_delete)  # Remove task from the database session
               db.session.commit()  # Commit the change to the database
           return redirect(url_for("home"))  # Redirect to the home page

由于你已将自定义监控代码应用到你的应用程序中，因此你可以像第一次创建它时一样在终端中运行。