Lire Cloud Logging et Cloud Monitoring

Les bibliothèques

pip install google-adk==2.10.0 google-cloud-logging google-cloud-monitoring python-dotenv

Les deux clients trouvent seuls leurs identifiants (ADC). On les crée une seule fois, à la demande, avec functools.lru_cache :

@lru_cache
def _logging():
    from google.cloud import logging as cloud_logging
    return cloud_logging.Client(project=_projet())

@lru_cache
def _monitoring():
    from google.cloud import monitoring_v3
    return monitoring_v3.MetricServiceClient()

L'import à l'intérieur de la fonction permet aux tests de remplacer _logging et _monitoring sans jamais contacter Google Cloud.

Le filtre Cloud Logging

Le langage de requête de Cloud Logging combine des conditions sur les champs d'une entrée. Notre outil construit un filtre simple et borné :

heures, max_entrees = max(1, min(int(heures), 72)), max(1, min(int(max_entrees), 1000))
depuis = (datetime.now(timezone.utc) - timedelta(hours=heures)).isoformat()
filtre = f'severity>=ERROR AND timestamp>="{depuis}"'
if journal:
    filtre += f' AND logName="projects/{_projet()}/logs/{journal}"'
entrees = _logging().list_entries(filter_=filtre, order_by=cloud_logging.DESCENDING,
                                  max_results=max_entrees, page_size=max_entrees)

Tester le filtre à la main d'abord

Avant de le coder, essayez toujours un filtre avec gcloud :

gcloud logging read 'severity>=ERROR AND logName="projects/'"$PROJECT_ID"'/logs/lab-supervision"' \
  --project "$PROJECT_ID" --freshness 1h \
  --format 'table(timestamp,severity,jsonPayload.service,jsonPayload.message)'

La métrique log_entry_count

Cloud Logging alimente automatiquement une métrique système dans Cloud Monitoring : logging.googleapis.com/log_entry_count, le nombre d'entrées écrites, avec les libellés log (nom du journal) et severity. Pas besoin de la créer.

series = _monitoring().list_time_series(request={
    "name": f"projects/{_projet()}",
    "filter": 'metric.type="logging.googleapis.com/log_entry_count" AND metric.labels.log="lab-supervision"',
    "interval": monitoring_v3.TimeInterval(start_time=fin - timedelta(hours=heures), end_time=fin),
    "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL,
})
totaux = defaultdict(int)
for s in series:
    totaux[s.metric.labels.get("severity", "?")] += sum(int(p.value.int64_value) for p in s.points)

flowchart LR
  E["Entrées écrites dans lab-supervision"] --> L["Cloud Logging"]
  L -->|"automatique, quelques minutes de retard"| M["Métrique log_entry_count (libellés log, severity)"]
  M -->|"list_time_series"| T["compter_par_severite : {ERROR: 7, WARNING: 2, ...}"]
  L -->|"list_entries"| R["lire_erreurs : entrées détaillées"]

Pourquoi lire les deux ? Les journaux donnent le détail (quels messages) mais on n'en lit qu'un échantillon borné ; la métrique donne le volume total par sévérité, même quand il y a des milliers d'entrées.

Résultat obtenu

Ce que les deux outils ont renvoyé lors du test réel du 2026-10-15, environ cinq minutes après l'écriture de l'incident fictif :

OUTIL  lire_erreurs {'heures': 1, 'journal': 'lab-supervision'}
OUTIL  compter_par_severite {'heures': 1, 'journal': 'lab-supervision'}
  -> {'entrees_lues': 8} 2 corrélation(s)
  -> {'par_severite': {'CRITICAL': 1, 'DEFAULT': 2, 'ERROR': 7, 'INFO': 1, 'WARNING': 2}}

lire_erreurs a lu les 8 entrées ERROR et CRITICAL ; la métrique compte toutes les sévérités. Les 2 entrées DEFAULT sont des entrées de diagnostic que la bibliothèque google-cloud-logging ajoute automatiquement à la première écriture (champ logging.googleapis.com/diagnostic) : elles n'ont rien à voir avec l'incident.

Exercice

Ajoutez un paramètre service à lire_erreurs pour ne garder que les entrées d'un service Cloud Run donné. Écrivez la partie de filtre correspondante.

def lire_erreurs(heures: int = 1, journal: str = "", max_entrees: int = 200, service: str = "") -> dict:
    """...
        service: nom d'un service Cloud Run à cibler ; vide = tous.
    """
    ...
    if service:
        if not re.fullmatch(r"[a-z0-9-]{1,63}", service):
            return {"erreur": "nom de service invalide"}
        filtre += f' AND resource.type="cloud_run_revision" AND resource.labels.service_name="{service}"'

Je sais faire

• Écrire un filtre Cloud Logging borné dans le temps et par sévérité
• Lire des entrées avec google-cloud-logging
• Lire la métrique système log_entry_count avec google-cloud-monitoring
• Borner et valider les arguments d'un outil de lecture