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_columnapp = 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 databasereturn 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 datanew_task = Task(description=task_description)  # Create new Task instancedb.session.add(new_task)  # Add new task to database sessiondb.session.commit()  # Commit changes to the databasereturn 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 IDif task_to_delete:db.session.delete(task_to_delete)  # Remove task from the database sessiondb.session.commit()  # Commit the change to the databasereturn 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 databasereturn 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 datanew_task = Task(description=task_description)  # Create new Task instancedb.session.add(new_task)  # Add new task to database sessiondb.session.commit()  # Commit changes to the databasereturn 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 IDif task_to_delete:db.session.delete(task_to_delete)  # Remove task from the database sessiondb.session.commit()  # Commit the change to the databasereturn redirect(url_for("home"))  # Redirect to the home page

由于你已將自定義監控代碼應用到你的應用程序中，因此你可以像第一次創建它時一樣在終端中運行。