Przejdź do sekcji TLDR celem uzyskania szybkiej odpowiedzi na pytanie w tytule artykułu.
Zapisz się na newsletter, jeśli nie chcesz przegapić kolejnych publikacji.
Studium przypadku
W środowisku rozproszonym, opartym o mikro-serwisy, istotną rolę odgrywa instrumentacja (ang. instrumentation), czyli dodanie fragmentów kodu, które umożliwiają dokładne monitorowanie działania aplikacji oraz mierzenie jej wydajności. Dzięki temu jesteśmy w stanie skutecznie zidentyfikować usługę, która może stanowić wąskie gardło całego systemu.
W jednym z projektów Django, nad którym swego czasu pracowałem, zaimplementowano instrumentację z wykorzystaniem narzędzi OpenTelemetry. Do propagacji kontekstu użyto propagatora B3, zgodnego ze standardem OpenTelemetry, który informacje o kontekście przekazuje poprzez nagłówki HTTP, o nazwie rozpoczynającej się prefiksem X-B3-
.
W pewnym momencie zaszła potrzeba uruchomienia projektu na chmurze Google (Google Cloud Platform) i okazało się, że nasza usługa chmurowa Google używa trochę innego nagłówka do przekazywania informacji o kontekście, a mianowicie: X-Cloud-Trace-Context
.
Bez wdawania się w szczegóły decyzyjne: nie chcieliśmy na tym etapie zmieniać sposobu propagacji w usłudze, więc musieliśmy odczytać wartość trace_id
z nagłówka X-Cloud-Trace-Context
i umieścić ją w innym nagłówku X-B3-TraceId
na poziomie żądania (HTTP request), zanim jeszcze zostanie obsłużone przez kod widoków.
Rozwiązanie z użyciem Django middleware
Żeby zmodyfikować samo żądanie HTTP, modyfikując nagłówki, najlepiej posłużyć się mechanizmem middleware. W dosłownym tłumaczeniu jest to oprogramowanie pośredniczące i możemy je sobie wyobrazić jako system wtyczek, przez które przechodzą żądania (requests) oraz odpowiedzi (responses) HTTP podczas standardowej komunikacji:
![](https://vilya.pl/wp-content/uploads/2024/07/middleware.png)
W uproszczeniu: kod middleware ma dostęp do danych żądania (HttpRequest
), zanim zostanie ono obsłużone przez metodę widoku, a także do danych odpowiedzi (HttpResponse
) zwróconej przez widok, zanim trafi ona do użytkownika.
Django pozwala na definiowanie własnych klas/funkcji middleware, które później należy zarejestrować w ustawieniu settings.MIDDLEWARE
.
Warto zwrócić uwagę, że kolejność rejestracji komponentów middleware ma znaczenie. Na powyższym schemacie, żądanie zostanie obsłużone w następującej kolejności: Middleware 1
-> Middleware 2
-> Middleware 3
. Odpowiedź natomiast przejdzie przez warstwę pośredniczącą w odwrotnej kolejności: Middleware 3
-> Middleware 2
-> Middleware 1
.
Podsumowując, komponent middleware jest odpowiednim miejscem na realizację naszego założenia, czyli przepisania wartości odczytanej z nagłówka X-Cloud-Trace-Context
i umieszczeniu jej w innym nagłówku żądania HTTP.
PoC
- Stwórzmy możliwie najprostszy projekt Django i dodajmy do niego aplikację o nazwie telemetry. Jeśli nigdy wcześniej nie pracowałeś(-aś) z Django, polecam zapoznać się z oficjalnym wprowadzeniem.
- Zdefiniujmy prostą metodę widoku (dostępną pod ścieżką
/telemetry
), której zadaniem jest zwrócenie nagłówków żądania HTTP w formacie JSON:
from django.http import JsonResponse
def index(request):
return JsonResponse(dict(request.headers))
# ...
urlpatterns = [
path("telemetry/", include("telemetry.urls")),
# ...
]
# ...
urlpatterns = [
path("", views.index, name="index"),
]
- Po uruchomieniu aplikacji z użyciem lokalnego serwera deweloperskiego (
python manage.py runserver
), możemy wysłać żądanie z ustawionym nagłówkiemX-B3-Traceid
i ujrzeć przykładową odpowiedź (przejrzyście sformatowaną dzięki narzędziu jq):
$ curl -s -H "X-B3-TraceId: 463ac35c9f6413ad48485a3953bb6124" http://localhost:8000/telemetry/ | jq
{
"Content-Length": "",
"Content-Type": "text/plain",
"Host": "localhost:8000",
"User-Agent": "curl/7.68.0",
"Accept": "*/*",
"X-B3-Traceid": "463ac35c9f6413ad48485a3953bb6124"
}
- Zgodnie z podejściem TDD (Test-Driven Development), zacznijmy od napisania automatycznych testów funkcjonalnych. Rozważmy 2 proste scenariusze:
- Fragment wartości nagłówka
X-Cloud-Trace-Context
jest przeniesiony doX-B3-Traceid
, jeśli takowego nie ma w żądaniu. - Jeśli nagłówek
X-B3-Traceid
jest już zdefiniowany w żądaniu, to zignoruj nagłówekX-Cloud-Trace-Context
.
- Fragment wartości nagłówka
import json
from django.test import Client, TestCase
class TestTelemetryMiddleware(TestCase):
X_CLOUD_TRACE_CONTEXT_HEADER_NAME = "X-Cloud-Trace-Context"
X_B3_TRACEID_HEADER_NAME = "X-B3-Traceid"
def setUp(self):
self.client = Client()
def test_copies_trace_id_from_X_CLOUD_TRACE_CONTEXT_to_X_B3_TRACEID_request_header(
self,
):
# Arrange
valid_trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"
headers = {
self.X_CLOUD_TRACE_CONTEXT_HEADER_NAME: self.__build_trace_context(
valid_trace_id
)
}
# Act
response_data = self.__send_get_request("/telemetry/", headers)
# Assert
assert response_data[self.X_B3_TRACEID_HEADER_NAME] == valid_trace_id
def test_ignores_X_CLOUD_TRACE_CONTEXT_if_X_B3_TRACEID_already_exists(self):
# Arrange
valid_trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"
valid_b3_trace_id = "f5196e6103f4ad68e25576e59a7c5699"
headers = {
self.X_CLOUD_TRACE_CONTEXT_HEADER_NAME: self.__build_trace_context(
valid_trace_id
),
self.X_B3_TRACEID_HEADER_NAME: valid_b3_trace_id,
}
# Act
response_data = self.__send_get_request("/telemetry/", headers)
# Assert
assert response_data[self.X_B3_TRACEID_HEADER_NAME] == valid_b3_trace_id
def __build_trace_context(
self, trace_id: str, span_id: str = "17725314949316355921", sampled: bool = True
) -> str:
return f"{trace_id}/{span_id};o={int(sampled)}"
def __send_get_request(self, path: str, headers: dict) -> dict:
# Make headers compatible with WSGI environment variables.
meta_headers = {
f"HTTP_{name.replace('-', '_')}": value for name, value in headers.items()
}
response = self.client.get(path, **meta_headers)
return json.loads(response.content.decode("utf-8"))
- Implementacja klasy
TelemetryMiddleware
, realizującej pożądaną przez nas funkcjonalność, wymaga prawidłowego sparsowania zawartości nagłówkaX-Cloud-Trace-Context
. Co prawda nie jest to trudne zadanie, ale zamiast wyważać otwarte drzwi, możemy podejrzeć, jak to zostało zrobione w udostępnionej przez Google bibliotece open source: opentelemetry-operations-python.
import re
from django.http import HttpRequest
# Make sure the regular expresson is compiled only once.
X_CLOUD_TRACE_CONTEXT_RE = re.compile(
r"(?P<trace_id>[0-9a-f]{32})\/(?P<span_id>[\d]{1,20})(;o=(?P<trace_flags>\d+))?"
)
class TelemetryMiddleware:
X_CLOUD_TRACE_CONTEXT_HEADER_NAME = "X-Cloud-Trace-Context"
X_B3_TRACEID_HEADER_NAME = "X-B3-Traceid"
X_B3_TRACEID_META_HEADER_NAME = f"HTTP_{X_B3_TRACEID_HEADER_NAME.replace('-', '_')}"
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request: HttpRequest):
if self.X_B3_TRACEID_HEADER_NAME not in request.headers:
trace_id = self.__extract_trace_id_from_context_header(request)
request.META[self.X_B3_TRACEID_META_HEADER_NAME] = trace_id
# Enforce the "HttpRequest.headers" cached property to be refreshed.
request.__dict__.pop("headers", None)
response = self.get_response(request)
# Code to be executed for each request/response after
# the view is called.
return response
def __extract_trace_id_from_context_header(self, request: HttpRequest) -> str:
trace_context_header = request.headers.get(
self.X_CLOUD_TRACE_CONTEXT_HEADER_NAME
)
if trace_context_header is None:
return ""
match = re.fullmatch(X_CLOUD_TRACE_CONTEXT_RE, trace_context_header)
if match is None:
return ""
return match.group("trace_id")
- Warto zwrócić uwagę na fragment kodu
request.__dict__.pop("headers", None)
, który powoduje, że wartośćHttpRequest.headers
(cached property) zostanie odświeżona przy kolejnej próbie odczytu. Jest to konieczne, żeby właściwość zawierała wprowadzone przez nas zmiany dorequest.META
. - Ostatnim krokiem jest zarejestrowanie klasy middleware w ustawieniach:
# ...
MIDDLEWARE = [
"telemetry.middleware.TelemetryMiddleware",
# ...
]
# ...
- Po uruchomieniu testów (
python manage.py test
), wszystkie powinny zakończyć się sukcesem.
TLDR
Podsumowując, jeśli chcemy zmodyfikować nagłówki żądania HTTP przed obsłużeniem przez kod widoków Django:
- Definiujemy klasę (lub metodę) middleware, w której modyfikujemy nagłówki żądania:
class MyMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
# Kod wykonywany dla każdego żądania, zanim trafi do widoku.
# Dodajemy nowy nagłówek w odpowiednim formacie:
# HTTP_MY_HEADER oznacza nagłówek "My-Header".
request.META["HTTP_MY_HEADER"] = "header_value"
# Wymuszamy odświeżenie cached property HttpReqest.headers.
request.__dict__.pop("headers", None)
response = self.get_response(request)
# Kod wykonywany dla każdej odpowiedzi, zwróconej przez widok.
return response
- Rejestrujemy komponent w ustawieniach
settings.py
:
MIDDLEWARE = [
"path.to.MyMiddleware",
# ...
]
Przydatne materiały
- Unit Testing Django Middleware – jak przetestować sam komponent middleware przy użyciu testów jednostkowych.
- StackOverflow: Django overwrite header value in request object – odrobinę szersze wyjaśnienie dotyczące odświeżenia właściwości
HttpRequest.headers
.