← Google Cloud pour les devs : gcloud et Cloud Logging

Écrire des logs structurés depuis Python

≈ 30 minutes · Jalon M3 · Prérequis : Lire et filtrer les logs : Logs Explorer et gcloud logging read

Le gain du jour : faire écrire à ton service Python des logs JSON que Cloud Logging comprend tout seul (sévérité, trace, champs métier), puis retrouver une ligne précise avec un seul filtre jsonPayload.userId="…".

Un log structuré, c’est quoi ?

Un log texte est une phrase : Paiement refusé pour u-42, montant 30. Un humain la lit. Une machine doit la deviner.

Un log structuré est un objet avec des champs nommés :

{"severity": "WARNING", "message": "Paiement refusé", "userId": "u-42", "amount": 30}

Cloud Logging range chaque champ à sa place. Tu peux alors filtrer sur userId, trier par severity, compter les amount. Avec du texte, tu en serais réduit à des expressions régulières fragiles.

Deux bénéfices concrets :

• La sévérité est juste. Un print("ERROR ...") en texte arrive avec la sévérité DEFAULT. Ton filtre severity>=ERROR ne le voit pas.
• Les champs sont interrogeables. jsonPayload.userId="u-42" est exact, rapide et ne casse pas si tu reformules ton message.

Il existe deux chemins pour y arriver depuis Python. On commence par le plus simple.

Chemin 1 : une ligne JSON sur stdout

Sur Cloud Run, Cloud Run functions, GKE et App Engine, un agent de logging est déjà intégré. Il lit ce que ton conteneur écrit sur stdout et stderr. Si une ligne est un objet JSON sur une seule ligne, il la transforme en entrée structurée. Rien à installer.

 🐍 ton code ── print(json) ──▶ stdout ──▶ 🤖 agent intégré ──▶ 🗂 Cloud Logging
                                                                   (jsonPayload)

L’agent reconnaît quelques champs spéciaux au premier niveau de l’objet. Ils ne restent pas dans jsonPayload : ils remplissent les champs officiels de l’entrée.

Tous les autres champs (userId, amount, ce que tu veux) restent dans jsonPayload.

Pas besoin d’appeler print à la main. Le module standard logging fait le travail, il suffit de lui donner un formateur qui produit du JSON :

import json
import logging
import sys


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "severity": record.levelname,  # DEBUG, INFO, WARNING, ERROR, CRITICAL
            "message": record.getMessage(),
            "logger": record.name,
        }
        # Champs métier passés via extra={"json_fields": {...}}
        entry.update(getattr(record, "json_fields", {}))
        if record.exc_info:
            entry["stack_trace"] = self.formatException(record.exc_info)
        return json.dumps(entry, ensure_ascii=False)


handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])

log = logging.getLogger("panier")
log.warning(
    "Paiement refusé",
    extra={"json_fields": {"userId": "u-42", "amount": 30, "appVersion": "3.2.1"}},
)

Sortie, une seule ligne :

{"severity": "WARNING", "message": "Paiement refusé", "logger": "panier", "userId": "u-42", "amount": 30, "appVersion": "3.2.1"}

Les noms de niveaux de Python (DEBUG, INFO, WARNING, ERROR, CRITICAL) existent tous dans Cloud Logging. Ils passent tels quels.

Chemin 2 : la bibliothèque google-cloud-logging

La bibliothèque officielle fait la même chose, plus quelques remplissages automatiques.

pip install google-cloud-logging

import logging
import google.cloud.logging

client = google.cloud.logging.Client()
client.setup_logging(log_level=logging.INFO)

logging.info(
    "Commande validée",
    extra={"json_fields": {"userId": "u-42", "orderId": "o-981"}},
)

client.setup_logging() branche un handler sur le logger racine de Python. Ensuite tu utilises logging comme d’habitude. Le handler choisi dépend de l’endroit où le code tourne :

Le dictionnaire json_fields se retrouve dans jsonPayload, et ton texte dans jsonPayload.message. Bonus : la bibliothèque ajoute d’elle-même sourceLocation (fichier, ligne, fonction) et un label python_logger avec le nom du logger.

En local, CloudLoggingHandler appelle l’API. Il lui faut des identifiants. Le plus simple est l’ADC (Application Default Credentials) de ton compte :

gcloud auth application-default login
gcloud config set project mon-projet

Ton compte doit avoir le droit d’écrire des logs (rôle roles/logging.logWriter). Sur Cloud Run, c’est le compte de service du service qui sert, aucune clé à gérer.

