Observer, chiffrer, nettoyer

Six semaines après la mise en service, la directrice financière pose deux questions : « combien de documents sont passés hier, et combien en erreur ? » et « combien ça coûte ? ». Si la réponse demande d'ouvrir la console et de compter des lignes de journal, le pipeline n'est pas exploitable. La troisième question viendra de l'équipe plateforme à la fin du projet pilote : « est-ce qu'il reste quelque chose de facturé ? ». Cette leçon donne une réponse vérifiable à chacune.

Le mécanisme

flowchart LR
  W["Worker<br/>1 ligne JSON par événement"] --> CL["Cloud Logging<br/>jsonPayload interrogeable"]
  CL --> M["Métrique basée sur les journaux<br/>lab-pipeline-evt-documents{status}"]
  M --> MON["Cloud Monitoring<br/>tableaux, alertes"]
  W --> BQ[("BigQuery<br/>tokens, latence, tentative")]
  BQ --> COUT["Coût par document<br/>(requête SQL)"]

Deux sources, deux usages :

• Les journaux structurés : le worker écrit une ligne JSON par événement sur la sortie standard. Cloud Run la transmet à Cloud Logging, qui en fait un jsonPayload dont chaque champ est filtrable (jsonPayload.event="document_echec"). Le champ severity est reconnu et devient la sévérité de l'entrée ; logging.googleapis.com/trace rattache la ligne à la trace de la requête. Une métrique basée sur les journaux compte ces lignes, ventilées par le champ status : c'est elle qui alimente les tableaux de bord et les alertes.
• BigQuery : chaque ligne porte le modèle, les tokens d'entrée et de sortie, la latence Gemini et le numéro de tentative. Le coût et la qualité se calculent en SQL, sans outil de plus.

La mise en œuvre