Relier les logs à la requête

Une requête HTTP produit souvent plusieurs lignes de log. Pour les voir ensemble, chaque ligne doit porter le même trace. Le load balancer de Google ajoute à chaque requête entrante un en-tête de trace, sous l’une de ces deux formes :

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01
X-Cloud-Trace-Context: 4bf92f3577b34da6a3ce929d0e0e4736/12345;o=1

Le morceau qui compte est le trace ID (32 caractères hexadécimaux). Avec la bibliothèque du chemin 2, Flask et Django sont gérés tout seuls : le filtre lit ces en-têtes et remplit trace et spanId. Pour FastAPI, ou avec le formateur maison du chemin 1, tu le fais toi-même. Voici la version Flask, sans bibliothèque :

import logging
import os

from flask import Flask, request

PROJECT = os.environ["GOOGLE_CLOUD_PROJECT"]
app = Flask(__name__)
log = logging.getLogger("api")


def trace_fields() -> dict:
    header = request.headers.get("traceparent", "")
    parts = header.split("-")
    if len(parts) == 4:
        trace_id = parts[1]
    else:
        trace_id = request.headers.get("X-Cloud-Trace-Context", "").split("/")[0]
    if not trace_id:
        return {}
    return {"logging.googleapis.com/trace": f"projects/{PROJECT}/traces/{trace_id}"}


@app.post("/panier/items")
def ajouter_item():
    user_id = request.json["userId"]
    log.info("Ajout au panier", extra={"json_fields": {**trace_fields(), "userId": user_id}})
    return {"ok": True}

Le préfixe projects/PROJET/traces/ est obligatoire. Sans lui, Cloud Logging ne relie pas la ligne à la requête. Avec lui, le Logs Explorer affiche tes lignes sous la ligne de requête de Cloud Run, et trace="projects/…" les retrouve toutes.

Les bonnes pratiques

• Choisis la sévérité pour l’humain qui sera réveillé. INFO pour un événement normal, WARNING pour un problème rattrapé, ERROR pour un échec que quelqu’un doit regarder. Une erreur de validation du client (400) n’est pas un ERROR serveur.
• Un événement par ligne. Une ligne dit une chose, avec tous ses champs. Pas trois lignes à recoller.
• Des noms de champs stables. userId, requestId, appVersion : choisis une écriture et garde-la partout. jsonPayload.userId et jsonPayload.user_id sont deux champs différents, la casse compte dans les filtres.
• Jamais de secrets ni de données personnelles. Pas de mot de passe, de token, de numéro de carte, d’email en clair. Logue un identifiant interne (userId), pas la personne. Un log se copie, s’exporte et se garde longtemps.
• Pas de log dans une boucle chaude. Une ligne par article dans un panier de 10 000 articles, c’est 10 000 entrées. Logue un résumé (count, durationMs) à la fin.
• DEBUG coûte cher en prod. L’ingestion de logs est facturée au volume au-delà d’un quota gratuit. Garde log_level=logging.INFO en production et active DEBUG par variable d’environnement, le temps d’une enquête.

Vérifier que ça marche

Déploie, déclenche une requête, puis interroge Cloud Logging depuis ton terminal :

gcloud logging read 'jsonPayload.userId="u-42"' --limit 5

Pour ne garder que les problèmes du service, avec la sortie JSON complète :

gcloud logging read 'resource.labels.service_name="api-panier" AND severity>=WARNING' \
  --limit 5 --format json

Ce que tu dois voir : severity au premier niveau (et non DEFAULT), tes champs métier dans jsonPayload, et un trace qui commence par projects/. Si tu vois un textPayload contenant ton JSON, la ligne n’a pas été reconnue : elle était sans doute sur plusieurs lignes ou mal formée.

Vérifie ta compréhension

À toi de jouer

1. Ajoute le JsonFormatter du chemin 1 à un petit service Python et vérifie en local qu’il écrit une seule ligne JSON par appel.

2. Logue un événement métier avec un userId dans json_fields, déploie sur Cloud Run, puis retrouve-le avec gcloud logging read ‘jsonPayload.userId=”…”’ —limit 5.

3. Dans une route Flask, lis l’en-tête traceparent ou X-Cloud-Trace-Context, attache le trace, et vérifie que tes lignes s’affichent sous la requête dans le Logs Explorer.

4. Relis tes appels de log : aucun secret, aucune donnée personnelle, aucun log dans une boucle chaude.

🗂 Fiche de référence associée : Google Cloud et Cloud Logging — antisèche.