class JsonFormatter(logging.Formatter):
    """Une ligne JSON par événement : Cloud Logging en fait un jsonPayload interrogeable."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {"severity": record.levelname, "message": record.getMessage()}
        entry.update(getattr(record, "fields", {}))
        if record.exc_info:
            entry["stack_trace"] = self.formatException(record.exc_info)
        return json.dumps(entry, ensure_ascii=False, default=str)

description: Documents traités par le worker, par statut (ok, rejected, duplicate)
filter: >-
  resource.type="cloud_run_revision"
  AND resource.labels.service_name="lab-pipeline-evt-worker"
  AND jsonPayload.event="document_traite"
labelExtractors:
  status: EXTRACT(jsonPayload.status)
metricDescriptor:
  metricKind: DELTA
  valueType: INT64
  labels:
    - key: status
      valueType: STRING

La requête du coût par document, plafonnée comme toutes les autres :

bq query --use_legacy_sql=false --maximum_bytes_billed=104857600 '
SELECT COUNT(*) n, SUM(input_tokens) entree, SUM(output_tokens) sortie,
       APPROX_QUANTILES(latency_ms, 2)[OFFSET(1)] latence_mediane, MAX(delivery_attempt) max_tentative
FROM `<projet>.lab_pipeline_evt.analyses` WHERE status = "ok" AND processed_at > "2026-10-16 13:00:00"'

Résultat obtenu (2026-10-16)

Journaux du worker pendant l'e2e, regroupés par événement et statut :

$ gcloud logging read 'resource.labels.service_name="lab-pipeline-evt-worker" AND jsonPayload.event!=""' ...
      9 document_traite  ok
      1 document_traite  rejected
      1 document_traite  duplicate
     10 message_empoisonne   message sans attribut eventType : il ne vient pas de Cloud Storage

Valeurs de la métrique sur la dernière heure, lues par l'API Cloud Monitoring : {'ok': 9, 'rejected': 1, 'duplicate': 1}. Et la requête de coût :

+---+--------+--------+-----------------+---------------+
| n | entree | sortie | latence_mediane | max_tentative |
+---+--------+--------+-----------------+---------------+
| 9 |   3696 |    968 |            1101 |             1 |
+---+--------+--------+-----------------+---------------+

Ce que coûtent 1 000 documents

Calcul à partir de ces mesures (environ 410 tokens d'entrée et 110 de sortie par document) et des tarifs publics de Vertex AI au 2026-10-16 (Gemini 2.5 Flash : 0,30 $par million de tokens d'entrée, 2,50$ en sortie) :

Deux enseignements. La sortie coûte plus que l'entrée, alors qu'elle est quatre fois plus courte : un résumé en deux phrases plutôt qu'en cinq est la première économie. Et BigQuery facture 10 Mo par requête même quand il lit 969 octets (mesuré par --dry_run sur la requête seen()) : à fort volume, la déduplication à la lecture (leçon 2) coûte moins que deux requêtes par document.

Nettoyer

$ make cloud-status
run.service lab-pipeline-evt-worker
pubsub.topic lab-pipeline-evt-docs
pubsub.topic lab-pipeline-evt-dlq
pubsub.subscription lab-pipeline-evt-push
pubsub.subscription lab-pipeline-evt-dlq-sub
storage.bucket lab-pipeline-evt-<projet>
artifacts.repo lab-pipeline-evt
logging.metric lab-pipeline-evt-documents
iam.sa lab-pipeline-evt-push@<projet>.iam.gserviceaccount.com
iam.sa lab-pipeline-evt-run@<projet>.iam.gserviceaccount.com
iam.binding roles/aiplatform.user
iam.binding roles/bigquery.jobUser
bigquery.dataset lab_pipeline_evt

$ make cloud-down
==> Abonnements et sujets Pub/Sub
==> Service Cloud Run, métrique, dépôt d'images
==> Bucket (objets, notification et sources de build compris) et dataset BigQuery
==> Droits au niveau projet, puis comptes de service
iam.sa lab-pipeline-evt-push@<projet>.iam.gserviceaccount.com
ÉCHEC : des ressources du lab subsistent.

Ce premier échec est réel et instructif : le compte de service était supprimé, mais la liste IAM, à cohérence différée, le montrait encore. make cloud-status lancé dix secondes plus tard répondait « Aucune ressource ». Le script vérifie désormais trois fois à 15 s d'intervalle avant de conclure ; relancé, il donne :

$ make cloud-down
==> Abonnements et sujets Pub/Sub
==> Service Cloud Run, métrique, dépôt d'images
==> Bucket (objets, notification et sources de build compris) et dataset BigQuery
==> Droits au niveau projet, puis comptes de service
Aucune ressource du lab pipeline-evt dans le projet.

Restent, par conception, les journaux déjà écrits (conservés 30 jours dans le bucket de journaux _Default, puis effacés) et l'historique des builds Cloud Build, qui ne sont pas facturés au repos.

Ce qui casse en production

• Métrique qui compte des lignes, pas des documents. Un document_traite est écrit par livraison réussie : les duplicate gonflent le total. Parade : ventiler par status, et prendre ok + rejected pour le volume de documents.
• Journaux non structurés. Un print() laissé dans le code arrive en textPayload, sans sévérité : les filtres et la métrique l'ignorent, une erreur passe inaperçue. Parade : un seul logger configuré, et pas de print dans le worker.
• Coût de la journalisation. Écrire le texte complet du document ou la réponse Gemini dans les journaux multiplie le volume ingéré (facturé au-delà de l'allocation gratuite) et y copie des données personnelles. Parade : identifiants et mesures dans les journaux, contenu dans BigQuery.
• Nettoyage faussement réussi ou faussement raté. Un script qui supprime sans vérifier laisse un compte ou une liaison ; un script qui vérifie trop tôt échoue sur une liste en retard, comme ici. Parade : vérification par préfixe et label, répétée quelques secondes plus tard, code de sortie non nul si quelque chose subsiste.

La règle de décision

Une ligne JSON par événement métier, avec les identifiants qui permettent de relier journal, message et ligne BigQuery ; une métrique par décision du contrat (ok, rejected, duplicate, échec) et une alerte sur les échecs et la file de lettres mortes. Le coût se calcule à partir des tokens mesurés, pas estimés. Un environnement de test se supprime par la commande qui l'a créé, et la suppression se prouve par une liste vide.

Pratique

Exercice. Écrivez la politique d'alerte qui prévient l'astreinte quand la file de lettres mortes contient au moins un message depuis plus de 10 minutes. Quelle métrique, quelle condition, quel délai ?

displayName: "Lettres mortes : lab-pipeline-evt"
combiner: OR
conditions:
  - displayName: "Messages non distribués dans lab-pipeline-evt-dlq-sub"
    conditionThreshold:
      filter: >-
        resource.type="pubsub_subscription"
        AND resource.labels.subscription_id="lab-pipeline-evt-dlq-sub"
        AND metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages"
      comparison: COMPARISON_GT
      thresholdValue: 0
      duration: 600s
      aggregations:
        - alignmentPeriod: 60s
          perSeriesAligner: ALIGN_MAX

• J'écris des journaux JSON que Cloud Logging indexe, avec sévérité et identifiants.
• Je crée une métrique basée sur les journaux ventilée par statut.
• Je calcule le coût par document à partir des tokens mesurés.
• Je supprime l'environnement et je prouve qu'il ne reste rien.